source3/tdb/tdb.c

   1 /*
   2    Unix SMB/Netbios implementation.
   3    Version 3.0
   4    Samba database functions
   5    Copyright (C) Andrew Tridgell              1999-2000
   6    Copyright (C) Luke Kenneth Casson Leighton      2000
   7    Copyright (C) Paul `Rusty' Russell              2000
   8    Copyright (C) Jeremy Allison                    2000
   9
  10    This program is free software; you can redistribute it and/or modify
  11    it under the terms of the GNU General Public License as published by
  12    the Free Software Foundation; either version 2 of the License, or
  13    (at your option) any later version.
  14
  15    This program is distributed in the hope that it will be useful,
  16    but WITHOUT ANY WARRANTY; without even the implied warranty of
  17    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  18    GNU General Public License for more details.
  19
  20    You should have received a copy of the GNU General Public License
  21    along with this program; if not, write to the Free Software
  22    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  23 */
  24 #ifdef STANDALONE
  25 #if HAVE_CONFIG_H
  26 #include <config.h>
  27 #endif
  28
  29 #include <stdlib.h>
  30 #include <stdio.h>
  31 #include <fcntl.h>
  32 #include <unistd.h>
  33 #include <string.h>
  34 #include <fcntl.h>
  35 #include <errno.h>
  36 #include <sys/mman.h>
  37 #include <sys/stat.h>
  38 #include "tdb.h"
  39 #include "spinlock.h"
  40 #else
  41 #include "includes.h"
  42 #endif
  43
  44 #define TDB_MAGIC_FOOD "TDB file\n"
  45 #define TDB_VERSION (0x26011967 + 6)
  46 #define TDB_MAGIC (0x26011999U)
  47 #define TDB_FREE_MAGIC (~TDB_MAGIC)
  48 #define TDB_DEAD_MAGIC (0xFEE1DEAD)
  49 #define TDB_ALIGNMENT 4
  50 #define MIN_REC_SIZE (2*sizeof(struct list_struct) + TDB_ALIGNMENT)
  51 #define DEFAULT_HASH_SIZE 131
  52 #define TDB_PAGE_SIZE 0x2000
  53 #define FREELIST_TOP (sizeof(struct tdb_header))
  54 #define TDB_ALIGN(x,a) (((x) + (a)-1) & ~((a)-1))
  55 #define TDB_BYTEREV(x) (((x<<24)|(x&0xFF00)<<8)|((x>>8)&0xFF00)|(x>>24))
  56 #define TDB_DEAD(r) ((r)->magic == TDB_DEAD_MAGIC)
  57 #define TDB_BAD_MAGIC(r) ((r)->magic != TDB_MAGIC && !TDB_DEAD(r))
  58 #define TDB_HASH_TOP(hash) (FREELIST_TOP + (BUCKET(hash)+1)*sizeof(tdb_off))
  59
  60 /* lock offsets */
  61 #define GLOBAL_LOCK 0
  62 #define ACTIVE_LOCK 4
  63
  64 #ifndef MAP_FILE
  65 #define MAP_FILE 0
  66 #endif
  67
  68 #define BUCKET(hash) ((hash) % tdb->header.hash_size)
  69 TDB_DATA tdb_null;
  70
  71 /* all contexts, to ensure no double-opens (fcntl locks don't nest!) */
  72 static TDB_CONTEXT *tdbs = NULL;
  73
  74 static void *tdb_munmap(void *ptr, tdb_len size)
  75 {
  76 #ifdef HAVE_MMAP
  77         munmap(ptr, size);
  78 #endif
  79         return NULL;
  80 }
  81
  82 static void *tdb_mmap(tdb_len size, int readonly, int fd)
  83 {
  84         void *ret = NULL;
  85 #ifdef HAVE_MMAP
  86         ret = mmap(NULL, size, PROT_READ | (readonly ? 0 : PROT_WRITE), MAP_SHARED|MAP_FILE, fd, 0);
  87 #endif
  88         return ret;
  89 }
  90
  91 /* Endian conversion: we only ever deal with 4 byte quantities */
  92 static void *convert(void *buf, u32 size)
  93 {
  94         u32 i, *p = buf;
  95         for (i = 0; i < size / 4; i++) p[i] = TDB_BYTEREV(p[i]);
  96         return buf;
  97 }
  98 #define DOCONV() (tdb->flags & TDB_CONVERT)
  99 #define CONVERT(x) (DOCONV() ? convert(&x, sizeof(x)) : &x)
 100
 101 /* the body of the database is made of one list_struct for the free space
 102    plus a separate data list for each hash value */
 103 struct list_struct {
 104         tdb_off next; /* offset of the next record in the list */
 105         tdb_len rec_len; /* total byte length of record */
 106         tdb_len key_len; /* byte length of key */
 107         tdb_len data_len; /* byte length of data */
 108         u32 full_hash; /* the full 32 bit hash of the key */
 109         u32 magic;   /* try to catch errors */
 110         /* the following union is implied:
 111            union {
 112               char record[rec_len];
 113               struct {
 114                 char key[key_len];
 115                 char data[data_len];
 116               }
 117               u32 totalsize; (tailer)
 118            } */
 119 };
 120
 121 /* a byte range locking function - return 0 on success
 122    this functions locks/unlocks 1 byte at the specified offset */
 123 static int tdb_brlock(TDB_CONTEXT *tdb, tdb_off offset,
 124                       int rw_type, int lck_type)
 125 {
 126         struct flock fl;
 127
 128         if (tdb->flags & TDB_NOLOCK) return 0;
 129         if (tdb->read_only) return -1;
 130
 131         fl.l_type = rw_type;
 132         fl.l_whence = SEEK_SET;
 133         fl.l_start = offset;
 134         fl.l_len = 1;
 135         fl.l_pid = 0;
 136
 137         if (fcntl(tdb->fd,lck_type,&fl)) return TDB_ERRCODE(TDB_ERR_LOCK, -1);
 138         return 0;
 139 }
 140
 141 /* lock a list in the database. list -1 is the alloc list */
 142 static int tdb_lock(TDB_CONTEXT *tdb, int list, int ltype)
 143 {
 144         if (list < -1 || list >= (int)tdb->header.hash_size) return -1;
 145         if (tdb->flags & TDB_NOLOCK) return 0;
 146
 147         /* Since fcntl locks don't nest, we do a lock for the first one,
 148            and simply bump the count for future ones */
 149         if (tdb->locked[list+1].count == 0) {
 150                 if (tdb->header.rwlocks) {
 151                         if (tdb_spinlock(tdb, list, ltype)) return -1;
 152                 } else if (tdb_brlock(tdb,FREELIST_TOP+4*list,ltype,F_SETLKW))
 153                         return -1;
 154                 tdb->locked[list+1].ltype = ltype;
 155         }
 156         tdb->locked[list+1].count++;
 157         return 0;
 158 }
 159
 160 /* unlock the database: returns void because it's too late for errors. */
 161 static void tdb_unlock(TDB_CONTEXT *tdb, int list, int ltype)
 162 {
 163         if (tdb->flags & TDB_NOLOCK) return;
 164
 165         /* Sanity checks */
 166         if (list < -1 || list >= (int)tdb->header.hash_size) return;
 167         if (tdb->locked[list+1].count==0) return;
 168
 169         if (tdb->locked[list+1].count == 1) {
 170                 /* Down to last nested lock: unlock underneath */
 171                 if (tdb->header.rwlocks) tdb_spinunlock(tdb, list, ltype);
 172                 else tdb_brlock(tdb, FREELIST_TOP+4*list, F_UNLCK, F_SETLKW);
 173         }
 174         tdb->locked[list+1].count--;
 175 }
 176
 177 /* This is based on the hash agorithm from gdbm */
 178 static u32 tdb_hash(TDB_DATA *key)
 179 {
 180         u32 value;      /* Used to compute the hash value.  */
 181         u32   i;        /* Used to cycle through random values. */
 182
 183         /* Set the initial value from the key size. */
 184         for (value = 0x238F13AF * key->dsize, i=0; i < key->dsize; i++)
 185                 value = (value + (key->dptr[i] << (i*5 % 24)));
 186
 187         return (1103515243 * value + 12345);
 188 }
 189
 190 /* check for an out of bounds access - if it is out of bounds then
 191    see if the database has been expanded by someone else and expand
 192    if necessary */
 193 static int tdb_oob(TDB_CONTEXT *tdb, tdb_off offset)
 194 {
 195         struct stat st;
 196         if (offset <= tdb->map_size) return 0;
 197         if (tdb->flags & TDB_INTERNAL) return 0;
 198
 199         fstat(tdb->fd, &st);
 200         if (st.st_size <= (size_t)offset) return TDB_ERRCODE(TDB_ERR_IO, -1);
 201
 202         /* Unmap, update size, remap */
 203         if (tdb->map_ptr) tdb->map_ptr=tdb_munmap(tdb->map_ptr, tdb->map_size);
 204         tdb->map_size = st.st_size;
 205         if (!(tdb->flags & TDB_NOMMAP))
 206                 tdb->map_ptr = tdb_mmap(tdb->map_size, tdb->read_only,tdb->fd);
 207         return 0;
 208 }
 209
 210 /* write a lump of data at a specified offset */
 211 static int tdb_write(TDB_CONTEXT *tdb, tdb_off off, void *buf, tdb_len len)
 212 {
 213         if (tdb_oob(tdb, off + len) != 0) return -1;
 214
 215         if (tdb->map_ptr) memcpy(off + (char *)tdb->map_ptr, buf, len);
 216         else if (lseek(tdb->fd, off, SEEK_SET) != off
 217                  || write(tdb->fd, buf, len) != (ssize_t)len)
 218                 return TDB_ERRCODE(TDB_ERR_IO, -1);
 219         return 0;
 220 }
 221
 222 /* read a lump of data at a specified offset, maybe convert */
 223 static int tdb_read(TDB_CONTEXT *tdb,tdb_off off,void *buf,tdb_len len,int cv)
 224 {
 225         if (tdb_oob(tdb, off + len) != 0) return -1;
 226
 227         if (tdb->map_ptr) memcpy(buf, off + (char *)tdb->map_ptr, len);
 228         else if (lseek(tdb->fd, off, SEEK_SET) != off
 229                  || read(tdb->fd, buf, len) != (ssize_t)len)
 230                 return TDB_ERRCODE(TDB_ERR_IO, -1);
 231         if (cv) convert(buf, len);
 232         return 0;
 233 }
 234
 235 /* read a lump of data, allocating the space for it */
 236 static char *tdb_alloc_read(TDB_CONTEXT *tdb, tdb_off offset, tdb_len len)
 237 {
 238         char *buf;
 239
 240         if (!(buf = malloc(len))) return TDB_ERRCODE(TDB_ERR_OOM, buf);
 241         if (tdb_read(tdb, offset, buf, len, 0) == -1) {
 242                 free(buf);
 243                 return NULL;
 244         }
 245         return buf;
 246 }
 247
 248 /* read/write a tdb_off */
 249 static int ofs_read(TDB_CONTEXT *tdb, tdb_off offset, tdb_off *d)
 250 {
 251         return tdb_read(tdb, offset, (char*)d, sizeof(*d), DOCONV());
 252 }
 253 static int ofs_write(TDB_CONTEXT *tdb, tdb_off offset, tdb_off *d)
 254 {
 255         tdb_off off = *d;
 256         return tdb_write(tdb, offset, CONVERT(off), sizeof(*d));
 257 }
 258
 259 /* read/write a record */
 260 static int rec_read(TDB_CONTEXT *tdb, tdb_off offset, struct list_struct *rec)
 261 {
 262         if (tdb_read(tdb, offset, rec, sizeof(*rec),DOCONV()) == -1) return -1;
 263         if (TDB_BAD_MAGIC(rec)) return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
 264         return tdb_oob(tdb, rec->next);
 265 }
 266 static int rec_write(TDB_CONTEXT *tdb, tdb_off offset, struct list_struct *rec)
 267 {
 268         struct list_struct r = *rec;
 269         return tdb_write(tdb, offset, CONVERT(r), sizeof(r));
 270 }
 271
 272 /* read a freelist record and check for simple errors */
 273 static int rec_free_read(TDB_CONTEXT *tdb, tdb_off off, struct list_struct*rec)
 274 {
 275         if (tdb_read(tdb, off, rec, sizeof(*rec),DOCONV()) == -1) return -1;
 276         if (rec->magic != TDB_FREE_MAGIC) {
 277 #ifdef TDB_DEBUG
 278                 printf("bad magic 0x%08x at offset %d\n",
 279                         rec->magic, off);
 280 #endif
 281                 return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
 282         }
 283         if (tdb_oob(tdb, rec->next) != 0) return -1;
 284         return 0;
 285 }
 286
 287 /* update a record tailer (must hold allocation lock) */
 288 static int update_tailer(TDB_CONTEXT *tdb, tdb_off offset,
 289                          const struct list_struct *rec)
 290 {
 291         tdb_off totalsize;
 292
 293         /* Offset of tailer from record header */
 294         totalsize = sizeof(*rec) + rec->rec_len;
 295         return ofs_write(tdb, offset + totalsize - sizeof(tdb_off),
 296                          &totalsize);
 297 }
 298
 299 #ifdef TDB_DEBUG
 300 void tdb_printfreelist(TDB_CONTEXT *tdb)
 301 {
 302         tdb_off offset, rec_ptr, last_ptr;
 303         struct list_struct rec, lastrec, newrec;
 304
 305         tdb_lock(tdb, -1, F_WRLCK);
 306
 307         last_ptr = 0;
 308         offset = FREELIST_TOP;
 309
 310         /* read in the freelist top */
 311         if (ofs_read(tdb, offset, &rec_ptr) == -1) {
 312                 return;
 313         }
 314
 315         printf("freelist top=[0x%08x]\n", rec_ptr );
 316         while (rec_ptr) {
 317                 if (tdb_read(tdb, rec_ptr, (char *)&rec, sizeof(rec), DOCONV()) == -1) {
 318                         return;
 319                 }
 320
 321                 if (rec.magic != TDB_FREE_MAGIC) {
 322                         printf("bad magic 0x%08x in free list\n", rec.magic);
 323                         return;
 324                 }
 325
 326                 printf("entry offset=[0x%08x], rec.rec_len = [0x%08x]\n", rec.next, rec.rec_len );
 327
 328                 /* move to the next record */
 329                 rec_ptr = rec.next;
 330         }
 331
 332         tdb_unlock(tdb, -1, F_WRLCK);
 333 }
 334 #endif
 335
 336 /* Remove an element from the freelist.  Must have alloc lock. */
 337 static int remove_from_freelist(TDB_CONTEXT *tdb, tdb_off off, tdb_off next)
 338 {
 339         tdb_off last_ptr, i;
 340
 341         /* read in the freelist top */
 342         last_ptr = FREELIST_TOP;
 343         while (ofs_read(tdb, last_ptr, &i) != -1 && i != 0) {
 344                 if (i == off) {
 345                         /* We've found it! */
 346                         return ofs_write(tdb, last_ptr, &next);
 347                 }
 348                 /* Follow chain (next offset is at start of record) */
 349                 last_ptr = i;
 350         }
 351         return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
 352 }
 353
 354 /* Add an element into the freelist. Merge adjacent records if
 355    neccessary. */
 356 static int tdb_free(TDB_CONTEXT *tdb, tdb_off offset, struct list_struct *rec)
 357 {
 358         tdb_off right, left;
 359
 360         /* Allocation and tailer lock */
 361         if (tdb_lock(tdb, -1, F_WRLCK) != 0) return -1;
 362
 363         /* Look right first (I'm an Australian, dammit) */
 364         right = offset + sizeof(*rec) + rec->rec_len;
 365         if (tdb_oob(tdb, right + sizeof(*rec)) == 0) {
 366                 struct list_struct r;
 367
 368                 if (tdb_read(tdb, right, &r, sizeof(r), DOCONV()) == -1)
 369                         goto fail;
 370
 371                 /* If it's free, expand to include it. */
 372                 if (r.magic == TDB_FREE_MAGIC) {
 373                         if (remove_from_freelist(tdb, right, r.next) == -1)
 374                                 goto fail;
 375                         rec->rec_len += sizeof(r) + r.rec_len;
 376                 }
 377         }
 378
 379         /* Look left */
 380         left = offset - 4;
 381         if (left > TDB_HASH_TOP(tdb->header.hash_size-1)) {
 382                 struct list_struct l;
 383                 tdb_off leftsize;
 384
 385                 /* Read in tailer and jump back to header */
 386                 if (ofs_read(tdb, left, &leftsize) == -1) goto fail;
 387                 left = offset - leftsize;
 388
 389                 /* Now read in record */
 390                 if (tdb_read(tdb, left, &l, sizeof(l), DOCONV()) == -1)
 391                         goto fail;
 392
 393                 /* If it's free, expand to include it. */
 394                 if (l.magic == TDB_FREE_MAGIC) {
 395                         if (remove_from_freelist(tdb, left, l.next) == -1)
 396                                 goto fail;
 397                         offset = left;
 398                         rec->rec_len += leftsize;
 399                 }
 400         }
 401         if (update_tailer(tdb, offset, rec) == -1) goto fail;
 402
 403         /* Now, prepend to free list */
 404         rec->magic = TDB_FREE_MAGIC;
 405
 406         if (ofs_read(tdb, FREELIST_TOP, &rec->next) == -1) goto fail;
 407         if (rec_write(tdb, offset, rec) == -1) goto fail;
 408         if (ofs_write(tdb, FREELIST_TOP, &offset) == -1) goto fail;
 409
 410         /* And we're done. */
 411         tdb_unlock(tdb, -1, F_WRLCK);
 412         return 0;
 413
 414  fail:
 415         tdb_unlock(tdb, -1, F_WRLCK);
 416         return -1;
 417 }
 418
 419 /* expand the database at least size bytes by expanding the underlying
 420    file and doing the mmap again if necessary */
 421 static int tdb_expand(TDB_CONTEXT *tdb, tdb_off size)
 422 {
 423         struct list_struct rec;
 424         tdb_off offset;
 425         char b = 0;
 426
 427         if (tdb_lock(tdb, -1, F_WRLCK) == -1) return 0;
 428
 429         /* must know about any previous expansions by another process */
 430         tdb_oob(tdb, tdb->map_size + 1);
 431
 432         /* always make room for at least 10 more records, and round
 433            the database up to a multiple of TDB_PAGE_SIZE */
 434         size = TDB_ALIGN(tdb->map_size + size*10, TDB_PAGE_SIZE) - tdb->map_size;
 435
 436         /* expand the file itself */
 437         if (!(tdb->flags & TDB_INTERNAL)) {
 438                 lseek(tdb->fd, tdb->map_size + size - 1, SEEK_SET);
 439                 if (write(tdb->fd, &b, 1) != 1) goto fail;
 440         }
 441
 442         if (!(tdb->flags & TDB_INTERNAL) && tdb->map_ptr)
 443                 tdb->map_ptr = tdb_munmap(tdb->map_ptr, tdb->map_size);
 444
 445         tdb->map_size += size;
 446
 447         if (tdb->flags & TDB_INTERNAL)
 448                 tdb->map_ptr = realloc(tdb->map_ptr, tdb->map_size);
 449
 450         /* form a new freelist record */
 451         memset(&rec,'\0',sizeof(rec));
 452         rec.rec_len = size - sizeof(rec);
 453
 454         /* link it into the free list */
 455         offset = tdb->map_size - size;
 456         if (tdb_free(tdb, offset, &rec) == -1) goto fail;
 457
 458         if (!(tdb->flags & TDB_NOMMAP))
 459                 tdb->map_ptr = tdb_mmap(tdb->map_size, 0, tdb->fd);
 460
 461         tdb_unlock(tdb, -1, F_WRLCK);
 462         return 0;
 463  fail:
 464         tdb_unlock(tdb, -1, F_WRLCK);
 465         return -1;
 466 }
 467
 468 /* allocate some space from the free list. The offset returned points
 469    to a unconnected list_struct within the database with room for at
 470    least length bytes of total data
 471
 472    0 is returned if the space could not be allocated
 473  */
 474 static tdb_off tdb_allocate(TDB_CONTEXT *tdb, tdb_len length,
 475                             struct list_struct *rec)
 476 {
 477         tdb_off rec_ptr, last_ptr, newrec_ptr;
 478         struct list_struct newrec;
 479
 480         if (tdb_lock(tdb, -1, F_WRLCK) == -1) return 0;
 481
 482         /* Extra bytes required for tailer */
 483         length += sizeof(tdb_off);
 484
 485  again:
 486         last_ptr = FREELIST_TOP;
 487
 488         /* read in the freelist top */
 489         if (ofs_read(tdb, FREELIST_TOP, &rec_ptr) == -1) goto fail;
 490
 491         /* keep looking until we find a freelist record big enough */
 492         while (rec_ptr) {
 493                 if (rec_free_read(tdb, rec_ptr, rec) == -1) goto fail;
 494
 495                 if (rec->rec_len >= length) {
 496                         /* found it - now possibly split it up  */
 497                         if (rec->rec_len > length + MIN_REC_SIZE) {
 498                                 /* Length of left piece */
 499                                 length = TDB_ALIGN(length, TDB_ALIGNMENT);
 500
 501                                 /* Right piece to go on free list */
 502                                 newrec.rec_len = rec->rec_len
 503                                         - (sizeof(*rec) + length);
 504                                 newrec_ptr = rec_ptr + sizeof(*rec) + length;
 505
 506                                 /* And left record is shortened */
 507                                 rec->rec_len = length;
 508                         } else
 509                                 newrec_ptr = 0;
 510
 511                         /* Remove allocated record from the free list */
 512                         if (ofs_write(tdb, last_ptr, &rec->next) == -1)
 513                                 goto fail;
 514
 515                         /* Update header: do this before we drop alloc
 516                            lock, otherwise tdb_free() might try to
 517                            merge with us, thinking we're free.
 518                            (Thanks Jeremy Allison). */
 519                         rec->magic = TDB_MAGIC;
 520                         if (rec_write(tdb, rec_ptr, rec) == -1)
 521                                 goto fail;
 522
 523                         /* Did we create new block? */
 524                         if (newrec_ptr) {
 525                                 /* Update allocated record tailer (we
 526                                    shortened it). */
 527                                 if (update_tailer(tdb, rec_ptr, rec) == -1)
 528                                         goto fail;
 529
 530                                 /* Free new record */
 531                                 if (tdb_free(tdb, newrec_ptr, &newrec) == -1)
 532                                         goto fail;
 533                         }
 534
 535                         /* all done - return the new record offset */
 536                         tdb_unlock(tdb, -1, F_WRLCK);
 537                         return rec_ptr;
 538                 }
 539                 /* move to the next record */
 540                 last_ptr = rec_ptr;
 541                 rec_ptr = rec->next;
 542         }
 543         /* we didn't find enough space. See if we can expand the
 544            database and if we can then try again */
 545         if (tdb_expand(tdb, length + sizeof(*rec)) == 0) goto again;
 546  fail:
 547         tdb_unlock(tdb, -1, F_WRLCK);
 548         return 0;
 549 }
 550
 551 /* initialise a new database with a specified hash size */
 552 static int tdb_new_database(TDB_CONTEXT *tdb, int hash_size)
 553 {
 554         struct tdb_header *newdb;
 555         int size, ret;
 556
 557         /* We make it up in memory, then write it out if not internal */
 558         size = sizeof(struct tdb_header) + (hash_size+1)*sizeof(tdb_off);
 559         if (!(newdb = calloc(size, 1))) return TDB_ERRCODE(TDB_ERR_OOM, -1);
 560
 561         /* Fill in the header */
 562         newdb->version = TDB_VERSION;
 563         newdb->hash_size = hash_size;
 564 #ifdef USE_SPINLOCKS
 565         newdb->rwlocks = size;
 566 #endif
 567         if (tdb->flags & TDB_INTERNAL) {
 568                 tdb->map_size = size;
 569                 tdb->map_ptr = (char *)newdb;
 570                 memcpy(&tdb->header, newdb, sizeof(tdb->header));
 571                 /* Convert the `ondisk' version if asked. */
 572                 CONVERT(*newdb);
 573                 return 0;
 574         }
 575         lseek(tdb->fd, 0, SEEK_SET);
 576         ftruncate(tdb->fd, 0);
 577         /* This creates an endian-converted header, as if read from disk */
 578         CONVERT(*newdb);
 579         memcpy(&tdb->header, newdb, sizeof(tdb->header));
 580         /* Don't endian-convert the magic food! */
 581         memcpy(newdb->magic_food, TDB_MAGIC_FOOD, strlen(TDB_MAGIC_FOOD)+1);
 582         if (write(tdb->fd, newdb, size) != size) ret = -1;
 583         else ret = tdb_create_rwlocks(tdb->fd, hash_size);
 584
 585         free(newdb);
 586         return ret;
 587 }
 588
 589 /* Returns 0 on fail.  On success, return offset of record, and fills
 590    in rec */
 591 static tdb_off tdb_find(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash,
 592                         struct list_struct *r)
 593 {
 594         tdb_off rec_ptr;
 595
 596         /* read in the hash top */
 597         if (ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1) return 0;
 598
 599         /* keep looking until we find the right record */
 600         while (rec_ptr) {
 601                 if (rec_read(tdb, rec_ptr, r) == -1) return 0;
 602
 603                 if (!TDB_DEAD(r) && hash==r->full_hash && key.dsize==r->key_len) {
 604                         char *k;
 605                         /* a very likely hit - read the key */
 606                         k = tdb_alloc_read(tdb, rec_ptr + sizeof(*r),
 607                                            r->key_len);
 608                         if (!k) return 0;
 609
 610                         if (memcmp(key.dptr, k, key.dsize) == 0) {
 611                                 free(k);
 612                                 return rec_ptr;
 613                         }
 614                         free(k);
 615                 }
 616                 rec_ptr = r->next;
 617         }
 618         return TDB_ERRCODE(TDB_ERR_NOEXIST, 0);
 619 }
 620
 621 /* If they do lockkeys, check that this hash is one they locked */
 622 static int tdb_keylocked(TDB_CONTEXT *tdb, u32 hash)
 623 {
 624         u32 i;
 625         if (!tdb->lockedkeys) return 1;
 626         for (i = 0; i < tdb->lockedkeys[0]; i++)
 627                 if (tdb->lockedkeys[i+1] == hash) return 1;
 628         return TDB_ERRCODE(TDB_ERR_NOLOCK, 0);
 629 }
 630
 631 /* As tdb_find, but if you succeed, keep the lock */
 632 static tdb_off tdb_find_lock(TDB_CONTEXT *tdb, TDB_DATA key, int locktype,
 633                              struct list_struct *rec)
 634 {
 635         u32 hash, rec_ptr;
 636
 637         hash = tdb_hash(&key);
 638         if (!tdb_keylocked(tdb, hash)) return 0;
 639         if (tdb_lock(tdb, BUCKET(hash), locktype) == -1) return 0;
 640         if (!(rec_ptr = tdb_find(tdb, key, hash, rec)))
 641                 tdb_unlock(tdb, BUCKET(hash), locktype);
 642         return rec_ptr;
 643 }
 644
 645 enum TDB_ERROR tdb_error(TDB_CONTEXT *tdb)
 646 {
 647         return tdb->ecode;
 648 }
 649
 650 static struct tdb_errname {
 651         enum TDB_ERROR ecode; const char *estring;
 652 } emap[] = { {TDB_SUCCESS, "Success"},
 653              {TDB_ERR_CORRUPT, "Corrupt database"},
 654              {TDB_ERR_IO, "IO Error"},
 655              {TDB_ERR_LOCK, "Locking error"},
 656              {TDB_ERR_OOM, "Out of memory"},
 657              {TDB_ERR_EXISTS, "Record exists"},
 658              {TDB_ERR_NOLOCK, "Lock exists on other keys"},
 659              {TDB_ERR_NOEXIST, "Record does not exist"} };
 660
 661 /* Error string for the last tdb error */
 662 const char *tdb_errorstr(TDB_CONTEXT *tdb)
 663 {
 664         u32 i;
 665         for (i = 0; i < sizeof(emap) / sizeof(struct tdb_errname); i++)
 666                 if (tdb->ecode == emap[i].ecode) return emap[i].estring;
 667         return "Invalid error code";
 668 }
 669
 670 /* update an entry in place - this only works if the new data size
 671    is <= the old data size and the key exists.
 672    on failure return -1
 673 */
 674 static int tdb_update(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA dbuf)
 675 {
 676         struct list_struct rec;
 677         tdb_off rec_ptr;
 678         int ret = -1;
 679
 680         /* find entry */
 681         if (!(rec_ptr = tdb_find_lock(tdb, key, F_WRLCK, &rec))) return -1;
 682
 683         /* must be long enough key, data and tailer */
 684         if (rec.rec_len < key.dsize + dbuf.dsize + sizeof(tdb_off)) goto out;
 685
 686         if (tdb_write(tdb, rec_ptr + sizeof(rec) + rec.key_len,
 687                       dbuf.dptr, dbuf.dsize) == -1)
 688                 goto out;
 689
 690         if (dbuf.dsize != rec.data_len) {
 691                 /* update size */
 692                 rec.data_len = dbuf.dsize;
 693                 ret = rec_write(tdb, rec_ptr, &rec);
 694         }
 695         else ret = 0;
 696  out:
 697         tdb_unlock(tdb, BUCKET(rec.full_hash), F_WRLCK);
 698         return ret;
 699 }
 700
 701 /* find an entry in the database given a key */
 702 TDB_DATA tdb_fetch(TDB_CONTEXT *tdb, TDB_DATA key)
 703 {
 704         tdb_off rec_ptr;
 705         struct list_struct rec;
 706         TDB_DATA ret;
 707
 708         /* find which hash bucket it is in */
 709         if (!(rec_ptr = tdb_find_lock(tdb,key,F_RDLCK,&rec))) return tdb_null;
 710
 711         ret.dptr = tdb_alloc_read(tdb, rec_ptr + sizeof(rec) + rec.key_len,
 712                                   rec.data_len);
 713         ret.dsize = rec.data_len;
 714         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
 715         return ret;
 716 }
 717
 718 /* check if an entry in the database exists
 719
 720    note that 1 is returned if the key is found and 0 is returned if not found
 721    this doesn't match the conventions in the rest of this module, but is
 722    compatible with gdbm
 723 */
 724 int tdb_exists(TDB_CONTEXT *tdb, TDB_DATA key)
 725 {
 726         struct list_struct rec;
 727
 728         if (tdb_find_lock(tdb, key, F_RDLCK, &rec) == 0) return 0;
 729         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
 730         return 1;
 731 }
 732
 733 /* record lock stops delete underneath */
 734 static int lock_record(TDB_CONTEXT *tdb, tdb_off off)
 735 {
 736         return off ? tdb_brlock(tdb, off, F_RDLCK, F_SETLKW) : 0;
 737 }
 738 /* write locks override our own fcntl readlocks, so check it here */
 739 static int write_lock_record(TDB_CONTEXT *tdb, tdb_off off)
 740 {
 741         struct tdb_traverse_lock *i;
 742         for (i = &tdb->travlocks; i; i = i->next) if (i->off == off) return -1;
 743         return tdb_brlock(tdb, off, F_WRLCK, F_SETLK);
 744 }
 745 static int write_unlock_record(TDB_CONTEXT *tdb, tdb_off off)
 746 {
 747         return tdb_brlock(tdb, off, F_UNLCK, F_SETLK);
 748 }
 749 /* fcntl locks don't stack: avoid unlocking someone else's */
 750 static int unlock_record(TDB_CONTEXT *tdb, tdb_off off)
 751 {
 752         struct tdb_traverse_lock *i;
 753         u32 count = 0;
 754
 755         if (off == 0) return 0;
 756         for (i = &tdb->travlocks; i; i = i->next) if (i->off == off) count++;
 757         return (count == 1 ? tdb_brlock(tdb, off, F_UNLCK, F_SETLKW) : 0);
 758 }
 759
 760 /* actually delete an entry in the database given the offset */
 761 static int do_delete(TDB_CONTEXT *tdb, tdb_off rec_ptr, struct list_struct*rec)
 762 {
 763         tdb_off last_ptr, i;
 764         struct list_struct lastrec;
 765
 766         if (tdb->read_only) return -1;
 767
 768         if (write_lock_record(tdb, rec_ptr) == -1) {
 769                 /* Someone traversing here: mark it as dead */
 770                 rec->magic = TDB_DEAD_MAGIC;
 771                 return rec_write(tdb, rec_ptr, rec);
 772         }
 773         write_unlock_record(tdb, rec_ptr);
 774
 775         /* find previous record in hash chain */
 776         if (ofs_read(tdb, TDB_HASH_TOP(rec->full_hash), &i) == -1) return -1;
 777         for (last_ptr = 0; i != rec_ptr; last_ptr = i, i = lastrec.next)
 778                 if (rec_read(tdb, i, &lastrec) == -1) return -1;
 779
 780         /* unlink it: next ptr is at start of record. */
 781         if (last_ptr == 0) last_ptr = TDB_HASH_TOP(rec->full_hash);
 782         if (ofs_write(tdb, last_ptr, &rec->next) == -1) return -1;
 783
 784         /* recover the space */
 785         if (tdb_free(tdb, rec_ptr, rec) == -1) return -1;
 786         return 0;
 787 }
 788
 789 /* Uses traverse lock: 0 = finish, -1 = error, other = record offset */
 790 static int tdb_next_lock(TDB_CONTEXT *tdb, struct tdb_traverse_lock *tlock,
 791                          struct list_struct *rec)
 792 {
 793         int want_next = (tlock->off != 0);
 794
 795         /* No traversal allows if you've called tdb_lockkeys() */
 796         if (tdb->lockedkeys) return TDB_ERRCODE(TDB_ERR_NOLOCK, -1);
 797
 798         /* Lock each chain from the start one. */
 799         for (; tlock->hash < tdb->header.hash_size; tlock->hash++) {
 800                 if (tdb_lock(tdb, tlock->hash, F_WRLCK) == -1) return -1;
 801
 802                 /* No previous record?  Start at top of chain. */
 803                 if (!tlock->off) {
 804                         if (ofs_read(tdb, TDB_HASH_TOP(tlock->hash),
 805                                      &tlock->off) == -1)
 806                                 goto fail;
 807                 } else {
 808                         /* Otherwise unlock the previous record. */
 809                         unlock_record(tdb, tlock->off);
 810                 }
 811
 812                 if (want_next) {
 813                         /* We have offset of old record: grab next */
 814                         if (rec_read(tdb, tlock->off, rec) == -1) goto fail;
 815                         tlock->off = rec->next;
 816                 }
 817
 818                 /* Iterate through chain */
 819                 while( tlock->off) {
 820                         tdb_off current;
 821                         if (rec_read(tdb, tlock->off, rec) == -1) goto fail;
 822                         if (!TDB_DEAD(rec)) {
 823                                 /* Woohoo: we found one! */
 824                                 lock_record(tdb, tlock->off);
 825                                 return tlock->off;
 826                         }
 827                         /* Try to clean dead ones from old traverses */
 828                         current = tlock->off;
 829                         tlock->off = rec->next;
 830                         do_delete(tdb, current, rec);
 831                 }
 832                 tdb_unlock(tdb, tlock->hash, F_WRLCK);
 833                 want_next = 0;
 834         }
 835         /* We finished iteration without finding anything */
 836         return TDB_ERRCODE(TDB_SUCCESS, 0);
 837
 838  fail:
 839         tlock->off = 0;
 840         tdb_unlock(tdb, tlock->hash, F_WRLCK);
 841         return -1;
 842 }
 843
 844 /* traverse the entire database - calling fn(tdb, key, data) on each element.
 845    return -1 on error or the record count traversed
 846    if fn is NULL then it is not called
 847    a non-zero return value from fn() indicates that the traversal should stop
 848   */
 849 int tdb_traverse(TDB_CONTEXT *tdb, tdb_traverse_func fn, void *state)
 850 {
 851         TDB_DATA key, dbuf;
 852         struct list_struct rec;
 853         struct tdb_traverse_lock tl = { NULL, 0, 0 };
 854         int ret, count = 0;
 855
 856         /* This was in the initializaton, above, but the IRIX compiler
 857          * did not like it.  crh
 858          */
 859         tl.next = tdb->travlocks.next;
 860
 861         /* fcntl locks don't stack: beware traverse inside traverse */
 862         tdb->travlocks.next = &tl;
 863
 864         /* tdb_next_lock places locks on the record returned, and its chain */
 865         while ((ret = tdb_next_lock(tdb, &tl, &rec)) > 0) {
 866                 count++;
 867                 /* now read the full record */
 868                 key.dptr = tdb_alloc_read(tdb, tl.off + sizeof(rec),
 869                                           rec.key_len + rec.data_len);
 870                 if (!key.dptr) {
 871                         tdb_unlock(tdb, tl.hash, F_WRLCK);
 872                         unlock_record(tdb, tl.off);
 873                         tdb->travlocks.next = tl.next;
 874                         return -1;
 875                 }
 876                 key.dsize = rec.key_len;
 877                 dbuf.dptr = key.dptr + rec.key_len;
 878                 dbuf.dsize = rec.data_len;
 879
 880                 /* Drop chain lock, call out */
 881                 tdb_unlock(tdb, tl.hash, F_WRLCK);
 882                 if (fn && fn(tdb, key, dbuf, state)) {
 883                         /* They want us to terminate traversal */
 884                         unlock_record(tdb, tl.off);
 885                         tdb->travlocks.next = tl.next;
 886                         free(key.dptr);
 887                         return count;
 888                 }
 889                 free(key.dptr);
 890         }
 891         tdb->travlocks.next = tl.next;
 892         if (ret < 0) return -1;
 893         else return count;
 894 }
 895
 896 /* find the first entry in the database and return its key */
 897 TDB_DATA tdb_firstkey(TDB_CONTEXT *tdb)
 898 {
 899         TDB_DATA key;
 900         struct list_struct rec;
 901
 902         /* release any old lock */
 903         unlock_record(tdb, tdb->travlocks.off);
 904         tdb->travlocks.off = tdb->travlocks.hash = 0;
 905
 906         if (tdb_next_lock(tdb, &tdb->travlocks, &rec) <= 0) return tdb_null;
 907         /* now read the key */
 908         key.dsize = rec.key_len;
 909         key.dptr =tdb_alloc_read(tdb,tdb->travlocks.off+sizeof(rec),key.dsize);
 910         tdb_unlock(tdb, BUCKET(tdb->travlocks.hash), F_WRLCK);
 911         return key;
 912 }
 913
 914 /* find the next entry in the database, returning its key */
 915 TDB_DATA tdb_nextkey(TDB_CONTEXT *tdb, TDB_DATA oldkey)
 916 {
 917         u32 oldhash;
 918         TDB_DATA key = tdb_null;
 919         struct list_struct rec;
 920         char *k = NULL;
 921
 922         /* Is locked key the old key?  If so, traverse will be reliable. */
 923         if (tdb->travlocks.off) {
 924                 if (tdb_lock(tdb,tdb->travlocks.hash,F_WRLCK)) return tdb_null;
 925                 if (rec_read(tdb, tdb->travlocks.off, &rec) == -1
 926                     || !(k = tdb_alloc_read(tdb,tdb->travlocks.off+sizeof(rec),
 927                                             rec.key_len))
 928                     || memcmp(k, oldkey.dptr, oldkey.dsize) != 0) {
 929                         /* No, it wasn't: unlock it and start from scratch */
 930                         unlock_record(tdb, tdb->travlocks.off);
 931                         tdb_unlock(tdb, tdb->travlocks.hash, F_WRLCK);
 932                         tdb->travlocks.off = 0;
 933                 }
 934
 935                 if (k)
 936                         free(k);
 937         }
 938
 939         if (!tdb->travlocks.off) {
 940                 /* No previous element: do normal find, and lock record */
 941                 tdb->travlocks.off = tdb_find_lock(tdb, oldkey, F_WRLCK, &rec);
 942                 if (!tdb->travlocks.off) return tdb_null;
 943                 tdb->travlocks.hash = BUCKET(rec.full_hash);
 944                 lock_record(tdb, tdb->travlocks.off);
 945         }
 946         oldhash = tdb->travlocks.hash;
 947
 948         /* Grab next record: locks chain and returned record,
 949            unlocks old record */
 950         if (tdb_next_lock(tdb, &tdb->travlocks, &rec) > 0) {
 951                 key.dsize = rec.key_len;
 952                 key.dptr = tdb_alloc_read(tdb, tdb->travlocks.off+sizeof(rec),
 953                                           key.dsize);
 954                 /* Unlock the chain of this new record */
 955                 tdb_unlock(tdb, tdb->travlocks.hash, F_WRLCK);
 956         }
 957         /* Unlock the chain of old record */
 958         tdb_unlock(tdb, BUCKET(oldhash), F_WRLCK);
 959         return key;
 960 }
 961
 962 /* delete an entry in the database given a key */
 963 int tdb_delete(TDB_CONTEXT *tdb, TDB_DATA key)
 964 {
 965         tdb_off rec_ptr;
 966         struct list_struct rec;
 967         int ret;
 968
 969         if (!(rec_ptr = tdb_find_lock(tdb, key, F_WRLCK, &rec))) return -1;
 970         ret = do_delete(tdb, rec_ptr, &rec);
 971         tdb_unlock(tdb, BUCKET(rec.full_hash), F_WRLCK);
 972         return ret;
 973 }
 974
 975 /* store an element in the database, replacing any existing element
 976    with the same key
 977
 978    return 0 on success, -1 on failure
 979 */
 980 int tdb_store(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA dbuf, int flag)
 981 {
 982         struct list_struct rec;
 983         u32 hash;
 984         tdb_off rec_ptr;
 985         char *p = NULL;
 986         int ret = 0;
 987
 988         /* find which hash bucket it is in */
 989         hash = tdb_hash(&key);
 990         if (!tdb_keylocked(tdb, hash)) return -1;
 991         if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1) return -1;
 992
 993         /* check for it existing, on insert. */
 994         if (flag == TDB_INSERT) {
 995                 if (tdb_exists(tdb, key)) {
 996                         tdb->ecode = TDB_ERR_EXISTS;
 997                         goto fail;
 998                 }
 999         } else {
1000                 /* first try in-place update, on modify or replace. */
1001                 if (tdb_update(tdb, key, dbuf) == 0) goto out;
1002                 if (flag == TDB_MODIFY && tdb->ecode == TDB_ERR_NOEXIST)
1003                         goto fail;
1004         }
1005         /* reset the error code potentially set by the tdb_update() */
1006         tdb->ecode = TDB_SUCCESS;
1007
1008         /* delete any existing record - if it doesn't exist we don't
1009            care.  Doing this first reduces fragmentation, and avoids
1010            coalescing with `allocated' block before it's updated. */
1011         if (flag != TDB_INSERT) tdb_delete(tdb, key);
1012
1013         /* now we're into insert / modify / replace of a record which
1014          * we know could not be optimised by an in-place store (for
1015          * various reasons).  */
1016         if (!(rec_ptr = tdb_allocate(tdb, key.dsize + dbuf.dsize, &rec)))
1017                 goto fail;
1018
1019         /* Read hash top into next ptr */
1020         if (ofs_read(tdb, TDB_HASH_TOP(hash), &rec.next) == -1) goto fail;
1021
1022         rec.key_len = key.dsize;
1023         rec.data_len = dbuf.dsize;
1024         rec.full_hash = hash;
1025         rec.magic = TDB_MAGIC;
1026
1027         if (!(p = (char *)malloc(key.dsize + dbuf.dsize))) {
1028                 tdb->ecode = TDB_ERR_OOM;
1029                 goto fail;
1030         }
1031
1032         memcpy(p, key.dptr, key.dsize);
1033         memcpy(p+key.dsize, dbuf.dptr, dbuf.dsize);
1034         /* write out and point the top of the hash chain at it */
1035         if (rec_write(tdb, rec_ptr, &rec) == -1
1036             || tdb_write(tdb, rec_ptr+sizeof(rec), p, key.dsize+dbuf.dsize)==-1
1037             || ofs_write(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1) {
1038         fail:
1039                 ret = -1;
1040         }
1041  out:
1042         free(p);
1043         tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
1044         return ret;
1045 }
1046
1047 /* open the database, creating it if necessary
1048
1049    The open_flags and mode are passed straight to the open call on the
1050    database file. A flags value of O_WRONLY is invalid. The hash size
1051    is advisory, use zero for a default value.
1052
1053    return is NULL on error */
1054 TDB_CONTEXT *tdb_open(char *name, int hash_size, int tdb_flags,
1055                       int open_flags, mode_t mode)
1056 {
1057         TDB_CONTEXT tdb, *ret, *i;
1058         struct stat st;
1059         int rev = 0, locked;
1060
1061         memset(&tdb, 0, sizeof(tdb));
1062         tdb.fd = -1;
1063         tdb.name = NULL;
1064         tdb.map_ptr = NULL;
1065         tdb.lockedkeys = NULL;
1066         tdb.flags = tdb_flags;
1067
1068         if ((open_flags & O_ACCMODE) == O_WRONLY) goto fail;
1069         if (hash_size == 0) hash_size = DEFAULT_HASH_SIZE;
1070         if ((open_flags & O_ACCMODE) == O_RDONLY) {
1071                 tdb.read_only = 1;
1072                 /* read only databases don't do locking */
1073                 tdb.flags |= TDB_NOLOCK;
1074         }
1075
1076         /* internal databases don't mmap or lock, and start off cleared */
1077         if (tdb.flags & TDB_INTERNAL) {
1078                 tdb.flags |= (TDB_NOLOCK | TDB_NOMMAP);
1079                 tdb.flags &= ~TDB_CLEAR_IF_FIRST;
1080                 tdb_new_database(&tdb, hash_size);
1081                 goto internal;
1082         }
1083
1084         if ((tdb.fd = open(name, open_flags, mode)) == -1) goto fail;
1085
1086         /* ensure there is only one process initialising at once */
1087         tdb_brlock(&tdb, GLOBAL_LOCK, F_WRLCK, F_SETLKW);
1088
1089         /* we need to zero database if we are the only one with it open */
1090         if ((locked = (tdb_brlock(&tdb, ACTIVE_LOCK, F_WRLCK, F_SETLK) == 0))
1091             && (tdb_flags & TDB_CLEAR_IF_FIRST)) {
1092                 open_flags |= O_CREAT;
1093                 ftruncate(tdb.fd, 0);
1094         }
1095
1096         if (read(tdb.fd, &tdb.header, sizeof(tdb.header)) != sizeof(tdb.header)
1097             || strcmp(tdb.header.magic_food, TDB_MAGIC_FOOD) != 0
1098             || (tdb.header.version != TDB_VERSION
1099                 && !(rev = (tdb.header.version==TDB_BYTEREV(TDB_VERSION))))) {
1100                 /* its not a valid database - possibly initialise it */
1101                 if (!(open_flags & O_CREAT)
1102                     || tdb_new_database(&tdb, hash_size) == -1) goto fail;
1103                 rev = (tdb.flags & TDB_CONVERT);
1104         }
1105         if (!rev) tdb.flags &= ~TDB_CONVERT;
1106         else {
1107                 tdb.flags |= TDB_CONVERT;
1108                 convert(&tdb.header, sizeof(tdb.header));
1109         }
1110         fstat(tdb.fd, &st);
1111         /* Is it already in the open list?  If so, fail. */
1112         for (i = tdbs; i; i = i->next) {
1113                 if (i->device == st.st_dev && i->inode == st.st_ino) {
1114                         errno = EBUSY;
1115                         close(tdb.fd);
1116                         return NULL;
1117                 }
1118         }
1119
1120         /* map the database and fill in the return structure */
1121         tdb.name = (char *)strdup(name);
1122         tdb.map_size = st.st_size;
1123         tdb.device = st.st_dev;
1124         tdb.inode = st.st_ino;
1125         tdb.locked = calloc(tdb.header.hash_size+1, sizeof(tdb.locked[0]));
1126         if (!tdb.locked) goto fail;
1127         if (!(tdb.flags & TDB_NOMMAP))
1128                 tdb.map_ptr = tdb_mmap(st.st_size, tdb.read_only, tdb.fd);
1129         if (locked) {
1130                 tdb_clear_spinlocks(&tdb);
1131                 tdb_brlock(&tdb, ACTIVE_LOCK, F_UNLCK, F_SETLK);
1132         }
1133         /* leave this lock in place to indicate it's in use */
1134         tdb_brlock(&tdb, ACTIVE_LOCK, F_RDLCK, F_SETLKW);
1135
1136  internal:
1137         if (!(ret = malloc(sizeof(tdb)))) goto fail;
1138         *ret = tdb;
1139         tdb_brlock(&tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW);
1140         ret->next = tdbs;
1141         tdbs = ret;
1142         return ret;
1143
1144  fail:
1145         if (tdb.name) free(tdb.name);
1146         if (tdb.fd != -1) close(tdb.fd);
1147         if (tdb.map_ptr) tdb_munmap(tdb.map_ptr, tdb.map_size);
1148         return NULL;
1149 }
1150
1151 /* close a database */
1152 int tdb_close(TDB_CONTEXT *tdb)
1153 {
1154         TDB_CONTEXT **i;
1155         int ret = 0;
1156
1157         if (tdb->map_ptr) {
1158                 if (tdb->flags & TDB_INTERNAL) free(tdb->map_ptr);
1159                 else tdb_munmap(tdb->map_ptr, tdb->map_size);
1160         }
1161         if (tdb->name) free(tdb->name);
1162         if (tdb->fd != -1) {
1163                 ret = close(tdb->fd);
1164         }
1165         if (tdb->locked) free(tdb->locked);
1166         if (tdb->lockedkeys) free(tdb->lockedkeys);
1167
1168         /* Remove from contexts list */
1169         for (i = &tdbs; *i; i = &(*i)->next) {
1170                 if (*i == tdb) {
1171                         *i = tdb->next;
1172                         break;
1173                 }
1174         }
1175
1176         memset(tdb, 0, sizeof(*tdb));
1177         free(tdb);
1178
1179         return ret;
1180 }
1181
1182 /* lock/unlock entire database */
1183 int tdb_lockall(TDB_CONTEXT *tdb)
1184 {
1185         u32 i;
1186
1187         /* There are no locks on read-only dbs */
1188         if (tdb->read_only) return TDB_ERRCODE(TDB_ERR_LOCK, -1);
1189         if (tdb->lockedkeys) return TDB_ERRCODE(TDB_ERR_NOLOCK, -1);
1190         for (i = 0; i < tdb->header.hash_size; i++)
1191                 if (tdb_lock(tdb, i, F_WRLCK))
1192                         break;
1193
1194         /* If error, release locks we have... */
1195         if (i < tdb->header.hash_size) {
1196                 u32 j;
1197
1198                 for ( j = 0; j < i; j++)
1199                         tdb_unlock(tdb, j, F_WRLCK);
1200                 return TDB_ERRCODE(TDB_ERR_NOLOCK, -1);
1201         }
1202
1203         return 0;
1204 }
1205 void tdb_unlockall(TDB_CONTEXT *tdb)
1206 {
1207         u32 i;
1208         for (i=0; i < tdb->header.hash_size; i++) tdb_unlock(tdb, i, F_WRLCK);
1209 }
1210
1211 int tdb_lockkeys(TDB_CONTEXT *tdb, u32 number, TDB_DATA keys[])
1212 {
1213         u32 i, j, hash;
1214
1215         /* Can't lock more keys if already locked */
1216         if (tdb->lockedkeys) return TDB_ERRCODE(TDB_ERR_NOLOCK, -1);
1217         if (!(tdb->lockedkeys = malloc(sizeof(u32) * (number+1))))
1218                 return TDB_ERRCODE(TDB_ERR_OOM, -1);
1219         /* First number in array is # keys */
1220         tdb->lockedkeys[0] = number;
1221
1222         /* Insertion sort by bucket */
1223         for (i = 0; i < number; i++) {
1224                 hash = tdb_hash(&keys[i]);
1225                 for (j = 0;
1226                      j < i && BUCKET(tdb->lockedkeys[j+1]) < BUCKET(hash);
1227                      j++);
1228                 memmove(&tdb->lockedkeys[j+2], &tdb->lockedkeys[j+1],
1229                         sizeof(u32) * (i-j));
1230                 tdb->lockedkeys[j+1] = hash;
1231         }
1232         /* Finally, lock in order */
1233         for (i = 0; i < number; i++)
1234                 if (tdb_lock(tdb, i, F_WRLCK))
1235                         break;
1236
1237         /* If error, release locks we have... */
1238         if (i < number) {
1239                 for ( j = 0; j < i; j++)
1240                         tdb_unlock(tdb, j, F_WRLCK);
1241                 free(tdb->lockedkeys);
1242                 tdb->lockedkeys = NULL;
1243                 return TDB_ERRCODE(TDB_ERR_NOLOCK, -1);
1244         }
1245         return 0;
1246 }
1247
1248 /* Unlock the keys previously locked by tdb_lockkeys() */
1249 void tdb_unlockkeys(TDB_CONTEXT *tdb)
1250 {
1251         u32 i;
1252         for (i = 0; i < tdb->lockedkeys[0]; i++)
1253                 tdb_unlock(tdb, tdb->lockedkeys[i+1], F_WRLCK);
1254         free(tdb->lockedkeys);
1255         tdb->lockedkeys = NULL;
1256 }
1257
1258 /* lock/unlock one hash chain. This is meant to be used to reduce
1259    contention - it cannot guarantee how many records will be locked */
1260 int tdb_chainlock(TDB_CONTEXT *tdb, TDB_DATA key)
1261 {
1262         return tdb_lock(tdb, BUCKET(tdb_hash(&key)), F_WRLCK);
1263 }
1264 void tdb_chainunlock(TDB_CONTEXT *tdb, TDB_DATA key)
1265 {
1266         tdb_unlock(tdb, BUCKET(tdb_hash(&key)), F_WRLCK);
1267 }