Home | History | Annotate | Download | only in ext2fs
      1 /*
      2 URL: svn://svnanon.samba.org/samba/branches/SAMBA_4_0/source/lib/tdb/common
      3 Rev: 23590
      4 Last Changed Date: 2007-06-22 13:36:10 -0400 (Fri, 22 Jun 2007)
      5 */
      6  /*
      7    trivial database library - standalone version
      8 
      9    Copyright (C) Andrew Tridgell              1999-2005
     10    Copyright (C) Jeremy Allison               2000-2006
     11    Copyright (C) Paul `Rusty' Russell         2000
     12 
     13      ** NOTE! The following LGPL license applies to the tdb
     14      ** library. This does NOT imply that all of Samba is released
     15      ** under the LGPL
     16 
     17    This library is free software; you can redistribute it and/or
     18    modify it under the terms of the GNU Lesser General Public
     19    License as published by the Free Software Foundation; either
     20    version 2 of the License, or (at your option) any later version.
     21 
     22    This library is distributed in the hope that it will be useful,
     23    but WITHOUT ANY WARRANTY; without even the implied warranty of
     24    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
     25    Lesser General Public License for more details.
     26 
     27    You should have received a copy of the GNU Lesser General Public
     28    License along with this library; if not, write to the Free Software
     29    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
     30 */
     31 
     32 #ifdef CONFIG_STAND_ALONE
     33 #define HAVE_MMAP
     34 #define HAVE_STRDUP
     35 #define HAVE_SYS_MMAN_H
     36 #define HAVE_UTIME_H
     37 #define HAVE_UTIME
     38 #endif
     39 #define _XOPEN_SOURCE 600
     40 
     41 #include <unistd.h>
     42 #include <stdio.h>
     43 #include <stdlib.h>
     44 #include <stdarg.h>
     45 #include <stddef.h>
     46 #include <errno.h>
     47 #include <string.h>
     48 #ifdef HAVE_SYS_SELECT_H
     49 #include <sys/select.h>
     50 #endif
     51 #include <sys/time.h>
     52 #include <sys/types.h>
     53 #include <time.h>
     54 #ifdef HAVE_UTIME_H
     55 #include <utime.h>
     56 #endif
     57 #include <sys/stat.h>
     58 #include <sys/file.h>
     59 #include <fcntl.h>
     60 
     61 #ifdef HAVE_SYS_MMAN_H
     62 #include <sys/mman.h>
     63 #endif
     64 
     65 #ifndef MAP_FILE
     66 #define MAP_FILE 0
     67 #endif
     68 
     69 #ifndef MAP_FAILED
     70 #define MAP_FAILED ((void *)-1)
     71 #endif
     72 
     73 #ifndef HAVE_STRDUP
     74 #define strdup rep_strdup
     75 static char *rep_strdup(const char *s)
     76 {
     77 	char *ret;
     78 	int length;
     79 	if (!s)
     80 		return NULL;
     81 
     82 	if (!length)
     83 		length = strlen(s);
     84 
     85 	ret = malloc(length + 1);
     86 	if (ret) {
     87 		strncpy(ret, s, length);
     88 		ret[length] = '\0';
     89 	}
     90 	return ret;
     91 }
     92 #endif
     93 
     94 #ifndef PRINTF_ATTRIBUTE
     95 #if (__GNUC__ >= 3) && (__GNUC_MINOR__ >= 1 )
     96 /** Use gcc attribute to check printf fns.  a1 is the 1-based index of
     97  * the parameter containing the format, and a2 the index of the first
     98  * argument. Note that some gcc 2.x versions don't handle this
     99  * properly **/
    100 #define PRINTF_ATTRIBUTE(a1, a2) __attribute__ ((format (__printf__, a1, a2)))
    101 #else
    102 #define PRINTF_ATTRIBUTE(a1, a2)
    103 #endif
    104 #endif
    105 
    106 typedef int bool;
    107 
    108 #include "tdb.h"
    109 
    110 #ifndef u32
    111 #define u32 unsigned
    112 #endif
    113 
    114 #ifndef HAVE_GETPAGESIZE
    115 #define getpagesize() 0x2000
    116 #endif
    117 
    118 typedef u32 tdb_len_t;
    119 typedef u32 tdb_off_t;
    120 
    121 #ifndef offsetof
    122 #define offsetof(t,f) ((unsigned int)&((t *)0)->f)
    123 #endif
    124 
    125 #define TDB_MAGIC_FOOD "TDB file\n"
    126 #define TDB_VERSION (0x26011967 + 6)
    127 #define TDB_MAGIC (0x26011999U)
    128 #define TDB_FREE_MAGIC (~TDB_MAGIC)
    129 #define TDB_DEAD_MAGIC (0xFEE1DEAD)
    130 #define TDB_RECOVERY_MAGIC (0xf53bc0e7U)
    131 #define TDB_ALIGNMENT 4
    132 #define MIN_REC_SIZE (2*sizeof(struct list_struct) + TDB_ALIGNMENT)
    133 #define DEFAULT_HASH_SIZE 131
    134 #define FREELIST_TOP (sizeof(struct tdb_header))
    135 #define TDB_ALIGN(x,a) (((x) + (a)-1) & ~((a)-1))
    136 #define TDB_BYTEREV(x) (((((x)&0xff)<<24)|((x)&0xFF00)<<8)|(((x)>>8)&0xFF00)|((x)>>24))
    137 #define TDB_DEAD(r) ((r)->magic == TDB_DEAD_MAGIC)
    138 #define TDB_BAD_MAGIC(r) ((r)->magic != TDB_MAGIC && !TDB_DEAD(r))
    139 #define TDB_HASH_TOP(hash) (FREELIST_TOP + (BUCKET(hash)+1)*sizeof(tdb_off_t))
    140 #define TDB_HASHTABLE_SIZE(tdb) ((tdb->header.hash_size+1)*sizeof(tdb_off_t))
    141 #define TDB_DATA_START(hash_size) TDB_HASH_TOP(hash_size-1)
    142 #define TDB_RECOVERY_HEAD offsetof(struct tdb_header, recovery_start)
    143 #define TDB_SEQNUM_OFS    offsetof(struct tdb_header, sequence_number)
    144 #define TDB_PAD_BYTE 0x42
    145 #define TDB_PAD_U32  0x42424242
    146 
    147 /* NB assumes there is a local variable called "tdb" that is the
    148  * current context, also takes doubly-parenthesized print-style
    149  * argument. */
    150 #define TDB_LOG(x) tdb->log.log_fn x
    151 
    152 /* lock offsets */
    153 #define GLOBAL_LOCK      0
    154 #define ACTIVE_LOCK      4
    155 #define TRANSACTION_LOCK 8
    156 
    157 /* free memory if the pointer is valid and zero the pointer */
    158 #ifndef SAFE_FREE
    159 #define SAFE_FREE(x) do { if ((x) != NULL) {free(x); (x)=NULL;} } while(0)
    160 #endif
    161 
    162 #define BUCKET(hash) ((hash) % tdb->header.hash_size)
    163 
    164 #define DOCONV() (tdb->flags & TDB_CONVERT)
    165 #define CONVERT(x) (DOCONV() ? tdb_convert(&x, sizeof(x)) : &x)
    166 
    167 
    168 /* the body of the database is made of one list_struct for the free space
    169    plus a separate data list for each hash value */
    170 struct list_struct {
    171 	tdb_off_t next; /* offset of the next record in the list */
    172 	tdb_len_t rec_len; /* total byte length of record */
    173 	tdb_len_t key_len; /* byte length of key */
    174 	tdb_len_t data_len; /* byte length of data */
    175 	u32 full_hash; /* the full 32 bit hash of the key */
    176 	u32 magic;   /* try to catch errors */
    177 	/* the following union is implied:
    178 		union {
    179 			char record[rec_len];
    180 			struct {
    181 				char key[key_len];
    182 				char data[data_len];
    183 			}
    184 			u32 totalsize; (tailer)
    185 		}
    186 	*/
    187 };
    188 
    189 
    190 /* this is stored at the front of every database */
    191 struct tdb_header {
    192 	char magic_food[32]; /* for /etc/magic */
    193 	u32 version; /* version of the code */
    194 	u32 hash_size; /* number of hash entries */
    195 	tdb_off_t rwlocks; /* obsolete - kept to detect old formats */
    196 	tdb_off_t recovery_start; /* offset of transaction recovery region */
    197 	tdb_off_t sequence_number; /* used when TDB_SEQNUM is set */
    198 	tdb_off_t reserved[29];
    199 };
    200 
    201 struct tdb_lock_type {
    202 	int list;
    203 	u32 count;
    204 	u32 ltype;
    205 };
    206 
    207 struct tdb_traverse_lock {
    208 	struct tdb_traverse_lock *next;
    209 	u32 off;
    210 	u32 hash;
    211 	int lock_rw;
    212 };
    213 
    214 
    215 struct tdb_methods {
    216 	int (*tdb_read)(struct tdb_context *, tdb_off_t , void *, tdb_len_t , int );
    217 	int (*tdb_write)(struct tdb_context *, tdb_off_t, const void *, tdb_len_t);
    218 	void (*next_hash_chain)(struct tdb_context *, u32 *);
    219 	int (*tdb_oob)(struct tdb_context *, tdb_off_t , int );
    220 	int (*tdb_expand_file)(struct tdb_context *, tdb_off_t , tdb_off_t );
    221 	int (*tdb_brlock)(struct tdb_context *, tdb_off_t , int, int, int, size_t);
    222 };
    223 
    224 struct tdb_context {
    225 	char *name; /* the name of the database */
    226 	void *map_ptr; /* where it is currently mapped */
    227 	int fd; /* open file descriptor for the database */
    228 	tdb_len_t map_size; /* how much space has been mapped */
    229 	int read_only; /* opened read-only */
    230 	int traverse_read; /* read-only traversal */
    231 	struct tdb_lock_type global_lock;
    232 	int num_lockrecs;
    233 	struct tdb_lock_type *lockrecs; /* only real locks, all with count>0 */
    234 	enum TDB_ERROR ecode; /* error code for last tdb error */
    235 	struct tdb_header header; /* a cached copy of the header */
    236 	u32 flags; /* the flags passed to tdb_open */
    237 	struct tdb_traverse_lock travlocks; /* current traversal locks */
    238 	struct tdb_context *next; /* all tdbs to avoid multiple opens */
    239 	dev_t device;	/* uniquely identifies this tdb */
    240 	ino_t inode;	/* uniquely identifies this tdb */
    241 	struct tdb_logging_context log;
    242 	unsigned int (*hash_fn)(TDB_DATA *key);
    243 	int open_flags; /* flags used in the open - needed by reopen */
    244 	unsigned int num_locks; /* number of chain locks held */
    245 	const struct tdb_methods *methods;
    246 	struct tdb_transaction *transaction;
    247 	int page_size;
    248 	int max_dead_records;
    249 	bool have_transaction_lock;
    250 };
    251 
    252 
    253 /*
    254   internal prototypes
    255 */
    256 static int tdb_munmap(struct tdb_context *tdb);
    257 static void tdb_mmap(struct tdb_context *tdb);
    258 static int tdb_lock(struct tdb_context *tdb, int list, int ltype);
    259 static int tdb_unlock(struct tdb_context *tdb, int list, int ltype);
    260 static int tdb_brlock(struct tdb_context *tdb, tdb_off_t offset, int rw_type, int lck_type, int probe, size_t len);
    261 static int tdb_transaction_lock(struct tdb_context *tdb, int ltype);
    262 static int tdb_transaction_unlock(struct tdb_context *tdb);
    263 static int tdb_brlock_upgrade(struct tdb_context *tdb, tdb_off_t offset, size_t len);
    264 static int tdb_write_lock_record(struct tdb_context *tdb, tdb_off_t off);
    265 static int tdb_write_unlock_record(struct tdb_context *tdb, tdb_off_t off);
    266 static int tdb_ofs_read(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d);
    267 static int tdb_ofs_write(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d);
    268 static void *tdb_convert(void *buf, u32 size);
    269 static int tdb_free(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec);
    270 static tdb_off_t tdb_allocate(struct tdb_context *tdb, tdb_len_t length, struct list_struct *rec);
    271 static int tdb_ofs_read(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d);
    272 static int tdb_ofs_write(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d);
    273 static int tdb_lock_record(struct tdb_context *tdb, tdb_off_t off);
    274 static int tdb_unlock_record(struct tdb_context *tdb, tdb_off_t off);
    275 static int tdb_rec_read(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec);
    276 static int tdb_rec_write(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec);
    277 static int tdb_do_delete(struct tdb_context *tdb, tdb_off_t rec_ptr, struct list_struct *rec);
    278 static unsigned char *tdb_alloc_read(struct tdb_context *tdb, tdb_off_t offset, tdb_len_t len);
    279 static int tdb_parse_data(struct tdb_context *tdb, TDB_DATA key,
    280 		   tdb_off_t offset, tdb_len_t len,
    281 		   int (*parser)(TDB_DATA key, TDB_DATA data,
    282 				 void *private_data),
    283 		   void *private_data);
    284 static tdb_off_t tdb_find_lock_hash(struct tdb_context *tdb, TDB_DATA key, u32 hash, int locktype,
    285 			   struct list_struct *rec);
    286 static void tdb_io_init(struct tdb_context *tdb);
    287 static int tdb_expand(struct tdb_context *tdb, tdb_off_t size);
    288 static int tdb_rec_free_read(struct tdb_context *tdb, tdb_off_t off,
    289 		      struct list_struct *rec);
    290 
    291 
    292 /* file: error.c */
    293 
    294 enum TDB_ERROR tdb_error(struct tdb_context *tdb)
    295 {
    296 	return tdb->ecode;
    297 }
    298 
    299 static struct tdb_errname {
    300 	enum TDB_ERROR ecode; const char *estring;
    301 } emap[] = { {TDB_SUCCESS, "Success"},
    302 	     {TDB_ERR_CORRUPT, "Corrupt database"},
    303 	     {TDB_ERR_IO, "IO Error"},
    304 	     {TDB_ERR_LOCK, "Locking error"},
    305 	     {TDB_ERR_OOM, "Out of memory"},
    306 	     {TDB_ERR_EXISTS, "Record exists"},
    307 	     {TDB_ERR_NOLOCK, "Lock exists on other keys"},
    308 	     {TDB_ERR_EINVAL, "Invalid parameter"},
    309 	     {TDB_ERR_NOEXIST, "Record does not exist"},
    310 	     {TDB_ERR_RDONLY, "write not permitted"} };
    311 
    312 /* Error string for the last tdb error */
    313 const char *tdb_errorstr(struct tdb_context *tdb)
    314 {
    315 	u32 i;
    316 	for (i = 0; i < sizeof(emap) / sizeof(struct tdb_errname); i++)
    317 		if (tdb->ecode == emap[i].ecode)
    318 			return emap[i].estring;
    319 	return "Invalid error code";
    320 }
    321 
    322 /* file: lock.c */
    323 
    324 #define TDB_MARK_LOCK 0x80000000
    325 
    326 /* a byte range locking function - return 0 on success
    327    this functions locks/unlocks 1 byte at the specified offset.
    328 
    329    On error, errno is also set so that errors are passed back properly
    330    through tdb_open().
    331 
    332    note that a len of zero means lock to end of file
    333 */
    334 int tdb_brlock(struct tdb_context *tdb, tdb_off_t offset,
    335 	       int rw_type, int lck_type, int probe, size_t len)
    336 {
    337 	struct flock fl;
    338 	int ret;
    339 
    340 	if (tdb->flags & TDB_NOLOCK) {
    341 		return 0;
    342 	}
    343 
    344 	if ((rw_type == F_WRLCK) && (tdb->read_only || tdb->traverse_read)) {
    345 		tdb->ecode = TDB_ERR_RDONLY;
    346 		return -1;
    347 	}
    348 
    349 	fl.l_type = rw_type;
    350 	fl.l_whence = SEEK_SET;
    351 	fl.l_start = offset;
    352 	fl.l_len = len;
    353 	fl.l_pid = 0;
    354 
    355 	do {
    356 		ret = fcntl(tdb->fd,lck_type,&fl);
    357 	} while (ret == -1 && errno == EINTR);
    358 
    359 	if (ret == -1) {
    360 		/* Generic lock error. errno set by fcntl.
    361 		 * EAGAIN is an expected return from non-blocking
    362 		 * locks. */
    363 		if (!probe && lck_type != F_SETLK) {
    364 			/* Ensure error code is set for log fun to examine. */
    365 			tdb->ecode = TDB_ERR_LOCK;
    366 			TDB_LOG((tdb, TDB_DEBUG_TRACE,"tdb_brlock failed (fd=%d) at offset %d rw_type=%d lck_type=%d len=%d\n",
    367 				 tdb->fd, offset, rw_type, lck_type, (int)len));
    368 		}
    369 		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
    370 	}
    371 	return 0;
    372 }
    373 
    374 
    375 /*
    376   upgrade a read lock to a write lock. This needs to be handled in a
    377   special way as some OSes (such as solaris) have too conservative
    378   deadlock detection and claim a deadlock when progress can be
    379   made. For those OSes we may loop for a while.
    380 */
    381 int tdb_brlock_upgrade(struct tdb_context *tdb, tdb_off_t offset, size_t len)
    382 {
    383 	int count = 1000;
    384 	while (count--) {
    385 		struct timeval tv;
    386 		if (tdb_brlock(tdb, offset, F_WRLCK, F_SETLKW, 1, len) == 0) {
    387 			return 0;
    388 		}
    389 		if (errno != EDEADLK) {
    390 			break;
    391 		}
    392 		/* sleep for as short a time as we can - more portable than usleep() */
    393 		tv.tv_sec = 0;
    394 		tv.tv_usec = 1;
    395 		select(0, NULL, NULL, NULL, &tv);
    396 	}
    397 	TDB_LOG((tdb, TDB_DEBUG_TRACE,"tdb_brlock_upgrade failed at offset %d\n", offset));
    398 	return -1;
    399 }
    400 
    401 
    402 /* lock a list in the database. list -1 is the alloc list */
    403 static int _tdb_lock(struct tdb_context *tdb, int list, int ltype, int op)
    404 {
    405 	struct tdb_lock_type *new_lck;
    406 	int i;
    407 	bool mark_lock = ((ltype & TDB_MARK_LOCK) == TDB_MARK_LOCK);
    408 
    409 	ltype &= ~TDB_MARK_LOCK;
    410 
    411 	/* a global lock allows us to avoid per chain locks */
    412 	if (tdb->global_lock.count &&
    413 	    (ltype == tdb->global_lock.ltype || ltype == F_RDLCK)) {
    414 		return 0;
    415 	}
    416 
    417 	if (tdb->global_lock.count) {
    418 		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
    419 	}
    420 
    421 	if (list < -1 || list >= (int)tdb->header.hash_size) {
    422 		TDB_LOG((tdb, TDB_DEBUG_ERROR,"tdb_lock: invalid list %d for ltype=%d\n",
    423 			   list, ltype));
    424 		return -1;
    425 	}
    426 	if (tdb->flags & TDB_NOLOCK)
    427 		return 0;
    428 
    429 	for (i=0; i<tdb->num_lockrecs; i++) {
    430 		if (tdb->lockrecs[i].list == list) {
    431 			if (tdb->lockrecs[i].count == 0) {
    432 				/*
    433 				 * Can't happen, see tdb_unlock(). It should
    434 				 * be an assert.
    435 				 */
    436 				TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_lock: "
    437 					 "lck->count == 0 for list %d", list));
    438 			}
    439 			/*
    440 			 * Just increment the in-memory struct, posix locks
    441 			 * don't stack.
    442 			 */
    443 			tdb->lockrecs[i].count++;
    444 			return 0;
    445 		}
    446 	}
    447 
    448 	new_lck = (struct tdb_lock_type *)realloc(
    449 		tdb->lockrecs,
    450 		sizeof(*tdb->lockrecs) * (tdb->num_lockrecs+1));
    451 	if (new_lck == NULL) {
    452 		errno = ENOMEM;
    453 		return -1;
    454 	}
    455 	tdb->lockrecs = new_lck;
    456 
    457 	/* Since fcntl locks don't nest, we do a lock for the first one,
    458 	   and simply bump the count for future ones */
    459 	if (!mark_lock &&
    460 	    tdb->methods->tdb_brlock(tdb,FREELIST_TOP+4*list, ltype, op,
    461 				     0, 1)) {
    462 		return -1;
    463 	}
    464 
    465 	tdb->num_locks++;
    466 
    467 	tdb->lockrecs[tdb->num_lockrecs].list = list;
    468 	tdb->lockrecs[tdb->num_lockrecs].count = 1;
    469 	tdb->lockrecs[tdb->num_lockrecs].ltype = ltype;
    470 	tdb->num_lockrecs += 1;
    471 
    472 	return 0;
    473 }
    474 
    475 /* lock a list in the database. list -1 is the alloc list */
    476 int tdb_lock(struct tdb_context *tdb, int list, int ltype)
    477 {
    478 	int ret;
    479 	ret = _tdb_lock(tdb, list, ltype, F_SETLKW);
    480 	if (ret) {
    481 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_lock failed on list %d "
    482 			 "ltype=%d (%s)\n",  list, ltype, strerror(errno)));
    483 	}
    484 	return ret;
    485 }
    486 
    487 /* lock a list in the database. list -1 is the alloc list. non-blocking lock */
    488 int tdb_lock_nonblock(struct tdb_context *tdb, int list, int ltype)
    489 {
    490 	return _tdb_lock(tdb, list, ltype, F_SETLK);
    491 }
    492 
    493 
    494 /* unlock the database: returns void because it's too late for errors. */
    495 	/* changed to return int it may be interesting to know there
    496 	   has been an error  --simo */
    497 int tdb_unlock(struct tdb_context *tdb, int list, int ltype)
    498 {
    499 	int ret = -1;
    500 	int i;
    501 	struct tdb_lock_type *lck = NULL;
    502 	bool mark_lock = ((ltype & TDB_MARK_LOCK) == TDB_MARK_LOCK);
    503 
    504 	ltype &= ~TDB_MARK_LOCK;
    505 
    506 	/* a global lock allows us to avoid per chain locks */
    507 	if (tdb->global_lock.count &&
    508 	    (ltype == tdb->global_lock.ltype || ltype == F_RDLCK)) {
    509 		return 0;
    510 	}
    511 
    512 	if (tdb->global_lock.count) {
    513 		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
    514 	}
    515 
    516 	if (tdb->flags & TDB_NOLOCK)
    517 		return 0;
    518 
    519 	/* Sanity checks */
    520 	if (list < -1 || list >= (int)tdb->header.hash_size) {
    521 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlock: list %d invalid (%d)\n", list, tdb->header.hash_size));
    522 		return ret;
    523 	}
    524 
    525 	for (i=0; i<tdb->num_lockrecs; i++) {
    526 		if (tdb->lockrecs[i].list == list) {
    527 			lck = &tdb->lockrecs[i];
    528 			break;
    529 		}
    530 	}
    531 
    532 	if ((lck == NULL) || (lck->count == 0)) {
    533 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlock: count is 0\n"));
    534 		return -1;
    535 	}
    536 
    537 	if (lck->count > 1) {
    538 		lck->count--;
    539 		return 0;
    540 	}
    541 
    542 	/*
    543 	 * This lock has count==1 left, so we need to unlock it in the
    544 	 * kernel. We don't bother with decrementing the in-memory array
    545 	 * element, we're about to overwrite it with the last array element
    546 	 * anyway.
    547 	 */
    548 
    549 	if (mark_lock) {
    550 		ret = 0;
    551 	} else {
    552 		ret = tdb->methods->tdb_brlock(tdb, FREELIST_TOP+4*list, F_UNLCK,
    553 					       F_SETLKW, 0, 1);
    554 	}
    555 	tdb->num_locks--;
    556 
    557 	/*
    558 	 * Shrink the array by overwriting the element just unlocked with the
    559 	 * last array element.
    560 	 */
    561 
    562 	if (tdb->num_lockrecs > 1) {
    563 		*lck = tdb->lockrecs[tdb->num_lockrecs-1];
    564 	}
    565 	tdb->num_lockrecs -= 1;
    566 
    567 	/*
    568 	 * We don't bother with realloc when the array shrinks, but if we have
    569 	 * a completely idle tdb we should get rid of the locked array.
    570 	 */
    571 
    572 	if (tdb->num_lockrecs == 0) {
    573 		SAFE_FREE(tdb->lockrecs);
    574 	}
    575 
    576 	if (ret)
    577 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlock: An error occurred unlocking!\n"));
    578 	return ret;
    579 }
    580 
    581 /*
    582   get the transaction lock
    583  */
    584 int tdb_transaction_lock(struct tdb_context *tdb, int ltype)
    585 {
    586 	if (tdb->have_transaction_lock || tdb->global_lock.count) {
    587 		return 0;
    588 	}
    589 	if (tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, ltype,
    590 				     F_SETLKW, 0, 1) == -1) {
    591 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_lock: failed to get transaction lock\n"));
    592 		tdb->ecode = TDB_ERR_LOCK;
    593 		return -1;
    594 	}
    595 	tdb->have_transaction_lock = 1;
    596 	return 0;
    597 }
    598 
    599 /*
    600   release the transaction lock
    601  */
    602 int tdb_transaction_unlock(struct tdb_context *tdb)
    603 {
    604 	int ret;
    605 	if (!tdb->have_transaction_lock) {
    606 		return 0;
    607 	}
    608 	ret = tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, F_UNLCK, F_SETLKW, 0, 1);
    609 	if (ret == 0) {
    610 		tdb->have_transaction_lock = 0;
    611 	}
    612 	return ret;
    613 }
    614 
    615 
    616 
    617 
    618 /* lock/unlock entire database */
    619 static int _tdb_lockall(struct tdb_context *tdb, int ltype, int op)
    620 {
    621 	bool mark_lock = ((ltype & TDB_MARK_LOCK) == TDB_MARK_LOCK);
    622 
    623 	ltype &= ~TDB_MARK_LOCK;
    624 
    625 	/* There are no locks on read-only dbs */
    626 	if (tdb->read_only || tdb->traverse_read)
    627 		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
    628 
    629 	if (tdb->global_lock.count && tdb->global_lock.ltype == ltype) {
    630 		tdb->global_lock.count++;
    631 		return 0;
    632 	}
    633 
    634 	if (tdb->global_lock.count) {
    635 		/* a global lock of a different type exists */
    636 		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
    637 	}
    638 
    639 	if (tdb->num_locks != 0) {
    640 		/* can't combine global and chain locks */
    641 		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
    642 	}
    643 
    644 	if (!mark_lock &&
    645 	    tdb->methods->tdb_brlock(tdb, FREELIST_TOP, ltype, op,
    646 				     0, 4*tdb->header.hash_size)) {
    647 		if (op == F_SETLKW) {
    648 			TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_lockall failed (%s)\n", strerror(errno)));
    649 		}
    650 		return -1;
    651 	}
    652 
    653 	tdb->global_lock.count = 1;
    654 	tdb->global_lock.ltype = ltype;
    655 
    656 	return 0;
    657 }
    658 
    659 
    660 
    661 /* unlock entire db */
    662 static int _tdb_unlockall(struct tdb_context *tdb, int ltype)
    663 {
    664 	bool mark_lock = ((ltype & TDB_MARK_LOCK) == TDB_MARK_LOCK);
    665 
    666 	ltype &= ~TDB_MARK_LOCK;
    667 
    668 	/* There are no locks on read-only dbs */
    669 	if (tdb->read_only || tdb->traverse_read) {
    670 		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
    671 	}
    672 
    673 	if (tdb->global_lock.ltype != ltype || tdb->global_lock.count == 0) {
    674 		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
    675 	}
    676 
    677 	if (tdb->global_lock.count > 1) {
    678 		tdb->global_lock.count--;
    679 		return 0;
    680 	}
    681 
    682 	if (!mark_lock &&
    683 	    tdb->methods->tdb_brlock(tdb, FREELIST_TOP, F_UNLCK, F_SETLKW,
    684 				     0, 4*tdb->header.hash_size)) {
    685 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlockall failed (%s)\n", strerror(errno)));
    686 		return -1;
    687 	}
    688 
    689 	tdb->global_lock.count = 0;
    690 	tdb->global_lock.ltype = 0;
    691 
    692 	return 0;
    693 }
    694 
    695 /* lock entire database with write lock */
    696 int tdb_lockall(struct tdb_context *tdb)
    697 {
    698 	return _tdb_lockall(tdb, F_WRLCK, F_SETLKW);
    699 }
    700 
    701 /* lock entire database with write lock - mark only */
    702 int tdb_lockall_mark(struct tdb_context *tdb)
    703 {
    704 	return _tdb_lockall(tdb, F_WRLCK | TDB_MARK_LOCK, F_SETLKW);
    705 }
    706 
    707 /* unlock entire database with write lock - unmark only */
    708 int tdb_lockall_unmark(struct tdb_context *tdb)
    709 {
    710 	return _tdb_unlockall(tdb, F_WRLCK | TDB_MARK_LOCK);
    711 }
    712 
    713 /* lock entire database with write lock - nonblocking varient */
    714 int tdb_lockall_nonblock(struct tdb_context *tdb)
    715 {
    716 	return _tdb_lockall(tdb, F_WRLCK, F_SETLK);
    717 }
    718 
    719 /* unlock entire database with write lock */
    720 int tdb_unlockall(struct tdb_context *tdb)
    721 {
    722 	return _tdb_unlockall(tdb, F_WRLCK);
    723 }
    724 
    725 /* lock entire database with read lock */
    726 int tdb_lockall_read(struct tdb_context *tdb)
    727 {
    728 	return _tdb_lockall(tdb, F_RDLCK, F_SETLKW);
    729 }
    730 
    731 /* lock entire database with read lock - nonblock varient */
    732 int tdb_lockall_read_nonblock(struct tdb_context *tdb)
    733 {
    734 	return _tdb_lockall(tdb, F_RDLCK, F_SETLK);
    735 }
    736 
    737 /* unlock entire database with read lock */
    738 int tdb_unlockall_read(struct tdb_context *tdb)
    739 {
    740 	return _tdb_unlockall(tdb, F_RDLCK);
    741 }
    742 
    743 /* lock/unlock one hash chain. This is meant to be used to reduce
    744    contention - it cannot guarantee how many records will be locked */
    745 int tdb_chainlock(struct tdb_context *tdb, TDB_DATA key)
    746 {
    747 	return tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
    748 }
    749 
    750 /* lock/unlock one hash chain, non-blocking. This is meant to be used
    751    to reduce contention - it cannot guarantee how many records will be
    752    locked */
    753 int tdb_chainlock_nonblock(struct tdb_context *tdb, TDB_DATA key)
    754 {
    755 	return tdb_lock_nonblock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
    756 }
    757 
    758 /* mark a chain as locked without actually locking it. Warning! use with great caution! */
    759 int tdb_chainlock_mark(struct tdb_context *tdb, TDB_DATA key)
    760 {
    761 	return tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK | TDB_MARK_LOCK);
    762 }
    763 
    764 /* unmark a chain as locked without actually locking it. Warning! use with great caution! */
    765 int tdb_chainlock_unmark(struct tdb_context *tdb, TDB_DATA key)
    766 {
    767 	return tdb_unlock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK | TDB_MARK_LOCK);
    768 }
    769 
    770 int tdb_chainunlock(struct tdb_context *tdb, TDB_DATA key)
    771 {
    772 	return tdb_unlock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
    773 }
    774 
    775 int tdb_chainlock_read(struct tdb_context *tdb, TDB_DATA key)
    776 {
    777 	return tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_RDLCK);
    778 }
    779 
    780 int tdb_chainunlock_read(struct tdb_context *tdb, TDB_DATA key)
    781 {
    782 	return tdb_unlock(tdb, BUCKET(tdb->hash_fn(&key)), F_RDLCK);
    783 }
    784 
    785 
    786 
    787 /* record lock stops delete underneath */
    788 int tdb_lock_record(struct tdb_context *tdb, tdb_off_t off)
    789 {
    790 	return off ? tdb->methods->tdb_brlock(tdb, off, F_RDLCK, F_SETLKW, 0, 1) : 0;
    791 }
    792 
    793 /*
    794   Write locks override our own fcntl readlocks, so check it here.
    795   Note this is meant to be F_SETLK, *not* F_SETLKW, as it's not
    796   an error to fail to get the lock here.
    797 */
    798 int tdb_write_lock_record(struct tdb_context *tdb, tdb_off_t off)
    799 {
    800 	struct tdb_traverse_lock *i;
    801 	for (i = &tdb->travlocks; i; i = i->next)
    802 		if (i->off == off)
    803 			return -1;
    804 	return tdb->methods->tdb_brlock(tdb, off, F_WRLCK, F_SETLK, 1, 1);
    805 }
    806 
    807 /*
    808   Note this is meant to be F_SETLK, *not* F_SETLKW, as it's not
    809   an error to fail to get the lock here.
    810 */
    811 int tdb_write_unlock_record(struct tdb_context *tdb, tdb_off_t off)
    812 {
    813 	return tdb->methods->tdb_brlock(tdb, off, F_UNLCK, F_SETLK, 0, 1);
    814 }
    815 
    816 /* fcntl locks don't stack: avoid unlocking someone else's */
    817 int tdb_unlock_record(struct tdb_context *tdb, tdb_off_t off)
    818 {
    819 	struct tdb_traverse_lock *i;
    820 	u32 count = 0;
    821 
    822 	if (off == 0)
    823 		return 0;
    824 	for (i = &tdb->travlocks; i; i = i->next)
    825 		if (i->off == off)
    826 			count++;
    827 	return (count == 1 ? tdb->methods->tdb_brlock(tdb, off, F_UNLCK, F_SETLKW, 0, 1) : 0);
    828 }
    829 
    830 /* file: io.c */
    831 
    832 /* check for an out of bounds access - if it is out of bounds then
    833    see if the database has been expanded by someone else and expand
    834    if necessary
    835    note that "len" is the minimum length needed for the db
    836 */
    837 static int tdb_oob(struct tdb_context *tdb, tdb_off_t len, int probe)
    838 {
    839 	struct stat st;
    840 	if (len <= tdb->map_size)
    841 		return 0;
    842 	if (tdb->flags & TDB_INTERNAL) {
    843 		if (!probe) {
    844 			/* Ensure ecode is set for log fn. */
    845 			tdb->ecode = TDB_ERR_IO;
    846 			TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_oob len %d beyond internal malloc size %d\n",
    847 				 (int)len, (int)tdb->map_size));
    848 		}
    849 		return TDB_ERRCODE(TDB_ERR_IO, -1);
    850 	}
    851 
    852 	if (fstat(tdb->fd, &st) == -1) {
    853 		return TDB_ERRCODE(TDB_ERR_IO, -1);
    854 	}
    855 
    856 	if (st.st_size < (size_t)len) {
    857 		if (!probe) {
    858 			/* Ensure ecode is set for log fn. */
    859 			tdb->ecode = TDB_ERR_IO;
    860 			TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_oob len %d beyond eof at %d\n",
    861 				 (int)len, (int)st.st_size));
    862 		}
    863 		return TDB_ERRCODE(TDB_ERR_IO, -1);
    864 	}
    865 
    866 	/* Unmap, update size, remap */
    867 	if (tdb_munmap(tdb) == -1)
    868 		return TDB_ERRCODE(TDB_ERR_IO, -1);
    869 	tdb->map_size = st.st_size;
    870 	tdb_mmap(tdb);
    871 	return 0;
    872 }
    873 
    874 /* write a lump of data at a specified offset */
    875 static int tdb_write(struct tdb_context *tdb, tdb_off_t off,
    876 		     const void *buf, tdb_len_t len)
    877 {
    878 	if (len == 0) {
    879 		return 0;
    880 	}
    881 
    882 	if (tdb->read_only || tdb->traverse_read) {
    883 		tdb->ecode = TDB_ERR_RDONLY;
    884 		return -1;
    885 	}
    886 
    887 	if (tdb->methods->tdb_oob(tdb, off + len, 0) != 0)
    888 		return -1;
    889 
    890 	if (tdb->map_ptr) {
    891 		memcpy(off + (char *)tdb->map_ptr, buf, len);
    892 	} else if (pwrite(tdb->fd, buf, len, off) != (ssize_t)len) {
    893 		/* Ensure ecode is set for log fn. */
    894 		tdb->ecode = TDB_ERR_IO;
    895 		TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_write failed at %d len=%d (%s)\n",
    896 			   off, len, strerror(errno)));
    897 		return TDB_ERRCODE(TDB_ERR_IO, -1);
    898 	}
    899 	return 0;
    900 }
    901 
    902 /* Endian conversion: we only ever deal with 4 byte quantities */
    903 void *tdb_convert(void *buf, u32 size)
    904 {
    905 	u32 i, *p = (u32 *)buf;
    906 	for (i = 0; i < size / 4; i++)
    907 		p[i] = TDB_BYTEREV(p[i]);
    908 	return buf;
    909 }
    910 
    911 
    912 /* read a lump of data at a specified offset, maybe convert */
    913 static int tdb_read(struct tdb_context *tdb, tdb_off_t off, void *buf,
    914 		    tdb_len_t len, int cv)
    915 {
    916 	if (tdb->methods->tdb_oob(tdb, off + len, 0) != 0) {
    917 		return -1;
    918 	}
    919 
    920 	if (tdb->map_ptr) {
    921 		memcpy(buf, off + (char *)tdb->map_ptr, len);
    922 	} else {
    923 		ssize_t ret = pread(tdb->fd, buf, len, off);
    924 		if (ret != (ssize_t)len) {
    925 			/* Ensure ecode is set for log fn. */
    926 			tdb->ecode = TDB_ERR_IO;
    927 			TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_read failed at %d "
    928 				 "len=%d ret=%d (%s) map_size=%d\n",
    929 				 (int)off, (int)len, (int)ret, strerror(errno),
    930 				 (int)tdb->map_size));
    931 			return TDB_ERRCODE(TDB_ERR_IO, -1);
    932 		}
    933 	}
    934 	if (cv) {
    935 		tdb_convert(buf, len);
    936 	}
    937 	return 0;
    938 }
    939 
    940 
    941 
    942 /*
    943   do an unlocked scan of the hash table heads to find the next non-zero head. The value
    944   will then be confirmed with the lock held
    945 */
    946 static void tdb_next_hash_chain(struct tdb_context *tdb, u32 *chain)
    947 {
    948 	u32 h = *chain;
    949 	if (tdb->map_ptr) {
    950 		for (;h < tdb->header.hash_size;h++) {
    951 			if (0 != *(u32 *)(TDB_HASH_TOP(h) + (unsigned char *)tdb->map_ptr)) {
    952 				break;
    953 			}
    954 		}
    955 	} else {
    956 		u32 off=0;
    957 		for (;h < tdb->header.hash_size;h++) {
    958 			if (tdb_ofs_read(tdb, TDB_HASH_TOP(h), &off) != 0 || off != 0) {
    959 				break;
    960 			}
    961 		}
    962 	}
    963 	(*chain) = h;
    964 }
    965 
    966 
    967 int tdb_munmap(struct tdb_context *tdb)
    968 {
    969 	if (tdb->flags & TDB_INTERNAL)
    970 		return 0;
    971 
    972 #ifdef HAVE_MMAP
    973 	if (tdb->map_ptr) {
    974 		int ret = munmap(tdb->map_ptr, tdb->map_size);
    975 		if (ret != 0)
    976 			return ret;
    977 	}
    978 #endif
    979 	tdb->map_ptr = NULL;
    980 	return 0;
    981 }
    982 
    983 void tdb_mmap(struct tdb_context *tdb)
    984 {
    985 	if (tdb->flags & TDB_INTERNAL)
    986 		return;
    987 
    988 #ifdef HAVE_MMAP
    989 	if (!(tdb->flags & TDB_NOMMAP)) {
    990 		tdb->map_ptr = mmap(NULL, tdb->map_size,
    991 				    PROT_READ|(tdb->read_only? 0:PROT_WRITE),
    992 				    MAP_SHARED|MAP_FILE, tdb->fd, 0);
    993 
    994 		/*
    995 		 * NB. When mmap fails it returns MAP_FAILED *NOT* NULL !!!!
    996 		 */
    997 
    998 		if (tdb->map_ptr == MAP_FAILED) {
    999 			tdb->map_ptr = NULL;
   1000 			TDB_LOG((tdb, TDB_DEBUG_WARNING, "tdb_mmap failed for size %d (%s)\n",
   1001 				 tdb->map_size, strerror(errno)));
   1002 		}
   1003 	} else {
   1004 		tdb->map_ptr = NULL;
   1005 	}
   1006 #else
   1007 	tdb->map_ptr = NULL;
   1008 #endif
   1009 }
   1010 
   1011 /* expand a file.  we prefer to use ftruncate, as that is what posix
   1012   says to use for mmap expansion */
   1013 static int tdb_expand_file(struct tdb_context *tdb, tdb_off_t size, tdb_off_t addition)
   1014 {
   1015 	char buf[1024];
   1016 
   1017 	if (tdb->read_only || tdb->traverse_read) {
   1018 		tdb->ecode = TDB_ERR_RDONLY;
   1019 		return -1;
   1020 	}
   1021 
   1022 	if (ftruncate(tdb->fd, size+addition) == -1) {
   1023 		char b = 0;
   1024 		if (pwrite(tdb->fd,  &b, 1, (size+addition) - 1) != 1) {
   1025 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "expand_file to %d failed (%s)\n",
   1026 				 size+addition, strerror(errno)));
   1027 			return -1;
   1028 		}
   1029 	}
   1030 
   1031 	/* now fill the file with something. This ensures that the
   1032 	   file isn't sparse, which would be very bad if we ran out of
   1033 	   disk. This must be done with write, not via mmap */
   1034 	memset(buf, TDB_PAD_BYTE, sizeof(buf));
   1035 	while (addition) {
   1036 		int n = addition>sizeof(buf)?sizeof(buf):addition;
   1037 		int ret = pwrite(tdb->fd, buf, n, size);
   1038 		if (ret != n) {
   1039 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "expand_file write of %d failed (%s)\n",
   1040 				   n, strerror(errno)));
   1041 			return -1;
   1042 		}
   1043 		addition -= n;
   1044 		size += n;
   1045 	}
   1046 	return 0;
   1047 }
   1048 
   1049 
   1050 /* expand the database at least size bytes by expanding the underlying
   1051    file and doing the mmap again if necessary */
   1052 int tdb_expand(struct tdb_context *tdb, tdb_off_t size)
   1053 {
   1054 	struct list_struct rec;
   1055 	tdb_off_t offset;
   1056 
   1057 	if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
   1058 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "lock failed in tdb_expand\n"));
   1059 		return -1;
   1060 	}
   1061 
   1062 	/* must know about any previous expansions by another process */
   1063 	tdb->methods->tdb_oob(tdb, tdb->map_size + 1, 1);
   1064 
   1065 	/* always make room for at least 10 more records, and round
   1066            the database up to a multiple of the page size */
   1067 	size = TDB_ALIGN(tdb->map_size + size*10, tdb->page_size) - tdb->map_size;
   1068 
   1069 	if (!(tdb->flags & TDB_INTERNAL))
   1070 		tdb_munmap(tdb);
   1071 
   1072 	/*
   1073 	 * We must ensure the file is unmapped before doing this
   1074 	 * to ensure consistency with systems like OpenBSD where
   1075 	 * writes and mmaps are not consistent.
   1076 	 */
   1077 
   1078 	/* expand the file itself */
   1079 	if (!(tdb->flags & TDB_INTERNAL)) {
   1080 		if (tdb->methods->tdb_expand_file(tdb, tdb->map_size, size) != 0)
   1081 			goto fail;
   1082 	}
   1083 
   1084 	tdb->map_size += size;
   1085 
   1086 	if (tdb->flags & TDB_INTERNAL) {
   1087 		char *new_map_ptr = (char *)realloc(tdb->map_ptr,
   1088 						    tdb->map_size);
   1089 		if (!new_map_ptr) {
   1090 			tdb->map_size -= size;
   1091 			goto fail;
   1092 		}
   1093 		tdb->map_ptr = new_map_ptr;
   1094 	} else {
   1095 		/*
   1096 		 * We must ensure the file is remapped before adding the space
   1097 		 * to ensure consistency with systems like OpenBSD where
   1098 		 * writes and mmaps are not consistent.
   1099 		 */
   1100 
   1101 		/* We're ok if the mmap fails as we'll fallback to read/write */
   1102 		tdb_mmap(tdb);
   1103 	}
   1104 
   1105 	/* form a new freelist record */
   1106 	memset(&rec,'\0',sizeof(rec));
   1107 	rec.rec_len = size - sizeof(rec);
   1108 
   1109 	/* link it into the free list */
   1110 	offset = tdb->map_size - size;
   1111 	if (tdb_free(tdb, offset, &rec) == -1)
   1112 		goto fail;
   1113 
   1114 	tdb_unlock(tdb, -1, F_WRLCK);
   1115 	return 0;
   1116  fail:
   1117 	tdb_unlock(tdb, -1, F_WRLCK);
   1118 	return -1;
   1119 }
   1120 
   1121 /* read/write a tdb_off_t */
   1122 int tdb_ofs_read(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d)
   1123 {
   1124 	return tdb->methods->tdb_read(tdb, offset, (char*)d, sizeof(*d), DOCONV());
   1125 }
   1126 
   1127 int tdb_ofs_write(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d)
   1128 {
   1129 	tdb_off_t off = *d;
   1130 	return tdb->methods->tdb_write(tdb, offset, CONVERT(off), sizeof(*d));
   1131 }
   1132 
   1133 
   1134 /* read a lump of data, allocating the space for it */
   1135 unsigned char *tdb_alloc_read(struct tdb_context *tdb, tdb_off_t offset, tdb_len_t len)
   1136 {
   1137 	unsigned char *buf;
   1138 
   1139 	/* some systems don't like zero length malloc */
   1140 	if (len == 0) {
   1141 		len = 1;
   1142 	}
   1143 
   1144 	if (!(buf = (unsigned char *)malloc(len))) {
   1145 		/* Ensure ecode is set for log fn. */
   1146 		tdb->ecode = TDB_ERR_OOM;
   1147 		TDB_LOG((tdb, TDB_DEBUG_ERROR,"tdb_alloc_read malloc failed len=%d (%s)\n",
   1148 			   len, strerror(errno)));
   1149 		return TDB_ERRCODE(TDB_ERR_OOM, buf);
   1150 	}
   1151 	if (tdb->methods->tdb_read(tdb, offset, buf, len, 0) == -1) {
   1152 		SAFE_FREE(buf);
   1153 		return NULL;
   1154 	}
   1155 	return buf;
   1156 }
   1157 
   1158 /* Give a piece of tdb data to a parser */
   1159 
   1160 int tdb_parse_data(struct tdb_context *tdb, TDB_DATA key,
   1161 		   tdb_off_t offset, tdb_len_t len,
   1162 		   int (*parser)(TDB_DATA key, TDB_DATA data,
   1163 				 void *private_data),
   1164 		   void *private_data)
   1165 {
   1166 	TDB_DATA data;
   1167 	int result;
   1168 
   1169 	data.dsize = len;
   1170 
   1171 	if ((tdb->transaction == NULL) && (tdb->map_ptr != NULL)) {
   1172 		/*
   1173 		 * Optimize by avoiding the malloc/memcpy/free, point the
   1174 		 * parser directly at the mmap area.
   1175 		 */
   1176 		if (tdb->methods->tdb_oob(tdb, offset+len, 0) != 0) {
   1177 			return -1;
   1178 		}
   1179 		data.dptr = offset + (unsigned char *)tdb->map_ptr;
   1180 		return parser(key, data, private_data);
   1181 	}
   1182 
   1183 	if (!(data.dptr = tdb_alloc_read(tdb, offset, len))) {
   1184 		return -1;
   1185 	}
   1186 
   1187 	result = parser(key, data, private_data);
   1188 	free(data.dptr);
   1189 	return result;
   1190 }
   1191 
   1192 /* read/write a record */
   1193 int tdb_rec_read(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec)
   1194 {
   1195 	if (tdb->methods->tdb_read(tdb, offset, rec, sizeof(*rec),DOCONV()) == -1)
   1196 		return -1;
   1197 	if (TDB_BAD_MAGIC(rec)) {
   1198 		/* Ensure ecode is set for log fn. */
   1199 		tdb->ecode = TDB_ERR_CORRUPT;
   1200 		TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_rec_read bad magic 0x%x at offset=%d\n", rec->magic, offset));
   1201 		return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
   1202 	}
   1203 	return tdb->methods->tdb_oob(tdb, rec->next+sizeof(*rec), 0);
   1204 }
   1205 
   1206 int tdb_rec_write(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec)
   1207 {
   1208 	struct list_struct r = *rec;
   1209 	return tdb->methods->tdb_write(tdb, offset, CONVERT(r), sizeof(r));
   1210 }
   1211 
   1212 static const struct tdb_methods io_methods = {
   1213 	tdb_read,
   1214 	tdb_write,
   1215 	tdb_next_hash_chain,
   1216 	tdb_oob,
   1217 	tdb_expand_file,
   1218 	tdb_brlock
   1219 };
   1220 
   1221 /*
   1222   initialise the default methods table
   1223 */
   1224 void tdb_io_init(struct tdb_context *tdb)
   1225 {
   1226 	tdb->methods = &io_methods;
   1227 }
   1228 
   1229 /* file: transaction.c */
   1230 
   1231 /*
   1232   transaction design:
   1233 
   1234   - only allow a single transaction at a time per database. This makes
   1235     using the transaction API simpler, as otherwise the caller would
   1236     have to cope with temporary failures in transactions that conflict
   1237     with other current transactions
   1238 
   1239   - keep the transaction recovery information in the same file as the
   1240     database, using a special 'transaction recovery' record pointed at
   1241     by the header. This removes the need for extra journal files as
   1242     used by some other databases
   1243 
   1244   - dynamically allocated the transaction recover record, re-using it
   1245     for subsequent transactions. If a larger record is needed then
   1246     tdb_free() the old record to place it on the normal tdb freelist
   1247     before allocating the new record
   1248 
   1249   - during transactions, keep a linked list of writes all that have
   1250     been performed by intercepting all tdb_write() calls. The hooked
   1251     transaction versions of tdb_read() and tdb_write() check this
   1252     linked list and try to use the elements of the list in preference
   1253     to the real database.
   1254 
   1255   - don't allow any locks to be held when a transaction starts,
   1256     otherwise we can end up with deadlock (plus lack of lock nesting
   1257     in posix locks would mean the lock is lost)
   1258 
   1259   - if the caller gains a lock during the transaction but doesn't
   1260     release it then fail the commit
   1261 
   1262   - allow for nested calls to tdb_transaction_start(), re-using the
   1263     existing transaction record. If the inner transaction is cancelled
   1264     then a subsequent commit will fail
   1265 
   1266   - keep a mirrored copy of the tdb hash chain heads to allow for the
   1267     fast hash heads scan on traverse, updating the mirrored copy in
   1268     the transaction version of tdb_write
   1269 
   1270   - allow callers to mix transaction and non-transaction use of tdb,
   1271     although once a transaction is started then an exclusive lock is
   1272     gained until the transaction is committed or cancelled
   1273 
   1274   - the commit stategy involves first saving away all modified data
   1275     into a linearised buffer in the transaction recovery area, then
   1276     marking the transaction recovery area with a magic value to
   1277     indicate a valid recovery record. In total 4 fsync/msync calls are
   1278     needed per commit to prevent race conditions. It might be possible
   1279     to reduce this to 3 or even 2 with some more work.
   1280 
   1281   - check for a valid recovery record on open of the tdb, while the
   1282     global lock is held. Automatically recover from the transaction
   1283     recovery area if needed, then continue with the open as
   1284     usual. This allows for smooth crash recovery with no administrator
   1285     intervention.
   1286 
   1287   - if TDB_NOSYNC is passed to flags in tdb_open then transactions are
   1288     still available, but no transaction recovery area is used and no
   1289     fsync/msync calls are made.
   1290 
   1291 */
   1292 
   1293 struct tdb_transaction_el {
   1294 	struct tdb_transaction_el *next, *prev;
   1295 	tdb_off_t offset;
   1296 	tdb_len_t length;
   1297 	unsigned char *data;
   1298 };
   1299 
   1300 /*
   1301   hold the context of any current transaction
   1302 */
   1303 struct tdb_transaction {
   1304 	/* we keep a mirrored copy of the tdb hash heads here so
   1305 	   tdb_next_hash_chain() can operate efficiently */
   1306 	u32 *hash_heads;
   1307 
   1308 	/* the original io methods - used to do IOs to the real db */
   1309 	const struct tdb_methods *io_methods;
   1310 
   1311 	/* the list of transaction elements. We use a doubly linked
   1312 	   list with a last pointer to allow us to keep the list
   1313 	   ordered, with first element at the front of the list. It
   1314 	   needs to be doubly linked as the read/write traversals need
   1315 	   to be backwards, while the commit needs to be forwards */
   1316 	struct tdb_transaction_el *elements, *elements_last;
   1317 
   1318 	/* non-zero when an internal transaction error has
   1319 	   occurred. All write operations will then fail until the
   1320 	   transaction is ended */
   1321 	int transaction_error;
   1322 
   1323 	/* when inside a transaction we need to keep track of any
   1324 	   nested tdb_transaction_start() calls, as these are allowed,
   1325 	   but don't create a new transaction */
   1326 	int nesting;
   1327 
   1328 	/* old file size before transaction */
   1329 	tdb_len_t old_map_size;
   1330 };
   1331 
   1332 
   1333 /*
   1334   read while in a transaction. We need to check first if the data is in our list
   1335   of transaction elements, then if not do a real read
   1336 */
   1337 static int transaction_read(struct tdb_context *tdb, tdb_off_t off, void *buf,
   1338 			    tdb_len_t len, int cv)
   1339 {
   1340 	struct tdb_transaction_el *el;
   1341 
   1342 	/* we need to walk the list backwards to get the most recent data */
   1343 	for (el=tdb->transaction->elements_last;el;el=el->prev) {
   1344 		tdb_len_t partial;
   1345 
   1346 		if (off+len <= el->offset) {
   1347 			continue;
   1348 		}
   1349 		if (off >= el->offset + el->length) {
   1350 			continue;
   1351 		}
   1352 
   1353 		/* an overlapping read - needs to be split into up to
   1354 		   2 reads and a memcpy */
   1355 		if (off < el->offset) {
   1356 			partial = el->offset - off;
   1357 			if (transaction_read(tdb, off, buf, partial, cv) != 0) {
   1358 				goto fail;
   1359 			}
   1360 			len -= partial;
   1361 			off += partial;
   1362 			buf = (void *)(partial + (char *)buf);
   1363 		}
   1364 		if (off + len <= el->offset + el->length) {
   1365 			partial = len;
   1366 		} else {
   1367 			partial = el->offset + el->length - off;
   1368 		}
   1369 		memcpy(buf, el->data + (off - el->offset), partial);
   1370 		if (cv) {
   1371 			tdb_convert(buf, len);
   1372 		}
   1373 		len -= partial;
   1374 		off += partial;
   1375 		buf = (void *)(partial + (char *)buf);
   1376 
   1377 		if (len != 0 && transaction_read(tdb, off, buf, len, cv) != 0) {
   1378 			goto fail;
   1379 		}
   1380 
   1381 		return 0;
   1382 	}
   1383 
   1384 	/* its not in the transaction elements - do a real read */
   1385 	return tdb->transaction->io_methods->tdb_read(tdb, off, buf, len, cv);
   1386 
   1387 fail:
   1388 	TDB_LOG((tdb, TDB_DEBUG_FATAL, "transaction_read: failed at off=%d len=%d\n", off, len));
   1389 	tdb->ecode = TDB_ERR_IO;
   1390 	tdb->transaction->transaction_error = 1;
   1391 	return -1;
   1392 }
   1393 
   1394 
   1395 /*
   1396   write while in a transaction
   1397 */
   1398 static int transaction_write(struct tdb_context *tdb, tdb_off_t off,
   1399 			     const void *buf, tdb_len_t len)
   1400 {
   1401 	struct tdb_transaction_el *el, *best_el=NULL;
   1402 
   1403 	if (len == 0) {
   1404 		return 0;
   1405 	}
   1406 
   1407 	/* if the write is to a hash head, then update the transaction
   1408 	   hash heads */
   1409 	if (len == sizeof(tdb_off_t) && off >= FREELIST_TOP &&
   1410 	    off < FREELIST_TOP+TDB_HASHTABLE_SIZE(tdb)) {
   1411 		u32 chain = (off-FREELIST_TOP) / sizeof(tdb_off_t);
   1412 		memcpy(&tdb->transaction->hash_heads[chain], buf, len);
   1413 	}
   1414 
   1415 	/* first see if we can replace an existing entry */
   1416 	for (el=tdb->transaction->elements_last;el;el=el->prev) {
   1417 		tdb_len_t partial;
   1418 
   1419 		if (best_el == NULL && off == el->offset+el->length) {
   1420 			best_el = el;
   1421 		}
   1422 
   1423 		if (off+len <= el->offset) {
   1424 			continue;
   1425 		}
   1426 		if (off >= el->offset + el->length) {
   1427 			continue;
   1428 		}
   1429 
   1430 		/* an overlapping write - needs to be split into up to
   1431 		   2 writes and a memcpy */
   1432 		if (off < el->offset) {
   1433 			partial = el->offset - off;
   1434 			if (transaction_write(tdb, off, buf, partial) != 0) {
   1435 				goto fail;
   1436 			}
   1437 			len -= partial;
   1438 			off += partial;
   1439 			buf = (const void *)(partial + (const char *)buf);
   1440 		}
   1441 		if (off + len <= el->offset + el->length) {
   1442 			partial = len;
   1443 		} else {
   1444 			partial = el->offset + el->length - off;
   1445 		}
   1446 		memcpy(el->data + (off - el->offset), buf, partial);
   1447 		len -= partial;
   1448 		off += partial;
   1449 		buf = (const void *)(partial + (const char *)buf);
   1450 
   1451 		if (len != 0 && transaction_write(tdb, off, buf, len) != 0) {
   1452 			goto fail;
   1453 		}
   1454 
   1455 		return 0;
   1456 	}
   1457 
   1458 	/* see if we can append the new entry to an existing entry */
   1459 	if (best_el && best_el->offset + best_el->length == off &&
   1460 	    (off+len < tdb->transaction->old_map_size ||
   1461 	     off > tdb->transaction->old_map_size)) {
   1462 		unsigned char *data = best_el->data;
   1463 		el = best_el;
   1464 		el->data = (unsigned char *)realloc(el->data,
   1465 						    el->length + len);
   1466 		if (el->data == NULL) {
   1467 			tdb->ecode = TDB_ERR_OOM;
   1468 			tdb->transaction->transaction_error = 1;
   1469 			el->data = data;
   1470 			return -1;
   1471 		}
   1472 		if (buf) {
   1473 			memcpy(el->data + el->length, buf, len);
   1474 		} else {
   1475 			memset(el->data + el->length, TDB_PAD_BYTE, len);
   1476 		}
   1477 		el->length += len;
   1478 		return 0;
   1479 	}
   1480 
   1481 	/* add a new entry at the end of the list */
   1482 	el = (struct tdb_transaction_el *)malloc(sizeof(*el));
   1483 	if (el == NULL) {
   1484 		tdb->ecode = TDB_ERR_OOM;
   1485 		tdb->transaction->transaction_error = 1;
   1486 		return -1;
   1487 	}
   1488 	el->next = NULL;
   1489 	el->prev = tdb->transaction->elements_last;
   1490 	el->offset = off;
   1491 	el->length = len;
   1492 	el->data = (unsigned char *)malloc(len);
   1493 	if (el->data == NULL) {
   1494 		free(el);
   1495 		tdb->ecode = TDB_ERR_OOM;
   1496 		tdb->transaction->transaction_error = 1;
   1497 		return -1;
   1498 	}
   1499 	if (buf) {
   1500 		memcpy(el->data, buf, len);
   1501 	} else {
   1502 		memset(el->data, TDB_PAD_BYTE, len);
   1503 	}
   1504 	if (el->prev) {
   1505 		el->prev->next = el;
   1506 	} else {
   1507 		tdb->transaction->elements = el;
   1508 	}
   1509 	tdb->transaction->elements_last = el;
   1510 	return 0;
   1511 
   1512 fail:
   1513 	TDB_LOG((tdb, TDB_DEBUG_FATAL, "transaction_write: failed at off=%d len=%d\n", off, len));
   1514 	tdb->ecode = TDB_ERR_IO;
   1515 	tdb->transaction->transaction_error = 1;
   1516 	return -1;
   1517 }
   1518 
   1519 /*
   1520   accelerated hash chain head search, using the cached hash heads
   1521 */
   1522 static void transaction_next_hash_chain(struct tdb_context *tdb, u32 *chain)
   1523 {
   1524 	u32 h = *chain;
   1525 	for (;h < tdb->header.hash_size;h++) {
   1526 		/* the +1 takes account of the freelist */
   1527 		if (0 != tdb->transaction->hash_heads[h+1]) {
   1528 			break;
   1529 		}
   1530 	}
   1531 	(*chain) = h;
   1532 }
   1533 
   1534 /*
   1535   out of bounds check during a transaction
   1536 */
   1537 static int transaction_oob(struct tdb_context *tdb, tdb_off_t len, int probe)
   1538 {
   1539 	if (len <= tdb->map_size) {
   1540 		return 0;
   1541 	}
   1542 	return TDB_ERRCODE(TDB_ERR_IO, -1);
   1543 }
   1544 
   1545 /*
   1546   transaction version of tdb_expand().
   1547 */
   1548 static int transaction_expand_file(struct tdb_context *tdb, tdb_off_t size,
   1549 				   tdb_off_t addition)
   1550 {
   1551 	/* add a write to the transaction elements, so subsequent
   1552 	   reads see the zero data */
   1553 	if (transaction_write(tdb, size, NULL, addition) != 0) {
   1554 		return -1;
   1555 	}
   1556 
   1557 	return 0;
   1558 }
   1559 
   1560 /*
   1561   brlock during a transaction - ignore them
   1562 */
   1563 static int transaction_brlock(struct tdb_context *tdb, tdb_off_t offset,
   1564 			      int rw_type, int lck_type, int probe, size_t len)
   1565 {
   1566 	return 0;
   1567 }
   1568 
   1569 static const struct tdb_methods transaction_methods = {
   1570 	transaction_read,
   1571 	transaction_write,
   1572 	transaction_next_hash_chain,
   1573 	transaction_oob,
   1574 	transaction_expand_file,
   1575 	transaction_brlock
   1576 };
   1577 
   1578 
   1579 /*
   1580   start a tdb transaction. No token is returned, as only a single
   1581   transaction is allowed to be pending per tdb_context
   1582 */
   1583 int tdb_transaction_start(struct tdb_context *tdb)
   1584 {
   1585 	/* some sanity checks */
   1586 	if (tdb->read_only || (tdb->flags & TDB_INTERNAL) || tdb->traverse_read) {
   1587 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: cannot start a transaction on a read-only or internal db\n"));
   1588 		tdb->ecode = TDB_ERR_EINVAL;
   1589 		return -1;
   1590 	}
   1591 
   1592 	/* cope with nested tdb_transaction_start() calls */
   1593 	if (tdb->transaction != NULL) {
   1594 		tdb->transaction->nesting++;
   1595 		TDB_LOG((tdb, TDB_DEBUG_TRACE, "tdb_transaction_start: nesting %d\n",
   1596 			 tdb->transaction->nesting));
   1597 		return 0;
   1598 	}
   1599 
   1600 	if (tdb->num_locks != 0 || tdb->global_lock.count) {
   1601 		/* the caller must not have any locks when starting a
   1602 		   transaction as otherwise we'll be screwed by lack
   1603 		   of nested locks in posix */
   1604 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: cannot start a transaction with locks held\n"));
   1605 		tdb->ecode = TDB_ERR_LOCK;
   1606 		return -1;
   1607 	}
   1608 
   1609 	if (tdb->travlocks.next != NULL) {
   1610 		/* you cannot use transactions inside a traverse (although you can use
   1611 		   traverse inside a transaction) as otherwise you can end up with
   1612 		   deadlock */
   1613 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: cannot start a transaction within a traverse\n"));
   1614 		tdb->ecode = TDB_ERR_LOCK;
   1615 		return -1;
   1616 	}
   1617 
   1618 	tdb->transaction = (struct tdb_transaction *)
   1619 		calloc(sizeof(struct tdb_transaction), 1);
   1620 	if (tdb->transaction == NULL) {
   1621 		tdb->ecode = TDB_ERR_OOM;
   1622 		return -1;
   1623 	}
   1624 
   1625 	/* get the transaction write lock. This is a blocking lock. As
   1626 	   discussed with Volker, there are a number of ways we could
   1627 	   make this async, which we will probably do in the future */
   1628 	if (tdb_transaction_lock(tdb, F_WRLCK) == -1) {
   1629 		SAFE_FREE(tdb->transaction);
   1630 		return -1;
   1631 	}
   1632 
   1633 	/* get a read lock from the freelist to the end of file. This
   1634 	   is upgraded to a write lock during the commit */
   1635 	if (tdb_brlock(tdb, FREELIST_TOP, F_RDLCK, F_SETLKW, 0, 0) == -1) {
   1636 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: failed to get hash locks\n"));
   1637 		tdb->ecode = TDB_ERR_LOCK;
   1638 		goto fail;
   1639 	}
   1640 
   1641 	/* setup a copy of the hash table heads so the hash scan in
   1642 	   traverse can be fast */
   1643 	tdb->transaction->hash_heads = (u32 *)
   1644 		calloc(tdb->header.hash_size+1, sizeof(u32));
   1645 	if (tdb->transaction->hash_heads == NULL) {
   1646 		tdb->ecode = TDB_ERR_OOM;
   1647 		goto fail;
   1648 	}
   1649 	if (tdb->methods->tdb_read(tdb, FREELIST_TOP, tdb->transaction->hash_heads,
   1650 				   TDB_HASHTABLE_SIZE(tdb), 0) != 0) {
   1651 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_start: failed to read hash heads\n"));
   1652 		tdb->ecode = TDB_ERR_IO;
   1653 		goto fail;
   1654 	}
   1655 
   1656 	/* make sure we know about any file expansions already done by
   1657 	   anyone else */
   1658 	tdb->methods->tdb_oob(tdb, tdb->map_size + 1, 1);
   1659 	tdb->transaction->old_map_size = tdb->map_size;
   1660 
   1661 	/* finally hook the io methods, replacing them with
   1662 	   transaction specific methods */
   1663 	tdb->transaction->io_methods = tdb->methods;
   1664 	tdb->methods = &transaction_methods;
   1665 
   1666 	/* by calling this transaction write here, we ensure that we don't grow the
   1667 	   transaction linked list due to hash table updates */
   1668 	if (transaction_write(tdb, FREELIST_TOP, tdb->transaction->hash_heads,
   1669 			      TDB_HASHTABLE_SIZE(tdb)) != 0) {
   1670 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_start: failed to prime hash table\n"));
   1671 		tdb->ecode = TDB_ERR_IO;
   1672 		tdb->methods = tdb->transaction->io_methods;
   1673 		goto fail;
   1674 	}
   1675 
   1676 	return 0;
   1677 
   1678 fail:
   1679 	tdb_brlock(tdb, FREELIST_TOP, F_UNLCK, F_SETLKW, 0, 0);
   1680 	tdb_transaction_unlock(tdb);
   1681 	SAFE_FREE(tdb->transaction->hash_heads);
   1682 	SAFE_FREE(tdb->transaction);
   1683 	return -1;
   1684 }
   1685 
   1686 
   1687 /*
   1688   cancel the current transaction
   1689 */
   1690 int tdb_transaction_cancel(struct tdb_context *tdb)
   1691 {
   1692 	if (tdb->transaction == NULL) {
   1693 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_cancel: no transaction\n"));
   1694 		return -1;
   1695 	}
   1696 
   1697 	if (tdb->transaction->nesting != 0) {
   1698 		tdb->transaction->transaction_error = 1;
   1699 		tdb->transaction->nesting--;
   1700 		return 0;
   1701 	}
   1702 
   1703 	tdb->map_size = tdb->transaction->old_map_size;
   1704 
   1705 	/* free all the transaction elements */
   1706 	while (tdb->transaction->elements) {
   1707 		struct tdb_transaction_el *el = tdb->transaction->elements;
   1708 		tdb->transaction->elements = el->next;
   1709 		free(el->data);
   1710 		free(el);
   1711 	}
   1712 
   1713 	/* remove any global lock created during the transaction */
   1714 	if (tdb->global_lock.count != 0) {
   1715 		tdb_brlock(tdb, FREELIST_TOP, F_UNLCK, F_SETLKW, 0, 4*tdb->header.hash_size);
   1716 		tdb->global_lock.count = 0;
   1717 	}
   1718 
   1719 	/* remove any locks created during the transaction */
   1720 	if (tdb->num_locks != 0) {
   1721 		int i;
   1722 		for (i=0;i<tdb->num_lockrecs;i++) {
   1723 			tdb_brlock(tdb,FREELIST_TOP+4*tdb->lockrecs[i].list,
   1724 				   F_UNLCK,F_SETLKW, 0, 1);
   1725 		}
   1726 		tdb->num_locks = 0;
   1727 		tdb->num_lockrecs = 0;
   1728 		SAFE_FREE(tdb->lockrecs);
   1729 	}
   1730 
   1731 	/* restore the normal io methods */
   1732 	tdb->methods = tdb->transaction->io_methods;
   1733 
   1734 	tdb_brlock(tdb, FREELIST_TOP, F_UNLCK, F_SETLKW, 0, 0);
   1735 	tdb_transaction_unlock(tdb);
   1736 	SAFE_FREE(tdb->transaction->hash_heads);
   1737 	SAFE_FREE(tdb->transaction);
   1738 
   1739 	return 0;
   1740 }
   1741 
   1742 /*
   1743   sync to disk
   1744 */
   1745 static int transaction_sync(struct tdb_context *tdb, tdb_off_t offset, tdb_len_t length)
   1746 {
   1747 	if (fsync(tdb->fd) != 0) {
   1748 		tdb->ecode = TDB_ERR_IO;
   1749 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction: fsync failed\n"));
   1750 		return -1;
   1751 	}
   1752 #ifdef MS_SYNC
   1753 	if (tdb->map_ptr) {
   1754 		tdb_off_t moffset = offset & ~(tdb->page_size-1);
   1755 		if (msync(moffset + (char *)tdb->map_ptr,
   1756 			  length + (offset - moffset), MS_SYNC) != 0) {
   1757 			tdb->ecode = TDB_ERR_IO;
   1758 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction: msync failed - %s\n",
   1759 				 strerror(errno)));
   1760 			return -1;
   1761 		}
   1762 	}
   1763 #endif
   1764 	return 0;
   1765 }
   1766 
   1767 
   1768 /*
   1769   work out how much space the linearised recovery data will consume
   1770 */
   1771 static tdb_len_t tdb_recovery_size(struct tdb_context *tdb)
   1772 {
   1773 	struct tdb_transaction_el *el;
   1774 	tdb_len_t recovery_size = 0;
   1775 
   1776 	recovery_size = sizeof(u32);
   1777 	for (el=tdb->transaction->elements;el;el=el->next) {
   1778 		if (el->offset >= tdb->transaction->old_map_size) {
   1779 			continue;
   1780 		}
   1781 		recovery_size += 2*sizeof(tdb_off_t) + el->length;
   1782 	}
   1783 
   1784 	return recovery_size;
   1785 }
   1786 
   1787 /*
   1788   allocate the recovery area, or use an existing recovery area if it is
   1789   large enough
   1790 */
   1791 static int tdb_recovery_allocate(struct tdb_context *tdb,
   1792 				 tdb_len_t *recovery_size,
   1793 				 tdb_off_t *recovery_offset,
   1794 				 tdb_len_t *recovery_max_size)
   1795 {
   1796 	struct list_struct rec;
   1797 	const struct tdb_methods *methods = tdb->transaction->io_methods;
   1798 	tdb_off_t recovery_head;
   1799 
   1800 	if (tdb_ofs_read(tdb, TDB_RECOVERY_HEAD, &recovery_head) == -1) {
   1801 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to read recovery head\n"));
   1802 		return -1;
   1803 	}
   1804 
   1805 	rec.rec_len = 0;
   1806 
   1807 	if (recovery_head != 0 &&
   1808 	    methods->tdb_read(tdb, recovery_head, &rec, sizeof(rec), DOCONV()) == -1) {
   1809 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to read recovery record\n"));
   1810 		return -1;
   1811 	}
   1812 
   1813 	*recovery_size = tdb_recovery_size(tdb);
   1814 
   1815 	if (recovery_head != 0 && *recovery_size <= rec.rec_len) {
   1816 		/* it fits in the existing area */
   1817 		*recovery_max_size = rec.rec_len;
   1818 		*recovery_offset = recovery_head;
   1819 		return 0;
   1820 	}
   1821 
   1822 	/* we need to free up the old recovery area, then allocate a
   1823 	   new one at the end of the file. Note that we cannot use
   1824 	   tdb_allocate() to allocate the new one as that might return
   1825 	   us an area that is being currently used (as of the start of
   1826 	   the transaction) */
   1827 	if (recovery_head != 0) {
   1828 		if (tdb_free(tdb, recovery_head, &rec) == -1) {
   1829 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to free previous recovery area\n"));
   1830 			return -1;
   1831 		}
   1832 	}
   1833 
   1834 	/* the tdb_free() call might have increased the recovery size */
   1835 	*recovery_size = tdb_recovery_size(tdb);
   1836 
   1837 	/* round up to a multiple of page size */
   1838 	*recovery_max_size = TDB_ALIGN(sizeof(rec) + *recovery_size, tdb->page_size) - sizeof(rec);
   1839 	*recovery_offset = tdb->map_size;
   1840 	recovery_head = *recovery_offset;
   1841 
   1842 	if (methods->tdb_expand_file(tdb, tdb->transaction->old_map_size,
   1843 				     (tdb->map_size - tdb->transaction->old_map_size) +
   1844 				     sizeof(rec) + *recovery_max_size) == -1) {
   1845 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to create recovery area\n"));
   1846 		return -1;
   1847 	}
   1848 
   1849 	/* remap the file (if using mmap) */
   1850 	methods->tdb_oob(tdb, tdb->map_size + 1, 1);
   1851 
   1852 	/* we have to reset the old map size so that we don't try to expand the file
   1853 	   again in the transaction commit, which would destroy the recovery area */
   1854 	tdb->transaction->old_map_size = tdb->map_size;
   1855 
   1856 	/* write the recovery header offset and sync - we can sync without a race here
   1857 	   as the magic ptr in the recovery record has not been set */
   1858 	CONVERT(recovery_head);
   1859 	if (methods->tdb_write(tdb, TDB_RECOVERY_HEAD,
   1860 			       &recovery_head, sizeof(tdb_off_t)) == -1) {
   1861 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to write recovery head\n"));
   1862 		return -1;
   1863 	}
   1864 
   1865 	return 0;
   1866 }
   1867 
   1868 
   1869 /*
   1870   setup the recovery data that will be used on a crash during commit
   1871 */
   1872 static int transaction_setup_recovery(struct tdb_context *tdb,
   1873 				      tdb_off_t *magic_offset)
   1874 {
   1875 	struct tdb_transaction_el *el;
   1876 	tdb_len_t recovery_size;
   1877 	unsigned char *data, *p;
   1878 	const struct tdb_methods *methods = tdb->transaction->io_methods;
   1879 	struct list_struct *rec;
   1880 	tdb_off_t recovery_offset, recovery_max_size;
   1881 	tdb_off_t old_map_size = tdb->transaction->old_map_size;
   1882 	u32 magic, tailer;
   1883 
   1884 	/*
   1885 	  check that the recovery area has enough space
   1886 	*/
   1887 	if (tdb_recovery_allocate(tdb, &recovery_size,
   1888 				  &recovery_offset, &recovery_max_size) == -1) {
   1889 		return -1;
   1890 	}
   1891 
   1892 	data = (unsigned char *)malloc(recovery_size + sizeof(*rec));
   1893 	if (data == NULL) {
   1894 		tdb->ecode = TDB_ERR_OOM;
   1895 		return -1;
   1896 	}
   1897 
   1898 	rec = (struct list_struct *)data;
   1899 	memset(rec, 0, sizeof(*rec));
   1900 
   1901 	rec->magic    = 0;
   1902 	rec->data_len = recovery_size;
   1903 	rec->rec_len  = recovery_max_size;
   1904 	rec->key_len  = old_map_size;
   1905 	CONVERT(rec);
   1906 
   1907 	/* build the recovery data into a single blob to allow us to do a single
   1908 	   large write, which should be more efficient */
   1909 	p = data + sizeof(*rec);
   1910 	for (el=tdb->transaction->elements;el;el=el->next) {
   1911 		if (el->offset >= old_map_size) {
   1912 			continue;
   1913 		}
   1914 		if (el->offset + el->length > tdb->transaction->old_map_size) {
   1915 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: transaction data over new region boundary\n"));
   1916 			free(data);
   1917 			tdb->ecode = TDB_ERR_CORRUPT;
   1918 			return -1;
   1919 		}
   1920 		memcpy(p, &el->offset, 4);
   1921 		memcpy(p+4, &el->length, 4);
   1922 		if (DOCONV()) {
   1923 			tdb_convert(p, 8);
   1924 		}
   1925 		/* the recovery area contains the old data, not the
   1926 		   new data, so we have to call the original tdb_read
   1927 		   method to get it */
   1928 		if (methods->tdb_read(tdb, el->offset, p + 8, el->length, 0) != 0) {
   1929 			free(data);
   1930 			tdb->ecode = TDB_ERR_IO;
   1931 			return -1;
   1932 		}
   1933 		p += 8 + el->length;
   1934 	}
   1935 
   1936 	/* and the tailer */
   1937 	tailer = sizeof(*rec) + recovery_max_size;
   1938 	memcpy(p, &tailer, 4);
   1939 	CONVERT(p);
   1940 
   1941 	/* write the recovery data to the recovery area */
   1942 	if (methods->tdb_write(tdb, recovery_offset, data, sizeof(*rec) + recovery_size) == -1) {
   1943 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: failed to write recovery data\n"));
   1944 		free(data);
   1945 		tdb->ecode = TDB_ERR_IO;
   1946 		return -1;
   1947 	}
   1948 
   1949 	/* as we don't have ordered writes, we have to sync the recovery
   1950 	   data before we update the magic to indicate that the recovery
   1951 	   data is present */
   1952 	if (transaction_sync(tdb, recovery_offset, sizeof(*rec) + recovery_size) == -1) {
   1953 		free(data);
   1954 		return -1;
   1955 	}
   1956 
   1957 	free(data);
   1958 
   1959 	magic = TDB_RECOVERY_MAGIC;
   1960 	CONVERT(magic);
   1961 
   1962 	*magic_offset = recovery_offset + offsetof(struct list_struct, magic);
   1963 
   1964 	if (methods->tdb_write(tdb, *magic_offset, &magic, sizeof(magic)) == -1) {
   1965 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: failed to write recovery magic\n"));
   1966 		tdb->ecode = TDB_ERR_IO;
   1967 		return -1;
   1968 	}
   1969 
   1970 	/* ensure the recovery magic marker is on disk */
   1971 	if (transaction_sync(tdb, *magic_offset, sizeof(magic)) == -1) {
   1972 		return -1;
   1973 	}
   1974 
   1975 	return 0;
   1976 }
   1977 
   1978 /*
   1979   commit the current transaction
   1980 */
   1981 int tdb_transaction_commit(struct tdb_context *tdb)
   1982 {
   1983 	const struct tdb_methods *methods;
   1984 	tdb_off_t magic_offset = 0;
   1985 	u32 zero = 0;
   1986 
   1987 	if (tdb->transaction == NULL) {
   1988 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_commit: no transaction\n"));
   1989 		return -1;
   1990 	}
   1991 
   1992 	if (tdb->transaction->transaction_error) {
   1993 		tdb->ecode = TDB_ERR_IO;
   1994 		tdb_transaction_cancel(tdb);
   1995 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_commit: transaction error pending\n"));
   1996 		return -1;
   1997 	}
   1998 
   1999 	if (tdb->transaction->nesting != 0) {
   2000 		tdb->transaction->nesting--;
   2001 		return 0;
   2002 	}
   2003 
   2004 	/* check for a null transaction */
   2005 	if (tdb->transaction->elements == NULL) {
   2006 		tdb_transaction_cancel(tdb);
   2007 		return 0;
   2008 	}
   2009 
   2010 	methods = tdb->transaction->io_methods;
   2011 
   2012 	/* if there are any locks pending then the caller has not
   2013 	   nested their locks properly, so fail the transaction */
   2014 	if (tdb->num_locks || tdb->global_lock.count) {
   2015 		tdb->ecode = TDB_ERR_LOCK;
   2016 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_commit: locks pending on commit\n"));
   2017 		tdb_transaction_cancel(tdb);
   2018 		return -1;
   2019 	}
   2020 
   2021 	/* upgrade the main transaction lock region to a write lock */
   2022 	if (tdb_brlock_upgrade(tdb, FREELIST_TOP, 0) == -1) {
   2023 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: failed to upgrade hash locks\n"));
   2024 		tdb->ecode = TDB_ERR_LOCK;
   2025 		tdb_transaction_cancel(tdb);
   2026 		return -1;
   2027 	}
   2028 
   2029 	/* get the global lock - this prevents new users attaching to the database
   2030 	   during the commit */
   2031 	if (tdb_brlock(tdb, GLOBAL_LOCK, F_WRLCK, F_SETLKW, 0, 1) == -1) {
   2032 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_commit: failed to get global lock\n"));
   2033 		tdb->ecode = TDB_ERR_LOCK;
   2034 		tdb_transaction_cancel(tdb);
   2035 		return -1;
   2036 	}
   2037 
   2038 	if (!(tdb->flags & TDB_NOSYNC)) {
   2039 		/* write the recovery data to the end of the file */
   2040 		if (transaction_setup_recovery(tdb, &magic_offset) == -1) {
   2041 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_commit: failed to setup recovery data\n"));
   2042 			tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
   2043 			tdb_transaction_cancel(tdb);
   2044 			return -1;
   2045 		}
   2046 	}
   2047 
   2048 	/* expand the file to the new size if needed */
   2049 	if (tdb->map_size != tdb->transaction->old_map_size) {
   2050 		if (methods->tdb_expand_file(tdb, tdb->transaction->old_map_size,
   2051 					     tdb->map_size -
   2052 					     tdb->transaction->old_map_size) == -1) {
   2053 			tdb->ecode = TDB_ERR_IO;
   2054 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_commit: expansion failed\n"));
   2055 			tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
   2056 			tdb_transaction_cancel(tdb);
   2057 			return -1;
   2058 		}
   2059 		tdb->map_size = tdb->transaction->old_map_size;
   2060 		methods->tdb_oob(tdb, tdb->map_size + 1, 1);
   2061 	}
   2062 
   2063 	/* perform all the writes */
   2064 	while (tdb->transaction->elements) {
   2065 		struct tdb_transaction_el *el = tdb->transaction->elements;
   2066 
   2067 		if (methods->tdb_write(tdb, el->offset, el->data, el->length) == -1) {
   2068 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_commit: write failed during commit\n"));
   2069 
   2070 			/* we've overwritten part of the data and
   2071 			   possibly expanded the file, so we need to
   2072 			   run the crash recovery code */
   2073 			tdb->methods = methods;
   2074 			tdb_transaction_recover(tdb);
   2075 
   2076 			tdb_transaction_cancel(tdb);
   2077 			tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
   2078 
   2079 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_commit: write failed\n"));
   2080 			return -1;
   2081 		}
   2082 		tdb->transaction->elements = el->next;
   2083 		free(el->data);
   2084 		free(el);
   2085 	}
   2086 
   2087 	if (!(tdb->flags & TDB_NOSYNC)) {
   2088 		/* ensure the new data is on disk */
   2089 		if (transaction_sync(tdb, 0, tdb->map_size) == -1) {
   2090 			return -1;
   2091 		}
   2092 
   2093 		/* remove the recovery marker */
   2094 		if (methods->tdb_write(tdb, magic_offset, &zero, 4) == -1) {
   2095 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_commit: failed to remove recovery magic\n"));
   2096 			return -1;
   2097 		}
   2098 
   2099 		/* ensure the recovery marker has been removed on disk */
   2100 		if (transaction_sync(tdb, magic_offset, 4) == -1) {
   2101 			return -1;
   2102 		}
   2103 	}
   2104 
   2105 	tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
   2106 
   2107 	/*
   2108 	  TODO: maybe write to some dummy hdr field, or write to magic
   2109 	  offset without mmap, before the last sync, instead of the
   2110 	  utime() call
   2111 	*/
   2112 
   2113 	/* on some systems (like Linux 2.6.x) changes via mmap/msync
   2114 	   don't change the mtime of the file, this means the file may
   2115 	   not be backed up (as tdb rounding to block sizes means that
   2116 	   file size changes are quite rare too). The following forces
   2117 	   mtime changes when a transaction completes */
   2118 #ifdef HAVE_UTIME
   2119 	utime(tdb->name, NULL);
   2120 #endif
   2121 
   2122 	/* use a transaction cancel to free memory and remove the
   2123 	   transaction locks */
   2124 	tdb_transaction_cancel(tdb);
   2125 	return 0;
   2126 }
   2127 
   2128 
   2129 /*
   2130   recover from an aborted transaction. Must be called with exclusive
   2131   database write access already established (including the global
   2132   lock to prevent new processes attaching)
   2133 */
   2134 int tdb_transaction_recover(struct tdb_context *tdb)
   2135 {
   2136 	tdb_off_t recovery_head, recovery_eof;
   2137 	unsigned char *data, *p;
   2138 	u32 zero = 0;
   2139 	struct list_struct rec;
   2140 
   2141 	/* find the recovery area */
   2142 	if (tdb_ofs_read(tdb, TDB_RECOVERY_HEAD, &recovery_head) == -1) {
   2143 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to read recovery head\n"));
   2144 		tdb->ecode = TDB_ERR_IO;
   2145 		return -1;
   2146 	}
   2147 
   2148 	if (recovery_head == 0) {
   2149 		/* we have never allocated a recovery record */
   2150 		return 0;
   2151 	}
   2152 
   2153 	/* read the recovery record */
   2154 	if (tdb->methods->tdb_read(tdb, recovery_head, &rec,
   2155 				   sizeof(rec), DOCONV()) == -1) {
   2156 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to read recovery record\n"));
   2157 		tdb->ecode = TDB_ERR_IO;
   2158 		return -1;
   2159 	}
   2160 
   2161 	if (rec.magic != TDB_RECOVERY_MAGIC) {
   2162 		/* there is no valid recovery data */
   2163 		return 0;
   2164 	}
   2165 
   2166 	if (tdb->read_only) {
   2167 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: attempt to recover read only database\n"));
   2168 		tdb->ecode = TDB_ERR_CORRUPT;
   2169 		return -1;
   2170 	}
   2171 
   2172 	recovery_eof = rec.key_len;
   2173 
   2174 	data = (unsigned char *)malloc(rec.data_len);
   2175 	if (data == NULL) {
   2176 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to allocate recovery data\n"));
   2177 		tdb->ecode = TDB_ERR_OOM;
   2178 		return -1;
   2179 	}
   2180 
   2181 	/* read the full recovery data */
   2182 	if (tdb->methods->tdb_read(tdb, recovery_head + sizeof(rec), data,
   2183 				   rec.data_len, 0) == -1) {
   2184 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to read recovery data\n"));
   2185 		tdb->ecode = TDB_ERR_IO;
   2186 		return -1;
   2187 	}
   2188 
   2189 	/* recover the file data */
   2190 	p = data;
   2191 	while (p+8 < data + rec.data_len) {
   2192 		u32 ofs, len;
   2193 		if (DOCONV()) {
   2194 			tdb_convert(p, 8);
   2195 		}
   2196 		memcpy(&ofs, p, 4);
   2197 		memcpy(&len, p+4, 4);
   2198 
   2199 		if (tdb->methods->tdb_write(tdb, ofs, p+8, len) == -1) {
   2200 			free(data);
   2201 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to recover %d bytes at offset %d\n", len, ofs));
   2202 			tdb->ecode = TDB_ERR_IO;
   2203 			return -1;
   2204 		}
   2205 		p += 8 + len;
   2206 	}
   2207 
   2208 	free(data);
   2209 
   2210 	if (transaction_sync(tdb, 0, tdb->map_size) == -1) {
   2211 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to sync recovery\n"));
   2212 		tdb->ecode = TDB_ERR_IO;
   2213 		return -1;
   2214 	}
   2215 
   2216 	/* if the recovery area is after the recovered eof then remove it */
   2217 	if (recovery_eof <= recovery_head) {
   2218 		if (tdb_ofs_write(tdb, TDB_RECOVERY_HEAD, &zero) == -1) {
   2219 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to remove recovery head\n"));
   2220 			tdb->ecode = TDB_ERR_IO;
   2221 			return -1;
   2222 		}
   2223 	}
   2224 
   2225 	/* remove the recovery magic */
   2226 	if (tdb_ofs_write(tdb, recovery_head + offsetof(struct list_struct, magic),
   2227 			  &zero) == -1) {
   2228 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to remove recovery magic\n"));
   2229 		tdb->ecode = TDB_ERR_IO;
   2230 		return -1;
   2231 	}
   2232 
   2233 	/* reduce the file size to the old size */
   2234 	tdb_munmap(tdb);
   2235 	if (ftruncate(tdb->fd, recovery_eof) != 0) {
   2236 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to reduce to recovery size\n"));
   2237 		tdb->ecode = TDB_ERR_IO;
   2238 		return -1;
   2239 	}
   2240 	tdb->map_size = recovery_eof;
   2241 	tdb_mmap(tdb);
   2242 
   2243 	if (transaction_sync(tdb, 0, recovery_eof) == -1) {
   2244 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to sync2 recovery\n"));
   2245 		tdb->ecode = TDB_ERR_IO;
   2246 		return -1;
   2247 	}
   2248 
   2249 	TDB_LOG((tdb, TDB_DEBUG_TRACE, "tdb_transaction_recover: recovered %d byte database\n",
   2250 		 recovery_eof));
   2251 
   2252 	/* all done */
   2253 	return 0;
   2254 }
   2255 
   2256 /* file: freelist.c */
   2257 
   2258 /* read a freelist record and check for simple errors */
   2259 static int tdb_rec_free_read(struct tdb_context *tdb, tdb_off_t off, struct list_struct *rec)
   2260 {
   2261 	if (tdb->methods->tdb_read(tdb, off, rec, sizeof(*rec),DOCONV()) == -1)
   2262 		return -1;
   2263 
   2264 	if (rec->magic == TDB_MAGIC) {
   2265 		/* this happens when a app is showdown while deleting a record - we should
   2266 		   not completely fail when this happens */
   2267 		TDB_LOG((tdb, TDB_DEBUG_WARNING, "tdb_rec_free_read non-free magic 0x%x at offset=%d - fixing\n",
   2268 			 rec->magic, off));
   2269 		rec->magic = TDB_FREE_MAGIC;
   2270 		if (tdb->methods->tdb_write(tdb, off, rec, sizeof(*rec)) == -1)
   2271 			return -1;
   2272 	}
   2273 
   2274 	if (rec->magic != TDB_FREE_MAGIC) {
   2275 		/* Ensure ecode is set for log fn. */
   2276 		tdb->ecode = TDB_ERR_CORRUPT;
   2277 		TDB_LOG((tdb, TDB_DEBUG_WARNING, "tdb_rec_free_read bad magic 0x%x at offset=%d\n",
   2278 			   rec->magic, off));
   2279 		return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
   2280 	}
   2281 	if (tdb->methods->tdb_oob(tdb, rec->next+sizeof(*rec), 0) != 0)
   2282 		return -1;
   2283 	return 0;
   2284 }
   2285 
   2286 
   2287 
   2288 /* Remove an element from the freelist.  Must have alloc lock. */
   2289 static int remove_from_freelist(struct tdb_context *tdb, tdb_off_t off, tdb_off_t next)
   2290 {
   2291 	tdb_off_t last_ptr, i;
   2292 
   2293 	/* read in the freelist top */
   2294 	last_ptr = FREELIST_TOP;
   2295 	while (tdb_ofs_read(tdb, last_ptr, &i) != -1 && i != 0) {
   2296 		if (i == off) {
   2297 			/* We've found it! */
   2298 			return tdb_ofs_write(tdb, last_ptr, &next);
   2299 		}
   2300 		/* Follow chain (next offset is at start of record) */
   2301 		last_ptr = i;
   2302 	}
   2303 	TDB_LOG((tdb, TDB_DEBUG_FATAL,"remove_from_freelist: not on list at off=%d\n", off));
   2304 	return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
   2305 }
   2306 
   2307 
   2308 /* update a record tailer (must hold allocation lock) */
   2309 static int update_tailer(struct tdb_context *tdb, tdb_off_t offset,
   2310 			 const struct list_struct *rec)
   2311 {
   2312 	tdb_off_t totalsize;
   2313 
   2314 	/* Offset of tailer from record header */
   2315 	totalsize = sizeof(*rec) + rec->rec_len;
   2316 	return tdb_ofs_write(tdb, offset + totalsize - sizeof(tdb_off_t),
   2317 			 &totalsize);
   2318 }
   2319 
   2320 /* Add an element into the freelist. Merge adjacent records if
   2321    neccessary. */
   2322 int tdb_free(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec)
   2323 {
   2324 	tdb_off_t right, left;
   2325 
   2326 	/* Allocation and tailer lock */
   2327 	if (tdb_lock(tdb, -1, F_WRLCK) != 0)
   2328 		return -1;
   2329 
   2330 	/* set an initial tailer, so if we fail we don't leave a bogus record */
   2331 	if (update_tailer(tdb, offset, rec) != 0) {
   2332 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free: update_tailer failed!\n"));
   2333 		goto fail;
   2334 	}
   2335 
   2336 	/* Look right first (I'm an Australian, dammit) */
   2337 	right = offset + sizeof(*rec) + rec->rec_len;
   2338 	if (right + sizeof(*rec) <= tdb->map_size) {
   2339 		struct list_struct r;
   2340 
   2341 		if (tdb->methods->tdb_read(tdb, right, &r, sizeof(r), DOCONV()) == -1) {
   2342 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free: right read failed at %u\n", right));
   2343 			goto left;
   2344 		}
   2345 
   2346 		/* If it's free, expand to include it. */
   2347 		if (r.magic == TDB_FREE_MAGIC) {
   2348 			if (remove_from_freelist(tdb, right, r.next) == -1) {
   2349 				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free: right free failed at %u\n", right));
   2350 				goto left;
   2351 			}
   2352 			rec->rec_len += sizeof(r) + r.rec_len;
   2353 		}
   2354 	}
   2355 
   2356 left:
   2357 	/* Look left */
   2358 	left = offset - sizeof(tdb_off_t);
   2359 	if (left > TDB_DATA_START(tdb->header.hash_size)) {
   2360 		struct list_struct l;
   2361 		tdb_off_t leftsize;
   2362 
   2363 		/* Read in tailer and jump back to header */
   2364 		if (tdb_ofs_read(tdb, left, &leftsize) == -1) {
   2365 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free: left offset read failed at %u\n", left));
   2366 			goto update;
   2367 		}
   2368 
   2369 		/* it could be uninitialised data */
   2370 		if (leftsize == 0 || leftsize == TDB_PAD_U32) {
   2371 			goto update;
   2372 		}
   2373 
   2374 		left = offset - leftsize;
   2375 
   2376 		/* Now read in record */
   2377 		if (tdb->methods->tdb_read(tdb, left, &l, sizeof(l), DOCONV()) == -1) {
   2378 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free: left read failed at %u (%u)\n", left, leftsize));
   2379 			goto update;
   2380 		}
   2381 
   2382 		/* If it's free, expand to include it. */
   2383 		if (l.magic == TDB_FREE_MAGIC) {
   2384 			if (remove_from_freelist(tdb, left, l.next) == -1) {
   2385 				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free: left free failed at %u\n", left));
   2386 				goto update;
   2387 			} else {
   2388 				offset = left;
   2389 				rec->rec_len += leftsize;
   2390 			}
   2391 		}
   2392 	}
   2393 
   2394 update:
   2395 	if (update_tailer(tdb, offset, rec) == -1) {
   2396 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free: update_tailer failed at %u\n", offset));
   2397 		goto fail;
   2398 	}
   2399 
   2400 	/* Now, prepend to free list */
   2401 	rec->magic = TDB_FREE_MAGIC;
   2402 
   2403 	if (tdb_ofs_read(tdb, FREELIST_TOP, &rec->next) == -1 ||
   2404 	    tdb_rec_write(tdb, offset, rec) == -1 ||
   2405 	    tdb_ofs_write(tdb, FREELIST_TOP, &offset) == -1) {
   2406 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free record write failed at offset=%d\n", offset));
   2407 		goto fail;
   2408 	}
   2409 
   2410 	/* And we're done. */
   2411 	tdb_unlock(tdb, -1, F_WRLCK);
   2412 	return 0;
   2413 
   2414  fail:
   2415 	tdb_unlock(tdb, -1, F_WRLCK);
   2416 	return -1;
   2417 }
   2418 
   2419 
   2420 /*
   2421    the core of tdb_allocate - called when we have decided which
   2422    free list entry to use
   2423  */
   2424 static tdb_off_t tdb_allocate_ofs(struct tdb_context *tdb, tdb_len_t length, tdb_off_t rec_ptr,
   2425 				struct list_struct *rec, tdb_off_t last_ptr)
   2426 {
   2427 	struct list_struct newrec;
   2428 	tdb_off_t newrec_ptr;
   2429 
   2430 	memset(&newrec, '\0', sizeof(newrec));
   2431 
   2432 	/* found it - now possibly split it up  */
   2433 	if (rec->rec_len > length + MIN_REC_SIZE) {
   2434 		/* Length of left piece */
   2435 		length = TDB_ALIGN(length, TDB_ALIGNMENT);
   2436 
   2437 		/* Right piece to go on free list */
   2438 		newrec.rec_len = rec->rec_len - (sizeof(*rec) + length);
   2439 		newrec_ptr = rec_ptr + sizeof(*rec) + length;
   2440 
   2441 		/* And left record is shortened */
   2442 		rec->rec_len = length;
   2443 	} else {
   2444 		newrec_ptr = 0;
   2445 	}
   2446 
   2447 	/* Remove allocated record from the free list */
   2448 	if (tdb_ofs_write(tdb, last_ptr, &rec->next) == -1) {
   2449 		return 0;
   2450 	}
   2451 
   2452 	/* Update header: do this before we drop alloc
   2453 	   lock, otherwise tdb_free() might try to
   2454 	   merge with us, thinking we're free.
   2455 	   (Thanks Jeremy Allison). */
   2456 	rec->magic = TDB_MAGIC;
   2457 	if (tdb_rec_write(tdb, rec_ptr, rec) == -1) {
   2458 		return 0;
   2459 	}
   2460 
   2461 	/* Did we create new block? */
   2462 	if (newrec_ptr) {
   2463 		/* Update allocated record tailer (we
   2464 		   shortened it). */
   2465 		if (update_tailer(tdb, rec_ptr, rec) == -1) {
   2466 			return 0;
   2467 		}
   2468 
   2469 		/* Free new record */
   2470 		if (tdb_free(tdb, newrec_ptr, &newrec) == -1) {
   2471 			return 0;
   2472 		}
   2473 	}
   2474 
   2475 	/* all done - return the new record offset */
   2476 	return rec_ptr;
   2477 }
   2478 
   2479 /* allocate some space from the free list. The offset returned points
   2480    to a unconnected list_struct within the database with room for at
   2481    least length bytes of total data
   2482 
   2483    0 is returned if the space could not be allocated
   2484  */
   2485 tdb_off_t tdb_allocate(struct tdb_context *tdb, tdb_len_t length, struct list_struct *rec)
   2486 {
   2487 	tdb_off_t rec_ptr, last_ptr, newrec_ptr;
   2488 	struct {
   2489 		tdb_off_t rec_ptr, last_ptr;
   2490 		tdb_len_t rec_len;
   2491 	} bestfit;
   2492 
   2493 	if (tdb_lock(tdb, -1, F_WRLCK) == -1)
   2494 		return 0;
   2495 
   2496 	/* Extra bytes required for tailer */
   2497 	length += sizeof(tdb_off_t);
   2498 
   2499  again:
   2500 	last_ptr = FREELIST_TOP;
   2501 
   2502 	/* read in the freelist top */
   2503 	if (tdb_ofs_read(tdb, FREELIST_TOP, &rec_ptr) == -1)
   2504 		goto fail;
   2505 
   2506 	bestfit.rec_ptr = 0;
   2507 	bestfit.last_ptr = 0;
   2508 	bestfit.rec_len = 0;
   2509 
   2510 	/*
   2511 	   this is a best fit allocation strategy. Originally we used
   2512 	   a first fit strategy, but it suffered from massive fragmentation
   2513 	   issues when faced with a slowly increasing record size.
   2514 	 */
   2515 	while (rec_ptr) {
   2516 		if (tdb_rec_free_read(tdb, rec_ptr, rec) == -1) {
   2517 			goto fail;
   2518 		}
   2519 
   2520 		if (rec->rec_len >= length) {
   2521 			if (bestfit.rec_ptr == 0 ||
   2522 			    rec->rec_len < bestfit.rec_len) {
   2523 				bestfit.rec_len = rec->rec_len;
   2524 				bestfit.rec_ptr = rec_ptr;
   2525 				bestfit.last_ptr = last_ptr;
   2526 				/* consider a fit to be good enough if
   2527 				   we aren't wasting more than half
   2528 				   the space */
   2529 				if (bestfit.rec_len < 2*length) {
   2530 					break;
   2531 				}
   2532 			}
   2533 		}
   2534 
   2535 		/* move to the next record */
   2536 		last_ptr = rec_ptr;
   2537 		rec_ptr = rec->next;
   2538 	}
   2539 
   2540 	if (bestfit.rec_ptr != 0) {
   2541 		if (tdb_rec_free_read(tdb, bestfit.rec_ptr, rec) == -1) {
   2542 			goto fail;
   2543 		}
   2544 
   2545 		newrec_ptr = tdb_allocate_ofs(tdb, length, bestfit.rec_ptr, rec, bestfit.last_ptr);
   2546 		tdb_unlock(tdb, -1, F_WRLCK);
   2547 		return newrec_ptr;
   2548 	}
   2549 
   2550 	/* we didn't find enough space. See if we can expand the
   2551 	   database and if we can then try again */
   2552 	if (tdb_expand(tdb, length + sizeof(*rec)) == 0)
   2553 		goto again;
   2554  fail:
   2555 	tdb_unlock(tdb, -1, F_WRLCK);
   2556 	return 0;
   2557 }
   2558 
   2559 /* file: freelistcheck.c */
   2560 
   2561 /* Check the freelist is good and contains no loops.
   2562    Very memory intensive - only do this as a consistency
   2563    checker. Heh heh - uses an in memory tdb as the storage
   2564    for the "seen" record list. For some reason this strikes
   2565    me as extremely clever as I don't have to write another tree
   2566    data structure implementation :-).
   2567  */
   2568 
   2569 static int seen_insert(struct tdb_context *mem_tdb, tdb_off_t rec_ptr)
   2570 {
   2571 	TDB_DATA key, data;
   2572 
   2573 	memset(&data, '\0', sizeof(data));
   2574 	key.dptr = (unsigned char *)&rec_ptr;
   2575 	key.dsize = sizeof(rec_ptr);
   2576 	return tdb_store(mem_tdb, key, data, TDB_INSERT);
   2577 }
   2578 
   2579 int tdb_validate_freelist(struct tdb_context *tdb, int *pnum_entries)
   2580 {
   2581 	struct tdb_context *mem_tdb = NULL;
   2582 	struct list_struct rec;
   2583 	tdb_off_t rec_ptr, last_ptr;
   2584 	int ret = -1;
   2585 
   2586 	*pnum_entries = 0;
   2587 
   2588 	mem_tdb = tdb_open("flval", tdb->header.hash_size,
   2589 				TDB_INTERNAL, O_RDWR, 0600);
   2590 	if (!mem_tdb) {
   2591 		return -1;
   2592 	}
   2593 
   2594 	if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
   2595 		tdb_close(mem_tdb);
   2596 		return 0;
   2597 	}
   2598 
   2599 	last_ptr = FREELIST_TOP;
   2600 
   2601 	/* Store the FREELIST_TOP record. */
   2602 	if (seen_insert(mem_tdb, last_ptr) == -1) {
   2603 		ret = TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
   2604 		goto fail;
   2605 	}
   2606 
   2607 	/* read in the freelist top */
   2608 	if (tdb_ofs_read(tdb, FREELIST_TOP, &rec_ptr) == -1) {
   2609 		goto fail;
   2610 	}
   2611 
   2612 	while (rec_ptr) {
   2613 
   2614 		/* If we can't store this record (we've seen it
   2615 		   before) then the free list has a loop and must
   2616 		   be corrupt. */
   2617 
   2618 		if (seen_insert(mem_tdb, rec_ptr)) {
   2619 			ret = TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
   2620 			goto fail;
   2621 		}
   2622 
   2623 		if (tdb_rec_free_read(tdb, rec_ptr, &rec) == -1) {
   2624 			goto fail;
   2625 		}
   2626 
   2627 		/* move to the next record */
   2628 		last_ptr = rec_ptr;
   2629 		rec_ptr = rec.next;
   2630 		*pnum_entries += 1;
   2631 	}
   2632 
   2633 	ret = 0;
   2634 
   2635   fail:
   2636 
   2637 	tdb_close(mem_tdb);
   2638 	tdb_unlock(tdb, -1, F_WRLCK);
   2639 	return ret;
   2640 }
   2641 
   2642 /* file: traverse.c */
   2643 
   2644 /* Uses traverse lock: 0 = finish, -1 = error, other = record offset */
   2645 static int tdb_next_lock(struct tdb_context *tdb, struct tdb_traverse_lock *tlock,
   2646 			 struct list_struct *rec)
   2647 {
   2648 	int want_next = (tlock->off != 0);
   2649 
   2650 	/* Lock each chain from the start one. */
   2651 	for (; tlock->hash < tdb->header.hash_size; tlock->hash++) {
   2652 		if (!tlock->off && tlock->hash != 0) {
   2653 			/* this is an optimisation for the common case where
   2654 			   the hash chain is empty, which is particularly
   2655 			   common for the use of tdb with ldb, where large
   2656 			   hashes are used. In that case we spend most of our
   2657 			   time in tdb_brlock(), locking empty hash chains.
   2658 
   2659 			   To avoid this, we do an unlocked pre-check to see
   2660 			   if the hash chain is empty before starting to look
   2661 			   inside it. If it is empty then we can avoid that
   2662 			   hash chain. If it isn't empty then we can't believe
   2663 			   the value we get back, as we read it without a
   2664 			   lock, so instead we get the lock and re-fetch the
   2665 			   value below.
   2666 
   2667 			   Notice that not doing this optimisation on the
   2668 			   first hash chain is critical. We must guarantee
   2669 			   that we have done at least one fcntl lock at the
   2670 			   start of a search to guarantee that memory is
   2671 			   coherent on SMP systems. If records are added by
   2672 			   others during the search then thats OK, and we
   2673 			   could possibly miss those with this trick, but we
   2674 			   could miss them anyway without this trick, so the
   2675 			   semantics don't change.
   2676 
   2677 			   With a non-indexed ldb search this trick gains us a
   2678 			   factor of around 80 in speed on a linux 2.6.x
   2679 			   system (testing using ldbtest).
   2680 			*/
   2681 			tdb->methods->next_hash_chain(tdb, &tlock->hash);
   2682 			if (tlock->hash == tdb->header.hash_size) {
   2683 				continue;
   2684 			}
   2685 		}
   2686 
   2687 		if (tdb_lock(tdb, tlock->hash, tlock->lock_rw) == -1)
   2688 			return -1;
   2689 
   2690 		/* No previous record?  Start at top of chain. */
   2691 		if (!tlock->off) {
   2692 			if (tdb_ofs_read(tdb, TDB_HASH_TOP(tlock->hash),
   2693 				     &tlock->off) == -1)
   2694 				goto fail;
   2695 		} else {
   2696 			/* Otherwise unlock the previous record. */
   2697 			if (tdb_unlock_record(tdb, tlock->off) != 0)
   2698 				goto fail;
   2699 		}
   2700 
   2701 		if (want_next) {
   2702 			/* We have offset of old record: grab next */
   2703 			if (tdb_rec_read(tdb, tlock->off, rec) == -1)
   2704 				goto fail;
   2705 			tlock->off = rec->next;
   2706 		}
   2707 
   2708 		/* Iterate through chain */
   2709 		while( tlock->off) {
   2710 			tdb_off_t current;
   2711 			if (tdb_rec_read(tdb, tlock->off, rec) == -1)
   2712 				goto fail;
   2713 
   2714 			/* Detect infinite loops. From "Shlomi Yaakobovich" <Shlomi (at) exanet.com>. */
   2715 			if (tlock->off == rec->next) {
   2716 				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_next_lock: loop detected.\n"));
   2717 				goto fail;
   2718 			}
   2719 
   2720 			if (!TDB_DEAD(rec)) {
   2721 				/* Woohoo: we found one! */
   2722 				if (tdb_lock_record(tdb, tlock->off) != 0)
   2723 					goto fail;
   2724 				return tlock->off;
   2725 			}
   2726 
   2727 			/* Try to clean dead ones from old traverses */
   2728 			current = tlock->off;
   2729 			tlock->off = rec->next;
   2730 			if (!(tdb->read_only || tdb->traverse_read) &&
   2731 			    tdb_do_delete(tdb, current, rec) != 0)
   2732 				goto fail;
   2733 		}
   2734 		tdb_unlock(tdb, tlock->hash, tlock->lock_rw);
   2735 		want_next = 0;
   2736 	}
   2737 	/* We finished iteration without finding anything */
   2738 	return TDB_ERRCODE(TDB_SUCCESS, 0);
   2739 
   2740  fail:
   2741 	tlock->off = 0;
   2742 	if (tdb_unlock(tdb, tlock->hash, tlock->lock_rw) != 0)
   2743 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_next_lock: On error unlock failed!\n"));
   2744 	return -1;
   2745 }
   2746 
   2747 /* traverse the entire database - calling fn(tdb, key, data) on each element.
   2748    return -1 on error or the record count traversed
   2749    if fn is NULL then it is not called
   2750    a non-zero return value from fn() indicates that the traversal should stop
   2751   */
   2752 static int tdb_traverse_internal(struct tdb_context *tdb,
   2753 				 tdb_traverse_func fn, void *private_data,
   2754 				 struct tdb_traverse_lock *tl)
   2755 {
   2756 	TDB_DATA key, dbuf;
   2757 	struct list_struct rec;
   2758 	int ret, count = 0;
   2759 
   2760 	/* This was in the initializaton, above, but the IRIX compiler
   2761 	 * did not like it.  crh
   2762 	 */
   2763 	tl->next = tdb->travlocks.next;
   2764 
   2765 	/* fcntl locks don't stack: beware traverse inside traverse */
   2766 	tdb->travlocks.next = tl;
   2767 
   2768 	/* tdb_next_lock places locks on the record returned, and its chain */
   2769 	while ((ret = tdb_next_lock(tdb, tl, &rec)) > 0) {
   2770 		count++;
   2771 		/* now read the full record */
   2772 		key.dptr = tdb_alloc_read(tdb, tl->off + sizeof(rec),
   2773 					  rec.key_len + rec.data_len);
   2774 		if (!key.dptr) {
   2775 			ret = -1;
   2776 			if (tdb_unlock(tdb, tl->hash, tl->lock_rw) != 0)
   2777 				goto out;
   2778 			if (tdb_unlock_record(tdb, tl->off) != 0)
   2779 				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_traverse: key.dptr == NULL and unlock_record failed!\n"));
   2780 			goto out;
   2781 		}
   2782 		key.dsize = rec.key_len;
   2783 		dbuf.dptr = key.dptr + rec.key_len;
   2784 		dbuf.dsize = rec.data_len;
   2785 
   2786 		/* Drop chain lock, call out */
   2787 		if (tdb_unlock(tdb, tl->hash, tl->lock_rw) != 0) {
   2788 			ret = -1;
   2789 			SAFE_FREE(key.dptr);
   2790 			goto out;
   2791 		}
   2792 		if (fn && fn(tdb, key, dbuf, private_data)) {
   2793 			/* They want us to terminate traversal */
   2794 			ret = count;
   2795 			if (tdb_unlock_record(tdb, tl->off) != 0) {
   2796 				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_traverse: unlock_record failed!\n"));;
   2797 				ret = -1;
   2798 			}
   2799 			SAFE_FREE(key.dptr);
   2800 			goto out;
   2801 		}
   2802 		SAFE_FREE(key.dptr);
   2803 	}
   2804 out:
   2805 	tdb->travlocks.next = tl->next;
   2806 	if (ret < 0)
   2807 		return -1;
   2808 	else
   2809 		return count;
   2810 }
   2811 
   2812 
   2813 /*
   2814   a write style traverse - temporarily marks the db read only
   2815 */
   2816 int tdb_traverse_read(struct tdb_context *tdb,
   2817 		      tdb_traverse_func fn, void *private_data)
   2818 {
   2819 	struct tdb_traverse_lock tl = { NULL, 0, 0, F_RDLCK };
   2820 	int ret;
   2821 
   2822 	/* we need to get a read lock on the transaction lock here to
   2823 	   cope with the lock ordering semantics of solaris10 */
   2824 	if (tdb_transaction_lock(tdb, F_RDLCK)) {
   2825 		return -1;
   2826 	}
   2827 
   2828 	tdb->traverse_read++;
   2829 	ret = tdb_traverse_internal(tdb, fn, private_data, &tl);
   2830 	tdb->traverse_read--;
   2831 
   2832 	tdb_transaction_unlock(tdb);
   2833 
   2834 	return ret;
   2835 }
   2836 
   2837 /*
   2838   a write style traverse - needs to get the transaction lock to
   2839   prevent deadlocks
   2840 */
   2841 int tdb_traverse(struct tdb_context *tdb,
   2842 		 tdb_traverse_func fn, void *private_data)
   2843 {
   2844 	struct tdb_traverse_lock tl = { NULL, 0, 0, F_WRLCK };
   2845 	int ret;
   2846 
   2847 	if (tdb->read_only || tdb->traverse_read) {
   2848 		return tdb_traverse_read(tdb, fn, private_data);
   2849 	}
   2850 
   2851 	if (tdb_transaction_lock(tdb, F_WRLCK)) {
   2852 		return -1;
   2853 	}
   2854 
   2855 	ret = tdb_traverse_internal(tdb, fn, private_data, &tl);
   2856 
   2857 	tdb_transaction_unlock(tdb);
   2858 
   2859 	return ret;
   2860 }
   2861 
   2862 
   2863 /* find the first entry in the database and return its key */
   2864 TDB_DATA tdb_firstkey(struct tdb_context *tdb)
   2865 {
   2866 	TDB_DATA key;
   2867 	struct list_struct rec;
   2868 
   2869 	/* release any old lock */
   2870 	if (tdb_unlock_record(tdb, tdb->travlocks.off) != 0)
   2871 		return tdb_null;
   2872 	tdb->travlocks.off = tdb->travlocks.hash = 0;
   2873 	tdb->travlocks.lock_rw = F_RDLCK;
   2874 
   2875 	/* Grab first record: locks chain and returned record. */
   2876 	if (tdb_next_lock(tdb, &tdb->travlocks, &rec) <= 0)
   2877 		return tdb_null;
   2878 	/* now read the key */
   2879 	key.dsize = rec.key_len;
   2880 	key.dptr =tdb_alloc_read(tdb,tdb->travlocks.off+sizeof(rec),key.dsize);
   2881 
   2882 	/* Unlock the hash chain of the record we just read. */
   2883 	if (tdb_unlock(tdb, tdb->travlocks.hash, tdb->travlocks.lock_rw) != 0)
   2884 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_firstkey: error occurred while tdb_unlocking!\n"));
   2885 	return key;
   2886 }
   2887 
   2888 /* find the next entry in the database, returning its key */
   2889 TDB_DATA tdb_nextkey(struct tdb_context *tdb, TDB_DATA oldkey)
   2890 {
   2891 	u32 oldhash;
   2892 	TDB_DATA key = tdb_null;
   2893 	struct list_struct rec;
   2894 	unsigned char *k = NULL;
   2895 
   2896 	/* Is locked key the old key?  If so, traverse will be reliable. */
   2897 	if (tdb->travlocks.off) {
   2898 		if (tdb_lock(tdb,tdb->travlocks.hash,tdb->travlocks.lock_rw))
   2899 			return tdb_null;
   2900 		if (tdb_rec_read(tdb, tdb->travlocks.off, &rec) == -1
   2901 		    || !(k = tdb_alloc_read(tdb,tdb->travlocks.off+sizeof(rec),
   2902 					    rec.key_len))
   2903 		    || memcmp(k, oldkey.dptr, oldkey.dsize) != 0) {
   2904 			/* No, it wasn't: unlock it and start from scratch */
   2905 			if (tdb_unlock_record(tdb, tdb->travlocks.off) != 0) {
   2906 				SAFE_FREE(k);
   2907 				return tdb_null;
   2908 			}
   2909 			if (tdb_unlock(tdb, tdb->travlocks.hash, tdb->travlocks.lock_rw) != 0) {
   2910 				SAFE_FREE(k);
   2911 				return tdb_null;
   2912 			}
   2913 			tdb->travlocks.off = 0;
   2914 		}
   2915 
   2916 		SAFE_FREE(k);
   2917 	}
   2918 
   2919 	if (!tdb->travlocks.off) {
   2920 		/* No previous element: do normal find, and lock record */
   2921 		tdb->travlocks.off = tdb_find_lock_hash(tdb, oldkey, tdb->hash_fn(&oldkey), tdb->travlocks.lock_rw, &rec);
   2922 		if (!tdb->travlocks.off)
   2923 			return tdb_null;
   2924 		tdb->travlocks.hash = BUCKET(rec.full_hash);
   2925 		if (tdb_lock_record(tdb, tdb->travlocks.off) != 0) {
   2926 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_nextkey: lock_record failed (%s)!\n", strerror(errno)));
   2927 			return tdb_null;
   2928 		}
   2929 	}
   2930 	oldhash = tdb->travlocks.hash;
   2931 
   2932 	/* Grab next record: locks chain and returned record,
   2933 	   unlocks old record */
   2934 	if (tdb_next_lock(tdb, &tdb->travlocks, &rec) > 0) {
   2935 		key.dsize = rec.key_len;
   2936 		key.dptr = tdb_alloc_read(tdb, tdb->travlocks.off+sizeof(rec),
   2937 					  key.dsize);
   2938 		/* Unlock the chain of this new record */
   2939 		if (tdb_unlock(tdb, tdb->travlocks.hash, tdb->travlocks.lock_rw) != 0)
   2940 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_nextkey: WARNING tdb_unlock failed!\n"));
   2941 	}
   2942 	/* Unlock the chain of old record */
   2943 	if (tdb_unlock(tdb, BUCKET(oldhash), tdb->travlocks.lock_rw) != 0)
   2944 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_nextkey: WARNING tdb_unlock failed!\n"));
   2945 	return key;
   2946 }
   2947 
   2948 /* file: dump.c */
   2949 
   2950 static tdb_off_t tdb_dump_record(struct tdb_context *tdb, int hash,
   2951 				 tdb_off_t offset)
   2952 {
   2953 	struct list_struct rec;
   2954 	tdb_off_t tailer_ofs, tailer;
   2955 
   2956 	if (tdb->methods->tdb_read(tdb, offset, (char *)&rec,
   2957 				   sizeof(rec), DOCONV()) == -1) {
   2958 		printf("ERROR: failed to read record at %u\n", offset);
   2959 		return 0;
   2960 	}
   2961 
   2962 	printf(" rec: hash=%d offset=0x%08x next=0x%08x rec_len=%d "
   2963 	       "key_len=%d data_len=%d full_hash=0x%x magic=0x%x\n",
   2964 	       hash, offset, rec.next, rec.rec_len, rec.key_len, rec.data_len,
   2965 	       rec.full_hash, rec.magic);
   2966 
   2967 	tailer_ofs = offset + sizeof(rec) + rec.rec_len - sizeof(tdb_off_t);
   2968 
   2969 	if (tdb_ofs_read(tdb, tailer_ofs, &tailer) == -1) {
   2970 		printf("ERROR: failed to read tailer at %u\n", tailer_ofs);
   2971 		return rec.next;
   2972 	}
   2973 
   2974 	if (tailer != rec.rec_len + sizeof(rec)) {
   2975 		printf("ERROR: tailer does not match record! tailer=%u totalsize=%u\n",
   2976 				(unsigned int)tailer, (unsigned int)(rec.rec_len + sizeof(rec)));
   2977 	}
   2978 	return rec.next;
   2979 }
   2980 
   2981 static int tdb_dump_chain(struct tdb_context *tdb, int i)
   2982 {
   2983 	tdb_off_t rec_ptr, top;
   2984 
   2985 	top = TDB_HASH_TOP(i);
   2986 
   2987 	if (tdb_lock(tdb, i, F_WRLCK) != 0)
   2988 		return -1;
   2989 
   2990 	if (tdb_ofs_read(tdb, top, &rec_ptr) == -1)
   2991 		return tdb_unlock(tdb, i, F_WRLCK);
   2992 
   2993 	if (rec_ptr)
   2994 		printf("hash=%d\n", i);
   2995 
   2996 	while (rec_ptr) {
   2997 		rec_ptr = tdb_dump_record(tdb, i, rec_ptr);
   2998 	}
   2999 
   3000 	return tdb_unlock(tdb, i, F_WRLCK);
   3001 }
   3002 
   3003 void tdb_dump_all(struct tdb_context *tdb)
   3004 {
   3005 	int i;
   3006 	for (i=0;i<tdb->header.hash_size;i++) {
   3007 		tdb_dump_chain(tdb, i);
   3008 	}
   3009 	printf("freelist:\n");
   3010 	tdb_dump_chain(tdb, -1);
   3011 }
   3012 
   3013 int tdb_printfreelist(struct tdb_context *tdb)
   3014 {
   3015 	int ret;
   3016 	long total_free = 0;
   3017 	tdb_off_t offset, rec_ptr;
   3018 	struct list_struct rec;
   3019 
   3020 	if ((ret = tdb_lock(tdb, -1, F_WRLCK)) != 0)
   3021 		return ret;
   3022 
   3023 	offset = FREELIST_TOP;
   3024 
   3025 	/* read in the freelist top */
   3026 	if (tdb_ofs_read(tdb, offset, &rec_ptr) == -1) {
   3027 		tdb_unlock(tdb, -1, F_WRLCK);
   3028 		return 0;
   3029 	}
   3030 
   3031 	printf("freelist top=[0x%08x]\n", rec_ptr );
   3032 	while (rec_ptr) {
   3033 		if (tdb->methods->tdb_read(tdb, rec_ptr, (char *)&rec,
   3034 					   sizeof(rec), DOCONV()) == -1) {
   3035 			tdb_unlock(tdb, -1, F_WRLCK);
   3036 			return -1;
   3037 		}
   3038 
   3039 		if (rec.magic != TDB_FREE_MAGIC) {
   3040 			printf("bad magic 0x%08x in free list\n", rec.magic);
   3041 			tdb_unlock(tdb, -1, F_WRLCK);
   3042 			return -1;
   3043 		}
   3044 
   3045 		printf("entry offset=[0x%08x], rec.rec_len = [0x%08x (%d)] (end = 0x%08x)\n",
   3046 		       rec_ptr, rec.rec_len, rec.rec_len, rec_ptr + rec.rec_len);
   3047 		total_free += rec.rec_len;
   3048 
   3049 		/* move to the next record */
   3050 		rec_ptr = rec.next;
   3051 	}
   3052 	printf("total rec_len = [0x%08x (%d)]\n", (int)total_free,
   3053                (int)total_free);
   3054 
   3055 	return tdb_unlock(tdb, -1, F_WRLCK);
   3056 }
   3057 
   3058 /* file: tdb.c */
   3059 
   3060 TDB_DATA tdb_null;
   3061 
   3062 /*
   3063   non-blocking increment of the tdb sequence number if the tdb has been opened using
   3064   the TDB_SEQNUM flag
   3065 */
   3066 void tdb_increment_seqnum_nonblock(struct tdb_context *tdb)
   3067 {
   3068 	tdb_off_t seqnum=0;
   3069 
   3070 	if (!(tdb->flags & TDB_SEQNUM)) {
   3071 		return;
   3072 	}
   3073 
   3074 	/* we ignore errors from this, as we have no sane way of
   3075 	   dealing with them.
   3076 	*/
   3077 	tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &seqnum);
   3078 	seqnum++;
   3079 	tdb_ofs_write(tdb, TDB_SEQNUM_OFS, &seqnum);
   3080 }
   3081 
   3082 /*
   3083   increment the tdb sequence number if the tdb has been opened using
   3084   the TDB_SEQNUM flag
   3085 */
   3086 static void tdb_increment_seqnum(struct tdb_context *tdb)
   3087 {
   3088 	if (!(tdb->flags & TDB_SEQNUM)) {
   3089 		return;
   3090 	}
   3091 
   3092 	if (tdb_brlock(tdb, TDB_SEQNUM_OFS, F_WRLCK, F_SETLKW, 1, 1) != 0) {
   3093 		return;
   3094 	}
   3095 
   3096 	tdb_increment_seqnum_nonblock(tdb);
   3097 
   3098 	tdb_brlock(tdb, TDB_SEQNUM_OFS, F_UNLCK, F_SETLKW, 1, 1);
   3099 }
   3100 
   3101 static int tdb_key_compare(TDB_DATA key, TDB_DATA data, void *private_data)
   3102 {
   3103 	return memcmp(data.dptr, key.dptr, data.dsize);
   3104 }
   3105 
   3106 /* Returns 0 on fail.  On success, return offset of record, and fills
   3107    in rec */
   3108 static tdb_off_t tdb_find(struct tdb_context *tdb, TDB_DATA key, u32 hash,
   3109 			struct list_struct *r)
   3110 {
   3111 	tdb_off_t rec_ptr;
   3112 
   3113 	/* read in the hash top */
   3114 	if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
   3115 		return 0;
   3116 
   3117 	/* keep looking until we find the right record */
   3118 	while (rec_ptr) {
   3119 		if (tdb_rec_read(tdb, rec_ptr, r) == -1)
   3120 			return 0;
   3121 
   3122 		if (!TDB_DEAD(r) && hash==r->full_hash
   3123 		    && key.dsize==r->key_len
   3124 		    && tdb_parse_data(tdb, key, rec_ptr + sizeof(*r),
   3125 				      r->key_len, tdb_key_compare,
   3126 				      NULL) == 0) {
   3127 			return rec_ptr;
   3128 		}
   3129 		rec_ptr = r->next;
   3130 	}
   3131 	return TDB_ERRCODE(TDB_ERR_NOEXIST, 0);
   3132 }
   3133 
   3134 /* As tdb_find, but if you succeed, keep the lock */
   3135 tdb_off_t tdb_find_lock_hash(struct tdb_context *tdb, TDB_DATA key, u32 hash, int locktype,
   3136 			   struct list_struct *rec)
   3137 {
   3138 	u32 rec_ptr;
   3139 
   3140 	if (tdb_lock(tdb, BUCKET(hash), locktype) == -1)
   3141 		return 0;
   3142 	if (!(rec_ptr = tdb_find(tdb, key, hash, rec)))
   3143 		tdb_unlock(tdb, BUCKET(hash), locktype);
   3144 	return rec_ptr;
   3145 }
   3146 
   3147 
   3148 /* update an entry in place - this only works if the new data size
   3149    is <= the old data size and the key exists.
   3150    on failure return -1.
   3151 */
   3152 static int tdb_update_hash(struct tdb_context *tdb, TDB_DATA key, u32 hash, TDB_DATA dbuf)
   3153 {
   3154 	struct list_struct rec;
   3155 	tdb_off_t rec_ptr;
   3156 
   3157 	/* find entry */
   3158 	if (!(rec_ptr = tdb_find(tdb, key, hash, &rec)))
   3159 		return -1;
   3160 
   3161 	/* must be long enough key, data and tailer */
   3162 	if (rec.rec_len < key.dsize + dbuf.dsize + sizeof(tdb_off_t)) {
   3163 		tdb->ecode = TDB_SUCCESS; /* Not really an error */
   3164 		return -1;
   3165 	}
   3166 
   3167 	if (tdb->methods->tdb_write(tdb, rec_ptr + sizeof(rec) + rec.key_len,
   3168 		      dbuf.dptr, dbuf.dsize) == -1)
   3169 		return -1;
   3170 
   3171 	if (dbuf.dsize != rec.data_len) {
   3172 		/* update size */
   3173 		rec.data_len = dbuf.dsize;
   3174 		return tdb_rec_write(tdb, rec_ptr, &rec);
   3175 	}
   3176 
   3177 	return 0;
   3178 }
   3179 
   3180 /* find an entry in the database given a key */
   3181 /* If an entry doesn't exist tdb_err will be set to
   3182  * TDB_ERR_NOEXIST. If a key has no data attached
   3183  * then the TDB_DATA will have zero length but
   3184  * a non-zero pointer
   3185  */
   3186 TDB_DATA tdb_fetch(struct tdb_context *tdb, TDB_DATA key)
   3187 {
   3188 	tdb_off_t rec_ptr;
   3189 	struct list_struct rec;
   3190 	TDB_DATA ret;
   3191 	u32 hash;
   3192 
   3193 	/* find which hash bucket it is in */
   3194 	hash = tdb->hash_fn(&key);
   3195 	if (!(rec_ptr = tdb_find_lock_hash(tdb,key,hash,F_RDLCK,&rec)))
   3196 		return tdb_null;
   3197 
   3198 	ret.dptr = tdb_alloc_read(tdb, rec_ptr + sizeof(rec) + rec.key_len,
   3199 				  rec.data_len);
   3200 	ret.dsize = rec.data_len;
   3201 	tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
   3202 	return ret;
   3203 }
   3204 
   3205 /*
   3206  * Find an entry in the database and hand the record's data to a parsing
   3207  * function. The parsing function is executed under the chain read lock, so it
   3208  * should be fast and should not block on other syscalls.
   3209  *
   3210  * DONT CALL OTHER TDB CALLS FROM THE PARSER, THIS MIGHT LEAD TO SEGFAULTS.
   3211  *
   3212  * For mmapped tdb's that do not have a transaction open it points the parsing
   3213  * function directly at the mmap area, it avoids the malloc/memcpy in this
   3214  * case. If a transaction is open or no mmap is available, it has to do
   3215  * malloc/read/parse/free.
   3216  *
   3217  * This is interesting for all readers of potentially large data structures in
   3218  * the tdb records, ldb indexes being one example.
   3219  */
   3220 
   3221 int tdb_parse_record(struct tdb_context *tdb, TDB_DATA key,
   3222 		     int (*parser)(TDB_DATA key, TDB_DATA data,
   3223 				   void *private_data),
   3224 		     void *private_data)
   3225 {
   3226 	tdb_off_t rec_ptr;
   3227 	struct list_struct rec;
   3228 	int ret;
   3229 	u32 hash;
   3230 
   3231 	/* find which hash bucket it is in */
   3232 	hash = tdb->hash_fn(&key);
   3233 
   3234 	if (!(rec_ptr = tdb_find_lock_hash(tdb,key,hash,F_RDLCK,&rec))) {
   3235 		return TDB_ERRCODE(TDB_ERR_NOEXIST, 0);
   3236 	}
   3237 
   3238 	ret = tdb_parse_data(tdb, key, rec_ptr + sizeof(rec) + rec.key_len,
   3239 			     rec.data_len, parser, private_data);
   3240 
   3241 	tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
   3242 
   3243 	return ret;
   3244 }
   3245 
   3246 /* check if an entry in the database exists
   3247 
   3248    note that 1 is returned if the key is found and 0 is returned if not found
   3249    this doesn't match the conventions in the rest of this module, but is
   3250    compatible with gdbm
   3251 */
   3252 static int tdb_exists_hash(struct tdb_context *tdb, TDB_DATA key, u32 hash)
   3253 {
   3254 	struct list_struct rec;
   3255 
   3256 	if (tdb_find_lock_hash(tdb, key, hash, F_RDLCK, &rec) == 0)
   3257 		return 0;
   3258 	tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
   3259 	return 1;
   3260 }
   3261 
   3262 int tdb_exists(struct tdb_context *tdb, TDB_DATA key)
   3263 {
   3264 	u32 hash = tdb->hash_fn(&key);
   3265 	return tdb_exists_hash(tdb, key, hash);
   3266 }
   3267 
   3268 /* actually delete an entry in the database given the offset */
   3269 int tdb_do_delete(struct tdb_context *tdb, tdb_off_t rec_ptr, struct list_struct*rec)
   3270 {
   3271 	tdb_off_t last_ptr, i;
   3272 	struct list_struct lastrec;
   3273 
   3274 	if (tdb->read_only || tdb->traverse_read) return -1;
   3275 
   3276 	if (tdb_write_lock_record(tdb, rec_ptr) == -1) {
   3277 		/* Someone traversing here: mark it as dead */
   3278 		rec->magic = TDB_DEAD_MAGIC;
   3279 		return tdb_rec_write(tdb, rec_ptr, rec);
   3280 	}
   3281 	if (tdb_write_unlock_record(tdb, rec_ptr) != 0)
   3282 		return -1;
   3283 
   3284 	/* find previous record in hash chain */
   3285 	if (tdb_ofs_read(tdb, TDB_HASH_TOP(rec->full_hash), &i) == -1)
   3286 		return -1;
   3287 	for (last_ptr = 0; i != rec_ptr; last_ptr = i, i = lastrec.next)
   3288 		if (tdb_rec_read(tdb, i, &lastrec) == -1)
   3289 			return -1;
   3290 
   3291 	/* unlink it: next ptr is at start of record. */
   3292 	if (last_ptr == 0)
   3293 		last_ptr = TDB_HASH_TOP(rec->full_hash);
   3294 	if (tdb_ofs_write(tdb, last_ptr, &rec->next) == -1)
   3295 		return -1;
   3296 
   3297 	/* recover the space */
   3298 	if (tdb_free(tdb, rec_ptr, rec) == -1)
   3299 		return -1;
   3300 	return 0;
   3301 }
   3302 
   3303 static int tdb_count_dead(struct tdb_context *tdb, u32 hash)
   3304 {
   3305 	int res = 0;
   3306 	tdb_off_t rec_ptr;
   3307 	struct list_struct rec;
   3308 
   3309 	/* read in the hash top */
   3310 	if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
   3311 		return 0;
   3312 
   3313 	while (rec_ptr) {
   3314 		if (tdb_rec_read(tdb, rec_ptr, &rec) == -1)
   3315 			return 0;
   3316 
   3317 		if (rec.magic == TDB_DEAD_MAGIC) {
   3318 			res += 1;
   3319 		}
   3320 		rec_ptr = rec.next;
   3321 	}
   3322 	return res;
   3323 }
   3324 
   3325 /*
   3326  * Purge all DEAD records from a hash chain
   3327  */
   3328 static int tdb_purge_dead(struct tdb_context *tdb, u32 hash)
   3329 {
   3330 	int res = -1;
   3331 	struct list_struct rec;
   3332 	tdb_off_t rec_ptr;
   3333 
   3334 	if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
   3335 		return -1;
   3336 	}
   3337 
   3338 	/* read in the hash top */
   3339 	if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
   3340 		goto fail;
   3341 
   3342 	while (rec_ptr) {
   3343 		tdb_off_t next;
   3344 
   3345 		if (tdb_rec_read(tdb, rec_ptr, &rec) == -1) {
   3346 			goto fail;
   3347 		}
   3348 
   3349 		next = rec.next;
   3350 
   3351 		if (rec.magic == TDB_DEAD_MAGIC
   3352 		    && tdb_do_delete(tdb, rec_ptr, &rec) == -1) {
   3353 			goto fail;
   3354 		}
   3355 		rec_ptr = next;
   3356 	}
   3357 	res = 0;
   3358  fail:
   3359 	tdb_unlock(tdb, -1, F_WRLCK);
   3360 	return res;
   3361 }
   3362 
   3363 /* delete an entry in the database given a key */
   3364 static int tdb_delete_hash(struct tdb_context *tdb, TDB_DATA key, u32 hash)
   3365 {
   3366 	tdb_off_t rec_ptr;
   3367 	struct list_struct rec;
   3368 	int ret;
   3369 
   3370 	if (tdb->max_dead_records != 0) {
   3371 
   3372 		/*
   3373 		 * Allow for some dead records per hash chain, mainly for
   3374 		 * tdb's with a very high create/delete rate like locking.tdb.
   3375 		 */
   3376 
   3377 		if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
   3378 			return -1;
   3379 
   3380 		if (tdb_count_dead(tdb, hash) >= tdb->max_dead_records) {
   3381 			/*
   3382 			 * Don't let the per-chain freelist grow too large,
   3383 			 * delete all existing dead records
   3384 			 */
   3385 			tdb_purge_dead(tdb, hash);
   3386 		}
   3387 
   3388 		if (!(rec_ptr = tdb_find(tdb, key, hash, &rec))) {
   3389 			tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
   3390 			return -1;
   3391 		}
   3392 
   3393 		/*
   3394 		 * Just mark the record as dead.
   3395 		 */
   3396 		rec.magic = TDB_DEAD_MAGIC;
   3397 		ret = tdb_rec_write(tdb, rec_ptr, &rec);
   3398 	}
   3399 	else {
   3400 		if (!(rec_ptr = tdb_find_lock_hash(tdb, key, hash, F_WRLCK,
   3401 						   &rec)))
   3402 			return -1;
   3403 
   3404 		ret = tdb_do_delete(tdb, rec_ptr, &rec);
   3405 	}
   3406 
   3407 	if (ret == 0) {
   3408 		tdb_increment_seqnum(tdb);
   3409 	}
   3410 
   3411 	if (tdb_unlock(tdb, BUCKET(rec.full_hash), F_WRLCK) != 0)
   3412 		TDB_LOG((tdb, TDB_DEBUG_WARNING, "tdb_delete: WARNING tdb_unlock failed!\n"));
   3413 	return ret;
   3414 }
   3415 
   3416 int tdb_delete(struct tdb_context *tdb, TDB_DATA key)
   3417 {
   3418 	u32 hash = tdb->hash_fn(&key);
   3419 	return tdb_delete_hash(tdb, key, hash);
   3420 }
   3421 
   3422 /*
   3423  * See if we have a dead record around with enough space
   3424  */
   3425 static tdb_off_t tdb_find_dead(struct tdb_context *tdb, u32 hash,
   3426 			       struct list_struct *r, tdb_len_t length)
   3427 {
   3428 	tdb_off_t rec_ptr;
   3429 
   3430 	/* read in the hash top */
   3431 	if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
   3432 		return 0;
   3433 
   3434 	/* keep looking until we find the right record */
   3435 	while (rec_ptr) {
   3436 		if (tdb_rec_read(tdb, rec_ptr, r) == -1)
   3437 			return 0;
   3438 
   3439 		if (TDB_DEAD(r) && r->rec_len >= length) {
   3440 			/*
   3441 			 * First fit for simple coding, TODO: change to best
   3442 			 * fit
   3443 			 */
   3444 			return rec_ptr;
   3445 		}
   3446 		rec_ptr = r->next;
   3447 	}
   3448 	return 0;
   3449 }
   3450 
   3451 /* store an element in the database, replacing any existing element
   3452    with the same key
   3453 
   3454    return 0 on success, -1 on failure
   3455 */
   3456 int tdb_store(struct tdb_context *tdb, TDB_DATA key, TDB_DATA dbuf, int flag)
   3457 {
   3458 	struct list_struct rec;
   3459 	u32 hash;
   3460 	tdb_off_t rec_ptr;
   3461 	char *p = NULL;
   3462 	int ret = -1;
   3463 
   3464 	if (tdb->read_only || tdb->traverse_read) {
   3465 		tdb->ecode = TDB_ERR_RDONLY;
   3466 		return -1;
   3467 	}
   3468 
   3469 	/* find which hash bucket it is in */
   3470 	hash = tdb->hash_fn(&key);
   3471 	if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
   3472 		return -1;
   3473 
   3474 	/* check for it existing, on insert. */
   3475 	if (flag == TDB_INSERT) {
   3476 		if (tdb_exists_hash(tdb, key, hash)) {
   3477 			tdb->ecode = TDB_ERR_EXISTS;
   3478 			goto fail;
   3479 		}
   3480 	} else {
   3481 		/* first try in-place update, on modify or replace. */
   3482 		if (tdb_update_hash(tdb, key, hash, dbuf) == 0) {
   3483 			goto done;
   3484 		}
   3485 		if (tdb->ecode == TDB_ERR_NOEXIST &&
   3486 		    flag == TDB_MODIFY) {
   3487 			/* if the record doesn't exist and we are in TDB_MODIFY mode then
   3488 			 we should fail the store */
   3489 			goto fail;
   3490 		}
   3491 	}
   3492 	/* reset the error code potentially set by the tdb_update() */
   3493 	tdb->ecode = TDB_SUCCESS;
   3494 
   3495 	/* delete any existing record - if it doesn't exist we don't
   3496            care.  Doing this first reduces fragmentation, and avoids
   3497            coalescing with `allocated' block before it's updated. */
   3498 	if (flag != TDB_INSERT)
   3499 		tdb_delete_hash(tdb, key, hash);
   3500 
   3501 	/* Copy key+value *before* allocating free space in case malloc
   3502 	   fails and we are left with a dead spot in the tdb. */
   3503 
   3504 	if (!(p = (char *)malloc(key.dsize + dbuf.dsize))) {
   3505 		tdb->ecode = TDB_ERR_OOM;
   3506 		goto fail;
   3507 	}
   3508 
   3509 	memcpy(p, key.dptr, key.dsize);
   3510 	if (dbuf.dsize)
   3511 		memcpy(p+key.dsize, dbuf.dptr, dbuf.dsize);
   3512 
   3513 	if (tdb->max_dead_records != 0) {
   3514 		/*
   3515 		 * Allow for some dead records per hash chain, look if we can
   3516 		 * find one that can hold the new record. We need enough space
   3517 		 * for key, data and tailer. If we find one, we don't have to
   3518 		 * consult the central freelist.
   3519 		 */
   3520 		rec_ptr = tdb_find_dead(
   3521 			tdb, hash, &rec,
   3522 			key.dsize + dbuf.dsize + sizeof(tdb_off_t));
   3523 
   3524 		if (rec_ptr != 0) {
   3525 			rec.key_len = key.dsize;
   3526 			rec.data_len = dbuf.dsize;
   3527 			rec.full_hash = hash;
   3528 			rec.magic = TDB_MAGIC;
   3529 			if (tdb_rec_write(tdb, rec_ptr, &rec) == -1
   3530 			    || tdb->methods->tdb_write(
   3531 				    tdb, rec_ptr + sizeof(rec),
   3532 				    p, key.dsize + dbuf.dsize) == -1) {
   3533 				goto fail;
   3534 			}
   3535 			goto done;
   3536 		}
   3537 	}
   3538 
   3539 	/*
   3540 	 * We have to allocate some space from the freelist, so this means we
   3541 	 * have to lock it. Use the chance to purge all the DEAD records from
   3542 	 * the hash chain under the freelist lock.
   3543 	 */
   3544 
   3545 	if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
   3546 		goto fail;
   3547 	}
   3548 
   3549 	if ((tdb->max_dead_records != 0)
   3550 	    && (tdb_purge_dead(tdb, hash) == -1)) {
   3551 		tdb_unlock(tdb, -1, F_WRLCK);
   3552 		goto fail;
   3553 	}
   3554 
   3555 	/* we have to allocate some space */
   3556 	rec_ptr = tdb_allocate(tdb, key.dsize + dbuf.dsize, &rec);
   3557 
   3558 	tdb_unlock(tdb, -1, F_WRLCK);
   3559 
   3560 	if (rec_ptr == 0) {
   3561 		goto fail;
   3562 	}
   3563 
   3564 	/* Read hash top into next ptr */
   3565 	if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec.next) == -1)
   3566 		goto fail;
   3567 
   3568 	rec.key_len = key.dsize;
   3569 	rec.data_len = dbuf.dsize;
   3570 	rec.full_hash = hash;
   3571 	rec.magic = TDB_MAGIC;
   3572 
   3573 	/* write out and point the top of the hash chain at it */
   3574 	if (tdb_rec_write(tdb, rec_ptr, &rec) == -1
   3575 	    || tdb->methods->tdb_write(tdb, rec_ptr+sizeof(rec), p, key.dsize+dbuf.dsize)==-1
   3576 	    || tdb_ofs_write(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1) {
   3577 		/* Need to tdb_unallocate() here */
   3578 		goto fail;
   3579 	}
   3580 
   3581  done:
   3582 	ret = 0;
   3583  fail:
   3584 	if (ret == 0) {
   3585 		tdb_increment_seqnum(tdb);
   3586 	}
   3587 
   3588 	SAFE_FREE(p);
   3589 	tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
   3590 	return ret;
   3591 }
   3592 
   3593 
   3594 /* Append to an entry. Create if not exist. */
   3595 int tdb_append(struct tdb_context *tdb, TDB_DATA key, TDB_DATA new_dbuf)
   3596 {
   3597 	u32 hash;
   3598 	TDB_DATA dbuf;
   3599 	int ret = -1;
   3600 
   3601 	/* find which hash bucket it is in */
   3602 	hash = tdb->hash_fn(&key);
   3603 	if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
   3604 		return -1;
   3605 
   3606 	dbuf = tdb_fetch(tdb, key);
   3607 
   3608 	if (dbuf.dptr == NULL) {
   3609 		dbuf.dptr = (unsigned char *)malloc(new_dbuf.dsize);
   3610 	} else {
   3611 		unsigned char *new_dptr = (unsigned char *)realloc(dbuf.dptr,
   3612 						     dbuf.dsize + new_dbuf.dsize);
   3613 		if (new_dptr == NULL) {
   3614 			free(dbuf.dptr);
   3615 		}
   3616 		dbuf.dptr = new_dptr;
   3617 	}
   3618 
   3619 	if (dbuf.dptr == NULL) {
   3620 		tdb->ecode = TDB_ERR_OOM;
   3621 		goto failed;
   3622 	}
   3623 
   3624 	memcpy(dbuf.dptr + dbuf.dsize, new_dbuf.dptr, new_dbuf.dsize);
   3625 	dbuf.dsize += new_dbuf.dsize;
   3626 
   3627 	ret = tdb_store(tdb, key, dbuf, 0);
   3628 
   3629 failed:
   3630 	tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
   3631 	SAFE_FREE(dbuf.dptr);
   3632 	return ret;
   3633 }
   3634 
   3635 
   3636 /*
   3637   return the name of the current tdb file
   3638   useful for external logging functions
   3639 */
   3640 const char *tdb_name(struct tdb_context *tdb)
   3641 {
   3642 	return tdb->name;
   3643 }
   3644 
   3645 /*
   3646   return the underlying file descriptor being used by tdb, or -1
   3647   useful for external routines that want to check the device/inode
   3648   of the fd
   3649 */
   3650 int tdb_fd(struct tdb_context *tdb)
   3651 {
   3652 	return tdb->fd;
   3653 }
   3654 
   3655 /*
   3656   return the current logging function
   3657   useful for external tdb routines that wish to log tdb errors
   3658 */
   3659 tdb_log_func tdb_log_fn(struct tdb_context *tdb)
   3660 {
   3661 	return tdb->log.log_fn;
   3662 }
   3663 
   3664 
   3665 /*
   3666   get the tdb sequence number. Only makes sense if the writers opened
   3667   with TDB_SEQNUM set. Note that this sequence number will wrap quite
   3668   quickly, so it should only be used for a 'has something changed'
   3669   test, not for code that relies on the count of the number of changes
   3670   made. If you want a counter then use a tdb record.
   3671 
   3672   The aim of this sequence number is to allow for a very lightweight
   3673   test of a possible tdb change.
   3674 */
   3675 int tdb_get_seqnum(struct tdb_context *tdb)
   3676 {
   3677 	tdb_off_t seqnum=0;
   3678 
   3679 	tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &seqnum);
   3680 	return seqnum;
   3681 }
   3682 
   3683 int tdb_hash_size(struct tdb_context *tdb)
   3684 {
   3685 	return tdb->header.hash_size;
   3686 }
   3687 
   3688 size_t tdb_map_size(struct tdb_context *tdb)
   3689 {
   3690 	return tdb->map_size;
   3691 }
   3692 
   3693 int tdb_get_flags(struct tdb_context *tdb)
   3694 {
   3695 	return tdb->flags;
   3696 }
   3697 
   3698 
   3699 /*
   3700   enable sequence number handling on an open tdb
   3701 */
   3702 void tdb_enable_seqnum(struct tdb_context *tdb)
   3703 {
   3704 	tdb->flags |= TDB_SEQNUM;
   3705 }
   3706 
   3707 /* file: open.c */
   3708 
   3709 /* all contexts, to ensure no double-opens (fcntl locks don't nest!) */
   3710 static struct tdb_context *tdbs = NULL;
   3711 
   3712 
   3713 /* This is based on the hash algorithm from gdbm */
   3714 static unsigned int default_tdb_hash(TDB_DATA *key)
   3715 {
   3716 	u32 value;	/* Used to compute the hash value.  */
   3717 	u32   i;	/* Used to cycle through random values. */
   3718 
   3719 	/* Set the initial value from the key size. */
   3720 	for (value = 0x238F13AF * key->dsize, i=0; i < key->dsize; i++)
   3721 		value = (value + (key->dptr[i] << (i*5 % 24)));
   3722 
   3723 	return (1103515243 * value + 12345);
   3724 }
   3725 
   3726 
   3727 /* initialise a new database with a specified hash size */
   3728 static int tdb_new_database(struct tdb_context *tdb, int hash_size)
   3729 {
   3730 	struct tdb_header *newdb;
   3731 	int size, ret = -1;
   3732 
   3733 	/* We make it up in memory, then write it out if not internal */
   3734 	size = sizeof(struct tdb_header) + (hash_size+1)*sizeof(tdb_off_t);
   3735 	if (!(newdb = (struct tdb_header *)calloc(size, 1)))
   3736 		return TDB_ERRCODE(TDB_ERR_OOM, -1);
   3737 
   3738 	/* Fill in the header */
   3739 	newdb->version = TDB_VERSION;
   3740 	newdb->hash_size = hash_size;
   3741 	if (tdb->flags & TDB_INTERNAL) {
   3742 		tdb->map_size = size;
   3743 		tdb->map_ptr = (char *)newdb;
   3744 		memcpy(&tdb->header, newdb, sizeof(tdb->header));
   3745 		/* Convert the `ondisk' version if asked. */
   3746 		CONVERT(*newdb);
   3747 		return 0;
   3748 	}
   3749 	if (lseek(tdb->fd, 0, SEEK_SET) == -1)
   3750 		goto fail;
   3751 
   3752 	if (ftruncate(tdb->fd, 0) == -1)
   3753 		goto fail;
   3754 
   3755 	/* This creates an endian-converted header, as if read from disk */
   3756 	CONVERT(*newdb);
   3757 	memcpy(&tdb->header, newdb, sizeof(tdb->header));
   3758 	/* Don't endian-convert the magic food! */
   3759 	memcpy(newdb->magic_food, TDB_MAGIC_FOOD, strlen(TDB_MAGIC_FOOD)+1);
   3760 	if (write(tdb->fd, newdb, size) != size) {
   3761 		ret = -1;
   3762 	} else {
   3763 		ret = 0;
   3764 	}
   3765 
   3766   fail:
   3767 	SAFE_FREE(newdb);
   3768 	return ret;
   3769 }
   3770 
   3771 
   3772 
   3773 static int tdb_already_open(dev_t device,
   3774 			    ino_t ino)
   3775 {
   3776 	struct tdb_context *i;
   3777 
   3778 	for (i = tdbs; i; i = i->next) {
   3779 		if (i->device == device && i->inode == ino) {
   3780 			return 1;
   3781 		}
   3782 	}
   3783 
   3784 	return 0;
   3785 }
   3786 
   3787 /* open the database, creating it if necessary
   3788 
   3789    The open_flags and mode are passed straight to the open call on the
   3790    database file. A flags value of O_WRONLY is invalid. The hash size
   3791    is advisory, use zero for a default value.
   3792 
   3793    Return is NULL on error, in which case errno is also set.  Don't
   3794    try to call tdb_error or tdb_errname, just do strerror(errno).
   3795 
   3796    @param name may be NULL for internal databases. */
   3797 struct tdb_context *tdb_open(const char *name, int hash_size, int tdb_flags,
   3798 		      int open_flags, mode_t mode)
   3799 {
   3800 	return tdb_open_ex(name, hash_size, tdb_flags, open_flags, mode, NULL, NULL);
   3801 }
   3802 
   3803 /* a default logging function */
   3804 static void null_log_fn(struct tdb_context *tdb, enum tdb_debug_level level, const char *fmt, ...) PRINTF_ATTRIBUTE(3, 4);
   3805 static void null_log_fn(struct tdb_context *tdb, enum tdb_debug_level level, const char *fmt, ...)
   3806 {
   3807 }
   3808 
   3809 
   3810 struct tdb_context *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
   3811 				int open_flags, mode_t mode,
   3812 				const struct tdb_logging_context *log_ctx,
   3813 				tdb_hash_func hash_fn)
   3814 {
   3815 	struct tdb_context *tdb;
   3816 	struct stat st;
   3817 	int rev = 0, locked = 0;
   3818 	unsigned char *vp;
   3819 	u32 vertest;
   3820 
   3821 	if (!(tdb = (struct tdb_context *)calloc(1, sizeof *tdb))) {
   3822 		/* Can't log this */
   3823 		errno = ENOMEM;
   3824 		goto fail;
   3825 	}
   3826 	tdb_io_init(tdb);
   3827 	tdb->fd = -1;
   3828 	tdb->name = NULL;
   3829 	tdb->map_ptr = NULL;
   3830 	tdb->flags = tdb_flags;
   3831 	tdb->open_flags = open_flags;
   3832 	if (log_ctx) {
   3833 		tdb->log = *log_ctx;
   3834 	} else {
   3835 		tdb->log.log_fn = null_log_fn;
   3836 		tdb->log.log_private = NULL;
   3837 	}
   3838 	tdb->hash_fn = hash_fn ? hash_fn : default_tdb_hash;
   3839 
   3840 	/* cache the page size */
   3841 	tdb->page_size = getpagesize();
   3842 	if (tdb->page_size <= 0) {
   3843 		tdb->page_size = 0x2000;
   3844 	}
   3845 
   3846 	if ((open_flags & O_ACCMODE) == O_WRONLY) {
   3847 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_open_ex: can't open tdb %s write-only\n",
   3848 			 name));
   3849 		errno = EINVAL;
   3850 		goto fail;
   3851 	}
   3852 
   3853 	if (hash_size == 0)
   3854 		hash_size = DEFAULT_HASH_SIZE;
   3855 	if ((open_flags & O_ACCMODE) == O_RDONLY) {
   3856 		tdb->read_only = 1;
   3857 		/* read only databases don't do locking or clear if first */
   3858 		tdb->flags |= TDB_NOLOCK;
   3859 		tdb->flags &= ~TDB_CLEAR_IF_FIRST;
   3860 	}
   3861 
   3862 	/* internal databases don't mmap or lock, and start off cleared */
   3863 	if (tdb->flags & TDB_INTERNAL) {
   3864 		tdb->flags |= (TDB_NOLOCK | TDB_NOMMAP);
   3865 		tdb->flags &= ~TDB_CLEAR_IF_FIRST;
   3866 		if (tdb_new_database(tdb, hash_size) != 0) {
   3867 			TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_open_ex: tdb_new_database failed!"));
   3868 			goto fail;
   3869 		}
   3870 		goto internal;
   3871 	}
   3872 
   3873 	if ((tdb->fd = open(name, open_flags, mode)) == -1) {
   3874 		TDB_LOG((tdb, TDB_DEBUG_WARNING, "tdb_open_ex: could not open file %s: %s\n",
   3875 			 name, strerror(errno)));
   3876 		goto fail;	/* errno set by open(2) */
   3877 	}
   3878 
   3879 	/* ensure there is only one process initialising at once */
   3880 	if (tdb->methods->tdb_brlock(tdb, GLOBAL_LOCK, F_WRLCK, F_SETLKW, 0, 1) == -1) {
   3881 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_open_ex: failed to get global lock on %s: %s\n",
   3882 			 name, strerror(errno)));
   3883 		goto fail;	/* errno set by tdb_brlock */
   3884 	}
   3885 
   3886 	/* we need to zero database if we are the only one with it open */
   3887 	if ((tdb_flags & TDB_CLEAR_IF_FIRST) &&
   3888 	    (locked = (tdb->methods->tdb_brlock(tdb, ACTIVE_LOCK, F_WRLCK, F_SETLK, 0, 1) == 0))) {
   3889 		open_flags |= O_CREAT;
   3890 		if (ftruncate(tdb->fd, 0) == -1) {
   3891 			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_open_ex: "
   3892 				 "failed to truncate %s: %s\n",
   3893 				 name, strerror(errno)));
   3894 			goto fail; /* errno set by ftruncate */
   3895 		}
   3896 	}
   3897 
   3898 	if (read(tdb->fd, &tdb->header, sizeof(tdb->header)) != sizeof(tdb->header)
   3899 	    || strcmp(tdb->header.magic_food, TDB_MAGIC_FOOD) != 0
   3900 	    || (tdb->header.version != TDB_VERSION
   3901 		&& !(rev = (tdb->header.version==TDB_BYTEREV(TDB_VERSION))))) {
   3902 		/* its not a valid database - possibly initialise it */
   3903 		if (!(open_flags & O_CREAT) || tdb_new_database(tdb, hash_size) == -1) {
   3904 			errno = EIO; /* ie bad format or something */
   3905 			goto fail;
   3906 		}
   3907 		rev = (tdb->flags & TDB_CONVERT);
   3908 	}
   3909 	vp = (unsigned char *)&tdb->header.version;
   3910 	vertest = (((u32)vp[0]) << 24) | (((u32)vp[1]) << 16) |
   3911 		  (((u32)vp[2]) << 8) | (u32)vp[3];
   3912 	tdb->flags |= (vertest==TDB_VERSION) ? TDB_BIGENDIAN : 0;
   3913 	if (!rev)
   3914 		tdb->flags &= ~TDB_CONVERT;
   3915 	else {
   3916 		tdb->flags |= TDB_CONVERT;
   3917 		tdb_convert(&tdb->header, sizeof(tdb->header));
   3918 	}
   3919 	if (fstat(tdb->fd, &st) == -1)
   3920 		goto fail;
   3921 
   3922 	if (tdb->header.rwlocks != 0) {
   3923 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_open_ex: spinlocks no longer supported\n"));
   3924 		goto fail;
   3925 	}
   3926 
   3927 	/* Is it already in the open list?  If so, fail. */
   3928 	if (tdb_already_open(st.st_dev, st.st_ino)) {
   3929 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_open_ex: "
   3930 			 "%s (%d,%d) is already open in this process\n",
   3931 			 name, (int)st.st_dev, (int)st.st_ino));
   3932 		errno = EBUSY;
   3933 		goto fail;
   3934 	}
   3935 
   3936 	if (!(tdb->name = (char *)strdup(name))) {
   3937 		errno = ENOMEM;
   3938 		goto fail;
   3939 	}
   3940 
   3941 	tdb->map_size = st.st_size;
   3942 	tdb->device = st.st_dev;
   3943 	tdb->inode = st.st_ino;
   3944 	tdb->max_dead_records = 0;
   3945 	tdb_mmap(tdb);
   3946 	if (locked) {
   3947 		if (tdb->methods->tdb_brlock(tdb, ACTIVE_LOCK, F_UNLCK, F_SETLK, 0, 1) == -1) {
   3948 			TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_open_ex: "
   3949 				 "failed to take ACTIVE_LOCK on %s: %s\n",
   3950 				 name, strerror(errno)));
   3951 			goto fail;
   3952 		}
   3953 
   3954 	}
   3955 
   3956 	/* We always need to do this if the CLEAR_IF_FIRST flag is set, even if
   3957 	   we didn't get the initial exclusive lock as we need to let all other
   3958 	   users know we're using it. */
   3959 
   3960 	if (tdb_flags & TDB_CLEAR_IF_FIRST) {
   3961 		/* leave this lock in place to indicate it's in use */
   3962 		if (tdb->methods->tdb_brlock(tdb, ACTIVE_LOCK, F_RDLCK, F_SETLKW, 0, 1) == -1)
   3963 			goto fail;
   3964 	}
   3965 
   3966 	/* if needed, run recovery */
   3967 	if (tdb_transaction_recover(tdb) == -1) {
   3968 		goto fail;
   3969 	}
   3970 
   3971  internal:
   3972 	/* Internal (memory-only) databases skip all the code above to
   3973 	 * do with disk files, and resume here by releasing their
   3974 	 * global lock and hooking into the active list. */
   3975 	if (tdb->methods->tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1) == -1)
   3976 		goto fail;
   3977 	tdb->next = tdbs;
   3978 	tdbs = tdb;
   3979 	return tdb;
   3980 
   3981  fail:
   3982 	{ int save_errno = errno;
   3983 
   3984 	if (!tdb)
   3985 		return NULL;
   3986 
   3987 	if (tdb->map_ptr) {
   3988 		if (tdb->flags & TDB_INTERNAL)
   3989 			SAFE_FREE(tdb->map_ptr);
   3990 		else
   3991 			tdb_munmap(tdb);
   3992 	}
   3993 	SAFE_FREE(tdb->name);
   3994 	if (tdb->fd != -1)
   3995 		if (close(tdb->fd) != 0)
   3996 			TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_open_ex: failed to close tdb->fd on error!\n"));
   3997 	SAFE_FREE(tdb);
   3998 	errno = save_errno;
   3999 	return NULL;
   4000 	}
   4001 }
   4002 
   4003 /*
   4004  * Set the maximum number of dead records per hash chain
   4005  */
   4006 
   4007 void tdb_set_max_dead(struct tdb_context *tdb, int max_dead)
   4008 {
   4009 	tdb->max_dead_records = max_dead;
   4010 }
   4011 
   4012 /**
   4013  * Close a database.
   4014  *
   4015  * @returns -1 for error; 0 for success.
   4016  **/
   4017 int tdb_close(struct tdb_context *tdb)
   4018 {
   4019 	struct tdb_context **i;
   4020 	int ret = 0;
   4021 
   4022 	if (tdb->transaction) {
   4023 		tdb_transaction_cancel(tdb);
   4024 	}
   4025 
   4026 	if (tdb->map_ptr) {
   4027 		if (tdb->flags & TDB_INTERNAL)
   4028 			SAFE_FREE(tdb->map_ptr);
   4029 		else
   4030 			tdb_munmap(tdb);
   4031 	}
   4032 	SAFE_FREE(tdb->name);
   4033 	if (tdb->fd != -1)
   4034 		ret = close(tdb->fd);
   4035 	SAFE_FREE(tdb->lockrecs);
   4036 
   4037 	/* Remove from contexts list */
   4038 	for (i = &tdbs; *i; i = &(*i)->next) {
   4039 		if (*i == tdb) {
   4040 			*i = tdb->next;
   4041 			break;
   4042 		}
   4043 	}
   4044 
   4045 	memset(tdb, 0, sizeof(*tdb));
   4046 	SAFE_FREE(tdb);
   4047 
   4048 	return ret;
   4049 }
   4050 
   4051 /* register a loging function */
   4052 void tdb_set_logging_function(struct tdb_context *tdb,
   4053                               const struct tdb_logging_context *log_ctx)
   4054 {
   4055         tdb->log = *log_ctx;
   4056 }
   4057 
   4058 void *tdb_get_logging_private(struct tdb_context *tdb)
   4059 {
   4060 	return tdb->log.log_private;
   4061 }
   4062 
   4063 /* reopen a tdb - this can be used after a fork to ensure that we have an independent
   4064    seek pointer from our parent and to re-establish locks */
   4065 int tdb_reopen(struct tdb_context *tdb)
   4066 {
   4067 	struct stat st;
   4068 
   4069 	if (tdb->flags & TDB_INTERNAL) {
   4070 		return 0; /* Nothing to do. */
   4071 	}
   4072 
   4073 	if (tdb->num_locks != 0 || tdb->global_lock.count) {
   4074 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_reopen: reopen not allowed with locks held\n"));
   4075 		goto fail;
   4076 	}
   4077 
   4078 	if (tdb->transaction != 0) {
   4079 		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_reopen: reopen not allowed inside a transaction\n"));
   4080 		goto fail;
   4081 	}
   4082 
   4083 	if (tdb_munmap(tdb) != 0) {
   4084 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_reopen: munmap failed (%s)\n", strerror(errno)));
   4085 		goto fail;
   4086 	}
   4087 	if (close(tdb->fd) != 0)
   4088 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_reopen: WARNING closing tdb->fd failed!\n"));
   4089 	tdb->fd = open(tdb->name, tdb->open_flags & ~(O_CREAT|O_TRUNC), 0);
   4090 	if (tdb->fd == -1) {
   4091 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_reopen: open failed (%s)\n", strerror(errno)));
   4092 		goto fail;
   4093 	}
   4094 	if ((tdb->flags & TDB_CLEAR_IF_FIRST) &&
   4095 	    (tdb->methods->tdb_brlock(tdb, ACTIVE_LOCK, F_RDLCK, F_SETLKW, 0, 1) == -1)) {
   4096 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_reopen: failed to obtain active lock\n"));
   4097 		goto fail;
   4098 	}
   4099 	if (fstat(tdb->fd, &st) != 0) {
   4100 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_reopen: fstat failed (%s)\n", strerror(errno)));
   4101 		goto fail;
   4102 	}
   4103 	if (st.st_ino != tdb->inode || st.st_dev != tdb->device) {
   4104 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_reopen: file dev/inode has changed!\n"));
   4105 		goto fail;
   4106 	}
   4107 	tdb_mmap(tdb);
   4108 
   4109 	return 0;
   4110 
   4111 fail:
   4112 	tdb_close(tdb);
   4113 	return -1;
   4114 }
   4115 
   4116 /* reopen all tdb's */
   4117 int tdb_reopen_all(int parent_longlived)
   4118 {
   4119 	struct tdb_context *tdb;
   4120 
   4121 	for (tdb=tdbs; tdb; tdb = tdb->next) {
   4122 		/*
   4123 		 * If the parent is longlived (ie. a
   4124 		 * parent daemon architecture), we know
   4125 		 * it will keep it's active lock on a
   4126 		 * tdb opened with CLEAR_IF_FIRST. Thus
   4127 		 * for child processes we don't have to
   4128 		 * add an active lock. This is essential
   4129 		 * to improve performance on systems that
   4130 		 * keep POSIX locks as a non-scalable data
   4131 		 * structure in the kernel.
   4132 		 */
   4133 		if (parent_longlived) {
   4134 			/* Ensure no clear-if-first. */
   4135 			tdb->flags &= ~TDB_CLEAR_IF_FIRST;
   4136 		}
   4137 
   4138 		if (tdb_reopen(tdb) != 0)
   4139 			return -1;
   4140 	}
   4141 
   4142 	return 0;
   4143 }
   4144