lib/tdb/common/tdb.c

   1  /*
   2    Unix SMB/CIFS implementation.
   3
   4    trivial database library
   5
   6    Copyright (C) Andrew Tridgell              1999-2005
   7    Copyright (C) Paul `Rusty' Russell              2000
   8    Copyright (C) Jeremy Allison                    2000-2003
   9
  10      ** NOTE! The following LGPL license applies to the tdb
  11      ** library. This does NOT imply that all of Samba is released
  12      ** under the LGPL
  13
  14    This library is free software; you can redistribute it and/or
  15    modify it under the terms of the GNU Lesser General Public
  16    License as published by the Free Software Foundation; either
  17    version 3 of the License, or (at your option) any later version.
  18
  19    This library is distributed in the hope that it will be useful,
  20    but WITHOUT ANY WARRANTY; without even the implied warranty of
  21    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  22    Lesser General Public License for more details.
  23
  24    You should have received a copy of the GNU Lesser General Public
  25    License along with this library; if not, see <http://www.gnu.org/licenses/>.
  26 */
  27
  28 #include "tdb_private.h"
  29
  30 _PUBLIC_ TDB_DATA tdb_null;
  31
  32 /*
  33   non-blocking increment of the tdb sequence number if the tdb has been opened using
  34   the TDB_SEQNUM flag
  35 */
  36 _PUBLIC_ void tdb_increment_seqnum_nonblock(struct tdb_context *tdb)
  37 {
  38         tdb_off_t seqnum=0;
  39
  40         if (!(tdb->flags & TDB_SEQNUM)) {
  41                 return;
  42         }
  43
  44         /* we ignore errors from this, as we have no sane way of
  45            dealing with them.
  46         */
  47         tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &seqnum);
  48         seqnum++;
  49         tdb_ofs_write(tdb, TDB_SEQNUM_OFS, &seqnum);
  50 }
  51
  52 /*
  53   increment the tdb sequence number if the tdb has been opened using
  54   the TDB_SEQNUM flag
  55 */
  56 static void tdb_increment_seqnum(struct tdb_context *tdb)
  57 {
  58         if (!(tdb->flags & TDB_SEQNUM)) {
  59                 return;
  60         }
  61
  62         if (tdb->transaction != NULL) {
  63                 tdb_increment_seqnum_nonblock(tdb);
  64                 return;
  65         }
  66
  67         if (tdb_nest_lock(tdb, TDB_SEQNUM_OFS, F_WRLCK,
  68                           TDB_LOCK_WAIT|TDB_LOCK_PROBE) != 0) {
  69                 return;
  70         }
  71
  72         tdb_increment_seqnum_nonblock(tdb);
  73
  74         tdb_nest_unlock(tdb, TDB_SEQNUM_OFS, F_WRLCK, false);
  75 }
  76
  77 static int tdb_key_compare(TDB_DATA key, TDB_DATA data, void *private_data)
  78 {
  79         return memcmp(data.dptr, key.dptr, data.dsize);
  80 }
  81
  82 void tdb_chainwalk_init(struct tdb_chainwalk_ctx *ctx, tdb_off_t ptr)
  83 {
  84         *ctx = (struct tdb_chainwalk_ctx) { .slow_ptr = ptr };
  85 }
  86
  87 bool tdb_chainwalk_check(struct tdb_context *tdb,
  88                          struct tdb_chainwalk_ctx *ctx,
  89                          tdb_off_t next_ptr)
  90 {
  91         int ret;
  92
  93         if (ctx->slow_chase) {
  94                 ret = tdb_ofs_read(tdb, ctx->slow_ptr, &ctx->slow_ptr);
  95                 if (ret == -1) {
  96                         return false;
  97                 }
  98         }
  99         ctx->slow_chase = !ctx->slow_chase;
 100
 101         if (next_ptr == ctx->slow_ptr) {
 102                 tdb->ecode = TDB_ERR_CORRUPT;
 103                 TDB_LOG((tdb, TDB_DEBUG_ERROR,
 104                          "tdb_chainwalk_check: circular chain\n"));
 105                 return false;
 106         }
 107
 108         return true;
 109 }
 110
 111 /* Returns 0 on fail.  On success, return offset of record, and fills
 112    in rec */
 113 static tdb_off_t tdb_find(struct tdb_context *tdb, TDB_DATA key, uint32_t hash,
 114                         struct tdb_record *r)
 115 {
 116         tdb_off_t rec_ptr;
 117         struct tdb_chainwalk_ctx chainwalk;
 118
 119         /* read in the hash top */
 120         if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
 121                 return 0;
 122
 123         tdb_chainwalk_init(&chainwalk, rec_ptr);
 124
 125         /* keep looking until we find the right record */
 126         while (rec_ptr) {
 127                 bool ok;
 128
 129                 if (tdb_rec_read(tdb, rec_ptr, r) == -1)
 130                         return 0;
 131
 132                 if (!TDB_DEAD(r) && hash==r->full_hash
 133                     && key.dsize==r->key_len
 134                     && tdb_parse_data(tdb, key, rec_ptr + sizeof(*r),
 135                                       r->key_len, tdb_key_compare,
 136                                       NULL) == 0) {
 137                         return rec_ptr;
 138                 }
 139                 rec_ptr = r->next;
 140
 141                 ok = tdb_chainwalk_check(tdb, &chainwalk, rec_ptr);
 142                 if (!ok) {
 143                         return 0;
 144                 }
 145         }
 146         tdb->ecode = TDB_ERR_NOEXIST;
 147         return 0;
 148 }
 149
 150 /* As tdb_find, but if you succeed, keep the lock */
 151 tdb_off_t tdb_find_lock_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash, int locktype,
 152                            struct tdb_record *rec)
 153 {
 154         uint32_t rec_ptr;
 155
 156         if (tdb_lock(tdb, BUCKET(hash), locktype) == -1)
 157                 return 0;
 158         if (!(rec_ptr = tdb_find(tdb, key, hash, rec)))
 159                 tdb_unlock(tdb, BUCKET(hash), locktype);
 160         return rec_ptr;
 161 }
 162
 163 static TDB_DATA _tdb_fetch(struct tdb_context *tdb, TDB_DATA key);
 164
 165 struct tdb_update_hash_state {
 166         const TDB_DATA *dbufs;
 167         int num_dbufs;
 168         tdb_len_t dbufs_len;
 169 };
 170
 171 static int tdb_update_hash_cmp(TDB_DATA key, TDB_DATA data, void *private_data)
 172 {
 173         struct tdb_update_hash_state *state = private_data;
 174         unsigned char *dptr = data.dptr;
 175         int i;
 176
 177         if (state->dbufs_len != data.dsize) {
 178                 return -1;
 179         }
 180
 181         for (i=0; i<state->num_dbufs; i++) {
 182                 TDB_DATA dbuf = state->dbufs[i];
 183                 if( dbuf.dsize > 0) {
 184                         int ret;
 185                         ret = memcmp(dptr, dbuf.dptr, dbuf.dsize);
 186                         if (ret != 0) {
 187                                 return -1;
 188                         }
 189                         dptr += dbuf.dsize;
 190                 }
 191         }
 192
 193         return 0;
 194 }
 195
 196 /* update an entry in place - this only works if the new data size
 197    is <= the old data size and the key exists.
 198    on failure return -1.
 199 */
 200 static int tdb_update_hash(struct tdb_context *tdb, TDB_DATA key,
 201                            uint32_t hash,
 202                            const TDB_DATA *dbufs, int num_dbufs,
 203                            tdb_len_t dbufs_len)
 204 {
 205         struct tdb_record rec;
 206         tdb_off_t rec_ptr, ofs;
 207         int i;
 208
 209         /* find entry */
 210         if (!(rec_ptr = tdb_find(tdb, key, hash, &rec)))
 211                 return -1;
 212
 213         /* it could be an exact duplicate of what is there - this is
 214          * surprisingly common (eg. with a ldb re-index). */
 215         if (rec.data_len == dbufs_len) {
 216                 struct tdb_update_hash_state state = {
 217                         .dbufs = dbufs, .num_dbufs = num_dbufs,
 218                         .dbufs_len = dbufs_len
 219                 };
 220                 int ret;
 221
 222                 ret = tdb_parse_record(tdb, key, tdb_update_hash_cmp, &state);
 223                 if (ret == 0) {
 224                         return 0;
 225                 }
 226         }
 227
 228         /* must be long enough key, data and tailer */
 229         if (rec.rec_len < key.dsize + dbufs_len + sizeof(tdb_off_t)) {
 230                 tdb->ecode = TDB_SUCCESS; /* Not really an error */
 231                 return -1;
 232         }
 233
 234         ofs = rec_ptr + sizeof(rec) + rec.key_len;
 235
 236         for (i=0; i<num_dbufs; i++) {
 237                 TDB_DATA dbuf = dbufs[i];
 238                 int ret;
 239
 240                 ret = tdb->methods->tdb_write(tdb, ofs, dbuf.dptr, dbuf.dsize);
 241                 if (ret == -1) {
 242                         return -1;
 243                 }
 244                 ofs += dbuf.dsize;
 245         }
 246
 247         if (dbufs_len != rec.data_len) {
 248                 /* update size */
 249                 rec.data_len = dbufs_len;
 250                 return tdb_rec_write(tdb, rec_ptr, &rec);
 251         }
 252
 253         return 0;
 254 }
 255
 256 /* find an entry in the database given a key */
 257 /* If an entry doesn't exist tdb_err will be set to
 258  * TDB_ERR_NOEXIST. If a key has no data attached
 259  * then the TDB_DATA will have zero length but
 260  * a non-zero pointer
 261  */
 262 static TDB_DATA _tdb_fetch(struct tdb_context *tdb, TDB_DATA key)
 263 {
 264         tdb_off_t rec_ptr;
 265         struct tdb_record rec;
 266         TDB_DATA ret;
 267         uint32_t hash;
 268
 269         /* find which hash bucket it is in */
 270         hash = tdb->hash_fn(&key);
 271         if (!(rec_ptr = tdb_find_lock_hash(tdb,key,hash,F_RDLCK,&rec)))
 272                 return tdb_null;
 273
 274         ret.dptr = tdb_alloc_read(tdb, rec_ptr + sizeof(rec) + rec.key_len,
 275                                   rec.data_len);
 276         ret.dsize = rec.data_len;
 277         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
 278         return ret;
 279 }
 280
 281 _PUBLIC_ TDB_DATA tdb_fetch(struct tdb_context *tdb, TDB_DATA key)
 282 {
 283         TDB_DATA ret = _tdb_fetch(tdb, key);
 284
 285         tdb_trace_1rec_retrec(tdb, "tdb_fetch", key, ret);
 286         return ret;
 287 }
 288
 289 /*
 290  * Find an entry in the database and hand the record's data to a parsing
 291  * function. The parsing function is executed under the chain read lock, so it
 292  * should be fast and should not block on other syscalls.
 293  *
 294  * DON'T CALL OTHER TDB CALLS FROM THE PARSER, THIS MIGHT LEAD TO SEGFAULTS.
 295  *
 296  * For mmapped tdb's that do not have a transaction open it points the parsing
 297  * function directly at the mmap area, it avoids the malloc/memcpy in this
 298  * case. If a transaction is open or no mmap is available, it has to do
 299  * malloc/read/parse/free.
 300  *
 301  * This is interesting for all readers of potentially large data structures in
 302  * the tdb records, ldb indexes being one example.
 303  *
 304  * Return -1 if the record was not found.
 305  */
 306
 307 _PUBLIC_ int tdb_parse_record(struct tdb_context *tdb, TDB_DATA key,
 308                      int (*parser)(TDB_DATA key, TDB_DATA data,
 309                                    void *private_data),
 310                      void *private_data)
 311 {
 312         tdb_off_t rec_ptr;
 313         struct tdb_record rec;
 314         int ret;
 315         uint32_t hash;
 316
 317         /* find which hash bucket it is in */
 318         hash = tdb->hash_fn(&key);
 319
 320         if (!(rec_ptr = tdb_find_lock_hash(tdb,key,hash,F_RDLCK,&rec))) {
 321                 /* record not found */
 322                 tdb_trace_1rec_ret(tdb, "tdb_parse_record", key, -1);
 323                 tdb->ecode = TDB_ERR_NOEXIST;
 324                 return -1;
 325         }
 326         tdb_trace_1rec_ret(tdb, "tdb_parse_record", key, 0);
 327
 328         ret = tdb_parse_data(tdb, key, rec_ptr + sizeof(rec) + rec.key_len,
 329                              rec.data_len, parser, private_data);
 330
 331         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
 332
 333         return ret;
 334 }
 335
 336 /* check if an entry in the database exists
 337
 338    note that 1 is returned if the key is found and 0 is returned if not found
 339    this doesn't match the conventions in the rest of this module, but is
 340    compatible with gdbm
 341 */
 342 static int tdb_exists_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash)
 343 {
 344         struct tdb_record rec;
 345
 346         if (tdb_find_lock_hash(tdb, key, hash, F_RDLCK, &rec) == 0)
 347                 return 0;
 348         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
 349         return 1;
 350 }
 351
 352 _PUBLIC_ int tdb_exists(struct tdb_context *tdb, TDB_DATA key)
 353 {
 354         uint32_t hash = tdb->hash_fn(&key);
 355         int ret;
 356
 357         ret = tdb_exists_hash(tdb, key, hash);
 358         tdb_trace_1rec_ret(tdb, "tdb_exists", key, ret);
 359         return ret;
 360 }
 361
 362 /*
 363  * Move a dead record to the freelist. The hash chain and freelist
 364  * must be locked.
 365  */
 366 static int tdb_del_dead(struct tdb_context *tdb,
 367                         uint32_t last_ptr,
 368                         uint32_t rec_ptr,
 369                         struct tdb_record *rec,
 370                         bool *deleted)
 371 {
 372         int ret;
 373
 374         ret = tdb_write_lock_record(tdb, rec_ptr);
 375         if (ret == -1) {
 376                 /* Someone traversing here: Just leave it dead */
 377                 return 0;
 378         }
 379         ret = tdb_write_unlock_record(tdb, rec_ptr);
 380         if (ret == -1) {
 381                 return -1;
 382         }
 383         ret = tdb_ofs_write(tdb, last_ptr, &rec->next);
 384         if (ret == -1) {
 385                 return -1;
 386         }
 387
 388         *deleted = true;
 389
 390         ret = tdb_free(tdb, rec_ptr, rec);
 391         return ret;
 392 }
 393
 394 /*
 395  * Walk the hash chain and leave tdb->max_dead_records around. Move
 396  * the rest of dead records to the freelist.
 397  */
 398 int tdb_trim_dead(struct tdb_context *tdb, uint32_t hash)
 399 {
 400         struct tdb_chainwalk_ctx chainwalk;
 401         struct tdb_record rec;
 402         tdb_off_t last_ptr, rec_ptr;
 403         bool locked_freelist = false;
 404         int num_dead = 0;
 405         int ret;
 406
 407         last_ptr = TDB_HASH_TOP(hash);
 408
 409         /*
 410          * Init chainwalk with the pointer to the hash top. It might
 411          * be that the very first record in the chain is a dead one
 412          * that we have to delete.
 413          */
 414         tdb_chainwalk_init(&chainwalk, last_ptr);
 415
 416         ret = tdb_ofs_read(tdb, last_ptr, &rec_ptr);
 417         if (ret == -1) {
 418                 return -1;
 419         }
 420
 421         while (rec_ptr != 0) {
 422                 bool deleted = false;
 423                 uint32_t next;
 424
 425                 ret = tdb_rec_read(tdb, rec_ptr, &rec);
 426                 if (ret == -1) {
 427                         goto fail;
 428                 }
 429
 430                 /*
 431                  * Make a copy of rec.next: Further down we might
 432                  * delete and put the record on the freelist. Make
 433                  * sure that modifications in that code path can't
 434                  * break the chainwalk here.
 435                  */
 436                 next = rec.next;
 437
 438                 if (rec.magic == TDB_DEAD_MAGIC) {
 439                         num_dead += 1;
 440
 441                         if (num_dead > tdb->max_dead_records) {
 442
 443                                 if (!locked_freelist) {
 444                                         /*
 445                                          * Lock the freelist only if
 446                                          * it's really required.
 447                                          */
 448                                         ret = tdb_lock(tdb, -1, F_WRLCK);
 449                                         if (ret == -1) {
 450                                                 goto fail;
 451                                         };
 452                                         locked_freelist = true;
 453                                 }
 454
 455                                 ret = tdb_del_dead(
 456                                         tdb,
 457                                         last_ptr,
 458                                         rec_ptr,
 459                                         &rec,
 460                                         &deleted);
 461
 462                                 if (ret == -1) {
 463                                         goto fail;
 464                                 }
 465                         }
 466                 }
 467
 468                 /*
 469                  * Don't do the chainwalk check if "rec_ptr" was
 470                  * deleted. We reduced the chain, and the chainwalk
 471                  * check might catch up early. Imagine a valid chain
 472                  * with just dead records: We never can bump the
 473                  * "slow" pointer in chainwalk_check, as there isn't
 474                  * anything left to jump to and compare.
 475                  */
 476                 if (!deleted) {
 477                         bool ok;
 478
 479                         last_ptr = rec_ptr;
 480
 481                         ok = tdb_chainwalk_check(tdb, &chainwalk, next);
 482                         if (!ok) {
 483                                 ret = -1;
 484                                 goto fail;
 485                         }
 486                 }
 487                 rec_ptr = next;
 488         }
 489         ret = 0;
 490 fail:
 491         if (locked_freelist) {
 492                 tdb_unlock(tdb, -1, F_WRLCK);
 493         }
 494         return ret;
 495 }
 496
 497 /* delete an entry in the database given a key */
 498 static int tdb_delete_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash)
 499 {
 500         tdb_off_t rec_ptr;
 501         struct tdb_record rec;
 502         int ret;
 503
 504         if (tdb->read_only || tdb->traverse_read) {
 505                 tdb->ecode = TDB_ERR_RDONLY;
 506                 return -1;
 507         }
 508
 509         rec_ptr = tdb_find_lock_hash(tdb, key, hash, F_WRLCK, &rec);
 510         if (rec_ptr == 0) {
 511                 return -1;
 512         }
 513
 514         /*
 515          * Mark the record dead
 516          */
 517         rec.magic = TDB_DEAD_MAGIC;
 518         ret = tdb_rec_write(tdb, rec_ptr, &rec);
 519         if (ret == -1) {
 520                 goto done;
 521         }
 522
 523         tdb_increment_seqnum(tdb);
 524
 525         ret = tdb_trim_dead(tdb, hash);
 526 done:
 527         if (tdb_unlock(tdb, BUCKET(hash), F_WRLCK) != 0)
 528                 TDB_LOG((tdb, TDB_DEBUG_WARNING, "tdb_delete: WARNING tdb_unlock failed!\n"));
 529         return ret;
 530 }
 531
 532 _PUBLIC_ int tdb_delete(struct tdb_context *tdb, TDB_DATA key)
 533 {
 534         uint32_t hash = tdb->hash_fn(&key);
 535         int ret;
 536
 537         ret = tdb_delete_hash(tdb, key, hash);
 538         tdb_trace_1rec_ret(tdb, "tdb_delete", key, ret);
 539         return ret;
 540 }
 541
 542 /*
 543  * See if we have a dead record around with enough space
 544  */
 545 tdb_off_t tdb_find_dead(struct tdb_context *tdb, uint32_t hash,
 546                         struct tdb_record *r, tdb_len_t length,
 547                         tdb_off_t *p_last_ptr)
 548 {
 549         tdb_off_t rec_ptr, last_ptr;
 550         struct tdb_chainwalk_ctx chainwalk;
 551         tdb_off_t best_rec_ptr = 0;
 552         tdb_off_t best_last_ptr = 0;
 553         struct tdb_record best = { .rec_len = UINT32_MAX };
 554
 555         length += sizeof(tdb_off_t); /* tailer */
 556
 557         last_ptr = TDB_HASH_TOP(hash);
 558
 559         /* read in the hash top */
 560         if (tdb_ofs_read(tdb, last_ptr, &rec_ptr) == -1)
 561                 return 0;
 562
 563         tdb_chainwalk_init(&chainwalk, rec_ptr);
 564
 565         /* keep looking until we find the right record */
 566         while (rec_ptr) {
 567                 bool ok;
 568
 569                 if (tdb_rec_read(tdb, rec_ptr, r) == -1)
 570                         return 0;
 571
 572                 if (TDB_DEAD(r) && (r->rec_len >= length) &&
 573                     (r->rec_len < best.rec_len)) {
 574                         best_rec_ptr = rec_ptr;
 575                         best_last_ptr = last_ptr;
 576                         best = *r;
 577                 }
 578                 last_ptr = rec_ptr;
 579                 rec_ptr = r->next;
 580
 581                 ok = tdb_chainwalk_check(tdb, &chainwalk, rec_ptr);
 582                 if (!ok) {
 583                         return 0;
 584                 }
 585         }
 586
 587         if (best.rec_len == UINT32_MAX) {
 588                 return 0;
 589         }
 590
 591         *r = best;
 592         *p_last_ptr = best_last_ptr;
 593         return best_rec_ptr;
 594 }
 595
 596 static int _tdb_storev(struct tdb_context *tdb, TDB_DATA key,
 597                        const TDB_DATA *dbufs, int num_dbufs,
 598                        int flag, uint32_t hash)
 599 {
 600         struct tdb_record rec;
 601         tdb_off_t rec_ptr, ofs;
 602         tdb_len_t rec_len, dbufs_len;
 603         int i;
 604         int ret = -1;
 605
 606         dbufs_len = 0;
 607
 608         for (i=0; i<num_dbufs; i++) {
 609                 size_t dsize = dbufs[i].dsize;
 610
 611                 if ((dsize != 0) && (dbufs[i].dptr == NULL)) {
 612                         tdb->ecode = TDB_ERR_EINVAL;
 613                         goto fail;
 614                 }
 615
 616                 dbufs_len += dsize;
 617                 if (dbufs_len < dsize) {
 618                         tdb->ecode = TDB_ERR_OOM;
 619                         goto fail;
 620                 }
 621         }
 622
 623         rec_len = key.dsize + dbufs_len;
 624         if ((rec_len < key.dsize) || (rec_len < dbufs_len)) {
 625                 tdb->ecode = TDB_ERR_OOM;
 626                 goto fail;
 627         }
 628
 629         /* check for it existing, on insert. */
 630         if (flag == TDB_INSERT) {
 631                 if (tdb_exists_hash(tdb, key, hash)) {
 632                         tdb->ecode = TDB_ERR_EXISTS;
 633                         goto fail;
 634                 }
 635         } else {
 636                 /* first try in-place update, on modify or replace. */
 637                 if (tdb_update_hash(tdb, key, hash, dbufs, num_dbufs,
 638                                     dbufs_len) == 0) {
 639                         goto done;
 640                 }
 641                 if (tdb->ecode == TDB_ERR_NOEXIST &&
 642                     flag == TDB_MODIFY) {
 643                         /* if the record doesn't exist and we are in TDB_MODIFY mode then
 644                          we should fail the store */
 645                         goto fail;
 646                 }
 647         }
 648         /* reset the error code potentially set by the tdb_update_hash() */
 649         tdb->ecode = TDB_SUCCESS;
 650
 651         /* delete any existing record - if it doesn't exist we don't
 652            care.  Doing this first reduces fragmentation, and avoids
 653            coalescing with `allocated' block before it's updated. */
 654         if (flag != TDB_INSERT)
 655                 tdb_delete_hash(tdb, key, hash);
 656
 657         /* we have to allocate some space */
 658         rec_ptr = tdb_allocate(tdb, hash, rec_len, &rec);
 659
 660         if (rec_ptr == 0) {
 661                 goto fail;
 662         }
 663
 664         /* Read hash top into next ptr */
 665         if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec.next) == -1)
 666                 goto fail;
 667
 668         rec.key_len = key.dsize;
 669         rec.data_len = dbufs_len;
 670         rec.full_hash = hash;
 671         rec.magic = TDB_MAGIC;
 672
 673         ofs = rec_ptr;
 674
 675         /* write out and point the top of the hash chain at it */
 676         ret = tdb_rec_write(tdb, ofs, &rec);
 677         if (ret == -1) {
 678                 goto fail;
 679         }
 680         ofs += sizeof(rec);
 681
 682         ret = tdb->methods->tdb_write(tdb, ofs, key.dptr, key.dsize);
 683         if (ret == -1) {
 684                 goto fail;
 685         }
 686         ofs += key.dsize;
 687
 688         for (i=0; i<num_dbufs; i++) {
 689                 if (dbufs[i].dsize == 0) {
 690                         continue;
 691                 }
 692
 693                 ret = tdb->methods->tdb_write(tdb, ofs, dbufs[i].dptr,
 694                                               dbufs[i].dsize);
 695                 if (ret == -1) {
 696                         goto fail;
 697                 }
 698                 ofs += dbufs[i].dsize;
 699         }
 700
 701         ret = tdb_ofs_write(tdb, TDB_HASH_TOP(hash), &rec_ptr);
 702         if (ret == -1) {
 703                 /* Need to tdb_unallocate() here */
 704                 goto fail;
 705         }
 706
 707  done:
 708         ret = 0;
 709  fail:
 710         if (ret == 0) {
 711                 tdb_increment_seqnum(tdb);
 712         }
 713         return ret;
 714 }
 715
 716 static int _tdb_store(struct tdb_context *tdb, TDB_DATA key,
 717                       TDB_DATA dbuf, int flag, uint32_t hash)
 718 {
 719         return _tdb_storev(tdb, key, &dbuf, 1, flag, hash);
 720 }
 721
 722 /* store an element in the database, replacing any existing element
 723    with the same key
 724
 725    return 0 on success, -1 on failure
 726 */
 727 _PUBLIC_ int tdb_store(struct tdb_context *tdb, TDB_DATA key, TDB_DATA dbuf, int flag)
 728 {
 729         uint32_t hash;
 730         int ret;
 731
 732         if (tdb->read_only || tdb->traverse_read) {
 733                 tdb->ecode = TDB_ERR_RDONLY;
 734                 tdb_trace_2rec_flag_ret(tdb, "tdb_store", key, dbuf, flag, -1);
 735                 return -1;
 736         }
 737
 738         /* find which hash bucket it is in */
 739         hash = tdb->hash_fn(&key);
 740         if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
 741                 return -1;
 742
 743         ret = _tdb_store(tdb, key, dbuf, flag, hash);
 744         tdb_trace_2rec_flag_ret(tdb, "tdb_store", key, dbuf, flag, ret);
 745         tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
 746         return ret;
 747 }
 748
 749 _PUBLIC_ int tdb_storev(struct tdb_context *tdb, TDB_DATA key,
 750                         const TDB_DATA *dbufs, int num_dbufs, int flag)
 751 {
 752         uint32_t hash;
 753         int ret;
 754
 755         if (tdb->read_only || tdb->traverse_read) {
 756                 tdb->ecode = TDB_ERR_RDONLY;
 757                 tdb_trace_1plusn_rec_flag_ret(tdb, "tdb_storev", key,
 758                                               dbufs, num_dbufs, flag, -1);
 759                 return -1;
 760         }
 761
 762         /* find which hash bucket it is in */
 763         hash = tdb->hash_fn(&key);
 764         if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
 765                 return -1;
 766
 767         ret = _tdb_storev(tdb, key, dbufs, num_dbufs, flag, hash);
 768         tdb_trace_1plusn_rec_flag_ret(tdb, "tdb_storev", key,
 769                                       dbufs, num_dbufs, flag, -1);
 770         tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
 771         return ret;
 772 }
 773
 774 /* Append to an entry. Create if not exist. */
 775 _PUBLIC_ int tdb_append(struct tdb_context *tdb, TDB_DATA key, TDB_DATA new_dbuf)
 776 {
 777         uint32_t hash;
 778         TDB_DATA dbufs[2];
 779         int ret = -1;
 780
 781         /* find which hash bucket it is in */
 782         hash = tdb->hash_fn(&key);
 783         if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
 784                 return -1;
 785
 786         dbufs[0] = _tdb_fetch(tdb, key);
 787         dbufs[1] = new_dbuf;
 788
 789         ret = _tdb_storev(tdb, key, dbufs, 2, 0, hash);
 790         tdb_trace_2rec_retrec(tdb, "tdb_append", key, dbufs[0], dbufs[1]);
 791
 792         tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
 793         SAFE_FREE(dbufs[0].dptr);
 794         return ret;
 795 }
 796
 797
 798 /*
 799   return the name of the current tdb file
 800   useful for external logging functions
 801 */
 802 _PUBLIC_ const char *tdb_name(struct tdb_context *tdb)
 803 {
 804         return tdb->name;
 805 }
 806
 807 /*
 808   return the underlying file descriptor being used by tdb, or -1
 809   useful for external routines that want to check the device/inode
 810   of the fd
 811 */
 812 _PUBLIC_ int tdb_fd(struct tdb_context *tdb)
 813 {
 814         return tdb->fd;
 815 }
 816
 817 /*
 818   return the current logging function
 819   useful for external tdb routines that wish to log tdb errors
 820 */
 821 _PUBLIC_ tdb_log_func tdb_log_fn(struct tdb_context *tdb)
 822 {
 823         return tdb->log.log_fn;
 824 }
 825
 826
 827 /*
 828   get the tdb sequence number. Only makes sense if the writers opened
 829   with TDB_SEQNUM set. Note that this sequence number will wrap quite
 830   quickly, so it should only be used for a 'has something changed'
 831   test, not for code that relies on the count of the number of changes
 832   made. If you want a counter then use a tdb record.
 833
 834   The aim of this sequence number is to allow for a very lightweight
 835   test of a possible tdb change.
 836 */
 837 _PUBLIC_ int tdb_get_seqnum(struct tdb_context *tdb)
 838 {
 839         tdb_off_t seqnum=0;
 840
 841         tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &seqnum);
 842         return seqnum;
 843 }
 844
 845 _PUBLIC_ int tdb_hash_size(struct tdb_context *tdb)
 846 {
 847         return tdb->hash_size;
 848 }
 849
 850 _PUBLIC_ size_t tdb_map_size(struct tdb_context *tdb)
 851 {
 852         return tdb->map_size;
 853 }
 854
 855 _PUBLIC_ int tdb_get_flags(struct tdb_context *tdb)
 856 {
 857         return tdb->flags;
 858 }
 859
 860 _PUBLIC_ void tdb_add_flags(struct tdb_context *tdb, unsigned flags)
 861 {
 862         if ((flags & TDB_ALLOW_NESTING) &&
 863             (flags & TDB_DISALLOW_NESTING)) {
 864                 tdb->ecode = TDB_ERR_NESTING;
 865                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_add_flags: "
 866                         "allow_nesting and disallow_nesting are not allowed together!"));
 867                 return;
 868         }
 869
 870         if (flags & TDB_ALLOW_NESTING) {
 871                 tdb->flags &= ~TDB_DISALLOW_NESTING;
 872         }
 873         if (flags & TDB_DISALLOW_NESTING) {
 874                 tdb->flags &= ~TDB_ALLOW_NESTING;
 875         }
 876
 877         tdb->flags |= flags;
 878 }
 879
 880 _PUBLIC_ void tdb_remove_flags(struct tdb_context *tdb, unsigned flags)
 881 {
 882         if ((flags & TDB_ALLOW_NESTING) &&
 883             (flags & TDB_DISALLOW_NESTING)) {
 884                 tdb->ecode = TDB_ERR_NESTING;
 885                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_remove_flags: "
 886                         "allow_nesting and disallow_nesting are not allowed together!"));
 887                 return;
 888         }
 889
 890         if ((flags & TDB_NOLOCK) &&
 891             (tdb->feature_flags & TDB_FEATURE_FLAG_MUTEX) &&
 892             (tdb->mutexes == NULL)) {
 893                 tdb->ecode = TDB_ERR_LOCK;
 894                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_remove_flags: "
 895                          "Can not remove NOLOCK flag on mutexed databases"));
 896                 return;
 897         }
 898
 899         if (flags & TDB_ALLOW_NESTING) {
 900                 tdb->flags |= TDB_DISALLOW_NESTING;
 901         }
 902         if (flags & TDB_DISALLOW_NESTING) {
 903                 tdb->flags |= TDB_ALLOW_NESTING;
 904         }
 905
 906         tdb->flags &= ~flags;
 907 }
 908
 909
 910 /*
 911   enable sequence number handling on an open tdb
 912 */
 913 _PUBLIC_ void tdb_enable_seqnum(struct tdb_context *tdb)
 914 {
 915         tdb->flags |= TDB_SEQNUM;
 916 }
 917
 918
 919 /*
 920   add a region of the file to the freelist. Length is the size of the region in bytes,
 921   which includes the free list header that needs to be added
 922  */
 923 static int tdb_free_region(struct tdb_context *tdb, tdb_off_t offset, ssize_t length)
 924 {
 925         struct tdb_record rec;
 926         if (length <= sizeof(rec)) {
 927                 /* the region is not worth adding */
 928                 return 0;
 929         }
 930         if (length + offset > tdb->map_size) {
 931                 TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_free_region: adding region beyond end of file\n"));
 932                 return -1;
 933         }
 934         memset(&rec,'\0',sizeof(rec));
 935         rec.rec_len = length - sizeof(rec);
 936         if (tdb_free(tdb, offset, &rec) == -1) {
 937                 TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_free_region: failed to add free record\n"));
 938                 return -1;
 939         }
 940         return 0;
 941 }
 942
 943 /*
 944   wipe the entire database, deleting all records. This can be done
 945   very fast by using a allrecord lock. The entire data portion of the
 946   file becomes a single entry in the freelist.
 947
 948   This code carefully steps around the recovery area, leaving it alone
 949  */
 950 _PUBLIC_ int tdb_wipe_all(struct tdb_context *tdb)
 951 {
 952         uint32_t i;
 953         tdb_off_t offset = 0;
 954         ssize_t data_len;
 955         tdb_off_t recovery_head;
 956         tdb_len_t recovery_size = 0;
 957
 958         if (tdb_lockall(tdb) != 0) {
 959                 return -1;
 960         }
 961
 962         tdb_trace(tdb, "tdb_wipe_all");
 963
 964         /* see if the tdb has a recovery area, and remember its size
 965            if so. We don't want to lose this as otherwise each
 966            tdb_wipe_all() in a transaction will increase the size of
 967            the tdb by the size of the recovery area */
 968         if (tdb_ofs_read(tdb, TDB_RECOVERY_HEAD, &recovery_head) == -1) {
 969                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_wipe_all: failed to read recovery head\n"));
 970                 goto failed;
 971         }
 972
 973         if (recovery_head != 0) {
 974                 struct tdb_record rec;
 975                 if (tdb->methods->tdb_read(tdb, recovery_head, &rec, sizeof(rec), DOCONV()) == -1) {
 976                         TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_wipe_all: failed to read recovery record\n"));
 977                         return -1;
 978                 }
 979                 recovery_size = rec.rec_len + sizeof(rec);
 980         }
 981
 982         /* wipe the hashes */
 983         for (i=0;i<tdb->hash_size;i++) {
 984                 if (tdb_ofs_write(tdb, TDB_HASH_TOP(i), &offset) == -1) {
 985                         TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_wipe_all: failed to write hash %d\n", i));
 986                         goto failed;
 987                 }
 988         }
 989
 990         /* wipe the freelist */
 991         if (tdb_ofs_write(tdb, FREELIST_TOP, &offset) == -1) {
 992                 TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_wipe_all: failed to write freelist\n"));
 993                 goto failed;
 994         }
 995
 996         /* add all the rest of the file to the freelist, possibly leaving a gap
 997            for the recovery area */
 998         if (recovery_size == 0) {
 999                 /* the simple case - the whole file can be used as a freelist */
1000                 data_len = (tdb->map_size - TDB_DATA_START(tdb->hash_size));
1001                 if (tdb_free_region(tdb, TDB_DATA_START(tdb->hash_size), data_len) != 0) {
1002                         goto failed;
1003                 }
1004         } else {
1005                 /* we need to add two freelist entries - one on either
1006                    side of the recovery area
1007
1008                    Note that we cannot shift the recovery area during
1009                    this operation. Only the transaction.c code may
1010                    move the recovery area or we risk subtle data
1011                    corruption
1012                 */
1013                 data_len = (recovery_head - TDB_DATA_START(tdb->hash_size));
1014                 if (tdb_free_region(tdb, TDB_DATA_START(tdb->hash_size), data_len) != 0) {
1015                         goto failed;
1016                 }
1017                 /* and the 2nd free list entry after the recovery area - if any */
1018                 data_len = tdb->map_size - (recovery_head+recovery_size);
1019                 if (tdb_free_region(tdb, recovery_head+recovery_size, data_len) != 0) {
1020                         goto failed;
1021                 }
1022         }
1023
1024         tdb_increment_seqnum_nonblock(tdb);
1025
1026         if (tdb_unlockall(tdb) != 0) {
1027                 TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_wipe_all: failed to unlock\n"));
1028                 goto failed;
1029         }
1030
1031         return 0;
1032
1033 failed:
1034         tdb_unlockall(tdb);
1035         return -1;
1036 }
1037
1038 struct traverse_state {
1039         bool error;
1040         struct tdb_context *dest_db;
1041 };
1042
1043 /*
1044   traverse function for repacking
1045  */
1046 static int repack_traverse(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *private_data)
1047 {
1048         struct traverse_state *state = (struct traverse_state *)private_data;
1049         if (tdb_store(state->dest_db, key, data, TDB_INSERT) != 0) {
1050                 state->error = true;
1051                 return -1;
1052         }
1053         return 0;
1054 }
1055
1056 /*
1057   repack a tdb
1058  */
1059 _PUBLIC_ int tdb_repack(struct tdb_context *tdb)
1060 {
1061         struct tdb_context *tmp_db;
1062         struct traverse_state state;
1063
1064         tdb_trace(tdb, "tdb_repack");
1065
1066         if (tdb_transaction_start(tdb) != 0) {
1067                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Failed to start transaction\n"));
1068                 return -1;
1069         }
1070
1071         tmp_db = tdb_open("tmpdb", tdb_hash_size(tdb), TDB_INTERNAL, O_RDWR|O_CREAT, 0);
1072         if (tmp_db == NULL) {
1073                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Failed to create tmp_db\n"));
1074                 tdb_transaction_cancel(tdb);
1075                 return -1;
1076         }
1077
1078         state.error = false;
1079         state.dest_db = tmp_db;
1080
1081         if (tdb_traverse_read(tdb, repack_traverse, &state) == -1) {
1082                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Failed to traverse copying out\n"));
1083                 tdb_transaction_cancel(tdb);
1084                 tdb_close(tmp_db);
1085                 return -1;
1086         }
1087
1088         if (state.error) {
1089                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Error during traversal\n"));
1090                 tdb_transaction_cancel(tdb);
1091                 tdb_close(tmp_db);
1092                 return -1;
1093         }
1094
1095         if (tdb_wipe_all(tdb) != 0) {
1096                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Failed to wipe database\n"));
1097                 tdb_transaction_cancel(tdb);
1098                 tdb_close(tmp_db);
1099                 return -1;
1100         }
1101
1102         state.error = false;
1103         state.dest_db = tdb;
1104
1105         if (tdb_traverse_read(tmp_db, repack_traverse, &state) == -1) {
1106                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Failed to traverse copying back\n"));
1107                 tdb_transaction_cancel(tdb);
1108                 tdb_close(tmp_db);
1109                 return -1;
1110         }
1111
1112         if (state.error) {
1113                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Error during second traversal\n"));
1114                 tdb_transaction_cancel(tdb);
1115                 tdb_close(tmp_db);
1116                 return -1;
1117         }
1118
1119         tdb_close(tmp_db);
1120
1121         if (tdb_transaction_commit(tdb) != 0) {
1122                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Failed to commit\n"));
1123                 return -1;
1124         }
1125
1126         return 0;
1127 }
1128
1129 /* Even on files, we can get partial writes due to signals. */
1130 bool tdb_write_all(int fd, const void *buf, size_t count)
1131 {
1132         while (count) {
1133                 ssize_t ret;
1134                 ret = write(fd, buf, count);
1135                 if (ret < 0)
1136                         return false;
1137                 buf = (const char *)buf + ret;
1138                 count -= ret;
1139         }
1140         return true;
1141 }
1142
1143 bool tdb_add_off_t(tdb_off_t a, tdb_off_t b, tdb_off_t *pret)
1144 {
1145         tdb_off_t ret = a + b;
1146
1147         if ((ret < a) || (ret < b)) {
1148                 return false;
1149         }
1150         *pret = ret;
1151         return true;
1152 }
1153
1154 #ifdef TDB_TRACE
1155 static void tdb_trace_write(struct tdb_context *tdb, const char *str)
1156 {
1157         if (!tdb_write_all(tdb->tracefd, str, strlen(str))) {
1158                 close(tdb->tracefd);
1159                 tdb->tracefd = -1;
1160         }
1161 }
1162
1163 static void tdb_trace_start(struct tdb_context *tdb)
1164 {
1165         tdb_off_t seqnum=0;
1166         char msg[sizeof(tdb_off_t) * 4 + 1];
1167
1168         tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &seqnum);
1169         snprintf(msg, sizeof(msg), "%u ", seqnum);
1170         tdb_trace_write(tdb, msg);
1171 }
1172
1173 static void tdb_trace_end(struct tdb_context *tdb)
1174 {
1175         tdb_trace_write(tdb, "\n");
1176 }
1177
1178 static void tdb_trace_end_ret(struct tdb_context *tdb, int ret)
1179 {
1180         char msg[sizeof(ret) * 4 + 4];
1181         snprintf(msg, sizeof(msg), " = %i\n", ret);
1182         tdb_trace_write(tdb, msg);
1183 }
1184
1185 static void tdb_trace_record(struct tdb_context *tdb, TDB_DATA rec)
1186 {
1187         char msg[20 + rec.dsize*2], *p;
1188         unsigned int i;
1189
1190         /* We differentiate zero-length records from non-existent ones. */
1191         if (rec.dptr == NULL) {
1192                 tdb_trace_write(tdb, " NULL");
1193                 return;
1194         }
1195
1196         /* snprintf here is purely cargo-cult programming. */
1197         p = msg;
1198         p += snprintf(p, sizeof(msg), " %zu:", rec.dsize);
1199         for (i = 0; i < rec.dsize; i++)
1200                 p += snprintf(p, 2, "%02x", rec.dptr[i]);
1201
1202         tdb_trace_write(tdb, msg);
1203 }
1204
1205 void tdb_trace(struct tdb_context *tdb, const char *op)
1206 {
1207         tdb_trace_start(tdb);
1208         tdb_trace_write(tdb, op);
1209         tdb_trace_end(tdb);
1210 }
1211
1212 void tdb_trace_seqnum(struct tdb_context *tdb, uint32_t seqnum, const char *op)
1213 {
1214         char msg[sizeof(tdb_off_t) * 4 + 1];
1215
1216         snprintf(msg, sizeof(msg), "%u ", seqnum);
1217         tdb_trace_write(tdb, msg);
1218         tdb_trace_write(tdb, op);
1219         tdb_trace_end(tdb);
1220 }
1221
1222 void tdb_trace_open(struct tdb_context *tdb, const char *op,
1223                     unsigned hash_size, unsigned tdb_flags, unsigned open_flags)
1224 {
1225         char msg[128];
1226
1227         snprintf(msg, sizeof(msg),
1228                  "%s %u 0x%x 0x%x", op, hash_size, tdb_flags, open_flags);
1229         tdb_trace_start(tdb);
1230         tdb_trace_write(tdb, msg);
1231         tdb_trace_end(tdb);
1232 }
1233
1234 void tdb_trace_ret(struct tdb_context *tdb, const char *op, int ret)
1235 {
1236         tdb_trace_start(tdb);
1237         tdb_trace_write(tdb, op);
1238         tdb_trace_end_ret(tdb, ret);
1239 }
1240
1241 void tdb_trace_retrec(struct tdb_context *tdb, const char *op, TDB_DATA ret)
1242 {
1243         tdb_trace_start(tdb);
1244         tdb_trace_write(tdb, op);
1245         tdb_trace_write(tdb, " =");
1246         tdb_trace_record(tdb, ret);
1247         tdb_trace_end(tdb);
1248 }
1249
1250 void tdb_trace_1rec(struct tdb_context *tdb, const char *op,
1251                     TDB_DATA rec)
1252 {
1253         tdb_trace_start(tdb);
1254         tdb_trace_write(tdb, op);
1255         tdb_trace_record(tdb, rec);
1256         tdb_trace_end(tdb);
1257 }
1258
1259 void tdb_trace_1rec_ret(struct tdb_context *tdb, const char *op,
1260                         TDB_DATA rec, int ret)
1261 {
1262         tdb_trace_start(tdb);
1263         tdb_trace_write(tdb, op);
1264         tdb_trace_record(tdb, rec);
1265         tdb_trace_end_ret(tdb, ret);
1266 }
1267
1268 void tdb_trace_1rec_retrec(struct tdb_context *tdb, const char *op,
1269                            TDB_DATA rec, TDB_DATA ret)
1270 {
1271         tdb_trace_start(tdb);
1272         tdb_trace_write(tdb, op);
1273         tdb_trace_record(tdb, rec);
1274         tdb_trace_write(tdb, " =");
1275         tdb_trace_record(tdb, ret);
1276         tdb_trace_end(tdb);
1277 }
1278
1279 void tdb_trace_2rec_flag_ret(struct tdb_context *tdb, const char *op,
1280                              TDB_DATA rec1, TDB_DATA rec2, unsigned flag,
1281                              int ret)
1282 {
1283         char msg[1 + sizeof(ret) * 4];
1284
1285         snprintf(msg, sizeof(msg), " %#x", flag);
1286         tdb_trace_start(tdb);
1287         tdb_trace_write(tdb, op);
1288         tdb_trace_record(tdb, rec1);
1289         tdb_trace_record(tdb, rec2);
1290         tdb_trace_write(tdb, msg);
1291         tdb_trace_end_ret(tdb, ret);
1292 }
1293
1294 void tdb_trace_1plusn_rec_flag_ret(struct tdb_context *tdb, const char *op,
1295                                    TDB_DATA rec,
1296                                    const TDB_DATA *recs, int num_recs,
1297                                    unsigned flag, int ret)
1298 {
1299         char msg[1 + sizeof(ret) * 4];
1300         int i;
1301
1302         snprintf(msg, sizeof(msg), " %#x", flag);
1303         tdb_trace_start(tdb);
1304         tdb_trace_write(tdb, op);
1305         tdb_trace_record(tdb, rec);
1306         for (i=0; i<num_recs; i++) {
1307                 tdb_trace_record(tdb, recs[i]);
1308         }
1309         tdb_trace_write(tdb, msg);
1310         tdb_trace_end_ret(tdb, ret);
1311 }
1312
1313 void tdb_trace_2rec_retrec(struct tdb_context *tdb, const char *op,
1314                            TDB_DATA rec1, TDB_DATA rec2, TDB_DATA ret)
1315 {
1316         tdb_trace_start(tdb);
1317         tdb_trace_write(tdb, op);
1318         tdb_trace_record(tdb, rec1);
1319         tdb_trace_record(tdb, rec2);
1320         tdb_trace_write(tdb, " =");
1321         tdb_trace_record(tdb, ret);
1322         tdb_trace_end(tdb);
1323 }
1324 #endif