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                 result->value_valid = true;
 546                 return result;
 547         }
 548
 549         result->value.dsize = ctdb_data.dsize - sizeof(struct ctdb_ltdb_header);
 550         result->value.dptr = NULL;
 551
 552         if ((result->value.dsize != 0)
 553             && !(result->value.dptr = (uint8_t *)talloc_memdup(
 554                          result, ctdb_data.dptr + sizeof(struct ctdb_ltdb_header),
 555                          result->value.dsize))) {
 556                 DEBUG(0, ("talloc failed\n"));
 557                 TALLOC_FREE(result);
 558                 return NULL;
 559         }
 560         result->value_valid = true;
 561
 562         SAFE_FREE(ctdb_data.dptr);
 563
 564         return result;
 565 }
 566
 567 static int db_ctdb_record_destructor(struct db_record **recp)
 568 {
 569         struct db_record *rec = talloc_get_type_abort(*recp, struct db_record);
 570         struct db_ctdb_transaction_handle *h = talloc_get_type_abort(
 571                 rec->private_data, struct db_ctdb_transaction_handle);
 572         int ret = h->ctx->db->transaction_commit(h->ctx->db);
 573         if (ret != 0) {
 574                 DEBUG(0,(__location__ " transaction_commit failed\n"));
 575         }
 576         return 0;
 577 }
 578
 579 /*
 580   auto-create a transaction for persistent databases
 581  */
 582 static struct db_record *db_ctdb_fetch_locked_persistent(struct db_ctdb_ctx *ctx,
 583                                                          TALLOC_CTX *mem_ctx,
 584                                                          TDB_DATA key)
 585 {
 586         int res;
 587         struct db_record *rec, **recp;
 588
 589         res = db_ctdb_transaction_start(ctx->db);
 590         if (res == -1) {
 591                 return NULL;
 592         }
 593
 594         rec = db_ctdb_fetch_locked_transaction(ctx, mem_ctx, key);
 595         if (rec == NULL) {
 596                 ctx->db->transaction_cancel(ctx->db);
 597                 return NULL;
 598         }
 599
 600         /* destroy this transaction when we release the lock */
 601         recp = talloc(rec, struct db_record *);
 602         if (recp == NULL) {
 603                 ctx->db->transaction_cancel(ctx->db);
 604                 talloc_free(rec);
 605                 return NULL;
 606         }
 607         *recp = rec;
 608         talloc_set_destructor(recp, db_ctdb_record_destructor);
 609         return rec;
 610 }
 611
 612
 613 /*
 614   stores a record inside a transaction
 615  */
 616 static NTSTATUS db_ctdb_transaction_store(struct db_ctdb_transaction_handle *h,
 617                                           TDB_DATA key, TDB_DATA data)
 618 {
 619         TALLOC_CTX *tmp_ctx = talloc_new(h);
 620         TDB_DATA rec;
 621         struct ctdb_ltdb_header header;
 622
 623         ZERO_STRUCT(header);
 624
 625         /* we need the header so we can update the RSN */
 626
 627         if (!pull_newest_from_marshall_buffer(h->m_write, key, &header,
 628                                               NULL, NULL)) {
 629
 630                 rec = tdb_fetch(h->ctx->wtdb->tdb, key);
 631
 632                 if (rec.dptr != NULL) {
 633                         memcpy(&header, rec.dptr,
 634                                sizeof(struct ctdb_ltdb_header));
 635                         rec.dsize -= sizeof(struct ctdb_ltdb_header);
 636
 637                         /*
 638                          * a special case, we are writing the same
 639                          * data that is there now
 640                          */
 641                         if (data.dsize == rec.dsize &&
 642                             memcmp(data.dptr,
 643                                    rec.dptr + sizeof(struct ctdb_ltdb_header),
 644                                    data.dsize) == 0) {
 645                                 SAFE_FREE(rec.dptr);
 646                                 talloc_free(tmp_ctx);
 647                                 return NT_STATUS_OK;
 648                         }
 649                 }
 650                 SAFE_FREE(rec.dptr);
 651         }
 652
 653         header.dmaster = get_my_vnn();
 654         header.rsn++;
 655
 656         h->m_write = db_ctdb_marshall_add(h, h->m_write, h->ctx->db_id, 0, key, &header, data);
 657         if (h->m_write == NULL) {
 658                 DEBUG(0,(__location__ " Failed to add to marshalling record\n"));
 659                 talloc_free(tmp_ctx);
 660                 return NT_STATUS_NO_MEMORY;
 661         }
 662
 663         talloc_free(tmp_ctx);
 664         return NT_STATUS_OK;
 665 }
 666
 667
 668 /*
 669    a record store inside a transaction
 670  */
 671 static NTSTATUS db_ctdb_storev_transaction(
 672         struct db_record *rec, const TDB_DATA *dbufs, int num_dbufs, int flag)
 673 {
 674         struct db_ctdb_transaction_handle *h = talloc_get_type_abort(
 675                 rec->private_data, struct db_ctdb_transaction_handle);
 676         NTSTATUS status;
 677         TDB_DATA data;
 678
 679         data = dbwrap_merge_dbufs(rec, dbufs, num_dbufs);
 680         if (data.dptr == NULL) {
 681                 return NT_STATUS_NO_MEMORY;
 682         }
 683
 684         status = db_ctdb_transaction_store(h, rec->key, data);
 685
 686         TALLOC_FREE(data.dptr);
 687
 688         return status;
 689 }
 690
 691 /*
 692    a record delete inside a transaction
 693  */
 694 static NTSTATUS db_ctdb_delete_transaction(struct db_record *rec)
 695 {
 696         struct db_ctdb_transaction_handle *h = talloc_get_type_abort(
 697                 rec->private_data, struct db_ctdb_transaction_handle);
 698         NTSTATUS status;
 699
 700         status =  db_ctdb_transaction_store(h, rec->key, tdb_null);
 701         return status;
 702 }
 703
 704 static void db_ctdb_fetch_db_seqnum_parser(
 705         TDB_DATA key, struct ctdb_ltdb_header *header,
 706         TDB_DATA data, void *private_data)
 707 {
 708         uint64_t *seqnum = (uint64_t *)private_data;
 709
 710         if (data.dsize != sizeof(uint64_t)) {
 711                 *seqnum = 0;
 712                 return;
 713         }
 714         memcpy(seqnum, data.dptr, sizeof(*seqnum));
 715 }
 716
 717 /**
 718  * Fetch the db sequence number of a persistent db directly from the db.
 719  */
 720 static NTSTATUS db_ctdb_fetch_db_seqnum_from_db(struct db_ctdb_ctx *db,
 721                                                 uint64_t *seqnum)
 722 {
 723         NTSTATUS status;
 724         TDB_DATA key;
 725
 726         if (seqnum == NULL) {
 727                 return NT_STATUS_INVALID_PARAMETER;
 728         }
 729
 730         key = string_term_tdb_data(CTDB_DB_SEQNUM_KEY);
 731
 732         status = db_ctdb_ltdb_parse(
 733                 db, key, db_ctdb_fetch_db_seqnum_parser, seqnum);
 734
 735         if (NT_STATUS_IS_OK(status)) {
 736                 return NT_STATUS_OK;
 737         }
 738         if (NT_STATUS_EQUAL(status, NT_STATUS_NOT_FOUND)) {
 739                 *seqnum = 0;
 740                 return NT_STATUS_OK;
 741         }
 742         return status;
 743 }
 744
 745 /**
 746  * Store the database sequence number inside a transaction.
 747  */
 748 static NTSTATUS db_ctdb_store_db_seqnum(struct db_ctdb_transaction_handle *h,
 749                                         uint64_t seqnum)
 750 {
 751         NTSTATUS status;
 752         TDB_DATA key = string_term_tdb_data(CTDB_DB_SEQNUM_KEY);
 753         TDB_DATA data = { .dptr=(uint8_t *)&seqnum, .dsize=sizeof(seqnum) };
 754
 755         status = db_ctdb_transaction_store(h, key, data);
 756
 757         return status;
 758 }
 759
 760 /*
 761   commit a transaction
 762  */
 763 static int db_ctdb_transaction_commit(struct db_context *db)
 764 {
 765         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
 766                                                         struct db_ctdb_ctx);
 767         NTSTATUS rets;
 768         int32_t status;
 769         struct db_ctdb_transaction_handle *h = ctx->transaction;
 770         uint64_t old_seqnum, new_seqnum;
 771         int ret;
 772
 773         if (h == NULL) {
 774                 DEBUG(0,(__location__ " transaction commit with no open transaction on db 0x%08x\n", ctx->db_id));
 775                 return -1;
 776         }
 777
 778         if (h->nested_cancel) {
 779                 db->transaction_cancel(db);
 780                 DEBUG(5,(__location__ " Failed transaction commit after nested cancel\n"));
 781                 return -1;
 782         }
 783
 784         if (h->nesting != 0) {
 785                 h->nesting--;
 786                 DEBUG(5, (__location__ " transaction commit on db 0x%08x: nesting %d -> %d\n",
 787                           ctx->db_id, ctx->transaction->nesting + 1, ctx->transaction->nesting));
 788                 return 0;
 789         }
 790
 791         if (h->m_write == NULL) {
 792                 /*
 793                  * No changes were made, so don't change the seqnum,
 794                  * don't push to other node, just exit with success.
 795                  */
 796                 ret = 0;
 797                 goto done;
 798         }
 799
 800         DEBUG(5,(__location__ " transaction commit on db 0x%08x\n", ctx->db_id));
 801
 802         /*
 803          * As the last db action before committing, bump the database sequence
 804          * number. Note that this undoes all changes to the seqnum records
 805          * performed under the transaction. This record is not meant to be
 806          * modified by user interaction. It is for internal use only...
 807          */
 808         rets = db_ctdb_fetch_db_seqnum_from_db(ctx, &old_seqnum);
 809         if (!NT_STATUS_IS_OK(rets)) {
 810                 DEBUG(1, (__location__ " failed to fetch the db sequence number "
 811                           "in transaction commit on db 0x%08x\n", ctx->db_id));
 812                 ret = -1;
 813                 goto done;
 814         }
 815
 816         new_seqnum = old_seqnum + 1;
 817
 818         rets = db_ctdb_store_db_seqnum(h, new_seqnum);
 819         if (!NT_STATUS_IS_OK(rets)) {
 820                 DEBUG(1, (__location__ "failed to store the db sequence number "
 821                           " in transaction commit on db 0x%08x\n", ctx->db_id));
 822                 ret = -1;
 823                 goto done;
 824         }
 825
 826 again:
 827         /* tell ctdbd to commit to the other nodes */
 828         ret = ctdbd_control_local(messaging_ctdb_connection(),
 829                                   CTDB_CONTROL_TRANS3_COMMIT,
 830                                   h->ctx->db_id, 0,
 831                                   db_ctdb_marshall_finish(h->m_write),
 832                                   NULL, NULL, &status);
 833         if ((ret != 0) || status != 0) {
 834                 /*
 835                  * The TRANS3_COMMIT control should only possibly fail when a
 836                  * recovery has been running concurrently. In any case, the db
 837                  * will be the same on all nodes, either the new copy or the
 838                  * old copy.  This can be detected by comparing the old and new
 839                  * local sequence numbers.
 840                  */
 841                 rets = db_ctdb_fetch_db_seqnum_from_db(ctx, &new_seqnum);
 842                 if (!NT_STATUS_IS_OK(rets)) {
 843                         DEBUG(1, (__location__ " failed to refetch db sequence "
 844                                   "number after failed TRANS3_COMMIT\n"));
 845                         ret = -1;
 846                         goto done;
 847                 }
 848
 849                 if (new_seqnum == old_seqnum) {
 850                         /* Recovery prevented all our changes: retry. */
 851                         goto again;
 852                 }
 853                 if (new_seqnum != (old_seqnum + 1)) {
 854                         DEBUG(0, (__location__ " ERROR: new_seqnum[%lu] != "
 855                                   "old_seqnum[%lu] + (0 or 1) after failed "
 856                                   "TRANS3_COMMIT - this should not happen!\n",
 857                                   (unsigned long)new_seqnum,
 858                                   (unsigned long)old_seqnum));
 859                         ret = -1;
 860                         goto done;
 861                 }
 862                 /*
 863                  * Recovery propagated our changes to all nodes, completing
 864                  * our commit for us - succeed.
 865                  */
 866         }
 867
 868         ret = 0;
 869
 870 done:
 871         h->ctx->transaction = NULL;
 872         talloc_free(h);
 873         return ret;
 874 }
 875
 876
 877 /*
 878   cancel a transaction
 879  */
 880 static int db_ctdb_transaction_cancel(struct db_context *db)
 881 {
 882         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
 883                                                         struct db_ctdb_ctx);
 884         struct db_ctdb_transaction_handle *h = ctx->transaction;
 885
 886         if (h == NULL) {
 887                 DEBUG(0,(__location__ " transaction cancel with no open transaction on db 0x%08x\n", ctx->db_id));
 888                 return -1;
 889         }
 890
 891         if (h->nesting != 0) {
 892                 h->nesting--;
 893                 h->nested_cancel = true;
 894                 DEBUG(5, (__location__ " transaction cancel on db 0x%08x: nesting %d -> %d\n",
 895                           ctx->db_id, ctx->transaction->nesting + 1, ctx->transaction->nesting));
 896                 return 0;
 897         }
 898
 899         DEBUG(5,(__location__ " Cancel transaction on db 0x%08x\n", ctx->db_id));
 900
 901         ctx->transaction = NULL;
 902         talloc_free(h);
 903         return 0;
 904 }
 905
 906
 907 static NTSTATUS db_ctdb_storev(struct db_record *rec,
 908                                const TDB_DATA *dbufs, int num_dbufs, int flag)
 909 {
 910         struct db_ctdb_rec *crec = talloc_get_type_abort(
 911                 rec->private_data, struct db_ctdb_rec);
 912         NTSTATUS status;
 913
 914         status = db_ctdb_ltdb_store(crec->ctdb_ctx, rec->key, &(crec->header),
 915                                     dbufs, num_dbufs);
 916         return status;
 917 }
 918
 919
 920
 921 static NTSTATUS db_ctdb_send_schedule_for_deletion(struct db_record *rec)
 922 {
 923         NTSTATUS status = NT_STATUS_OK;
 924         int ret;
 925         struct ctdb_control_schedule_for_deletion *dd;
 926         TDB_DATA indata;
 927         int32_t cstatus;
 928         struct db_ctdb_rec *crec = talloc_get_type_abort(
 929                 rec->private_data, struct db_ctdb_rec);
 930         struct db_ctdb_ctx *ctx = crec->ctdb_ctx;
 931
 932         indata.dsize = offsetof(struct ctdb_control_schedule_for_deletion, key) + rec->key.dsize;
 933         indata.dptr = talloc_zero_array(crec, uint8_t, indata.dsize);
 934         if (indata.dptr == NULL) {
 935                 DEBUG(0, (__location__ " talloc failed!\n"));
 936                 return NT_STATUS_NO_MEMORY;
 937         }
 938
 939         dd = (struct ctdb_control_schedule_for_deletion *)(void *)indata.dptr;
 940         dd->db_id = ctx->db_id;
 941         dd->hdr = crec->header;
 942         dd->keylen = rec->key.dsize;
 943         memcpy(dd->key, rec->key.dptr, rec->key.dsize);
 944
 945         ret = ctdbd_control_local(messaging_ctdb_connection(),
 946                                   CTDB_CONTROL_SCHEDULE_FOR_DELETION,
 947                                   crec->ctdb_ctx->db_id,
 948                                   CTDB_CTRL_FLAG_NOREPLY, /* flags */
 949                                   indata,
 950                                   NULL, /* mem_ctx */
 951                                   NULL, /* outdata */
 952                                   &cstatus);
 953         talloc_free(indata.dptr);
 954
 955         if ((ret != 0) || cstatus != 0) {
 956                 DEBUG(1, (__location__ " Error sending local control "
 957                           "SCHEDULE_FOR_DELETION: %s, cstatus = %"PRIi32"\n",
 958                           strerror(ret), cstatus));
 959                 if (ret != 0) {
 960                         status = map_nt_error_from_unix(ret);
 961                 } else {
 962                         status = NT_STATUS_UNSUCCESSFUL;
 963                 }
 964         }
 965
 966         return status;
 967 }
 968
 969 static NTSTATUS db_ctdb_delete(struct db_record *rec)
 970 {
 971         NTSTATUS status;
 972
 973         /*
 974          * We have to store the header with empty data. TODO: Fix the
 975          * tdb-level cleanup
 976          */
 977
 978         status = db_ctdb_storev(rec, &tdb_null, 1, 0);
 979         if (!NT_STATUS_IS_OK(status)) {
 980                 return status;
 981         }
 982
 983         status = db_ctdb_send_schedule_for_deletion(rec);
 984         return status;
 985 }
 986
 987 static int db_ctdb_record_destr(struct db_record* data)
 988 {
 989         struct db_ctdb_rec *crec = talloc_get_type_abort(
 990                 data->private_data, struct db_ctdb_rec);
 991         int threshold;
 992         int ret;
 993         struct timeval before;
 994         double timediff;
 995
 996         DEBUG(10, (DEBUGLEVEL > 10
 997                    ? "Unlocking db %u key %s\n"
 998                    : "Unlocking db %u key %.20s\n",
 999                    (int)crec->ctdb_ctx->db_id,
1000                    hex_encode_talloc(data, (unsigned char *)data->key.dptr,
1001                               data->key.dsize)));
1002
1003         before = timeval_current();
1004
1005         ret = tdb_chainunlock(crec->ctdb_ctx->wtdb->tdb, data->key);
1006
1007         timediff = timeval_elapsed(&before);
1008         timediff *= 1000;       /* get us milliseconds */
1009
1010         if (timediff > crec->ctdb_ctx->warn_unlock_msecs) {
1011                 char *key;
1012                 key = hex_encode_talloc(talloc_tos(),
1013                                         (unsigned char *)data->key.dptr,
1014                                         data->key.dsize);
1015                 DEBUG(0, ("tdb_chainunlock on db %s, key %s took %f milliseconds\n",
1016                           tdb_name(crec->ctdb_ctx->wtdb->tdb), key,
1017                           timediff));
1018                 TALLOC_FREE(key);
1019         }
1020
1021         if (ret != 0) {
1022                 DEBUG(0, ("tdb_chainunlock failed\n"));
1023                 return -1;
1024         }
1025
1026         threshold = crec->ctdb_ctx->warn_locktime_msecs;
1027         if (threshold != 0) {
1028                 timediff = timeval_elapsed(&crec->lock_time) * 1000;
1029                 if (timediff > threshold) {
1030                         const char *key;
1031
1032                         key = hex_encode_talloc(data,
1033                                                 (unsigned char *)data->key.dptr,
1034                                                 data->key.dsize);
1035                         DEBUG(0, ("Held tdb lock on db %s, key %s "
1036                                   "%f milliseconds\n",
1037                                   tdb_name(crec->ctdb_ctx->wtdb->tdb),
1038                                   key, timediff));
1039                 }
1040         }
1041
1042         return 0;
1043 }
1044
1045 /**
1046  * Check whether we have a valid local copy of the given record,
1047  * either for reading or for writing.
1048  */
1049 static bool db_ctdb_can_use_local_hdr(const struct ctdb_ltdb_header *hdr,
1050                                       uint32_t my_vnn, bool read_only)
1051 {
1052         if (hdr->dmaster != my_vnn) {
1053                 /* If we're not dmaster, it must be r/o copy. */
1054                 return read_only && (hdr->flags & CTDB_REC_RO_HAVE_READONLY);
1055         }
1056
1057         /*
1058          * If we want write access, no one may have r/o copies.
1059          */
1060         return read_only || !(hdr->flags & CTDB_REC_RO_HAVE_DELEGATIONS);
1061 }
1062
1063 static bool db_ctdb_can_use_local_copy(TDB_DATA ctdb_data, uint32_t my_vnn,
1064                                        bool read_only)
1065 {
1066         if (ctdb_data.dptr == NULL) {
1067                 return false;
1068         }
1069
1070         if (ctdb_data.dsize < sizeof(struct ctdb_ltdb_header)) {
1071                 return false;
1072         }
1073
1074         return db_ctdb_can_use_local_hdr(
1075                 (struct ctdb_ltdb_header *)ctdb_data.dptr, my_vnn, read_only);
1076 }
1077
1078 static struct db_record *fetch_locked_internal(struct db_ctdb_ctx *ctx,
1079                                                TALLOC_CTX *mem_ctx,
1080                                                TDB_DATA key,
1081                                                bool tryonly)
1082 {
1083         struct db_record *result;
1084         struct db_ctdb_rec *crec;
1085         TDB_DATA ctdb_data;
1086         int migrate_attempts;
1087         struct timeval migrate_start;
1088         struct timeval chainlock_start;
1089         struct timeval ctdb_start_time;
1090         double chainlock_time = 0;
1091         double ctdb_time = 0;
1092         int duration_msecs;
1093         int lockret;
1094         int ret;
1095
1096         if (!(result = talloc(mem_ctx, struct db_record))) {
1097                 DEBUG(0, ("talloc failed\n"));
1098                 return NULL;
1099         }
1100
1101         if (!(crec = talloc_zero(result, struct db_ctdb_rec))) {
1102                 DEBUG(0, ("talloc failed\n"));
1103                 TALLOC_FREE(result);
1104                 return NULL;
1105         }
1106
1107         result->db = ctx->db;
1108         result->private_data = (void *)crec;
1109         crec->ctdb_ctx = ctx;
1110
1111         result->key.dsize = key.dsize;
1112         result->key.dptr = (uint8_t *)talloc_memdup(result, key.dptr,
1113                                                     key.dsize);
1114         if (result->key.dptr == NULL) {
1115                 DEBUG(0, ("talloc failed\n"));
1116                 TALLOC_FREE(result);
1117                 return NULL;
1118         }
1119
1120         migrate_attempts = 0;
1121         GetTimeOfDay(&migrate_start);
1122
1123         /*
1124          * Do a blocking lock on the record
1125          */
1126 again:
1127
1128         if (DEBUGLEVEL >= 10) {
1129                 char *keystr = hex_encode_talloc(result, key.dptr, key.dsize);
1130                 DEBUG(10, (DEBUGLEVEL > 10
1131                            ? "Locking db %u key %s\n"
1132                            : "Locking db %u key %.20s\n",
1133                            (int)crec->ctdb_ctx->db_id, keystr));
1134                 TALLOC_FREE(keystr);
1135         }
1136
1137         GetTimeOfDay(&chainlock_start);
1138         lockret = tryonly
1139                 ? tdb_chainlock_nonblock(ctx->wtdb->tdb, key)
1140                 : tdb_chainlock(ctx->wtdb->tdb, key);
1141         chainlock_time += timeval_elapsed(&chainlock_start);
1142
1143         if (lockret != 0) {
1144                 DEBUG(3, ("tdb_chainlock failed\n"));
1145                 TALLOC_FREE(result);
1146                 return NULL;
1147         }
1148
1149         result->storev = db_ctdb_storev;
1150         result->delete_rec = db_ctdb_delete;
1151         talloc_set_destructor(result, db_ctdb_record_destr);
1152
1153         ctdb_data = tdb_fetch(ctx->wtdb->tdb, key);
1154
1155         /*
1156          * See if we have a valid record and we are the dmaster. If so, we can
1157          * take the shortcut and just return it.
1158          */
1159
1160         if (!db_ctdb_can_use_local_copy(ctdb_data, get_my_vnn(), false)) {
1161                 SAFE_FREE(ctdb_data.dptr);
1162                 tdb_chainunlock(ctx->wtdb->tdb, key);
1163                 talloc_set_destructor(result, NULL);
1164
1165                 if (tryonly && (migrate_attempts != 0)) {
1166                         DEBUG(5, ("record migrated away again\n"));
1167                         TALLOC_FREE(result);
1168                         return NULL;
1169                 }
1170
1171                 migrate_attempts += 1;
1172
1173                 DEBUG(10, ("ctdb_data.dptr = %p, dmaster = %"PRIu32" "
1174                            "(%"PRIu32") %"PRIu32"\n",
1175                            ctdb_data.dptr, ctdb_data.dptr ?
1176                            ((struct ctdb_ltdb_header *)ctdb_data.dptr)->dmaster :
1177                            UINT32_MAX,
1178                            get_my_vnn(),
1179                            ctdb_data.dptr ?
1180                            ((struct ctdb_ltdb_header *)ctdb_data.dptr)->flags : 0));
1181
1182                 GetTimeOfDay(&ctdb_start_time);
1183                 ret = ctdbd_migrate(messaging_ctdb_connection(), ctx->db_id,
1184                                     key);
1185                 ctdb_time += timeval_elapsed(&ctdb_start_time);
1186
1187                 if (ret != 0) {
1188                         DEBUG(5, ("ctdbd_migrate failed: %s\n",
1189                                   strerror(ret)));
1190                         TALLOC_FREE(result);
1191                         return NULL;
1192                 }
1193                 /* now its migrated, try again */
1194                 goto again;
1195         }
1196
1197         {
1198                 double duration;
1199                 duration = timeval_elapsed(&migrate_start);
1200
1201                 /*
1202                  * Convert the duration to milliseconds to avoid a
1203                  * floating-point division of
1204                  * lp_parm_int("migrate_duration") by 1000.
1205                  */
1206                 duration_msecs = duration * 1000;
1207         }
1208
1209         if ((migrate_attempts > ctx->warn_migrate_attempts) ||
1210             (duration_msecs > ctx->warn_migrate_msecs)) {
1211                 int chain = 0;
1212
1213                 if (tdb_get_flags(ctx->wtdb->tdb) & TDB_INCOMPATIBLE_HASH) {
1214                         chain = tdb_jenkins_hash(&key) %
1215                                 tdb_hash_size(ctx->wtdb->tdb);
1216                 }
1217
1218                 DEBUG(0, ("db_ctdb_fetch_locked for %s key %s, chain %d "
1219                           "needed %d attempts, %d milliseconds, "
1220                           "chainlock: %f ms, CTDB %f ms\n",
1221                           tdb_name(ctx->wtdb->tdb),
1222                           hex_encode_talloc(talloc_tos(),
1223                                             (unsigned char *)key.dptr,
1224                                             key.dsize),
1225                           chain,
1226                           migrate_attempts, duration_msecs,
1227                           chainlock_time * 1000.0,
1228                           ctdb_time * 1000.0));
1229         }
1230
1231         GetTimeOfDay(&crec->lock_time);
1232
1233         memcpy(&crec->header, ctdb_data.dptr, sizeof(crec->header));
1234
1235         result->value.dsize = ctdb_data.dsize - sizeof(crec->header);
1236         result->value.dptr = NULL;
1237
1238         if (result->value.dsize != 0) {
1239                 result->value.dptr = talloc_memdup(
1240                         result, ctdb_data.dptr + sizeof(crec->header),
1241                         result->value.dsize);
1242                 if (result->value.dptr == NULL) {
1243                         DBG_ERR("talloc failed\n");
1244                         TALLOC_FREE(result);
1245                         return NULL;
1246                 }
1247         }
1248         result->value_valid = true;
1249
1250         SAFE_FREE(ctdb_data.dptr);
1251
1252         return result;
1253 }
1254
1255 static struct db_record *db_ctdb_fetch_locked(struct db_context *db,
1256                                               TALLOC_CTX *mem_ctx,
1257                                               TDB_DATA key)
1258 {
1259         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1260                                                         struct db_ctdb_ctx);
1261
1262         if (ctx->transaction != NULL) {
1263                 return db_ctdb_fetch_locked_transaction(ctx, mem_ctx, key);
1264         }
1265
1266         if (db->persistent) {
1267                 return db_ctdb_fetch_locked_persistent(ctx, mem_ctx, key);
1268         }
1269
1270         return fetch_locked_internal(ctx, mem_ctx, key, false);
1271 }
1272
1273 static struct db_record *db_ctdb_try_fetch_locked(struct db_context *db,
1274                                                   TALLOC_CTX *mem_ctx,
1275                                                   TDB_DATA key)
1276 {
1277         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1278                                                         struct db_ctdb_ctx);
1279
1280         if (ctx->transaction != NULL) {
1281                 return db_ctdb_fetch_locked_transaction(ctx, mem_ctx, key);
1282         }
1283
1284         if (db->persistent) {
1285                 return db_ctdb_fetch_locked_persistent(ctx, mem_ctx, key);
1286         }
1287
1288         return fetch_locked_internal(ctx, mem_ctx, key, true);
1289 }
1290
1291 struct db_ctdb_parse_record_state {
1292         void (*parser)(TDB_DATA key, TDB_DATA data, void *private_data);
1293         void *private_data;
1294         uint32_t my_vnn;
1295         bool ask_for_readonly_copy;
1296         bool done;
1297         bool empty_record;
1298 };
1299
1300 static void db_ctdb_parse_record_parser(
1301         TDB_DATA key, struct ctdb_ltdb_header *header,
1302         TDB_DATA data, void *private_data)
1303 {
1304         struct db_ctdb_parse_record_state *state =
1305                 (struct db_ctdb_parse_record_state *)private_data;
1306         state->parser(key, data, state->private_data);
1307 }
1308
1309 static void db_ctdb_parse_record_parser_nonpersistent(
1310         TDB_DATA key, struct ctdb_ltdb_header *header,
1311         TDB_DATA data, void *private_data)
1312 {
1313         struct db_ctdb_parse_record_state *state =
1314                 (struct db_ctdb_parse_record_state *)private_data;
1315
1316         if (db_ctdb_can_use_local_hdr(header, state->my_vnn, true)) {
1317                 /*
1318                  * A record consisting only of the ctdb header can be
1319                  * a validly created empty record or a tombstone
1320                  * record of a deleted record (not vacuumed yet). Mark
1321                  * it accordingly.
1322                  */
1323                 state->empty_record = (data.dsize == 0);
1324                 if (!state->empty_record) {
1325                         state->parser(key, data, state->private_data);
1326                 }
1327                 state->done = true;
1328         } else {
1329                 /*
1330                  * We found something in the db, so it seems that this record,
1331                  * while not usable locally right now, is popular. Ask for a
1332                  * R/O copy.
1333                  */
1334                 state->ask_for_readonly_copy = true;
1335         }
1336 }
1337
1338 static NTSTATUS db_ctdb_try_parse_local_record(struct db_ctdb_ctx *ctx,
1339                                                TDB_DATA key,
1340                                                struct db_ctdb_parse_record_state *state)
1341 {
1342         NTSTATUS status;
1343
1344         if (ctx->transaction != NULL) {
1345                 struct db_ctdb_transaction_handle *h = ctx->transaction;
1346                 bool found;
1347
1348                 /*
1349                  * Transactions only happen for persistent db's.
1350                  */
1351
1352                 found = parse_newest_in_marshall_buffer(
1353                         h->m_write, key, db_ctdb_parse_record_parser, state);
1354
1355                 if (found) {
1356                         return NT_STATUS_OK;
1357                 }
1358         }
1359
1360         if (ctx->db->persistent) {
1361                 /*
1362                  * Persistent db, but not found in the transaction buffer
1363                  */
1364                 return db_ctdb_ltdb_parse(
1365                         ctx, key, db_ctdb_parse_record_parser, state);
1366         }
1367
1368         state->done = false;
1369         state->ask_for_readonly_copy = false;
1370
1371         status = db_ctdb_ltdb_parse(
1372                 ctx, key, db_ctdb_parse_record_parser_nonpersistent, state);
1373         if (NT_STATUS_IS_OK(status) && state->done) {
1374                 if (state->empty_record) {
1375                         /*
1376                          * We know authoritatively, that this is an empty
1377                          * record. Since ctdb does not distinguish between empty
1378                          * and deleted records, this can be a record stored as
1379                          * empty or a not-yet-vacuumed tombstone record of a
1380                          * deleted record. Now Samba right now can live without
1381                          * empty records, so we can safely report this record
1382                          * as non-existing.
1383                          *
1384                          * See bugs 10008 and 12005.
1385                          */
1386                         return NT_STATUS_NOT_FOUND;
1387                 }
1388                 return NT_STATUS_OK;
1389         }
1390
1391         return NT_STATUS_MORE_PROCESSING_REQUIRED;
1392 }
1393
1394 static NTSTATUS db_ctdb_parse_record(struct db_context *db, TDB_DATA key,
1395                                      void (*parser)(TDB_DATA key,
1396                                                     TDB_DATA data,
1397                                                     void *private_data),
1398                                      void *private_data)
1399 {
1400         struct db_ctdb_ctx *ctx = talloc_get_type_abort(
1401                 db->private_data, struct db_ctdb_ctx);
1402         struct db_ctdb_parse_record_state state;
1403         NTSTATUS status;
1404         int ret;
1405
1406         state.parser = parser;
1407         state.private_data = private_data;
1408         state.my_vnn = get_my_vnn();
1409         state.empty_record = false;
1410
1411         status = db_ctdb_try_parse_local_record(ctx, key, &state);
1412         if (!NT_STATUS_EQUAL(status, NT_STATUS_MORE_PROCESSING_REQUIRED)) {
1413                 return status;
1414         }
1415
1416         ret = ctdbd_parse(messaging_ctdb_connection(), ctx->db_id, key,
1417                           state.ask_for_readonly_copy, parser, private_data);
1418         if (ret != 0) {
1419                 if (ret == ENOENT) {
1420                         /*
1421                          * This maps to
1422                          * NT_STATUS_OBJECT_NAME_NOT_FOUND. Our upper
1423                          * layers expect NT_STATUS_NOT_FOUND for "no
1424                          * record around". We need to convert dbwrap
1425                          * to 0/errno away from NTSTATUS ... :-)
1426                          */
1427                         return NT_STATUS_NOT_FOUND;
1428                 }
1429                 return map_nt_error_from_unix(ret);
1430         }
1431         return NT_STATUS_OK;
1432 }
1433
1434 static void db_ctdb_parse_record_done(struct tevent_req *subreq);
1435
1436 static struct tevent_req *db_ctdb_parse_record_send(
1437         TALLOC_CTX *mem_ctx,
1438         struct tevent_context *ev,
1439         struct db_context *db,
1440         TDB_DATA key,
1441         void (*parser)(TDB_DATA key,
1442                        TDB_DATA data,
1443                        void *private_data),
1444         void *private_data,
1445         enum dbwrap_req_state *req_state)
1446 {
1447         struct db_ctdb_ctx *ctx = talloc_get_type_abort(
1448                 db->private_data, struct db_ctdb_ctx);
1449         struct tevent_req *req = NULL;
1450         struct tevent_req *subreq = NULL;
1451         struct db_ctdb_parse_record_state *state = NULL;
1452         NTSTATUS status;
1453
1454         req = tevent_req_create(mem_ctx, &state,
1455                                 struct db_ctdb_parse_record_state);
1456         if (req == NULL) {
1457                 *req_state = DBWRAP_REQ_ERROR;
1458                 return NULL;
1459
1460         }
1461
1462         *state = (struct db_ctdb_parse_record_state) {
1463                 .parser = parser,
1464                 .private_data = private_data,
1465                 .my_vnn = get_my_vnn(),
1466                 .empty_record = false,
1467         };
1468
1469         status = db_ctdb_try_parse_local_record(ctx, key, state);
1470         if (!NT_STATUS_EQUAL(status, NT_STATUS_MORE_PROCESSING_REQUIRED)) {
1471                 if (tevent_req_nterror(req, status)) {
1472                         *req_state = DBWRAP_REQ_ERROR;
1473                         return tevent_req_post(req, ev);
1474                 }
1475                 *req_state = DBWRAP_REQ_DONE;
1476                 tevent_req_done(req);
1477                 return tevent_req_post(req, ev);
1478         }
1479
1480         subreq = ctdbd_parse_send(state,
1481                                   ev,
1482                                   ctdb_async_ctx.async_conn,
1483                                   ctx->db_id,
1484                                   key,
1485                                   state->ask_for_readonly_copy,
1486                                   parser,
1487                                   private_data,
1488                                   req_state);
1489         if (tevent_req_nomem(subreq, req)) {
1490                 *req_state = DBWRAP_REQ_ERROR;
1491                 return tevent_req_post(req, ev);
1492         }
1493         tevent_req_set_callback(subreq, db_ctdb_parse_record_done, req);
1494
1495         return req;
1496 }
1497
1498 static void db_ctdb_parse_record_done(struct tevent_req *subreq)
1499 {
1500         struct tevent_req *req = tevent_req_callback_data(
1501                 subreq, struct tevent_req);
1502         int ret;
1503
1504         ret = ctdbd_parse_recv(subreq);
1505         TALLOC_FREE(subreq);
1506         if (ret != 0) {
1507                 if (ret == ENOENT) {
1508                         /*
1509                          * This maps to NT_STATUS_OBJECT_NAME_NOT_FOUND. Our
1510                          * upper layers expect NT_STATUS_NOT_FOUND for "no
1511                          * record around". We need to convert dbwrap to 0/errno
1512                          * away from NTSTATUS ... :-)
1513                          */
1514                         tevent_req_nterror(req, NT_STATUS_NOT_FOUND);
1515                         return;
1516                 }
1517                 tevent_req_nterror(req, map_nt_error_from_unix(ret));
1518                 return;
1519         }
1520
1521         tevent_req_done(req);
1522         return;
1523 }
1524
1525 static NTSTATUS db_ctdb_parse_record_recv(struct tevent_req *req)
1526 {
1527         return tevent_req_simple_recv_ntstatus(req);
1528 }
1529
1530 struct traverse_state {
1531         struct db_context *db;
1532         int (*fn)(struct db_record *rec, void *private_data);
1533         void *private_data;
1534         int count;
1535 };
1536
1537 static void traverse_callback(TDB_DATA key, TDB_DATA data, void *private_data)
1538 {
1539         struct traverse_state *state = (struct traverse_state *)private_data;
1540         struct db_record *rec = NULL;
1541         TALLOC_CTX *tmp_ctx = NULL;
1542
1543         tmp_ctx = talloc_new(state->db);
1544         if (tmp_ctx == NULL) {
1545                 DBG_ERR("talloc_new failed\n");
1546                 return;
1547         }
1548
1549         /* we have to give them a locked record to prevent races */
1550         rec = db_ctdb_fetch_locked(state->db, tmp_ctx, key);
1551         if (rec != NULL && rec->value.dsize > 0) {
1552                 state->fn(rec, state->private_data);
1553                 state->count++;
1554         }
1555         talloc_free(tmp_ctx);
1556 }
1557
1558 static int traverse_persistent_callback(TDB_CONTEXT *tdb, TDB_DATA kbuf, TDB_DATA dbuf,
1559                                         void *private_data)
1560 {
1561         struct traverse_state *state = (struct traverse_state *)private_data;
1562         struct db_record *rec;
1563         TALLOC_CTX *tmp_ctx = talloc_new(state->db);
1564         int ret = 0;
1565
1566         /*
1567          * Skip the __db_sequence_number__ key:
1568          * This is used for persistent transactions internally.
1569          */
1570         if (kbuf.dsize == strlen(CTDB_DB_SEQNUM_KEY) + 1 &&
1571             strcmp((const char*)kbuf.dptr, CTDB_DB_SEQNUM_KEY) == 0)
1572         {
1573                 goto done;
1574         }
1575
1576         /* we have to give them a locked record to prevent races */
1577         rec = db_ctdb_fetch_locked(state->db, tmp_ctx, kbuf);
1578         if (rec && rec->value.dsize > 0) {
1579                 ret = state->fn(rec, state->private_data);
1580         }
1581
1582 done:
1583         talloc_free(tmp_ctx);
1584         return ret;
1585 }
1586
1587 /* wrapper to use traverse_persistent_callback with dbwrap */
1588 static int traverse_persistent_callback_dbwrap(struct db_record *rec, void* data)
1589 {
1590         return traverse_persistent_callback(NULL, rec->key, rec->value, data);
1591 }
1592
1593 static int db_ctdbd_traverse(uint32_t db_id,
1594                              void (*fn)(TDB_DATA key, TDB_DATA data,
1595                                         void *private_data),
1596                              void *private_data)
1597 {
1598         struct ctdbd_connection *conn;
1599         int ret;
1600
1601         become_root();
1602         ret = ctdbd_init_connection(talloc_tos(), lp_ctdbd_socket(),
1603                                     lp_ctdb_timeout(), &conn);
1604         unbecome_root();
1605         if (ret != 0) {
1606                 DBG_WARNING("ctdbd_init_connection failed: %s\n",
1607                             strerror(ret));
1608                 return ret;
1609         }
1610
1611         ret = ctdbd_traverse(conn, db_id, fn, private_data);
1612         TALLOC_FREE(conn);
1613
1614         if (ret != 0) {
1615                 DBG_WARNING("ctdbd_traverse failed: %s\n",
1616                             strerror(ret));
1617                 return ret;
1618         }
1619
1620         return 0;
1621 }
1622
1623
1624 static int db_ctdb_traverse(struct db_context *db,
1625                             int (*fn)(struct db_record *rec,
1626                                       void *private_data),
1627                             void *private_data)
1628 {
1629         int ret;
1630         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1631                                                         struct db_ctdb_ctx);
1632         struct traverse_state state;
1633
1634         state = (struct traverse_state) {
1635                 .db = db,
1636                 .fn = fn,
1637                 .private_data = private_data,
1638         };
1639
1640         if (db->persistent) {
1641                 struct tdb_context *ltdb = ctx->wtdb->tdb;
1642
1643                 /* for persistent databases we don't need to do a ctdb traverse,
1644                    we can do a faster local traverse */
1645                 ret = tdb_traverse(ltdb, traverse_persistent_callback, &state);
1646                 if (ret < 0) {
1647                         return ret;
1648                 }
1649                 if (ctx->transaction && ctx->transaction->m_write) {
1650                         /*
1651                          * we now have to handle keys not yet
1652                          * present at transaction start
1653                          */
1654                         struct db_context *newkeys = db_open_rbt(talloc_tos());
1655                         struct ctdb_marshall_buffer *mbuf = ctx->transaction->m_write;
1656                         struct ctdb_rec_data_old *rec=NULL;
1657                         uint32_t i;
1658                         int count = 0;
1659                         NTSTATUS status;
1660
1661                         if (newkeys == NULL) {
1662                                 return -1;
1663                         }
1664
1665                         for (i=0; i<mbuf->count; i++) {
1666                                 TDB_DATA key;
1667                                 rec = db_ctdb_marshall_loop_next_key(
1668                                         mbuf, rec, &key);
1669                                 SMB_ASSERT(rec != NULL);
1670
1671                                 if (!tdb_exists(ltdb, key)) {
1672                                         dbwrap_store(newkeys, key, tdb_null, 0);
1673                                 }
1674                         }
1675                         status = dbwrap_traverse(newkeys,
1676                                                  traverse_persistent_callback_dbwrap,
1677                                                  &state,
1678                                                  &count);
1679                         talloc_free(newkeys);
1680                         if (!NT_STATUS_IS_OK(status)) {
1681                                 return -1;
1682                         }
1683                         ret += count;
1684                 }
1685                 return ret;
1686         }
1687
1688         ret = db_ctdbd_traverse(ctx->db_id, traverse_callback, &state);
1689         if (ret != 0) {
1690                 return -1;
1691         }
1692         return state.count;
1693 }
1694
1695 static NTSTATUS db_ctdb_storev_deny(struct db_record *rec,
1696                                     const TDB_DATA *dbufs, int num_dbufs, int flag)
1697 {
1698         return NT_STATUS_MEDIA_WRITE_PROTECTED;
1699 }
1700
1701 static NTSTATUS db_ctdb_delete_deny(struct db_record *rec)
1702 {
1703         return NT_STATUS_MEDIA_WRITE_PROTECTED;
1704 }
1705
1706 static void traverse_read_callback(TDB_DATA key, TDB_DATA data, void *private_data)
1707 {
1708         struct traverse_state *state = (struct traverse_state *)private_data;
1709         struct db_record rec;
1710
1711         ZERO_STRUCT(rec);
1712         rec.db = state->db;
1713         rec.key = key;
1714         rec.value = data;
1715         rec.storev = db_ctdb_storev_deny;
1716         rec.delete_rec = db_ctdb_delete_deny;
1717         rec.private_data = NULL;
1718         rec.value_valid = true;
1719         state->fn(&rec, state->private_data);
1720         state->count++;
1721 }
1722
1723 static int traverse_persistent_callback_read(TDB_CONTEXT *tdb, TDB_DATA kbuf, TDB_DATA dbuf,
1724                                         void *private_data)
1725 {
1726         struct traverse_state *state = (struct traverse_state *)private_data;
1727         struct db_record rec;
1728
1729         /*
1730          * Skip the __db_sequence_number__ key:
1731          * This is used for persistent transactions internally.
1732          */
1733         if (kbuf.dsize == strlen(CTDB_DB_SEQNUM_KEY) + 1 &&
1734             strcmp((const char*)kbuf.dptr, CTDB_DB_SEQNUM_KEY) == 0)
1735         {
1736                 return 0;
1737         }
1738
1739         ZERO_STRUCT(rec);
1740         rec.db = state->db;
1741         rec.key = kbuf;
1742         rec.value = dbuf;
1743         rec.value_valid = true;
1744         rec.storev = db_ctdb_storev_deny;
1745         rec.delete_rec = db_ctdb_delete_deny;
1746         rec.private_data = NULL;
1747
1748         if (rec.value.dsize <= sizeof(struct ctdb_ltdb_header)) {
1749                 /* a deleted record */
1750                 return 0;
1751         }
1752         rec.value.dsize -= sizeof(struct ctdb_ltdb_header);
1753         rec.value.dptr += sizeof(struct ctdb_ltdb_header);
1754
1755         state->count++;
1756         return state->fn(&rec, state->private_data);
1757 }
1758
1759 static int db_ctdb_traverse_read(struct db_context *db,
1760                                  int (*fn)(struct db_record *rec,
1761                                            void *private_data),
1762                                  void *private_data)
1763 {
1764         int ret;
1765         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1766                                                         struct db_ctdb_ctx);
1767         struct traverse_state state;
1768
1769         state = (struct traverse_state) {
1770                 .db = db,
1771                 .fn = fn,
1772                 .private_data = private_data,
1773         };
1774
1775         if (db->persistent) {
1776                 /* for persistent databases we don't need to do a ctdb traverse,
1777                    we can do a faster local traverse */
1778                 int nrecs;
1779
1780                 nrecs = tdb_traverse_read(ctx->wtdb->tdb,
1781                                           traverse_persistent_callback_read,
1782                                           &state);
1783                 if (nrecs == -1) {
1784                         return -1;
1785                 }
1786                 return state.count;
1787         }
1788
1789         ret = db_ctdbd_traverse(ctx->db_id, traverse_read_callback, &state);
1790         if (ret != 0) {
1791                 return -1;
1792         }
1793         return state.count;
1794 }
1795
1796 static int db_ctdb_get_seqnum(struct db_context *db)
1797 {
1798         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1799                                                         struct db_ctdb_ctx);
1800         return tdb_get_seqnum(ctx->wtdb->tdb);
1801 }
1802
1803 static size_t db_ctdb_id(struct db_context *db, uint8_t *id, size_t idlen)
1804 {
1805         struct db_ctdb_ctx *ctx = talloc_get_type_abort(
1806                 db->private_data, struct db_ctdb_ctx);
1807
1808         if (idlen >= sizeof(ctx->db_id)) {
1809                 memcpy(id, &ctx->db_id, sizeof(ctx->db_id));
1810         }
1811
1812         return sizeof(ctx->db_id);
1813 }
1814
1815 struct db_context *db_open_ctdb(TALLOC_CTX *mem_ctx,
1816                                 struct messaging_context *msg_ctx,
1817                                 const char *name,
1818                                 int hash_size, int tdb_flags,
1819                                 int open_flags, mode_t mode,
1820                                 enum dbwrap_lock_order lock_order,
1821                                 uint64_t dbwrap_flags)
1822 {
1823         struct db_context *result;
1824         struct db_ctdb_ctx *db_ctdb;
1825         char *db_path;
1826         struct loadparm_context *lp_ctx;
1827         TDB_DATA data;
1828         TDB_DATA outdata = {0};
1829         bool persistent = (tdb_flags & TDB_CLEAR_IF_FIRST) == 0;
1830         int32_t cstatus;
1831         int ret;
1832
1833         if (!lp_clustering()) {
1834                 DEBUG(10, ("Clustering disabled -- no ctdb\n"));
1835                 return NULL;
1836         }
1837
1838         if (!(result = talloc_zero(mem_ctx, struct db_context))) {
1839                 DEBUG(0, ("talloc failed\n"));
1840                 TALLOC_FREE(result);
1841                 return NULL;
1842         }
1843
1844         if (!(db_ctdb = talloc(result, struct db_ctdb_ctx))) {
1845                 DEBUG(0, ("talloc failed\n"));
1846                 TALLOC_FREE(result);
1847                 return NULL;
1848         }
1849
1850         result->name = talloc_strdup(result, name);
1851         if (result->name == NULL) {
1852                 DEBUG(0, ("talloc failed\n"));
1853                 TALLOC_FREE(result);
1854                 return NULL;
1855         }
1856
1857         db_ctdb->transaction = NULL;
1858         db_ctdb->db = result;
1859
1860         ret = ctdbd_db_attach(messaging_ctdb_connection(), name,
1861                               &db_ctdb->db_id, persistent);
1862         if (ret != 0) {
1863                 DEBUG(0, ("ctdbd_db_attach failed for %s: %s\n", name,
1864                           strerror(ret)));
1865                 TALLOC_FREE(result);
1866                 return NULL;
1867         }
1868
1869         if (tdb_flags & TDB_SEQNUM) {
1870                 data.dptr = (uint8_t *)&db_ctdb->db_id;
1871                 data.dsize = sizeof(db_ctdb->db_id);
1872
1873                 ret = ctdbd_control_local(messaging_ctdb_connection(),
1874                                           CTDB_CONTROL_ENABLE_SEQNUM,
1875                                           0, 0, data,
1876                                           NULL, NULL, &cstatus);
1877                 if ((ret != 0) || cstatus != 0) {
1878                         DBG_ERR("ctdb_control for enable seqnum "
1879                                 "failed: %s\n", strerror(ret));
1880                         TALLOC_FREE(result);
1881                         return NULL;
1882                 }
1883         }
1884
1885         db_path = ctdbd_dbpath(messaging_ctdb_connection(), db_ctdb,
1886                                db_ctdb->db_id);
1887         if (db_path == NULL) {
1888                 DBG_ERR("ctdbd_dbpath failed\n");
1889                 TALLOC_FREE(result);
1890                 return NULL;
1891         }
1892
1893         result->persistent = persistent;
1894         result->lock_order = lock_order;
1895
1896         data.dptr = (uint8_t *)&db_ctdb->db_id;
1897         data.dsize = sizeof(db_ctdb->db_id);
1898
1899         ret = ctdbd_control_local(messaging_ctdb_connection(),
1900                                   CTDB_CONTROL_DB_OPEN_FLAGS,
1901                                   0, 0, data, NULL, &outdata, &cstatus);
1902         if (ret != 0) {
1903                 DBG_ERR(" ctdb control for db_open_flags "
1904                          "failed: %s\n", strerror(ret));
1905                 TALLOC_FREE(result);
1906                 return NULL;
1907         }
1908
1909         if (cstatus != 0 || outdata.dsize != sizeof(int)) {
1910                 DBG_ERR("ctdb_control for db_open_flags failed\n");
1911                 TALLOC_FREE(outdata.dptr);
1912                 TALLOC_FREE(result);
1913                 return NULL;
1914         }
1915
1916         tdb_flags = *(int *)outdata.dptr;
1917         TALLOC_FREE(outdata.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 }