source3/lib/dbwrap/dbwrap_ctdb.c

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