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 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 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_brlock(tdb, TDB_SEQNUM_OFS, F_WRLCK, F_SETLKW, 1, 1) != 0) {
  63                 return;
  64         }
  65
  66         tdb_increment_seqnum_nonblock(tdb);
  67
  68         tdb_brlock(tdb, TDB_SEQNUM_OFS, F_UNLCK, F_SETLKW, 1, 1);
  69 }
  70
  71 static int tdb_key_compare(TDB_DATA key, TDB_DATA data, void *private_data)
  72 {
  73         return memcmp(data.dptr, key.dptr, data.dsize);
  74 }
  75
  76 /* Returns 0 on fail.  On success, return offset of record, and fills
  77    in rec */
  78 static tdb_off_t tdb_find(struct tdb_context *tdb, TDB_DATA key, uint32_t hash,
  79                         struct tdb_record *r)
  80 {
  81         tdb_off_t rec_ptr;
  82
  83         /* read in the hash top */
  84         if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
  85                 return 0;
  86
  87         /* keep looking until we find the right record */
  88         while (rec_ptr) {
  89                 if (tdb_rec_read(tdb, rec_ptr, r) == -1)
  90                         return 0;
  91
  92                 if (!TDB_DEAD(r) && hash==r->full_hash
  93                     && key.dsize==r->key_len
  94                     && tdb_parse_data(tdb, key, rec_ptr + sizeof(*r),
  95                                       r->key_len, tdb_key_compare,
  96                                       NULL) == 0) {
  97                         return rec_ptr;
  98                 }
  99                 /* detect tight infinite loop */
 100                 if (rec_ptr == r->next) {
 101                         tdb->ecode = TDB_ERR_CORRUPT;
 102                         TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_find: loop detected.\n"));
 103                         return 0;
 104                 }
 105                 rec_ptr = r->next;
 106         }
 107         tdb->ecode = TDB_ERR_NOEXIST;
 108         return 0;
 109 }
 110
 111 /* As tdb_find, but if you succeed, keep the lock */
 112 tdb_off_t tdb_find_lock_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash, int locktype,
 113                            struct tdb_record *rec)
 114 {
 115         uint32_t rec_ptr;
 116
 117         if (tdb_lock(tdb, BUCKET(hash), locktype) == -1)
 118                 return 0;
 119         if (!(rec_ptr = tdb_find(tdb, key, hash, rec)))
 120                 tdb_unlock(tdb, BUCKET(hash), locktype);
 121         return rec_ptr;
 122 }
 123
 124 static TDB_DATA _tdb_fetch(struct tdb_context *tdb, TDB_DATA key);
 125
 126 /* update an entry in place - this only works if the new data size
 127    is <= the old data size and the key exists.
 128    on failure return -1.
 129 */
 130 static int tdb_update_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash, TDB_DATA dbuf)
 131 {
 132         struct tdb_record rec;
 133         tdb_off_t rec_ptr;
 134
 135         /* find entry */
 136         if (!(rec_ptr = tdb_find(tdb, key, hash, &rec)))
 137                 return -1;
 138
 139         /* it could be an exact duplicate of what is there - this is
 140          * surprisingly common (eg. with a ldb re-index). */
 141         if (rec.key_len == key.dsize &&
 142             rec.data_len == dbuf.dsize &&
 143             rec.full_hash == hash) {
 144                 TDB_DATA data = _tdb_fetch(tdb, key);
 145                 if (data.dsize == dbuf.dsize &&
 146                     memcmp(data.dptr, dbuf.dptr, data.dsize) == 0) {
 147                         if (data.dptr) {
 148                                 free(data.dptr);
 149                         }
 150                         return 0;
 151                 }
 152                 if (data.dptr) {
 153                         free(data.dptr);
 154                 }
 155         }
 156
 157
 158         /* must be long enough key, data and tailer */
 159         if (rec.rec_len < key.dsize + dbuf.dsize + sizeof(tdb_off_t)) {
 160                 tdb->ecode = TDB_SUCCESS; /* Not really an error */
 161                 return -1;
 162         }
 163
 164         if (tdb->methods->tdb_write(tdb, rec_ptr + sizeof(rec) + rec.key_len,
 165                       dbuf.dptr, dbuf.dsize) == -1)
 166                 return -1;
 167
 168         if (dbuf.dsize != rec.data_len) {
 169                 /* update size */
 170                 rec.data_len = dbuf.dsize;
 171                 return tdb_rec_write(tdb, rec_ptr, &rec);
 172         }
 173
 174         return 0;
 175 }
 176
 177 /* find an entry in the database given a key */
 178 /* If an entry doesn't exist tdb_err will be set to
 179  * TDB_ERR_NOEXIST. If a key has no data attached
 180  * then the TDB_DATA will have zero length but
 181  * a non-zero pointer
 182  */
 183 static TDB_DATA _tdb_fetch(struct tdb_context *tdb, TDB_DATA key)
 184 {
 185         tdb_off_t rec_ptr;
 186         struct tdb_record rec;
 187         TDB_DATA ret;
 188         uint32_t hash;
 189
 190         /* find which hash bucket it is in */
 191         hash = tdb->hash_fn(&key);
 192         if (!(rec_ptr = tdb_find_lock_hash(tdb,key,hash,F_RDLCK,&rec)))
 193                 return tdb_null;
 194
 195         ret.dptr = tdb_alloc_read(tdb, rec_ptr + sizeof(rec) + rec.key_len,
 196                                   rec.data_len);
 197         ret.dsize = rec.data_len;
 198         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
 199         return ret;
 200 }
 201
 202 TDB_DATA tdb_fetch(struct tdb_context *tdb, TDB_DATA key)
 203 {
 204         TDB_DATA ret = _tdb_fetch(tdb, key);
 205
 206         tdb_trace_1rec_retrec(tdb, "tdb_fetch", key, ret);
 207         return ret;
 208 }
 209
 210 /*
 211  * Find an entry in the database and hand the record's data to a parsing
 212  * function. The parsing function is executed under the chain read lock, so it
 213  * should be fast and should not block on other syscalls.
 214  *
 215  * DONT CALL OTHER TDB CALLS FROM THE PARSER, THIS MIGHT LEAD TO SEGFAULTS.
 216  *
 217  * For mmapped tdb's that do not have a transaction open it points the parsing
 218  * function directly at the mmap area, it avoids the malloc/memcpy in this
 219  * case. If a transaction is open or no mmap is available, it has to do
 220  * malloc/read/parse/free.
 221  *
 222  * This is interesting for all readers of potentially large data structures in
 223  * the tdb records, ldb indexes being one example.
 224  */
 225
 226 int tdb_parse_record(struct tdb_context *tdb, TDB_DATA key,
 227                      int (*parser)(TDB_DATA key, TDB_DATA data,
 228                                    void *private_data),
 229                      void *private_data)
 230 {
 231         tdb_off_t rec_ptr;
 232         struct tdb_record rec;
 233         int ret;
 234         uint32_t hash;
 235
 236         /* find which hash bucket it is in */
 237         hash = tdb->hash_fn(&key);
 238
 239         if (!(rec_ptr = tdb_find_lock_hash(tdb,key,hash,F_RDLCK,&rec))) {
 240                 tdb_trace_1rec_ret(tdb, "tdb_parse_record", key, -1);
 241                 tdb->ecode = TDB_ERR_NOEXIST;
 242                 return 0;
 243         }
 244         tdb_trace_1rec_ret(tdb, "tdb_parse_record", key, 0);
 245
 246         ret = tdb_parse_data(tdb, key, rec_ptr + sizeof(rec) + rec.key_len,
 247                              rec.data_len, parser, private_data);
 248
 249         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
 250
 251         return ret;
 252 }
 253
 254 /* check if an entry in the database exists
 255
 256    note that 1 is returned if the key is found and 0 is returned if not found
 257    this doesn't match the conventions in the rest of this module, but is
 258    compatible with gdbm
 259 */
 260 static int tdb_exists_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash)
 261 {
 262         struct tdb_record rec;
 263
 264         if (tdb_find_lock_hash(tdb, key, hash, F_RDLCK, &rec) == 0)
 265                 return 0;
 266         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
 267         return 1;
 268 }
 269
 270 int tdb_exists(struct tdb_context *tdb, TDB_DATA key)
 271 {
 272         uint32_t hash = tdb->hash_fn(&key);
 273         int ret;
 274
 275         ret = tdb_exists_hash(tdb, key, hash);
 276         tdb_trace_1rec_ret(tdb, "tdb_exists", key, ret);
 277         return ret;
 278 }
 279
 280 /* actually delete an entry in the database given the offset */
 281 int tdb_do_delete(struct tdb_context *tdb, tdb_off_t rec_ptr, struct tdb_record *rec)
 282 {
 283         tdb_off_t last_ptr, i;
 284         struct tdb_record lastrec;
 285
 286         if (tdb->read_only || tdb->traverse_read) return -1;
 287
 288         if (((tdb->traverse_write != 0) && (!TDB_DEAD(rec))) ||
 289             tdb_write_lock_record(tdb, rec_ptr) == -1) {
 290                 /* Someone traversing here: mark it as dead */
 291                 rec->magic = TDB_DEAD_MAGIC;
 292                 return tdb_rec_write(tdb, rec_ptr, rec);
 293         }
 294         if (tdb_write_unlock_record(tdb, rec_ptr) != 0)
 295                 return -1;
 296
 297         /* find previous record in hash chain */
 298         if (tdb_ofs_read(tdb, TDB_HASH_TOP(rec->full_hash), &i) == -1)
 299                 return -1;
 300         for (last_ptr = 0; i != rec_ptr; last_ptr = i, i = lastrec.next)
 301                 if (tdb_rec_read(tdb, i, &lastrec) == -1)
 302                         return -1;
 303
 304         /* unlink it: next ptr is at start of record. */
 305         if (last_ptr == 0)
 306                 last_ptr = TDB_HASH_TOP(rec->full_hash);
 307         if (tdb_ofs_write(tdb, last_ptr, &rec->next) == -1)
 308                 return -1;
 309
 310         /* recover the space */
 311         if (tdb_free(tdb, rec_ptr, rec) == -1)
 312                 return -1;
 313         return 0;
 314 }
 315
 316 static int tdb_count_dead(struct tdb_context *tdb, uint32_t hash)
 317 {
 318         int res = 0;
 319         tdb_off_t rec_ptr;
 320         struct tdb_record rec;
 321
 322         /* read in the hash top */
 323         if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
 324                 return 0;
 325
 326         while (rec_ptr) {
 327                 if (tdb_rec_read(tdb, rec_ptr, &rec) == -1)
 328                         return 0;
 329
 330                 if (rec.magic == TDB_DEAD_MAGIC) {
 331                         res += 1;
 332                 }
 333                 rec_ptr = rec.next;
 334         }
 335         return res;
 336 }
 337
 338 /*
 339  * Purge all DEAD records from a hash chain
 340  */
 341 static int tdb_purge_dead(struct tdb_context *tdb, uint32_t hash)
 342 {
 343         int res = -1;
 344         struct tdb_record rec;
 345         tdb_off_t rec_ptr;
 346
 347         if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
 348                 return -1;
 349         }
 350
 351         /* read in the hash top */
 352         if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
 353                 goto fail;
 354
 355         while (rec_ptr) {
 356                 tdb_off_t next;
 357
 358                 if (tdb_rec_read(tdb, rec_ptr, &rec) == -1) {
 359                         goto fail;
 360                 }
 361
 362                 next = rec.next;
 363
 364                 if (rec.magic == TDB_DEAD_MAGIC
 365                     && tdb_do_delete(tdb, rec_ptr, &rec) == -1) {
 366                         goto fail;
 367                 }
 368                 rec_ptr = next;
 369         }
 370         res = 0;
 371  fail:
 372         tdb_unlock(tdb, -1, F_WRLCK);
 373         return res;
 374 }
 375
 376 /* delete an entry in the database given a key */
 377 static int tdb_delete_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash)
 378 {
 379         tdb_off_t rec_ptr;
 380         struct tdb_record rec;
 381         int ret;
 382
 383         if (tdb->max_dead_records != 0) {
 384
 385                 /*
 386                  * Allow for some dead records per hash chain, mainly for
 387                  * tdb's with a very high create/delete rate like locking.tdb.
 388                  */
 389
 390                 if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
 391                         return -1;
 392
 393                 if (tdb_count_dead(tdb, hash) >= tdb->max_dead_records) {
 394                         /*
 395                          * Don't let the per-chain freelist grow too large,
 396                          * delete all existing dead records
 397                          */
 398                         tdb_purge_dead(tdb, hash);
 399                 }
 400
 401                 if (!(rec_ptr = tdb_find(tdb, key, hash, &rec))) {
 402                         tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
 403                         return -1;
 404                 }
 405
 406                 /*
 407                  * Just mark the record as dead.
 408                  */
 409                 rec.magic = TDB_DEAD_MAGIC;
 410                 ret = tdb_rec_write(tdb, rec_ptr, &rec);
 411         }
 412         else {
 413                 if (!(rec_ptr = tdb_find_lock_hash(tdb, key, hash, F_WRLCK,
 414                                                    &rec)))
 415                         return -1;
 416
 417                 ret = tdb_do_delete(tdb, rec_ptr, &rec);
 418         }
 419
 420         if (ret == 0) {
 421                 tdb_increment_seqnum(tdb);
 422         }
 423
 424         if (tdb_unlock(tdb, BUCKET(rec.full_hash), F_WRLCK) != 0)
 425                 TDB_LOG((tdb, TDB_DEBUG_WARNING, "tdb_delete: WARNING tdb_unlock failed!\n"));
 426         return ret;
 427 }
 428
 429 int tdb_delete(struct tdb_context *tdb, TDB_DATA key)
 430 {
 431         uint32_t hash = tdb->hash_fn(&key);
 432         int ret;
 433
 434         ret = tdb_delete_hash(tdb, key, hash);
 435         tdb_trace_1rec_ret(tdb, "tdb_delete", key, ret);
 436         return ret;
 437 }
 438
 439 /*
 440  * See if we have a dead record around with enough space
 441  */
 442 static tdb_off_t tdb_find_dead(struct tdb_context *tdb, uint32_t hash,
 443                                struct tdb_record *r, tdb_len_t length)
 444 {
 445         tdb_off_t rec_ptr;
 446
 447         /* read in the hash top */
 448         if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
 449                 return 0;
 450
 451         /* keep looking until we find the right record */
 452         while (rec_ptr) {
 453                 if (tdb_rec_read(tdb, rec_ptr, r) == -1)
 454                         return 0;
 455
 456                 if (TDB_DEAD(r) && r->rec_len >= length) {
 457                         /*
 458                          * First fit for simple coding, TODO: change to best
 459                          * fit
 460                          */
 461                         return rec_ptr;
 462                 }
 463                 rec_ptr = r->next;
 464         }
 465         return 0;
 466 }
 467
 468 static int _tdb_store(struct tdb_context *tdb, TDB_DATA key,
 469                        TDB_DATA dbuf, int flag, uint32_t hash)
 470 {
 471         struct tdb_record rec;
 472         tdb_off_t rec_ptr;
 473         char *p = NULL;
 474         int ret = -1;
 475
 476         /* check for it existing, on insert. */
 477         if (flag == TDB_INSERT) {
 478                 if (tdb_exists_hash(tdb, key, hash)) {
 479                         tdb->ecode = TDB_ERR_EXISTS;
 480                         goto fail;
 481                 }
 482         } else {
 483                 /* first try in-place update, on modify or replace. */
 484                 if (tdb_update_hash(tdb, key, hash, dbuf) == 0) {
 485                         goto done;
 486                 }
 487                 if (tdb->ecode == TDB_ERR_NOEXIST &&
 488                     flag == TDB_MODIFY) {
 489                         /* if the record doesn't exist and we are in TDB_MODIFY mode then
 490                          we should fail the store */
 491                         goto fail;
 492                 }
 493         }
 494         /* reset the error code potentially set by the tdb_update() */
 495         tdb->ecode = TDB_SUCCESS;
 496
 497         /* delete any existing record - if it doesn't exist we don't
 498            care.  Doing this first reduces fragmentation, and avoids
 499            coalescing with `allocated' block before it's updated. */
 500         if (flag != TDB_INSERT)
 501                 tdb_delete_hash(tdb, key, hash);
 502
 503         /* Copy key+value *before* allocating free space in case malloc
 504            fails and we are left with a dead spot in the tdb. */
 505
 506         if (!(p = (char *)malloc(key.dsize + dbuf.dsize))) {
 507                 tdb->ecode = TDB_ERR_OOM;
 508                 goto fail;
 509         }
 510
 511         memcpy(p, key.dptr, key.dsize);
 512         if (dbuf.dsize)
 513                 memcpy(p+key.dsize, dbuf.dptr, dbuf.dsize);
 514
 515         if (tdb->max_dead_records != 0) {
 516                 /*
 517                  * Allow for some dead records per hash chain, look if we can
 518                  * find one that can hold the new record. We need enough space
 519                  * for key, data and tailer. If we find one, we don't have to
 520                  * consult the central freelist.
 521                  */
 522                 rec_ptr = tdb_find_dead(
 523                         tdb, hash, &rec,
 524                         key.dsize + dbuf.dsize + sizeof(tdb_off_t));
 525
 526                 if (rec_ptr != 0) {
 527                         rec.key_len = key.dsize;
 528                         rec.data_len = dbuf.dsize;
 529                         rec.full_hash = hash;
 530                         rec.magic = TDB_MAGIC;
 531                         if (tdb_rec_write(tdb, rec_ptr, &rec) == -1
 532                             || tdb->methods->tdb_write(
 533                                     tdb, rec_ptr + sizeof(rec),
 534                                     p, key.dsize + dbuf.dsize) == -1) {
 535                                 goto fail;
 536                         }
 537                         goto done;
 538                 }
 539         }
 540
 541         /*
 542          * We have to allocate some space from the freelist, so this means we
 543          * have to lock it. Use the chance to purge all the DEAD records from
 544          * the hash chain under the freelist lock.
 545          */
 546
 547         if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
 548                 goto fail;
 549         }
 550
 551         if ((tdb->max_dead_records != 0)
 552             && (tdb_purge_dead(tdb, hash) == -1)) {
 553                 tdb_unlock(tdb, -1, F_WRLCK);
 554                 goto fail;
 555         }
 556
 557         /* we have to allocate some space */
 558         rec_ptr = tdb_allocate(tdb, key.dsize + dbuf.dsize, &rec);
 559
 560         tdb_unlock(tdb, -1, F_WRLCK);
 561
 562         if (rec_ptr == 0) {
 563                 goto fail;
 564         }
 565
 566         /* Read hash top into next ptr */
 567         if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec.next) == -1)
 568                 goto fail;
 569
 570         rec.key_len = key.dsize;
 571         rec.data_len = dbuf.dsize;
 572         rec.full_hash = hash;
 573         rec.magic = TDB_MAGIC;
 574
 575         /* write out and point the top of the hash chain at it */
 576         if (tdb_rec_write(tdb, rec_ptr, &rec) == -1
 577             || tdb->methods->tdb_write(tdb, rec_ptr+sizeof(rec), p, key.dsize+dbuf.dsize)==-1
 578             || tdb_ofs_write(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1) {
 579                 /* Need to tdb_unallocate() here */
 580                 goto fail;
 581         }
 582
 583  done:
 584         ret = 0;
 585  fail:
 586         if (ret == 0) {
 587                 tdb_increment_seqnum(tdb);
 588         }
 589
 590         SAFE_FREE(p);
 591         return ret;
 592 }
 593
 594 /* store an element in the database, replacing any existing element
 595    with the same key
 596
 597    return 0 on success, -1 on failure
 598 */
 599 int tdb_store(struct tdb_context *tdb, TDB_DATA key, TDB_DATA dbuf, int flag)
 600 {
 601         uint32_t hash;
 602         int ret;
 603
 604         if (tdb->read_only || tdb->traverse_read) {
 605                 tdb->ecode = TDB_ERR_RDONLY;
 606                 tdb_trace_2rec_flag_ret(tdb, "tdb_store", key, dbuf, flag, -1);
 607                 return -1;
 608         }
 609
 610         /* find which hash bucket it is in */
 611         hash = tdb->hash_fn(&key);
 612         if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
 613                 return -1;
 614
 615         ret = _tdb_store(tdb, key, dbuf, flag, hash);
 616         tdb_trace_2rec_flag_ret(tdb, "tdb_store", key, dbuf, flag, ret);
 617         tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
 618         return ret;
 619 }
 620
 621 /* Append to an entry. Create if not exist. */
 622 int tdb_append(struct tdb_context *tdb, TDB_DATA key, TDB_DATA new_dbuf)
 623 {
 624         uint32_t hash;
 625         TDB_DATA dbuf;
 626         int ret = -1;
 627
 628         /* find which hash bucket it is in */
 629         hash = tdb->hash_fn(&key);
 630         if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
 631                 return -1;
 632
 633         dbuf = _tdb_fetch(tdb, key);
 634
 635         if (dbuf.dptr == NULL) {
 636                 dbuf.dptr = (unsigned char *)malloc(new_dbuf.dsize);
 637         } else {
 638                 unsigned int new_len = dbuf.dsize + new_dbuf.dsize;
 639                 unsigned char *new_dptr;
 640
 641                 /* realloc '0' is special: don't do that. */
 642                 if (new_len == 0)
 643                         new_len = 1;
 644                 new_dptr = (unsigned char *)realloc(dbuf.dptr, new_len);
 645                 if (new_dptr == NULL) {
 646                         free(dbuf.dptr);
 647                 }
 648                 dbuf.dptr = new_dptr;
 649         }
 650
 651         if (dbuf.dptr == NULL) {
 652                 tdb->ecode = TDB_ERR_OOM;
 653                 goto failed;
 654         }
 655
 656         memcpy(dbuf.dptr + dbuf.dsize, new_dbuf.dptr, new_dbuf.dsize);
 657         dbuf.dsize += new_dbuf.dsize;
 658
 659         ret = _tdb_store(tdb, key, dbuf, 0, hash);
 660         tdb_trace_2rec_retrec(tdb, "tdb_append", key, new_dbuf, dbuf);
 661
 662 failed:
 663         tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
 664         SAFE_FREE(dbuf.dptr);
 665         return ret;
 666 }
 667
 668
 669 /*
 670   return the name of the current tdb file
 671   useful for external logging functions
 672 */
 673 const char *tdb_name(struct tdb_context *tdb)
 674 {
 675         return tdb->name;
 676 }
 677
 678 /*
 679   return the underlying file descriptor being used by tdb, or -1
 680   useful for external routines that want to check the device/inode
 681   of the fd
 682 */
 683 int tdb_fd(struct tdb_context *tdb)
 684 {
 685         return tdb->fd;
 686 }
 687
 688 /*
 689   return the current logging function
 690   useful for external tdb routines that wish to log tdb errors
 691 */
 692 tdb_log_func tdb_log_fn(struct tdb_context *tdb)
 693 {
 694         return tdb->log.log_fn;
 695 }
 696
 697
 698 /*
 699   get the tdb sequence number. Only makes sense if the writers opened
 700   with TDB_SEQNUM set. Note that this sequence number will wrap quite
 701   quickly, so it should only be used for a 'has something changed'
 702   test, not for code that relies on the count of the number of changes
 703   made. If you want a counter then use a tdb record.
 704
 705   The aim of this sequence number is to allow for a very lightweight
 706   test of a possible tdb change.
 707 */
 708 int tdb_get_seqnum(struct tdb_context *tdb)
 709 {
 710         tdb_off_t seqnum=0;
 711
 712         tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &seqnum);
 713         return seqnum;
 714 }
 715
 716 int tdb_hash_size(struct tdb_context *tdb)
 717 {
 718         return tdb->header.hash_size;
 719 }
 720
 721 size_t tdb_map_size(struct tdb_context *tdb)
 722 {
 723         return tdb->map_size;
 724 }
 725
 726 int tdb_get_flags(struct tdb_context *tdb)
 727 {
 728         return tdb->flags;
 729 }
 730
 731 void tdb_add_flags(struct tdb_context *tdb, unsigned flags)
 732 {
 733         if ((flags & TDB_ALLOW_NESTING) &&
 734             (flags & TDB_DISALLOW_NESTING)) {
 735                 tdb->ecode = TDB_ERR_NESTING;
 736                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_add_flags: "
 737                         "allow_nesting and disallow_nesting are not allowed together!"));
 738                 return;
 739         }
 740
 741         if (flags & TDB_ALLOW_NESTING) {
 742                 tdb->flags &= ~TDB_DISALLOW_NESTING;
 743         }
 744         if (flags & TDB_DISALLOW_NESTING) {
 745                 tdb->flags &= ~TDB_ALLOW_NESTING;
 746         }
 747
 748         tdb->flags |= flags;
 749 }
 750
 751 void tdb_remove_flags(struct tdb_context *tdb, unsigned flags)
 752 {
 753         if ((flags & TDB_ALLOW_NESTING) &&
 754             (flags & TDB_DISALLOW_NESTING)) {
 755                 tdb->ecode = TDB_ERR_NESTING;
 756                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_remove_flags: "
 757                         "allow_nesting and disallow_nesting are not allowed together!"));
 758                 return;
 759         }
 760
 761         if (flags & TDB_ALLOW_NESTING) {
 762                 tdb->flags |= TDB_DISALLOW_NESTING;
 763         }
 764         if (flags & TDB_DISALLOW_NESTING) {
 765                 tdb->flags |= TDB_ALLOW_NESTING;
 766         }
 767
 768         tdb->flags &= ~flags;
 769 }
 770
 771
 772 /*
 773   enable sequence number handling on an open tdb
 774 */
 775 void tdb_enable_seqnum(struct tdb_context *tdb)
 776 {
 777         tdb->flags |= TDB_SEQNUM;
 778 }
 779
 780
 781 /*
 782   add a region of the file to the freelist. Length is the size of the region in bytes,
 783   which includes the free list header that needs to be added
 784  */
 785 static int tdb_free_region(struct tdb_context *tdb, tdb_off_t offset, ssize_t length)
 786 {
 787         struct tdb_record rec;
 788         if (length <= sizeof(rec)) {
 789                 /* the region is not worth adding */
 790                 return 0;
 791         }
 792         if (length + offset > tdb->map_size) {
 793                 TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_free_region: adding region beyond end of file\n"));
 794                 return -1;
 795         }
 796         memset(&rec,'\0',sizeof(rec));
 797         rec.rec_len = length - sizeof(rec);
 798         if (tdb_free(tdb, offset, &rec) == -1) {
 799                 TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_free_region: failed to add free record\n"));
 800                 return -1;
 801         }
 802         return 0;
 803 }
 804
 805 /*
 806   wipe the entire database, deleting all records. This can be done
 807   very fast by using a global lock. The entire data portion of the
 808   file becomes a single entry in the freelist.
 809
 810   This code carefully steps around the recovery area, leaving it alone
 811  */
 812 int tdb_wipe_all(struct tdb_context *tdb)
 813 {
 814         int i;
 815         tdb_off_t offset = 0;
 816         ssize_t data_len;
 817         tdb_off_t recovery_head;
 818         tdb_len_t recovery_size = 0;
 819
 820         if (tdb_lockall(tdb) != 0) {
 821                 return -1;
 822         }
 823
 824         tdb_trace(tdb, "tdb_wipe_all");
 825
 826         /* see if the tdb has a recovery area, and remember its size
 827            if so. We don't want to lose this as otherwise each
 828            tdb_wipe_all() in a transaction will increase the size of
 829            the tdb by the size of the recovery area */
 830         if (tdb_ofs_read(tdb, TDB_RECOVERY_HEAD, &recovery_head) == -1) {
 831                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_wipe_all: failed to read recovery head\n"));
 832                 goto failed;
 833         }
 834
 835         if (recovery_head != 0) {
 836                 struct tdb_record rec;
 837                 if (tdb->methods->tdb_read(tdb, recovery_head, &rec, sizeof(rec), DOCONV()) == -1) {
 838                         TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_wipe_all: failed to read recovery record\n"));
 839                         return -1;
 840                 }
 841                 recovery_size = rec.rec_len + sizeof(rec);
 842         }
 843
 844         /* wipe the hashes */
 845         for (i=0;i<tdb->header.hash_size;i++) {
 846                 if (tdb_ofs_write(tdb, TDB_HASH_TOP(i), &offset) == -1) {
 847                         TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_wipe_all: failed to write hash %d\n", i));
 848                         goto failed;
 849                 }
 850         }
 851
 852         /* wipe the freelist */
 853         if (tdb_ofs_write(tdb, FREELIST_TOP, &offset) == -1) {
 854                 TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_wipe_all: failed to write freelist\n"));
 855                 goto failed;
 856         }
 857
 858         /* add all the rest of the file to the freelist, possibly leaving a gap
 859            for the recovery area */
 860         if (recovery_size == 0) {
 861                 /* the simple case - the whole file can be used as a freelist */
 862                 data_len = (tdb->map_size - TDB_DATA_START(tdb->header.hash_size));
 863                 if (tdb_free_region(tdb, TDB_DATA_START(tdb->header.hash_size), data_len) != 0) {
 864                         goto failed;
 865                 }
 866         } else {
 867                 /* we need to add two freelist entries - one on either
 868                    side of the recovery area
 869
 870                    Note that we cannot shift the recovery area during
 871                    this operation. Only the transaction.c code may
 872                    move the recovery area or we risk subtle data
 873                    corruption
 874                 */
 875                 data_len = (recovery_head - TDB_DATA_START(tdb->header.hash_size));
 876                 if (tdb_free_region(tdb, TDB_DATA_START(tdb->header.hash_size), data_len) != 0) {
 877                         goto failed;
 878                 }
 879                 /* and the 2nd free list entry after the recovery area - if any */
 880                 data_len = tdb->map_size - (recovery_head+recovery_size);
 881                 if (tdb_free_region(tdb, recovery_head+recovery_size, data_len) != 0) {
 882                         goto failed;
 883                 }
 884         }
 885
 886         if (tdb_unlockall(tdb) != 0) {
 887                 TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_wipe_all: failed to unlock\n"));
 888                 goto failed;
 889         }
 890
 891         return 0;
 892
 893 failed:
 894         tdb_unlockall(tdb);
 895         return -1;
 896 }
 897
 898 struct traverse_state {
 899         bool error;
 900         struct tdb_context *dest_db;
 901 };
 902
 903 /*
 904   traverse function for repacking
 905  */
 906 static int repack_traverse(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *private_data)
 907 {
 908         struct traverse_state *state = (struct traverse_state *)private_data;
 909         if (tdb_store(state->dest_db, key, data, TDB_INSERT) != 0) {
 910                 state->error = true;
 911                 return -1;
 912         }
 913         return 0;
 914 }
 915
 916 /*
 917   repack a tdb
 918  */
 919 int tdb_repack(struct tdb_context *tdb)
 920 {
 921         struct tdb_context *tmp_db;
 922         struct traverse_state state;
 923
 924         tdb_trace(tdb, "tdb_repack");
 925
 926         if (tdb_transaction_start(tdb) != 0) {
 927                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Failed to start transaction\n"));
 928                 return -1;
 929         }
 930
 931         tmp_db = tdb_open("tmpdb", tdb_hash_size(tdb), TDB_INTERNAL, O_RDWR|O_CREAT, 0);
 932         if (tmp_db == NULL) {
 933                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Failed to create tmp_db\n"));
 934                 tdb_transaction_cancel(tdb);
 935                 return -1;
 936         }
 937
 938         state.error = false;
 939         state.dest_db = tmp_db;
 940
 941         if (tdb_traverse_read(tdb, repack_traverse, &state) == -1) {
 942                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Failed to traverse copying out\n"));
 943                 tdb_transaction_cancel(tdb);
 944                 tdb_close(tmp_db);
 945                 return -1;
 946         }
 947
 948         if (state.error) {
 949                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Error during traversal\n"));
 950                 tdb_transaction_cancel(tdb);
 951                 tdb_close(tmp_db);
 952                 return -1;
 953         }
 954
 955         if (tdb_wipe_all(tdb) != 0) {
 956                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Failed to wipe database\n"));
 957                 tdb_transaction_cancel(tdb);
 958                 tdb_close(tmp_db);
 959                 return -1;
 960         }
 961
 962         state.error = false;
 963         state.dest_db = tdb;
 964
 965         if (tdb_traverse_read(tmp_db, repack_traverse, &state) == -1) {
 966                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Failed to traverse copying back\n"));
 967                 tdb_transaction_cancel(tdb);
 968                 tdb_close(tmp_db);
 969                 return -1;
 970         }
 971
 972         if (state.error) {
 973                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Error during second traversal\n"));
 974                 tdb_transaction_cancel(tdb);
 975                 tdb_close(tmp_db);
 976                 return -1;
 977         }
 978
 979         tdb_close(tmp_db);
 980
 981         if (tdb_transaction_commit(tdb) != 0) {
 982                 TDB_LOG((tdb, TDB_DEBUG_FATAL, __location__ " Failed to commit\n"));
 983                 return -1;
 984         }
 985
 986         return 0;
 987 }
 988
 989 #ifdef TDB_TRACE
 990 static void tdb_trace_write(struct tdb_context *tdb, const char *str)
 991 {
 992         if (write(tdb->tracefd, str, strlen(str)) != strlen(str)) {
 993                 close(tdb->tracefd);
 994                 tdb->tracefd = -1;
 995         }
 996 }
 997
 998 static void tdb_trace_start(struct tdb_context *tdb)
 999 {
1000         tdb_off_t seqnum=0;
1001         char msg[sizeof(tdb_off_t) * 4 + 1];
1002
1003         tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &seqnum);
1004         snprintf(msg, sizeof(msg), "%u ", seqnum);
1005         tdb_trace_write(tdb, msg);
1006 }
1007
1008 static void tdb_trace_end(struct tdb_context *tdb)
1009 {
1010         tdb_trace_write(tdb, "\n");
1011 }
1012
1013 static void tdb_trace_end_ret(struct tdb_context *tdb, int ret)
1014 {
1015         char msg[sizeof(ret) * 4 + 4];
1016         snprintf(msg, sizeof(msg), " = %i\n", ret);
1017         tdb_trace_write(tdb, msg);
1018 }
1019
1020 static void tdb_trace_record(struct tdb_context *tdb, TDB_DATA rec)
1021 {
1022         char msg[20 + rec.dsize*2], *p;
1023         unsigned int i;
1024
1025         /* We differentiate zero-length records from non-existent ones. */
1026         if (rec.dptr == NULL) {
1027                 tdb_trace_write(tdb, " NULL");
1028                 return;
1029         }
1030
1031         /* snprintf here is purely cargo-cult programming. */
1032         p = msg;
1033         p += snprintf(p, sizeof(msg), " %zu:", rec.dsize);
1034         for (i = 0; i < rec.dsize; i++)
1035                 p += snprintf(p, 2, "%02x", rec.dptr[i]);
1036
1037         tdb_trace_write(tdb, msg);
1038 }
1039
1040 void tdb_trace(struct tdb_context *tdb, const char *op)
1041 {
1042         tdb_trace_start(tdb);
1043         tdb_trace_write(tdb, op);
1044         tdb_trace_end(tdb);
1045 }
1046
1047 void tdb_trace_seqnum(struct tdb_context *tdb, uint32_t seqnum, const char *op)
1048 {
1049         char msg[sizeof(tdb_off_t) * 4 + 1];
1050
1051         snprintf(msg, sizeof(msg), "%u ", seqnum);
1052         tdb_trace_write(tdb, msg);
1053         tdb_trace_write(tdb, op);
1054         tdb_trace_end(tdb);
1055 }
1056
1057 void tdb_trace_open(struct tdb_context *tdb, const char *op,
1058                     unsigned hash_size, unsigned tdb_flags, unsigned open_flags)
1059 {
1060         char msg[128];
1061
1062         snprintf(msg, sizeof(msg),
1063                  "%s %u 0x%x 0x%x", op, hash_size, tdb_flags, open_flags);
1064         tdb_trace_start(tdb);
1065         tdb_trace_write(tdb, msg);
1066         tdb_trace_end(tdb);
1067 }
1068
1069 void tdb_trace_ret(struct tdb_context *tdb, const char *op, int ret)
1070 {
1071         tdb_trace_start(tdb);
1072         tdb_trace_write(tdb, op);
1073         tdb_trace_end_ret(tdb, ret);
1074 }
1075
1076 void tdb_trace_retrec(struct tdb_context *tdb, const char *op, TDB_DATA ret)
1077 {
1078         tdb_trace_start(tdb);
1079         tdb_trace_write(tdb, op);
1080         tdb_trace_write(tdb, " =");
1081         tdb_trace_record(tdb, ret);
1082         tdb_trace_end(tdb);
1083 }
1084
1085 void tdb_trace_1rec(struct tdb_context *tdb, const char *op,
1086                     TDB_DATA rec)
1087 {
1088         tdb_trace_start(tdb);
1089         tdb_trace_write(tdb, op);
1090         tdb_trace_record(tdb, rec);
1091         tdb_trace_end(tdb);
1092 }
1093
1094 void tdb_trace_1rec_ret(struct tdb_context *tdb, const char *op,
1095                         TDB_DATA rec, int ret)
1096 {
1097         tdb_trace_start(tdb);
1098         tdb_trace_write(tdb, op);
1099         tdb_trace_record(tdb, rec);
1100         tdb_trace_end_ret(tdb, ret);
1101 }
1102
1103 void tdb_trace_1rec_retrec(struct tdb_context *tdb, const char *op,
1104                            TDB_DATA rec, TDB_DATA ret)
1105 {
1106         tdb_trace_start(tdb);
1107         tdb_trace_write(tdb, op);
1108         tdb_trace_record(tdb, rec);
1109         tdb_trace_write(tdb, " =");
1110         tdb_trace_record(tdb, ret);
1111         tdb_trace_end(tdb);
1112 }
1113
1114 void tdb_trace_2rec_flag_ret(struct tdb_context *tdb, const char *op,
1115                              TDB_DATA rec1, TDB_DATA rec2, unsigned flag,
1116                              int ret)
1117 {
1118         char msg[1 + sizeof(ret) * 4];
1119
1120         snprintf(msg, sizeof(msg), " %#x", flag);
1121         tdb_trace_start(tdb);
1122         tdb_trace_write(tdb, op);
1123         tdb_trace_record(tdb, rec1);
1124         tdb_trace_record(tdb, rec2);
1125         tdb_trace_write(tdb, msg);
1126         tdb_trace_end_ret(tdb, ret);
1127 }
1128
1129 void tdb_trace_2rec_retrec(struct tdb_context *tdb, const char *op,
1130                            TDB_DATA rec1, TDB_DATA rec2, TDB_DATA ret)
1131 {
1132         tdb_trace_start(tdb);
1133         tdb_trace_write(tdb, op);
1134         tdb_trace_record(tdb, rec1);
1135         tdb_trace_record(tdb, rec2);
1136         tdb_trace_write(tdb, " =");
1137         tdb_trace_record(tdb, ret);
1138         tdb_trace_end(tdb);
1139 }
1140 #endif