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