r16945: Sync trunk -> 3.0 for 3.0.24 code. Still need

to do the upper layer directories but this is what
everyone is waiting for....

Jeremy.

16 files changed, 3553 insertions, 2578 deletions

diff --git a/source/tdb/dump.c b/source/tdb/dump.c
new file mode 100644
index 00000000000..577f23aac68
--- /dev/null
+++ b/source/tdb/dump.c
@@ -0,0 +1,135 @@
+ /* 
+   Unix SMB/CIFS implementation.
+
+   trivial database library
+
+   Copyright (C) Andrew Tridgell              1999-2005
+   Copyright (C) Paul `Rusty' Russell		   2000
+   Copyright (C) Jeremy Allison			   2000-2003
+   
+     ** NOTE! The following LGPL license applies to the tdb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+   
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "tdb_private.h"
+
+static tdb_off_t tdb_dump_record(struct tdb_context *tdb, tdb_off_t offset)
+{
+	struct list_struct rec;
+	tdb_off_t tailer_ofs, tailer;
+
+	if (tdb->methods->tdb_read(tdb, offset, (char *)&rec, 
+				   sizeof(rec), DOCONV()) == -1) {
+		printf("ERROR: failed to read record at %u\n", offset);
+		return 0;
+	}
+
+	printf(" rec: offset=0x%08x next=0x%08x rec_len=%d key_len=%d data_len=%d full_hash=0x%x magic=0x%x\n",
+	       offset, rec.next, rec.rec_len, rec.key_len, rec.data_len, rec.full_hash, rec.magic);
+
+	tailer_ofs = offset + sizeof(rec) + rec.rec_len - sizeof(tdb_off_t);
+
+	if (tdb_ofs_read(tdb, tailer_ofs, &tailer) == -1) {
+		printf("ERROR: failed to read tailer at %u\n", tailer_ofs);
+		return rec.next;
+	}
+
+	if (tailer != rec.rec_len + sizeof(rec)) {
+		printf("ERROR: tailer does not match record! tailer=%u totalsize=%u\n",
+				(unsigned int)tailer, (unsigned int)(rec.rec_len + sizeof(rec)));
+	}
+	return rec.next;
+}
+
+static int tdb_dump_chain(struct tdb_context *tdb, int i)
+{
+	tdb_off_t rec_ptr, top;
+
+	top = TDB_HASH_TOP(i);
+
+	if (tdb_lock(tdb, i, F_WRLCK) != 0)
+		return -1;
+
+	if (tdb_ofs_read(tdb, top, &rec_ptr) == -1)
+		return tdb_unlock(tdb, i, F_WRLCK);
+
+	if (rec_ptr)
+		printf("hash=%d\n", i);
+
+	while (rec_ptr) {
+		rec_ptr = tdb_dump_record(tdb, rec_ptr);
+	}
+
+	return tdb_unlock(tdb, i, F_WRLCK);
+}
+
+void tdb_dump_all(struct tdb_context *tdb)
+{
+	int i;
+	for (i=0;i<tdb->header.hash_size;i++) {
+		tdb_dump_chain(tdb, i);
+	}
+	printf("freelist:\n");
+	tdb_dump_chain(tdb, -1);
+}
+
+int tdb_printfreelist(struct tdb_context *tdb)
+{
+	int ret;
+	long total_free = 0;
+	tdb_off_t offset, rec_ptr;
+	struct list_struct rec;
+
+	if ((ret = tdb_lock(tdb, -1, F_WRLCK)) != 0)
+		return ret;
+
+	offset = FREELIST_TOP;
+
+	/* read in the freelist top */
+	if (tdb_ofs_read(tdb, offset, &rec_ptr) == -1) {
+		tdb_unlock(tdb, -1, F_WRLCK);
+		return 0;
+	}
+
+	printf("freelist top=[0x%08x]\n", rec_ptr );
+	while (rec_ptr) {
+		if (tdb->methods->tdb_read(tdb, rec_ptr, (char *)&rec, 
+					   sizeof(rec), DOCONV()) == -1) {
+			tdb_unlock(tdb, -1, F_WRLCK);
+			return -1;
+		}
+
+		if (rec.magic != TDB_FREE_MAGIC) {
+			printf("bad magic 0x%08x in free list\n", rec.magic);
+			tdb_unlock(tdb, -1, F_WRLCK);
+			return -1;
+		}
+
+		printf("entry offset=[0x%08x], rec.rec_len = [0x%08x (%d)] (end = 0x%08x)\n", 
+		       rec_ptr, rec.rec_len, rec.rec_len, rec_ptr + rec.rec_len);
+		total_free += rec.rec_len;
+
+		/* move to the next record */
+		rec_ptr = rec.next;
+	}
+	printf("total rec_len = [0x%08x (%d)]\n", (int)total_free, 
+               (int)total_free);
+
+	return tdb_unlock(tdb, -1, F_WRLCK);
+}
+
diff --git a/source/tdb/error.c b/source/tdb/error.c
new file mode 100644
index 00000000000..4cf33a29ab2
--- /dev/null
+++ b/source/tdb/error.c
@@ -0,0 +1,58 @@
+ /* 
+   Unix SMB/CIFS implementation.
+
+   trivial database library
+
+   Copyright (C) Andrew Tridgell              1999-2005
+   Copyright (C) Paul `Rusty' Russell		   2000
+   Copyright (C) Jeremy Allison			   2000-2003
+   
+     ** NOTE! The following LGPL license applies to the tdb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+   
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "tdb_private.h"
+
+enum TDB_ERROR tdb_error(struct tdb_context *tdb)
+{
+	return tdb->ecode;
+}
+
+static struct tdb_errname {
+	enum TDB_ERROR ecode; const char *estring;
+} emap[] = { {TDB_SUCCESS, "Success"},
+	     {TDB_ERR_CORRUPT, "Corrupt database"},
+	     {TDB_ERR_IO, "IO Error"},
+	     {TDB_ERR_LOCK, "Locking error"},
+	     {TDB_ERR_OOM, "Out of memory"},
+	     {TDB_ERR_EXISTS, "Record exists"},
+	     {TDB_ERR_NOLOCK, "Lock exists on other keys"},
+	     {TDB_ERR_EINVAL, "Invalid parameter"},
+	     {TDB_ERR_NOEXIST, "Record does not exist"},
+	     {TDB_ERR_RDONLY, "write not permitted"} };
+
+/* Error string for the last tdb error */
+const char *tdb_errorstr(struct tdb_context *tdb)
+{
+	u32 i;
+	for (i = 0; i < sizeof(emap) / sizeof(struct tdb_errname); i++)
+		if (tdb->ecode == emap[i].ecode)
+			return emap[i].estring;
+	return "Invalid error code";
+}
+
diff --git a/source/tdb/freelist.c b/source/tdb/freelist.c
new file mode 100644
index 00000000000..34837511647
--- /dev/null
+++ b/source/tdb/freelist.c
@@ -0,0 +1,331 @@
+ /* 
+   Unix SMB/CIFS implementation.
+
+   trivial database library
+
+   Copyright (C) Andrew Tridgell              1999-2005
+   Copyright (C) Paul `Rusty' Russell		   2000
+   Copyright (C) Jeremy Allison			   2000-2003
+   
+     ** NOTE! The following LGPL license applies to the tdb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+   
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "tdb_private.h"
+
+/* read a freelist record and check for simple errors */
+static int rec_free_read(struct tdb_context *tdb, tdb_off_t off, struct list_struct *rec)
+{
+	if (tdb->methods->tdb_read(tdb, off, rec, sizeof(*rec),DOCONV()) == -1)
+		return -1;
+
+	if (rec->magic == TDB_MAGIC) {
+		/* this happens when a app is showdown while deleting a record - we should
+		   not completely fail when this happens */
+		TDB_LOG((tdb, 0,"rec_free_read non-free magic 0x%x at offset=%d - fixing\n", 
+			 rec->magic, off));
+		rec->magic = TDB_FREE_MAGIC;
+		if (tdb->methods->tdb_write(tdb, off, rec, sizeof(*rec)) == -1)
+			return -1;
+	}
+
+	if (rec->magic != TDB_FREE_MAGIC) {
+		/* Ensure ecode is set for log fn. */
+		tdb->ecode = TDB_ERR_CORRUPT;
+		TDB_LOG((tdb, 0,"rec_free_read bad magic 0x%x at offset=%d\n", 
+			   rec->magic, off));
+		return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
+	}
+	if (tdb->methods->tdb_oob(tdb, rec->next+sizeof(*rec), 0) != 0)
+		return -1;
+	return 0;
+}
+
+
+
+/* Remove an element from the freelist.  Must have alloc lock. */
+static int remove_from_freelist(struct tdb_context *tdb, tdb_off_t off, tdb_off_t next)
+{
+	tdb_off_t last_ptr, i;
+
+	/* read in the freelist top */
+	last_ptr = FREELIST_TOP;
+	while (tdb_ofs_read(tdb, last_ptr, &i) != -1 && i != 0) {
+		if (i == off) {
+			/* We've found it! */
+			return tdb_ofs_write(tdb, last_ptr, &next);
+		}
+		/* Follow chain (next offset is at start of record) */
+		last_ptr = i;
+	}
+	TDB_LOG((tdb, 0,"remove_from_freelist: not on list at off=%d\n", off));
+	return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
+}
+
+
+/* update a record tailer (must hold allocation lock) */
+static int update_tailer(struct tdb_context *tdb, tdb_off_t offset,
+			 const struct list_struct *rec)
+{
+	tdb_off_t totalsize;
+
+	/* Offset of tailer from record header */
+	totalsize = sizeof(*rec) + rec->rec_len;
+	return tdb_ofs_write(tdb, offset + totalsize - sizeof(tdb_off_t),
+			 &totalsize);
+}
+
+/* Add an element into the freelist. Merge adjacent records if
+   neccessary. */
+int tdb_free(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec)
+{
+	tdb_off_t right, left;
+
+	/* Allocation and tailer lock */
+	if (tdb_lock(tdb, -1, F_WRLCK) != 0)
+		return -1;
+
+	/* set an initial tailer, so if we fail we don't leave a bogus record */
+	if (update_tailer(tdb, offset, rec) != 0) {
+		TDB_LOG((tdb, 0, "tdb_free: upfate_tailer failed!\n"));
+		goto fail;
+	}
+
+	/* Look right first (I'm an Australian, dammit) */
+	right = offset + sizeof(*rec) + rec->rec_len;
+	if (right + sizeof(*rec) <= tdb->map_size) {
+		struct list_struct r;
+
+		if (tdb->methods->tdb_read(tdb, right, &r, sizeof(r), DOCONV()) == -1) {
+			TDB_LOG((tdb, 0, "tdb_free: right read failed at %u\n", right));
+			goto left;
+		}
+
+		/* If it's free, expand to include it. */
+		if (r.magic == TDB_FREE_MAGIC) {
+			if (remove_from_freelist(tdb, right, r.next) == -1) {
+				TDB_LOG((tdb, 0, "tdb_free: right free failed at %u\n", right));
+				goto left;
+			}
+			rec->rec_len += sizeof(r) + r.rec_len;
+		}
+	}
+
+left:
+	/* Look left */
+	left = offset - sizeof(tdb_off_t);
+	if (left > TDB_DATA_START(tdb->header.hash_size)) {
+		struct list_struct l;
+		tdb_off_t leftsize;
+		
+		/* Read in tailer and jump back to header */
+		if (tdb_ofs_read(tdb, left, &leftsize) == -1) {
+			TDB_LOG((tdb, 0, "tdb_free: left offset read failed at %u\n", left));
+			goto update;
+		}
+
+		/* it could be uninitialised data */
+		if (leftsize == 0 || leftsize == TDB_PAD_U32) {
+			goto update;
+		}
+
+		left = offset - leftsize;
+
+		/* Now read in record */
+		if (tdb->methods->tdb_read(tdb, left, &l, sizeof(l), DOCONV()) == -1) {
+			TDB_LOG((tdb, 0, "tdb_free: left read failed at %u (%u)\n", left, leftsize));
+			goto update;
+		}
+
+		/* If it's free, expand to include it. */
+		if (l.magic == TDB_FREE_MAGIC) {
+			if (remove_from_freelist(tdb, left, l.next) == -1) {
+				TDB_LOG((tdb, 0, "tdb_free: left free failed at %u\n", left));
+				goto update;
+			} else {
+				offset = left;
+				rec->rec_len += leftsize;
+			}
+		}
+	}
+
+update:
+	if (update_tailer(tdb, offset, rec) == -1) {
+		TDB_LOG((tdb, 0, "tdb_free: update_tailer failed at %u\n", offset));
+		goto fail;
+	}
+
+	/* Now, prepend to free list */
+	rec->magic = TDB_FREE_MAGIC;
+
+	if (tdb_ofs_read(tdb, FREELIST_TOP, &rec->next) == -1 ||
+	    tdb_rec_write(tdb, offset, rec) == -1 ||
+	    tdb_ofs_write(tdb, FREELIST_TOP, &offset) == -1) {
+		TDB_LOG((tdb, 0, "tdb_free record write failed at offset=%d\n", offset));
+		goto fail;
+	}
+
+	/* And we're done. */
+	tdb_unlock(tdb, -1, F_WRLCK);
+	return 0;
+
+ fail:
+	tdb_unlock(tdb, -1, F_WRLCK);
+	return -1;
+}
+
+
+/* 
+   the core of tdb_allocate - called when we have decided which
+   free list entry to use
+ */
+static tdb_off_t tdb_allocate_ofs(struct tdb_context *tdb, tdb_len_t length, tdb_off_t rec_ptr,
+				struct list_struct *rec, tdb_off_t last_ptr)
+{
+	struct list_struct newrec;
+	tdb_off_t newrec_ptr;
+
+	memset(&newrec, '\0', sizeof(newrec));
+
+	/* found it - now possibly split it up  */
+	if (rec->rec_len > length + MIN_REC_SIZE) {
+		/* Length of left piece */
+		length = TDB_ALIGN(length, TDB_ALIGNMENT);
+		
+		/* Right piece to go on free list */
+		newrec.rec_len = rec->rec_len - (sizeof(*rec) + length);
+		newrec_ptr = rec_ptr + sizeof(*rec) + length;
+		
+		/* And left record is shortened */
+		rec->rec_len = length;
+	} else {
+		newrec_ptr = 0;
+	}
+	
+	/* Remove allocated record from the free list */
+	if (tdb_ofs_write(tdb, last_ptr, &rec->next) == -1) {
+		return 0;
+	}
+	
+	/* Update header: do this before we drop alloc
+	   lock, otherwise tdb_free() might try to
+	   merge with us, thinking we're free.
+	   (Thanks Jeremy Allison). */
+	rec->magic = TDB_MAGIC;
+	if (tdb_rec_write(tdb, rec_ptr, rec) == -1) {
+		return 0;
+	}
+	
+	/* Did we create new block? */
+	if (newrec_ptr) {
+		/* Update allocated record tailer (we
+		   shortened it). */
+		if (update_tailer(tdb, rec_ptr, rec) == -1) {
+			return 0;
+		}
+		
+		/* Free new record */
+		if (tdb_free(tdb, newrec_ptr, &newrec) == -1) {
+			return 0;
+		}
+	}
+	
+	/* all done - return the new record offset */
+	return rec_ptr;
+}
+
+/* allocate some space from the free list. The offset returned points
+   to a unconnected list_struct within the database with room for at
+   least length bytes of total data
+
+   0 is returned if the space could not be allocated
+ */
+tdb_off_t tdb_allocate(struct tdb_context *tdb, tdb_len_t length, struct list_struct *rec)
+{
+	tdb_off_t rec_ptr, last_ptr, newrec_ptr;
+	struct {
+		tdb_off_t rec_ptr, last_ptr;
+		tdb_len_t rec_len;
+	} bestfit;
+
+	if (tdb_lock(tdb, -1, F_WRLCK) == -1)
+		return 0;
+
+	/* Extra bytes required for tailer */
+	length += sizeof(tdb_off_t);
+
+ again:
+	last_ptr = FREELIST_TOP;
+
+	/* read in the freelist top */
+	if (tdb_ofs_read(tdb, FREELIST_TOP, &rec_ptr) == -1)
+		goto fail;
+
+	bestfit.rec_ptr = 0;
+	bestfit.last_ptr = 0;
+	bestfit.rec_len = 0;
+
+	/* 
+	   this is a best fit allocation strategy. Originally we used
+	   a first fit strategy, but it suffered from massive fragmentation
+	   issues when faced with a slowly increasing record size.
+	 */
+	while (rec_ptr) {
+		if (rec_free_read(tdb, rec_ptr, rec) == -1) {
+			goto fail;
+		}
+
+		if (rec->rec_len >= length) {
+			if (bestfit.rec_ptr == 0 ||
+			    rec->rec_len < bestfit.rec_len) {
+				bestfit.rec_len = rec->rec_len;
+				bestfit.rec_ptr = rec_ptr;
+				bestfit.last_ptr = last_ptr;
+				/* consider a fit to be good enough if
+				   we aren't wasting more than half
+				   the space */
+				if (bestfit.rec_len < 2*length) {
+					break;
+				}
+			}
+		}
+
+		/* move to the next record */
+		last_ptr = rec_ptr;
+		rec_ptr = rec->next;
+	}
+
+	if (bestfit.rec_ptr != 0) {
+		if (rec_free_read(tdb, bestfit.rec_ptr, rec) == -1) {
+			goto fail;
+		}
+
+		newrec_ptr = tdb_allocate_ofs(tdb, length, bestfit.rec_ptr, rec, bestfit.last_ptr);
+		tdb_unlock(tdb, -1, F_WRLCK);
+		return newrec_ptr;
+	}
+
+	/* we didn't find enough space. See if we can expand the
+	   database and if we can then try again */
+	if (tdb_expand(tdb, length + sizeof(*rec)) == 0)
+		goto again;
+ fail:
+	tdb_unlock(tdb, -1, F_WRLCK);
+	return 0;
+}
+
diff --git a/source/tdb/io.c b/source/tdb/io.c
new file mode 100644
index 00000000000..ac6f4cdf573
--- /dev/null
+++ b/source/tdb/io.c
@@ -0,0 +1,412 @@
+ /* 
+   Unix SMB/CIFS implementation.
+
+   trivial database library
+
+   Copyright (C) Andrew Tridgell              1999-2005
+   Copyright (C) Paul `Rusty' Russell		   2000
+   Copyright (C) Jeremy Allison			   2000-2003
+   
+     ** NOTE! The following LGPL license applies to the tdb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+   
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+
+#include "tdb_private.h"
+
+#ifndef HAVE_PREAD
+ static ssize_t pread(int fd, void *buf, size_t count, off_t offset)
+{
+	if (lseek(fd, offset, SEEK_SET) != offset) {
+		errno = EIO;
+		return -1;
+	}
+	return read(fd, buf, count);
+}
+#endif
+
+#ifndef HAVE_PWRITE
+ static ssize_t pwrite(int fd, const void *buf, size_t count, off_t offset)
+{
+	if (lseek(fd, offset, SEEK_SET) != offset) {
+		errno = EIO;
+		return -1;
+	}
+	return write(fd, buf, count);
+}
+#endif
+
+/* check for an out of bounds access - if it is out of bounds then
+   see if the database has been expanded by someone else and expand
+   if necessary 
+   note that "len" is the minimum length needed for the db
+*/
+static int tdb_oob(struct tdb_context *tdb, tdb_off_t len, int probe)
+{
+	struct stat st;
+	if (len <= tdb->map_size)
+		return 0;
+	if (tdb->flags & TDB_INTERNAL) {
+		if (!probe) {
+			/* Ensure ecode is set for log fn. */
+			tdb->ecode = TDB_ERR_IO;
+			TDB_LOG((tdb, 0,"tdb_oob len %d beyond internal malloc size %d\n",
+				 (int)len, (int)tdb->map_size));
+		}
+		return TDB_ERRCODE(TDB_ERR_IO, -1);
+	}
+
+	if (fstat(tdb->fd, &st) == -1) {
+		return TDB_ERRCODE(TDB_ERR_IO, -1);
+	}
+
+	if (st.st_size < (size_t)len) {
+		if (!probe) {
+			/* Ensure ecode is set for log fn. */
+			tdb->ecode = TDB_ERR_IO;
+			TDB_LOG((tdb, 0,"tdb_oob len %d beyond eof at %d\n",
+				 (int)len, (int)st.st_size));
+		}
+		return TDB_ERRCODE(TDB_ERR_IO, -1);
+	}
+
+	/* Unmap, update size, remap */
+	if (tdb_munmap(tdb) == -1)
+		return TDB_ERRCODE(TDB_ERR_IO, -1);
+	tdb->map_size = st.st_size;
+	tdb_mmap(tdb);
+	return 0;
+}
+
+/* write a lump of data at a specified offset */
+static int tdb_write(struct tdb_context *tdb, tdb_off_t off, 
+		     const void *buf, tdb_len_t len)
+{
+	if (len == 0) {
+		return 0;
+	}
+
+	if (tdb->read_only || tdb->traverse_read) {
+		tdb->ecode = TDB_ERR_RDONLY;
+		return -1;
+	}
+
+	if (tdb->methods->tdb_oob(tdb, off + len, 0) != 0)
+		return -1;
+
+	if (tdb->map_ptr) {
+		memcpy(off + (char *)tdb->map_ptr, buf, len);
+	} else if (pwrite(tdb->fd, buf, len, off) != (ssize_t)len) {
+		/* Ensure ecode is set for log fn. */
+		tdb->ecode = TDB_ERR_IO;
+		TDB_LOG((tdb, 0,"tdb_write failed at %d len=%d (%s)\n",
+			   off, len, strerror(errno)));
+		return TDB_ERRCODE(TDB_ERR_IO, -1);
+	}
+	return 0;
+}
+
+/* Endian conversion: we only ever deal with 4 byte quantities */
+void *tdb_convert(void *buf, u32 size)
+{
+	u32 i, *p = buf;
+	for (i = 0; i < size / 4; i++)
+		p[i] = TDB_BYTEREV(p[i]);
+	return buf;
+}
+
+
+/* read a lump of data at a specified offset, maybe convert */
+static int tdb_read(struct tdb_context *tdb, tdb_off_t off, void *buf, 
+		    tdb_len_t len, int cv)
+{
+	if (tdb->methods->tdb_oob(tdb, off + len, 0) != 0) {
+		return -1;
+	}
+
+	if (tdb->map_ptr) {
+		memcpy(buf, off + (char *)tdb->map_ptr, len);
+	} else {
+		ssize_t ret = pread(tdb->fd, buf, len, off);
+		if (ret != (ssize_t)len) {
+			/* Ensure ecode is set for log fn. */
+			tdb->ecode = TDB_ERR_IO;
+			TDB_LOG((tdb, 0,"tdb_read failed at %d len=%d ret=%d (%s) map_size=%d\n",
+				 off, len, (int)ret, strerror(errno), (int)tdb->map_size));
+			return TDB_ERRCODE(TDB_ERR_IO, -1);
+		}
+	}
+	if (cv) {
+		tdb_convert(buf, len);
+	}
+	return 0;
+}
+
+
+
+/*
+  do an unlocked scan of the hash table heads to find the next non-zero head. The value
+  will then be confirmed with the lock held
+*/		
+static void tdb_next_hash_chain(struct tdb_context *tdb, u32 *chain)
+{
+	u32 h = *chain;
+	if (tdb->map_ptr) {
+		for (;h < tdb->header.hash_size;h++) {
+			if (0 != *(u32 *)(TDB_HASH_TOP(h) + (unsigned char *)tdb->map_ptr)) {
+				break;
+			}
+		}
+	} else {
+		u32 off=0;
+		for (;h < tdb->header.hash_size;h++) {
+			if (tdb_ofs_read(tdb, TDB_HASH_TOP(h), &off) != 0 || off != 0) {
+				break;
+			}
+		}
+	}
+	(*chain) = h;
+}
+
+
+int tdb_munmap(struct tdb_context *tdb)
+{
+	if (tdb->flags & TDB_INTERNAL)
+		return 0;
+
+#ifdef HAVE_MMAP
+	if (tdb->map_ptr) {
+		int ret = munmap(tdb->map_ptr, tdb->map_size);
+		if (ret != 0)
+			return ret;
+	}
+#endif
+	tdb->map_ptr = NULL;
+	return 0;
+}
+
+void tdb_mmap(struct tdb_context *tdb)
+{
+	if (tdb->flags & TDB_INTERNAL)
+		return;
+
+#ifdef HAVE_MMAP
+	if (!(tdb->flags & TDB_NOMMAP)) {
+		tdb->map_ptr = mmap(NULL, tdb->map_size, 
+				    PROT_READ|(tdb->read_only? 0:PROT_WRITE), 
+				    MAP_SHARED|MAP_FILE, tdb->fd, 0);
+
+		/*
+		 * NB. When mmap fails it returns MAP_FAILED *NOT* NULL !!!!
+		 */
+
+		if (tdb->map_ptr == MAP_FAILED) {
+			tdb->map_ptr = NULL;
+			TDB_LOG((tdb, 2, "tdb_mmap failed for size %d (%s)\n", 
+				 tdb->map_size, strerror(errno)));
+		}
+	} else {
+		tdb->map_ptr = NULL;
+	}
+#else
+	tdb->map_ptr = NULL;
+#endif
+}
+
+/* expand a file.  we prefer to use ftruncate, as that is what posix
+  says to use for mmap expansion */
+static int tdb_expand_file(struct tdb_context *tdb, tdb_off_t size, tdb_off_t addition)
+{
+	char buf[1024];
+
+	if (tdb->read_only || tdb->traverse_read) {
+		tdb->ecode = TDB_ERR_RDONLY;
+		return -1;
+	}
+
+	if (ftruncate(tdb->fd, size+addition) == -1) {
+		char b = 0;
+		if (pwrite(tdb->fd,  &b, 1, (size+addition) - 1) != 1) {
+			TDB_LOG((tdb, 0, "expand_file to %d failed (%s)\n", 
+				 size+addition, strerror(errno)));
+			return -1;
+		}
+	}
+
+	/* now fill the file with something. This ensures that the
+	   file isn't sparse, which would be very bad if we ran out of
+	   disk. This must be done with write, not via mmap */
+	memset(buf, TDB_PAD_BYTE, sizeof(buf));
+	while (addition) {
+		int n = addition>sizeof(buf)?sizeof(buf):addition;
+		int ret = pwrite(tdb->fd, buf, n, size);
+		if (ret != n) {
+			TDB_LOG((tdb, 0, "expand_file write of %d failed (%s)\n", 
+				   n, strerror(errno)));
+			return -1;
+		}
+		addition -= n;
+		size += n;
+	}
+	return 0;
+}
+
+
+/* expand the database at least size bytes by expanding the underlying
+   file and doing the mmap again if necessary */
+int tdb_expand(struct tdb_context *tdb, tdb_off_t size)
+{
+	struct list_struct rec;
+	tdb_off_t offset;
+
+	if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
+		TDB_LOG((tdb, 0, "lock failed in tdb_expand\n"));
+		return -1;
+	}
+
+	/* must know about any previous expansions by another process */
+	tdb->methods->tdb_oob(tdb, tdb->map_size + 1, 1);
+
+	/* always make room for at least 10 more records, and round
+           the database up to a multiple of the page size */
+	size = TDB_ALIGN(tdb->map_size + size*10, tdb->page_size) - tdb->map_size;
+
+	if (!(tdb->flags & TDB_INTERNAL))
+		tdb_munmap(tdb);
+
+	/*
+	 * We must ensure the file is unmapped before doing this
+	 * to ensure consistency with systems like OpenBSD where
+	 * writes and mmaps are not consistent.
+	 */
+
+	/* expand the file itself */
+	if (!(tdb->flags & TDB_INTERNAL)) {
+		if (tdb->methods->tdb_expand_file(tdb, tdb->map_size, size) != 0)
+			goto fail;
+	}
+
+	tdb->map_size += size;
+
+	if (tdb->flags & TDB_INTERNAL) {
+		char *new_map_ptr = realloc(tdb->map_ptr, tdb->map_size);
+		if (!new_map_ptr) {
+			tdb->map_size -= size;
+			goto fail;
+		}
+		tdb->map_ptr = new_map_ptr;
+	} else {
+		/*
+		 * We must ensure the file is remapped before adding the space
+		 * to ensure consistency with systems like OpenBSD where
+		 * writes and mmaps are not consistent.
+		 */
+
+		/* We're ok if the mmap fails as we'll fallback to read/write */
+		tdb_mmap(tdb);
+	}
+
+	/* form a new freelist record */
+	memset(&rec,'\0',sizeof(rec));
+	rec.rec_len = size - sizeof(rec);
+
+	/* link it into the free list */
+	offset = tdb->map_size - size;
+	if (tdb_free(tdb, offset, &rec) == -1)
+		goto fail;
+
+	tdb_unlock(tdb, -1, F_WRLCK);
+	return 0;
+ fail:
+	tdb_unlock(tdb, -1, F_WRLCK);
+	return -1;
+}
+
+/* read/write a tdb_off_t */
+int tdb_ofs_read(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d)
+{
+	return tdb->methods->tdb_read(tdb, offset, (char*)d, sizeof(*d), DOCONV());
+}
+
+int tdb_ofs_write(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d)
+{
+	tdb_off_t off = *d;
+	return tdb->methods->tdb_write(tdb, offset, CONVERT(off), sizeof(*d));
+}
+
+
+/* read a lump of data, allocating the space for it */
+char *tdb_alloc_read(struct tdb_context *tdb, tdb_off_t offset, tdb_len_t len)
+{
+	char *buf;
+
+	/* some systems don't like zero length malloc */
+	if (len == 0) {
+		len = 1;
+	}
+
+	if (!(buf = malloc(len))) {
+		/* Ensure ecode is set for log fn. */
+		tdb->ecode = TDB_ERR_OOM;
+		TDB_LOG((tdb, 0,"tdb_alloc_read malloc failed len=%d (%s)\n",
+			   len, strerror(errno)));
+		return TDB_ERRCODE(TDB_ERR_OOM, buf);
+	}
+	if (tdb->methods->tdb_read(tdb, offset, buf, len, 0) == -1) {
+		SAFE_FREE(buf);
+		return NULL;
+	}
+	return buf;
+}
+
+/* read/write a record */
+int tdb_rec_read(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec)
+{
+	if (tdb->methods->tdb_read(tdb, offset, rec, sizeof(*rec),DOCONV()) == -1)
+		return -1;
+	if (TDB_BAD_MAGIC(rec)) {
+		/* Ensure ecode is set for log fn. */
+		tdb->ecode = TDB_ERR_CORRUPT;
+		TDB_LOG((tdb, 0,"tdb_rec_read bad magic 0x%x at offset=%d\n", rec->magic, offset));
+		return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
+	}
+	return tdb->methods->tdb_oob(tdb, rec->next+sizeof(*rec), 0);
+}
+
+int tdb_rec_write(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec)
+{
+	struct list_struct r = *rec;
+	return tdb->methods->tdb_write(tdb, offset, CONVERT(r), sizeof(r));
+}
+
+static const struct tdb_methods io_methods = {
+	tdb_read,
+	tdb_write,
+	tdb_next_hash_chain,
+	tdb_oob,
+	tdb_expand_file,
+	tdb_brlock
+};
+
+/*
+  initialise the default methods table
+*/
+void tdb_io_init(struct tdb_context *tdb)
+{
+	tdb->methods = &io_methods;
+}
diff --git a/source/tdb/lock.c b/source/tdb/lock.c
new file mode 100644
index 00000000000..a96d77a229a
--- /dev/null
+++ b/source/tdb/lock.c
@@ -0,0 +1,276 @@
+ /* 
+   Unix SMB/CIFS implementation.
+
+   trivial database library
+
+   Copyright (C) Andrew Tridgell              1999-2005
+   Copyright (C) Paul `Rusty' Russell		   2000
+   Copyright (C) Jeremy Allison			   2000-2003
+   
+     ** NOTE! The following LGPL license applies to the tdb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+   
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "tdb_private.h"
+
+/* a byte range locking function - return 0 on success
+   this functions locks/unlocks 1 byte at the specified offset.
+
+   On error, errno is also set so that errors are passed back properly
+   through tdb_open(). 
+
+   note that a len of zero means lock to end of file
+*/
+int tdb_brlock_len(struct tdb_context *tdb, tdb_off_t offset, 
+		   int rw_type, int lck_type, int probe, size_t len)
+{
+	struct flock fl;
+	int ret;
+
+	if (tdb->flags & TDB_NOLOCK) {
+		return 0;
+	}
+
+	if ((rw_type == F_WRLCK) && (tdb->read_only || tdb->traverse_read)) {
+		tdb->ecode = TDB_ERR_RDONLY;
+		return -1;
+	}
+
+	fl.l_type = rw_type;
+	fl.l_whence = SEEK_SET;
+	fl.l_start = offset;
+	fl.l_len = len;
+	fl.l_pid = 0;
+
+	do {
+		ret = fcntl(tdb->fd,lck_type,&fl);
+	} while (ret == -1 && errno == EINTR);
+
+	if (ret == -1) {
+		/* Generic lock error. errno set by fcntl.
+		 * EAGAIN is an expected return from non-blocking
+		 * locks. */
+		if (!probe && lck_type != F_SETLK) {
+			/* Ensure error code is set for log fun to examine. */
+			tdb->ecode = TDB_ERR_LOCK;
+			TDB_LOG((tdb, 5,"tdb_brlock failed (fd=%d) at offset %d rw_type=%d lck_type=%d len=%d\n", 
+				 tdb->fd, offset, rw_type, lck_type, (int)len));
+		}
+		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
+	}
+	return 0;
+}
+
+
+/*
+  upgrade a read lock to a write lock. This needs to be handled in a
+  special way as some OSes (such as solaris) have too conservative
+  deadlock detection and claim a deadlock when progress can be
+  made. For those OSes we may loop for a while.  
+*/
+int tdb_brlock_upgrade(struct tdb_context *tdb, tdb_off_t offset, size_t len)
+{
+	int count = 1000;
+	while (count--) {
+		struct timeval tv;
+		if (tdb_brlock_len(tdb, offset, F_WRLCK, F_SETLKW, 1, len) == 0) {
+			return 0;
+		}
+		if (errno != EDEADLK) {
+			break;
+		}
+		/* sleep for as short a time as we can - more portable than usleep() */
+		tv.tv_sec = 0;
+		tv.tv_usec = 1;
+		select(0, NULL, NULL, NULL, &tv);
+	}
+	TDB_LOG((tdb, 5,"tdb_brlock_upgrade failed at offset %d\n", offset));
+	return -1;
+}
+
+
+/* a byte range locking function - return 0 on success
+   this functions locks/unlocks 1 byte at the specified offset.
+
+   On error, errno is also set so that errors are passed back properly
+   through tdb_open(). */
+int tdb_brlock(struct tdb_context *tdb, tdb_off_t offset, 
+	       int rw_type, int lck_type, int probe)
+{
+	return tdb_brlock_len(tdb, offset, rw_type, lck_type, probe, 1);
+}
+
+/* lock a list in the database. list -1 is the alloc list */
+int tdb_lock(struct tdb_context *tdb, int list, int ltype)
+{
+	if (list < -1 || list >= (int)tdb->header.hash_size) {
+		TDB_LOG((tdb, 0,"tdb_lock: invalid list %d for ltype=%d\n", 
+			   list, ltype));
+		return -1;
+	}
+	if (tdb->flags & TDB_NOLOCK)
+		return 0;
+
+	/* Since fcntl locks don't nest, we do a lock for the first one,
+	   and simply bump the count for future ones */
+	if (tdb->locked[list+1].count == 0) {
+		if (tdb->methods->tdb_brlock(tdb,FREELIST_TOP+4*list,ltype,F_SETLKW, 0)) {
+			TDB_LOG((tdb, 0,"tdb_lock failed on list %d ltype=%d (%s)\n", 
+				 list, ltype, strerror(errno)));
+			return -1;
+		}
+		tdb->locked[list+1].ltype = ltype;
+		tdb->num_locks++;
+	}
+	tdb->locked[list+1].count++;
+	return 0;
+}
+
+/* unlock the database: returns void because it's too late for errors. */
+	/* changed to return int it may be interesting to know there
+	   has been an error  --simo */
+int tdb_unlock(struct tdb_context *tdb, int list, int ltype)
+{
+	int ret = -1;
+
+	if (tdb->flags & TDB_NOLOCK)
+		return 0;
+
+	/* Sanity checks */
+	if (list < -1 || list >= (int)tdb->header.hash_size) {
+		TDB_LOG((tdb, 0, "tdb_unlock: list %d invalid (%d)\n", list, tdb->header.hash_size));
+		return ret;
+	}
+
+	if (tdb->locked[list+1].count==0) {
+		TDB_LOG((tdb, 0, "tdb_unlock: count is 0\n"));
+		return ret;
+	}
+
+	if (tdb->locked[list+1].count == 1) {
+		/* Down to last nested lock: unlock underneath */
+		ret = tdb->methods->tdb_brlock(tdb, FREELIST_TOP+4*list, F_UNLCK, F_SETLKW, 0);
+		tdb->num_locks--;
+	} else {
+		ret = 0;
+	}
+	tdb->locked[list+1].count--;
+
+	if (ret)
+		TDB_LOG((tdb, 0,"tdb_unlock: An error occurred unlocking!\n")); 
+	return ret;
+}
+
+
+
+/* lock/unlock entire database */
+int tdb_lockall(struct tdb_context *tdb)
+{
+	u32 i;
+
+	/* There are no locks on read-only dbs */
+	if (tdb->read_only || tdb->traverse_read)
+		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
+	for (i = 0; i < tdb->header.hash_size; i++) 
+		if (tdb_lock(tdb, i, F_WRLCK))
+			break;
+
+	/* If error, release locks we have... */
+	if (i < tdb->header.hash_size) {
+		u32 j;
+
+		for ( j = 0; j < i; j++)
+			tdb_unlock(tdb, j, F_WRLCK);
+		return TDB_ERRCODE(TDB_ERR_NOLOCK, -1);
+	}
+
+	return 0;
+}
+void tdb_unlockall(struct tdb_context *tdb)
+{
+	u32 i;
+	for (i=0; i < tdb->header.hash_size; i++)
+		tdb_unlock(tdb, i, F_WRLCK);
+}
+
+/* lock/unlock one hash chain. This is meant to be used to reduce
+   contention - it cannot guarantee how many records will be locked */
+int tdb_chainlock(struct tdb_context *tdb, TDB_DATA key)
+{
+	return tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
+}
+
+int tdb_chainunlock(struct tdb_context *tdb, TDB_DATA key)
+{
+	return tdb_unlock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
+}
+
+int tdb_chainlock_read(struct tdb_context *tdb, TDB_DATA key)
+{
+	return tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_RDLCK);
+}
+
+int tdb_chainunlock_read(struct tdb_context *tdb, TDB_DATA key)
+{
+	return tdb_unlock(tdb, BUCKET(tdb->hash_fn(&key)), F_RDLCK);
+}
+
+
+
+/* record lock stops delete underneath */
+int tdb_lock_record(struct tdb_context *tdb, tdb_off_t off)
+{
+	return off ? tdb->methods->tdb_brlock(tdb, off, F_RDLCK, F_SETLKW, 0) : 0;
+}
+
+/*
+  Write locks override our own fcntl readlocks, so check it here.
+  Note this is meant to be F_SETLK, *not* F_SETLKW, as it's not
+  an error to fail to get the lock here.
+*/
+int tdb_write_lock_record(struct tdb_context *tdb, tdb_off_t off)
+{
+	struct tdb_traverse_lock *i;
+	for (i = &tdb->travlocks; i; i = i->next)
+		if (i->off == off)
+			return -1;
+	return tdb->methods->tdb_brlock(tdb, off, F_WRLCK, F_SETLK, 1);
+}
+
+/*
+  Note this is meant to be F_SETLK, *not* F_SETLKW, as it's not
+  an error to fail to get the lock here.
+*/
+int tdb_write_unlock_record(struct tdb_context *tdb, tdb_off_t off)
+{
+	return tdb->methods->tdb_brlock(tdb, off, F_UNLCK, F_SETLK, 0);
+}
+
+/* fcntl locks don't stack: avoid unlocking someone else's */
+int tdb_unlock_record(struct tdb_context *tdb, tdb_off_t off)
+{
+	struct tdb_traverse_lock *i;
+	u32 count = 0;
+
+	if (off == 0)
+		return 0;
+	for (i = &tdb->travlocks; i; i = i->next)
+		if (i->off == off)
+			count++;
+	return (count == 1 ? tdb->methods->tdb_brlock(tdb, off, F_UNLCK, F_SETLKW, 0) : 0);
+}
diff --git a/source/tdb/open.c b/source/tdb/open.c
new file mode 100644
index 00000000000..cb4a9cd9945
--- /dev/null
+++ b/source/tdb/open.c
@@ -0,0 +1,453 @@
+ /* 
+   Unix SMB/CIFS implementation.
+
+   trivial database library
+
+   Copyright (C) Andrew Tridgell              1999-2005
+   Copyright (C) Paul `Rusty' Russell		   2000
+   Copyright (C) Jeremy Allison			   2000-2003
+   
+     ** NOTE! The following LGPL license applies to the tdb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+   
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "tdb_private.h"
+
+/* all contexts, to ensure no double-opens (fcntl locks don't nest!) */
+static struct tdb_context *tdbs = NULL;
+
+
+/* This is based on the hash algorithm from gdbm */
+static unsigned int default_tdb_hash(TDB_DATA *key)
+{
+	u32 value;	/* Used to compute the hash value.  */
+	u32   i;	/* Used to cycle through random values. */
+
+	/* Set the initial value from the key size. */
+	for (value = 0x238F13AF * key->dsize, i=0; i < key->dsize; i++)
+		value = (value + (key->dptr[i] << (i*5 % 24)));
+
+	return (1103515243 * value + 12345);  
+}
+
+
+/* initialise a new database with a specified hash size */
+static int tdb_new_database(struct tdb_context *tdb, int hash_size)
+{
+	struct tdb_header *newdb;
+	int size, ret = -1;
+
+	/* We make it up in memory, then write it out if not internal */
+	size = sizeof(struct tdb_header) + (hash_size+1)*sizeof(tdb_off_t);
+	if (!(newdb = calloc(size, 1)))
+		return TDB_ERRCODE(TDB_ERR_OOM, -1);
+
+	/* Fill in the header */
+	newdb->version = TDB_VERSION;
+	newdb->hash_size = hash_size;
+	if (tdb->flags & TDB_INTERNAL) {
+		tdb->map_size = size;
+		tdb->map_ptr = (char *)newdb;
+		memcpy(&tdb->header, newdb, sizeof(tdb->header));
+		/* Convert the `ondisk' version if asked. */
+		CONVERT(*newdb);
+		return 0;
+	}
+	if (lseek(tdb->fd, 0, SEEK_SET) == -1)
+		goto fail;
+
+	if (ftruncate(tdb->fd, 0) == -1)
+		goto fail;
+
+	/* This creates an endian-converted header, as if read from disk */
+	CONVERT(*newdb);
+	memcpy(&tdb->header, newdb, sizeof(tdb->header));
+	/* Don't endian-convert the magic food! */
+	memcpy(newdb->magic_food, TDB_MAGIC_FOOD, strlen(TDB_MAGIC_FOOD)+1);
+	if (write(tdb->fd, newdb, size) != size) {
+		ret = -1;
+	} else {
+		ret = 0;
+	}
+
+  fail:
+	SAFE_FREE(newdb);
+	return ret;
+}
+
+
+
+static int tdb_already_open(dev_t device,
+			    ino_t ino)
+{
+	struct tdb_context *i;
+	
+	for (i = tdbs; i; i = i->next) {
+		if (i->device == device && i->inode == ino) {
+			return 1;
+		}
+	}
+
+	return 0;
+}
+
+/* open the database, creating it if necessary 
+
+   The open_flags and mode are passed straight to the open call on the
+   database file. A flags value of O_WRONLY is invalid. The hash size
+   is advisory, use zero for a default value.
+
+   Return is NULL on error, in which case errno is also set.  Don't 
+   try to call tdb_error or tdb_errname, just do strerror(errno).
+
+   @param name may be NULL for internal databases. */
+struct tdb_context *tdb_open(const char *name, int hash_size, int tdb_flags,
+		      int open_flags, mode_t mode)
+{
+	return tdb_open_ex(name, hash_size, tdb_flags, open_flags, mode, NULL, NULL);
+}
+
+/* a default logging function */
+static void null_log_fn(struct tdb_context *tdb, int level, const char *fmt, ...)
+{
+}
+
+
+struct tdb_context *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
+			 int open_flags, mode_t mode,
+			 tdb_log_func log_fn,
+			 tdb_hash_func hash_fn)
+{
+	struct tdb_context *tdb;
+	struct stat st;
+	int rev = 0, locked = 0;
+	unsigned char *vp;
+	u32 vertest;
+
+	if (!(tdb = calloc(1, sizeof *tdb))) {
+		/* Can't log this */
+		errno = ENOMEM;
+		goto fail;
+	}
+	tdb_io_init(tdb);
+	tdb->fd = -1;
+	tdb->name = NULL;
+	tdb->map_ptr = NULL;
+	tdb->flags = tdb_flags;
+	tdb->open_flags = open_flags;
+	tdb->log_fn = log_fn?log_fn:null_log_fn;
+	tdb->hash_fn = hash_fn ? hash_fn : default_tdb_hash;
+
+	/* cache the page size */
+	tdb->page_size = getpagesize();
+	if (tdb->page_size <= 0) {
+		tdb->page_size = 0x2000;
+	}
+
+	if ((open_flags & O_ACCMODE) == O_WRONLY) {
+		TDB_LOG((tdb, 0, "tdb_open_ex: can't open tdb %s write-only\n",
+			 name));
+		errno = EINVAL;
+		goto fail;
+	}
+	
+	if (hash_size == 0)
+		hash_size = DEFAULT_HASH_SIZE;
+	if ((open_flags & O_ACCMODE) == O_RDONLY) {
+		tdb->read_only = 1;
+		/* read only databases don't do locking or clear if first */
+		tdb->flags |= TDB_NOLOCK;
+		tdb->flags &= ~TDB_CLEAR_IF_FIRST;
+	}
+
+	/* internal databases don't mmap or lock, and start off cleared */
+	if (tdb->flags & TDB_INTERNAL) {
+		tdb->flags |= (TDB_NOLOCK | TDB_NOMMAP);
+		tdb->flags &= ~TDB_CLEAR_IF_FIRST;
+		if (tdb_new_database(tdb, hash_size) != 0) {
+			TDB_LOG((tdb, 0, "tdb_open_ex: tdb_new_database failed!"));
+			goto fail;
+		}
+		goto internal;
+	}
+
+	if ((tdb->fd = open(name, open_flags, mode)) == -1) {
+		TDB_LOG((tdb, 5, "tdb_open_ex: could not open file %s: %s\n",
+			 name, strerror(errno)));
+		goto fail;	/* errno set by open(2) */
+	}
+
+	/* ensure there is only one process initialising at once */
+	if (tdb->methods->tdb_brlock(tdb, GLOBAL_LOCK, F_WRLCK, F_SETLKW, 0) == -1) {
+		TDB_LOG((tdb, 0, "tdb_open_ex: failed to get global lock on %s: %s\n",
+			 name, strerror(errno)));
+		goto fail;	/* errno set by tdb_brlock */
+	}
+
+	/* we need to zero database if we are the only one with it open */
+	if ((tdb_flags & TDB_CLEAR_IF_FIRST) &&
+	    (locked = (tdb->methods->tdb_brlock(tdb, ACTIVE_LOCK, F_WRLCK, F_SETLK, 0) == 0))) {
+		open_flags |= O_CREAT;
+		if (ftruncate(tdb->fd, 0) == -1) {
+			TDB_LOG((tdb, 0, "tdb_open_ex: "
+				 "failed to truncate %s: %s\n",
+				 name, strerror(errno)));
+			goto fail; /* errno set by ftruncate */
+		}
+	}
+
+	if (read(tdb->fd, &tdb->header, sizeof(tdb->header)) != sizeof(tdb->header)
+	    || strcmp(tdb->header.magic_food, TDB_MAGIC_FOOD) != 0
+	    || (tdb->header.version != TDB_VERSION
+		&& !(rev = (tdb->header.version==TDB_BYTEREV(TDB_VERSION))))) {
+		/* its not a valid database - possibly initialise it */
+		if (!(open_flags & O_CREAT) || tdb_new_database(tdb, hash_size) == -1) {
+			errno = EIO; /* ie bad format or something */
+			goto fail;
+		}
+		rev = (tdb->flags & TDB_CONVERT);
+	}
+	vp = (unsigned char *)&tdb->header.version;
+	vertest = (((u32)vp[0]) << 24) | (((u32)vp[1]) << 16) |
+		  (((u32)vp[2]) << 8) | (u32)vp[3];
+	tdb->flags |= (vertest==TDB_VERSION) ? TDB_BIGENDIAN : 0;
+	if (!rev)
+		tdb->flags &= ~TDB_CONVERT;
+	else {
+		tdb->flags |= TDB_CONVERT;
+		tdb_convert(&tdb->header, sizeof(tdb->header));
+	}
+	if (fstat(tdb->fd, &st) == -1)
+		goto fail;
+
+	if (tdb->header.rwlocks != 0) {
+		TDB_LOG((tdb, 5, "tdb_open_ex: spinlocks no longer supported\n"));
+		goto fail;
+	}
+
+	/* Is it already in the open list?  If so, fail. */
+	if (tdb_already_open(st.st_dev, st.st_ino)) {
+		TDB_LOG((tdb, 2, "tdb_open_ex: "
+			 "%s (%d,%d) is already open in this process\n",
+			 name, (int)st.st_dev, (int)st.st_ino));
+		errno = EBUSY;
+		goto fail;
+	}
+
+	if (!(tdb->name = (char *)strdup(name))) {
+		errno = ENOMEM;
+		goto fail;
+	}
+
+	tdb->map_size = st.st_size;
+	tdb->device = st.st_dev;
+	tdb->inode = st.st_ino;
+	tdb->locked = calloc(tdb->header.hash_size+1, sizeof(tdb->locked[0]));
+	if (!tdb->locked) {
+		TDB_LOG((tdb, 2, "tdb_open_ex: "
+			 "failed to allocate lock structure for %s\n",
+			 name));
+		errno = ENOMEM;
+		goto fail;
+	}
+	tdb_mmap(tdb);
+	if (locked) {
+		if (tdb->methods->tdb_brlock(tdb, ACTIVE_LOCK, F_UNLCK, F_SETLK, 0) == -1) {
+			TDB_LOG((tdb, 0, "tdb_open_ex: "
+				 "failed to take ACTIVE_LOCK on %s: %s\n",
+				 name, strerror(errno)));
+			goto fail;
+		}
+
+	}
+
+	/* We always need to do this if the CLEAR_IF_FIRST flag is set, even if
+	   we didn't get the initial exclusive lock as we need to let all other
+	   users know we're using it. */
+
+	if (tdb_flags & TDB_CLEAR_IF_FIRST) {
+		/* leave this lock in place to indicate it's in use */
+		if (tdb->methods->tdb_brlock(tdb, ACTIVE_LOCK, F_RDLCK, F_SETLKW, 0) == -1)
+			goto fail;
+	}
+
+	/* if needed, run recovery */
+	if (tdb_transaction_recover(tdb) == -1) {
+		goto fail;
+	}
+
+ internal:
+	/* Internal (memory-only) databases skip all the code above to
+	 * do with disk files, and resume here by releasing their
+	 * global lock and hooking into the active list. */
+	if (tdb->methods->tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0) == -1)
+		goto fail;
+	tdb->next = tdbs;
+	tdbs = tdb;
+	return tdb;
+
+ fail:
+	{ int save_errno = errno;
+
+	if (!tdb)
+		return NULL;
+	
+	if (tdb->map_ptr) {
+		if (tdb->flags & TDB_INTERNAL)
+			SAFE_FREE(tdb->map_ptr);
+		else
+			tdb_munmap(tdb);
+	}
+	SAFE_FREE(tdb->name);
+	if (tdb->fd != -1)
+		if (close(tdb->fd) != 0)
+			TDB_LOG((tdb, 5, "tdb_open_ex: failed to close tdb->fd on error!\n"));
+	SAFE_FREE(tdb->locked);
+	SAFE_FREE(tdb);
+	errno = save_errno;
+	return NULL;
+	}
+}
+
+/**
+ * Close a database.
+ *
+ * @returns -1 for error; 0 for success.
+ **/
+int tdb_close(struct tdb_context *tdb)
+{
+	struct tdb_context **i;
+	int ret = 0;
+
+	if (tdb->transaction) {
+		tdb_transaction_cancel(tdb);
+	}
+
+	if (tdb->map_ptr) {
+		if (tdb->flags & TDB_INTERNAL)
+			SAFE_FREE(tdb->map_ptr);
+		else
+			tdb_munmap(tdb);
+	}
+	SAFE_FREE(tdb->name);
+	if (tdb->fd != -1)
+		ret = close(tdb->fd);
+	SAFE_FREE(tdb->locked);
+
+	/* Remove from contexts list */
+	for (i = &tdbs; *i; i = &(*i)->next) {
+		if (*i == tdb) {
+			*i = tdb->next;
+			break;
+		}
+	}
+
+	memset(tdb, 0, sizeof(*tdb));
+	SAFE_FREE(tdb);
+
+	return ret;
+}
+
+/* register a loging function */
+void tdb_logging_function(struct tdb_context *tdb, void (*fn)(struct tdb_context *, int , const char *, ...))
+{
+	tdb->log_fn = fn?fn:null_log_fn;
+}
+
+
+/* reopen a tdb - this can be used after a fork to ensure that we have an independent
+   seek pointer from our parent and to re-establish locks */
+int tdb_reopen(struct tdb_context *tdb)
+{
+	struct stat st;
+
+	if (tdb->flags & TDB_INTERNAL) {
+		return 0; /* Nothing to do. */
+	}
+
+	if (tdb->num_locks != 0) {
+		TDB_LOG((tdb, 0, "tdb_reopen: reopen not allowed with locks held\n"));
+		goto fail;
+	}
+
+	if (tdb->transaction != 0) {
+		TDB_LOG((tdb, 0, "tdb_reopen: reopen not allowed inside a transaction\n"));
+		goto fail;
+	}
+
+	if (tdb_munmap(tdb) != 0) {
+		TDB_LOG((tdb, 0, "tdb_reopen: munmap failed (%s)\n", strerror(errno)));
+		goto fail;
+	}
+	if (close(tdb->fd) != 0)
+		TDB_LOG((tdb, 0, "tdb_reopen: WARNING closing tdb->fd failed!\n"));
+	tdb->fd = open(tdb->name, tdb->open_flags & ~(O_CREAT|O_TRUNC), 0);
+	if (tdb->fd == -1) {
+		TDB_LOG((tdb, 0, "tdb_reopen: open failed (%s)\n", strerror(errno)));
+		goto fail;
+	}
+	if ((tdb->flags & TDB_CLEAR_IF_FIRST) && 
+	    (tdb->methods->tdb_brlock(tdb, ACTIVE_LOCK, F_RDLCK, F_SETLKW, 0) == -1)) {
+		TDB_LOG((tdb, 0, "tdb_reopen: failed to obtain active lock\n"));
+		goto fail;
+	}
+	if (fstat(tdb->fd, &st) != 0) {
+		TDB_LOG((tdb, 0, "tdb_reopen: fstat failed (%s)\n", strerror(errno)));
+		goto fail;
+	}
+	if (st.st_ino != tdb->inode || st.st_dev != tdb->device) {
+		TDB_LOG((tdb, 0, "tdb_reopen: file dev/inode has changed!\n"));
+		goto fail;
+	}
+	tdb_mmap(tdb);
+
+	return 0;
+
+fail:
+	tdb_close(tdb);
+	return -1;
+}
+
+/* reopen all tdb's */
+int tdb_reopen_all(int parent_longlived)
+{
+	struct tdb_context *tdb;
+
+	for (tdb=tdbs; tdb; tdb = tdb->next) {
+		/*
+		 * If the parent is longlived (ie. a
+		 * parent daemon architecture), we know
+		 * it will keep it's active lock on a
+		 * tdb opened with CLEAR_IF_FIRST. Thus
+		 * for child processes we don't have to
+		 * add an active lock. This is essential
+		 * to improve performance on systems that
+		 * keep POSIX locks as a non-scalable data
+		 * structure in the kernel.
+		 */
+		if (parent_longlived) {
+			/* Ensure no clear-if-first. */
+			tdb->flags &= ~TDB_CLEAR_IF_FIRST;
+		}
+
+		if (tdb_reopen(tdb) != 0)
+			return -1;
+	}
+
+	return 0;
+}
diff --git a/source/tdb/spinlock.c b/source/tdb/spinlock.c
deleted file mode 100644
index e42a6901c71..00000000000
--- a/source/tdb/spinlock.c
+++ /dev/null
@@ -1,472 +0,0 @@
-/* 
-   Unix SMB/CIFS implementation.
-
-   trivial database library 
-
-   Copyright (C) Anton Blanchard                   2001
-   
-     ** NOTE! The following LGPL license applies to the tdb
-     ** library. This does NOT imply that all of Samba is released
-     ** under the LGPL
-   
-   This library is free software; you can redistribute it and/or
-   modify it under the terms of the GNU Lesser General Public
-   License as published by the Free Software Foundation; either
-   version 2 of the License, or (at your option) any later version.
-
-   This library is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
-   Lesser General Public License for more details.
-   
-   You should have received a copy of the GNU Lesser General Public
-   License along with this library; if not, write to the Free Software
-   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
-*/
-#if HAVE_CONFIG_H
-#include <config.h>
-#endif
-
-#ifdef STANDALONE
-#include <stdlib.h>
-#include <stdio.h>
-#include <unistd.h>
-#include <string.h>
-#include <fcntl.h>
-#include <errno.h>
-#include <sys/stat.h>
-#include <time.h>
-#include <signal.h>
-#include "tdb.h"
-#include "spinlock.h"
-
-#define DEBUG
-#else
-#include "includes.h"
-#endif
-
-#ifdef USE_SPINLOCKS
-
-/*
- * ARCH SPECIFIC
- */
-
-#if defined(SPARC_SPINLOCKS)
-
-static inline int __spin_trylock(spinlock_t *lock)
-{
-	unsigned int result;
-
-	asm volatile("ldstub    [%1], %0"
-		: "=r" (result)
-		: "r" (lock)
-		: "memory");
-
-	return (result == 0) ? 0 : EBUSY;
-}
-
-static inline void __spin_unlock(spinlock_t *lock)
-{
-	asm volatile("":::"memory");
-	*lock = 0;
-}
-
-static inline void __spin_lock_init(spinlock_t *lock)
-{
-	*lock = 0;
-}
-
-static inline int __spin_is_locked(spinlock_t *lock)
-{
-	return (*lock != 0);
-}
-
-#elif defined(POWERPC_SPINLOCKS) 
-
-static inline int __spin_trylock(spinlock_t *lock)
-{
-	unsigned int result;
-
-	__asm__ __volatile__(
-"1:	lwarx		%0,0,%1\n\
-	cmpwi		0,%0,0\n\
-	li		%0,0\n\
-	bne-		2f\n\
-	li		%0,1\n\
-	stwcx.		%0,0,%1\n\
-	bne-		1b\n\
-	isync\n\
-2:"	: "=&r"(result)
-	: "r"(lock)
-	: "cr0", "memory");
-
-	return (result == 1) ? 0 : EBUSY;
-}
-
-static inline void __spin_unlock(spinlock_t *lock)
-{
-	asm volatile("eieio":::"memory");
-	*lock = 0;
-}
-
-static inline void __spin_lock_init(spinlock_t *lock)
-{
-	*lock = 0;
-}
-
-static inline int __spin_is_locked(spinlock_t *lock)
-{
-	return (*lock != 0);
-}
-
-#elif defined(INTEL_SPINLOCKS) 
-
-static inline int __spin_trylock(spinlock_t *lock)
-{
-	int oldval;
-
-	asm volatile("xchgl %0,%1"
-		: "=r" (oldval), "=m" (*lock)
-		: "0" (0)
-		: "memory");
-
-	return oldval > 0 ? 0 : EBUSY;
-}
-
-static inline void __spin_unlock(spinlock_t *lock)
-{
-	asm volatile("":::"memory");
-	*lock = 1;
-}
-
-static inline void __spin_lock_init(spinlock_t *lock)
-{
-	*lock = 1;
-}
-
-static inline int __spin_is_locked(spinlock_t *lock)
-{
-	return (*lock != 1);
-}
-
-#elif defined(MIPS_SPINLOCKS) && defined(sgi) && (_COMPILER_VERSION >= 730)
-
-/* Implement spinlocks on IRIX using the MIPSPro atomic fetch operations. See
- * sync(3) for the details of the intrinsic operations.
- *
- * "sgi" and "_COMPILER_VERSION" are always defined by MIPSPro.
- */
-
-#ifdef STANDALONE
-
-/* MIPSPro 7.3 has "__inline" as an extension, but not "inline. */
-#define inline __inline
-
-#endif /* STANDALONE */
-
-/* Returns 0 if the lock is acquired, EBUSY otherwise. */
-static inline int __spin_trylock(spinlock_t *lock)
-{
-        unsigned int val;
-        val = __lock_test_and_set(lock, 1);
-        return val == 0 ? 0 : EBUSY;
-}
-
-static inline void __spin_unlock(spinlock_t *lock)
-{
-        __lock_release(lock);
-}
-
-static inline void __spin_lock_init(spinlock_t *lock)
-{
-        __lock_release(lock);
-}
-
-/* Returns 1 if the lock is held, 0 otherwise. */
-static inline int __spin_is_locked(spinlock_t *lock)
-{
-        unsigned int val;
-        val = __add_and_fetch(lock, 0);
-	return val;
-}
-
-#elif defined(MIPS_SPINLOCKS) 
-
-static inline unsigned int load_linked(unsigned long addr)
-{
-	unsigned int res;
-
-	__asm__ __volatile__("ll\t%0,(%1)"
-		: "=r" (res)
-		: "r" (addr));
-
-	return res;
-}
-
-static inline unsigned int store_conditional(unsigned long addr, unsigned int value)
-{
-	unsigned int res;
-
-	__asm__ __volatile__("sc\t%0,(%2)"
-		: "=r" (res)
-		: "0" (value), "r" (addr));
-	return res;
-}
-
-static inline int __spin_trylock(spinlock_t *lock)
-{
-	unsigned int mw;
-
-	do {
-		mw = load_linked(lock);
-		if (mw) 
-			return EBUSY;
-	} while (!store_conditional(lock, 1));
-
-	asm volatile("":::"memory");
-
-	return 0;
-}
-
-static inline void __spin_unlock(spinlock_t *lock)
-{
-	asm volatile("":::"memory");
-	*lock = 0;
-}
-
-static inline void __spin_lock_init(spinlock_t *lock)
-{
-	*lock = 0;
-}
-
-static inline int __spin_is_locked(spinlock_t *lock)
-{
-	return (*lock != 0);
-}
-
-#else
-#error Need to implement spinlock code in spinlock.c
-#endif
-
-/*
- * OS SPECIFIC
- */
-
-static void yield_cpu(void)
-{
-	struct timespec tm;
-
-#ifdef USE_SCHED_YIELD
-	sched_yield();
-#else
-	/* Linux will busy loop for delays < 2ms on real time tasks */
-	tm.tv_sec = 0;
-	tm.tv_nsec = 2000000L + 1;
-	nanosleep(&tm, NULL);
-#endif
-}
-
-/*
- * GENERIC
- */
-
-static int smp_machine = 0;
-
-static inline void __spin_lock(spinlock_t *lock)
-{
-	int ntries = 0;
-
-	while(__spin_trylock(lock)) {
-		while(__spin_is_locked(lock)) {
-			if (smp_machine && ntries++ < MAX_BUSY_LOOPS)
-				continue;
-			yield_cpu();
-		}
-	}
-}
-
-static void __read_lock(tdb_rwlock_t *rwlock)
-{
-	int ntries = 0;
-
-	while(1) {
-		__spin_lock(&rwlock->lock);
-
-		if (!(rwlock->count & RWLOCK_BIAS)) {
-			rwlock->count++;
-			__spin_unlock(&rwlock->lock);
-			return;
-		}
-	
-		__spin_unlock(&rwlock->lock);
-
-		while(rwlock->count & RWLOCK_BIAS) {
-			if (smp_machine && ntries++ < MAX_BUSY_LOOPS)
-				continue;
-			yield_cpu();
-		}
-	}
-}
-
-static void __write_lock(tdb_rwlock_t *rwlock)
-{
-	int ntries = 0;
-
-	while(1) {
-		__spin_lock(&rwlock->lock);
-
-		if (rwlock->count == 0) {
-			rwlock->count |= RWLOCK_BIAS;
-			__spin_unlock(&rwlock->lock);
-			return;
-		}
-
-		__spin_unlock(&rwlock->lock);
-
-		while(rwlock->count != 0) {
-			if (smp_machine && ntries++ < MAX_BUSY_LOOPS)
-				continue;
-			yield_cpu();
-		}
-	}
-}
-
-static void __write_unlock(tdb_rwlock_t *rwlock)
-{
-	__spin_lock(&rwlock->lock);
-
-#ifdef DEBUG
-	if (!(rwlock->count & RWLOCK_BIAS))
-		fprintf(stderr, "bug: write_unlock\n");
-#endif
-
-	rwlock->count &= ~RWLOCK_BIAS;
-	__spin_unlock(&rwlock->lock);
-}
-
-static void __read_unlock(tdb_rwlock_t *rwlock)
-{
-	__spin_lock(&rwlock->lock);
-
-#ifdef DEBUG
-	if (!rwlock->count)
-		fprintf(stderr, "bug: read_unlock\n");
-
-	if (rwlock->count & RWLOCK_BIAS)
-		fprintf(stderr, "bug: read_unlock\n");
-#endif
-
-	rwlock->count--;
-	__spin_unlock(&rwlock->lock);
-}
-
-/* TDB SPECIFIC */
-
-/* lock a list in the database. list -1 is the alloc list */
-int tdb_spinlock(TDB_CONTEXT *tdb, int list, int rw_type)
-{
-	tdb_rwlock_t *rwlocks;
-
-	if (!tdb->map_ptr) return -1;
-	rwlocks = (tdb_rwlock_t *)((char *)tdb->map_ptr + tdb->header.rwlocks);
-
-	switch(rw_type) {
-	case F_RDLCK:
-		__read_lock(&rwlocks[list+1]);
-		break;
-
-	case F_WRLCK:
-		__write_lock(&rwlocks[list+1]);
-		break;
-
-	default:
-		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
-	}
-	return 0;
-}
-
-/* unlock the database. */
-int tdb_spinunlock(TDB_CONTEXT *tdb, int list, int rw_type)
-{
-	tdb_rwlock_t *rwlocks;
-
-	if (!tdb->map_ptr) return -1;
-	rwlocks = (tdb_rwlock_t *)((char *)tdb->map_ptr + tdb->header.rwlocks);
-
-	switch(rw_type) {
-	case F_RDLCK:
-		__read_unlock(&rwlocks[list+1]);
-		break;
-
-	case F_WRLCK:
-		__write_unlock(&rwlocks[list+1]);
-		break;
-
-	default:
-		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
-	}
-
-	return 0;
-}
-
-int tdb_create_rwlocks(int fd, unsigned int hash_size)
-{
-	unsigned size, i;
-	tdb_rwlock_t *rwlocks;
-
-	size = TDB_SPINLOCK_SIZE(hash_size);
-	rwlocks = malloc(size);
-	if (!rwlocks)
-		return -1;
-
-	for(i = 0; i < hash_size+1; i++) {
-		__spin_lock_init(&rwlocks[i].lock);
-		rwlocks[i].count = 0;
-	}
-
-	/* Write it out (appending to end) */
-	if (write(fd, rwlocks, size) != size) {
-		free(rwlocks);
-		return -1;
-	}
-	smp_machine = this_is_smp();
-	free(rwlocks);
-	return 0;
-}
-
-int tdb_clear_spinlocks(TDB_CONTEXT *tdb)
-{
-	tdb_rwlock_t *rwlocks;
-	unsigned i;
-
-	if (tdb->header.rwlocks == 0) return 0;
-	if (!tdb->map_ptr) return -1;
-
-	/* We're mmapped here */
-	rwlocks = (tdb_rwlock_t *)((char *)tdb->map_ptr + tdb->header.rwlocks);
-	for(i = 0; i < tdb->header.hash_size+1; i++) {
-		__spin_lock_init(&rwlocks[i].lock);
-		rwlocks[i].count = 0;
-	}
-	return 0;
-}
-#else
-int tdb_create_rwlocks(int fd, unsigned int hash_size) { return 0; }
-int tdb_spinlock(TDB_CONTEXT *tdb, int list, int rw_type) { return -1; }
-int tdb_spinunlock(TDB_CONTEXT *tdb, int list, int rw_type) { return -1; }
-
-/* Non-spinlock version: remove spinlock pointer */
-int tdb_clear_spinlocks(TDB_CONTEXT *tdb)
-{
-	tdb_off off = (tdb_off)((char *)&tdb->header.rwlocks
-				- (char *)&tdb->header);
-
-	tdb->header.rwlocks = 0;
-	if (lseek(tdb->fd, off, SEEK_SET) != off
-	    || write(tdb->fd, (void *)&tdb->header.rwlocks,
-		     sizeof(tdb->header.rwlocks)) 
-	    != sizeof(tdb->header.rwlocks))
-		return -1;
-	return 0;
-}
-#endif
diff --git a/source/tdb/spinlock.h b/source/tdb/spinlock.h
deleted file mode 100644
index 967fe37457f..00000000000
--- a/source/tdb/spinlock.h
+++ /dev/null
@@ -1,59 +0,0 @@
-#ifndef __SPINLOCK_H__
-#define __SPINLOCK_H__
-
-#ifdef HAVE_CONFIG_H
-#include <config.h>
-#endif
-
-#include "tdb.h"
-
-#ifdef USE_SPINLOCKS
-
-#define RWLOCK_BIAS 0x1000UL
-
-/* OS SPECIFIC */
-#define MAX_BUSY_LOOPS 1000
-#undef USE_SCHED_YIELD
-
-/* ARCH SPECIFIC */
-/* We should make sure these are padded to a cache line */
-#if defined(SPARC_SPINLOCKS)
-typedef volatile char spinlock_t;
-#elif defined(POWERPC_SPINLOCKS)
-typedef volatile unsigned long spinlock_t;
-#elif defined(INTEL_SPINLOCKS)
-typedef volatile int spinlock_t;
-#elif defined(MIPS_SPINLOCKS)
-typedef volatile unsigned long spinlock_t;
-#else
-#error Need to implement spinlock code in spinlock.h
-#endif
-
-typedef struct {
-	spinlock_t lock;
-	volatile int count;
-} tdb_rwlock_t;
-
-int tdb_spinlock(TDB_CONTEXT *tdb, int list, int rw_type);
-int tdb_spinunlock(TDB_CONTEXT *tdb, int list, int rw_type);
-int tdb_create_rwlocks(int fd, unsigned int hash_size);
-int tdb_clear_spinlocks(TDB_CONTEXT *tdb);
-
-#define TDB_SPINLOCK_SIZE(hash_size) (((hash_size) + 1) * sizeof(tdb_rwlock_t))
-
-#else /* !USE_SPINLOCKS */
-#if 0
-#define tdb_create_rwlocks(fd, hash_size) 0
-#define tdb_spinlock(tdb, list, rw_type) (-1)
-#define tdb_spinunlock(tdb, list, rw_type) (-1)
-#else
-int tdb_spinlock(TDB_CONTEXT *tdb, int list, int rw_type);
-int tdb_spinunlock(TDB_CONTEXT *tdb, int list, int rw_type);
-int tdb_create_rwlocks(int fd, unsigned int hash_size);
-#endif
-int tdb_clear_spinlocks(TDB_CONTEXT *tdb);
-#define TDB_SPINLOCK_SIZE(hash_size) 0
-
-#endif
-
-#endif
diff --git a/source/tdb/tdb.c b/source/tdb/tdb.c
index ad73a1d9aa0..1fae6b6de1d 100644
--- a/source/tdb/tdb.c
+++ b/source/tdb/tdb.c
@@ -3,7 +3,7 @@
 
    trivial database library
 
-   Copyright (C) Andrew Tridgell              1999-2004
+   Copyright (C) Andrew Tridgell              1999-2005
    Copyright (C) Paul `Rusty' Russell		   2000
    Copyright (C) Jeremy Allison			   2000-2003
    
@@ -20,1004 +20,57 @@
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
    Lesser General Public License for more details.
-   
+
    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */
 
+#include "tdb_private.h"
 
-/* NOTE: If you use tdbs under valgrind, and in particular if you run
- * tdbtorture, you may get spurious "uninitialized value" warnings.  I
- * think this is because valgrind doesn't understand that the mmap'd
- * area may be written to by other processes.  Memory can, from the
- * point of view of the grinded process, spontaneously become
- * initialized.
- *
- * I can think of a few solutions.  [mbp 20030311]
- *
- * 1 - Write suppressions for Valgrind so that it doesn't complain
- * about this.  Probably the most reasonable but people need to
- * remember to use them.
- *
- * 2 - Use IO not mmap when running under valgrind.  Not so nice.
- *
- * 3 - Use the special valgrind macros to mark memory as valid at the
- * right time.  Probably too hard -- the process just doesn't know.
- */ 
-
-#ifdef STANDALONE
-#if HAVE_CONFIG_H
-#include <config.h>
-#endif
-
-#include <stdlib.h>
-#include <stdio.h>
-#include <fcntl.h>
-#include <unistd.h>
-#include <string.h>
-#include <fcntl.h>
-#include <errno.h>
-#include <sys/mman.h>
-#include <sys/stat.h>
-#include <signal.h>
-#include "tdb.h"
-#include "spinlock.h"
-#else
-#include "includes.h"
-
-#if defined(PARANOID_MALLOC_CHECKER)
-#ifdef malloc
-#undef malloc
-#endif
-
-#ifdef realloc
-#undef realloc
-#endif
-
-#ifdef calloc
-#undef calloc
-#endif
-
-#ifdef strdup
-#undef strdup
-#endif
-
-#ifdef strndup
-#undef strndup
-#endif
-
-#endif
-
-#endif
-
-#define TDB_MAGIC_FOOD "TDB file\n"
-#define TDB_VERSION (0x26011967 + 6)
-#define TDB_MAGIC (0x26011999U)
-#define TDB_FREE_MAGIC (~TDB_MAGIC)
-#define TDB_DEAD_MAGIC (0xFEE1DEAD)
-#define TDB_ALIGNMENT 4
-#define MIN_REC_SIZE (2*sizeof(struct list_struct) + TDB_ALIGNMENT)
-#define DEFAULT_HASH_SIZE 131
-#define TDB_PAGE_SIZE 0x2000
-#define FREELIST_TOP (sizeof(struct tdb_header))
-#define TDB_ALIGN(x,a) (((x) + (a)-1) & ~((a)-1))
-#define TDB_BYTEREV(x) (((((x)&0xff)<<24)|((x)&0xFF00)<<8)|(((x)>>8)&0xFF00)|((x)>>24))
-#define TDB_DEAD(r) ((r)->magic == TDB_DEAD_MAGIC)
-#define TDB_BAD_MAGIC(r) ((r)->magic != TDB_MAGIC && !TDB_DEAD(r))
-#define TDB_HASH_TOP(hash) (FREELIST_TOP + (BUCKET(hash)+1)*sizeof(tdb_off))
-#define TDB_DATA_START(hash_size) (TDB_HASH_TOP(hash_size-1) + TDB_SPINLOCK_SIZE(hash_size))
-
-
-/* NB assumes there is a local variable called "tdb" that is the
- * current context, also takes doubly-parenthesized print-style
- * argument. */
-#define TDB_LOG(x) (tdb->log_fn?((tdb->log_fn x),0) : 0)
-
-/* lock offsets */
-#define GLOBAL_LOCK 0
-#define ACTIVE_LOCK 4
-
-#ifndef MAP_FILE
-#define MAP_FILE 0
-#endif
-
-#ifndef MAP_FAILED
-#define MAP_FAILED ((void *)-1)
-#endif
-
-/* free memory if the pointer is valid and zero the pointer */
-#ifndef SAFE_FREE
-#define SAFE_FREE(x) do { if ((x) != NULL) {free((x)); (x)=NULL;} } while(0)
-#endif
-
-#define BUCKET(hash) ((hash) % tdb->header.hash_size)
 TDB_DATA tdb_null;
 
-/* all contexts, to ensure no double-opens (fcntl locks don't nest!) */
-static TDB_CONTEXT *tdbs = NULL;
-
-static int tdb_munmap(TDB_CONTEXT *tdb)
+/*
+  increment the tdb sequence number if the tdb has been opened using
+  the TDB_SEQNUM flag
+*/
+static void tdb_increment_seqnum(struct tdb_context *tdb)
 {
-	if (tdb->flags & TDB_INTERNAL)
-		return 0;
-
-#ifdef HAVE_MMAP
-	if (tdb->map_ptr) {
-		int ret = munmap(tdb->map_ptr, tdb->map_size);
-		if (ret != 0)
-			return ret;
+	tdb_off_t seqnum=0;
+	
+	if (!(tdb->flags & TDB_SEQNUM)) {
+		return;
 	}
-#endif
-	tdb->map_ptr = NULL;
-	return 0;
-}
 
-static void tdb_mmap(TDB_CONTEXT *tdb)
-{
-	if (tdb->flags & TDB_INTERNAL)
+	if (tdb_brlock(tdb, TDB_SEQNUM_OFS, F_WRLCK, F_SETLKW, 1) != 0) {
 		return;
-
-#ifdef HAVE_MMAP
-	if (!(tdb->flags & TDB_NOMMAP)) {
-		tdb->map_ptr = mmap(NULL, tdb->map_size, 
-				    PROT_READ|(tdb->read_only? 0:PROT_WRITE), 
-				    MAP_SHARED|MAP_FILE, tdb->fd, 0);
-
-		/*
-		 * NB. When mmap fails it returns MAP_FAILED *NOT* NULL !!!!
-		 */
-
-		if (tdb->map_ptr == MAP_FAILED) {
-			tdb->map_ptr = NULL;
-			TDB_LOG((tdb, 2, "tdb_mmap failed for size %d (%s)\n", 
-				 tdb->map_size, strerror(errno)));
-		}
-	} else {
-		tdb->map_ptr = NULL;
 	}
-#else
-	tdb->map_ptr = NULL;
-#endif
-}
 
-/* Endian conversion: we only ever deal with 4 byte quantities */
-static void *convert(void *buf, u32 size)
-{
-	u32 i, *p = buf;
-	for (i = 0; i < size / 4; i++)
-		p[i] = TDB_BYTEREV(p[i]);
-	return buf;
-}
-#define DOCONV() (tdb->flags & TDB_CONVERT)
-#define CONVERT(x) (DOCONV() ? convert(&x, sizeof(x)) : &x)
-
-/* the body of the database is made of one list_struct for the free space
-   plus a separate data list for each hash value */
-struct list_struct {
-	tdb_off next; /* offset of the next record in the list */
-	tdb_len rec_len; /* total byte length of record */
-	tdb_len key_len; /* byte length of key */
-	tdb_len data_len; /* byte length of data */
-	u32 full_hash; /* the full 32 bit hash of the key */
-	u32 magic;   /* try to catch errors */
-	/* the following union is implied:
-		union {
-			char record[rec_len];
-			struct {
-				char key[key_len];
-				char data[data_len];
-			}
-			u32 totalsize; (tailer)
-		}
+	/* we ignore errors from this, as we have no sane way of
+	   dealing with them.
 	*/
-};
-
-/***************************************************************
- Allow a caller to set a "alarm" flag that tdb can check to abort
- a blocking lock on SIGALRM.
-***************************************************************/
-
-static sig_atomic_t *palarm_fired;
-
-void tdb_set_lock_alarm(sig_atomic_t *palarm)
-{
-	palarm_fired = palarm;
-}
-
-/* a byte range locking function - return 0 on success
-   this functions locks/unlocks 1 byte at the specified offset.
-
-   On error, errno is also set so that errors are passed back properly
-   through tdb_open(). */
-static int tdb_brlock(TDB_CONTEXT *tdb, tdb_off offset, 
-		      int rw_type, int lck_type, int probe)
-{
-	struct flock fl;
-	int ret;
-
-	if (tdb->flags & TDB_NOLOCK)
-		return 0;
-	if ((rw_type == F_WRLCK) && (tdb->read_only)) {
-		errno = EACCES;
-		return -1;
-	}
-
-	fl.l_type = rw_type;
-	fl.l_whence = SEEK_SET;
-	fl.l_start = offset;
-	fl.l_len = 1;
-	fl.l_pid = 0;
-
-	do {
-		ret = fcntl(tdb->fd,lck_type,&fl);
-		if (ret == -1 && errno == EINTR && palarm_fired && *palarm_fired)
-			break;
-	} while (ret == -1 && errno == EINTR);
-
-	if (ret == -1) {
-		if (!probe && lck_type != F_SETLK) {
-			/* Ensure error code is set for log fun to examine. */
-			if (errno == EINTR && palarm_fired && *palarm_fired)
-				tdb->ecode = TDB_ERR_LOCK_TIMEOUT;
-			else
-				tdb->ecode = TDB_ERR_LOCK;
-			TDB_LOG((tdb, 5,"tdb_brlock failed (fd=%d) at offset %d rw_type=%d lck_type=%d\n", 
-				 tdb->fd, offset, rw_type, lck_type));
-		}
-		/* Was it an alarm timeout ? */
-		if (errno == EINTR && palarm_fired && *palarm_fired) {
-			TDB_LOG((tdb, 5, "tdb_brlock timed out (fd=%d) at offset %d rw_type=%d lck_type=%d\n", 
-				 tdb->fd, offset, rw_type, lck_type));
-			return TDB_ERRCODE(TDB_ERR_LOCK_TIMEOUT, -1);
-		}
-		/* Otherwise - generic lock error. errno set by fcntl.
-		 * EAGAIN is an expected return from non-blocking
-		 * locks. */
-		if (errno != EAGAIN) {
-			TDB_LOG((tdb, 5, "tdb_brlock failed (fd=%d) at offset %d rw_type=%d lck_type=%d: %s\n", 
-				 tdb->fd, offset, rw_type, lck_type, 
-				 strerror(errno)));
-		}
-		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
-	}
-	return 0;
-}
-
-/* lock a list in the database. list -1 is the alloc list */
-static int tdb_lock(TDB_CONTEXT *tdb, int list, int ltype)
-{
-	if (list < -1 || list >= (int)tdb->header.hash_size) {
-		TDB_LOG((tdb, 0,"tdb_lock: invalid list %d for ltype=%d\n", 
-			   list, ltype));
-		return -1;
-	}
-	if (tdb->flags & TDB_NOLOCK)
-		return 0;
-
-	/* Since fcntl locks don't nest, we do a lock for the first one,
-	   and simply bump the count for future ones */
-	if (tdb->locked[list+1].count == 0) {
-		if (!tdb->read_only && tdb->header.rwlocks) {
-			if (tdb_spinlock(tdb, list, ltype)) {
-				TDB_LOG((tdb, 0, "tdb_lock spinlock failed on list %d ltype=%d\n", 
-					   list, ltype));
-				return -1;
-			}
-		} else if (tdb_brlock(tdb,FREELIST_TOP+4*list,ltype,F_SETLKW, 0)) {
-			TDB_LOG((tdb, 0,"tdb_lock failed on list %d ltype=%d (%s)\n", 
-					   list, ltype, strerror(errno)));
-			return -1;
-		}
-		tdb->locked[list+1].ltype = ltype;
-	}
-	tdb->locked[list+1].count++;
-	return 0;
-}
-
-/* unlock the database: returns void because it's too late for errors. */
-	/* changed to return int it may be interesting to know there
-	   has been an error  --simo */
-static int tdb_unlock(TDB_CONTEXT *tdb, int list, int ltype)
-{
-	int ret = -1;
-
-	if (tdb->flags & TDB_NOLOCK)
-		return 0;
-
-	/* Sanity checks */
-	if (list < -1 || list >= (int)tdb->header.hash_size) {
-		TDB_LOG((tdb, 0, "tdb_unlock: list %d invalid (%d)\n", list, tdb->header.hash_size));
-		return ret;
-	}
-
-	if (tdb->locked[list+1].count==0) {
-		TDB_LOG((tdb, 0, "tdb_unlock: count is 0\n"));
-		return ret;
-	}
-
-	if (tdb->locked[list+1].count == 1) {
-		/* Down to last nested lock: unlock underneath */
-		if (!tdb->read_only && tdb->header.rwlocks) {
-			ret = tdb_spinunlock(tdb, list, ltype);
-		} else {
-			ret = tdb_brlock(tdb, FREELIST_TOP+4*list, F_UNLCK, F_SETLKW, 0);
-		}
-	} else {
-		ret = 0;
-	}
-	tdb->locked[list+1].count--;
-
-	if (ret)
-		TDB_LOG((tdb, 0,"tdb_unlock: An error occurred unlocking!\n")); 
-	return ret;
-}
-
-/* check for an out of bounds access - if it is out of bounds then
-   see if the database has been expanded by someone else and expand
-   if necessary 
-   note that "len" is the minimum length needed for the db
-*/
-static int tdb_oob(TDB_CONTEXT *tdb, tdb_off len, int probe)
-{
-	struct stat st;
-	if (len <= tdb->map_size)
-		return 0;
-	if (tdb->flags & TDB_INTERNAL) {
-		if (!probe) {
-			/* Ensure ecode is set for log fn. */
-			tdb->ecode = TDB_ERR_IO;
-			TDB_LOG((tdb, 0,"tdb_oob len %d beyond internal malloc size %d\n",
-				 (int)len, (int)tdb->map_size));
-		}
-		return TDB_ERRCODE(TDB_ERR_IO, -1);
-	}
-
-	if (fstat(tdb->fd, &st) == -1)
-		return TDB_ERRCODE(TDB_ERR_IO, -1);
-
-	if (st.st_size < (size_t)len) {
-		if (!probe) {
-			/* Ensure ecode is set for log fn. */
-			tdb->ecode = TDB_ERR_IO;
-			TDB_LOG((tdb, 0,"tdb_oob len %d beyond eof at %d\n",
-				 (int)len, (int)st.st_size));
-		}
-		return TDB_ERRCODE(TDB_ERR_IO, -1);
-	}
-
-	/* Unmap, update size, remap */
-	if (tdb_munmap(tdb) == -1)
-		return TDB_ERRCODE(TDB_ERR_IO, -1);
-	tdb->map_size = st.st_size;
-	tdb_mmap(tdb);
-	return 0;
-}
-
-/* write a lump of data at a specified offset */
-static int tdb_write(TDB_CONTEXT *tdb, tdb_off off, void *buf, tdb_len len)
-{
-	if (tdb_oob(tdb, off + len, 0) != 0)
-		return -1;
-
-	if (tdb->map_ptr)
-		memcpy(off + (char *)tdb->map_ptr, buf, len);
-#ifdef HAVE_PWRITE
-	else if (pwrite(tdb->fd, buf, len, off) != (ssize_t)len) {
-#else
-	else if (lseek(tdb->fd, off, SEEK_SET) != off
-		 || write(tdb->fd, buf, len) != (ssize_t)len) {
-#endif
-		/* Ensure ecode is set for log fn. */
-		tdb->ecode = TDB_ERR_IO;
-		TDB_LOG((tdb, 0,"tdb_write failed at %d len=%d (%s)\n",
-			   off, len, strerror(errno)));
-		return TDB_ERRCODE(TDB_ERR_IO, -1);
-	}
-	return 0;
-}
-
-/* read a lump of data at a specified offset, maybe convert */
-static int tdb_read(TDB_CONTEXT *tdb,tdb_off off,void *buf,tdb_len len,int cv)
-{
-	if (tdb_oob(tdb, off + len, 0) != 0)
-		return -1;
-
-	if (tdb->map_ptr)
-		memcpy(buf, off + (char *)tdb->map_ptr, len);
-#ifdef HAVE_PREAD
-	else if (pread(tdb->fd, buf, len, off) != (ssize_t)len) {
-#else
-	else if (lseek(tdb->fd, off, SEEK_SET) != off
-		 || read(tdb->fd, buf, len) != (ssize_t)len) {
-#endif
-		/* Ensure ecode is set for log fn. */
-		tdb->ecode = TDB_ERR_IO;
-		TDB_LOG((tdb, 0,"tdb_read failed at %d len=%d (%s)\n",
-			   off, len, strerror(errno)));
-		return TDB_ERRCODE(TDB_ERR_IO, -1);
-	}
-	if (cv)
-		convert(buf, len);
-	return 0;
-}
-
-/* read a lump of data, allocating the space for it */
-static char *tdb_alloc_read(TDB_CONTEXT *tdb, tdb_off offset, tdb_len len)
-{
-	char *buf;
+	tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &seqnum);
+	seqnum++;
+	tdb_ofs_write(tdb, TDB_SEQNUM_OFS, &seqnum);
 
-	if (!(buf = malloc(len))) {
-		/* Ensure ecode is set for log fn. */
-		tdb->ecode = TDB_ERR_OOM;
-		TDB_LOG((tdb, 0,"tdb_alloc_read malloc failed len=%d (%s)\n",
-			   len, strerror(errno)));
-		return TDB_ERRCODE(TDB_ERR_OOM, buf);
-	}
-	if (tdb_read(tdb, offset, buf, len, 0) == -1) {
-		SAFE_FREE(buf);
-		return NULL;
-	}
-	return buf;
-}
-
-/* read/write a tdb_off */
-static int ofs_read(TDB_CONTEXT *tdb, tdb_off offset, tdb_off *d)
-{
-	return tdb_read(tdb, offset, (char*)d, sizeof(*d), DOCONV());
-}
-static int ofs_write(TDB_CONTEXT *tdb, tdb_off offset, tdb_off *d)
-{
-	tdb_off off = *d;
-	return tdb_write(tdb, offset, CONVERT(off), sizeof(*d));
+	tdb_brlock(tdb, TDB_SEQNUM_OFS, F_UNLCK, F_SETLKW, 1);
 }
 
-/* read/write a record */
-static int rec_read(TDB_CONTEXT *tdb, tdb_off offset, struct list_struct *rec)
-{
-	if (tdb_read(tdb, offset, rec, sizeof(*rec),DOCONV()) == -1)
-		return -1;
-	if (TDB_BAD_MAGIC(rec)) {
-		/* Ensure ecode is set for log fn. */
-		tdb->ecode = TDB_ERR_CORRUPT;
-		TDB_LOG((tdb, 0,"rec_read bad magic 0x%x at offset=%d\n", rec->magic, offset));
-		return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
-	}
-	return tdb_oob(tdb, rec->next+sizeof(*rec), 0);
-}
-static int rec_write(TDB_CONTEXT *tdb, tdb_off offset, struct list_struct *rec)
-{
-	struct list_struct r = *rec;
-	return tdb_write(tdb, offset, CONVERT(r), sizeof(r));
-}
-
-/* read a freelist record and check for simple errors */
-static int rec_free_read(TDB_CONTEXT *tdb, tdb_off off, struct list_struct *rec)
-{
-	if (tdb_read(tdb, off, rec, sizeof(*rec),DOCONV()) == -1)
-		return -1;
-
-	if (rec->magic == TDB_MAGIC) {
-		/* this happens when a app is showdown while deleting a record - we should
-		   not completely fail when this happens */
-		TDB_LOG((tdb, 0,"rec_free_read non-free magic 0x%x at offset=%d - fixing\n", 
-			 rec->magic, off));
-		rec->magic = TDB_FREE_MAGIC;
-		if (tdb_write(tdb, off, rec, sizeof(*rec)) == -1)
-			return -1;
-	}
-
-	if (rec->magic != TDB_FREE_MAGIC) {
-		/* Ensure ecode is set for log fn. */
-		tdb->ecode = TDB_ERR_CORRUPT;
-		TDB_LOG((tdb, 0,"rec_free_read bad magic 0x%x at offset=%d\n", 
-			   rec->magic, off));
-		return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
-	}
-	if (tdb_oob(tdb, rec->next+sizeof(*rec), 0) != 0)
-		return -1;
-	return 0;
-}
-
-/* update a record tailer (must hold allocation lock) */
-static int update_tailer(TDB_CONTEXT *tdb, tdb_off offset,
-			 const struct list_struct *rec)
-{
-	tdb_off totalsize;
-
-	/* Offset of tailer from record header */
-	totalsize = sizeof(*rec) + rec->rec_len;
-	return ofs_write(tdb, offset + totalsize - sizeof(tdb_off),
-			 &totalsize);
-}
-
-static tdb_off tdb_dump_record(TDB_CONTEXT *tdb, tdb_off offset)
-{
-	struct list_struct rec;
-	tdb_off tailer_ofs, tailer;
-
-	if (tdb_read(tdb, offset, (char *)&rec, sizeof(rec), DOCONV()) == -1) {
-		printf("ERROR: failed to read record at %u\n", offset);
-		return 0;
-	}
-
-	printf(" rec: offset=%u next=%d rec_len=%d key_len=%d data_len=%d full_hash=0x%x magic=0x%x\n",
-	       offset, rec.next, rec.rec_len, rec.key_len, rec.data_len, rec.full_hash, rec.magic);
-
-	tailer_ofs = offset + sizeof(rec) + rec.rec_len - sizeof(tdb_off);
-	if (ofs_read(tdb, tailer_ofs, &tailer) == -1) {
-		printf("ERROR: failed to read tailer at %u\n", tailer_ofs);
-		return rec.next;
-	}
-
-	if (tailer != rec.rec_len + sizeof(rec)) {
-		printf("ERROR: tailer does not match record! tailer=%u totalsize=%u\n",
-				(unsigned)tailer, (unsigned)(rec.rec_len + sizeof(rec)));
-	}
-	return rec.next;
-}
-
-static int tdb_dump_chain(TDB_CONTEXT *tdb, int i)
-{
-	tdb_off rec_ptr, top;
-	int hash_length = 0;
-
-	top = TDB_HASH_TOP(i);
-
-	if (tdb_lock(tdb, i, F_WRLCK) != 0)
-		return -1;
-
-	if (ofs_read(tdb, top, &rec_ptr) == -1)
-		return tdb_unlock(tdb, i, F_WRLCK);
-
-	if (rec_ptr)
-		printf("hash=%d\n", i);
-
-	while (rec_ptr) {
-		rec_ptr = tdb_dump_record(tdb, rec_ptr);
-		hash_length += 1;
-	}
-
-	printf("chain %d length %d\n", i, hash_length);
-
-	return tdb_unlock(tdb, i, F_WRLCK);
-}
-
-void tdb_dump_all(TDB_CONTEXT *tdb)
-{
-	int i;
-	for (i=0;i<tdb->header.hash_size;i++) {
-		tdb_dump_chain(tdb, i);
-	}
-	tdb_printfreelist(tdb);
-}
-
-int tdb_printfreelist(TDB_CONTEXT *tdb)
-{
-	int ret;
-	long total_free = 0;
-	tdb_off offset, rec_ptr;
-	struct list_struct rec;
-
-	if ((ret = tdb_lock(tdb, -1, F_WRLCK)) != 0)
-		return ret;
-
-	offset = FREELIST_TOP;
-
-	/* read in the freelist top */
-	if (ofs_read(tdb, offset, &rec_ptr) == -1) {
-		tdb_unlock(tdb, -1, F_WRLCK);
-		return 0;
-	}
-
-	printf("freelist top=[0x%08x]\n", rec_ptr );
-	while (rec_ptr) {
-		if (tdb_read(tdb, rec_ptr, (char *)&rec, sizeof(rec), DOCONV()) == -1) {
-			tdb_unlock(tdb, -1, F_WRLCK);
-			return -1;
-		}
-
-		if (rec.magic != TDB_FREE_MAGIC) {
-			printf("bad magic 0x%08x in free list\n", rec.magic);
-			tdb_unlock(tdb, -1, F_WRLCK);
-			return -1;
-		}
-
-		printf("entry offset=[0x%08x], rec.rec_len = [0x%08x (%d)]\n", rec.next, rec.rec_len, rec.rec_len );
-		total_free += rec.rec_len;
-
-		/* move to the next record */
-		rec_ptr = rec.next;
-	}
-	printf("total rec_len = [0x%08x (%d)]\n", (int)total_free, 
-               (int)total_free);
-
-	return tdb_unlock(tdb, -1, F_WRLCK);
-}
-
-/* Remove an element from the freelist.  Must have alloc lock. */
-static int remove_from_freelist(TDB_CONTEXT *tdb, tdb_off off, tdb_off next)
-{
-	tdb_off last_ptr, i;
-
-	/* read in the freelist top */
-	last_ptr = FREELIST_TOP;
-	while (ofs_read(tdb, last_ptr, &i) != -1 && i != 0) {
-		if (i == off) {
-			/* We've found it! */
-			return ofs_write(tdb, last_ptr, &next);
-		}
-		/* Follow chain (next offset is at start of record) */
-		last_ptr = i;
-	}
-	TDB_LOG((tdb, 0,"remove_from_freelist: not on list at off=%d\n", off));
-	return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
-}
-
-/* Add an element into the freelist. Merge adjacent records if
-   neccessary. */
-static int tdb_free(TDB_CONTEXT *tdb, tdb_off offset, struct list_struct *rec)
-{
-	tdb_off right, left;
-
-	/* Allocation and tailer lock */
-	if (tdb_lock(tdb, -1, F_WRLCK) != 0)
-		return -1;
-
-	/* set an initial tailer, so if we fail we don't leave a bogus record */
-	if (update_tailer(tdb, offset, rec) != 0) {
-		TDB_LOG((tdb, 0, "tdb_free: upfate_tailer failed!\n"));
-		goto fail;
-	}
-
-	/* Look right first (I'm an Australian, dammit) */
-	right = offset + sizeof(*rec) + rec->rec_len;
-	if (right + sizeof(*rec) <= tdb->map_size) {
-		struct list_struct r;
-
-		if (tdb_read(tdb, right, &r, sizeof(r), DOCONV()) == -1) {
-			TDB_LOG((tdb, 0, "tdb_free: right read failed at %u\n", right));
-			goto left;
-		}
-
-		/* If it's free, expand to include it. */
-		if (r.magic == TDB_FREE_MAGIC) {
-			if (remove_from_freelist(tdb, right, r.next) == -1) {
-				TDB_LOG((tdb, 0, "tdb_free: right free failed at %u\n", right));
-				goto left;
-			}
-			rec->rec_len += sizeof(r) + r.rec_len;
-		}
-	}
-
-left:
-	/* Look left */
-	left = offset - sizeof(tdb_off);
-	if (left > TDB_DATA_START(tdb->header.hash_size)) {
-		struct list_struct l;
-		tdb_off leftsize;
-		
-		/* Read in tailer and jump back to header */
-		if (ofs_read(tdb, left, &leftsize) == -1) {
-			TDB_LOG((tdb, 0, "tdb_free: left offset read failed at %u\n", left));
-			goto update;
-		}
-		left = offset - leftsize;
-
-		/* Now read in record */
-		if (tdb_read(tdb, left, &l, sizeof(l), DOCONV()) == -1) {
-			TDB_LOG((tdb, 0, "tdb_free: left read failed at %u (%u)\n", left, leftsize));
-			goto update;
-		}
-
-		/* If it's free, expand to include it. */
-		if (l.magic == TDB_FREE_MAGIC) {
-			if (remove_from_freelist(tdb, left, l.next) == -1) {
-				TDB_LOG((tdb, 0, "tdb_free: left free failed at %u\n", left));
-				goto update;
-			} else {
-				offset = left;
-				rec->rec_len += leftsize;
-			}
-		}
-	}
-
-update:
-	if (update_tailer(tdb, offset, rec) == -1) {
-		TDB_LOG((tdb, 0, "tdb_free: update_tailer failed at %u\n", offset));
-		goto fail;
-	}
-
-	/* Now, prepend to free list */
-	rec->magic = TDB_FREE_MAGIC;
-
-	if (ofs_read(tdb, FREELIST_TOP, &rec->next) == -1 ||
-	    rec_write(tdb, offset, rec) == -1 ||
-	    ofs_write(tdb, FREELIST_TOP, &offset) == -1) {
-		TDB_LOG((tdb, 0, "tdb_free record write failed at offset=%d\n", offset));
-		goto fail;
-	}
-
-	/* And we're done. */
-	tdb_unlock(tdb, -1, F_WRLCK);
-	return 0;
-
- fail:
-	tdb_unlock(tdb, -1, F_WRLCK);
-	return -1;
-}
-
-
-/* expand a file.  we prefer to use ftruncate, as that is what posix
-  says to use for mmap expansion */
-static int expand_file(TDB_CONTEXT *tdb, tdb_off size, tdb_off addition)
-{
-	char buf[1024];
-#if HAVE_FTRUNCATE_EXTEND
-	if (ftruncate(tdb->fd, size+addition) != 0) {
-		TDB_LOG((tdb, 0, "expand_file ftruncate to %d failed (%s)\n", 
-			   size+addition, strerror(errno)));
-		return -1;
-	}
-#else
-	char b = 0;
-
-#ifdef HAVE_PWRITE
-	if (pwrite(tdb->fd,  &b, 1, (size+addition) - 1) != 1) {
-#else
-	if (lseek(tdb->fd, (size+addition) - 1, SEEK_SET) != (size+addition) - 1 || 
-	    write(tdb->fd, &b, 1) != 1) {
-#endif
-		TDB_LOG((tdb, 0, "expand_file to %d failed (%s)\n", 
-			   size+addition, strerror(errno)));
-		return -1;
-	}
-#endif
-
-	/* now fill the file with something. This ensures that the file isn't sparse, which would be
-	   very bad if we ran out of disk. This must be done with write, not via mmap */
-	memset(buf, 0x42, sizeof(buf));
-	while (addition) {
-		int n = addition>sizeof(buf)?sizeof(buf):addition;
-#ifdef HAVE_PWRITE
-		int ret = pwrite(tdb->fd, buf, n, size);
-#else
-		int ret;
-		if (lseek(tdb->fd, size, SEEK_SET) != size)
-			return -1;
-		ret = write(tdb->fd, buf, n);
-#endif
-		if (ret != n) {
-			TDB_LOG((tdb, 0, "expand_file write of %d failed (%s)\n", 
-				   n, strerror(errno)));
-			return -1;
-		}
-		addition -= n;
-		size += n;
-	}
-	return 0;
-}
-
-
-/* expand the database at least size bytes by expanding the underlying
-   file and doing the mmap again if necessary */
-static int tdb_expand(TDB_CONTEXT *tdb, tdb_off size)
-{
-	struct list_struct rec;
-	tdb_off offset;
-
-	if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
-		TDB_LOG((tdb, 0, "lock failed in tdb_expand\n"));
-		return -1;
-	}
-
-	/* must know about any previous expansions by another process */
-	tdb_oob(tdb, tdb->map_size + 1, 1);
-
-	/* always make room for at least 10 more records, and round
-           the database up to a multiple of TDB_PAGE_SIZE */
-	size = TDB_ALIGN(tdb->map_size + size*10, TDB_PAGE_SIZE) - tdb->map_size;
-
-	if (!(tdb->flags & TDB_INTERNAL))
-		tdb_munmap(tdb);
-
-	/*
-	 * We must ensure the file is unmapped before doing this
-	 * to ensure consistency with systems like OpenBSD where
-	 * writes and mmaps are not consistent.
-	 */
-
-	/* expand the file itself */
-	if (!(tdb->flags & TDB_INTERNAL)) {
-		if (expand_file(tdb, tdb->map_size, size) != 0)
-			goto fail;
-	}
-
-	tdb->map_size += size;
-
-	if (tdb->flags & TDB_INTERNAL) {
-		char *new_map_ptr = realloc(tdb->map_ptr, tdb->map_size);
-		if (!new_map_ptr) {
-			tdb->map_size -= size;
-			goto fail;
-		}
-		tdb->map_ptr = new_map_ptr;
-	} else {
-		/*
-		 * We must ensure the file is remapped before adding the space
-		 * to ensure consistency with systems like OpenBSD where
-		 * writes and mmaps are not consistent.
-		 */
-
-		/* We're ok if the mmap fails as we'll fallback to read/write */
-		tdb_mmap(tdb);
-	}
-
-	/* form a new freelist record */
-	memset(&rec,'\0',sizeof(rec));
-	rec.rec_len = size - sizeof(rec);
-
-	/* link it into the free list */
-	offset = tdb->map_size - size;
-	if (tdb_free(tdb, offset, &rec) == -1)
-		goto fail;
-
-	tdb_unlock(tdb, -1, F_WRLCK);
-	return 0;
- fail:
-	tdb_unlock(tdb, -1, F_WRLCK);
-	return -1;
-}
-
-/* allocate some space from the free list. The offset returned points
-   to a unconnected list_struct within the database with room for at
-   least length bytes of total data
-
-   0 is returned if the space could not be allocated
- */
-static tdb_off tdb_allocate(TDB_CONTEXT *tdb, tdb_len length,
-			    struct list_struct *rec)
-{
-	tdb_off rec_ptr, last_ptr, newrec_ptr;
-	struct list_struct newrec;
-
-	memset(&newrec, '\0', sizeof(newrec));
-
-	if (tdb_lock(tdb, -1, F_WRLCK) == -1)
-		return 0;
-
-	/* Extra bytes required for tailer */
-	length += sizeof(tdb_off);
-
- again:
-	last_ptr = FREELIST_TOP;
-
-	/* read in the freelist top */
-	if (ofs_read(tdb, FREELIST_TOP, &rec_ptr) == -1)
-		goto fail;
-
-	/* keep looking until we find a freelist record big enough */
-	while (rec_ptr) {
-		if (rec_free_read(tdb, rec_ptr, rec) == -1)
-			goto fail;
-
-		if (rec->rec_len >= length) {
-			/* found it - now possibly split it up  */
-			if (rec->rec_len > length + MIN_REC_SIZE) {
-				/* Length of left piece */
-				length = TDB_ALIGN(length, TDB_ALIGNMENT);
-
-				/* Right piece to go on free list */
-				newrec.rec_len = rec->rec_len
-					- (sizeof(*rec) + length);
-				newrec_ptr = rec_ptr + sizeof(*rec) + length;
-
-				/* And left record is shortened */
-				rec->rec_len = length;
-			} else
-				newrec_ptr = 0;
-
-			/* Remove allocated record from the free list */
-			if (ofs_write(tdb, last_ptr, &rec->next) == -1)
-				goto fail;
-
-			/* Update header: do this before we drop alloc
-                           lock, otherwise tdb_free() might try to
-                           merge with us, thinking we're free.
-                           (Thanks Jeremy Allison). */
-			rec->magic = TDB_MAGIC;
-			if (rec_write(tdb, rec_ptr, rec) == -1)
-				goto fail;
-
-			/* Did we create new block? */
-			if (newrec_ptr) {
-				/* Update allocated record tailer (we
-                                   shortened it). */
-				if (update_tailer(tdb, rec_ptr, rec) == -1)
-					goto fail;
-
-				/* Free new record */
-				if (tdb_free(tdb, newrec_ptr, &newrec) == -1)
-					goto fail;
-			}
-
-			/* all done - return the new record offset */
-			tdb_unlock(tdb, -1, F_WRLCK);
-			return rec_ptr;
-		}
-		/* move to the next record */
-		last_ptr = rec_ptr;
-		rec_ptr = rec->next;
-	}
-	/* we didn't find enough space. See if we can expand the
-	   database and if we can then try again */
-	if (tdb_expand(tdb, length + sizeof(*rec)) == 0)
-		goto again;
- fail:
-	tdb_unlock(tdb, -1, F_WRLCK);
-	return 0;
-}
-
-/* initialise a new database with a specified hash size */
-static int tdb_new_database(TDB_CONTEXT *tdb, int hash_size)
-{
-	struct tdb_header *newdb;
-	int size, ret = -1;
-
-	/* We make it up in memory, then write it out if not internal */
-	size = sizeof(struct tdb_header) + (hash_size+1)*sizeof(tdb_off);
-	if (!(newdb = calloc(size, 1)))
-		return TDB_ERRCODE(TDB_ERR_OOM, -1);
-
-	/* Fill in the header */
-	newdb->version = TDB_VERSION;
-	newdb->hash_size = hash_size;
-#ifdef USE_SPINLOCKS
-	newdb->rwlocks = size;
-#endif
-	if (tdb->flags & TDB_INTERNAL) {
-		tdb->map_size = size;
-		tdb->map_ptr = (char *)newdb;
-		memcpy(&tdb->header, newdb, sizeof(tdb->header));
-		/* Convert the `ondisk' version if asked. */
-		CONVERT(*newdb);
-		return 0;
-	}
-	if (lseek(tdb->fd, 0, SEEK_SET) == -1)
-		goto fail;
-
-	if (ftruncate(tdb->fd, 0) == -1)
-		goto fail;
-
-	/* This creates an endian-converted header, as if read from disk */
-	CONVERT(*newdb);
-	memcpy(&tdb->header, newdb, sizeof(tdb->header));
-	/* Don't endian-convert the magic food! */
-	memcpy(newdb->magic_food, TDB_MAGIC_FOOD, strlen(TDB_MAGIC_FOOD)+1);
-	if (write(tdb->fd, newdb, size) != size)
-		ret = -1;
-	else
-		ret = tdb_create_rwlocks(tdb->fd, hash_size);
-
-  fail:
-	SAFE_FREE(newdb);
-	return ret;
-}
 
 /* Returns 0 on fail.  On success, return offset of record, and fills
    in rec */
-static tdb_off tdb_find(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash,
+static tdb_off_t tdb_find(struct tdb_context *tdb, TDB_DATA key, u32 hash,
 			struct list_struct *r)
 {
-	tdb_off rec_ptr;
+	tdb_off_t rec_ptr;
 	
 	/* read in the hash top */
-	if (ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
+	if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
 		return 0;
 
 	/* keep looking until we find the right record */
 	while (rec_ptr) {
-		if (rec_read(tdb, rec_ptr, r) == -1)
+		if (tdb_rec_read(tdb, rec_ptr, r) == -1)
 			return 0;
 
 		if (!TDB_DEAD(r) && hash==r->full_hash && key.dsize==r->key_len) {
@@ -1040,8 +93,8 @@ static tdb_off tdb_find(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash,
 }
 
 /* As tdb_find, but if you succeed, keep the lock */
-static tdb_off tdb_find_lock_hash(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash, int locktype,
-			     struct list_struct *rec)
+tdb_off_t tdb_find_lock_hash(struct tdb_context *tdb, TDB_DATA key, u32 hash, int locktype,
+			   struct list_struct *rec)
 {
 	u32 rec_ptr;
 
@@ -1052,60 +105,34 @@ static tdb_off tdb_find_lock_hash(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash, int
 	return rec_ptr;
 }
 
-enum TDB_ERROR tdb_error(TDB_CONTEXT *tdb)
-{
-	return tdb->ecode;
-}
-
-static struct tdb_errname {
-	enum TDB_ERROR ecode; const char *estring;
-} emap[] = { {TDB_SUCCESS, "Success"},
-	     {TDB_ERR_CORRUPT, "Corrupt database"},
-	     {TDB_ERR_IO, "IO Error"},
-	     {TDB_ERR_LOCK, "Locking error"},
-	     {TDB_ERR_OOM, "Out of memory"},
-	     {TDB_ERR_EXISTS, "Record exists"},
-	     {TDB_ERR_NOLOCK, "Lock exists on other keys"},
-	     {TDB_ERR_NOEXIST, "Record does not exist"} };
-
-/* Error string for the last tdb error */
-const char *tdb_errorstr(TDB_CONTEXT *tdb)
-{
-	u32 i;
-	for (i = 0; i < sizeof(emap) / sizeof(struct tdb_errname); i++)
-		if (tdb->ecode == emap[i].ecode)
-			return emap[i].estring;
-	return "Invalid error code";
-}
 
 /* update an entry in place - this only works if the new data size
    is <= the old data size and the key exists.
    on failure return -1.
 */
-
-static int tdb_update_hash(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash, TDB_DATA dbuf)
+static int tdb_update_hash(struct tdb_context *tdb, TDB_DATA key, u32 hash, TDB_DATA dbuf)
 {
 	struct list_struct rec;
-	tdb_off rec_ptr;
+	tdb_off_t rec_ptr;
 
 	/* find entry */
 	if (!(rec_ptr = tdb_find(tdb, key, hash, &rec)))
 		return -1;
 
 	/* must be long enough key, data and tailer */
-	if (rec.rec_len < key.dsize + dbuf.dsize + sizeof(tdb_off)) {
+	if (rec.rec_len < key.dsize + dbuf.dsize + sizeof(tdb_off_t)) {
 		tdb->ecode = TDB_SUCCESS; /* Not really an error */
 		return -1;
 	}
 
-	if (tdb_write(tdb, rec_ptr + sizeof(rec) + rec.key_len,
+	if (tdb->methods->tdb_write(tdb, rec_ptr + sizeof(rec) + rec.key_len,
 		      dbuf.dptr, dbuf.dsize) == -1)
 		return -1;
 
 	if (dbuf.dsize != rec.data_len) {
 		/* update size */
 		rec.data_len = dbuf.dsize;
-		return rec_write(tdb, rec_ptr, &rec);
+		return tdb_rec_write(tdb, rec_ptr, &rec);
 	}
  
 	return 0;
@@ -1114,13 +141,12 @@ static int tdb_update_hash(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash, TDB_DATA db
 /* find an entry in the database given a key */
 /* If an entry doesn't exist tdb_err will be set to
  * TDB_ERR_NOEXIST. If a key has no data attached
- * tdb_err will not be set. Both will return a
- * zero pptr and zero dsize.
+ * then the TDB_DATA will have zero length but
+ * a non-zero pointer
  */
-
-TDB_DATA tdb_fetch(TDB_CONTEXT *tdb, TDB_DATA key)
+TDB_DATA tdb_fetch(struct tdb_context *tdb, TDB_DATA key)
 {
-	tdb_off rec_ptr;
+	tdb_off_t rec_ptr;
 	struct list_struct rec;
 	TDB_DATA ret;
 	u32 hash;
@@ -1130,11 +156,8 @@ TDB_DATA tdb_fetch(TDB_CONTEXT *tdb, TDB_DATA key)
 	if (!(rec_ptr = tdb_find_lock_hash(tdb,key,hash,F_RDLCK,&rec)))
 		return tdb_null;
 
-	if (rec.data_len)
-		ret.dptr = tdb_alloc_read(tdb, rec_ptr + sizeof(rec) + rec.key_len,
-					  rec.data_len);
-	else
-		ret.dptr = NULL;
+	ret.dptr = tdb_alloc_read(tdb, rec_ptr + sizeof(rec) + rec.key_len,
+				  rec.data_len);
 	ret.dsize = rec.data_len;
 	tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
 	return ret;
@@ -1146,7 +169,7 @@ TDB_DATA tdb_fetch(TDB_CONTEXT *tdb, TDB_DATA key)
    this doesn't match the conventions in the rest of this module, but is
    compatible with gdbm
 */
-static int tdb_exists_hash(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash)
+static int tdb_exists_hash(struct tdb_context *tdb, TDB_DATA key, u32 hash)
 {
 	struct list_struct rec;
 	
@@ -1156,82 +179,39 @@ static int tdb_exists_hash(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash)
 	return 1;
 }
 
-int tdb_exists(TDB_CONTEXT *tdb, TDB_DATA key)
+int tdb_exists(struct tdb_context *tdb, TDB_DATA key)
 {
 	u32 hash = tdb->hash_fn(&key);
 	return tdb_exists_hash(tdb, key, hash);
 }
 
-/* record lock stops delete underneath */
-static int lock_record(TDB_CONTEXT *tdb, tdb_off off)
-{
-	return off ? tdb_brlock(tdb, off, F_RDLCK, F_SETLKW, 0) : 0;
-}
-/*
-  Write locks override our own fcntl readlocks, so check it here.
-  Note this is meant to be F_SETLK, *not* F_SETLKW, as it's not
-  an error to fail to get the lock here.
-*/
- 
-static int write_lock_record(TDB_CONTEXT *tdb, tdb_off off)
-{
-	struct tdb_traverse_lock *i;
-	for (i = &tdb->travlocks; i; i = i->next)
-		if (i->off == off)
-			return -1;
-	return tdb_brlock(tdb, off, F_WRLCK, F_SETLK, 1);
-}
-
-/*
-  Note this is meant to be F_SETLK, *not* F_SETLKW, as it's not
-  an error to fail to get the lock here.
-*/
-
-static int write_unlock_record(TDB_CONTEXT *tdb, tdb_off off)
-{
-	return tdb_brlock(tdb, off, F_UNLCK, F_SETLK, 0);
-}
-/* fcntl locks don't stack: avoid unlocking someone else's */
-static int unlock_record(TDB_CONTEXT *tdb, tdb_off off)
-{
-	struct tdb_traverse_lock *i;
-	u32 count = 0;
-
-	if (off == 0)
-		return 0;
-	for (i = &tdb->travlocks; i; i = i->next)
-		if (i->off == off)
-			count++;
-	return (count == 1 ? tdb_brlock(tdb, off, F_UNLCK, F_SETLKW, 0) : 0);
-}
-
 /* actually delete an entry in the database given the offset */
-static int do_delete(TDB_CONTEXT *tdb, tdb_off rec_ptr, struct list_struct*rec)
+int tdb_do_delete(struct tdb_context *tdb, tdb_off_t rec_ptr, struct list_struct*rec)
 {
-	tdb_off last_ptr, i;
+	tdb_off_t last_ptr, i;
 	struct list_struct lastrec;
 
-	if (tdb->read_only) return -1;
+	if (tdb->read_only || tdb->traverse_read) return -1;
 
-	if (write_lock_record(tdb, rec_ptr) == -1) {
+	if (tdb_write_lock_record(tdb, rec_ptr) == -1) {
 		/* Someone traversing here: mark it as dead */
 		rec->magic = TDB_DEAD_MAGIC;
-		return rec_write(tdb, rec_ptr, rec);
+		return tdb_rec_write(tdb, rec_ptr, rec);
 	}
-	if (write_unlock_record(tdb, rec_ptr) != 0)
+	if (tdb_write_unlock_record(tdb, rec_ptr) != 0)
 		return -1;
 
 	/* find previous record in hash chain */
-	if (ofs_read(tdb, TDB_HASH_TOP(rec->full_hash), &i) == -1)
+	if (tdb_ofs_read(tdb, TDB_HASH_TOP(rec->full_hash), &i) == -1)
 		return -1;
 	for (last_ptr = 0; i != rec_ptr; last_ptr = i, i = lastrec.next)
-		if (rec_read(tdb, i, &lastrec) == -1)
+		if (tdb_rec_read(tdb, i, &lastrec) == -1)
 			return -1;
 
 	/* unlink it: next ptr is at start of record. */
 	if (last_ptr == 0)
 		last_ptr = TDB_HASH_TOP(rec->full_hash);
-	if (ofs_write(tdb, last_ptr, &rec->next) == -1)
+	if (tdb_ofs_write(tdb, last_ptr, &rec->next) == -1)
 		return -1;
 
 	/* recover the space */
@@ -1240,283 +220,27 @@ static int do_delete(TDB_CONTEXT *tdb, tdb_off rec_ptr, struct list_struct*rec)
 	return 0;
 }
 
-/* Uses traverse lock: 0 = finish, -1 = error, other = record offset */
-static int tdb_next_lock(TDB_CONTEXT *tdb, struct tdb_traverse_lock *tlock,
-			 struct list_struct *rec)
-{
-	int want_next = (tlock->off != 0);
-
-	/* Lock each chain from the start one. */
-	for (; tlock->hash < tdb->header.hash_size; tlock->hash++) {
-
-		/* this is an optimisation for the common case where
-		   the hash chain is empty, which is particularly
-		   common for the use of tdb with ldb, where large
-		   hashes are used. In that case we spend most of our
-		   time in tdb_brlock(), locking empty hash chains.
-
-		   To avoid this, we do an unlocked pre-check to see
-		   if the hash chain is empty before starting to look
-		   inside it. If it is empty then we can avoid that
-		   hash chain. If it isn't empty then we can't believe
-		   the value we get back, as we read it without a
-		   lock, so instead we get the lock and re-fetch the
-		   value below.
-
-		   Notice that not doing this optimisation on the
-		   first hash chain is critical. We must guarantee
-		   that we have done at least one fcntl lock at the
-		   start of a search to guarantee that memory is
-		   coherent on SMP systems. If records are added by
-		   others during the search then thats OK, and we
-		   could possibly miss those with this trick, but we
-		   could miss them anyway without this trick, so the
-		   semantics don't change.
-
-		   With a non-indexed ldb search this trick gains us a
-		   factor of around 80 in speed on a linux 2.6.x
-		   system (testing using ldbtest).
-		 */
-		if (!tlock->off && tlock->hash != 0) {
-			u32 off;
-			if (tdb->map_ptr) {
-				for (;tlock->hash < tdb->header.hash_size;tlock->hash++) {
-					if (0 != *(u32 *)(TDB_HASH_TOP(tlock->hash) + (unsigned char *)tdb->map_ptr)) {
-						break;
-					}
-				}
-				if (tlock->hash == tdb->header.hash_size) {
-					continue;
-				}
-			} else {
-				if (ofs_read(tdb, TDB_HASH_TOP(tlock->hash), &off) == 0 &&
-				    off == 0) {
-					continue;
-				}
-			}
-		}
-
-		if (tdb_lock(tdb, tlock->hash, F_WRLCK) == -1)
-			return -1;
-
-		/* No previous record?  Start at top of chain. */
-		if (!tlock->off) {
-			if (ofs_read(tdb, TDB_HASH_TOP(tlock->hash),
-				     &tlock->off) == -1)
-				goto fail;
-		} else {
-			/* Otherwise unlock the previous record. */
-			if (unlock_record(tdb, tlock->off) != 0)
-				goto fail;
-		}
-
-		if (want_next) {
-			/* We have offset of old record: grab next */
-			if (rec_read(tdb, tlock->off, rec) == -1)
-				goto fail;
-			tlock->off = rec->next;
-		}
-
-		/* Iterate through chain */
-		while( tlock->off) {
-			tdb_off current;
-			if (rec_read(tdb, tlock->off, rec) == -1)
-				goto fail;
-
-			/* Detect infinite loops. From "Shlomi Yaakobovich" <Shlomi@exanet.com>. */
-			if (tlock->off == rec->next) {
-				TDB_LOG((tdb, 0, "tdb_next_lock: loop detected.\n"));
-				goto fail;
-			}
-
-			if (!TDB_DEAD(rec)) {
-				/* Woohoo: we found one! */
-				if (lock_record(tdb, tlock->off) != 0)
-					goto fail;
-				return tlock->off;
-			}
-
-			/* Try to clean dead ones from old traverses */
-			current = tlock->off;
-			tlock->off = rec->next;
-			if (!tdb->read_only && 
-			    do_delete(tdb, current, rec) != 0)
-				goto fail;
-		}
-		tdb_unlock(tdb, tlock->hash, F_WRLCK);
-		want_next = 0;
-	}
-	/* We finished iteration without finding anything */
-	return TDB_ERRCODE(TDB_SUCCESS, 0);
-
- fail:
-	tlock->off = 0;
-	if (tdb_unlock(tdb, tlock->hash, F_WRLCK) != 0)
-		TDB_LOG((tdb, 0, "tdb_next_lock: On error unlock failed!\n"));
-	return -1;
-}
-
-/* traverse the entire database - calling fn(tdb, key, data) on each element.
-   return -1 on error or the record count traversed
-   if fn is NULL then it is not called
-   a non-zero return value from fn() indicates that the traversal should stop
-  */
-int tdb_traverse(TDB_CONTEXT *tdb, tdb_traverse_func fn, void *private_val)
-{
-	TDB_DATA key, dbuf;
-	struct list_struct rec;
-	struct tdb_traverse_lock tl = { NULL, 0, 0 };
-	int ret, count = 0;
-
-	/* This was in the initializaton, above, but the IRIX compiler
-	 * did not like it.  crh
-	 */
-	tl.next = tdb->travlocks.next;
-
-	/* fcntl locks don't stack: beware traverse inside traverse */
-	tdb->travlocks.next = &tl;
-
-	/* tdb_next_lock places locks on the record returned, and its chain */
-	while ((ret = tdb_next_lock(tdb, &tl, &rec)) > 0) {
-		count++;
-		/* now read the full record */
-		key.dptr = tdb_alloc_read(tdb, tl.off + sizeof(rec), 
-					  rec.key_len + rec.data_len);
-		if (!key.dptr) {
-			ret = -1;
-			if (tdb_unlock(tdb, tl.hash, F_WRLCK) != 0)
-				goto out;
-			if (unlock_record(tdb, tl.off) != 0)
-				TDB_LOG((tdb, 0, "tdb_traverse: key.dptr == NULL and unlock_record failed!\n"));
-			goto out;
-		}
-		key.dsize = rec.key_len;
-		dbuf.dptr = key.dptr + rec.key_len;
-		dbuf.dsize = rec.data_len;
-
-		/* Drop chain lock, call out */
-		if (tdb_unlock(tdb, tl.hash, F_WRLCK) != 0) {
-			ret = -1;
-			SAFE_FREE(key.dptr);
-			goto out;
-		}
-		if (fn && fn(tdb, key, dbuf, private_val)) {
-			/* They want us to terminate traversal */
-			ret = count;
-			if (unlock_record(tdb, tl.off) != 0) {
-				TDB_LOG((tdb, 0, "tdb_traverse: unlock_record failed!\n"));;
-				ret = -1;
-			}
-			SAFE_FREE(key.dptr);
-			goto out;
-		}
-		SAFE_FREE(key.dptr);
-	}
-out:
-	tdb->travlocks.next = tl.next;
-	if (ret < 0)
-		return -1;
-	else
-		return count;
-}
-
-/* find the first entry in the database and return its key */
-TDB_DATA tdb_firstkey(TDB_CONTEXT *tdb)
-{
-	TDB_DATA key;
-	struct list_struct rec;
-
-	/* release any old lock */
-	if (unlock_record(tdb, tdb->travlocks.off) != 0)
-		return tdb_null;
-	tdb->travlocks.off = tdb->travlocks.hash = 0;
-
-	if (tdb_next_lock(tdb, &tdb->travlocks, &rec) <= 0)
-		return tdb_null;
-	/* now read the key */
-	key.dsize = rec.key_len;
-	key.dptr =tdb_alloc_read(tdb,tdb->travlocks.off+sizeof(rec),key.dsize);
-	if (tdb_unlock(tdb, BUCKET(tdb->travlocks.hash), F_WRLCK) != 0)
-		TDB_LOG((tdb, 0, "tdb_firstkey: error occurred while tdb_unlocking!\n"));
-	return key;
-}
-
-/* find the next entry in the database, returning its key */
-TDB_DATA tdb_nextkey(TDB_CONTEXT *tdb, TDB_DATA oldkey)
-{
-	u32 oldhash;
-	TDB_DATA key = tdb_null;
-	struct list_struct rec;
-	char *k = NULL;
-
-	/* Is locked key the old key?  If so, traverse will be reliable. */
-	if (tdb->travlocks.off) {
-		if (tdb_lock(tdb,tdb->travlocks.hash,F_WRLCK))
-			return tdb_null;
-		if (rec_read(tdb, tdb->travlocks.off, &rec) == -1
-		    || !(k = tdb_alloc_read(tdb,tdb->travlocks.off+sizeof(rec),
-					    rec.key_len))
-		    || memcmp(k, oldkey.dptr, oldkey.dsize) != 0) {
-			/* No, it wasn't: unlock it and start from scratch */
-			if (unlock_record(tdb, tdb->travlocks.off) != 0) {
-				SAFE_FREE(k);
-				return tdb_null;
-			}
-			if (tdb_unlock(tdb, tdb->travlocks.hash, F_WRLCK) != 0) {
-				SAFE_FREE(k);
-				return tdb_null;
-			}
-			tdb->travlocks.off = 0;
-		}
-
-		SAFE_FREE(k);
-	}
-
-	if (!tdb->travlocks.off) {
-		/* No previous element: do normal find, and lock record */
-		tdb->travlocks.off = tdb_find_lock_hash(tdb, oldkey, tdb->hash_fn(&oldkey), F_WRLCK, &rec);
-		if (!tdb->travlocks.off)
-			return tdb_null;
-		tdb->travlocks.hash = BUCKET(rec.full_hash);
-		if (lock_record(tdb, tdb->travlocks.off) != 0) {
-			TDB_LOG((tdb, 0, "tdb_nextkey: lock_record failed (%s)!\n", strerror(errno)));
-			return tdb_null;
-		}
-	}
-	oldhash = tdb->travlocks.hash;
-
-	/* Grab next record: locks chain and returned record,
-	   unlocks old record */
-	if (tdb_next_lock(tdb, &tdb->travlocks, &rec) > 0) {
-		key.dsize = rec.key_len;
-		key.dptr = tdb_alloc_read(tdb, tdb->travlocks.off+sizeof(rec),
-					  key.dsize);
-		/* Unlock the chain of this new record */
-		if (tdb_unlock(tdb, tdb->travlocks.hash, F_WRLCK) != 0)
-			TDB_LOG((tdb, 0, "tdb_nextkey: WARNING tdb_unlock failed!\n"));
-	}
-	/* Unlock the chain of old record */
-	if (tdb_unlock(tdb, BUCKET(oldhash), F_WRLCK) != 0)
-		TDB_LOG((tdb, 0, "tdb_nextkey: WARNING tdb_unlock failed!\n"));
-	return key;
-}
-
 /* delete an entry in the database given a key */
-static int tdb_delete_hash(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash)
+static int tdb_delete_hash(struct tdb_context *tdb, TDB_DATA key, u32 hash)
 {
-	tdb_off rec_ptr;
+	tdb_off_t rec_ptr;
 	struct list_struct rec;
 	int ret;
 
 	if (!(rec_ptr = tdb_find_lock_hash(tdb, key, hash, F_WRLCK, &rec)))
 		return -1;
-	ret = do_delete(tdb, rec_ptr, &rec);
+	ret = tdb_do_delete(tdb, rec_ptr, &rec);
+
+	if (ret == 0) {
+		tdb_increment_seqnum(tdb);
+	}
+
 	if (tdb_unlock(tdb, BUCKET(rec.full_hash), F_WRLCK) != 0)
 		TDB_LOG((tdb, 0, "tdb_delete: WARNING tdb_unlock failed!\n"));
 	return ret;
 }
 
-int tdb_delete(TDB_CONTEXT *tdb, TDB_DATA key)
+int tdb_delete(struct tdb_context *tdb, TDB_DATA key)
 {
 	u32 hash = tdb->hash_fn(&key);
 	return tdb_delete_hash(tdb, key, hash);
@@ -1527,14 +251,19 @@ int tdb_delete(TDB_CONTEXT *tdb, TDB_DATA key)
 
    return 0 on success, -1 on failure
 */
-int tdb_store(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA dbuf, int flag)
+int tdb_store(struct tdb_context *tdb, TDB_DATA key, TDB_DATA dbuf, int flag)
 {
 	struct list_struct rec;
 	u32 hash;
-	tdb_off rec_ptr;
+	tdb_off_t rec_ptr;
 	char *p = NULL;
 	int ret = 0;
 
+	if (tdb->read_only || tdb->traverse_read) {
+		tdb->ecode = TDB_ERR_RDONLY;
+		return -1;
+	}
+
 	/* find which hash bucket it is in */
 	hash = tdb->hash_fn(&key);
 	if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
@@ -1555,7 +284,7 @@ int tdb_store(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA dbuf, int flag)
 			/* if the record doesn't exist and we are in TDB_MODIFY mode then
 			 we should fail the store */
 			goto fail;
-	}
+		}
 	}
 	/* reset the error code potentially set by the tdb_update() */
 	tdb->ecode = TDB_SUCCESS;
@@ -1583,7 +312,7 @@ int tdb_store(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA dbuf, int flag)
 		goto fail;
 
 	/* Read hash top into next ptr */
-	if (ofs_read(tdb, TDB_HASH_TOP(hash), &rec.next) == -1)
+	if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec.next) == -1)
 		goto fail;
 
 	rec.key_len = key.dsize;
@@ -1592,12 +321,15 @@ int tdb_store(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA dbuf, int flag)
 	rec.magic = TDB_MAGIC;
 
 	/* write out and point the top of the hash chain at it */
-	if (rec_write(tdb, rec_ptr, &rec) == -1
-	    || tdb_write(tdb, rec_ptr+sizeof(rec), p, key.dsize+dbuf.dsize)==-1
-	    || ofs_write(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1) {
+	if (tdb_rec_write(tdb, rec_ptr, &rec) == -1
+	    || tdb->methods->tdb_write(tdb, rec_ptr+sizeof(rec), p, key.dsize+dbuf.dsize)==-1
+	    || tdb_ofs_write(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1) {
 		/* Need to tdb_unallocate() here */
 		goto fail;
 	}
+
+	tdb_increment_seqnum(tdb);
+
  out:
 	SAFE_FREE(p); 
 	tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
@@ -1607,521 +339,92 @@ fail:
 	goto out;
 }
 
-/* Attempt to append data to an entry in place - this only works if the new data size
-   is <= the old data size and the key exists.
-   on failure return -1. Record must be locked before calling.
-*/
-static int tdb_append_inplace(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash, TDB_DATA new_dbuf)
-{
-	struct list_struct rec;
-	tdb_off rec_ptr;
-
-	/* find entry */
-	if (!(rec_ptr = tdb_find(tdb, key, hash, &rec)))
-		return -1;
-
-	/* Append of 0 is always ok. */
-	if (new_dbuf.dsize == 0)
-		return 0;
-
-	/* must be long enough for key, old data + new data and tailer */
-	if (rec.rec_len < key.dsize + rec.data_len + new_dbuf.dsize + sizeof(tdb_off)) {
-		/* No room. */
-		tdb->ecode = TDB_SUCCESS; /* Not really an error */
-		return -1;
-	}
-
-	if (tdb_write(tdb, rec_ptr + sizeof(rec) + rec.key_len + rec.data_len,
-		      new_dbuf.dptr, new_dbuf.dsize) == -1)
-		return -1;
-
-	/* update size */
-	rec.data_len += new_dbuf.dsize;
-	return rec_write(tdb, rec_ptr, &rec);
-}
 
 /* Append to an entry. Create if not exist. */
-
-int tdb_append(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA new_dbuf)
+int tdb_append(struct tdb_context *tdb, TDB_DATA key, TDB_DATA new_dbuf)
 {
-	struct list_struct rec;
 	u32 hash;
-	tdb_off rec_ptr;
-	char *p = NULL;
-	int ret = 0;
-	size_t new_data_size = 0;
+	TDB_DATA dbuf;
+	int ret = -1;
 
 	/* find which hash bucket it is in */
 	hash = tdb->hash_fn(&key);
 	if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
 		return -1;
 
-	/* first try in-place. */
-	if (tdb_append_inplace(tdb, key, hash, new_dbuf) == 0)
-		goto out;
+	dbuf = tdb_fetch(tdb, key);
 
-	/* reset the error code potentially set by the tdb_append_inplace() */
-	tdb->ecode = TDB_SUCCESS;
-
-	/* find entry */
-	if (!(rec_ptr = tdb_find(tdb, key, hash, &rec))) {
-		if (tdb->ecode != TDB_ERR_NOEXIST)
-			goto fail;
-
-		/* Not found - create. */
-
-		ret = tdb_store(tdb, key, new_dbuf, TDB_INSERT);
-		goto out;
+	if (dbuf.dptr == NULL) {
+		dbuf.dptr = malloc(new_dbuf.dsize);
+	} else {
+		dbuf.dptr = realloc(dbuf.dptr, dbuf.dsize + new_dbuf.dsize);
 	}
 
-	new_data_size = rec.data_len + new_dbuf.dsize;
-
-	/* Copy key+old_value+value *before* allocating free space in case malloc
-	   fails and we are left with a dead spot in the tdb. */
-
-	if (!(p = (char *)malloc(key.dsize + new_data_size))) {
+	if (dbuf.dptr == NULL) {
 		tdb->ecode = TDB_ERR_OOM;
-		goto fail;
-	}
-
-	/* Copy the key in place. */
-	memcpy(p, key.dptr, key.dsize);
-
-	/* Now read the old data into place. */
-	if (rec.data_len &&
-		tdb_read(tdb, rec_ptr + sizeof(rec) + rec.key_len, p + key.dsize, rec.data_len, 0) == -1)
-			goto fail;
-
-	/* Finally append the new data. */
-	if (new_dbuf.dsize)
-		memcpy(p+key.dsize+rec.data_len, new_dbuf.dptr, new_dbuf.dsize);
-
-	/* delete any existing record - if it doesn't exist we don't
-           care.  Doing this first reduces fragmentation, and avoids
-           coalescing with `allocated' block before it's updated. */
-
-	tdb_delete_hash(tdb, key, hash);
-
-	if (!(rec_ptr = tdb_allocate(tdb, key.dsize + new_data_size, &rec)))
-		goto fail;
-
-	/* Read hash top into next ptr */
-	if (ofs_read(tdb, TDB_HASH_TOP(hash), &rec.next) == -1)
-		goto fail;
-
-	rec.key_len = key.dsize;
-	rec.data_len = new_data_size;
-	rec.full_hash = hash;
-	rec.magic = TDB_MAGIC;
-
-	/* write out and point the top of the hash chain at it */
-	if (rec_write(tdb, rec_ptr, &rec) == -1
-	    || tdb_write(tdb, rec_ptr+sizeof(rec), p, key.dsize+new_data_size)==-1
-	    || ofs_write(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1) {
-		/* Need to tdb_unallocate() here */
-		goto fail;
+		goto failed;
 	}
 
- out:
-	SAFE_FREE(p); 
-	tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
-	return ret;
-
-fail:
-	ret = -1;
-	goto out;
-}
+	memcpy(dbuf.dptr + dbuf.dsize, new_dbuf.dptr, new_dbuf.dsize);
+	dbuf.dsize += new_dbuf.dsize;
 
-static int tdb_already_open(dev_t device,
-			    ino_t ino)
-{
-	TDB_CONTEXT *i;
+	ret = tdb_store(tdb, key, dbuf, 0);
 	
-	for (i = tdbs; i; i = i->next) {
-		if (i->device == device && i->inode == ino) {
-			return 1;
-		}
-	}
-
-	return 0;
-}
-
-/* This is based on the hash algorithm from gdbm */
-static u32 default_tdb_hash(TDB_DATA *key)
-{
-	u32 value;	/* Used to compute the hash value.  */
-	u32   i;	/* Used to cycle through random values. */
-
-	/* Set the initial value from the key size. */
-	for (value = 0x238F13AF * key->dsize, i=0; i < key->dsize; i++)
-		value = (value + (key->dptr[i] << (i*5 % 24)));
-
-	return (1103515243 * value + 12345);  
-}
-
-/* open the database, creating it if necessary 
-
-   The open_flags and mode are passed straight to the open call on the
-   database file. A flags value of O_WRONLY is invalid. The hash size
-   is advisory, use zero for a default value.
-
-   Return is NULL on error, in which case errno is also set.  Don't 
-   try to call tdb_error or tdb_errname, just do strerror(errno).
-
-   @param name may be NULL for internal databases. */
-TDB_CONTEXT *tdb_open(const char *name, int hash_size, int tdb_flags,
-		      int open_flags, mode_t mode)
-{
-	return tdb_open_ex(name, hash_size, tdb_flags, open_flags, mode, NULL, NULL);
-}
-
-
-TDB_CONTEXT *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
-			 int open_flags, mode_t mode,
-			 tdb_log_func log_fn,
-			 tdb_hash_func hash_fn)
-{
-	TDB_CONTEXT *tdb;
-	struct stat st;
-	int rev = 0, locked = 0;
-	unsigned char *vp;
-	u32 vertest;
-
-	if (!(tdb = calloc(1, sizeof *tdb))) {
-		/* Can't log this */
-		errno = ENOMEM;
-		goto fail;
-	}
-	tdb->fd = -1;
-	tdb->name = NULL;
-	tdb->map_ptr = NULL;
-	tdb->flags = tdb_flags;
-	tdb->open_flags = open_flags;
-	tdb->log_fn = log_fn;
-	tdb->hash_fn = hash_fn ? hash_fn : default_tdb_hash;
-
-	if ((open_flags & O_ACCMODE) == O_WRONLY) {
-		TDB_LOG((tdb, 0, "tdb_open_ex: can't open tdb %s write-only\n",
-			 name));
-		errno = EINVAL;
-		goto fail;
-	}
-	
-	if (hash_size == 0)
-		hash_size = DEFAULT_HASH_SIZE;
-	if ((open_flags & O_ACCMODE) == O_RDONLY) {
-		tdb->read_only = 1;
-		/* read only databases don't do locking or clear if first */
-		tdb->flags |= TDB_NOLOCK;
-		tdb->flags &= ~TDB_CLEAR_IF_FIRST;
-	}
-
-	/* internal databases don't mmap or lock, and start off cleared */
-	if (tdb->flags & TDB_INTERNAL) {
-		tdb->flags |= (TDB_NOLOCK | TDB_NOMMAP);
-		tdb->flags &= ~TDB_CLEAR_IF_FIRST;
-		if (tdb_new_database(tdb, hash_size) != 0) {
-			TDB_LOG((tdb, 0, "tdb_open_ex: tdb_new_database failed!"));
-			goto fail;
-		}
-		goto internal;
-	}
-
-	if ((tdb->fd = open(name, open_flags, mode)) == -1) {
-		TDB_LOG((tdb, 5, "tdb_open_ex: could not open file %s: %s\n",
-			 name, strerror(errno)));
-		goto fail;	/* errno set by open(2) */
-	}
-
-	/* ensure there is only one process initialising at once */
-	if (tdb_brlock(tdb, GLOBAL_LOCK, F_WRLCK, F_SETLKW, 0) == -1) {
-		TDB_LOG((tdb, 0, "tdb_open_ex: failed to get global lock on %s: %s\n",
-			 name, strerror(errno)));
-		goto fail;	/* errno set by tdb_brlock */
-	}
-
-	/* we need to zero database if we are the only one with it open */
-	if ((tdb_flags & TDB_CLEAR_IF_FIRST) &&
-		(locked = (tdb_brlock(tdb, ACTIVE_LOCK, F_WRLCK, F_SETLK, 0) == 0))) {
-		open_flags |= O_CREAT;
-		if (ftruncate(tdb->fd, 0) == -1) {
-			TDB_LOG((tdb, 0, "tdb_open_ex: "
-				 "failed to truncate %s: %s\n",
-				 name, strerror(errno)));
-			goto fail; /* errno set by ftruncate */
-		}
-	}
-
-	if (read(tdb->fd, &tdb->header, sizeof(tdb->header)) != sizeof(tdb->header)
-	    || strcmp(tdb->header.magic_food, TDB_MAGIC_FOOD) != 0
-	    || (tdb->header.version != TDB_VERSION
-		&& !(rev = (tdb->header.version==TDB_BYTEREV(TDB_VERSION))))) {
-		/* its not a valid database - possibly initialise it */
-		if (!(open_flags & O_CREAT) || tdb_new_database(tdb, hash_size) == -1) {
-			errno = EIO; /* ie bad format or something */
-			goto fail;
-		}
-		rev = (tdb->flags & TDB_CONVERT);
-	}
-	vp = (unsigned char *)&tdb->header.version;
-	vertest = (((u32)vp[0]) << 24) | (((u32)vp[1]) << 16) |
-		  (((u32)vp[2]) << 8) | (u32)vp[3];
-	tdb->flags |= (vertest==TDB_VERSION) ? TDB_BIGENDIAN : 0;
-	if (!rev)
-		tdb->flags &= ~TDB_CONVERT;
-	else {
-		tdb->flags |= TDB_CONVERT;
-		convert(&tdb->header, sizeof(tdb->header));
-	}
-	if (fstat(tdb->fd, &st) == -1)
-		goto fail;
-
-	/* Is it already in the open list?  If so, fail. */
-	if (tdb_already_open(st.st_dev, st.st_ino)) {
-		TDB_LOG((tdb, 2, "tdb_open_ex: "
-			 "%s (%d,%d) is already open in this process\n",
-			 name, (int)st.st_dev, (int)st.st_ino));
-		errno = EBUSY;
-		goto fail;
-	}
-
-	if (!(tdb->name = (char *)strdup(name))) {
-		errno = ENOMEM;
-		goto fail;
-	}
-
-	tdb->map_size = st.st_size;
-	tdb->device = st.st_dev;
-	tdb->inode = st.st_ino;
-	tdb->locked = calloc(tdb->header.hash_size+1, sizeof(tdb->locked[0]));
-	if (!tdb->locked) {
-		TDB_LOG((tdb, 2, "tdb_open_ex: "
-			 "failed to allocate lock structure for %s\n",
-			 name));
-		errno = ENOMEM;
-		goto fail;
-	}
-	tdb_mmap(tdb);
-	if (locked) {
-		if (!tdb->read_only)
-			if (tdb_clear_spinlocks(tdb) != 0) {
-				TDB_LOG((tdb, 0, "tdb_open_ex: "
-				"failed to clear spinlock\n"));
-				goto fail;
-			}
-		if (tdb_brlock(tdb, ACTIVE_LOCK, F_UNLCK, F_SETLK, 0) == -1) {
-			TDB_LOG((tdb, 0, "tdb_open_ex: "
-				 "failed to take ACTIVE_LOCK on %s: %s\n",
-				 name, strerror(errno)));
-			goto fail;
-		}
-
-	}
-
-	/* We always need to do this if the CLEAR_IF_FIRST flag is set, even if
-	   we didn't get the initial exclusive lock as we need to let all other
-	   users know we're using it. */
-
-	if (tdb_flags & TDB_CLEAR_IF_FIRST) {
-		/* leave this lock in place to indicate it's in use */
-		if (tdb_brlock(tdb, ACTIVE_LOCK, F_RDLCK, F_SETLKW, 0) == -1)
-			goto fail;
-	}
-
-
- internal:
-	/* Internal (memory-only) databases skip all the code above to
-	 * do with disk files, and resume here by releasing their
-	 * global lock and hooking into the active list. */
-	if (tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0) == -1)
-		goto fail;
-	tdb->next = tdbs;
-	tdbs = tdb;
-	return tdb;
-
- fail:
-	{ int save_errno = errno;
-
-	if (!tdb)
-		return NULL;
-	
-	if (tdb->map_ptr) {
-		if (tdb->flags & TDB_INTERNAL)
-			SAFE_FREE(tdb->map_ptr);
-		else
-			tdb_munmap(tdb);
-	}
-	SAFE_FREE(tdb->name);
-	if (tdb->fd != -1)
-		if (close(tdb->fd) != 0)
-			TDB_LOG((tdb, 5, "tdb_open_ex: failed to close tdb->fd on error!\n"));
-	SAFE_FREE(tdb->locked);
-	SAFE_FREE(tdb);
-	errno = save_errno;
-	return NULL;
-	}
-}
-
-/**
- * Close a database.
- *
- * @returns -1 for error; 0 for success.
- **/
-int tdb_close(TDB_CONTEXT *tdb)
-{
-	TDB_CONTEXT **i;
-	int ret = 0;
-
-	if (tdb->map_ptr) {
-		if (tdb->flags & TDB_INTERNAL)
-			SAFE_FREE(tdb->map_ptr);
-		else
-			tdb_munmap(tdb);
-	}
-	SAFE_FREE(tdb->name);
-	if (tdb->fd != -1)
-		ret = close(tdb->fd);
-	SAFE_FREE(tdb->locked);
-
-	/* Remove from contexts list */
-	for (i = &tdbs; *i; i = &(*i)->next) {
-		if (*i == tdb) {
-			*i = tdb->next;
-			break;
-		}
-	}
-
-	memset(tdb, 0, sizeof(*tdb));
-	SAFE_FREE(tdb);
-
+failed:
+	tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
+	SAFE_FREE(dbuf.dptr);
 	return ret;
 }
 
-/* lock/unlock entire database */
-int tdb_lockall(TDB_CONTEXT *tdb)
-{
-	u32 i;
-
-	/* There are no locks on read-only dbs */
-	if (tdb->read_only)
-		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
-	for (i = 0; i < tdb->header.hash_size; i++) 
-		if (tdb_lock(tdb, i, F_WRLCK))
-			break;
-
-	/* If error, release locks we have... */
-	if (i < tdb->header.hash_size) {
-		u32 j;
-
-		for ( j = 0; j < i; j++)
-			tdb_unlock(tdb, j, F_WRLCK);
-		return TDB_ERRCODE(TDB_ERR_NOLOCK, -1);
-	}
-
-	return 0;
-}
-void tdb_unlockall(TDB_CONTEXT *tdb)
-{
-	u32 i;
-	for (i=0; i < tdb->header.hash_size; i++)
-		tdb_unlock(tdb, i, F_WRLCK);
-}
-
-/* lock/unlock one hash chain. This is meant to be used to reduce
-   contention - it cannot guarantee how many records will be locked */
-int tdb_chainlock(TDB_CONTEXT *tdb, TDB_DATA key)
-{
-	return tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
-}
 
-int tdb_chainunlock(TDB_CONTEXT *tdb, TDB_DATA key)
+/*
+  return the name of the current tdb file
+  useful for external logging functions
+*/
+const char *tdb_name(struct tdb_context *tdb)
 {
-	return tdb_unlock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
+	return tdb->name;
 }
 
-int tdb_chainlock_read(TDB_CONTEXT *tdb, TDB_DATA key)
+/*
+  return the underlying file descriptor being used by tdb, or -1
+  useful for external routines that want to check the device/inode
+  of the fd
+*/
+int tdb_fd(struct tdb_context *tdb)
 {
-	return tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_RDLCK);
+	return tdb->fd;
 }
 
-int tdb_chainunlock_read(TDB_CONTEXT *tdb, TDB_DATA key)
+/*
+  return the current logging function
+  useful for external tdb routines that wish to log tdb errors
+*/
+tdb_log_func tdb_log_fn(struct tdb_context *tdb)
 {
-	return tdb_unlock(tdb, BUCKET(tdb->hash_fn(&key)), F_RDLCK);
+	return tdb->log_fn;
 }
 
 
-/* register a loging function */
-void tdb_logging_function(TDB_CONTEXT *tdb, void (*fn)(TDB_CONTEXT *, int , const char *, ...))
-{
-	tdb->log_fn = fn;
-}
-
-/* reopen a tdb - this can be used after a fork to ensure that we have an independent
-   seek pointer from our parent and to re-establish locks */
-int tdb_reopen(TDB_CONTEXT *tdb)
+/*
+  get the tdb sequence number. Only makes sense if the writers opened
+  with TDB_SEQNUM set. Note that this sequence number will wrap quite
+  quickly, so it should only be used for a 'has something changed'
+  test, not for code that relies on the count of the number of changes
+  made. If you want a counter then use a tdb record.
+
+  The aim of this sequence number is to allow for a very lightweight
+  test of a possible tdb change.
+*/
+int tdb_get_seqnum(struct tdb_context *tdb)
 {
-	struct stat st;
+	tdb_off_t seqnum=0;
 
-	if (tdb->flags & TDB_INTERNAL)
-		return 0; /* Nothing to do. */
-	if (tdb_munmap(tdb) != 0) {
-		TDB_LOG((tdb, 0, "tdb_reopen: munmap failed (%s)\n", strerror(errno)));
-		goto fail;
-	}
-	if (close(tdb->fd) != 0)
-		TDB_LOG((tdb, 0, "tdb_reopen: WARNING closing tdb->fd failed!\n"));
-	tdb->fd = open(tdb->name, tdb->open_flags & ~(O_CREAT|O_TRUNC), 0);
-	if (tdb->fd == -1) {
-		TDB_LOG((tdb, 0, "tdb_reopen: open failed (%s)\n", strerror(errno)));
-		goto fail;
-	}
-	if ((tdb->flags & TDB_CLEAR_IF_FIRST) &&
-			(tdb_brlock(tdb, ACTIVE_LOCK, F_RDLCK, F_SETLKW, 0) == -1)) {
-		TDB_LOG((tdb, 0, "tdb_reopen: failed to obtain active lock\n"));
-		goto fail;
-	}
-	if (fstat(tdb->fd, &st) != 0) {
-		TDB_LOG((tdb, 0, "tdb_reopen: fstat failed (%s)\n", strerror(errno)));
-		goto fail;
-	}
-	if (st.st_ino != tdb->inode || st.st_dev != tdb->device) {
-		TDB_LOG((tdb, 0, "tdb_reopen: file dev/inode has changed!\n"));
-		goto fail;
-	}
-	tdb_mmap(tdb);
-
-	return 0;
-
-fail:
-	tdb_close(tdb);
-	return -1;
+	tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &seqnum);
+	return seqnum;
 }
 
-/* reopen all tdb's */
-int tdb_reopen_all(int parent_longlived)
+int tdb_hash_size(struct tdb_context *tdb)
 {
-	TDB_CONTEXT *tdb;
-
-	for (tdb=tdbs; tdb; tdb = tdb->next) {
-		/*
-		 * If the parent is longlived (ie. a
-		 * parent daemon architecture), we know
-		 * it will keep it's active lock on a
-		 * tdb opened with CLEAR_IF_FIRST. Thus
-		 * for child processes we don't have to
-		 * add an active lock. This is essential
-		 * to improve performance on systems that
-		 * keep POSIX locks as a non-scalable data
-		 * structure in the kernel.
-		 */
-		if (parent_longlived) {
-			/* Ensure no clear-if-first. */
-			tdb->flags &= ~TDB_CLEAR_IF_FIRST;
-		}
-
-		if (tdb_reopen(tdb) != 0)
-			return -1;
-	}
-
-	return 0;
+	return tdb->header.hash_size;
 }
diff --git a/source/tdb/tdb.h b/source/tdb/tdb.h
index b5b87ee5a54..82f8828c89d 100644
--- a/source/tdb/tdb.h
+++ b/source/tdb/tdb.h
@@ -3,9 +3,9 @@
 
 /* 
    Unix SMB/CIFS implementation.
-   
+
    trivial database library
-   
+
    Copyright (C) Andrew Tridgell 1999-2004
    
      ** NOTE! The following LGPL license applies to the tdb
@@ -21,7 +21,7 @@
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
    Lesser General Public License for more details.
-   
+
    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
@@ -31,17 +31,6 @@
 extern "C" {
 #endif
 
-#ifndef PRINTF_ATTRIBUTE
-/** Use gcc attribute to check printf fns.  a1 is the 1-based index of
- * the parameter containing the format, and a2 the index of the first
- * argument. Note that some gcc 2.x versions don't handle this
- * properly **/
-#if (__GNUC__ >= 3) && (__GNUC_MINOR__ >= 1 )
-#define PRINTF_ATTRIBUTE(a1, a2) __attribute__ ((format (__printf__, a1, a2)))
-#else
-#define PRINTF_ATTRIBUTE(a1, a2)
-#endif
-#endif
 
 /* flags to tdb_store() */
 #define TDB_REPLACE 1
@@ -56,102 +45,83 @@ extern "C" {
 #define TDB_NOMMAP   8 /* don't use mmap */
 #define TDB_CONVERT 16 /* convert endian (internal use) */
 #define TDB_BIGENDIAN 32 /* header is big-endian (internal use) */
+#define TDB_NOSYNC   64 /* don't use synchronous transactions */
+#define TDB_SEQNUM   128 /* maintain a sequence number */
 
 #define TDB_ERRCODE(code, ret) ((tdb->ecode = (code)), ret)
 
 /* error codes */
 enum TDB_ERROR {TDB_SUCCESS=0, TDB_ERR_CORRUPT, TDB_ERR_IO, TDB_ERR_LOCK, 
 		TDB_ERR_OOM, TDB_ERR_EXISTS, TDB_ERR_NOLOCK, TDB_ERR_LOCK_TIMEOUT,
-		TDB_ERR_NOEXIST};
-
-#ifndef u32
-#define u32 unsigned
-#endif
+		TDB_ERR_NOEXIST, TDB_ERR_EINVAL, TDB_ERR_RDONLY};
 
-typedef struct {
+typedef struct TDB_DATA {
 	char *dptr;
 	size_t dsize;
 } TDB_DATA;
 
-typedef u32 tdb_len;
-typedef u32 tdb_off;
-
-/* this is stored at the front of every database */
-struct tdb_header {
-	char magic_food[32]; /* for /etc/magic */
-	u32 version; /* version of the code */
-	u32 hash_size; /* number of hash entries */
-	tdb_off rwlocks;
-	tdb_off reserved[31];
-};
-
-struct tdb_lock_type {
-	u32 count;
-	u32 ltype;
-};
-
-struct tdb_traverse_lock {
-	struct tdb_traverse_lock *next;
-	u32 off;
-	u32 hash;
-};
+#ifndef PRINTF_ATTRIBUTE
+#if (__GNUC__ >= 3)
+/** Use gcc attribute to check printf fns.  a1 is the 1-based index of
+ * the parameter containing the format, and a2 the index of the first
+ * argument. Note that some gcc 2.x versions don't handle this
+ * properly **/
+#define PRINTF_ATTRIBUTE(a1, a2) __attribute__ ((format (__printf__, a1, a2)))
+#else
+#define PRINTF_ATTRIBUTE(a1, a2)
+#endif
+#endif
 
 /* this is the context structure that is returned from a db open */
-typedef struct tdb_context {
-	char *name; /* the name of the database */
-	void *map_ptr; /* where it is currently mapped */
-	int fd; /* open file descriptor for the database */
-	tdb_len map_size; /* how much space has been mapped */
-	int read_only; /* opened read-only */
-	struct tdb_lock_type *locked; /* array of chain locks */
-	enum TDB_ERROR ecode; /* error code for last tdb error */
-	struct tdb_header header; /* a cached copy of the header */
-	u32 flags; /* the flags passed to tdb_open */
-	struct tdb_traverse_lock travlocks; /* current traversal locks */
-	struct tdb_context *next; /* all tdbs to avoid multiple opens */
-	dev_t device;	/* uniquely identifies this tdb */
-	ino_t inode;	/* uniquely identifies this tdb */
-	void (*log_fn)(struct tdb_context *tdb, int level, const char *, ...) PRINTF_ATTRIBUTE(3,4); /* logging function */
-	u32 (*hash_fn)(TDB_DATA *key);
-	int open_flags; /* flags used in the open - needed by reopen */
-} TDB_CONTEXT;
-
-typedef int (*tdb_traverse_func)(TDB_CONTEXT *, TDB_DATA, TDB_DATA, void *);
-typedef void (*tdb_log_func)(TDB_CONTEXT *, int , const char *, ...);
-typedef u32 (*tdb_hash_func)(TDB_DATA *key);
-
-TDB_CONTEXT *tdb_open(const char *name, int hash_size, int tdb_flags,
+typedef struct tdb_context TDB_CONTEXT;
+
+typedef int (*tdb_traverse_func)(struct tdb_context *, TDB_DATA, TDB_DATA, void *);
+typedef void (*tdb_log_func)(struct tdb_context *, int , const char *, ...);
+typedef unsigned int (*tdb_hash_func)(TDB_DATA *key);
+
+struct tdb_context *tdb_open(const char *name, int hash_size, int tdb_flags,
 		      int open_flags, mode_t mode);
-TDB_CONTEXT *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
+struct tdb_context *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
 			 int open_flags, mode_t mode,
 			 tdb_log_func log_fn,
 			 tdb_hash_func hash_fn);
 
-int tdb_reopen(TDB_CONTEXT *tdb);
-int tdb_reopen_all(int);
-void tdb_logging_function(TDB_CONTEXT *tdb, tdb_log_func);
-enum TDB_ERROR tdb_error(TDB_CONTEXT *tdb);
-const char *tdb_errorstr(TDB_CONTEXT *tdb);
-TDB_DATA tdb_fetch(TDB_CONTEXT *tdb, TDB_DATA key);
-int tdb_delete(TDB_CONTEXT *tdb, TDB_DATA key);
-int tdb_store(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA dbuf, int flag);
-int tdb_append(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA new_dbuf);
-int tdb_close(TDB_CONTEXT *tdb);
-TDB_DATA tdb_firstkey(TDB_CONTEXT *tdb);
-TDB_DATA tdb_nextkey(TDB_CONTEXT *tdb, TDB_DATA key);
-int tdb_traverse(TDB_CONTEXT *tdb, tdb_traverse_func fn, void *);
-int tdb_exists(TDB_CONTEXT *tdb, TDB_DATA key);
-int tdb_lockall(TDB_CONTEXT *tdb);
-void tdb_unlockall(TDB_CONTEXT *tdb);
+int tdb_reopen(struct tdb_context *tdb);
+int tdb_reopen_all(int parent_longlived);
+void tdb_logging_function(struct tdb_context *tdb, tdb_log_func);
+enum TDB_ERROR tdb_error(struct tdb_context *tdb);
+const char *tdb_errorstr(struct tdb_context *tdb);
+TDB_DATA tdb_fetch(struct tdb_context *tdb, TDB_DATA key);
+int tdb_delete(struct tdb_context *tdb, TDB_DATA key);
+int tdb_store(struct tdb_context *tdb, TDB_DATA key, TDB_DATA dbuf, int flag);
+int tdb_append(struct tdb_context *tdb, TDB_DATA key, TDB_DATA new_dbuf);
+int tdb_close(struct tdb_context *tdb);
+TDB_DATA tdb_firstkey(struct tdb_context *tdb);
+TDB_DATA tdb_nextkey(struct tdb_context *tdb, TDB_DATA key);
+int tdb_traverse(struct tdb_context *tdb, tdb_traverse_func fn, void *);
+int tdb_traverse_read(struct tdb_context *tdb, tdb_traverse_func fn, void *);
+int tdb_exists(struct tdb_context *tdb, TDB_DATA key);
+int tdb_lockall(struct tdb_context *tdb);
+void tdb_unlockall(struct tdb_context *tdb);
+const char *tdb_name(struct tdb_context *tdb);
+int tdb_fd(struct tdb_context *tdb);
+tdb_log_func tdb_log_fn(struct tdb_context *tdb);
+int tdb_transaction_start(struct tdb_context *tdb);
+int tdb_transaction_commit(struct tdb_context *tdb);
+int tdb_transaction_cancel(struct tdb_context *tdb);
+int tdb_transaction_recover(struct tdb_context *tdb);
+int tdb_get_seqnum(struct tdb_context *tdb);
+int tdb_hash_size(struct tdb_context *tdb);
 
 /* Low level locking functions: use with care */
-void tdb_set_lock_alarm(sig_atomic_t *palarm);
-int tdb_chainlock(TDB_CONTEXT *tdb, TDB_DATA key);
-int tdb_chainunlock(TDB_CONTEXT *tdb, TDB_DATA key);
+int tdb_chainlock(struct tdb_context *tdb, TDB_DATA key);
+int tdb_chainunlock(struct tdb_context *tdb, TDB_DATA key);
+int tdb_chainlock_read(struct tdb_context *tdb, TDB_DATA key);
+int tdb_chainunlock_read(struct tdb_context *tdb, TDB_DATA key);
 
 /* Debug functions. Not used in production. */
-void tdb_dump_all(TDB_CONTEXT *tdb);
-int tdb_printfreelist(TDB_CONTEXT *tdb);
+void tdb_dump_all(struct tdb_context *tdb);
+int tdb_printfreelist(struct tdb_context *tdb);
 
 extern TDB_DATA tdb_null;
 
diff --git a/source/tdb/tdb_private.h b/source/tdb/tdb_private.h
new file mode 100644
index 00000000000..d10c1c46bf6
--- /dev/null
+++ b/source/tdb/tdb_private.h
@@ -0,0 +1,241 @@
+ /* 
+   Unix SMB/CIFS implementation.
+
+   trivial database library - private includes
+
+   Copyright (C) Andrew Tridgell              2005
+   
+     ** NOTE! The following LGPL license applies to the tdb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+   
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#ifndef _SAMBA_BUILD_
+#include "config.h"
+#include <stdlib.h>
+#include <stdio.h>
+#ifdef HAVE_STDINT_H
+#include <stdint.h>
+#endif
+#include <fcntl.h>
+#include <unistd.h>
+#include <string.h>
+#include <fcntl.h>
+#include <errno.h>
+#include <sys/mman.h>
+#include <sys/stat.h>
+#ifdef HAVE_SYS_SELECT_H
+#include <sys/select.h>
+#endif
+#ifdef HAVE_SYS_TIME_H
+#include <sys/time.h>
+#endif
+#include "tdb.h"
+
+#ifndef HAVE_PREAD_DECL
+ssize_t pread(int fd, void *buf, size_t count, off_t offset);
+#endif
+#ifndef HAVE_PWRITE_DECL
+ssize_t pwrite(int fd, const void *buf, size_t count, off_t offset);
+#endif
+
+#else
+#include "includes.h"
+#undef malloc
+#undef realloc
+#undef calloc
+#undef strdup
+#endif
+
+#ifndef u32
+#define u32 unsigned
+#endif
+
+#ifndef HAVE_GETPAGESIZE
+#define getpagesize() 0x2000
+#endif
+
+typedef u32 tdb_len_t;
+typedef u32 tdb_off_t;
+
+#ifndef offsetof
+#define offsetof(t,f) ((unsigned int)&((t *)0)->f)
+#endif
+
+#define TDB_MAGIC_FOOD "TDB file\n"
+#define TDB_VERSION (0x26011967 + 6)
+#define TDB_MAGIC (0x26011999U)
+#define TDB_FREE_MAGIC (~TDB_MAGIC)
+#define TDB_DEAD_MAGIC (0xFEE1DEAD)
+#define TDB_RECOVERY_MAGIC (0xf53bc0e7U)
+#define TDB_ALIGNMENT 4
+#define MIN_REC_SIZE (2*sizeof(struct list_struct) + TDB_ALIGNMENT)
+#define DEFAULT_HASH_SIZE 131
+#define FREELIST_TOP (sizeof(struct tdb_header))
+#define TDB_ALIGN(x,a) (((x) + (a)-1) & ~((a)-1))
+#define TDB_BYTEREV(x) (((((x)&0xff)<<24)|((x)&0xFF00)<<8)|(((x)>>8)&0xFF00)|((x)>>24))
+#define TDB_DEAD(r) ((r)->magic == TDB_DEAD_MAGIC)
+#define TDB_BAD_MAGIC(r) ((r)->magic != TDB_MAGIC && !TDB_DEAD(r))
+#define TDB_HASH_TOP(hash) (FREELIST_TOP + (BUCKET(hash)+1)*sizeof(tdb_off_t))
+#define TDB_HASHTABLE_SIZE(tdb) ((tdb->header.hash_size+1)*sizeof(tdb_off_t))
+#define TDB_DATA_START(hash_size) TDB_HASH_TOP(hash_size-1)
+#define TDB_RECOVERY_HEAD offsetof(struct tdb_header, recovery_start)
+#define TDB_SEQNUM_OFS    offsetof(struct tdb_header, sequence_number)
+#define TDB_PAD_BYTE 0x42
+#define TDB_PAD_U32  0x42424242
+
+/* NB assumes there is a local variable called "tdb" that is the
+ * current context, also takes doubly-parenthesized print-style
+ * argument. */
+#define TDB_LOG(x) tdb->log_fn x
+
+/* lock offsets */
+#define GLOBAL_LOCK      0
+#define ACTIVE_LOCK      4
+#define TRANSACTION_LOCK 8
+
+#ifndef MAP_FILE
+#define MAP_FILE 0
+#endif
+
+#ifndef MAP_FAILED
+#define MAP_FAILED ((void *)-1)
+#endif
+
+/* free memory if the pointer is valid and zero the pointer */
+#ifndef SAFE_FREE
+#define SAFE_FREE(x) do { if ((x) != NULL) {free(x); (x)=NULL;} } while(0)
+#endif
+
+#define BUCKET(hash) ((hash) % tdb->header.hash_size)
+
+#define DOCONV() (tdb->flags & TDB_CONVERT)
+#define CONVERT(x) (DOCONV() ? tdb_convert(&x, sizeof(x)) : &x)
+
+
+/* the body of the database is made of one list_struct for the free space
+   plus a separate data list for each hash value */
+struct list_struct {
+	tdb_off_t next; /* offset of the next record in the list */
+	tdb_len_t rec_len; /* total byte length of record */
+	tdb_len_t key_len; /* byte length of key */
+	tdb_len_t data_len; /* byte length of data */
+	u32 full_hash; /* the full 32 bit hash of the key */
+	u32 magic;   /* try to catch errors */
+	/* the following union is implied:
+		union {
+			char record[rec_len];
+			struct {
+				char key[key_len];
+				char data[data_len];
+			}
+			u32 totalsize; (tailer)
+		}
+	*/
+};
+
+
+/* this is stored at the front of every database */
+struct tdb_header {
+	char magic_food[32]; /* for /etc/magic */
+	u32 version; /* version of the code */
+	u32 hash_size; /* number of hash entries */
+	tdb_off_t rwlocks; /* obsolete - kept to detect old formats */
+	tdb_off_t recovery_start; /* offset of transaction recovery region */
+	tdb_off_t sequence_number; /* used when TDB_SEQNUM is set */
+	tdb_off_t reserved[29];
+};
+
+struct tdb_lock_type {
+	u32 count;
+	u32 ltype;
+};
+
+struct tdb_traverse_lock {
+	struct tdb_traverse_lock *next;
+	u32 off;
+	u32 hash;
+	int lock_rw;
+};
+
+
+struct tdb_methods {
+	int (*tdb_read)(struct tdb_context *, tdb_off_t , void *, tdb_len_t , int );
+	int (*tdb_write)(struct tdb_context *, tdb_off_t, const void *, tdb_len_t);
+	void (*next_hash_chain)(struct tdb_context *, u32 *);
+	int (*tdb_oob)(struct tdb_context *, tdb_off_t , int );
+	int (*tdb_expand_file)(struct tdb_context *, tdb_off_t , tdb_off_t );
+	int (*tdb_brlock)(struct tdb_context *, tdb_off_t , int, int, int);
+};
+
+struct tdb_context {
+	char *name; /* the name of the database */
+	void *map_ptr; /* where it is currently mapped */
+	int fd; /* open file descriptor for the database */
+	tdb_len_t map_size; /* how much space has been mapped */
+	int read_only; /* opened read-only */
+	int traverse_read; /* read-only traversal */
+	struct tdb_lock_type *locked; /* array of chain locks */
+	enum TDB_ERROR ecode; /* error code for last tdb error */
+	struct tdb_header header; /* a cached copy of the header */
+	u32 flags; /* the flags passed to tdb_open */
+	struct tdb_traverse_lock travlocks; /* current traversal locks */
+	struct tdb_context *next; /* all tdbs to avoid multiple opens */
+	dev_t device;	/* uniquely identifies this tdb */
+	ino_t inode;	/* uniquely identifies this tdb */
+	void (*log_fn)(struct tdb_context *tdb, int level, const char *, ...) PRINTF_ATTRIBUTE(3,4); /* logging function */
+	unsigned int (*hash_fn)(TDB_DATA *key);
+	int open_flags; /* flags used in the open - needed by reopen */
+	unsigned int num_locks; /* number of chain locks held */
+	const struct tdb_methods *methods;
+	struct tdb_transaction *transaction;
+	int page_size;
+};
+
+
+/*
+  internal prototypes
+*/
+int tdb_munmap(struct tdb_context *tdb);
+void tdb_mmap(struct tdb_context *tdb);
+int tdb_lock(struct tdb_context *tdb, int list, int ltype);
+int tdb_unlock(struct tdb_context *tdb, int list, int ltype);
+int tdb_brlock(struct tdb_context *tdb, tdb_off_t offset, int rw_type, int lck_type, int probe);
+int tdb_brlock_upgrade(struct tdb_context *tdb, tdb_off_t offset, size_t len);
+int tdb_brlock_len(struct tdb_context *tdb, tdb_off_t offset, 
+		   int rw_type, int lck_type, int probe, size_t len);
+int tdb_write_lock_record(struct tdb_context *tdb, tdb_off_t off);
+int tdb_write_unlock_record(struct tdb_context *tdb, tdb_off_t off);
+int tdb_ofs_read(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d);
+int tdb_ofs_write(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d);
+void *tdb_convert(void *buf, u32 size);
+int tdb_free(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec);
+tdb_off_t tdb_allocate(struct tdb_context *tdb, tdb_len_t length, struct list_struct *rec);
+int tdb_ofs_read(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d);
+int tdb_ofs_write(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d);
+int tdb_lock_record(struct tdb_context *tdb, tdb_off_t off);
+int tdb_unlock_record(struct tdb_context *tdb, tdb_off_t off);
+int tdb_rec_read(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec);
+int tdb_rec_write(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec);
+int tdb_do_delete(struct tdb_context *tdb, tdb_off_t rec_ptr, struct list_struct *rec);
+char *tdb_alloc_read(struct tdb_context *tdb, tdb_off_t offset, tdb_len_t len);
+tdb_off_t tdb_find_lock_hash(struct tdb_context *tdb, TDB_DATA key, u32 hash, int locktype,
+			   struct list_struct *rec);
+void tdb_io_init(struct tdb_context *tdb);
+int tdb_expand(struct tdb_context *tdb, tdb_off_t size);
+
+
diff --git a/source/tdb/tdbback.c b/source/tdb/tdbback.c
index 7967a6bbf54..c337924d6cf 100644
--- a/source/tdb/tdbback.c
+++ b/source/tdb/tdbback.c
@@ -77,7 +77,7 @@ static int copy_fn(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA dbuf, void *state)
 	TDB_CONTEXT *tdb_new = (TDB_CONTEXT *)state;
 
 	if (tdb_store(tdb_new, key, dbuf, TDB_INSERT) != 0) {
-		fprintf(stderr,"Failed to insert into %s\n", tdb_new->name);
+		fprintf(stderr,"Failed to insert into %s\n", tdb_name(tdb));
 		failed = 1;
 		return 1;
 	}
@@ -122,7 +122,7 @@ int backup_tdb(const char *old_name, const char *new_name)
 
 	/* create the new tdb */
 	unlink(tmp_name);
-	tdb_new = tdb_open(tmp_name, tdb->header.hash_size, 
+	tdb_new = tdb_open(tmp_name, tdb_hash_size(tdb),
 			   TDB_DEFAULT, O_RDWR|O_CREAT|O_EXCL, 
 			   st.st_mode & 0777);
 	if (!tdb_new) {
@@ -179,7 +179,7 @@ int backup_tdb(const char *old_name, const char *new_name)
 	}
 
 	/* make sure the new tdb has reached stable storage */
-	fsync(tdb_new->fd);
+	fsync(tdb_fd(tdb_new));
 
 	/* close the new tdb and rename it to .bak */
 	tdb_close(tdb_new);
diff --git a/source/tdb/tdbutil.c b/source/tdb/tdbutil.c
index fc45fa93606..e13f7e3e678 100644
--- a/source/tdb/tdbutil.c
+++ b/source/tdb/tdbutil.c
@@ -19,9 +19,21 @@
    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */
 
-#include "includes.h"
+#include "tdb_private.h"
 #include <fnmatch.h>
 
+/***************************************************************
+ Allow a caller to set a "alarm" flag that tdb can check to abort
+ a blocking lock on SIGALRM.
+***************************************************************/
+
+static sig_atomic_t *palarm_fired;
+
+static void tdb_set_lock_alarm(sig_atomic_t *palarm)
+{
+	palarm_fired = palarm;
+}
+
 /* these are little tdb utility functions that are meant to make
    dealing with a tdb database a little less cumbersome in Samba */
 
@@ -641,129 +653,6 @@ int tdb_unpack(char *buf, int bufsize, const char *fmt, ...)
 }
 
 
-/**
- * Pack SID passed by pointer
- *
- * @param pack_buf pointer to buffer which is to be filled with packed data
- * @param bufsize size of packing buffer
- * @param sid pointer to sid to be packed
- *
- * @return length of the packed representation of the whole structure
- **/
-size_t tdb_sid_pack(char* pack_buf, int bufsize, DOM_SID* sid)
-{
-	int idx;
-	size_t len = 0;
-	
-	if (!sid || !pack_buf) return -1;
-	
-	len += tdb_pack(pack_buf + len, bufsize - len, "bb", sid->sid_rev_num,
-	                sid->num_auths);
-	
-	for (idx = 0; idx < 6; idx++) {
-		len += tdb_pack(pack_buf + len, bufsize - len, "b", sid->id_auth[idx]);
-	}
-	
-	for (idx = 0; idx < MAXSUBAUTHS; idx++) {
-		len += tdb_pack(pack_buf + len, bufsize - len, "d", sid->sub_auths[idx]);
-	}
-	
-	return len;
-}
-
-
-/**
- * Unpack SID into a pointer
- *
- * @param pack_buf pointer to buffer with packed representation
- * @param bufsize size of the buffer
- * @param sid pointer to sid structure to be filled with unpacked data
- *
- * @return size of structure unpacked from buffer
- **/
-size_t tdb_sid_unpack(char* pack_buf, int bufsize, DOM_SID* sid)
-{
-	int idx, len = 0;
-	
-	if (!sid || !pack_buf) return -1;
-
-	len += tdb_unpack(pack_buf + len, bufsize - len, "bb",
-	                  &sid->sid_rev_num, &sid->num_auths);
-			  
-	for (idx = 0; idx < 6; idx++) {
-		len += tdb_unpack(pack_buf + len, bufsize - len, "b", &sid->id_auth[idx]);
-	}
-	
-	for (idx = 0; idx < MAXSUBAUTHS; idx++) {
-		len += tdb_unpack(pack_buf + len, bufsize - len, "d", &sid->sub_auths[idx]);
-	}
-	
-	return len;
-}
-
-
-/**
- * Pack TRUSTED_DOM_PASS passed by pointer
- *
- * @param pack_buf pointer to buffer which is to be filled with packed data
- * @param bufsize size of the buffer
- * @param pass pointer to trusted domain password to be packed
- *
- * @return length of the packed representation of the whole structure
- **/
-size_t tdb_trusted_dom_pass_pack(char* pack_buf, int bufsize, TRUSTED_DOM_PASS* pass)
-{
-	int idx, len = 0;
-	
-	if (!pack_buf || !pass) return -1;
-	
-	/* packing unicode domain name and password */
-	len += tdb_pack(pack_buf + len, bufsize - len, "d", pass->uni_name_len);
-	
-	for (idx = 0; idx < 32; idx++)
-		len +=  tdb_pack(pack_buf + len, bufsize - len, "w", pass->uni_name[idx]);
-	
-	len += tdb_pack(pack_buf + len, bufsize - len, "dPd", pass->pass_len,
-	                     pass->pass, pass->mod_time);
-
-	/* packing SID structure */
-	len += tdb_sid_pack(pack_buf + len, bufsize - len, &pass->domain_sid);
-
-	return len;
-}
-
-
-/**
- * Unpack TRUSTED_DOM_PASS passed by pointer
- *
- * @param pack_buf pointer to buffer with packed representation
- * @param bufsize size of the buffer
- * @param pass pointer to trusted domain password to be filled with unpacked data
- *
- * @return size of structure unpacked from buffer
- **/
-size_t tdb_trusted_dom_pass_unpack(char* pack_buf, int bufsize, TRUSTED_DOM_PASS* pass)
-{
-	int idx, len = 0;
-	
-	if (!pack_buf || !pass) return -1;
-
-	/* unpack unicode domain name and plaintext password */
-	len += tdb_unpack(pack_buf, bufsize - len, "d", &pass->uni_name_len);
-	
-	for (idx = 0; idx < 32; idx++)
-		len +=  tdb_unpack(pack_buf + len, bufsize - len, "w", &pass->uni_name[idx]);
-
-	len += tdb_unpack(pack_buf + len, bufsize - len, "dPd", &pass->pass_len, &pass->pass,
-	                  &pass->mod_time);
-	
-	/* unpack domain sid */
-	len += tdb_sid_unpack(pack_buf + len, bufsize - len, &pass->domain_sid);
-	
-	return len;	
-}
-
-
 /****************************************************************************
  Log tdb messages via DEBUG().
 ****************************************************************************/
@@ -806,15 +695,6 @@ TDB_CONTEXT *tdb_open_log(const char *name, int hash_size, int tdb_flags,
 }
 
 /****************************************************************************
-  return the name of the current tdb file useful for external logging
-  functions
-****************************************************************************/
-const char *tdb_name(struct tdb_context *tdb)
-{
-	return tdb->name;
-}
-
-/****************************************************************************
  Allow tdb_delete to be used as a tdb_traversal_fn.
 ****************************************************************************/
 
@@ -893,3 +773,43 @@ void tdb_search_list_free(TDB_LIST_NODE* node)
 		node = next_node;
 	};
 }
+
+size_t tdb_map_size(struct tdb_context *tdb)
+{
+	return tdb->map_size;
+}
+
+int tdb_get_flags(struct tdb_context *tdb)
+{
+	return tdb->flags;
+}
+
+/****************************************************************************
+ tdb_store, wrapped in a transaction. This way we make sure that a process
+ that dies within writing does not leave a corrupt tdb behind.
+****************************************************************************/
+
+int tdb_trans_store(struct tdb_context *tdb, TDB_DATA key, TDB_DATA dbuf,
+		    int flag)
+{
+	int res;
+
+	if ((res = tdb_transaction_start(tdb)) != 0) {
+		DEBUG(5, ("tdb_transaction_start failed\n"));
+		return res;
+	}
+
+	if ((res = tdb_store(tdb, key, dbuf, flag)) != 0) {
+		DEBUG(10, ("tdb_store failed\n"));
+		if (tdb_transaction_cancel(tdb) != 0) {
+			smb_panic("Cancelling transaction failed\n");
+		}
+		return res;
+	}
+
+	if ((res = tdb_transaction_commit(tdb)) != 0) {
+		DEBUG(5, ("tdb_transaction_commit failed\n"));
+	}
+
+	return res;
+}
diff --git a/source/tdb/tdbutil.h b/source/tdb/tdbutil.h
index 1a181a962f2..44351619f45 100644
--- a/source/tdb/tdbutil.h
+++ b/source/tdb/tdbutil.h
@@ -1,8 +1,7 @@
 /* 
    Unix SMB/CIFS implementation.
    tdb utility functions
-   Copyright (C) Andrew Tridgell   1999
-   Copyright (C) Rafal Szczesniak  2002
+   Copyright (C) Andrew Tridgell 1999
    
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -22,6 +21,7 @@
 #ifndef __TDBUTIL_H__
 #define __TDBUTIL_H__
 
+#include "tdb.h"
 
 /* single node of a list returned by tdb_search_keys */
 typedef struct keys_node 
@@ -31,8 +31,38 @@ typedef struct keys_node
 } TDB_LIST_NODE;
 
 
-TDB_LIST_NODE *tdb_search_keys(TDB_CONTEXT*, const char*);
+TDB_LIST_NODE *tdb_search_keys(struct tdb_context*, const char*);
 void tdb_search_list_free(TDB_LIST_NODE*);
-
+int32_t tdb_change_int32_atomic(struct tdb_context *tdb, const char *keystr, int32_t *oldval, int32_t change_val);
+int tdb_lock_bystring(struct tdb_context *tdb, const char *keyval);
+int tdb_lock_bystring_with_timeout(TDB_CONTEXT *tdb, const char *keyval,
+				   int timeout);
+void tdb_unlock_bystring(struct tdb_context *tdb, const char *keyval);
+int tdb_read_lock_bystring_with_timeout(TDB_CONTEXT *tdb, const char *keyval,
+					unsigned int timeout);
+void tdb_read_unlock_bystring(TDB_CONTEXT *tdb, const char *keyval);
+int32_t tdb_fetch_int32(struct tdb_context *tdb, const char *keystr);
+BOOL tdb_store_uint32(struct tdb_context *tdb, const char *keystr, uint32_t value);
+int tdb_store_int32(struct tdb_context *tdb, const char *keystr, int32_t v);
+BOOL tdb_fetch_uint32(struct tdb_context *tdb, const char *keystr, uint32_t *value);
+int tdb_traverse_delete_fn(struct tdb_context *the_tdb, TDB_DATA key, TDB_DATA dbuf,
+                     void *state);
+int tdb_store_bystring(struct tdb_context *tdb, const char *keystr, TDB_DATA data, int flags);
+TDB_DATA tdb_fetch_bystring(struct tdb_context *tdb, const char *keystr);
+int tdb_delete_bystring(struct tdb_context *tdb, const char *keystr);
+struct tdb_context *tdb_open_log(const char *name, int hash_size,
+				 int tdb_flags, int open_flags, mode_t mode);
+int tdb_unpack(char *buf, int bufsize, const char *fmt, ...);
+size_t tdb_pack(char *buf, int bufsize, const char *fmt, ...);
+TDB_DATA make_tdb_data(const char *dptr, size_t dsize);
+TDB_DATA string_tdb_data(const char *string);
+int tdb_trans_store(struct tdb_context *tdb, TDB_DATA key, TDB_DATA dbuf,
+		    int flag);
+BOOL tdb_change_uint32_atomic(TDB_CONTEXT *tdb, const char *keystr,
+			      uint32 *oldval, uint32 change_val);
+int tdb_chainlock_with_timeout( TDB_CONTEXT *tdb, TDB_DATA key,
+				unsigned int timeout);
+int tdb_get_flags(struct tdb_context *tdb);
+size_t tdb_map_size(struct tdb_context *tdb);
 
 #endif /* __TDBUTIL_H__ */
diff --git a/source/tdb/transaction.c b/source/tdb/transaction.c
new file mode 100644
index 00000000000..aa54c4ae3c8
--- /dev/null
+++ b/source/tdb/transaction.c
@@ -0,0 +1,1043 @@
+ /* 
+   Unix SMB/CIFS implementation.
+
+   trivial database library
+
+   Copyright (C) Andrew Tridgell              2005
+
+     ** NOTE! The following LGPL license applies to the tdb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+   
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "tdb_private.h"
+
+/*
+  transaction design:
+
+  - only allow a single transaction at a time per database. This makes
+    using the transaction API simpler, as otherwise the caller would
+    have to cope with temporary failures in transactions that conflict
+    with other current transactions
+
+  - keep the transaction recovery information in the same file as the
+    database, using a special 'transaction recovery' record pointed at
+    by the header. This removes the need for extra journal files as
+    used by some other databases
+
+  - dymacially allocated the transaction recover record, re-using it
+    for subsequent transactions. If a larger record is needed then
+    tdb_free() the old record to place it on the normal tdb freelist
+    before allocating the new record
+
+  - during transactions, keep a linked list of writes all that have
+    been performed by intercepting all tdb_write() calls. The hooked
+    transaction versions of tdb_read() and tdb_write() check this
+    linked list and try to use the elements of the list in preference
+    to the real database.
+
+  - don't allow any locks to be held when a transaction starts,
+    otherwise we can end up with deadlock (plus lack of lock nesting
+    in posix locks would mean the lock is lost)
+
+  - if the caller gains a lock during the transaction but doesn't
+    release it then fail the commit
+
+  - allow for nested calls to tdb_transaction_start(), re-using the
+    existing transaction record. If the inner transaction is cancelled
+    then a subsequent commit will fail
+ 
+  - keep a mirrored copy of the tdb hash chain heads to allow for the
+    fast hash heads scan on traverse, updating the mirrored copy in
+    the transaction version of tdb_write
+
+  - allow callers to mix transaction and non-transaction use of tdb,
+    although once a transaction is started then an exclusive lock is
+    gained until the transaction is committed or cancelled
+
+  - the commit stategy involves first saving away all modified data
+    into a linearised buffer in the transaction recovery area, then
+    marking the transaction recovery area with a magic value to
+    indicate a valid recovery record. In total 4 fsync/msync calls are
+    needed per commit to prevent race conditions. It might be possible
+    to reduce this to 3 or even 2 with some more work.
+
+  - check for a valid recovery record on open of the tdb, while the
+    global lock is held. Automatically recover from the transaction
+    recovery area if needed, then continue with the open as
+    usual. This allows for smooth crash recovery with no administrator
+    intervention.
+
+  - if TDB_NOSYNC is passed to flags in tdb_open then transactions are
+    still available, but no transaction recovery area is used and no
+    fsync/msync calls are made.
+
+*/
+
+
+/*
+  hold the context of any current transaction
+*/
+struct tdb_transaction {
+	/* we keep a mirrored copy of the tdb hash heads here so
+	   tdb_next_hash_chain() can operate efficiently */
+	u32 *hash_heads;
+
+	/* the original io methods - used to do IOs to the real db */
+	const struct tdb_methods *io_methods;
+
+	/* the list of transaction elements. We use a doubly linked
+	   list with a last pointer to allow us to keep the list
+	   ordered, with first element at the front of the list. It
+	   needs to be doubly linked as the read/write traversals need
+	   to be backwards, while the commit needs to be forwards */
+	struct tdb_transaction_el {
+		struct tdb_transaction_el *next, *prev;
+		tdb_off_t offset;
+		tdb_len_t length;
+		unsigned char *data;
+	} *elements, *elements_last;
+
+	/* non-zero when an internal transaction error has
+	   occurred. All write operations will then fail until the
+	   transaction is ended */
+	int transaction_error;
+
+	/* when inside a transaction we need to keep track of any
+	   nested tdb_transaction_start() calls, as these are allowed,
+	   but don't create a new transaction */
+	int nesting;
+
+	/* old file size before transaction */
+	tdb_len_t old_map_size;
+};
+
+
+/*
+  read while in a transaction. We need to check first if the data is in our list
+  of transaction elements, then if not do a real read
+*/
+static int transaction_read(struct tdb_context *tdb, tdb_off_t off, void *buf, 
+			    tdb_len_t len, int cv)
+{
+	struct tdb_transaction_el *el;
+
+	/* we need to walk the list backwards to get the most recent data */
+	for (el=tdb->transaction->elements_last;el;el=el->prev) {
+		tdb_len_t partial;
+
+		if (off+len <= el->offset) {
+			continue;
+		}
+		if (off >= el->offset + el->length) {
+			continue;
+		}
+
+		/* an overlapping read - needs to be split into up to
+		   2 reads and a memcpy */
+		if (off < el->offset) {
+			partial = el->offset - off;
+			if (transaction_read(tdb, off, buf, partial, cv) != 0) {
+				goto fail;
+			}
+			len -= partial;
+			off += partial;
+			buf = (void *)(partial + (char *)buf);
+		}
+		if (off + len <= el->offset + el->length) {
+			partial = len;
+		} else {
+			partial = el->offset + el->length - off;
+		}
+		memcpy(buf, el->data + (off - el->offset), partial);
+		if (cv) {
+			tdb_convert(buf, len);
+		}
+		len -= partial;
+		off += partial;
+		buf = (void *)(partial + (char *)buf);
+		
+		if (len != 0 && transaction_read(tdb, off, buf, len, cv) != 0) {
+			goto fail;
+		}
+
+		return 0;
+	}
+
+	/* its not in the transaction elements - do a real read */
+	return tdb->transaction->io_methods->tdb_read(tdb, off, buf, len, cv);
+
+fail:
+	TDB_LOG((tdb, 0, "transaction_read: failed at off=%d len=%d\n", off, len));
+	tdb->ecode = TDB_ERR_IO;
+	tdb->transaction->transaction_error = 1;
+	return -1;
+}
+
+
+/*
+  write while in a transaction
+*/
+static int transaction_write(struct tdb_context *tdb, tdb_off_t off, 
+			     const void *buf, tdb_len_t len)
+{
+	struct tdb_transaction_el *el, *best_el=NULL;
+
+	if (len == 0) {
+		return 0;
+	}
+	
+	/* if the write is to a hash head, then update the transaction
+	   hash heads */
+	if (len == sizeof(tdb_off_t) && off >= FREELIST_TOP &&
+	    off < FREELIST_TOP+TDB_HASHTABLE_SIZE(tdb)) {
+		u32 chain = (off-FREELIST_TOP) / sizeof(tdb_off_t);
+		memcpy(&tdb->transaction->hash_heads[chain], buf, len);
+	}
+
+	/* first see if we can replace an existing entry */
+	for (el=tdb->transaction->elements_last;el;el=el->prev) {
+		tdb_len_t partial;
+
+		if (best_el == NULL && off == el->offset+el->length) {
+			best_el = el;
+		}
+
+		if (off+len <= el->offset) {
+			continue;
+		}
+		if (off >= el->offset + el->length) {
+			continue;
+		}
+
+		/* an overlapping write - needs to be split into up to
+		   2 writes and a memcpy */
+		if (off < el->offset) {
+			partial = el->offset - off;
+			if (transaction_write(tdb, off, buf, partial) != 0) {
+				goto fail;
+			}
+			len -= partial;
+			off += partial;
+			buf = (const void *)(partial + (const char *)buf);
+		}
+		if (off + len <= el->offset + el->length) {
+			partial = len;
+		} else {
+			partial = el->offset + el->length - off;
+		}
+		memcpy(el->data + (off - el->offset), buf, partial);
+		len -= partial;
+		off += partial;
+		buf = (const void *)(partial + (const char *)buf);
+		
+		if (len != 0 && transaction_write(tdb, off, buf, len) != 0) {
+			goto fail;
+		}
+
+		return 0;
+	}
+
+	/* see if we can append the new entry to an existing entry */
+	if (best_el && best_el->offset + best_el->length == off && 
+	    (off+len < tdb->transaction->old_map_size ||
+	     off > tdb->transaction->old_map_size)) {
+		unsigned char *data = best_el->data;
+		el = best_el;
+		el->data = realloc(el->data, el->length + len);
+		if (el->data == NULL) {
+			tdb->ecode = TDB_ERR_OOM;
+			tdb->transaction->transaction_error = 1;
+			el->data = data;
+			return -1;
+		}
+		if (buf) {
+			memcpy(el->data + el->length, buf, len);
+		} else {
+			memset(el->data + el->length, TDB_PAD_BYTE, len);
+		}
+		el->length += len;
+		return 0;
+	}
+
+	/* add a new entry at the end of the list */
+	el = malloc(sizeof(*el));
+	if (el == NULL) {
+		tdb->ecode = TDB_ERR_OOM;
+		tdb->transaction->transaction_error = 1;		
+		return -1;
+	}
+	el->next = NULL;
+	el->prev = tdb->transaction->elements_last;
+	el->offset = off;
+	el->length = len;
+	el->data = malloc(len);
+	if (el->data == NULL) {
+		free(el);
+		tdb->ecode = TDB_ERR_OOM;
+		tdb->transaction->transaction_error = 1;		
+		return -1;
+	}
+	if (buf) {
+		memcpy(el->data, buf, len);
+	} else {
+		memset(el->data, TDB_PAD_BYTE, len);
+	}
+	if (el->prev) {
+		el->prev->next = el;
+	} else {
+		tdb->transaction->elements = el;
+	}
+	tdb->transaction->elements_last = el;
+	return 0;
+
+fail:
+	TDB_LOG((tdb, 0, "transaction_write: failed at off=%d len=%d\n", off, len));
+	tdb->ecode = TDB_ERR_IO;
+	tdb->transaction->transaction_error = 1;
+	return -1;
+}
+
+/*
+  accelerated hash chain head search, using the cached hash heads
+*/
+static void transaction_next_hash_chain(struct tdb_context *tdb, u32 *chain)
+{
+	u32 h = *chain;
+	for (;h < tdb->header.hash_size;h++) {
+		/* the +1 takes account of the freelist */
+		if (0 != tdb->transaction->hash_heads[h+1]) {
+			break;
+		}
+	}
+	(*chain) = h;
+}
+
+/*
+  out of bounds check during a transaction
+*/
+static int transaction_oob(struct tdb_context *tdb, tdb_off_t len, int probe)
+{
+	if (len <= tdb->map_size) {
+		return 0;
+	}
+	return TDB_ERRCODE(TDB_ERR_IO, -1);
+}
+
+/*
+  transaction version of tdb_expand().
+*/
+static int transaction_expand_file(struct tdb_context *tdb, tdb_off_t size, 
+				   tdb_off_t addition)
+{
+	/* add a write to the transaction elements, so subsequent
+	   reads see the zero data */
+	if (transaction_write(tdb, size, NULL, addition) != 0) {
+		return -1;
+	}
+
+	return 0;
+}
+
+/*
+  brlock during a transaction - ignore them
+*/
+int transaction_brlock(struct tdb_context *tdb, tdb_off_t offset, 
+		       int rw_type, int lck_type, int probe)
+{
+	return 0;
+}
+
+static const struct tdb_methods transaction_methods = {
+	transaction_read,
+	transaction_write,
+	transaction_next_hash_chain,
+	transaction_oob,
+	transaction_expand_file,
+	transaction_brlock
+};
+
+
+/*
+  start a tdb transaction. No token is returned, as only a single
+  transaction is allowed to be pending per tdb_context
+*/
+int tdb_transaction_start(struct tdb_context *tdb)
+{
+	/* some sanity checks */
+	if (tdb->read_only || (tdb->flags & TDB_INTERNAL) || tdb->traverse_read) {
+		TDB_LOG((tdb, 0, "tdb_transaction_start: cannot start a transaction on a read-only or internal db\n"));
+		tdb->ecode = TDB_ERR_EINVAL;
+		return -1;
+	}
+
+	/* cope with nested tdb_transaction_start() calls */
+	if (tdb->transaction != NULL) {
+		tdb->transaction->nesting++;
+		TDB_LOG((tdb, 0, "tdb_transaction_start: nesting %d\n", 
+			 tdb->transaction->nesting));
+		return 0;
+	}
+
+	if (tdb->num_locks != 0) {
+		/* the caller must not have any locks when starting a
+		   transaction as otherwise we'll be screwed by lack
+		   of nested locks in posix */
+		TDB_LOG((tdb, 0, "tdb_transaction_start: cannot start a transaction with locks held\n"));
+		tdb->ecode = TDB_ERR_LOCK;
+		return -1;
+	}
+
+	if (tdb->travlocks.next != NULL) {
+		/* you cannot use transactions inside a traverse (although you can use
+		   traverse inside a transaction) as otherwise you can end up with
+		   deadlock */
+		TDB_LOG((tdb, 0, "tdb_transaction_start: cannot start a transaction within a traverse\n"));
+		tdb->ecode = TDB_ERR_LOCK;
+		return -1;
+	}
+
+	tdb->transaction = calloc(sizeof(struct tdb_transaction), 1);
+	if (tdb->transaction == NULL) {
+		tdb->ecode = TDB_ERR_OOM;
+		return -1;
+	}
+
+	/* get the transaction write lock. This is a blocking lock. As
+	   discussed with Volker, there are a number of ways we could
+	   make this async, which we will probably do in the future */
+	if (tdb_brlock_len(tdb, TRANSACTION_LOCK, F_WRLCK, F_SETLKW, 0, 1) == -1) {
+		TDB_LOG((tdb, 0, "tdb_transaction_start: failed to get transaction lock\n"));
+		tdb->ecode = TDB_ERR_LOCK;
+		SAFE_FREE(tdb->transaction);
+		return -1;
+	}
+	
+	/* get a read lock from the freelist to the end of file. This
+	   is upgraded to a write lock during the commit */
+	if (tdb_brlock_len(tdb, FREELIST_TOP, F_RDLCK, F_SETLKW, 0, 0) == -1) {
+		TDB_LOG((tdb, 0, "tdb_transaction_start: failed to get hash locks\n"));
+		tdb->ecode = TDB_ERR_LOCK;
+		goto fail;
+	}
+
+	/* setup a copy of the hash table heads so the hash scan in
+	   traverse can be fast */
+	tdb->transaction->hash_heads = calloc(tdb->header.hash_size+1, sizeof(tdb_off_t));
+	if (tdb->transaction->hash_heads == NULL) {
+		tdb->ecode = TDB_ERR_OOM;
+		goto fail;
+	}
+	if (tdb->methods->tdb_read(tdb, FREELIST_TOP, tdb->transaction->hash_heads,
+				   TDB_HASHTABLE_SIZE(tdb), 0) != 0) {
+		TDB_LOG((tdb, 0, "tdb_transaction_start: failed to read hash heads\n"));
+		tdb->ecode = TDB_ERR_IO;
+		goto fail;
+	}
+
+	/* make sure we know about any file expansions already done by
+	   anyone else */
+	tdb->methods->tdb_oob(tdb, tdb->map_size + 1, 1);
+	tdb->transaction->old_map_size = tdb->map_size;
+
+	/* finally hook the io methods, replacing them with
+	   transaction specific methods */
+	tdb->transaction->io_methods = tdb->methods;
+	tdb->methods = &transaction_methods;
+
+	/* by calling this transaction write here, we ensure that we don't grow the
+	   transaction linked list due to hash table updates */
+	if (transaction_write(tdb, FREELIST_TOP, tdb->transaction->hash_heads, 
+			      TDB_HASHTABLE_SIZE(tdb)) != 0) {
+		TDB_LOG((tdb, 0, "tdb_transaction_start: failed to prime hash table\n"));
+		tdb->ecode = TDB_ERR_IO;
+		goto fail;
+	}
+
+	return 0;
+	
+fail:
+	tdb_brlock_len(tdb, FREELIST_TOP, F_UNLCK, F_SETLKW, 0, 0);
+	tdb_brlock_len(tdb, TRANSACTION_LOCK, F_UNLCK, F_SETLKW, 0, 1);
+	SAFE_FREE(tdb->transaction->hash_heads);
+	SAFE_FREE(tdb->transaction);
+	return -1;
+}
+
+
+/*
+  cancel the current transaction
+*/
+int tdb_transaction_cancel(struct tdb_context *tdb)
+{	
+	if (tdb->transaction == NULL) {
+		TDB_LOG((tdb, 0, "tdb_transaction_cancel: no transaction\n"));
+		return -1;
+	}
+
+	if (tdb->transaction->nesting != 0) {
+		tdb->transaction->transaction_error = 1;
+		tdb->transaction->nesting--;
+		return 0;
+	}		
+
+	tdb->map_size = tdb->transaction->old_map_size;
+
+	/* free all the transaction elements */
+	while (tdb->transaction->elements) {
+		struct tdb_transaction_el *el = tdb->transaction->elements;
+		tdb->transaction->elements = el->next;
+		free(el->data);
+		free(el);
+	}
+
+	/* remove any locks created during the transaction */
+	if (tdb->num_locks != 0) {
+		int h;
+		for (h=0;h<tdb->header.hash_size+1;h++) {
+			if (tdb->locked[h].count != 0) {
+				tdb_brlock_len(tdb,FREELIST_TOP+4*h,F_UNLCK,F_SETLKW, 0, 1);
+				tdb->locked[h].count = 0;
+			}
+		}
+		tdb->num_locks = 0;
+	}
+
+	/* restore the normal io methods */
+	tdb->methods = tdb->transaction->io_methods;
+
+	tdb_brlock_len(tdb, FREELIST_TOP, F_UNLCK, F_SETLKW, 0, 0);
+	tdb_brlock_len(tdb, TRANSACTION_LOCK, F_UNLCK, F_SETLKW, 0, 1);
+	SAFE_FREE(tdb->transaction->hash_heads);
+	SAFE_FREE(tdb->transaction);
+	
+	return 0;
+}
+
+/*
+  sync to disk
+*/
+static int transaction_sync(struct tdb_context *tdb, tdb_off_t offset, tdb_len_t length)
+{	
+	if (fsync(tdb->fd) != 0) {
+		tdb->ecode = TDB_ERR_IO;
+		TDB_LOG((tdb, 0, "tdb_transaction: fsync failed\n"));
+		return -1;
+	}
+#ifdef MS_SYNC
+	if (tdb->map_ptr) {
+		tdb_off_t moffset = offset & ~(tdb->page_size-1);
+		if (msync(moffset + (char *)tdb->map_ptr, 
+			  length + (offset - moffset), MS_SYNC) != 0) {
+			tdb->ecode = TDB_ERR_IO;
+			TDB_LOG((tdb, 0, "tdb_transaction: msync failed - %s\n",
+				 strerror(errno)));
+			return -1;
+		}
+	}
+#endif
+	return 0;
+}
+
+
+/*
+  work out how much space the linearised recovery data will consume
+*/
+static tdb_len_t tdb_recovery_size(struct tdb_context *tdb)
+{
+	struct tdb_transaction_el *el;
+	tdb_len_t recovery_size = 0;
+
+	recovery_size = sizeof(u32);
+	for (el=tdb->transaction->elements;el;el=el->next) {
+		if (el->offset >= tdb->transaction->old_map_size) {
+			continue;
+		}
+		recovery_size += 2*sizeof(tdb_off_t) + el->length;
+	}
+
+	return recovery_size;
+}
+
+/*
+  allocate the recovery area, or use an existing recovery area if it is
+  large enough
+*/
+static int tdb_recovery_allocate(struct tdb_context *tdb, 
+				 tdb_len_t *recovery_size,
+				 tdb_off_t *recovery_offset,
+				 tdb_len_t *recovery_max_size)
+{
+	struct list_struct rec;
+	const struct tdb_methods *methods = tdb->transaction->io_methods;
+	tdb_off_t recovery_head;
+
+	if (tdb_ofs_read(tdb, TDB_RECOVERY_HEAD, &recovery_head) == -1) {
+		TDB_LOG((tdb, 0, "tdb_recovery_allocate: failed to read recovery head\n"));
+		return -1;
+	}
+
+	rec.rec_len = 0;
+
+	if (recovery_head != 0 && 
+	    methods->tdb_read(tdb, recovery_head, &rec, sizeof(rec), DOCONV()) == -1) {
+		TDB_LOG((tdb, 0, "tdb_recovery_allocate: failed to read recovery record\n"));
+		return -1;
+	}
+
+	*recovery_size = tdb_recovery_size(tdb);
+
+	if (recovery_head != 0 && *recovery_size <= rec.rec_len) {
+		/* it fits in the existing area */
+		*recovery_max_size = rec.rec_len;
+		*recovery_offset = recovery_head;
+		return 0;
+	}
+
+	/* we need to free up the old recovery area, then allocate a
+	   new one at the end of the file. Note that we cannot use
+	   tdb_allocate() to allocate the new one as that might return
+	   us an area that is being currently used (as of the start of
+	   the transaction) */
+	if (recovery_head != 0) {
+		if (tdb_free(tdb, recovery_head, &rec) == -1) {
+			TDB_LOG((tdb, 0, "tdb_recovery_allocate: failed to free previous recovery area\n"));
+			return -1;
+		}
+	}
+
+	/* the tdb_free() call might have increased the recovery size */
+	*recovery_size = tdb_recovery_size(tdb);
+
+	/* round up to a multiple of page size */
+	*recovery_max_size = TDB_ALIGN(sizeof(rec) + *recovery_size, tdb->page_size) - sizeof(rec);
+	*recovery_offset = tdb->map_size;
+	recovery_head = *recovery_offset;
+
+	if (methods->tdb_expand_file(tdb, tdb->transaction->old_map_size, 
+				     (tdb->map_size - tdb->transaction->old_map_size) +
+				     sizeof(rec) + *recovery_max_size) == -1) {
+		TDB_LOG((tdb, 0, "tdb_recovery_allocate: failed to create recovery area\n"));
+		return -1;
+	}
+
+	/* remap the file (if using mmap) */
+	methods->tdb_oob(tdb, tdb->map_size + 1, 1);
+
+	/* we have to reset the old map size so that we don't try to expand the file
+	   again in the transaction commit, which would destroy the recovery area */
+	tdb->transaction->old_map_size = tdb->map_size;
+
+	/* write the recovery header offset and sync - we can sync without a race here
+	   as the magic ptr in the recovery record has not been set */
+	CONVERT(recovery_head);
+	if (methods->tdb_write(tdb, TDB_RECOVERY_HEAD, 
+			       &recovery_head, sizeof(tdb_off_t)) == -1) {
+		TDB_LOG((tdb, 0, "tdb_recovery_allocate: failed to write recovery head\n"));
+		return -1;
+	}
+
+	return 0;
+}
+
+
+/*
+  setup the recovery data that will be used on a crash during commit
+*/
+static int transaction_setup_recovery(struct tdb_context *tdb, 
+				      tdb_off_t *magic_offset)
+{
+	struct tdb_transaction_el *el;
+	tdb_len_t recovery_size;
+	unsigned char *data, *p;
+	const struct tdb_methods *methods = tdb->transaction->io_methods;
+	struct list_struct *rec;
+	tdb_off_t recovery_offset, recovery_max_size;
+	tdb_off_t old_map_size = tdb->transaction->old_map_size;
+	u32 magic, tailer;
+
+	/*
+	  check that the recovery area has enough space
+	*/
+	if (tdb_recovery_allocate(tdb, &recovery_size, 
+				  &recovery_offset, &recovery_max_size) == -1) {
+		return -1;
+	}
+
+	data = malloc(recovery_size + sizeof(*rec));
+	if (data == NULL) {
+		tdb->ecode = TDB_ERR_OOM;
+		return -1;
+	}
+
+	rec = (struct list_struct *)data;
+	memset(rec, 0, sizeof(*rec));
+
+	rec->magic    = 0;
+	rec->data_len = recovery_size;
+	rec->rec_len  = recovery_max_size;
+	rec->key_len  = old_map_size;
+	CONVERT(rec);
+
+	/* build the recovery data into a single blob to allow us to do a single
+	   large write, which should be more efficient */
+	p = data + sizeof(*rec);
+	for (el=tdb->transaction->elements;el;el=el->next) {
+		if (el->offset >= old_map_size) {
+			continue;
+		}
+		if (el->offset + el->length > tdb->transaction->old_map_size) {
+			TDB_LOG((tdb, 0, "tdb_transaction_commit: transaction data over new region boundary\n"));
+			free(data);
+			tdb->ecode = TDB_ERR_CORRUPT;
+			return -1;
+		}
+		memcpy(p, &el->offset, 4);
+		memcpy(p+4, &el->length, 4);
+		if (DOCONV()) {
+			tdb_convert(p, 8);
+		}
+		/* the recovery area contains the old data, not the
+		   new data, so we have to call the original tdb_read
+		   method to get it */
+		if (methods->tdb_read(tdb, el->offset, p + 8, el->length, 0) != 0) {
+			free(data);
+			tdb->ecode = TDB_ERR_IO;
+			return -1;
+		}
+		p += 8 + el->length;
+	}
+
+	/* and the tailer */
+	tailer = sizeof(*rec) + recovery_max_size;
+	memcpy(p, &tailer, 4);
+	CONVERT(p);
+
+	/* write the recovery data to the recovery area */
+	if (methods->tdb_write(tdb, recovery_offset, data, sizeof(*rec) + recovery_size) == -1) {
+		TDB_LOG((tdb, 0, "tdb_transaction_commit: failed to write recovery data\n"));
+		free(data);
+		tdb->ecode = TDB_ERR_IO;
+		return -1;
+	}
+
+	/* as we don't have ordered writes, we have to sync the recovery
+	   data before we update the magic to indicate that the recovery
+	   data is present */
+	if (transaction_sync(tdb, recovery_offset, sizeof(*rec) + recovery_size) == -1) {
+		free(data);
+		return -1;
+	}
+
+	free(data);
+
+	magic = TDB_RECOVERY_MAGIC;
+	CONVERT(magic);
+
+	*magic_offset = recovery_offset + offsetof(struct list_struct, magic);
+
+	if (methods->tdb_write(tdb, *magic_offset, &magic, sizeof(magic)) == -1) {
+		TDB_LOG((tdb, 0, "tdb_transaction_commit: failed to write recovery magic\n"));
+		tdb->ecode = TDB_ERR_IO;
+		return -1;
+	}
+
+	/* ensure the recovery magic marker is on disk */
+	if (transaction_sync(tdb, *magic_offset, sizeof(magic)) == -1) {
+		return -1;
+	}
+
+	return 0;
+}
+
+/*
+  commit the current transaction
+*/
+int tdb_transaction_commit(struct tdb_context *tdb)
+{	
+	const struct tdb_methods *methods;
+	tdb_off_t magic_offset = 0;
+	u32 zero = 0;
+
+	if (tdb->transaction == NULL) {
+		TDB_LOG((tdb, 0, "tdb_transaction_commit: no transaction\n"));
+		return -1;
+	}
+
+	if (tdb->transaction->transaction_error) {
+		tdb->ecode = TDB_ERR_IO;
+		tdb_transaction_cancel(tdb);
+		TDB_LOG((tdb, 0, "tdb_transaction_commit: transaction error pending\n"));
+		return -1;
+	}
+
+	if (tdb->transaction->nesting != 0) {
+		tdb->transaction->nesting--;
+		return 0;
+	}		
+
+	/* check for a null transaction */
+	if (tdb->transaction->elements == NULL) {
+		tdb_transaction_cancel(tdb);
+		return 0;
+	}
+
+	methods = tdb->transaction->io_methods;
+	
+	/* if there are any locks pending then the caller has not
+	   nested their locks properly, so fail the transaction */
+	if (tdb->num_locks) {
+		tdb->ecode = TDB_ERR_LOCK;
+		TDB_LOG((tdb, 0, "tdb_transaction_commit: locks pending on commit\n"));
+		tdb_transaction_cancel(tdb);
+		return -1;
+	}
+
+	/* upgrade the main transaction lock region to a write lock */
+	if (tdb_brlock_upgrade(tdb, FREELIST_TOP, 0) == -1) {
+		TDB_LOG((tdb, 0, "tdb_transaction_start: failed to upgrade hash locks\n"));
+		tdb->ecode = TDB_ERR_LOCK;
+		tdb_transaction_cancel(tdb);
+		return -1;
+	}
+
+	/* get the global lock - this prevents new users attaching to the database
+	   during the commit */
+	if (tdb_brlock_len(tdb, GLOBAL_LOCK, F_WRLCK, F_SETLKW, 0, 1) == -1) {
+		TDB_LOG((tdb, 0, "tdb_transaction_commit: failed to get global lock\n"));
+		tdb->ecode = TDB_ERR_LOCK;
+		tdb_transaction_cancel(tdb);
+		return -1;
+	}
+
+	if (!(tdb->flags & TDB_NOSYNC)) {
+		/* write the recovery data to the end of the file */
+		if (transaction_setup_recovery(tdb, &magic_offset) == -1) {
+			TDB_LOG((tdb, 0, "tdb_transaction_commit: failed to setup recovery data\n"));
+			tdb_brlock_len(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
+			tdb_transaction_cancel(tdb);
+			return -1;
+		}
+	}
+
+	/* expand the file to the new size if needed */
+	if (tdb->map_size != tdb->transaction->old_map_size) {
+		if (methods->tdb_expand_file(tdb, tdb->transaction->old_map_size, 
+					     tdb->map_size - 
+					     tdb->transaction->old_map_size) == -1) {
+			tdb->ecode = TDB_ERR_IO;
+			TDB_LOG((tdb, 0, "tdb_transaction_commit: expansion failed\n"));
+			tdb_brlock_len(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
+			tdb_transaction_cancel(tdb);
+			return -1;
+		}
+		tdb->map_size = tdb->transaction->old_map_size;
+		methods->tdb_oob(tdb, tdb->map_size + 1, 1);
+	}
+
+	/* perform all the writes */
+	while (tdb->transaction->elements) {
+		struct tdb_transaction_el *el = tdb->transaction->elements;
+
+		if (methods->tdb_write(tdb, el->offset, el->data, el->length) == -1) {
+			TDB_LOG((tdb, 0, "tdb_transaction_commit: write failed during commit\n"));
+			
+			/* we've overwritten part of the data and
+			   possibly expanded the file, so we need to
+			   run the crash recovery code */
+			tdb->methods = methods;
+			tdb_transaction_recover(tdb); 
+
+			tdb_transaction_cancel(tdb);
+			tdb_brlock_len(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
+
+			TDB_LOG((tdb, 0, "tdb_transaction_commit: write failed\n"));
+			return -1;
+		}
+		tdb->transaction->elements = el->next;
+		free(el->data); 
+		free(el);
+	} 
+
+	if (!(tdb->flags & TDB_NOSYNC)) {
+		/* ensure the new data is on disk */
+		if (transaction_sync(tdb, 0, tdb->map_size) == -1) {
+			return -1;
+		}
+
+		/* remove the recovery marker */
+		if (methods->tdb_write(tdb, magic_offset, &zero, 4) == -1) {
+			TDB_LOG((tdb, 0, "tdb_transaction_commit: failed to remove recovery magic\n"));
+			return -1;
+		}
+
+		/* ensure the recovery marker has been removed on disk */
+		if (transaction_sync(tdb, magic_offset, 4) == -1) {
+			return -1;
+		}
+	}
+
+	tdb_brlock_len(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
+
+	/*
+	  TODO: maybe write to some dummy hdr field, or write to magic
+	  offset without mmap, before the last sync, instead of the
+	  utime() call
+	*/
+
+	/* on some systems (like Linux 2.6.x) changes via mmap/msync
+	   don't change the mtime of the file, this means the file may
+	   not be backed up (as tdb rounding to block sizes means that
+	   file size changes are quite rare too). The following forces
+	   mtime changes when a transaction completes */
+#ifdef HAVE_UTIME
+	utime(tdb->name, NULL);
+#endif
+
+	/* use a transaction cancel to free memory and remove the
+	   transaction locks */
+	tdb_transaction_cancel(tdb);
+	return 0;
+}
+
+
+/*
+  recover from an aborted transaction. Must be called with exclusive
+  database write access already established (including the global
+  lock to prevent new processes attaching)
+*/
+int tdb_transaction_recover(struct tdb_context *tdb)
+{
+	tdb_off_t recovery_head, recovery_eof;
+	unsigned char *data, *p;
+	u32 zero = 0;
+	struct list_struct rec;
+
+	/* find the recovery area */
+	if (tdb_ofs_read(tdb, TDB_RECOVERY_HEAD, &recovery_head) == -1) {
+		TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to read recovery head\n"));
+		tdb->ecode = TDB_ERR_IO;
+		return -1;
+	}
+
+	if (recovery_head == 0) {
+		/* we have never allocated a recovery record */
+		return 0;
+	}
+
+	/* read the recovery record */
+	if (tdb->methods->tdb_read(tdb, recovery_head, &rec, 
+				   sizeof(rec), DOCONV()) == -1) {
+		TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to read recovery record\n"));		
+		tdb->ecode = TDB_ERR_IO;
+		return -1;
+	}
+
+	if (rec.magic != TDB_RECOVERY_MAGIC) {
+		/* there is no valid recovery data */
+		return 0;
+	}
+
+	if (tdb->read_only) {
+		TDB_LOG((tdb, 0, "tdb_transaction_recover: attempt to recover read only database\n"));
+		tdb->ecode = TDB_ERR_CORRUPT;
+		return -1;
+	}
+
+	recovery_eof = rec.key_len;
+
+	data = malloc(rec.data_len);
+	if (data == NULL) {
+		TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to allocate recovery data\n"));		
+		tdb->ecode = TDB_ERR_OOM;
+		return -1;
+	}
+
+	/* read the full recovery data */
+	if (tdb->methods->tdb_read(tdb, recovery_head + sizeof(rec), data,
+				   rec.data_len, 0) == -1) {
+		TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to read recovery data\n"));		
+		tdb->ecode = TDB_ERR_IO;
+		return -1;
+	}
+
+	/* recover the file data */
+	p = data;
+	while (p+8 < data + rec.data_len) {
+		u32 ofs, len;
+		if (DOCONV()) {
+			tdb_convert(p, 8);
+		}
+		memcpy(&ofs, p, 4);
+		memcpy(&len, p+4, 4);
+
+		if (tdb->methods->tdb_write(tdb, ofs, p+8, len) == -1) {
+			free(data);
+			TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to recover %d bytes at offset %d\n", len, ofs));
+			tdb->ecode = TDB_ERR_IO;
+			return -1;
+		}
+		p += 8 + len;
+	}
+
+	free(data);
+
+	if (transaction_sync(tdb, 0, tdb->map_size) == -1) {
+		TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to sync recovery\n"));
+		tdb->ecode = TDB_ERR_IO;
+		return -1;
+	}
+
+	/* if the recovery area is after the recovered eof then remove it */
+	if (recovery_eof <= recovery_head) {
+		if (tdb_ofs_write(tdb, TDB_RECOVERY_HEAD, &zero) == -1) {
+			TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to remove recovery head\n"));
+			tdb->ecode = TDB_ERR_IO;
+			return -1;			
+		}
+	}
+
+	/* remove the recovery magic */
+	if (tdb_ofs_write(tdb, recovery_head + offsetof(struct list_struct, magic), 
+			  &zero) == -1) {
+		TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to remove recovery magic\n"));
+		tdb->ecode = TDB_ERR_IO;
+		return -1;			
+	}
+	
+	/* reduce the file size to the old size */
+	tdb_munmap(tdb);
+	if (ftruncate(tdb->fd, recovery_eof) != 0) {
+		TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to reduce to recovery size\n"));
+		tdb->ecode = TDB_ERR_IO;
+		return -1;			
+	}
+	tdb->map_size = recovery_eof;
+	tdb_mmap(tdb);
+
+	if (transaction_sync(tdb, 0, recovery_eof) == -1) {
+		TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to sync2 recovery\n"));
+		tdb->ecode = TDB_ERR_IO;
+		return -1;
+	}
+
+	TDB_LOG((tdb, 0, "tdb_transaction_recover: recovered %d byte database\n", 
+		 recovery_eof));
+
+	/* all done */
+	return 0;
+}
diff --git a/source/tdb/traverse.c b/source/tdb/traverse.c
new file mode 100644
index 00000000000..d070e0a7662
--- /dev/null
+++ b/source/tdb/traverse.c
@@ -0,0 +1,334 @@
+ /* 
+   Unix SMB/CIFS implementation.
+
+   trivial database library
+
+   Copyright (C) Andrew Tridgell              1999-2005
+   Copyright (C) Paul `Rusty' Russell		   2000
+   Copyright (C) Jeremy Allison			   2000-2003
+   
+     ** NOTE! The following LGPL license applies to the tdb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+   
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "tdb_private.h"
+
+/* Uses traverse lock: 0 = finish, -1 = error, other = record offset */
+static int tdb_next_lock(struct tdb_context *tdb, struct tdb_traverse_lock *tlock,
+			 struct list_struct *rec)
+{
+	int want_next = (tlock->off != 0);
+
+	/* Lock each chain from the start one. */
+	for (; tlock->hash < tdb->header.hash_size; tlock->hash++) {
+		if (!tlock->off && tlock->hash != 0) {
+			/* this is an optimisation for the common case where
+			   the hash chain is empty, which is particularly
+			   common for the use of tdb with ldb, where large
+			   hashes are used. In that case we spend most of our
+			   time in tdb_brlock(), locking empty hash chains.
+			   
+			   To avoid this, we do an unlocked pre-check to see
+			   if the hash chain is empty before starting to look
+			   inside it. If it is empty then we can avoid that
+			   hash chain. If it isn't empty then we can't believe
+			   the value we get back, as we read it without a
+			   lock, so instead we get the lock and re-fetch the
+			   value below.
+			   
+			   Notice that not doing this optimisation on the
+			   first hash chain is critical. We must guarantee
+			   that we have done at least one fcntl lock at the
+			   start of a search to guarantee that memory is
+			   coherent on SMP systems. If records are added by
+			   others during the search then thats OK, and we
+			   could possibly miss those with this trick, but we
+			   could miss them anyway without this trick, so the
+			   semantics don't change.
+			   
+			   With a non-indexed ldb search this trick gains us a
+			   factor of around 80 in speed on a linux 2.6.x
+			   system (testing using ldbtest).
+			*/
+			tdb->methods->next_hash_chain(tdb, &tlock->hash);
+			if (tlock->hash == tdb->header.hash_size) {
+				continue;
+			}
+		}
+
+		if (tdb_lock(tdb, tlock->hash, tlock->lock_rw) == -1)
+			return -1;
+
+		/* No previous record?  Start at top of chain. */
+		if (!tlock->off) {
+			if (tdb_ofs_read(tdb, TDB_HASH_TOP(tlock->hash),
+				     &tlock->off) == -1)
+				goto fail;
+		} else {
+			/* Otherwise unlock the previous record. */
+			if (tdb_unlock_record(tdb, tlock->off) != 0)
+				goto fail;
+		}
+
+		if (want_next) {
+			/* We have offset of old record: grab next */
+			if (tdb_rec_read(tdb, tlock->off, rec) == -1)
+				goto fail;
+			tlock->off = rec->next;
+		}
+
+		/* Iterate through chain */
+		while( tlock->off) {
+			tdb_off_t current;
+			if (tdb_rec_read(tdb, tlock->off, rec) == -1)
+				goto fail;
+
+			/* Detect infinite loops. From "Shlomi Yaakobovich" <Shlomi@exanet.com>. */
+			if (tlock->off == rec->next) {
+				TDB_LOG((tdb, 0, "tdb_next_lock: loop detected.\n"));
+				goto fail;
+			}
+
+			if (!TDB_DEAD(rec)) {
+				/* Woohoo: we found one! */
+				if (tdb_lock_record(tdb, tlock->off) != 0)
+					goto fail;
+				return tlock->off;
+			}
+
+			/* Try to clean dead ones from old traverses */
+			current = tlock->off;
+			tlock->off = rec->next;
+			if (!(tdb->read_only || tdb->traverse_read) && 
+			    tdb_do_delete(tdb, current, rec) != 0)
+				goto fail;
+		}
+		tdb_unlock(tdb, tlock->hash, tlock->lock_rw);
+		want_next = 0;
+	}
+	/* We finished iteration without finding anything */
+	return TDB_ERRCODE(TDB_SUCCESS, 0);
+
+ fail:
+	tlock->off = 0;
+	if (tdb_unlock(tdb, tlock->hash, tlock->lock_rw) != 0)
+		TDB_LOG((tdb, 0, "tdb_next_lock: On error unlock failed!\n"));
+	return -1;
+}
+
+/* traverse the entire database - calling fn(tdb, key, data) on each element.
+   return -1 on error or the record count traversed
+   if fn is NULL then it is not called
+   a non-zero return value from fn() indicates that the traversal should stop
+  */
+static int tdb_traverse_internal(struct tdb_context *tdb, 
+				 tdb_traverse_func fn, void *private_data,
+				 struct tdb_traverse_lock *tl)
+{
+	TDB_DATA key, dbuf;
+	struct list_struct rec;
+	int ret, count = 0;
+
+	/* This was in the initializaton, above, but the IRIX compiler
+	 * did not like it.  crh
+	 */
+	tl->next = tdb->travlocks.next;
+
+	/* fcntl locks don't stack: beware traverse inside traverse */
+	tdb->travlocks.next = tl;
+
+	/* tdb_next_lock places locks on the record returned, and its chain */
+	while ((ret = tdb_next_lock(tdb, tl, &rec)) > 0) {
+		count++;
+		/* now read the full record */
+		key.dptr = tdb_alloc_read(tdb, tl->off + sizeof(rec), 
+					  rec.key_len + rec.data_len);
+		if (!key.dptr) {
+			ret = -1;
+			if (tdb_unlock(tdb, tl->hash, tl->lock_rw) != 0)
+				goto out;
+			if (tdb_unlock_record(tdb, tl->off) != 0)
+				TDB_LOG((tdb, 0, "tdb_traverse: key.dptr == NULL and unlock_record failed!\n"));
+			goto out;
+		}
+		key.dsize = rec.key_len;
+		dbuf.dptr = key.dptr + rec.key_len;
+		dbuf.dsize = rec.data_len;
+
+		/* Drop chain lock, call out */
+		if (tdb_unlock(tdb, tl->hash, tl->lock_rw) != 0) {
+			ret = -1;
+			SAFE_FREE(key.dptr);
+			goto out;
+		}
+		if (fn && fn(tdb, key, dbuf, private_data)) {
+			/* They want us to terminate traversal */
+			ret = count;
+			if (tdb_unlock_record(tdb, tl->off) != 0) {
+				TDB_LOG((tdb, 0, "tdb_traverse: unlock_record failed!\n"));;
+				ret = -1;
+			}
+			SAFE_FREE(key.dptr);
+			goto out;
+		}
+		SAFE_FREE(key.dptr);
+	}
+out:
+	tdb->travlocks.next = tl->next;
+	if (ret < 0)
+		return -1;
+	else
+		return count;
+}
+
+
+/*
+  a write style traverse - temporarily marks the db read only
+*/
+int tdb_traverse_read(struct tdb_context *tdb, 
+		      tdb_traverse_func fn, void *private_data)
+{
+	struct tdb_traverse_lock tl = { NULL, 0, 0, F_RDLCK };
+	int ret;
+	
+	/* we need to get a read lock on the transaction lock here to
+	   cope with the lock ordering semantics of solaris10 */
+	if (tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, F_RDLCK, F_SETLKW, 0) == -1) {
+		TDB_LOG((tdb, 0, "tdb_traverse_read: failed to get transaction lock\n"));
+		tdb->ecode = TDB_ERR_LOCK;
+		return -1;
+	}
+
+	tdb->traverse_read++;
+	ret = tdb_traverse_internal(tdb, fn, private_data, &tl);
+	tdb->traverse_read--;
+
+	tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, F_UNLCK, F_SETLKW, 0);
+
+	return ret;
+}
+
+/*
+  a write style traverse - needs to get the transaction lock to
+  prevent deadlocks
+*/
+int tdb_traverse(struct tdb_context *tdb, 
+		 tdb_traverse_func fn, void *private_data)
+{
+	struct tdb_traverse_lock tl = { NULL, 0, 0, F_WRLCK };
+	int ret;
+
+	if (tdb->read_only || tdb->traverse_read) {
+		return tdb_traverse_read(tdb, fn, private_data);
+	}
+	
+	if (tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, F_WRLCK, F_SETLKW, 0) == -1) {
+		TDB_LOG((tdb, 0, "tdb_traverse: failed to get transaction lock\n"));
+		tdb->ecode = TDB_ERR_LOCK;
+		return -1;
+	}
+
+	ret = tdb_traverse_internal(tdb, fn, private_data, &tl);
+
+	tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, F_UNLCK, F_SETLKW, 0);
+
+	return ret;
+}
+
+
+/* find the first entry in the database and return its key */
+TDB_DATA tdb_firstkey(struct tdb_context *tdb)
+{
+	TDB_DATA key;
+	struct list_struct rec;
+
+	/* release any old lock */
+	if (tdb_unlock_record(tdb, tdb->travlocks.off) != 0)
+		return tdb_null;
+	tdb->travlocks.off = tdb->travlocks.hash = 0;
+
+	if (tdb_next_lock(tdb, &tdb->travlocks, &rec) <= 0)
+		return tdb_null;
+	/* now read the key */
+	key.dsize = rec.key_len;
+	key.dptr =tdb_alloc_read(tdb,tdb->travlocks.off+sizeof(rec),key.dsize);
+	if (tdb_unlock(tdb, BUCKET(tdb->travlocks.hash), F_WRLCK) != 0)
+		TDB_LOG((tdb, 0, "tdb_firstkey: error occurred while tdb_unlocking!\n"));
+	return key;
+}
+
+/* find the next entry in the database, returning its key */
+TDB_DATA tdb_nextkey(struct tdb_context *tdb, TDB_DATA oldkey)
+{
+	u32 oldhash;
+	TDB_DATA key = tdb_null;
+	struct list_struct rec;
+	char *k = NULL;
+
+	/* Is locked key the old key?  If so, traverse will be reliable. */
+	if (tdb->travlocks.off) {
+		if (tdb_lock(tdb,tdb->travlocks.hash,F_WRLCK))
+			return tdb_null;
+		if (tdb_rec_read(tdb, tdb->travlocks.off, &rec) == -1
+		    || !(k = tdb_alloc_read(tdb,tdb->travlocks.off+sizeof(rec),
+					    rec.key_len))
+		    || memcmp(k, oldkey.dptr, oldkey.dsize) != 0) {
+			/* No, it wasn't: unlock it and start from scratch */
+			if (tdb_unlock_record(tdb, tdb->travlocks.off) != 0) {
+				SAFE_FREE(k);
+				return tdb_null;
+			}
+			if (tdb_unlock(tdb, tdb->travlocks.hash, F_WRLCK) != 0) {
+				SAFE_FREE(k);
+				return tdb_null;
+			}
+			tdb->travlocks.off = 0;
+		}
+
+		SAFE_FREE(k);
+	}
+
+	if (!tdb->travlocks.off) {
+		/* No previous element: do normal find, and lock record */
+		tdb->travlocks.off = tdb_find_lock_hash(tdb, oldkey, tdb->hash_fn(&oldkey), F_WRLCK, &rec);
+		if (!tdb->travlocks.off)
+			return tdb_null;
+		tdb->travlocks.hash = BUCKET(rec.full_hash);
+		if (tdb_lock_record(tdb, tdb->travlocks.off) != 0) {
+			TDB_LOG((tdb, 0, "tdb_nextkey: lock_record failed (%s)!\n", strerror(errno)));
+			return tdb_null;
+		}
+	}
+	oldhash = tdb->travlocks.hash;
+
+	/* Grab next record: locks chain and returned record,
+	   unlocks old record */
+	if (tdb_next_lock(tdb, &tdb->travlocks, &rec) > 0) {
+		key.dsize = rec.key_len;
+		key.dptr = tdb_alloc_read(tdb, tdb->travlocks.off+sizeof(rec),
+					  key.dsize);
+		/* Unlock the chain of this new record */
+		if (tdb_unlock(tdb, tdb->travlocks.hash, F_WRLCK) != 0)
+			TDB_LOG((tdb, 0, "tdb_nextkey: WARNING tdb_unlock failed!\n"));
+	}
+	/* Unlock the chain of old record */
+	if (tdb_unlock(tdb, BUCKET(oldhash), F_WRLCK) != 0)
+		TDB_LOG((tdb, 0, "tdb_nextkey: WARNING tdb_unlock failed!\n"));
+	return key;
+}
+