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