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