?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
s3: Slightly simplify db_ctdb_marshall_loop_next
[Samba/gebeck_regimport.git] / source3 / lib / dbwrap / dbwrap_ctdb.c
blob43b4b28d0db7453ab07ecca6d4c9d7a17da3022c
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 #ifdef CLUSTER_SUPPORT
31
32 /*
33 * It is not possible to include ctdb.h and tdb_compat.h (included via
34 * some other include above) without warnings. This fixes those
35 * warnings.
36 */
37
38 #ifdef typesafe_cb
39 #undef typesafe_cb
40 #endif
41
42 #ifdef typesafe_cb_preargs
43 #undef typesafe_cb_preargs
44 #endif
45
46 #ifdef typesafe_cb_postargs
47 #undef typesafe_cb_postargs
48 #endif
49
50 #include "ctdb.h"
51 #include "ctdb_private.h"
52 #include "ctdbd_conn.h"
53 #include "dbwrap/dbwrap.h"
54 #include "dbwrap/dbwrap_private.h"
55 #include "dbwrap/dbwrap_ctdb.h"
56 #include "g_lock.h"
57 #include "messages.h"
58
59 struct db_ctdb_transaction_handle {
60 struct db_ctdb_ctx *ctx;
61 /*
62 * we store the writes done under a transaction:
63 */
64 struct ctdb_marshall_buffer *m_write;
65 uint32_t nesting;
66 bool nested_cancel;
67 char *lock_name;
68 };
69
70 struct db_ctdb_ctx {
71 struct db_context *db;
72 struct tdb_wrap *wtdb;
73 uint32_t db_id;
74 struct db_ctdb_transaction_handle *transaction;
75 struct g_lock_ctx *lock_ctx;
76 };
77
78 struct db_ctdb_rec {
79 struct db_ctdb_ctx *ctdb_ctx;
80 struct ctdb_ltdb_header header;
81 struct timeval lock_time;
82 };
83
84 static NTSTATUS tdb_error_to_ntstatus(struct tdb_context *tdb)
85 {
86 enum TDB_ERROR tret = tdb_error(tdb);
87
88 return map_nt_error_from_tdb(tret);
89 }
90
91 struct db_ctdb_ltdb_parse_state {
92 void (*parser)(TDB_DATA key, struct ctdb_ltdb_header *header,
93 TDB_DATA data, void *private_data);
94 void *private_data;
95 };
96
97 static int db_ctdb_ltdb_parser(TDB_DATA key, TDB_DATA data,
98 void *private_data)
99 {
100 struct db_ctdb_ltdb_parse_state *state =
101 (struct db_ctdb_ltdb_parse_state *)private_data;
102
103 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
104 return -1;
105 }
106 state->parser(
107 key, (struct ctdb_ltdb_header *)data.dptr,
108 make_tdb_data(data.dptr + sizeof(struct ctdb_ltdb_header),
109 data.dsize - sizeof(struct ctdb_ltdb_header)),
110 state->private_data);
111 return 0;
112 }
113
114 static NTSTATUS db_ctdb_ltdb_parse(
115 struct db_ctdb_ctx *db, TDB_DATA key,
116 void (*parser)(TDB_DATA key, struct ctdb_ltdb_header *header,
117 TDB_DATA data, void *private_data),
118 void *private_data)
119 {
120 struct db_ctdb_ltdb_parse_state state;
121 int ret;
122
123 state.parser = parser;
124 state.private_data = private_data;
125
126 ret = tdb_parse_record(db->wtdb->tdb, key, db_ctdb_ltdb_parser,
127 &state);
128 if (ret == -1) {
129 return NT_STATUS_NOT_FOUND;
130 }
131 return NT_STATUS_OK;
132 }
133
134 struct db_ctdb_ltdb_fetch_state {
135 struct ctdb_ltdb_header *header;
136 TALLOC_CTX *mem_ctx;
137 TDB_DATA *data;
138 bool oom;
139 };
140
141 static void db_ctdb_ltdb_fetch_parser(
142 TDB_DATA key, struct ctdb_ltdb_header *header,
143 TDB_DATA data, void *private_data)
144 {
145 struct db_ctdb_ltdb_fetch_state *state =
146 (struct db_ctdb_ltdb_fetch_state *)private_data;
147
148 if (state->header != NULL) {
149 memcpy(state->header, header, sizeof(struct ctdb_ltdb_header));
150 }
151 if (state->data == NULL) {
152 return;
153 }
154 state->data->dsize = data.dsize;
155 if (data.dsize == 0) {
156 state->data->dptr = NULL;
157 return;
158 }
159 state->data->dptr = talloc_memdup(state->mem_ctx, data.dptr,
160 data.dsize);
161 if (state->data->dptr == NULL) {
162 state->oom = true;
163 return;
164 }
165 }
166
167 /**
168 * fetch a record from the tdb, separating out the header
169 * information and returning the body of the record.
170 */
171 static NTSTATUS db_ctdb_ltdb_fetch(struct db_ctdb_ctx *db,
172 TDB_DATA key,
173 struct ctdb_ltdb_header *header,
174 TALLOC_CTX *mem_ctx,
175 TDB_DATA *data)
176 {
177 struct db_ctdb_ltdb_fetch_state state;
178 NTSTATUS status;
179
180 state.header = header;
181 state.mem_ctx = mem_ctx;
182 state.data = data;
183 state.oom = false;
184
185 status = db_ctdb_ltdb_parse(db, key, db_ctdb_ltdb_fetch_parser,
186 &state);
187 if (!NT_STATUS_IS_OK(status)) {
188 if (data) {
189 ZERO_STRUCTP(data);
190 }
191 if (header) {
192 header->dmaster = (uint32_t)-1;
193 header->rsn = 0;
194 }
195 return status;
196 }
197 if (state.oom) {
198 return NT_STATUS_NO_MEMORY;
199 }
200 return NT_STATUS_OK;
201 }
202
203 /*
204 * Store a record together with the ctdb record header
205 * in the local copy of the database.
206 */
207 static NTSTATUS db_ctdb_ltdb_store(struct db_ctdb_ctx *db,
208 TDB_DATA key,
209 struct ctdb_ltdb_header *header,
210 TDB_DATA data)
211 {
212 TALLOC_CTX *tmp_ctx = talloc_stackframe();
213 TDB_DATA rec;
214 int ret;
215
216 rec.dsize = data.dsize + sizeof(struct ctdb_ltdb_header);
217 rec.dptr = (uint8_t *)talloc_size(tmp_ctx, rec.dsize);
218
219 if (rec.dptr == NULL) {
220 talloc_free(tmp_ctx);
221 return NT_STATUS_NO_MEMORY;
222 }
223
224 memcpy(rec.dptr, header, sizeof(struct ctdb_ltdb_header));
225 memcpy(sizeof(struct ctdb_ltdb_header) + (uint8_t *)rec.dptr, data.dptr, data.dsize);
226
227 ret = tdb_store(db->wtdb->tdb, key, rec, TDB_REPLACE);
228
229 talloc_free(tmp_ctx);
230
231 return (ret == 0) ? NT_STATUS_OK
232 : tdb_error_to_ntstatus(db->wtdb->tdb);
233
234 }
235
236 /*
237 form a ctdb_rec_data record from a key/data pair
238 */
239 static struct ctdb_rec_data *db_ctdb_marshall_record(TALLOC_CTX *mem_ctx, uint32_t reqid,
240 TDB_DATA key,
241 struct ctdb_ltdb_header *header,
242 TDB_DATA data)
243 {
244 size_t length;
245 struct ctdb_rec_data *d;
246
247 length = offsetof(struct ctdb_rec_data, data) + key.dsize +
248 data.dsize + sizeof(*header);
249 d = (struct ctdb_rec_data *)talloc_size(mem_ctx, length);
250 if (d == NULL) {
251 return NULL;
252 }
253 d->length = length;
254 d->reqid = reqid;
255 d->keylen = key.dsize;
256 memcpy(&d->data[0], key.dptr, key.dsize);
257
258 d->datalen = data.dsize + sizeof(*header);
259 memcpy(&d->data[key.dsize], header, sizeof(*header));
260 memcpy(&d->data[key.dsize+sizeof(*header)], data.dptr, data.dsize);
261 return d;
262 }
263
264
265 /* helper function for marshalling multiple records */
266 static struct ctdb_marshall_buffer *db_ctdb_marshall_add(TALLOC_CTX *mem_ctx,
267 struct ctdb_marshall_buffer *m,
268 uint64_t db_id,
269 uint32_t reqid,
270 TDB_DATA key,
271 struct ctdb_ltdb_header *header,
272 TDB_DATA data)
273 {
274 struct ctdb_rec_data *r;
275 size_t m_size, r_size;
276 struct ctdb_marshall_buffer *m2 = NULL;
277
278 r = db_ctdb_marshall_record(talloc_tos(), reqid, key, header, data);
279 if (r == NULL) {
280 talloc_free(m);
281 return NULL;
282 }
283
284 if (m == NULL) {
285 m = (struct ctdb_marshall_buffer *)talloc_zero_size(
286 mem_ctx, offsetof(struct ctdb_marshall_buffer, data));
287 if (m == NULL) {
288 goto done;
289 }
290 m->db_id = db_id;
291 }
292
293 m_size = talloc_get_size(m);
294 r_size = talloc_get_size(r);
295
296 m2 = (struct ctdb_marshall_buffer *)talloc_realloc_size(
297 mem_ctx, m, m_size + r_size);
298 if (m2 == NULL) {
299 talloc_free(m);
300 goto done;
301 }
302
303 memcpy(m_size + (uint8_t *)m2, r, r_size);
304
305 m2->count++;
306
307 done:
308 talloc_free(r);
309 return m2;
310 }
311
312 /* we've finished marshalling, return a data blob with the marshalled records */
313 static TDB_DATA db_ctdb_marshall_finish(struct ctdb_marshall_buffer *m)
314 {
315 TDB_DATA data;
316 data.dptr = (uint8_t *)m;
317 data.dsize = talloc_get_size(m);
318 return data;
319 }
320
321 /*
322 loop over a marshalling buffer
323
324 - pass r==NULL to start
325 - loop the number of times indicated by m->count
326 */
327 static struct ctdb_rec_data *db_ctdb_marshall_loop_next(struct ctdb_marshall_buffer *m, struct ctdb_rec_data *r,
328 uint32_t *reqid,
329 struct ctdb_ltdb_header *header,
330 TDB_DATA *key, TDB_DATA *data)
331 {
332 if (r == NULL) {
333 r = (struct ctdb_rec_data *)&m->data[0];
334 } else {
335 r = (struct ctdb_rec_data *)(r->length + (uint8_t *)r);
336 }
337
338 if (reqid != NULL) {
339 *reqid = r->reqid;
340 }
341
342 key->dptr = &r->data[0];
343 key->dsize = r->keylen;
344
345 if (data != NULL) {
346 data->dptr = &r->data[r->keylen];
347 data->dsize = r->datalen;
348 if (header != NULL) {
349 data->dptr += sizeof(*header);
350 data->dsize -= sizeof(*header);
351 }
352 }
353
354 if (header != NULL) {
355 if (r->datalen < sizeof(*header)) {
356 return NULL;
357 }
358 *header = *(struct ctdb_ltdb_header *)&r->data[r->keylen];
359 }
360
361 return r;
362 }
363
364 /**
365 * CTDB transaction destructor
366 */
367 static int db_ctdb_transaction_destructor(struct db_ctdb_transaction_handle *h)
368 {
369 NTSTATUS status;
370
371 status = g_lock_unlock(h->ctx->lock_ctx, h->lock_name);
372 if (!NT_STATUS_IS_OK(status)) {
373 DEBUG(0, ("g_lock_unlock failed for %s: %s\n", h->lock_name,
374 nt_errstr(status)));
375 return -1;
376 }
377 return 0;
378 }
379
380 /**
381 * CTDB dbwrap API: transaction_start function
382 * starts a transaction on a persistent database
383 */
384 static int db_ctdb_transaction_start(struct db_context *db)
385 {
386 struct db_ctdb_transaction_handle *h;
387 NTSTATUS status;
388 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
389 struct db_ctdb_ctx);
390
391 if (!db->persistent) {
392 DEBUG(0,("transactions not supported on non-persistent database 0x%08x\n",
393 ctx->db_id));
394 return -1;
395 }
396
397 if (ctx->transaction) {
398 ctx->transaction->nesting++;
399 DEBUG(5, (__location__ " transaction start on db 0x%08x: nesting %d -> %d\n",
400 ctx->db_id, ctx->transaction->nesting - 1, ctx->transaction->nesting));
401 return 0;
402 }
403
404 h = talloc_zero(db, struct db_ctdb_transaction_handle);
405 if (h == NULL) {
406 DEBUG(0,(__location__ " oom for transaction handle\n"));
407 return -1;
408 }
409
410 h->ctx = ctx;
411
412 h->lock_name = talloc_asprintf(h, "transaction_db_0x%08x",
413 (unsigned int)ctx->db_id);
414 if (h->lock_name == NULL) {
415 DEBUG(0, ("talloc_asprintf failed\n"));
416 TALLOC_FREE(h);
417 return -1;
418 }
419
420 /*
421 * Wait a day, i.e. forever...
422 */
423 status = g_lock_lock(ctx->lock_ctx, h->lock_name, G_LOCK_WRITE,
424 timeval_set(86400, 0));
425 if (!NT_STATUS_IS_OK(status)) {
426 DEBUG(0, ("g_lock_lock failed: %s\n", nt_errstr(status)));
427 TALLOC_FREE(h);
428 return -1;
429 }
430
431 talloc_set_destructor(h, db_ctdb_transaction_destructor);
432
433 ctx->transaction = h;
434
435 DEBUG(5,(__location__ " transaction started on db 0x%08x\n", ctx->db_id));
436
437 return 0;
438 }
439
440 static bool pull_newest_from_marshall_buffer(struct ctdb_marshall_buffer *buf,
441 TDB_DATA key,
442 struct ctdb_ltdb_header *pheader,
443 TALLOC_CTX *mem_ctx,
444 TDB_DATA *pdata)
445 {
446 struct ctdb_rec_data *rec = NULL;
447 struct ctdb_ltdb_header h;
448 bool found = false;
449 TDB_DATA data;
450 int i;
451
452 if (buf == NULL) {
453 return false;
454 }
455
456 ZERO_STRUCT(h);
457 ZERO_STRUCT(data);
458
459 /*
460 * Walk the list of records written during this
461 * transaction. If we want to read one we have already
462 * written, return the last written sample. Thus we do not do
463 * a "break;" for the first hit, this record might have been
464 * overwritten later.
465 */
466
467 for (i=0; i<buf->count; i++) {
468 TDB_DATA tkey, tdata;
469 uint32_t reqid;
470 struct ctdb_ltdb_header hdr;
471
472 ZERO_STRUCT(hdr);
473
474 rec = db_ctdb_marshall_loop_next(buf, rec, &reqid, &hdr, &tkey,
475 &tdata);
476 if (rec == NULL) {
477 return false;
478 }
479
480 if (tdb_data_equal(key, tkey)) {
481 found = true;
482 data = tdata;
483 h = hdr;
484 }
485 }
486
487 if (!found) {
488 return false;
489 }
490
491 if (pdata != NULL) {
492 data.dptr = (uint8_t *)talloc_memdup(mem_ctx, data.dptr,
493 data.dsize);
494 if ((data.dsize != 0) && (data.dptr == NULL)) {
495 return false;
496 }
497 *pdata = data;
498 }
499
500 if (pheader != NULL) {
501 *pheader = h;
502 }
503
504 return true;
505 }
506
507 /*
508 fetch a record inside a transaction
509 */
510 static NTSTATUS db_ctdb_transaction_fetch(struct db_ctdb_ctx *db,
511 TALLOC_CTX *mem_ctx,
512 TDB_DATA key, TDB_DATA *data)
513 {
514 struct db_ctdb_transaction_handle *h = db->transaction;
515 NTSTATUS status;
516 bool found;
517
518 found = pull_newest_from_marshall_buffer(h->m_write, key, NULL,
519 mem_ctx, data);
520 if (found) {
521 return NT_STATUS_OK;
522 }
523
524 status = db_ctdb_ltdb_fetch(h->ctx, key, NULL, mem_ctx, data);
525
526 if (NT_STATUS_EQUAL(status, NT_STATUS_NOT_FOUND)) {
527 *data = tdb_null;
528 }
529
530 return status;
531 }
532
533 /**
534 * Fetch a record from a persistent database
535 * without record locking and without an active transaction.
536 *
537 * This just fetches from the local database copy.
538 * Since the databases are kept in syc cluster-wide,
539 * there is no point in doing a ctdb call to fetch the
540 * record from the lmaster. It does even harm since migration
541 * of records bump their RSN and hence render the persistent
542 * database inconsistent.
543 */
544 static NTSTATUS db_ctdb_fetch_persistent(struct db_ctdb_ctx *db,
545 TALLOC_CTX *mem_ctx,
546 TDB_DATA key, TDB_DATA *data)
547 {
548 NTSTATUS status;
549
550 status = db_ctdb_ltdb_fetch(db, key, NULL, mem_ctx, data);
551
552 if (NT_STATUS_EQUAL(status, NT_STATUS_NOT_FOUND)) {
553 *data = tdb_null;
554 }
555
556 return status;
557 }
558
559 static NTSTATUS db_ctdb_store_transaction(struct db_record *rec, TDB_DATA data, int flag);
560 static NTSTATUS db_ctdb_delete_transaction(struct db_record *rec);
561
562 static struct db_record *db_ctdb_fetch_locked_transaction(struct db_ctdb_ctx *ctx,
563 TALLOC_CTX *mem_ctx,
564 TDB_DATA key)
565 {
566 struct db_record *result;
567 TDB_DATA ctdb_data;
568
569 if (!(result = talloc(mem_ctx, struct db_record))) {
570 DEBUG(0, ("talloc failed\n"));
571 return NULL;
572 }
573
574 result->private_data = ctx->transaction;
575
576 result->key.dsize = key.dsize;
577 result->key.dptr = (uint8_t *)talloc_memdup(result, key.dptr,
578 key.dsize);
579 if (result->key.dptr == NULL) {
580 DEBUG(0, ("talloc failed\n"));
581 TALLOC_FREE(result);
582 return NULL;
583 }
584
585 result->store = db_ctdb_store_transaction;
586 result->delete_rec = db_ctdb_delete_transaction;
587
588 if (pull_newest_from_marshall_buffer(ctx->transaction->m_write, key,
589 NULL, result, &result->value)) {
590 return result;
591 }
592
593 ctdb_data = tdb_fetch_compat(ctx->wtdb->tdb, key);
594 if (ctdb_data.dptr == NULL) {
595 /* create the record */
596 result->value = tdb_null;
597 return result;
598 }
599
600 result->value.dsize = ctdb_data.dsize - sizeof(struct ctdb_ltdb_header);
601 result->value.dptr = NULL;
602
603 if ((result->value.dsize != 0)
604 && !(result->value.dptr = (uint8_t *)talloc_memdup(
605 result, ctdb_data.dptr + sizeof(struct ctdb_ltdb_header),
606 result->value.dsize))) {
607 DEBUG(0, ("talloc failed\n"));
608 TALLOC_FREE(result);
609 }
610
611 SAFE_FREE(ctdb_data.dptr);
612
613 return result;
614 }
615
616 static int db_ctdb_record_destructor(struct db_record **recp)
617 {
618 struct db_record *rec = talloc_get_type_abort(*recp, struct db_record);
619 struct db_ctdb_transaction_handle *h = talloc_get_type_abort(
620 rec->private_data, struct db_ctdb_transaction_handle);
621 int ret = h->ctx->db->transaction_commit(h->ctx->db);
622 if (ret != 0) {
623 DEBUG(0,(__location__ " transaction_commit failed\n"));
624 }
625 return 0;
626 }
627
628 /*
629 auto-create a transaction for persistent databases
630 */
631 static struct db_record *db_ctdb_fetch_locked_persistent(struct db_ctdb_ctx *ctx,
632 TALLOC_CTX *mem_ctx,
633 TDB_DATA key)
634 {
635 int res;
636 struct db_record *rec, **recp;
637
638 res = db_ctdb_transaction_start(ctx->db);
639 if (res == -1) {
640 return NULL;
641 }
642
643 rec = db_ctdb_fetch_locked_transaction(ctx, mem_ctx, key);
644 if (rec == NULL) {
645 ctx->db->transaction_cancel(ctx->db);
646 return NULL;
647 }
648
649 /* destroy this transaction when we release the lock */
650 recp = talloc(rec, struct db_record *);
651 if (recp == NULL) {
652 ctx->db->transaction_cancel(ctx->db);
653 talloc_free(rec);
654 return NULL;
655 }
656 *recp = rec;
657 talloc_set_destructor(recp, db_ctdb_record_destructor);
658 return rec;
659 }
660
661
662 /*
663 stores a record inside a transaction
664 */
665 static NTSTATUS db_ctdb_transaction_store(struct db_ctdb_transaction_handle *h,
666 TDB_DATA key, TDB_DATA data)
667 {
668 TALLOC_CTX *tmp_ctx = talloc_new(h);
669 TDB_DATA rec;
670 struct ctdb_ltdb_header header;
671
672 ZERO_STRUCT(header);
673
674 /* we need the header so we can update the RSN */
675
676 if (!pull_newest_from_marshall_buffer(h->m_write, key, &header,
677 NULL, NULL)) {
678
679 rec = tdb_fetch_compat(h->ctx->wtdb->tdb, key);
680
681 if (rec.dptr != NULL) {
682 memcpy(&header, rec.dptr,
683 sizeof(struct ctdb_ltdb_header));
684 rec.dsize -= sizeof(struct ctdb_ltdb_header);
685
686 /*
687 * a special case, we are writing the same
688 * data that is there now
689 */
690 if (data.dsize == rec.dsize &&
691 memcmp(data.dptr,
692 rec.dptr + sizeof(struct ctdb_ltdb_header),
693 data.dsize) == 0) {
694 SAFE_FREE(rec.dptr);
695 talloc_free(tmp_ctx);
696 return NT_STATUS_OK;
697 }
698 }
699 SAFE_FREE(rec.dptr);
700 }
701
702 header.dmaster = get_my_vnn();
703 header.rsn++;
704
705 h->m_write = db_ctdb_marshall_add(h, h->m_write, h->ctx->db_id, 0, key, &header, data);
706 if (h->m_write == NULL) {
707 DEBUG(0,(__location__ " Failed to add to marshalling record\n"));
708 talloc_free(tmp_ctx);
709 return NT_STATUS_NO_MEMORY;
710 }
711
712 talloc_free(tmp_ctx);
713 return NT_STATUS_OK;
714 }
715
716
717 /*
718 a record store inside a transaction
719 */
720 static NTSTATUS db_ctdb_store_transaction(struct db_record *rec, TDB_DATA data, int flag)
721 {
722 struct db_ctdb_transaction_handle *h = talloc_get_type_abort(
723 rec->private_data, struct db_ctdb_transaction_handle);
724 NTSTATUS status;
725
726 status = db_ctdb_transaction_store(h, rec->key, data);
727 return status;
728 }
729
730 /*
731 a record delete inside a transaction
732 */
733 static NTSTATUS db_ctdb_delete_transaction(struct db_record *rec)
734 {
735 struct db_ctdb_transaction_handle *h = talloc_get_type_abort(
736 rec->private_data, struct db_ctdb_transaction_handle);
737 NTSTATUS status;
738
739 status = db_ctdb_transaction_store(h, rec->key, tdb_null);
740 return status;
741 }
742
743 static void db_ctdb_fetch_db_seqnum_parser(
744 TDB_DATA key, struct ctdb_ltdb_header *header,
745 TDB_DATA data, void *private_data)
746 {
747 uint64_t *seqnum = (uint64_t *)private_data;
748
749 if (data.dsize != sizeof(uint64_t)) {
750 *seqnum = 0;
751 return;
752 }
753 memcpy(seqnum, data.dptr, sizeof(*seqnum));
754 }
755
756 /**
757 * Fetch the db sequence number of a persistent db directly from the db.
758 */
759 static NTSTATUS db_ctdb_fetch_db_seqnum_from_db(struct db_ctdb_ctx *db,
760 uint64_t *seqnum)
761 {
762 NTSTATUS status;
763 TDB_DATA key;
764
765 if (seqnum == NULL) {
766 return NT_STATUS_INVALID_PARAMETER;
767 }
768
769 key = string_term_tdb_data(CTDB_DB_SEQNUM_KEY);
770
771 status = db_ctdb_ltdb_parse(
772 db, key, db_ctdb_fetch_db_seqnum_parser, seqnum);
773
774 if (NT_STATUS_IS_OK(status)) {
775 return NT_STATUS_OK;
776 }
777 if (NT_STATUS_EQUAL(status, NT_STATUS_NOT_FOUND)) {
778 *seqnum = 0;
779 return NT_STATUS_OK;
780 }
781 return status;
782 }
783
784 /**
785 * Store the database sequence number inside a transaction.
786 */
787 static NTSTATUS db_ctdb_store_db_seqnum(struct db_ctdb_transaction_handle *h,
788 uint64_t seqnum)
789 {
790 NTSTATUS status;
791 const char *keyname = CTDB_DB_SEQNUM_KEY;
792 TDB_DATA key;
793 TDB_DATA data;
794
795 key = string_term_tdb_data(keyname);
796
797 data.dptr = (uint8_t *)&seqnum;
798 data.dsize = sizeof(uint64_t);
799
800 status = db_ctdb_transaction_store(h, key, data);
801
802 return status;
803 }
804
805 /*
806 commit a transaction
807 */
808 static int db_ctdb_transaction_commit(struct db_context *db)
809 {
810 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
811 struct db_ctdb_ctx);
812 NTSTATUS rets;
813 int status;
814 struct db_ctdb_transaction_handle *h = ctx->transaction;
815 uint64_t old_seqnum, new_seqnum;
816 int ret;
817
818 if (h == NULL) {
819 DEBUG(0,(__location__ " transaction commit with no open transaction on db 0x%08x\n", ctx->db_id));
820 return -1;
821 }
822
823 if (h->nested_cancel) {
824 db->transaction_cancel(db);
825 DEBUG(5,(__location__ " Failed transaction commit after nested cancel\n"));
826 return -1;
827 }
828
829 if (h->nesting != 0) {
830 h->nesting--;
831 DEBUG(5, (__location__ " transaction commit on db 0x%08x: nesting %d -> %d\n",
832 ctx->db_id, ctx->transaction->nesting + 1, ctx->transaction->nesting));
833 return 0;
834 }
835
836 if (h->m_write == NULL) {
837 /*
838 * No changes were made, so don't change the seqnum,
839 * don't push to other node, just exit with success.
840 */
841 ret = 0;
842 goto done;
843 }
844
845 DEBUG(5,(__location__ " transaction commit on db 0x%08x\n", ctx->db_id));
846
847 /*
848 * As the last db action before committing, bump the database sequence
849 * number. Note that this undoes all changes to the seqnum records
850 * performed under the transaction. This record is not meant to be
851 * modified by user interaction. It is for internal use only...
852 */
853 rets = db_ctdb_fetch_db_seqnum_from_db(ctx, &old_seqnum);
854 if (!NT_STATUS_IS_OK(rets)) {
855 DEBUG(1, (__location__ " failed to fetch the db sequence number "
856 "in transaction commit on db 0x%08x\n", ctx->db_id));
857 ret = -1;
858 goto done;
859 }
860
861 new_seqnum = old_seqnum + 1;
862
863 rets = db_ctdb_store_db_seqnum(h, new_seqnum);
864 if (!NT_STATUS_IS_OK(rets)) {
865 DEBUG(1, (__location__ "failed to store the db sequence number "
866 " in transaction commit on db 0x%08x\n", ctx->db_id));
867 ret = -1;
868 goto done;
869 }
870
871 again:
872 /* tell ctdbd to commit to the other nodes */
873 rets = ctdbd_control_local(messaging_ctdbd_connection(),
874 CTDB_CONTROL_TRANS3_COMMIT,
875 h->ctx->db_id, 0,
876 db_ctdb_marshall_finish(h->m_write),
877 NULL, NULL, &status);
878 if (!NT_STATUS_IS_OK(rets) || status != 0) {
879 /*
880 * The TRANS3_COMMIT control should only possibly fail when a
881 * recovery has been running concurrently. In any case, the db
882 * will be the same on all nodes, either the new copy or the
883 * old copy. This can be detected by comparing the old and new
884 * local sequence numbers.
885 */
886 rets = db_ctdb_fetch_db_seqnum_from_db(ctx, &new_seqnum);
887 if (!NT_STATUS_IS_OK(rets)) {
888 DEBUG(1, (__location__ " failed to refetch db sequence "
889 "number after failed TRANS3_COMMIT\n"));
890 ret = -1;
891 goto done;
892 }
893
894 if (new_seqnum == old_seqnum) {
895 /* Recovery prevented all our changes: retry. */
896 goto again;
897 }
898 if (new_seqnum != (old_seqnum + 1)) {
899 DEBUG(0, (__location__ " ERROR: new_seqnum[%lu] != "
900 "old_seqnum[%lu] + (0 or 1) after failed "
901 "TRANS3_COMMIT - this should not happen!\n",
902 (unsigned long)new_seqnum,
903 (unsigned long)old_seqnum));
904 ret = -1;
905 goto done;
906 }
907 /*
908 * Recovery propagated our changes to all nodes, completing
909 * our commit for us - succeed.
910 */
911 }
912
913 ret = 0;
914
915 done:
916 h->ctx->transaction = NULL;
917 talloc_free(h);
918 return ret;
919 }
920
921
922 /*
923 cancel a transaction
924 */
925 static int db_ctdb_transaction_cancel(struct db_context *db)
926 {
927 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
928 struct db_ctdb_ctx);
929 struct db_ctdb_transaction_handle *h = ctx->transaction;
930
931 if (h == NULL) {
932 DEBUG(0,(__location__ " transaction cancel with no open transaction on db 0x%08x\n", ctx->db_id));
933 return -1;
934 }
935
936 if (h->nesting != 0) {
937 h->nesting--;
938 h->nested_cancel = true;
939 DEBUG(5, (__location__ " transaction cancel on db 0x%08x: nesting %d -> %d\n",
940 ctx->db_id, ctx->transaction->nesting + 1, ctx->transaction->nesting));
941 return 0;
942 }
943
944 DEBUG(5,(__location__ " Cancel transaction on db 0x%08x\n", ctx->db_id));
945
946 ctx->transaction = NULL;
947 talloc_free(h);
948 return 0;
949 }
950
951
952 static NTSTATUS db_ctdb_store(struct db_record *rec, TDB_DATA data, int flag)
953 {
954 struct db_ctdb_rec *crec = talloc_get_type_abort(
955 rec->private_data, struct db_ctdb_rec);
956
957 return db_ctdb_ltdb_store(crec->ctdb_ctx, rec->key, &(crec->header), data);
958 }
959
960
961
962 #ifdef HAVE_CTDB_CONTROL_SCHEDULE_FOR_DELETION_DECL
963 static NTSTATUS db_ctdb_send_schedule_for_deletion(struct db_record *rec)
964 {
965 NTSTATUS status;
966 struct ctdb_control_schedule_for_deletion *dd;
967 TDB_DATA indata;
968 int cstatus;
969 struct db_ctdb_rec *crec = talloc_get_type_abort(
970 rec->private_data, struct db_ctdb_rec);
971
972 indata.dsize = offsetof(struct ctdb_control_schedule_for_deletion, key) + rec->key.dsize;
973 indata.dptr = talloc_zero_array(crec, uint8_t, indata.dsize);
974 if (indata.dptr == NULL) {
975 DEBUG(0, (__location__ " talloc failed!\n"));
976 return NT_STATUS_NO_MEMORY;
977 }
978
979 dd = (struct ctdb_control_schedule_for_deletion *)(void *)indata.dptr;
980 dd->db_id = crec->ctdb_ctx->db_id;
981 dd->hdr = crec->header;
982 dd->keylen = rec->key.dsize;
983 memcpy(dd->key, rec->key.dptr, rec->key.dsize);
984
985 status = ctdbd_control_local(messaging_ctdbd_connection(),
986 CTDB_CONTROL_SCHEDULE_FOR_DELETION,
987 crec->ctdb_ctx->db_id,
988 CTDB_CTRL_FLAG_NOREPLY, /* flags */
989 indata,
990 NULL, /* outdata */
991 NULL, /* errmsg */
992 &cstatus);
993 talloc_free(indata.dptr);
994
995 if (!NT_STATUS_IS_OK(status) || cstatus != 0) {
996 DEBUG(1, (__location__ " Error sending local control "
997 "SCHEDULE_FOR_DELETION: %s, cstatus = %d\n",
998 nt_errstr(status), cstatus));
999 if (NT_STATUS_IS_OK(status)) {
1000 status = NT_STATUS_UNSUCCESSFUL;
1001 }
1002 }
1003
1004 return status;
1005 }
1006 #endif
1007
1008 static NTSTATUS db_ctdb_delete(struct db_record *rec)
1009 {
1010 TDB_DATA data;
1011 NTSTATUS status;
1012
1013 /*
1014 * We have to store the header with empty data. TODO: Fix the
1015 * tdb-level cleanup
1016 */
1017
1018 ZERO_STRUCT(data);
1019
1020 status = db_ctdb_store(rec, data, 0);
1021 if (!NT_STATUS_IS_OK(status)) {
1022 return status;
1023 }
1024
1025 #ifdef HAVE_CTDB_CONTROL_SCHEDULE_FOR_DELETION_DECL
1026 status = db_ctdb_send_schedule_for_deletion(rec);
1027 #endif
1028
1029 return status;
1030 }
1031
1032 static int db_ctdb_record_destr(struct db_record* data)
1033 {
1034 struct db_ctdb_rec *crec = talloc_get_type_abort(
1035 data->private_data, struct db_ctdb_rec);
1036 int threshold;
1037
1038 DEBUG(10, (DEBUGLEVEL > 10
1039 ? "Unlocking db %u key %s\n"
1040 : "Unlocking db %u key %.20s\n",
1041 (int)crec->ctdb_ctx->db_id,
1042 hex_encode_talloc(data, (unsigned char *)data->key.dptr,
1043 data->key.dsize)));
1044
1045 tdb_chainunlock(crec->ctdb_ctx->wtdb->tdb, data->key);
1046
1047 threshold = lp_ctdb_locktime_warn_threshold();
1048 if (threshold != 0) {
1049 double timediff = timeval_elapsed(&crec->lock_time);
1050 if ((timediff * 1000) > threshold) {
1051 const char *key;
1052
1053 key = hex_encode_talloc(data,
1054 (unsigned char *)data->key.dptr,
1055 data->key.dsize);
1056 DEBUG(0, ("Held tdb lock on db %s, key %s %f seconds\n",
1057 tdb_name(crec->ctdb_ctx->wtdb->tdb), key,
1058 timediff));
1059 }
1060 }
1061
1062 return 0;
1063 }
1064
1065 /**
1066 * Check whether we have a valid local copy of the given record,
1067 * either for reading or for writing.
1068 */
1069 static bool db_ctdb_can_use_local_copy(TDB_DATA ctdb_data, bool read_only)
1070 {
1071 struct ctdb_ltdb_header *hdr;
1072
1073 if (ctdb_data.dptr == NULL)
1074 return false;
1075
1076 if (ctdb_data.dsize < sizeof(struct ctdb_ltdb_header))
1077 return false;
1078
1079 hdr = (struct ctdb_ltdb_header *)ctdb_data.dptr;
1080
1081 #ifdef HAVE_CTDB_WANT_READONLY_DECL
1082 if (hdr->dmaster != get_my_vnn()) {
1083 /* If we're not dmaster, it must be r/o copy. */
1084 return read_only && (hdr->flags & CTDB_REC_RO_HAVE_READONLY);
1085 }
1086
1087 /*
1088 * If we want write access, no one may have r/o copies.
1089 */
1090 return read_only || !(hdr->flags & CTDB_REC_RO_HAVE_DELEGATIONS);
1091 #else
1092 return (hdr->dmaster == get_my_vnn());
1093 #endif
1094 }
1095
1096 static struct db_record *fetch_locked_internal(struct db_ctdb_ctx *ctx,
1097 TALLOC_CTX *mem_ctx,
1098 TDB_DATA key,
1099 bool tryonly)
1100 {
1101 struct db_record *result;
1102 struct db_ctdb_rec *crec;
1103 NTSTATUS status;
1104 TDB_DATA ctdb_data;
1105 int migrate_attempts = 0;
1106 int lockret;
1107
1108 if (!(result = talloc(mem_ctx, struct db_record))) {
1109 DEBUG(0, ("talloc failed\n"));
1110 return NULL;
1111 }
1112
1113 if (!(crec = talloc_zero(result, struct db_ctdb_rec))) {
1114 DEBUG(0, ("talloc failed\n"));
1115 TALLOC_FREE(result);
1116 return NULL;
1117 }
1118
1119 result->db = ctx->db;
1120 result->private_data = (void *)crec;
1121 crec->ctdb_ctx = ctx;
1122
1123 result->key.dsize = key.dsize;
1124 result->key.dptr = (uint8_t *)talloc_memdup(result, key.dptr,
1125 key.dsize);
1126 if (result->key.dptr == NULL) {
1127 DEBUG(0, ("talloc failed\n"));
1128 TALLOC_FREE(result);
1129 return NULL;
1130 }
1131
1132 /*
1133 * Do a blocking lock on the record
1134 */
1135 again:
1136
1137 if (DEBUGLEVEL >= 10) {
1138 char *keystr = hex_encode_talloc(result, key.dptr, key.dsize);
1139 DEBUG(10, (DEBUGLEVEL > 10
1140 ? "Locking db %u key %s\n"
1141 : "Locking db %u key %.20s\n",
1142 (int)crec->ctdb_ctx->db_id, keystr));
1143 TALLOC_FREE(keystr);
1144 }
1145
1146 lockret = tryonly
1147 ? tdb_chainlock_nonblock(ctx->wtdb->tdb, key)
1148 : tdb_chainlock(ctx->wtdb->tdb, key);
1149 if (lockret != 0) {
1150 DEBUG(3, ("tdb_chainlock failed\n"));
1151 TALLOC_FREE(result);
1152 return NULL;
1153 }
1154
1155 result->store = db_ctdb_store;
1156 result->delete_rec = db_ctdb_delete;
1157 talloc_set_destructor(result, db_ctdb_record_destr);
1158
1159 ctdb_data = tdb_fetch_compat(ctx->wtdb->tdb, key);
1160
1161 /*
1162 * See if we have a valid record and we are the dmaster. If so, we can
1163 * take the shortcut and just return it.
1164 */
1165
1166 if (!db_ctdb_can_use_local_copy(ctdb_data, false)) {
1167 SAFE_FREE(ctdb_data.dptr);
1168 tdb_chainunlock(ctx->wtdb->tdb, key);
1169 talloc_set_destructor(result, NULL);
1170
1171 if (tryonly && (migrate_attempts != 0)) {
1172 DEBUG(5, ("record migrated away again\n"));
1173 TALLOC_FREE(result);
1174 return NULL;
1175 }
1176
1177 migrate_attempts += 1;
1178
1179 DEBUG(10, ("ctdb_data.dptr = %p, dmaster = %u (%u) %u\n",
1180 ctdb_data.dptr, ctdb_data.dptr ?
1181 ((struct ctdb_ltdb_header *)ctdb_data.dptr)->dmaster : -1,
1182 get_my_vnn(),
1183 ctdb_data.dptr ?
1184 ((struct ctdb_ltdb_header *)ctdb_data.dptr)->flags : 0));
1185
1186 status = ctdbd_migrate(messaging_ctdbd_connection(), ctx->db_id,
1187 key);
1188 if (!NT_STATUS_IS_OK(status)) {
1189 DEBUG(5, ("ctdb_migrate failed: %s\n",
1190 nt_errstr(status)));
1191 TALLOC_FREE(result);
1192 return NULL;
1193 }
1194 /* now its migrated, try again */
1195 goto again;
1196 }
1197
1198 if (migrate_attempts > 10) {
1199 DEBUG(0, ("db_ctdb_fetch_locked for %s key %s needed %d "
1200 "attempts\n", tdb_name(ctx->wtdb->tdb),
1201 hex_encode_talloc(talloc_tos(),
1202 (unsigned char *)key.dptr,
1203 key.dsize),
1204 migrate_attempts));
1205 }
1206
1207 GetTimeOfDay(&crec->lock_time);
1208
1209 memcpy(&crec->header, ctdb_data.dptr, sizeof(crec->header));
1210
1211 result->value.dsize = ctdb_data.dsize - sizeof(crec->header);
1212 result->value.dptr = NULL;
1213
1214 if ((result->value.dsize != 0)
1215 && !(result->value.dptr = (uint8_t *)talloc_memdup(
1216 result, ctdb_data.dptr + sizeof(crec->header),
1217 result->value.dsize))) {
1218 DEBUG(0, ("talloc failed\n"));
1219 TALLOC_FREE(result);
1220 }
1221
1222 SAFE_FREE(ctdb_data.dptr);
1223
1224 return result;
1225 }
1226
1227 static struct db_record *db_ctdb_fetch_locked(struct db_context *db,
1228 TALLOC_CTX *mem_ctx,
1229 TDB_DATA key)
1230 {
1231 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1232 struct db_ctdb_ctx);
1233
1234 if (ctx->transaction != NULL) {
1235 return db_ctdb_fetch_locked_transaction(ctx, mem_ctx, key);
1236 }
1237
1238 if (db->persistent) {
1239 return db_ctdb_fetch_locked_persistent(ctx, mem_ctx, key);
1240 }
1241
1242 return fetch_locked_internal(ctx, mem_ctx, key, false);
1243 }
1244
1245 static struct db_record *db_ctdb_try_fetch_locked(struct db_context *db,
1246 TALLOC_CTX *mem_ctx,
1247 TDB_DATA key)
1248 {
1249 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1250 struct db_ctdb_ctx);
1251
1252 if (ctx->transaction != NULL) {
1253 return db_ctdb_fetch_locked_transaction(ctx, mem_ctx, key);
1254 }
1255
1256 if (db->persistent) {
1257 return db_ctdb_fetch_locked_persistent(ctx, mem_ctx, key);
1258 }
1259
1260 return fetch_locked_internal(ctx, mem_ctx, key, true);
1261 }
1262
1263 /*
1264 fetch (unlocked, no migration) operation on ctdb
1265 */
1266 static NTSTATUS db_ctdb_fetch(struct db_context *db, TALLOC_CTX *mem_ctx,
1267 TDB_DATA key, TDB_DATA *data)
1268 {
1269 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1270 struct db_ctdb_ctx);
1271 NTSTATUS status;
1272 TDB_DATA ctdb_data;
1273
1274 if (ctx->transaction) {
1275 return db_ctdb_transaction_fetch(ctx, mem_ctx, key, data);
1276 }
1277
1278 if (db->persistent) {
1279 return db_ctdb_fetch_persistent(ctx, mem_ctx, key, data);
1280 }
1281
1282 /* try a direct fetch */
1283 ctdb_data = tdb_fetch_compat(ctx->wtdb->tdb, key);
1284
1285 /*
1286 * See if we have a valid record and we are the dmaster. If so, we can
1287 * take the shortcut and just return it.
1288 * we bypass the dmaster check for persistent databases
1289 */
1290 if (db_ctdb_can_use_local_copy(ctdb_data, true)) {
1291 /*
1292 * We have a valid local copy - avoid the ctdb protocol op
1293 */
1294 data->dsize = ctdb_data.dsize - sizeof(struct ctdb_ltdb_header);
1295
1296 data->dptr = (uint8_t *)talloc_memdup(
1297 mem_ctx, ctdb_data.dptr+sizeof(struct ctdb_ltdb_header),
1298 data->dsize);
1299
1300 SAFE_FREE(ctdb_data.dptr);
1301
1302 if (data->dptr == NULL) {
1303 return NT_STATUS_NO_MEMORY;
1304 }
1305 return NT_STATUS_OK;
1306 }
1307
1308 SAFE_FREE(ctdb_data.dptr);
1309
1310 /*
1311 * We weren't able to get it locally - ask ctdb to fetch it for us.
1312 * If we already had *something*, it's probably worth making a local
1313 * read-only copy.
1314 */
1315 status = ctdbd_fetch(messaging_ctdbd_connection(), ctx->db_id, key,
1316 mem_ctx, data,
1317 ctdb_data.dsize >= sizeof(struct ctdb_ltdb_header));
1318 if (!NT_STATUS_IS_OK(status)) {
1319 DEBUG(5, ("ctdbd_fetch failed: %s\n", nt_errstr(status)));
1320 }
1321
1322 return status;
1323 }
1324
1325 static NTSTATUS db_ctdb_parse_record(struct db_context *db, TDB_DATA key,
1326 void (*parser)(TDB_DATA key,
1327 TDB_DATA data,
1328 void *private_data),
1329 void *private_data)
1330 {
1331 NTSTATUS status;
1332 TDB_DATA data;
1333
1334 status = db_ctdb_fetch(db, talloc_tos(), key, &data);
1335 if (!NT_STATUS_IS_OK(status)) {
1336 return status;
1337 }
1338 parser(key, data, private_data);
1339 TALLOC_FREE(data.dptr);
1340 return NT_STATUS_OK;
1341 }
1342
1343 struct traverse_state {
1344 struct db_context *db;
1345 int (*fn)(struct db_record *rec, void *private_data);
1346 void *private_data;
1347 int count;
1348 };
1349
1350 static void traverse_callback(TDB_DATA key, TDB_DATA data, void *private_data)
1351 {
1352 struct traverse_state *state = (struct traverse_state *)private_data;
1353 struct db_record *rec;
1354 TALLOC_CTX *tmp_ctx = talloc_new(state->db);
1355 /* we have to give them a locked record to prevent races */
1356 rec = db_ctdb_fetch_locked(state->db, tmp_ctx, key);
1357 if (rec && rec->value.dsize > 0) {
1358 state->fn(rec, state->private_data);
1359 }
1360 talloc_free(tmp_ctx);
1361 }
1362
1363 static int traverse_persistent_callback(TDB_CONTEXT *tdb, TDB_DATA kbuf, TDB_DATA dbuf,
1364 void *private_data)
1365 {
1366 struct traverse_state *state = (struct traverse_state *)private_data;
1367 struct db_record *rec;
1368 TALLOC_CTX *tmp_ctx = talloc_new(state->db);
1369 int ret = 0;
1370
1371 /*
1372 * Skip the __db_sequence_number__ key:
1373 * This is used for persistent transactions internally.
1374 */
1375 if (kbuf.dsize == strlen(CTDB_DB_SEQNUM_KEY) + 1 &&
1376 strcmp((const char*)kbuf.dptr, CTDB_DB_SEQNUM_KEY) == 0)
1377 {
1378 goto done;
1379 }
1380
1381 /* we have to give them a locked record to prevent races */
1382 rec = db_ctdb_fetch_locked(state->db, tmp_ctx, kbuf);
1383 if (rec && rec->value.dsize > 0) {
1384 ret = state->fn(rec, state->private_data);
1385 }
1386
1387 done:
1388 talloc_free(tmp_ctx);
1389 return ret;
1390 }
1391
1392 /* wrapper to use traverse_persistent_callback with dbwrap */
1393 static int traverse_persistent_callback_dbwrap(struct db_record *rec, void* data)
1394 {
1395 return traverse_persistent_callback(NULL, rec->key, rec->value, data);
1396 }
1397
1398
1399 static int db_ctdb_traverse(struct db_context *db,
1400 int (*fn)(struct db_record *rec,
1401 void *private_data),
1402 void *private_data)
1403 {
1404 NTSTATUS status;
1405 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1406 struct db_ctdb_ctx);
1407 struct traverse_state state;
1408
1409 state.db = db;
1410 state.fn = fn;
1411 state.private_data = private_data;
1412 state.count = 0;
1413
1414 if (db->persistent) {
1415 struct tdb_context *ltdb = ctx->wtdb->tdb;
1416 int ret;
1417
1418 /* for persistent databases we don't need to do a ctdb traverse,
1419 we can do a faster local traverse */
1420 ret = tdb_traverse(ltdb, traverse_persistent_callback, &state);
1421 if (ret < 0) {
1422 return ret;
1423 }
1424 if (ctx->transaction && ctx->transaction->m_write) {
1425 /*
1426 * we now have to handle keys not yet
1427 * present at transaction start
1428 */
1429 struct db_context *newkeys = db_open_rbt(talloc_tos());
1430 struct ctdb_marshall_buffer *mbuf = ctx->transaction->m_write;
1431 struct ctdb_rec_data *rec=NULL;
1432 int i;
1433 int count = 0;
1434
1435 if (newkeys == NULL) {
1436 return -1;
1437 }
1438
1439 for (i=0; i<mbuf->count; i++) {
1440 TDB_DATA key;
1441 rec =db_ctdb_marshall_loop_next(mbuf, rec,
1442 NULL, NULL,
1443 &key, NULL);
1444 SMB_ASSERT(rec != NULL);
1445
1446 if (!tdb_exists(ltdb, key)) {
1447 dbwrap_store(newkeys, key, tdb_null, 0);
1448 }
1449 }
1450 status = dbwrap_traverse(newkeys,
1451 traverse_persistent_callback_dbwrap,
1452 &state,
1453 &count);
1454 talloc_free(newkeys);
1455 if (!NT_STATUS_IS_OK(status)) {
1456 return -1;
1457 }
1458 ret += count;
1459 }
1460 return ret;
1461 }
1462
1463 status = ctdbd_traverse(ctx->db_id, traverse_callback, &state);
1464 if (!NT_STATUS_IS_OK(status)) {
1465 return -1;
1466 }
1467 return state.count;
1468 }
1469
1470 static NTSTATUS db_ctdb_store_deny(struct db_record *rec, TDB_DATA data, int flag)
1471 {
1472 return NT_STATUS_MEDIA_WRITE_PROTECTED;
1473 }
1474
1475 static NTSTATUS db_ctdb_delete_deny(struct db_record *rec)
1476 {
1477 return NT_STATUS_MEDIA_WRITE_PROTECTED;
1478 }
1479
1480 static void traverse_read_callback(TDB_DATA key, TDB_DATA data, void *private_data)
1481 {
1482 struct traverse_state *state = (struct traverse_state *)private_data;
1483 struct db_record rec;
1484 rec.key = key;
1485 rec.value = data;
1486 rec.store = db_ctdb_store_deny;
1487 rec.delete_rec = db_ctdb_delete_deny;
1488 rec.private_data = state->db;
1489 state->fn(&rec, state->private_data);
1490 state->count++;
1491 }
1492
1493 static int traverse_persistent_callback_read(TDB_CONTEXT *tdb, TDB_DATA kbuf, TDB_DATA dbuf,
1494 void *private_data)
1495 {
1496 struct traverse_state *state = (struct traverse_state *)private_data;
1497 struct db_record rec;
1498
1499 /*
1500 * Skip the __db_sequence_number__ key:
1501 * This is used for persistent transactions internally.
1502 */
1503 if (kbuf.dsize == strlen(CTDB_DB_SEQNUM_KEY) + 1 &&
1504 strcmp((const char*)kbuf.dptr, CTDB_DB_SEQNUM_KEY) == 0)
1505 {
1506 return 0;
1507 }
1508
1509 rec.key = kbuf;
1510 rec.value = dbuf;
1511 rec.store = db_ctdb_store_deny;
1512 rec.delete_rec = db_ctdb_delete_deny;
1513 rec.private_data = state->db;
1514
1515 if (rec.value.dsize <= sizeof(struct ctdb_ltdb_header)) {
1516 /* a deleted record */
1517 return 0;
1518 }
1519 rec.value.dsize -= sizeof(struct ctdb_ltdb_header);
1520 rec.value.dptr += sizeof(struct ctdb_ltdb_header);
1521
1522 state->count++;
1523 return state->fn(&rec, state->private_data);
1524 }
1525
1526 static int db_ctdb_traverse_read(struct db_context *db,
1527 int (*fn)(struct db_record *rec,
1528 void *private_data),
1529 void *private_data)
1530 {
1531 NTSTATUS status;
1532 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1533 struct db_ctdb_ctx);
1534 struct traverse_state state;
1535
1536 state.db = db;
1537 state.fn = fn;
1538 state.private_data = private_data;
1539 state.count = 0;
1540
1541 if (db->persistent) {
1542 /* for persistent databases we don't need to do a ctdb traverse,
1543 we can do a faster local traverse */
1544 return tdb_traverse_read(ctx->wtdb->tdb, traverse_persistent_callback_read, &state);
1545 }
1546
1547 status = ctdbd_traverse(ctx->db_id, traverse_read_callback, &state);
1548 if (!NT_STATUS_IS_OK(status)) {
1549 return -1;
1550 }
1551 return state.count;
1552 }
1553
1554 static int db_ctdb_get_seqnum(struct db_context *db)
1555 {
1556 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1557 struct db_ctdb_ctx);
1558 return tdb_get_seqnum(ctx->wtdb->tdb);
1559 }
1560
1561 static void db_ctdb_id(struct db_context *db, const uint8_t **id,
1562 size_t *idlen)
1563 {
1564 struct db_ctdb_ctx *ctx = talloc_get_type_abort(
1565 db->private_data, struct db_ctdb_ctx);
1566
1567 *id = (uint8_t *)&ctx->db_id;
1568 *idlen = sizeof(ctx->db_id);
1569 }
1570
1571 struct db_context *db_open_ctdb(TALLOC_CTX *mem_ctx,
1572 const char *name,
1573 int hash_size, int tdb_flags,
1574 int open_flags, mode_t mode,
1575 enum dbwrap_lock_order lock_order)
1576 {
1577 struct db_context *result;
1578 struct db_ctdb_ctx *db_ctdb;
1579 char *db_path;
1580 struct ctdbd_connection *conn;
1581 struct loadparm_context *lp_ctx;
1582 struct ctdb_db_priority prio;
1583 NTSTATUS status;
1584 int cstatus;
1585
1586 if (!lp_clustering()) {
1587 DEBUG(10, ("Clustering disabled -- no ctdb\n"));
1588 return NULL;
1589 }
1590
1591 if (!(result = talloc_zero(mem_ctx, struct db_context))) {
1592 DEBUG(0, ("talloc failed\n"));
1593 TALLOC_FREE(result);
1594 return NULL;
1595 }
1596
1597 if (!(db_ctdb = talloc(result, struct db_ctdb_ctx))) {
1598 DEBUG(0, ("talloc failed\n"));
1599 TALLOC_FREE(result);
1600 return NULL;
1601 }
1602
1603 db_ctdb->transaction = NULL;
1604 db_ctdb->db = result;
1605
1606 conn = messaging_ctdbd_connection();
1607 if (conn == NULL) {
1608 DEBUG(1, ("Could not connect to ctdb\n"));
1609 TALLOC_FREE(result);
1610 return NULL;
1611 }
1612
1613 if (!NT_STATUS_IS_OK(ctdbd_db_attach(conn, name, &db_ctdb->db_id, tdb_flags))) {
1614 DEBUG(0, ("ctdbd_db_attach failed for %s\n", name));
1615 TALLOC_FREE(result);
1616 return NULL;
1617 }
1618
1619 db_path = ctdbd_dbpath(conn, db_ctdb, db_ctdb->db_id);
1620
1621 result->persistent = ((tdb_flags & TDB_CLEAR_IF_FIRST) == 0);
1622 result->lock_order = lock_order;
1623
1624 /* only pass through specific flags */
1625 tdb_flags &= TDB_SEQNUM;
1626
1627 /* honor permissions if user has specified O_CREAT */
1628 if (open_flags & O_CREAT) {
1629 chmod(db_path, mode);
1630 }
1631
1632 prio.db_id = db_ctdb->db_id;
1633 prio.priority = lock_order;
1634
1635 status = ctdbd_control_local(
1636 conn, CTDB_CONTROL_SET_DB_PRIORITY, 0, 0,
1637 make_tdb_data((uint8_t *)&prio, sizeof(prio)),
1638 NULL, NULL, &cstatus);
1639
1640 if (!NT_STATUS_IS_OK(status) || (cstatus != 0)) {
1641 DEBUG(1, ("CTDB_CONTROL_SET_DB_PRIORITY failed: %s, %d\n",
1642 nt_errstr(status), cstatus));
1643 TALLOC_FREE(result);
1644 return NULL;
1645 }
1646
1647 lp_ctx = loadparm_init_s3(db_path, loadparm_s3_helpers());
1648
1649 db_ctdb->wtdb = tdb_wrap_open(db_ctdb, db_path, hash_size, tdb_flags,
1650 O_RDWR, 0, lp_ctx);
1651 talloc_unlink(db_path, lp_ctx);
1652 if (db_ctdb->wtdb == NULL) {
1653 DEBUG(0, ("Could not open tdb %s: %s\n", db_path, strerror(errno)));
1654 TALLOC_FREE(result);
1655 return NULL;
1656 }
1657 talloc_free(db_path);
1658
1659 if (result->persistent) {
1660 db_ctdb->lock_ctx = g_lock_ctx_init(db_ctdb,
1661 ctdb_conn_msg_ctx(conn));
1662 if (db_ctdb->lock_ctx == NULL) {
1663 DEBUG(0, ("g_lock_ctx_init failed\n"));
1664 TALLOC_FREE(result);
1665 return NULL;
1666 }
1667 }
1668
1669 result->private_data = (void *)db_ctdb;
1670 result->fetch_locked = db_ctdb_fetch_locked;
1671 result->try_fetch_locked = db_ctdb_try_fetch_locked;
1672 result->parse_record = db_ctdb_parse_record;
1673 result->traverse = db_ctdb_traverse;
1674 result->traverse_read = db_ctdb_traverse_read;
1675 result->get_seqnum = db_ctdb_get_seqnum;
1676 result->transaction_start = db_ctdb_transaction_start;
1677 result->transaction_commit = db_ctdb_transaction_commit;
1678 result->transaction_cancel = db_ctdb_transaction_cancel;
1679 result->id = db_ctdb_id;
1680 result->stored_callback = NULL;
1681
1682 DEBUG(3,("db_open_ctdb: opened database '%s' with dbid 0x%x\n",
1683 name, db_ctdb->db_id));
1684
1685 return result;
1686 }
1687
1688 #else /* CLUSTER_SUPPORT */
1689
1690 struct db_context *db_open_ctdb(TALLOC_CTX *mem_ctx,
1691 const char *name,
1692 int hash_size, int tdb_flags,
1693 int open_flags, mode_t mode,
1694 enum dbwrap_lock_order lock_order)
1695 {
1696 DEBUG(3, ("db_open_ctdb: no cluster support!\n"));
1697 errno = ENOSYS;
1698 return NULL;
1699 }
1700
1701 #endif
reg import