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