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