?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
s3:utils: let smbstatus report anonymous signing/encryption explicitly
[Samba.git] / source3 / lib / dbwrap / dbwrap_ctdb.c
blob807b7f5224860d40028efe1db4a9cce46ea20780
1 /*
2 Unix SMB/CIFS implementation.
3 Database interface wrapper around ctdbd
4 Copyright (C) Volker Lendecke 2007-2009
5 Copyright (C) Michael Adam 2009
6
7 This program is free software; you can redistribute it and/or modify
8 it under the terms of the GNU General Public License as published by
9 the Free Software Foundation; either version 3 of the License, or
10 (at your option) any later version.
11
12 This program is distributed in the hope that it will be useful,
13 but WITHOUT ANY WARRANTY; without even the implied warranty of
14 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 GNU General Public License for more details.
16
17 You should have received a copy of the GNU General Public License
18 along with this program. If not, see <http://www.gnu.org/licenses/>.
19 */
20
21 #include "includes.h"
22 #include "system/filesys.h"
23 #include "lib/tdb_wrap/tdb_wrap.h"
24 #include "util_tdb.h"
25 #include "dbwrap/dbwrap.h"
26 #include "dbwrap/dbwrap_ctdb.h"
27 #include "dbwrap/dbwrap_rbt.h"
28 #include "lib/param/param.h"
29
30 #include "ctdb/include/ctdb_protocol.h"
31 #include "ctdbd_conn.h"
32 #include "dbwrap/dbwrap.h"
33 #include "dbwrap/dbwrap_private.h"
34 #include "dbwrap/dbwrap_ctdb.h"
35 #include "g_lock.h"
36 #include "messages.h"
37 #include "messages_ctdb.h"
38 #include "lib/cluster_support.h"
39 #include "lib/util/tevent_ntstatus.h"
40
41 struct db_ctdb_transaction_handle {
42 struct db_ctdb_ctx *ctx;
43 /*
44 * we store the writes done under a transaction:
45 */
46 struct ctdb_marshall_buffer *m_write;
47 uint32_t nesting;
48 bool nested_cancel;
49 char *lock_name;
50 };
51
52 struct db_ctdb_ctx {
53 struct db_context *db;
54 struct tdb_wrap *wtdb;
55 uint32_t db_id;
56 struct db_ctdb_transaction_handle *transaction;
57 struct g_lock_ctx *lock_ctx;
58
59 /* thresholds for warning messages */
60 int warn_unlock_msecs;
61 int warn_migrate_msecs;
62 int warn_migrate_attempts;
63 int warn_locktime_msecs;
64 };
65
66 struct db_ctdb_rec {
67 struct db_ctdb_ctx *ctdb_ctx;
68 struct ctdb_ltdb_header header;
69 struct timeval lock_time;
70 };
71
72 struct ctdb_async_ctx {
73 bool initialized;
74 struct ctdbd_connection *async_conn;
75 };
76
77 static struct ctdb_async_ctx ctdb_async_ctx;
78
79 static int ctdb_async_ctx_init_internal(TALLOC_CTX *mem_ctx,
80 struct tevent_context *ev,
81 bool reinit)
82 {
83 int ret;
84
85 if (reinit) {
86 TALLOC_FREE(ctdb_async_ctx.async_conn);
87 ctdb_async_ctx.initialized = false;
88 }
89
90 if (ctdb_async_ctx.initialized) {
91 return 0;
92 }
93
94 become_root();
95 ret = ctdbd_init_async_connection(
96 mem_ctx,
97 lp_ctdbd_socket(),
98 lp_ctdb_timeout(),
99 &ctdb_async_ctx.async_conn);
100 unbecome_root();
101
102 if (ret != 0) {
103 DBG_ERR("ctdbd_init_async_connection(%s, timeout=%d) "
104 "failed: ret=%d %s\n",
105 lp_ctdbd_socket(),
106 lp_ctdb_timeout(),
107 ret, strerror(ret));
108 return ret;
109 }
110
111 SMB_ASSERT(ctdb_async_ctx.async_conn != NULL);
112
113 ctdb_async_ctx.initialized = true;
114 return 0;
115 }
116
117 static int ctdb_async_ctx_init(TALLOC_CTX *mem_ctx, struct tevent_context *ev)
118 {
119 return ctdb_async_ctx_init_internal(mem_ctx, ev, false);
120 }
121
122 int ctdb_async_ctx_reinit(TALLOC_CTX *mem_ctx, struct tevent_context *ev)
123 {
124 return ctdb_async_ctx_init_internal(mem_ctx, ev, true);
125 }
126
127 static NTSTATUS tdb_error_to_ntstatus(struct tdb_context *tdb)
128 {
129 enum TDB_ERROR tret = tdb_error(tdb);
130
131 return map_nt_error_from_tdb(tret);
132 }
133
134 struct db_ctdb_ltdb_parse_state {
135 void (*parser)(TDB_DATA key, struct ctdb_ltdb_header *header,
136 TDB_DATA data, void *private_data);
137 void *private_data;
138 };
139
140 static int db_ctdb_ltdb_parser(TDB_DATA key, TDB_DATA data,
141 void *private_data)
142 {
143 struct db_ctdb_ltdb_parse_state *state =
144 (struct db_ctdb_ltdb_parse_state *)private_data;
145
146 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
147 return -1;
148 }
149
150 state->parser(
151 key, (struct ctdb_ltdb_header *)data.dptr,
152 make_tdb_data(data.dptr + sizeof(struct ctdb_ltdb_header),
153 data.dsize - sizeof(struct ctdb_ltdb_header)),
154 state->private_data);
155 return 0;
156 }
157
158 static NTSTATUS db_ctdb_ltdb_parse(
159 struct db_ctdb_ctx *db, TDB_DATA key,
160 void (*parser)(TDB_DATA key, struct ctdb_ltdb_header *header,
161 TDB_DATA data, void *private_data),
162 void *private_data)
163 {
164 struct db_ctdb_ltdb_parse_state state;
165 int ret;
166
167 state.parser = parser;
168 state.private_data = private_data;
169
170 ret = tdb_parse_record(db->wtdb->tdb, key, db_ctdb_ltdb_parser,
171 &state);
172 if (ret == -1) {
173 return NT_STATUS_NOT_FOUND;
174 }
175 return NT_STATUS_OK;
176 }
177
178 /*
179 * Store a record together with the ctdb record header
180 * in the local copy of the database.
181 */
182 static NTSTATUS db_ctdb_ltdb_store(struct db_ctdb_ctx *db,
183 TDB_DATA key,
184 struct ctdb_ltdb_header *header,
185 const TDB_DATA *dbufs, int num_dbufs)
186 {
187 TDB_DATA recs[num_dbufs+1];
188 int ret;
189
190 recs[0] = (TDB_DATA) { .dptr = (uint8_t *)header,
191 .dsize = sizeof(struct ctdb_ltdb_header) };
192 memcpy(&recs[1], dbufs, sizeof(TDB_DATA) * num_dbufs);
193
194 ret = tdb_storev(db->wtdb->tdb, key, recs, num_dbufs + 1, TDB_REPLACE);
195
196 return (ret == 0) ? NT_STATUS_OK
197 : tdb_error_to_ntstatus(db->wtdb->tdb);
198
199 }
200
201 /*
202 form a ctdb_rec_data record from a key/data pair
203 */
204 static struct ctdb_rec_data_old *db_ctdb_marshall_record(TALLOC_CTX *mem_ctx, uint32_t reqid,
205 TDB_DATA key,
206 struct ctdb_ltdb_header *header,
207 TDB_DATA data)
208 {
209 size_t length;
210 struct ctdb_rec_data_old *d;
211
212 length = offsetof(struct ctdb_rec_data_old, data) + key.dsize +
213 data.dsize + sizeof(*header);
214 d = (struct ctdb_rec_data_old *)talloc_size(mem_ctx, length);
215 if (d == NULL) {
216 return NULL;
217 }
218 d->length = length;
219 d->reqid = reqid;
220 d->keylen = key.dsize;
221 memcpy(&d->data[0], key.dptr, key.dsize);
222
223 d->datalen = data.dsize + sizeof(*header);
224 memcpy(&d->data[key.dsize], header, sizeof(*header));
225 memcpy(&d->data[key.dsize+sizeof(*header)], data.dptr, data.dsize);
226 return d;
227 }
228
229
230 /* helper function for marshalling multiple records */
231 static struct ctdb_marshall_buffer *db_ctdb_marshall_add(TALLOC_CTX *mem_ctx,
232 struct ctdb_marshall_buffer *m,
233 uint32_t db_id,
234 uint32_t reqid,
235 TDB_DATA key,
236 struct ctdb_ltdb_header *header,
237 TDB_DATA data)
238 {
239 struct ctdb_rec_data_old *r;
240 size_t m_size, r_size;
241 struct ctdb_marshall_buffer *m2 = NULL;
242
243 r = db_ctdb_marshall_record(talloc_tos(), reqid, key, header, data);
244 if (r == NULL) {
245 talloc_free(m);
246 return NULL;
247 }
248
249 if (m == NULL) {
250 m = (struct ctdb_marshall_buffer *)talloc_zero_size(
251 mem_ctx, offsetof(struct ctdb_marshall_buffer, data));
252 if (m == NULL) {
253 goto done;
254 }
255 m->db_id = db_id;
256 }
257
258 m_size = talloc_get_size(m);
259 r_size = talloc_get_size(r);
260
261 m2 = (struct ctdb_marshall_buffer *)talloc_realloc_size(
262 mem_ctx, m, m_size + r_size);
263 if (m2 == NULL) {
264 talloc_free(m);
265 goto done;
266 }
267
268 memcpy(m_size + (uint8_t *)m2, r, r_size);
269
270 m2->count++;
271
272 done:
273 talloc_free(r);
274 return m2;
275 }
276
277 /* we've finished marshalling, return a data blob with the marshalled records */
278 static TDB_DATA db_ctdb_marshall_finish(struct ctdb_marshall_buffer *m)
279 {
280 TDB_DATA data;
281 data.dptr = (uint8_t *)m;
282 data.dsize = talloc_get_size(m);
283 return data;
284 }
285
286 /*
287 loop over a marshalling buffer
288
289 - pass r==NULL to start
290 - loop the number of times indicated by m->count
291 */
292 static struct ctdb_rec_data_old *db_ctdb_marshall_loop_next_key(
293 struct ctdb_marshall_buffer *m, struct ctdb_rec_data_old *r, TDB_DATA *key)
294 {
295 if (r == NULL) {
296 r = (struct ctdb_rec_data_old *)&m->data[0];
297 } else {
298 r = (struct ctdb_rec_data_old *)(r->length + (uint8_t *)r);
299 }
300
301 key->dptr = &r->data[0];
302 key->dsize = r->keylen;
303 return r;
304 }
305
306 static bool db_ctdb_marshall_buf_parse(
307 struct ctdb_rec_data_old *r, uint32_t *reqid,
308 struct ctdb_ltdb_header **header, TDB_DATA *data)
309 {
310 if (r->datalen < sizeof(struct ctdb_ltdb_header)) {
311 return false;
312 }
313
314 *reqid = r->reqid;
315
316 data->dptr = &r->data[r->keylen] + sizeof(struct ctdb_ltdb_header);
317 data->dsize = r->datalen - sizeof(struct ctdb_ltdb_header);
318
319 *header = (struct ctdb_ltdb_header *)&r->data[r->keylen];
320
321 return true;
322 }
323
324 /**
325 * CTDB transaction destructor
326 */
327 static int db_ctdb_transaction_destructor(struct db_ctdb_transaction_handle *h)
328 {
329 NTSTATUS status;
330
331 status = g_lock_unlock(h->ctx->lock_ctx,
332 string_term_tdb_data(h->lock_name));
333 if (!NT_STATUS_IS_OK(status)) {
334 DEBUG(0, ("g_lock_unlock failed for %s: %s\n", h->lock_name,
335 nt_errstr(status)));
336 return -1;
337 }
338 return 0;
339 }
340
341 /**
342 * CTDB dbwrap API: transaction_start function
343 * starts a transaction on a persistent database
344 */
345 static int db_ctdb_transaction_start(struct db_context *db)
346 {
347 struct db_ctdb_transaction_handle *h;
348 NTSTATUS status;
349 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
350 struct db_ctdb_ctx);
351
352 if (!db->persistent) {
353 DEBUG(0,("transactions not supported on non-persistent database 0x%08x\n",
354 ctx->db_id));
355 return -1;
356 }
357
358 if (ctx->transaction) {
359 ctx->transaction->nesting++;
360 DEBUG(5, (__location__ " transaction start on db 0x%08x: nesting %d -> %d\n",
361 ctx->db_id, ctx->transaction->nesting - 1, ctx->transaction->nesting));
362 return 0;
363 }
364
365 h = talloc_zero(db, struct db_ctdb_transaction_handle);
366 if (h == NULL) {
367 DEBUG(0,(__location__ " oom for transaction handle\n"));
368 return -1;
369 }
370
371 h->ctx = ctx;
372
373 h->lock_name = talloc_asprintf(h, "transaction_db_0x%08x",
374 (unsigned int)ctx->db_id);
375 if (h->lock_name == NULL) {
376 DEBUG(0, ("talloc_asprintf failed\n"));
377 TALLOC_FREE(h);
378 return -1;
379 }
380
381 /*
382 * Wait a day, i.e. forever...
383 */
384 status = g_lock_lock(ctx->lock_ctx,
385 string_term_tdb_data(h->lock_name),
386 G_LOCK_WRITE,
387 tevent_timeval_set(86400, 0),
388 NULL,
389 NULL);
390 if (!NT_STATUS_IS_OK(status)) {
391 DEBUG(0, ("g_lock_lock failed: %s\n", nt_errstr(status)));
392 TALLOC_FREE(h);
393 return -1;
394 }
395
396 talloc_set_destructor(h, db_ctdb_transaction_destructor);
397
398 ctx->transaction = h;
399
400 DEBUG(5,(__location__ " transaction started on db 0x%08x\n", ctx->db_id));
401
402 return 0;
403 }
404
405 static bool parse_newest_in_marshall_buffer(
406 struct ctdb_marshall_buffer *buf, TDB_DATA key,
407 void (*parser)(TDB_DATA key, struct ctdb_ltdb_header *header,
408 TDB_DATA data, void *private_data),
409 void *private_data)
410 {
411 struct ctdb_rec_data_old *rec = NULL;
412 struct ctdb_ltdb_header *h = NULL;
413 TDB_DATA data;
414 uint32_t i;
415
416 if (buf == NULL) {
417 return false;
418 }
419
420 /*
421 * Walk the list of records written during this
422 * transaction. If we want to read one we have already
423 * written, return the last written sample. Thus we do not do
424 * a "break;" for the first hit, this record might have been
425 * overwritten later.
426 */
427
428 for (i=0; i<buf->count; i++) {
429 TDB_DATA tkey;
430 uint32_t reqid;
431
432 rec = db_ctdb_marshall_loop_next_key(buf, rec, &tkey);
433 if (rec == NULL) {
434 return false;
435 }
436
437 if (!tdb_data_equal(key, tkey)) {
438 continue;
439 }
440
441 if (!db_ctdb_marshall_buf_parse(rec, &reqid, &h, &data)) {
442 return false;
443 }
444 }
445
446 if (h == NULL) {
447 return false;
448 }
449
450 parser(key, h, data, private_data);
451
452 return true;
453 }
454
455 struct pull_newest_from_marshall_buffer_state {
456 struct ctdb_ltdb_header *pheader;
457 TALLOC_CTX *mem_ctx;
458 TDB_DATA *pdata;
459 };
460
461 static void pull_newest_from_marshall_buffer_parser(
462 TDB_DATA key, struct ctdb_ltdb_header *header,
463 TDB_DATA data, void *private_data)
464 {
465 struct pull_newest_from_marshall_buffer_state *state =
466 (struct pull_newest_from_marshall_buffer_state *)private_data;
467
468 if (state->pheader != NULL) {
469 memcpy(state->pheader, header, sizeof(*state->pheader));
470 }
471 if (state->pdata != NULL) {
472 state->pdata->dsize = data.dsize;
473 state->pdata->dptr = (uint8_t *)talloc_memdup(
474 state->mem_ctx, data.dptr, data.dsize);
475 }
476 }
477
478 static bool pull_newest_from_marshall_buffer(struct ctdb_marshall_buffer *buf,
479 TDB_DATA key,
480 struct ctdb_ltdb_header *pheader,
481 TALLOC_CTX *mem_ctx,
482 TDB_DATA *pdata)
483 {
484 struct pull_newest_from_marshall_buffer_state state;
485
486 state.pheader = pheader;
487 state.mem_ctx = mem_ctx;
488 state.pdata = pdata;
489
490 if (!parse_newest_in_marshall_buffer(
491 buf, key, pull_newest_from_marshall_buffer_parser,
492 &state)) {
493 return false;
494 }
495 if ((pdata != NULL) && (pdata->dsize != 0) && (pdata->dptr == NULL)) {
496 /* ENOMEM */
497 return false;
498 }
499 return true;
500 }
501
502 static NTSTATUS db_ctdb_storev_transaction(struct db_record *rec,
503 const TDB_DATA *dbufs, int num_dbufs,
504 int flag);
505 static NTSTATUS db_ctdb_delete_transaction(struct db_record *rec);
506
507 static struct db_record *db_ctdb_fetch_locked_transaction(struct db_ctdb_ctx *ctx,
508 TALLOC_CTX *mem_ctx,
509 TDB_DATA key)
510 {
511 struct db_record *result;
512 TDB_DATA ctdb_data;
513
514 if (!(result = talloc(mem_ctx, struct db_record))) {
515 DEBUG(0, ("talloc failed\n"));
516 return NULL;
517 }
518
519 result->db = ctx->db;
520 result->private_data = ctx->transaction;
521
522 result->key.dsize = key.dsize;
523 result->key.dptr = (uint8_t *)talloc_memdup(result, key.dptr,
524 key.dsize);
525 if (result->key.dptr == NULL) {
526 DEBUG(0, ("talloc failed\n"));
527 TALLOC_FREE(result);
528 return NULL;
529 }
530
531 result->storev = db_ctdb_storev_transaction;
532 result->delete_rec = db_ctdb_delete_transaction;
533
534 if (ctx->transaction == NULL) {
535 DEBUG(0, ("no transaction available\n"));
536 TALLOC_FREE(result);
537 return NULL;
538 }
539 if (pull_newest_from_marshall_buffer(ctx->transaction->m_write, key,
540 NULL, result, &result->value)) {
541 result->value_valid = true;
542 return result;
543 }
544
545 ctdb_data = tdb_fetch(ctx->wtdb->tdb, key);
546 if (ctdb_data.dptr == NULL) {
547 /* create the record */
548 result->value = tdb_null;
549 result->value_valid = true;
550 return result;
551 }
552
553 result->value.dsize = ctdb_data.dsize - sizeof(struct ctdb_ltdb_header);
554 result->value.dptr = NULL;
555
556 if ((result->value.dsize != 0)
557 && !(result->value.dptr = (uint8_t *)talloc_memdup(
558 result, ctdb_data.dptr + sizeof(struct ctdb_ltdb_header),
559 result->value.dsize))) {
560 DEBUG(0, ("talloc failed\n"));
561 TALLOC_FREE(result);
562 return NULL;
563 }
564 result->value_valid = true;
565
566 SAFE_FREE(ctdb_data.dptr);
567
568 return result;
569 }
570
571 static int db_ctdb_record_destructor(struct db_record **recp)
572 {
573 struct db_record *rec = talloc_get_type_abort(*recp, struct db_record);
574 struct db_ctdb_transaction_handle *h = talloc_get_type_abort(
575 rec->private_data, struct db_ctdb_transaction_handle);
576 int ret = h->ctx->db->transaction_commit(h->ctx->db);
577 if (ret != 0) {
578 DEBUG(0,(__location__ " transaction_commit failed\n"));
579 }
580 return 0;
581 }
582
583 /*
584 auto-create a transaction for persistent databases
585 */
586 static struct db_record *db_ctdb_fetch_locked_persistent(struct db_ctdb_ctx *ctx,
587 TALLOC_CTX *mem_ctx,
588 TDB_DATA key)
589 {
590 int res;
591 struct db_record *rec, **recp;
592
593 res = db_ctdb_transaction_start(ctx->db);
594 if (res == -1) {
595 return NULL;
596 }
597
598 rec = db_ctdb_fetch_locked_transaction(ctx, mem_ctx, key);
599 if (rec == NULL) {
600 ctx->db->transaction_cancel(ctx->db);
601 return NULL;
602 }
603
604 /* destroy this transaction when we release the lock */
605 recp = talloc(rec, struct db_record *);
606 if (recp == NULL) {
607 ctx->db->transaction_cancel(ctx->db);
608 talloc_free(rec);
609 return NULL;
610 }
611 *recp = rec;
612 talloc_set_destructor(recp, db_ctdb_record_destructor);
613 return rec;
614 }
615
616
617 /*
618 stores a record inside a transaction
619 */
620 static NTSTATUS db_ctdb_transaction_store(struct db_ctdb_transaction_handle *h,
621 TDB_DATA key, TDB_DATA data)
622 {
623 TALLOC_CTX *tmp_ctx = talloc_new(h);
624 TDB_DATA rec;
625 struct ctdb_ltdb_header header;
626
627 ZERO_STRUCT(header);
628
629 /* we need the header so we can update the RSN */
630
631 if (!pull_newest_from_marshall_buffer(h->m_write, key, &header,
632 NULL, NULL)) {
633
634 rec = tdb_fetch(h->ctx->wtdb->tdb, key);
635
636 if (rec.dptr != NULL) {
637 memcpy(&header, rec.dptr,
638 sizeof(struct ctdb_ltdb_header));
639 rec.dsize -= sizeof(struct ctdb_ltdb_header);
640
641 /*
642 * a special case, we are writing the same
643 * data that is there now
644 */
645 if (data.dsize == rec.dsize &&
646 memcmp(data.dptr,
647 rec.dptr + sizeof(struct ctdb_ltdb_header),
648 data.dsize) == 0) {
649 SAFE_FREE(rec.dptr);
650 talloc_free(tmp_ctx);
651 return NT_STATUS_OK;
652 }
653 }
654 SAFE_FREE(rec.dptr);
655 }
656
657 header.dmaster = get_my_vnn();
658 header.rsn++;
659
660 h->m_write = db_ctdb_marshall_add(h, h->m_write, h->ctx->db_id, 0, key, &header, data);
661 if (h->m_write == NULL) {
662 DEBUG(0,(__location__ " Failed to add to marshalling record\n"));
663 talloc_free(tmp_ctx);
664 return NT_STATUS_NO_MEMORY;
665 }
666
667 talloc_free(tmp_ctx);
668 return NT_STATUS_OK;
669 }
670
671
672 /*
673 a record store inside a transaction
674 */
675 static NTSTATUS db_ctdb_storev_transaction(
676 struct db_record *rec, const TDB_DATA *dbufs, int num_dbufs, int flag)
677 {
678 struct db_ctdb_transaction_handle *h = talloc_get_type_abort(
679 rec->private_data, struct db_ctdb_transaction_handle);
680 NTSTATUS status;
681 TDB_DATA data = {0};
682
683 status = dbwrap_merge_dbufs(&data, rec, dbufs, num_dbufs);
684 if (!NT_STATUS_IS_OK(status)) {
685 return status;
686 }
687
688 status = db_ctdb_transaction_store(h, rec->key, data);
689
690 TALLOC_FREE(data.dptr);
691
692 return status;
693 }
694
695 /*
696 a record delete inside a transaction
697 */
698 static NTSTATUS db_ctdb_delete_transaction(struct db_record *rec)
699 {
700 struct db_ctdb_transaction_handle *h = talloc_get_type_abort(
701 rec->private_data, struct db_ctdb_transaction_handle);
702 NTSTATUS status;
703
704 status = db_ctdb_transaction_store(h, rec->key, tdb_null);
705 return status;
706 }
707
708 static void db_ctdb_fetch_db_seqnum_parser(
709 TDB_DATA key, struct ctdb_ltdb_header *header,
710 TDB_DATA data, void *private_data)
711 {
712 uint64_t *seqnum = (uint64_t *)private_data;
713
714 if (data.dsize != sizeof(uint64_t)) {
715 *seqnum = 0;
716 return;
717 }
718 memcpy(seqnum, data.dptr, sizeof(*seqnum));
719 }
720
721 /**
722 * Fetch the db sequence number of a persistent db directly from the db.
723 */
724 static NTSTATUS db_ctdb_fetch_db_seqnum_from_db(struct db_ctdb_ctx *db,
725 uint64_t *seqnum)
726 {
727 NTSTATUS status;
728 TDB_DATA key;
729
730 if (seqnum == NULL) {
731 return NT_STATUS_INVALID_PARAMETER;
732 }
733
734 key = string_term_tdb_data(CTDB_DB_SEQNUM_KEY);
735
736 status = db_ctdb_ltdb_parse(
737 db, key, db_ctdb_fetch_db_seqnum_parser, seqnum);
738
739 if (NT_STATUS_IS_OK(status)) {
740 return NT_STATUS_OK;
741 }
742 if (NT_STATUS_EQUAL(status, NT_STATUS_NOT_FOUND)) {
743 *seqnum = 0;
744 return NT_STATUS_OK;
745 }
746 return status;
747 }
748
749 /**
750 * Store the database sequence number inside a transaction.
751 */
752 static NTSTATUS db_ctdb_store_db_seqnum(struct db_ctdb_transaction_handle *h,
753 uint64_t seqnum)
754 {
755 NTSTATUS status;
756 TDB_DATA key = string_term_tdb_data(CTDB_DB_SEQNUM_KEY);
757 TDB_DATA data = { .dptr=(uint8_t *)&seqnum, .dsize=sizeof(seqnum) };
758
759 status = db_ctdb_transaction_store(h, key, data);
760
761 return status;
762 }
763
764 /*
765 commit a transaction
766 */
767 static int db_ctdb_transaction_commit(struct db_context *db)
768 {
769 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
770 struct db_ctdb_ctx);
771 NTSTATUS rets;
772 int32_t status;
773 struct db_ctdb_transaction_handle *h = ctx->transaction;
774 uint64_t old_seqnum, new_seqnum;
775 int ret;
776
777 if (h == NULL) {
778 DEBUG(0,(__location__ " transaction commit with no open transaction on db 0x%08x\n", ctx->db_id));
779 return -1;
780 }
781
782 if (h->nested_cancel) {
783 db->transaction_cancel(db);
784 DEBUG(5,(__location__ " Failed transaction commit after nested cancel\n"));
785 return -1;
786 }
787
788 if (h->nesting != 0) {
789 h->nesting--;
790 DEBUG(5, (__location__ " transaction commit on db 0x%08x: nesting %d -> %d\n",
791 ctx->db_id, ctx->transaction->nesting + 1, ctx->transaction->nesting));
792 return 0;
793 }
794
795 if (h->m_write == NULL) {
796 /*
797 * No changes were made, so don't change the seqnum,
798 * don't push to other node, just exit with success.
799 */
800 ret = 0;
801 goto done;
802 }
803
804 DEBUG(5,(__location__ " transaction commit on db 0x%08x\n", ctx->db_id));
805
806 /*
807 * As the last db action before committing, bump the database sequence
808 * number. Note that this undoes all changes to the seqnum records
809 * performed under the transaction. This record is not meant to be
810 * modified by user interaction. It is for internal use only...
811 */
812 rets = db_ctdb_fetch_db_seqnum_from_db(ctx, &old_seqnum);
813 if (!NT_STATUS_IS_OK(rets)) {
814 DEBUG(1, (__location__ " failed to fetch the db sequence number "
815 "in transaction commit on db 0x%08x\n", ctx->db_id));
816 ret = -1;
817 goto done;
818 }
819
820 new_seqnum = old_seqnum + 1;
821
822 rets = db_ctdb_store_db_seqnum(h, new_seqnum);
823 if (!NT_STATUS_IS_OK(rets)) {
824 DEBUG(1, (__location__ "failed to store the db sequence number "
825 " in transaction commit on db 0x%08x\n", ctx->db_id));
826 ret = -1;
827 goto done;
828 }
829
830 again:
831 /* tell ctdbd to commit to the other nodes */
832 ret = ctdbd_control_local(messaging_ctdb_connection(),
833 CTDB_CONTROL_TRANS3_COMMIT,
834 h->ctx->db_id, 0,
835 db_ctdb_marshall_finish(h->m_write),
836 NULL, NULL, &status);
837 if ((ret != 0) || status != 0) {
838 /*
839 * The TRANS3_COMMIT control should only possibly fail when a
840 * recovery has been running concurrently. In any case, the db
841 * will be the same on all nodes, either the new copy or the
842 * old copy. This can be detected by comparing the old and new
843 * local sequence numbers.
844 */
845 rets = db_ctdb_fetch_db_seqnum_from_db(ctx, &new_seqnum);
846 if (!NT_STATUS_IS_OK(rets)) {
847 DEBUG(1, (__location__ " failed to refetch db sequence "
848 "number after failed TRANS3_COMMIT\n"));
849 ret = -1;
850 goto done;
851 }
852
853 if (new_seqnum == old_seqnum) {
854 /* Recovery prevented all our changes: retry. */
855 goto again;
856 }
857 if (new_seqnum != (old_seqnum + 1)) {
858 DEBUG(0, (__location__ " ERROR: new_seqnum[%lu] != "
859 "old_seqnum[%lu] + (0 or 1) after failed "
860 "TRANS3_COMMIT - this should not happen!\n",
861 (unsigned long)new_seqnum,
862 (unsigned long)old_seqnum));
863 ret = -1;
864 goto done;
865 }
866 /*
867 * Recovery propagated our changes to all nodes, completing
868 * our commit for us - succeed.
869 */
870 }
871
872 ret = 0;
873
874 done:
875 h->ctx->transaction = NULL;
876 talloc_free(h);
877 return ret;
878 }
879
880
881 /*
882 cancel a transaction
883 */
884 static int db_ctdb_transaction_cancel(struct db_context *db)
885 {
886 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
887 struct db_ctdb_ctx);
888 struct db_ctdb_transaction_handle *h = ctx->transaction;
889
890 if (h == NULL) {
891 DEBUG(0,(__location__ " transaction cancel with no open transaction on db 0x%08x\n", ctx->db_id));
892 return -1;
893 }
894
895 if (h->nesting != 0) {
896 h->nesting--;
897 h->nested_cancel = true;
898 DEBUG(5, (__location__ " transaction cancel on db 0x%08x: nesting %d -> %d\n",
899 ctx->db_id, ctx->transaction->nesting + 1, ctx->transaction->nesting));
900 return 0;
901 }
902
903 DEBUG(5,(__location__ " Cancel transaction on db 0x%08x\n", ctx->db_id));
904
905 ctx->transaction = NULL;
906 talloc_free(h);
907 return 0;
908 }
909
910
911 static NTSTATUS db_ctdb_storev(struct db_record *rec,
912 const TDB_DATA *dbufs, int num_dbufs, int flag)
913 {
914 struct db_ctdb_rec *crec = talloc_get_type_abort(
915 rec->private_data, struct db_ctdb_rec);
916 NTSTATUS status;
917
918 status = db_ctdb_ltdb_store(crec->ctdb_ctx, rec->key, &(crec->header),
919 dbufs, num_dbufs);
920 return status;
921 }
922
923
924
925 static NTSTATUS db_ctdb_send_schedule_for_deletion(struct db_record *rec)
926 {
927 NTSTATUS status = NT_STATUS_OK;
928 int ret;
929 struct ctdb_control_schedule_for_deletion *dd;
930 TDB_DATA indata;
931 int32_t cstatus;
932 struct db_ctdb_rec *crec = talloc_get_type_abort(
933 rec->private_data, struct db_ctdb_rec);
934 struct db_ctdb_ctx *ctx = crec->ctdb_ctx;
935
936 indata.dsize = offsetof(struct ctdb_control_schedule_for_deletion, key) + rec->key.dsize;
937 indata.dptr = talloc_zero_array(crec, uint8_t, indata.dsize);
938 if (indata.dptr == NULL) {
939 DEBUG(0, (__location__ " talloc failed!\n"));
940 return NT_STATUS_NO_MEMORY;
941 }
942
943 dd = (struct ctdb_control_schedule_for_deletion *)(void *)indata.dptr;
944 dd->db_id = ctx->db_id;
945 dd->hdr = crec->header;
946 dd->keylen = rec->key.dsize;
947 memcpy(dd->key, rec->key.dptr, rec->key.dsize);
948
949 ret = ctdbd_control_local(messaging_ctdb_connection(),
950 CTDB_CONTROL_SCHEDULE_FOR_DELETION,
951 crec->ctdb_ctx->db_id,
952 CTDB_CTRL_FLAG_NOREPLY, /* flags */
953 indata,
954 NULL, /* mem_ctx */
955 NULL, /* outdata */
956 &cstatus);
957 talloc_free(indata.dptr);
958
959 if ((ret != 0) || cstatus != 0) {
960 DEBUG(1, (__location__ " Error sending local control "
961 "SCHEDULE_FOR_DELETION: %s, cstatus = %"PRIi32"\n",
962 strerror(ret), cstatus));
963 if (ret != 0) {
964 status = map_nt_error_from_unix(ret);
965 } else {
966 status = NT_STATUS_UNSUCCESSFUL;
967 }
968 }
969
970 return status;
971 }
972
973 static NTSTATUS db_ctdb_delete(struct db_record *rec)
974 {
975 NTSTATUS status;
976
977 /*
978 * We have to store the header with empty data. TODO: Fix the
979 * tdb-level cleanup
980 */
981
982 status = db_ctdb_storev(rec, &tdb_null, 1, 0);
983 if (!NT_STATUS_IS_OK(status)) {
984 return status;
985 }
986
987 status = db_ctdb_send_schedule_for_deletion(rec);
988 return status;
989 }
990
991 static int db_ctdb_record_destr(struct db_record* data)
992 {
993 struct db_ctdb_rec *crec = talloc_get_type_abort(
994 data->private_data, struct db_ctdb_rec);
995 int threshold;
996 int ret;
997 struct timeval before;
998 double timediff;
999
1000 DEBUG(10, (DEBUGLEVEL > 10
1001 ? "Unlocking db %u key %s\n"
1002 : "Unlocking db %u key %.20s\n",
1003 (int)crec->ctdb_ctx->db_id,
1004 hex_encode_talloc(data, (unsigned char *)data->key.dptr,
1005 data->key.dsize)));
1006
1007 before = timeval_current();
1008
1009 ret = tdb_chainunlock(crec->ctdb_ctx->wtdb->tdb, data->key);
1010
1011 timediff = timeval_elapsed(&before);
1012 timediff *= 1000; /* get us milliseconds */
1013
1014 if (timediff > crec->ctdb_ctx->warn_unlock_msecs) {
1015 char *key;
1016 key = hex_encode_talloc(talloc_tos(),
1017 (unsigned char *)data->key.dptr,
1018 data->key.dsize);
1019 DEBUG(0, ("tdb_chainunlock on db %s, key %s took %f milliseconds\n",
1020 tdb_name(crec->ctdb_ctx->wtdb->tdb), key,
1021 timediff));
1022 TALLOC_FREE(key);
1023 }
1024
1025 if (ret != 0) {
1026 DEBUG(0, ("tdb_chainunlock failed\n"));
1027 return -1;
1028 }
1029
1030 threshold = crec->ctdb_ctx->warn_locktime_msecs;
1031 if (threshold != 0) {
1032 timediff = timeval_elapsed(&crec->lock_time) * 1000;
1033 if (timediff > threshold) {
1034 const char *key;
1035
1036 key = hex_encode_talloc(data,
1037 (unsigned char *)data->key.dptr,
1038 data->key.dsize);
1039 DEBUG(0, ("Held tdb lock on db %s, key %s "
1040 "%f milliseconds\n",
1041 tdb_name(crec->ctdb_ctx->wtdb->tdb),
1042 key, timediff));
1043 }
1044 }
1045
1046 return 0;
1047 }
1048
1049 /**
1050 * Check whether we have a valid local copy of the given record,
1051 * either for reading or for writing.
1052 */
1053 static bool db_ctdb_can_use_local_hdr(const struct ctdb_ltdb_header *hdr,
1054 uint32_t my_vnn, bool read_only)
1055 {
1056 if (hdr->dmaster != my_vnn) {
1057 /* If we're not dmaster, it must be r/o copy. */
1058 return read_only && (hdr->flags & CTDB_REC_RO_HAVE_READONLY);
1059 }
1060
1061 /*
1062 * If we want write access, no one may have r/o copies.
1063 */
1064 return read_only || !(hdr->flags & CTDB_REC_RO_HAVE_DELEGATIONS);
1065 }
1066
1067 static bool db_ctdb_can_use_local_copy(TDB_DATA ctdb_data, uint32_t my_vnn,
1068 bool read_only)
1069 {
1070 if (ctdb_data.dptr == NULL) {
1071 return false;
1072 }
1073
1074 if (ctdb_data.dsize < sizeof(struct ctdb_ltdb_header)) {
1075 return false;
1076 }
1077
1078 return db_ctdb_can_use_local_hdr(
1079 (struct ctdb_ltdb_header *)ctdb_data.dptr, my_vnn, read_only);
1080 }
1081
1082 static struct db_record *fetch_locked_internal(struct db_ctdb_ctx *ctx,
1083 TALLOC_CTX *mem_ctx,
1084 TDB_DATA key)
1085 {
1086 struct db_record *result;
1087 struct db_ctdb_rec *crec;
1088 TDB_DATA ctdb_data;
1089 int migrate_attempts;
1090 struct timeval migrate_start;
1091 struct timeval chainlock_start;
1092 struct timeval ctdb_start_time;
1093 double chainlock_time = 0;
1094 double ctdb_time = 0;
1095 int duration_msecs;
1096 int lockret;
1097 int ret;
1098
1099 if (!(result = talloc(mem_ctx, struct db_record))) {
1100 DEBUG(0, ("talloc failed\n"));
1101 return NULL;
1102 }
1103
1104 if (!(crec = talloc_zero(result, struct db_ctdb_rec))) {
1105 DEBUG(0, ("talloc failed\n"));
1106 TALLOC_FREE(result);
1107 return NULL;
1108 }
1109
1110 result->db = ctx->db;
1111 result->private_data = (void *)crec;
1112 crec->ctdb_ctx = ctx;
1113
1114 result->key.dsize = key.dsize;
1115 result->key.dptr = (uint8_t *)talloc_memdup(result, key.dptr,
1116 key.dsize);
1117 if (result->key.dptr == NULL) {
1118 DEBUG(0, ("talloc failed\n"));
1119 TALLOC_FREE(result);
1120 return NULL;
1121 }
1122
1123 migrate_attempts = 0;
1124 GetTimeOfDay(&migrate_start);
1125
1126 /*
1127 * Do a blocking lock on the record
1128 */
1129 again:
1130
1131 if (DEBUGLEVEL >= 10) {
1132 char *keystr = hex_encode_talloc(result, key.dptr, key.dsize);
1133 DEBUG(10, (DEBUGLEVEL > 10
1134 ? "Locking db %u key %s\n"
1135 : "Locking db %u key %.20s\n",
1136 (int)crec->ctdb_ctx->db_id, keystr));
1137 TALLOC_FREE(keystr);
1138 }
1139
1140 GetTimeOfDay(&chainlock_start);
1141 lockret = tdb_chainlock(ctx->wtdb->tdb, key);
1142 chainlock_time += timeval_elapsed(&chainlock_start);
1143
1144 if (lockret != 0) {
1145 DEBUG(3, ("tdb_chainlock failed\n"));
1146 TALLOC_FREE(result);
1147 return NULL;
1148 }
1149
1150 result->storev = db_ctdb_storev;
1151 result->delete_rec = db_ctdb_delete;
1152 talloc_set_destructor(result, db_ctdb_record_destr);
1153
1154 ctdb_data = tdb_fetch(ctx->wtdb->tdb, key);
1155
1156 /*
1157 * See if we have a valid record and we are the dmaster. If so, we can
1158 * take the shortcut and just return it.
1159 */
1160
1161 if (!db_ctdb_can_use_local_copy(ctdb_data, get_my_vnn(), false)) {
1162 SAFE_FREE(ctdb_data.dptr);
1163 tdb_chainunlock(ctx->wtdb->tdb, key);
1164 talloc_set_destructor(result, NULL);
1165
1166 migrate_attempts += 1;
1167
1168 DEBUG(10, ("ctdb_data.dptr = %p, dmaster = %"PRIu32" "
1169 "(%"PRIu32") %"PRIu32"\n",
1170 ctdb_data.dptr, ctdb_data.dptr ?
1171 ((struct ctdb_ltdb_header *)ctdb_data.dptr)->dmaster :
1172 UINT32_MAX,
1173 get_my_vnn(),
1174 ctdb_data.dptr ?
1175 ((struct ctdb_ltdb_header *)ctdb_data.dptr)->flags : 0));
1176
1177 GetTimeOfDay(&ctdb_start_time);
1178 ret = ctdbd_migrate(messaging_ctdb_connection(), ctx->db_id,
1179 key);
1180 ctdb_time += timeval_elapsed(&ctdb_start_time);
1181
1182 if (ret != 0) {
1183 DEBUG(5, ("ctdbd_migrate failed: %s\n",
1184 strerror(ret)));
1185 TALLOC_FREE(result);
1186 return NULL;
1187 }
1188 /* now its migrated, try again */
1189 goto again;
1190 }
1191
1192 {
1193 double duration;
1194 duration = timeval_elapsed(&migrate_start);
1195
1196 /*
1197 * Convert the duration to milliseconds to avoid a
1198 * floating-point division of
1199 * lp_parm_int("migrate_duration") by 1000.
1200 */
1201 duration_msecs = duration * 1000;
1202 }
1203
1204 if ((migrate_attempts > ctx->warn_migrate_attempts) ||
1205 (duration_msecs > ctx->warn_migrate_msecs)) {
1206 int chain = 0;
1207
1208 if (tdb_get_flags(ctx->wtdb->tdb) & TDB_INCOMPATIBLE_HASH) {
1209 chain = tdb_jenkins_hash(&key) %
1210 tdb_hash_size(ctx->wtdb->tdb);
1211 }
1212
1213 DEBUG(0, ("db_ctdb_fetch_locked for %s key %s, chain %d "
1214 "needed %d attempts, %d milliseconds, "
1215 "chainlock: %f ms, CTDB %f ms\n",
1216 tdb_name(ctx->wtdb->tdb),
1217 hex_encode_talloc(talloc_tos(),
1218 (unsigned char *)key.dptr,
1219 key.dsize),
1220 chain,
1221 migrate_attempts, duration_msecs,
1222 chainlock_time * 1000.0,
1223 ctdb_time * 1000.0));
1224 }
1225
1226 GetTimeOfDay(&crec->lock_time);
1227
1228 memcpy(&crec->header, ctdb_data.dptr, sizeof(crec->header));
1229
1230 result->value.dsize = ctdb_data.dsize - sizeof(crec->header);
1231 result->value.dptr = NULL;
1232
1233 if (result->value.dsize != 0) {
1234 result->value.dptr = talloc_memdup(
1235 result, ctdb_data.dptr + sizeof(crec->header),
1236 result->value.dsize);
1237 if (result->value.dptr == NULL) {
1238 DBG_ERR("talloc failed\n");
1239 TALLOC_FREE(result);
1240 return NULL;
1241 }
1242 }
1243 result->value_valid = true;
1244
1245 SAFE_FREE(ctdb_data.dptr);
1246
1247 return result;
1248 }
1249
1250 static struct db_record *db_ctdb_fetch_locked(struct db_context *db,
1251 TALLOC_CTX *mem_ctx,
1252 TDB_DATA key)
1253 {
1254 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1255 struct db_ctdb_ctx);
1256
1257 if (ctx->transaction != NULL) {
1258 return db_ctdb_fetch_locked_transaction(ctx, mem_ctx, key);
1259 }
1260
1261 if (db->persistent) {
1262 return db_ctdb_fetch_locked_persistent(ctx, mem_ctx, key);
1263 }
1264
1265 return fetch_locked_internal(ctx, mem_ctx, key);
1266 }
1267
1268 struct db_ctdb_parse_record_state {
1269 void (*parser)(TDB_DATA key, TDB_DATA data, void *private_data);
1270 void *private_data;
1271 uint32_t my_vnn;
1272 bool ask_for_readonly_copy;
1273 bool done;
1274 bool empty_record;
1275 };
1276
1277 static void db_ctdb_parse_record_parser(
1278 TDB_DATA key, struct ctdb_ltdb_header *header,
1279 TDB_DATA data, void *private_data)
1280 {
1281 struct db_ctdb_parse_record_state *state =
1282 (struct db_ctdb_parse_record_state *)private_data;
1283 state->parser(key, data, state->private_data);
1284 }
1285
1286 static void db_ctdb_parse_record_parser_nonpersistent(
1287 TDB_DATA key, struct ctdb_ltdb_header *header,
1288 TDB_DATA data, void *private_data)
1289 {
1290 struct db_ctdb_parse_record_state *state =
1291 (struct db_ctdb_parse_record_state *)private_data;
1292
1293 if (db_ctdb_can_use_local_hdr(header, state->my_vnn, true)) {
1294 /*
1295 * A record consisting only of the ctdb header can be
1296 * a validly created empty record or a tombstone
1297 * record of a deleted record (not vacuumed yet). Mark
1298 * it accordingly.
1299 */
1300 state->empty_record = (data.dsize == 0);
1301 if (!state->empty_record) {
1302 state->parser(key, data, state->private_data);
1303 }
1304 state->done = true;
1305 } else {
1306 /*
1307 * We found something in the db, so it seems that this record,
1308 * while not usable locally right now, is popular. Ask for a
1309 * R/O copy.
1310 */
1311 state->ask_for_readonly_copy = true;
1312 }
1313 }
1314
1315 static NTSTATUS db_ctdb_try_parse_local_record(struct db_ctdb_ctx *ctx,
1316 TDB_DATA key,
1317 struct db_ctdb_parse_record_state *state)
1318 {
1319 NTSTATUS status;
1320
1321 if (ctx->transaction != NULL) {
1322 struct db_ctdb_transaction_handle *h = ctx->transaction;
1323 bool found;
1324
1325 /*
1326 * Transactions only happen for persistent db's.
1327 */
1328
1329 found = parse_newest_in_marshall_buffer(
1330 h->m_write, key, db_ctdb_parse_record_parser, state);
1331
1332 if (found) {
1333 return NT_STATUS_OK;
1334 }
1335 }
1336
1337 if (ctx->db->persistent) {
1338 /*
1339 * Persistent db, but not found in the transaction buffer
1340 */
1341 return db_ctdb_ltdb_parse(
1342 ctx, key, db_ctdb_parse_record_parser, state);
1343 }
1344
1345 state->done = false;
1346 state->ask_for_readonly_copy = false;
1347
1348 status = db_ctdb_ltdb_parse(
1349 ctx, key, db_ctdb_parse_record_parser_nonpersistent, state);
1350 if (NT_STATUS_IS_OK(status) && state->done) {
1351 if (state->empty_record) {
1352 /*
1353 * We know authoritatively, that this is an empty
1354 * record. Since ctdb does not distinguish between empty
1355 * and deleted records, this can be a record stored as
1356 * empty or a not-yet-vacuumed tombstone record of a
1357 * deleted record. Now Samba right now can live without
1358 * empty records, so we can safely report this record
1359 * as non-existing.
1360 *
1361 * See bugs 10008 and 12005.
1362 */
1363 return NT_STATUS_NOT_FOUND;
1364 }
1365 return NT_STATUS_OK;
1366 }
1367
1368 return NT_STATUS_MORE_PROCESSING_REQUIRED;
1369 }
1370
1371 static NTSTATUS db_ctdb_parse_record(struct db_context *db, TDB_DATA key,
1372 void (*parser)(TDB_DATA key,
1373 TDB_DATA data,
1374 void *private_data),
1375 void *private_data)
1376 {
1377 struct db_ctdb_ctx *ctx = talloc_get_type_abort(
1378 db->private_data, struct db_ctdb_ctx);
1379 struct db_ctdb_parse_record_state state;
1380 NTSTATUS status;
1381 int ret;
1382
1383 state.parser = parser;
1384 state.private_data = private_data;
1385 state.my_vnn = get_my_vnn();
1386 state.empty_record = false;
1387
1388 status = db_ctdb_try_parse_local_record(ctx, key, &state);
1389 if (!NT_STATUS_EQUAL(status, NT_STATUS_MORE_PROCESSING_REQUIRED)) {
1390 return status;
1391 }
1392
1393 ret = ctdbd_parse(messaging_ctdb_connection(), ctx->db_id, key,
1394 state.ask_for_readonly_copy, parser, private_data);
1395 if (ret != 0) {
1396 if (ret == ENOENT) {
1397 /*
1398 * This maps to
1399 * NT_STATUS_OBJECT_NAME_NOT_FOUND. Our upper
1400 * layers expect NT_STATUS_NOT_FOUND for "no
1401 * record around". We need to convert dbwrap
1402 * to 0/errno away from NTSTATUS ... :-)
1403 */
1404 return NT_STATUS_NOT_FOUND;
1405 }
1406 return map_nt_error_from_unix(ret);
1407 }
1408 return NT_STATUS_OK;
1409 }
1410
1411 static void db_ctdb_parse_record_done(struct tevent_req *subreq);
1412
1413 static struct tevent_req *db_ctdb_parse_record_send(
1414 TALLOC_CTX *mem_ctx,
1415 struct tevent_context *ev,
1416 struct db_context *db,
1417 TDB_DATA key,
1418 void (*parser)(TDB_DATA key,
1419 TDB_DATA data,
1420 void *private_data),
1421 void *private_data,
1422 enum dbwrap_req_state *req_state)
1423 {
1424 struct db_ctdb_ctx *ctx = talloc_get_type_abort(
1425 db->private_data, struct db_ctdb_ctx);
1426 struct tevent_req *req = NULL;
1427 struct tevent_req *subreq = NULL;
1428 struct db_ctdb_parse_record_state *state = NULL;
1429 NTSTATUS status;
1430
1431 req = tevent_req_create(mem_ctx, &state,
1432 struct db_ctdb_parse_record_state);
1433 if (req == NULL) {
1434 *req_state = DBWRAP_REQ_ERROR;
1435 return NULL;
1436
1437 }
1438
1439 *state = (struct db_ctdb_parse_record_state) {
1440 .parser = parser,
1441 .private_data = private_data,
1442 .my_vnn = get_my_vnn(),
1443 .empty_record = false,
1444 };
1445
1446 status = db_ctdb_try_parse_local_record(ctx, key, state);
1447 if (!NT_STATUS_EQUAL(status, NT_STATUS_MORE_PROCESSING_REQUIRED)) {
1448 if (tevent_req_nterror(req, status)) {
1449 *req_state = DBWRAP_REQ_ERROR;
1450 return tevent_req_post(req, ev);
1451 }
1452 *req_state = DBWRAP_REQ_DONE;
1453 tevent_req_done(req);
1454 return tevent_req_post(req, ev);
1455 }
1456
1457 subreq = ctdbd_parse_send(state,
1458 ev,
1459 ctdb_async_ctx.async_conn,
1460 ctx->db_id,
1461 key,
1462 state->ask_for_readonly_copy,
1463 parser,
1464 private_data,
1465 req_state);
1466 if (tevent_req_nomem(subreq, req)) {
1467 *req_state = DBWRAP_REQ_ERROR;
1468 return tevent_req_post(req, ev);
1469 }
1470 tevent_req_set_callback(subreq, db_ctdb_parse_record_done, req);
1471
1472 return req;
1473 }
1474
1475 static void db_ctdb_parse_record_done(struct tevent_req *subreq)
1476 {
1477 struct tevent_req *req = tevent_req_callback_data(
1478 subreq, struct tevent_req);
1479 int ret;
1480
1481 ret = ctdbd_parse_recv(subreq);
1482 TALLOC_FREE(subreq);
1483 if (ret != 0) {
1484 if (ret == ENOENT) {
1485 /*
1486 * This maps to NT_STATUS_OBJECT_NAME_NOT_FOUND. Our
1487 * upper layers expect NT_STATUS_NOT_FOUND for "no
1488 * record around". We need to convert dbwrap to 0/errno
1489 * away from NTSTATUS ... :-)
1490 */
1491 tevent_req_nterror(req, NT_STATUS_NOT_FOUND);
1492 return;
1493 }
1494 tevent_req_nterror(req, map_nt_error_from_unix(ret));
1495 return;
1496 }
1497
1498 tevent_req_done(req);
1499 return;
1500 }
1501
1502 static NTSTATUS db_ctdb_parse_record_recv(struct tevent_req *req)
1503 {
1504 return tevent_req_simple_recv_ntstatus(req);
1505 }
1506
1507 struct traverse_state {
1508 struct db_context *db;
1509 int (*fn)(struct db_record *rec, void *private_data);
1510 void *private_data;
1511 int count;
1512 };
1513
1514 static void traverse_callback(TDB_DATA key, TDB_DATA data, void *private_data)
1515 {
1516 struct traverse_state *state = (struct traverse_state *)private_data;
1517 struct db_record *rec = NULL;
1518 TALLOC_CTX *tmp_ctx = NULL;
1519
1520 tmp_ctx = talloc_new(state->db);
1521 if (tmp_ctx == NULL) {
1522 DBG_ERR("talloc_new failed\n");
1523 return;
1524 }
1525
1526 /* we have to give them a locked record to prevent races */
1527 rec = db_ctdb_fetch_locked(state->db, tmp_ctx, key);
1528 if (rec != NULL && rec->value.dsize > 0) {
1529 state->fn(rec, state->private_data);
1530 state->count++;
1531 }
1532 talloc_free(tmp_ctx);
1533 }
1534
1535 static int traverse_persistent_callback(TDB_CONTEXT *tdb, TDB_DATA kbuf, TDB_DATA dbuf,
1536 void *private_data)
1537 {
1538 struct traverse_state *state = (struct traverse_state *)private_data;
1539 struct db_record *rec;
1540 TALLOC_CTX *tmp_ctx = talloc_new(state->db);
1541 int ret = 0;
1542
1543 /*
1544 * Skip the __db_sequence_number__ key:
1545 * This is used for persistent transactions internally.
1546 */
1547 if (kbuf.dsize == strlen(CTDB_DB_SEQNUM_KEY) + 1 &&
1548 strcmp((const char*)kbuf.dptr, CTDB_DB_SEQNUM_KEY) == 0)
1549 {
1550 goto done;
1551 }
1552
1553 /* we have to give them a locked record to prevent races */
1554 rec = db_ctdb_fetch_locked(state->db, tmp_ctx, kbuf);
1555 if (rec && rec->value.dsize > 0) {
1556 ret = state->fn(rec, state->private_data);
1557 }
1558
1559 done:
1560 talloc_free(tmp_ctx);
1561 return ret;
1562 }
1563
1564 /* wrapper to use traverse_persistent_callback with dbwrap */
1565 static int traverse_persistent_callback_dbwrap(struct db_record *rec, void* data)
1566 {
1567 return traverse_persistent_callback(NULL, rec->key, rec->value, data);
1568 }
1569
1570 static int db_ctdbd_traverse(uint32_t db_id,
1571 void (*fn)(TDB_DATA key, TDB_DATA data,
1572 void *private_data),
1573 void *private_data)
1574 {
1575 struct ctdbd_connection *conn;
1576 int ret;
1577
1578 become_root();
1579 ret = ctdbd_init_connection(talloc_tos(), lp_ctdbd_socket(),
1580 lp_ctdb_timeout(), &conn);
1581 unbecome_root();
1582 if (ret != 0) {
1583 DBG_WARNING("ctdbd_init_connection failed: %s\n",
1584 strerror(ret));
1585 return ret;
1586 }
1587
1588 ret = ctdbd_traverse(conn, db_id, fn, private_data);
1589 TALLOC_FREE(conn);
1590
1591 if (ret != 0) {
1592 DBG_WARNING("ctdbd_traverse failed: %s\n",
1593 strerror(ret));
1594 return ret;
1595 }
1596
1597 return 0;
1598 }
1599
1600
1601 static int db_ctdb_traverse(struct db_context *db,
1602 int (*fn)(struct db_record *rec,
1603 void *private_data),
1604 void *private_data)
1605 {
1606 int ret;
1607 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1608 struct db_ctdb_ctx);
1609 struct traverse_state state;
1610
1611 state = (struct traverse_state) {
1612 .db = db,
1613 .fn = fn,
1614 .private_data = private_data,
1615 };
1616
1617 if (db->persistent) {
1618 struct tdb_context *ltdb = ctx->wtdb->tdb;
1619
1620 /* for persistent databases we don't need to do a ctdb traverse,
1621 we can do a faster local traverse */
1622 ret = tdb_traverse(ltdb, traverse_persistent_callback, &state);
1623 if (ret < 0) {
1624 return ret;
1625 }
1626 if (ctx->transaction && ctx->transaction->m_write) {
1627 /*
1628 * we now have to handle keys not yet
1629 * present at transaction start
1630 */
1631 struct db_context *newkeys = db_open_rbt(talloc_tos());
1632 struct ctdb_marshall_buffer *mbuf = ctx->transaction->m_write;
1633 struct ctdb_rec_data_old *rec=NULL;
1634 uint32_t i;
1635 int count = 0;
1636 NTSTATUS status;
1637
1638 if (newkeys == NULL) {
1639 return -1;
1640 }
1641
1642 for (i=0; i<mbuf->count; i++) {
1643 TDB_DATA key;
1644 rec = db_ctdb_marshall_loop_next_key(
1645 mbuf, rec, &key);
1646 SMB_ASSERT(rec != NULL);
1647
1648 if (!tdb_exists(ltdb, key)) {
1649 dbwrap_store(newkeys, key, tdb_null, 0);
1650 }
1651 }
1652 status = dbwrap_traverse(newkeys,
1653 traverse_persistent_callback_dbwrap,
1654 &state,
1655 &count);
1656 talloc_free(newkeys);
1657 if (!NT_STATUS_IS_OK(status)) {
1658 return -1;
1659 }
1660 ret += count;
1661 }
1662 return ret;
1663 }
1664
1665 ret = db_ctdbd_traverse(ctx->db_id, traverse_callback, &state);
1666 if (ret != 0) {
1667 return -1;
1668 }
1669 return state.count;
1670 }
1671
1672 static NTSTATUS db_ctdb_storev_deny(struct db_record *rec,
1673 const TDB_DATA *dbufs, int num_dbufs, int flag)
1674 {
1675 return NT_STATUS_MEDIA_WRITE_PROTECTED;
1676 }
1677
1678 static NTSTATUS db_ctdb_delete_deny(struct db_record *rec)
1679 {
1680 return NT_STATUS_MEDIA_WRITE_PROTECTED;
1681 }
1682
1683 static void traverse_read_callback(TDB_DATA key, TDB_DATA data, void *private_data)
1684 {
1685 struct traverse_state *state = (struct traverse_state *)private_data;
1686 struct db_record rec;
1687
1688 ZERO_STRUCT(rec);
1689 rec.db = state->db;
1690 rec.key = key;
1691 rec.value = data;
1692 rec.storev = db_ctdb_storev_deny;
1693 rec.delete_rec = db_ctdb_delete_deny;
1694 rec.private_data = NULL;
1695 rec.value_valid = true;
1696 state->fn(&rec, state->private_data);
1697 state->count++;
1698 }
1699
1700 static int traverse_persistent_callback_read(TDB_CONTEXT *tdb, TDB_DATA kbuf, TDB_DATA dbuf,
1701 void *private_data)
1702 {
1703 struct traverse_state *state = (struct traverse_state *)private_data;
1704 struct db_record rec;
1705
1706 /*
1707 * Skip the __db_sequence_number__ key:
1708 * This is used for persistent transactions internally.
1709 */
1710 if (kbuf.dsize == strlen(CTDB_DB_SEQNUM_KEY) + 1 &&
1711 strcmp((const char*)kbuf.dptr, CTDB_DB_SEQNUM_KEY) == 0)
1712 {
1713 return 0;
1714 }
1715
1716 ZERO_STRUCT(rec);
1717 rec.db = state->db;
1718 rec.key = kbuf;
1719 rec.value = dbuf;
1720 rec.value_valid = true;
1721 rec.storev = db_ctdb_storev_deny;
1722 rec.delete_rec = db_ctdb_delete_deny;
1723 rec.private_data = NULL;
1724
1725 if (rec.value.dsize <= sizeof(struct ctdb_ltdb_header)) {
1726 /* a deleted record */
1727 return 0;
1728 }
1729 rec.value.dsize -= sizeof(struct ctdb_ltdb_header);
1730 rec.value.dptr += sizeof(struct ctdb_ltdb_header);
1731
1732 state->count++;
1733 return state->fn(&rec, state->private_data);
1734 }
1735
1736 static int db_ctdb_traverse_read(struct db_context *db,
1737 int (*fn)(struct db_record *rec,
1738 void *private_data),
1739 void *private_data)
1740 {
1741 int ret;
1742 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1743 struct db_ctdb_ctx);
1744 struct traverse_state state;
1745
1746 state = (struct traverse_state) {
1747 .db = db,
1748 .fn = fn,
1749 .private_data = private_data,
1750 };
1751
1752 if (db->persistent) {
1753 /* for persistent databases we don't need to do a ctdb traverse,
1754 we can do a faster local traverse */
1755 int nrecs;
1756
1757 nrecs = tdb_traverse_read(ctx->wtdb->tdb,
1758 traverse_persistent_callback_read,
1759 &state);
1760 if (nrecs == -1) {
1761 return -1;
1762 }
1763 return state.count;
1764 }
1765
1766 ret = db_ctdbd_traverse(ctx->db_id, traverse_read_callback, &state);
1767 if (ret != 0) {
1768 return -1;
1769 }
1770 return state.count;
1771 }
1772
1773 static int db_ctdb_get_seqnum(struct db_context *db)
1774 {
1775 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1776 struct db_ctdb_ctx);
1777 return tdb_get_seqnum(ctx->wtdb->tdb);
1778 }
1779
1780 static size_t db_ctdb_id(struct db_context *db, uint8_t *id, size_t idlen)
1781 {
1782 struct db_ctdb_ctx *ctx = talloc_get_type_abort(
1783 db->private_data, struct db_ctdb_ctx);
1784
1785 if (idlen >= sizeof(ctx->db_id)) {
1786 memcpy(id, &ctx->db_id, sizeof(ctx->db_id));
1787 }
1788
1789 return sizeof(ctx->db_id);
1790 }
1791
1792 struct db_context *db_open_ctdb(TALLOC_CTX *mem_ctx,
1793 struct messaging_context *msg_ctx,
1794 const char *name,
1795 int hash_size, int tdb_flags,
1796 int open_flags, mode_t mode,
1797 enum dbwrap_lock_order lock_order,
1798 uint64_t dbwrap_flags)
1799 {
1800 struct db_context *result;
1801 struct db_ctdb_ctx *db_ctdb;
1802 char *db_path;
1803 struct loadparm_context *lp_ctx;
1804 TDB_DATA data;
1805 TDB_DATA outdata = {0};
1806 bool persistent = (tdb_flags & TDB_CLEAR_IF_FIRST) == 0;
1807 int32_t cstatus;
1808 int ret;
1809
1810 if (!lp_clustering()) {
1811 DEBUG(10, ("Clustering disabled -- no ctdb\n"));
1812 return NULL;
1813 }
1814
1815 if (!(result = talloc_zero(mem_ctx, struct db_context))) {
1816 DEBUG(0, ("talloc failed\n"));
1817 TALLOC_FREE(result);
1818 return NULL;
1819 }
1820
1821 if (!(db_ctdb = talloc(result, struct db_ctdb_ctx))) {
1822 DEBUG(0, ("talloc failed\n"));
1823 TALLOC_FREE(result);
1824 return NULL;
1825 }
1826
1827 result->name = talloc_strdup(result, name);
1828 if (result->name == NULL) {
1829 DEBUG(0, ("talloc failed\n"));
1830 TALLOC_FREE(result);
1831 return NULL;
1832 }
1833
1834 db_ctdb->transaction = NULL;
1835 db_ctdb->db = result;
1836
1837 ret = ctdbd_db_attach(messaging_ctdb_connection(), name,
1838 &db_ctdb->db_id, persistent);
1839 if (ret != 0) {
1840 DEBUG(0, ("ctdbd_db_attach failed for %s: %s\n", name,
1841 strerror(ret)));
1842 TALLOC_FREE(result);
1843 return NULL;
1844 }
1845
1846 if (tdb_flags & TDB_SEQNUM) {
1847 data.dptr = (uint8_t *)&db_ctdb->db_id;
1848 data.dsize = sizeof(db_ctdb->db_id);
1849
1850 ret = ctdbd_control_local(messaging_ctdb_connection(),
1851 CTDB_CONTROL_ENABLE_SEQNUM,
1852 0, 0, data,
1853 NULL, NULL, &cstatus);
1854 if ((ret != 0) || cstatus != 0) {
1855 DBG_ERR("ctdb_control for enable seqnum "
1856 "failed: %s\n", strerror(ret));
1857 TALLOC_FREE(result);
1858 return NULL;
1859 }
1860 }
1861
1862 db_path = ctdbd_dbpath(messaging_ctdb_connection(), db_ctdb,
1863 db_ctdb->db_id);
1864 if (db_path == NULL) {
1865 DBG_ERR("ctdbd_dbpath failed\n");
1866 TALLOC_FREE(result);
1867 return NULL;
1868 }
1869
1870 result->persistent = persistent;
1871 result->lock_order = lock_order;
1872
1873 data.dptr = (uint8_t *)&db_ctdb->db_id;
1874 data.dsize = sizeof(db_ctdb->db_id);
1875
1876 ret = ctdbd_control_local(messaging_ctdb_connection(),
1877 CTDB_CONTROL_DB_OPEN_FLAGS,
1878 0, 0, data, NULL, &outdata, &cstatus);
1879 if (ret != 0) {
1880 DBG_ERR(" ctdb control for db_open_flags "
1881 "failed: %s\n", strerror(ret));
1882 TALLOC_FREE(result);
1883 return NULL;
1884 }
1885
1886 if (cstatus != 0 || outdata.dsize != sizeof(int)) {
1887 DBG_ERR("ctdb_control for db_open_flags failed\n");
1888 TALLOC_FREE(outdata.dptr);
1889 TALLOC_FREE(result);
1890 return NULL;
1891 }
1892
1893 tdb_flags = *(int *)outdata.dptr;
1894 TALLOC_FREE(outdata.dptr);
1895
1896 if (!result->persistent) {
1897 ret = ctdb_async_ctx_init(NULL, messaging_tevent_context(msg_ctx));
1898 if (ret != 0) {
1899 DBG_ERR("ctdb_async_ctx_init failed: %s\n", strerror(ret));
1900 TALLOC_FREE(result);
1901 return NULL;
1902 }
1903 }
1904
1905 if (!result->persistent &&
1906 (dbwrap_flags & DBWRAP_FLAG_OPTIMIZE_READONLY_ACCESS))
1907 {
1908 TDB_DATA indata;
1909
1910 indata = make_tdb_data((uint8_t *)&db_ctdb->db_id,
1911 sizeof(db_ctdb->db_id));
1912
1913 ret = ctdbd_control_local(
1914 messaging_ctdb_connection(),
1915 CTDB_CONTROL_SET_DB_READONLY, 0, 0,
1916 indata, NULL, NULL, &cstatus);
1917 if ((ret != 0) || (cstatus != 0)) {
1918 DEBUG(1, ("CTDB_CONTROL_SET_DB_READONLY failed: "
1919 "%s, %"PRIi32"\n", strerror(ret), cstatus));
1920 TALLOC_FREE(result);
1921 return NULL;
1922 }
1923 }
1924
1925 lp_ctx = loadparm_init_s3(db_path, loadparm_s3_helpers());
1926
1927 if (hash_size == 0) {
1928 hash_size = lpcfg_tdb_hash_size(lp_ctx, db_path);
1929 }
1930
1931 db_ctdb->wtdb = tdb_wrap_open(db_ctdb, db_path, hash_size,
1932 lpcfg_tdb_flags(lp_ctx, tdb_flags),
1933 O_RDWR, 0);
1934 talloc_unlink(db_path, lp_ctx);
1935 if (db_ctdb->wtdb == NULL) {
1936 DEBUG(0, ("Could not open tdb %s: %s\n", db_path, strerror(errno)));
1937 TALLOC_FREE(result);
1938 return NULL;
1939 }
1940 talloc_free(db_path);
1941
1942 /* honor permissions if user has specified O_CREAT */
1943 if (open_flags & O_CREAT) {
1944 int fd;
1945 fd = tdb_fd(db_ctdb->wtdb->tdb);
1946 ret = fchmod(fd, mode);
1947 if (ret == -1) {
1948 DBG_WARNING("fchmod failed: %s\n",
1949 strerror(errno));
1950 TALLOC_FREE(result);
1951 return NULL;
1952 }
1953 }
1954
1955 if (result->persistent) {
1956 db_ctdb->lock_ctx = g_lock_ctx_init(db_ctdb, msg_ctx);
1957 if (db_ctdb->lock_ctx == NULL) {
1958 DEBUG(0, ("g_lock_ctx_init failed\n"));
1959 TALLOC_FREE(result);
1960 return NULL;
1961 }
1962 }
1963
1964 db_ctdb->warn_unlock_msecs = lp_parm_int(-1, "ctdb",
1965 "unlock_warn_threshold", 5);
1966 db_ctdb->warn_migrate_attempts = lp_parm_int(-1, "ctdb",
1967 "migrate_attempts", 10);
1968 db_ctdb->warn_migrate_msecs = lp_parm_int(-1, "ctdb",
1969 "migrate_duration", 5000);
1970 db_ctdb->warn_locktime_msecs = lp_ctdb_locktime_warn_threshold();
1971
1972 result->private_data = (void *)db_ctdb;
1973 result->fetch_locked = db_ctdb_fetch_locked;
1974 result->parse_record = db_ctdb_parse_record;
1975 result->parse_record_send = db_ctdb_parse_record_send;
1976 result->parse_record_recv = db_ctdb_parse_record_recv;
1977 result->traverse = db_ctdb_traverse;
1978 result->traverse_read = db_ctdb_traverse_read;
1979 result->get_seqnum = db_ctdb_get_seqnum;
1980 result->transaction_start = db_ctdb_transaction_start;
1981 result->transaction_commit = db_ctdb_transaction_commit;
1982 result->transaction_cancel = db_ctdb_transaction_cancel;
1983 result->id = db_ctdb_id;
1984
1985 DEBUG(3,("db_open_ctdb: opened database '%s' with dbid 0x%x\n",
1986 name, db_ctdb->db_id));
1987
1988 return result;
1989 }
Samba GIT mirror