source3/lib/dbwrap/dbwrap_ctdb.c

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