r19426: merge nearly all the differences between Samba3 tdb and Samba4

tdb. This includes:

 - the new tdb_lockall and tdb_lockall_read code, which will be needed
   for the ldb speedups

 - the tdb logging changes. This is an intermediate step to keep the
   differences between the two branches small. The plan is still to
   move to a tdb_init()/tdb_set_logging_function()/tdb_attach() style
   of open which will make things much cleaner.

 - the updated test suites and standalone tdb build code

 - use libreplace headers

There are still some small differences I haven't merged. I'll discuss
those on the list.
(This used to be commit 48903c75edfaf75dbd3e9d052e615552cdff39b4)

17 files changed, 558 insertions, 406 deletions

diff --git a/source3/tdb/Makefile.in b/source3/tdb/Makefile.in
index 1c23aaeea7..e158752d43 100644
--- a/source3/tdb/Makefile.in
+++ b/source3/tdb/Makefile.in
@@ -1,54 +1,81 @@
+#!gmake
 #
 # Makefile for tdb directory
 #
 
-CFLAGS = -Iinclude @CFLAGS@
 CC = @CC@
 prefix = @prefix@
 exec_prefix = @exec_prefix@
 bindir = @bindir@
 includedir = @includedir@
 libdir = @libdir@
+VPATH = @srcdir@:@libreplacedir@
+srcdir = @srcdir@
+builddir = @builddir@
+CFLAGS = -I$(srcdir)/include -Iinclude -I@libreplacedir@ @CFLAGS@
+
+.PHONY: test
 
 PROGS = bin/tdbtool bin/tdbtorture
-TDB_OBJ = common/tdb.o common/dump.o common/io.o common/lock.o \
-	common/open.o common/traverse.o common/freelist.o common/error.o \
-	common/transaction.o
+TDB_OBJ = @TDBOBJ@ @LIBREPLACEOBJ@
+
+DIRS = bin common tools
 
-all: $(PROGS)
+all: showflags dirs $(PROGS)
+
+showflags:
+	@echo 'tdb will be compiled with flags:'
+	@echo '  CFLAGS = $(CFLAGS)'
+	@echo '  LIBS = $(LIBS)'
 
 .c.o:
 	@echo Compiling $*.c
+	@mkdir -p `dirname $@`
 	@$(CC) $(CFLAGS) -c $< -o $@
 
+dirs:
+	@mkdir -p $(DIRS)
+
 install: all
 	mkdir -p $(bindir)
 	mkdir -p $(includedir)
 	mkdir -p $(libdir) 
 	mkdir -p $(libdir)/pkgconfig
 	cp $(PROGS) $(bindir)
-	cp include/tdb.h $(includedir)
+	cp $(srcdir)/include/tdb.h $(includedir)
 	cp tdb.pc $(libdir)/pkgconfig
 
-bin/tdbtest: tools/tdbtest.o $(TDB_OBJ)
-	$(CC) $(CFLAGS) -o bin/tdbtest tools/tdbtest.o $(TDB_OBJ) -lgdbm
+libtdb.a: $(TDB_OBJ)
+	ar -rv libtdb.a $(TDB_OBJ)
+
+bin/tdbtest: tools/tdbtest.o libtdb.a
+	$(CC) $(CFLAGS) -o bin/tdbtest tools/tdbtest.o -L. -ltdb -lgdbm
+
+bin/tdbtool: tools/tdbtool.o libtdb.a
+	$(CC) $(CFLAGS) -o bin/tdbtool tools/tdbtool.o -L. -ltdb
+
+bin/tdbtorture: tools/tdbtorture.o libtdb.a
+	$(CC) $(CFLAGS) -o bin/tdbtorture tools/tdbtorture.o -L. -ltdb
 
-bin/tdbtool: tools/tdbtool.o $(TDB_OBJ)
-	$(CC) $(CFLAGS) -o bin/tdbtool tools/tdbtool.o $(TDB_OBJ)
+bin/tdbdump: tools/tdbdump.o libtdb.a
+	$(CC) $(CFLAGS) -o bin/tdbdump tools/tdbdump.o -L. -ltdb
 
-bin/tdbtorture: tools/tdbtorture.o $(TDB_OBJ)
-	$(CC) $(CFLAGS) -o bin/tdbtorture tools/tdbtorture.o $(TDB_OBJ)
+bin/tdbbackup: tools/tdbbackup.o libtdb.a
+	$(CC) $(CFLAGS) -o bin/tdbbackup tools/tdbbackup.o -L. -ltdb
 
-bin/tdbdump: tools/tdbdump.o $(TDB_OBJ)
-	$(CC) $(CFLAGS) -o bin/tdbdump tools/tdbdump.o $(TDB_OBJ)
+test: bin/tdbtorture
+	bin/tdbtorture
 
-bin/tdbbackup: tools/tdbbackup.o $(TDB_OBJ)
-	$(CC) $(CFLAGS) -o bin/tdbbackup tools/tdbbackup.o $(TDB_OBJ)
+installcheck: test install
 
 clean:
-	rm -f $(PROGS) common/*.o tools/*.o *~ *.bak */*~ */*.bak *% core test.db test.tdb test.gdbm
+	rm -f $(PROGS) *.o *.a common/*.o tools/*.o tdb.pc
+	rm -f test.db test.tdb torture.tdb test.gdbm
 
-installcheck: install
-	$(bindir)/tdbtorture
+distclean: clean
+	rm -f *~ */*~
+	rm -f config.log config.status include/config.h config.cache
+	rm -f Makefile
 
-test: installcheck
+realdistclean: distclean
+	rm -f configure include/config.h.in
diff --git a/source3/tdb/aclocal.m4 b/source3/tdb/aclocal.m4
index cdc2a2fa52..5605e476ba 100644
--- a/source3/tdb/aclocal.m4
+++ b/source3/tdb/aclocal.m4
@@ -1,12 +1 @@
-dnl see if a declaration exists for a function or variable
-dnl defines HAVE_function_DECL if it exists
-dnl AC_HAVE_DECL(var, includes)
-AC_DEFUN(AC_HAVE_DECL,
-[
- AC_CACHE_CHECK([for $1 declaration],ac_cv_have_$1_decl,[
-    AC_TRY_COMPILE([$2],[int i = (int)$1],
-        ac_cv_have_$1_decl=yes,ac_cv_have_$1_decl=no)])
- if test x"$ac_cv_have_$1_decl" = x"yes"; then
-    AC_DEFINE([HAVE_]translit([$1], [a-z], [A-Z])[_DECL],1,[Whether $1() is available])
- fi
-])
+m4_include(libreplace.m4)
diff --git a/source3/tdb/autogen.sh b/source3/tdb/autogen.sh
index 89053ad9f8..bf84eeee19 100755
--- a/source3/tdb/autogen.sh
+++ b/source3/tdb/autogen.sh
@@ -1,7 +1,13 @@
 #!/bin/sh
 
-autoheader || exit 1
-autoconf || exit 1
+rm -rf autom4te.cache
+rm -f configure config.h.in
+
+IPATHS="-I libreplace -I lib/replace -I ../libreplace -I ../replace"
+autoconf $IPATHS || exit 1
+autoheader $IPATHS || exit 1
+
+rm -rf autom4te.cache
 
 echo "Now run ./configure and then make."
 exit 0
diff --git a/source3/tdb/common/freelist.c b/source3/tdb/common/freelist.c
index 3483751164..9d1ae59801 100644
--- a/source3/tdb/common/freelist.c
+++ b/source3/tdb/common/freelist.c
@@ -37,7 +37,7 @@ static int rec_free_read(struct tdb_context *tdb, tdb_off_t off, struct list_str
 	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", 
+		TDB_LOG((tdb, TDB_DEBUG_WARNING, "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)
@@ -47,7 +47,7 @@ static int rec_free_read(struct tdb_context *tdb, tdb_off_t off, struct list_str
 	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", 
+		TDB_LOG((tdb, TDB_DEBUG_WARNING, "rec_free_read bad magic 0x%x at offset=%d\n", 
 			   rec->magic, off));
 		return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
 	}
@@ -73,7 +73,7 @@ static int remove_from_freelist(struct tdb_context *tdb, tdb_off_t off, tdb_off_
 		/* 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));
+	TDB_LOG((tdb, TDB_DEBUG_FATAL,"remove_from_freelist: not on list at off=%d\n", off));
 	return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
 }
 
@@ -102,7 +102,7 @@ int tdb_free(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec)
 
 	/* 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"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free: update_tailer failed!\n"));
 		goto fail;
 	}
 
@@ -112,14 +112,14 @@ int tdb_free(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *rec)
 		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));
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "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));
+				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free: right free failed at %u\n", right));
 				goto left;
 			}
 			rec->rec_len += sizeof(r) + r.rec_len;
@@ -135,7 +135,7 @@ left:
 		
 		/* 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));
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free: left offset read failed at %u\n", left));
 			goto update;
 		}
 
@@ -148,14 +148,14 @@ left:
 
 		/* 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));
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "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));
+				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free: left free failed at %u\n", left));
 				goto update;
 			} else {
 				offset = left;
@@ -166,7 +166,7 @@ left:
 
 update:
 	if (update_tailer(tdb, offset, rec) == -1) {
-		TDB_LOG((tdb, 0, "tdb_free: update_tailer failed at %u\n", offset));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free: update_tailer failed at %u\n", offset));
 		goto fail;
 	}
 
@@ -176,7 +176,7 @@ update:
 	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));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_free record write failed at offset=%d\n", offset));
 		goto fail;
 	}
 
diff --git a/source3/tdb/common/io.c b/source3/tdb/common/io.c
index 5d7edfefff..9a6084888d 100644
--- a/source3/tdb/common/io.c
+++ b/source3/tdb/common/io.c
@@ -29,28 +29,6 @@
 
 #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 
@@ -65,7 +43,7 @@ static int tdb_oob(struct tdb_context *tdb, tdb_off_t len, int probe)
 		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",
+			TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_oob len %d beyond internal malloc size %d\n",
 				 (int)len, (int)tdb->map_size));
 		}
 		return TDB_ERRCODE(TDB_ERR_IO, -1);
@@ -79,7 +57,7 @@ static int tdb_oob(struct tdb_context *tdb, tdb_off_t len, int probe)
 		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",
+			TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_oob len %d beyond eof at %d\n",
 				 (int)len, (int)st.st_size));
 		}
 		return TDB_ERRCODE(TDB_ERR_IO, -1);
@@ -114,7 +92,7 @@ static int tdb_write(struct tdb_context *tdb, tdb_off_t off,
 	} 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",
+		TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_write failed at %d len=%d (%s)\n",
 			   off, len, strerror(errno)));
 		return TDB_ERRCODE(TDB_ERR_IO, -1);
 	}
@@ -146,8 +124,8 @@ static int tdb_read(struct tdb_context *tdb, tdb_off_t off, void *buf,
 		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));
+			TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_read failed at %d len=%d ret=%d (%s) map_size=%d\n",
+				 off, len, ret, strerror(errno), (int)tdb->map_size));
 			return TDB_ERRCODE(TDB_ERR_IO, -1);
 		}
 	}
@@ -217,7 +195,7 @@ void tdb_mmap(struct tdb_context *tdb)
 
 		if (tdb->map_ptr == MAP_FAILED) {
 			tdb->map_ptr = NULL;
-			TDB_LOG((tdb, 2, "tdb_mmap failed for size %d (%s)\n", 
+			TDB_LOG((tdb, TDB_DEBUG_WARNING, "tdb_mmap failed for size %d (%s)\n", 
 				 tdb->map_size, strerror(errno)));
 		}
 	} else {
@@ -242,7 +220,7 @@ static int tdb_expand_file(struct tdb_context *tdb, tdb_off_t size, tdb_off_t ad
 	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", 
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "expand_file to %d failed (%s)\n", 
 				 size+addition, strerror(errno)));
 			return -1;
 		}
@@ -256,7 +234,7 @@ static int tdb_expand_file(struct tdb_context *tdb, tdb_off_t size, tdb_off_t ad
 		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", 
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "expand_file write of %d failed (%s)\n", 
 				   n, strerror(errno)));
 			return -1;
 		}
@@ -275,7 +253,7 @@ int tdb_expand(struct tdb_context *tdb, tdb_off_t size)
 	tdb_off_t offset;
 
 	if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
-		TDB_LOG((tdb, 0, "lock failed in tdb_expand\n"));
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "lock failed in tdb_expand\n"));
 		return -1;
 	}
 
@@ -364,7 +342,7 @@ char *tdb_alloc_read(struct tdb_context *tdb, tdb_off_t offset, tdb_len_t len)
 	if (!(buf = (char *)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",
+		TDB_LOG((tdb, TDB_DEBUG_ERROR,"tdb_alloc_read malloc failed len=%d (%s)\n",
 			   len, strerror(errno)));
 		return TDB_ERRCODE(TDB_ERR_OOM, buf);
 	}
@@ -383,7 +361,7 @@ int tdb_rec_read(struct tdb_context *tdb, tdb_off_t offset, struct list_struct *
 	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));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL,"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);
diff --git a/source3/tdb/common/lock.c b/source3/tdb/common/lock.c
index a96d77a229..a5bff2d0b3 100644
--- a/source3/tdb/common/lock.c
+++ b/source3/tdb/common/lock.c
@@ -36,8 +36,8 @@
 
    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)
+int tdb_brlock(struct tdb_context *tdb, tdb_off_t offset, 
+	       int rw_type, int lck_type, int probe, size_t len)
 {
 	struct flock fl;
 	int ret;
@@ -68,7 +68,7 @@ int tdb_brlock_len(struct tdb_context *tdb, tdb_off_t offset,
 		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_LOG((tdb, TDB_DEBUG_TRACE,"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);
@@ -88,7 +88,7 @@ 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) {
+		if (tdb_brlock(tdb, offset, F_WRLCK, F_SETLKW, 1, len) == 0) {
 			return 0;
 		}
 		if (errno != EDEADLK) {
@@ -99,27 +99,26 @@ int tdb_brlock_upgrade(struct tdb_context *tdb, tdb_off_t offset, size_t len)
 		tv.tv_usec = 1;
 		select(0, NULL, NULL, NULL, &tv);
 	}
-	TDB_LOG((tdb, 5,"tdb_brlock_upgrade failed at offset %d\n", offset));
+	TDB_LOG((tdb, TDB_DEBUG_TRACE,"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)
 {
+	/* a global lock allows us to avoid per chain locks */
+	if (tdb->global_lock.count && 
+	    (ltype == tdb->global_lock.ltype || ltype == F_RDLCK)) {
+		return 0;
+	}
+
+	if (tdb->global_lock.count) {
+		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
+	}
+
 	if (list < -1 || list >= (int)tdb->header.hash_size) {
-		TDB_LOG((tdb, 0,"tdb_lock: invalid list %d for ltype=%d\n", 
+		TDB_LOG((tdb, TDB_DEBUG_ERROR,"tdb_lock: invalid list %d for ltype=%d\n", 
 			   list, ltype));
 		return -1;
 	}
@@ -129,8 +128,8 @@ int tdb_lock(struct tdb_context *tdb, int list, int ltype)
 	/* 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", 
+		if (tdb->methods->tdb_brlock(tdb,FREELIST_TOP+4*list,ltype,F_SETLKW, 0, 1)) {
+			TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_lock failed on list %d ltype=%d (%s)\n", 
 				 list, ltype, strerror(errno)));
 			return -1;
 		}
@@ -148,23 +147,33 @@ int tdb_unlock(struct tdb_context *tdb, int list, int ltype)
 {
 	int ret = -1;
 
+	/* a global lock allows us to avoid per chain locks */
+	if (tdb->global_lock.count && 
+	    (ltype == tdb->global_lock.ltype || ltype == F_RDLCK)) {
+		return 0;
+	}
+
+	if (tdb->global_lock.count) {
+		return TDB_ERRCODE(TDB_ERR_LOCK, -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));
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "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"));
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "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);
+		ret = tdb->methods->tdb_brlock(tdb, FREELIST_TOP+4*list, F_UNLCK, F_SETLKW, 0, 1);
 		tdb->num_locks--;
 	} else {
 		ret = 0;
@@ -172,40 +181,97 @@ int tdb_unlock(struct tdb_context *tdb, int list, int ltype)
 	tdb->locked[list+1].count--;
 
 	if (ret)
-		TDB_LOG((tdb, 0,"tdb_unlock: An error occurred unlocking!\n")); 
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlock: An error occurred unlocking!\n")); 
 	return ret;
 }
 
 
 
 /* lock/unlock entire database */
-int tdb_lockall(struct tdb_context *tdb)
+static int _tdb_lockall(struct tdb_context *tdb, int ltype)
 {
-	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;
+	if (tdb->global_lock.count && tdb->global_lock.ltype == ltype) {
+		tdb->global_lock.count++;
+		return 0;
+	}
+
+	if (tdb->global_lock.count) {
+		/* a global lock of a different type exists */
+		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
+	}
+	
+	if (tdb->num_locks != 0) {
+		/* can't combine global and chain locks */
+		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
+	}
+
+	if (tdb->methods->tdb_brlock(tdb, FREELIST_TOP, ltype, F_SETLKW, 
+				     0, 4*tdb->header.hash_size)) {
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_lockall failed (%s)\n", strerror(errno)));
+		return -1;
+	}
+
+	tdb->global_lock.count = 1;
+	tdb->global_lock.ltype = ltype;
+
+	return 0;
+}
+
+/* unlock entire db */
+static int _tdb_unlockall(struct tdb_context *tdb, int ltype)
+{
+	/* There are no locks on read-only dbs */
+	if (tdb->read_only || tdb->traverse_read) {
+		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
+	}
 
-		for ( j = 0; j < i; j++)
-			tdb_unlock(tdb, j, F_WRLCK);
-		return TDB_ERRCODE(TDB_ERR_NOLOCK, -1);
+	if (tdb->global_lock.ltype != ltype || tdb->global_lock.count == 0) {
+		return TDB_ERRCODE(TDB_ERR_LOCK, -1);
+	}
+
+	if (tdb->global_lock.count > 1) {
+		tdb->global_lock.count--;
+		return 0;
 	}
 
+	if (tdb->methods->tdb_brlock(tdb, FREELIST_TOP, F_UNLCK, F_SETLKW, 
+				     0, 4*tdb->header.hash_size)) {
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlockall failed (%s)\n", strerror(errno)));
+		return -1;
+	}
+
+	tdb->global_lock.count = 0;
+	tdb->global_lock.ltype = 0;
+
 	return 0;
 }
-void tdb_unlockall(struct tdb_context *tdb)
+
+/* lock entire database with write lock */
+int tdb_lockall(struct tdb_context *tdb)
+{
+	return _tdb_lockall(tdb, F_WRLCK);
+}
+
+/* unlock entire database with write lock */
+int tdb_unlockall(struct tdb_context *tdb)
+{
+	return _tdb_unlockall(tdb, F_WRLCK);
+}
+
+/* lock entire database with read lock */
+int tdb_lockall_read(struct tdb_context *tdb)
+{
+	return _tdb_lockall(tdb, F_RDLCK);
+}
+
+/* unlock entire database with read lock */
+int tdb_unlockall_read(struct tdb_context *tdb)
 {
-	u32 i;
-	for (i=0; i < tdb->header.hash_size; i++)
-		tdb_unlock(tdb, i, F_WRLCK);
+	return _tdb_unlockall(tdb, F_RDLCK);
 }
 
 /* lock/unlock one hash chain. This is meant to be used to reduce
@@ -235,7 +301,7 @@ int tdb_chainunlock_read(struct tdb_context *tdb, TDB_DATA key)
 /* 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;
+	return off ? tdb->methods->tdb_brlock(tdb, off, F_RDLCK, F_SETLKW, 0, 1) : 0;
 }
 
 /*
@@ -249,7 +315,7 @@ int tdb_write_lock_record(struct tdb_context *tdb, tdb_off_t off)
 	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);
+	return tdb->methods->tdb_brlock(tdb, off, F_WRLCK, F_SETLK, 1, 1);
 }
 
 /*
@@ -258,7 +324,7 @@ 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)
 {
-	return tdb->methods->tdb_brlock(tdb, off, F_UNLCK, F_SETLK, 0);
+	return tdb->methods->tdb_brlock(tdb, off, F_UNLCK, F_SETLK, 0, 1);
 }
 
 /* fcntl locks don't stack: avoid unlocking someone else's */
@@ -272,5 +338,5 @@ int tdb_unlock_record(struct tdb_context *tdb, tdb_off_t off)
 	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);
+	return (count == 1 ? tdb->methods->tdb_brlock(tdb, off, F_UNLCK, F_SETLKW, 0, 1) : 0);
 }
diff --git a/source3/tdb/common/open.c b/source3/tdb/common/open.c
index b53b4c9f5c..e1f21aa856 100644
--- a/source3/tdb/common/open.c
+++ b/source3/tdb/common/open.c
@@ -123,15 +123,16 @@ struct tdb_context *tdb_open(const char *name, int hash_size, int tdb_flags,
 }
 
 /* a default logging function */
-static void null_log_fn(struct tdb_context *tdb, int level, const char *fmt, ...)
+static void null_log_fn(struct tdb_context *tdb, enum tdb_debug_level level, const char *fmt, ...) PRINTF_ATTRIBUTE(3, 4);
+static void null_log_fn(struct tdb_context *tdb, enum tdb_debug_level 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)
+				int open_flags, mode_t mode,
+				const struct tdb_logging_context *log_ctx,
+				tdb_hash_func hash_fn)
 {
 	struct tdb_context *tdb;
 	struct stat st;
@@ -150,7 +151,12 @@ struct tdb_context *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
 	tdb->map_ptr = NULL;
 	tdb->flags = tdb_flags;
 	tdb->open_flags = open_flags;
-	tdb->log_fn = log_fn?log_fn:null_log_fn;
+	if (log_ctx) {
+		tdb->log = *log_ctx;
+	} else {
+		tdb->log.log_fn = null_log_fn;
+		tdb->log.log_private = NULL;
+	}
 	tdb->hash_fn = hash_fn ? hash_fn : default_tdb_hash;
 
 	/* cache the page size */
@@ -160,7 +166,7 @@ struct tdb_context *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
 	}
 
 	if ((open_flags & O_ACCMODE) == O_WRONLY) {
-		TDB_LOG((tdb, 0, "tdb_open_ex: can't open tdb %s write-only\n",
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_open_ex: can't open tdb %s write-only\n",
 			 name));
 		errno = EINVAL;
 		goto fail;
@@ -180,31 +186,31 @@ struct tdb_context *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
 		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!"));
+			TDB_LOG((tdb, TDB_DEBUG_ERROR, "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",
+		TDB_LOG((tdb, TDB_DEBUG_WARNING, "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",
+	if (tdb->methods->tdb_brlock(tdb, GLOBAL_LOCK, F_WRLCK, F_SETLKW, 0, 1) == -1) {
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "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))) {
+	    (locked = (tdb->methods->tdb_brlock(tdb, ACTIVE_LOCK, F_WRLCK, F_SETLK, 0, 1) == 0))) {
 		open_flags |= O_CREAT;
 		if (ftruncate(tdb->fd, 0) == -1) {
-			TDB_LOG((tdb, 0, "tdb_open_ex: "
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_open_ex: "
 				 "failed to truncate %s: %s\n",
 				 name, strerror(errno)));
 			goto fail; /* errno set by ftruncate */
@@ -236,13 +242,13 @@ struct tdb_context *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
 		goto fail;
 
 	if (tdb->header.rwlocks != 0) {
-		TDB_LOG((tdb, 5, "tdb_open_ex: spinlocks no longer supported\n"));
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "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: "
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_open_ex: "
 			 "%s (%d,%d) is already open in this process\n",
 			 name, (int)st.st_dev, (int)st.st_ino));
 		errno = EBUSY;
@@ -260,7 +266,7 @@ struct tdb_context *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
 	tdb->locked = (struct tdb_lock_type *)calloc(tdb->header.hash_size+1,
 						     sizeof(tdb->locked[0]));
 	if (!tdb->locked) {
-		TDB_LOG((tdb, 2, "tdb_open_ex: "
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_open_ex: "
 			 "failed to allocate lock structure for %s\n",
 			 name));
 		errno = ENOMEM;
@@ -268,8 +274,8 @@ struct tdb_context *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
 	}
 	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: "
+		if (tdb->methods->tdb_brlock(tdb, ACTIVE_LOCK, F_UNLCK, F_SETLK, 0, 1) == -1) {
+			TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_open_ex: "
 				 "failed to take ACTIVE_LOCK on %s: %s\n",
 				 name, strerror(errno)));
 			goto fail;
@@ -283,7 +289,7 @@ struct tdb_context *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
 
 	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)
+		if (tdb->methods->tdb_brlock(tdb, ACTIVE_LOCK, F_RDLCK, F_SETLKW, 0, 1) == -1)
 			goto fail;
 	}
 
@@ -296,7 +302,7 @@ struct tdb_context *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
 	/* 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)
+	if (tdb->methods->tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1) == -1)
 		goto fail;
 	tdb->next = tdbs;
 	tdbs = tdb;
@@ -317,7 +323,7 @@ struct tdb_context *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
 	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"));
+			TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_open_ex: failed to close tdb->fd on error!\n"));
 	SAFE_FREE(tdb->locked);
 	SAFE_FREE(tdb);
 	errno = save_errno;
@@ -365,11 +371,16 @@ int tdb_close(struct tdb_context *tdb)
 }
 
 /* register a loging function */
-void tdb_logging_function(struct tdb_context *tdb, void (*fn)(struct tdb_context *, int , const char *, ...))
+void tdb_set_logging_function(struct tdb_context *tdb,
+                              const struct tdb_logging_context *log)
 {
-	tdb->log_fn = fn?fn:null_log_fn;
+        tdb->log = *log;
 }
 
+void *tdb_get_logging_private(struct tdb_context *tdb)
+{
+	return tdb->log.log_private;
+}
 
 /* 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 */
@@ -381,38 +392,38 @@ int tdb_reopen(struct tdb_context *tdb)
 		return 0; /* Nothing to do. */
 	}
 
-	if (tdb->num_locks != 0) {
-		TDB_LOG((tdb, 0, "tdb_reopen: reopen not allowed with locks held\n"));
+	if (tdb->num_locks != 0 || tdb->global_lock.count) {
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "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"));
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "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)));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "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_LOG((tdb, TDB_DEBUG_FATAL, "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)));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "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"));
+	    (tdb->methods->tdb_brlock(tdb, ACTIVE_LOCK, F_RDLCK, F_SETLKW, 0, 1) == -1)) {
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "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)));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "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"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_reopen: file dev/inode has changed!\n"));
 		goto fail;
 	}
 	tdb_mmap(tdb);
diff --git a/source3/tdb/common/tdb.c b/source3/tdb/common/tdb.c
index 524d404a50..addb5b4378 100644
--- a/source3/tdb/common/tdb.c
+++ b/source3/tdb/common/tdb.c
@@ -42,7 +42,7 @@ static void tdb_increment_seqnum(struct tdb_context *tdb)
 		return;
 	}
 
-	if (tdb_brlock(tdb, TDB_SEQNUM_OFS, F_WRLCK, F_SETLKW, 1) != 0) {
+	if (tdb_brlock(tdb, TDB_SEQNUM_OFS, F_WRLCK, F_SETLKW, 1, 1) != 0) {
 		return;
 	}
 
@@ -53,7 +53,7 @@ static void tdb_increment_seqnum(struct tdb_context *tdb)
 	seqnum++;
 	tdb_ofs_write(tdb, TDB_SEQNUM_OFS, &seqnum);
 
-	tdb_brlock(tdb, TDB_SEQNUM_OFS, F_UNLCK, F_SETLKW, 1);
+	tdb_brlock(tdb, TDB_SEQNUM_OFS, F_UNLCK, F_SETLKW, 1, 1);
 }
 
 
@@ -236,7 +236,7 @@ static int tdb_delete_hash(struct tdb_context *tdb, TDB_DATA key, u32 hash)
 	}
 
 	if (tdb_unlock(tdb, BUCKET(rec.full_hash), F_WRLCK) != 0)
-		TDB_LOG((tdb, 0, "tdb_delete: WARNING tdb_unlock failed!\n"));
+		TDB_LOG((tdb, TDB_DEBUG_WARNING, "tdb_delete: WARNING tdb_unlock failed!\n"));
 	return ret;
 }
 
@@ -403,7 +403,7 @@ int tdb_fd(struct tdb_context *tdb)
 */
 tdb_log_func tdb_log_fn(struct tdb_context *tdb)
 {
-	return tdb->log_fn;
+	return tdb->log.log_fn;
 }
 
 
@@ -429,3 +429,14 @@ int tdb_hash_size(struct tdb_context *tdb)
 {
 	return tdb->header.hash_size;
 }
+
+size_t tdb_map_size(struct tdb_context *tdb)
+{
+	return tdb->map_size;
+}
+
+int tdb_get_flags(struct tdb_context *tdb)
+{
+	return tdb->flags;
+}
+
diff --git a/source3/tdb/common/tdb_private.h b/source3/tdb/common/tdb_private.h
index d4f94048a4..cd57365434 100644
--- a/source3/tdb/common/tdb_private.h
+++ b/source3/tdb/common/tdb_private.h
@@ -24,42 +24,13 @@
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */
 
-#ifndef _SAMBA_BUILD_
-#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 "replace.h"
+#include "system/filesys.h"
+#include "system/time.h"
+#include "system/shmem.h"
+#include "system/select.h"
 #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
@@ -100,21 +71,13 @@ typedef u32 tdb_off_t;
 /* 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
+#define TDB_LOG(x) tdb->log.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)
@@ -178,7 +141,7 @@ struct tdb_methods {
 	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);
+	int (*tdb_brlock)(struct tdb_context *, tdb_off_t , int, int, int, size_t);
 };
 
 struct tdb_context {
@@ -188,6 +151,7 @@ struct tdb_context {
 	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 global_lock;
 	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 */
@@ -196,7 +160,7 @@ struct tdb_context {
 	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 */
+	struct tdb_logging_context log;
 	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 */
@@ -213,10 +177,8 @@ 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(struct tdb_context *tdb, tdb_off_t offset, int rw_type, int lck_type, int probe, size_t len);
 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);
diff --git a/source3/tdb/common/tdbutil.c b/source3/tdb/common/tdbutil.c
index 45e1891324..3e18c09fbf 100644
--- a/source3/tdb/common/tdbutil.c
+++ b/source3/tdb/common/tdbutil.c
@@ -19,8 +19,11 @@
    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */
 
-#include "tdb_private.h"
-#include <fnmatch.h>
+#include "includes.h"
+#undef malloc
+#undef realloc
+#undef calloc
+#undef strdup
 
 /***************************************************************
  Allow a caller to set a "alarm" flag that tdb can check to abort
@@ -91,7 +94,7 @@ static int tdb_chainlock_with_timeout_internal( TDB_CONTEXT *tdb, TDB_DATA key,
 		CatchSignal(SIGALRM, SIGNAL_CAST SIG_IGN);
 		if (gotalarm) {
 			DEBUG(0,("tdb_chainlock_with_timeout_internal: alarm (%u) timed out for key %s in tdb %s\n",
-				timeout, key.dptr, tdb->name ));
+				timeout, key.dptr, tdb_name(tdb)));
 			/* TODO: If we time out waiting for a lock, it might
 			 * be nice to use F_GETLK to get the pid of the
 			 * process currently holding the lock and print that
@@ -657,7 +660,7 @@ int tdb_unpack(char *buf, int bufsize, const char *fmt, ...)
  Log tdb messages via DEBUG().
 ****************************************************************************/
 
-static void tdb_log(TDB_CONTEXT *tdb, int level, const char *format, ...)
+static void tdb_log(TDB_CONTEXT *tdb, enum tdb_debug_level level, const char *format, ...)
 {
 	va_list ap;
 	char *ptr = NULL;
@@ -669,7 +672,7 @@ static void tdb_log(TDB_CONTEXT *tdb, int level, const char *format, ...)
 	if (!ptr || !*ptr)
 		return;
 
-	DEBUG(level, ("tdb(%s): %s", tdb->name ? tdb->name : "unnamed", ptr));
+	DEBUG((int)level, ("tdb(%s): %s", tdb_name(tdb) ? tdb_name(tdb) : "unnamed", ptr));
 	SAFE_FREE(ptr);
 }
 
@@ -682,12 +685,16 @@ TDB_CONTEXT *tdb_open_log(const char *name, int hash_size, int tdb_flags,
 			  int open_flags, mode_t mode)
 {
 	TDB_CONTEXT *tdb;
+	struct tdb_logging_context log_ctx;
 
 	if (!lp_use_mmap())
 		tdb_flags |= TDB_NOMMAP;
 
+	log_ctx.log_fn = tdb_log;
+	log_ctx.log_private = NULL;
+
 	tdb = tdb_open_ex(name, hash_size, tdb_flags, 
-				    open_flags, mode, tdb_log, NULL);
+			  open_flags, mode, &log_ctx, NULL);
 	if (!tdb)
 		return NULL;
 
@@ -773,16 +780,6 @@ void tdb_search_list_free(TDB_LIST_NODE* 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.
diff --git a/source3/tdb/common/transaction.c b/source3/tdb/common/transaction.c
index 8e21e3ad87..a7ce5e3f61 100644
--- a/source3/tdb/common/transaction.c
+++ b/source3/tdb/common/transaction.c
@@ -183,7 +183,7 @@ static int transaction_read(struct tdb_context *tdb, tdb_off_t off, void *buf,
 	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_LOG((tdb, TDB_DEBUG_FATAL, "transaction_read: failed at off=%d len=%d\n", off, len));
 	tdb->ecode = TDB_ERR_IO;
 	tdb->transaction->transaction_error = 1;
 	return -1;
@@ -308,7 +308,7 @@ static int transaction_write(struct tdb_context *tdb, tdb_off_t off,
 	return 0;
 
 fail:
-	TDB_LOG((tdb, 0, "transaction_write: failed at off=%d len=%d\n", off, len));
+	TDB_LOG((tdb, TDB_DEBUG_FATAL, "transaction_write: failed at off=%d len=%d\n", off, len));
 	tdb->ecode = TDB_ERR_IO;
 	tdb->transaction->transaction_error = 1;
 	return -1;
@@ -359,7 +359,7 @@ static int transaction_expand_file(struct tdb_context *tdb, tdb_off_t size,
   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)
+		       int rw_type, int lck_type, int probe, size_t len)
 {
 	return 0;
 }
@@ -382,7 +382,7 @@ 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_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: cannot start a transaction on a read-only or internal db\n"));
 		tdb->ecode = TDB_ERR_EINVAL;
 		return -1;
 	}
@@ -390,16 +390,16 @@ int tdb_transaction_start(struct tdb_context *tdb)
 	/* 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_LOG((tdb, TDB_DEBUG_TRACE, "tdb_transaction_start: nesting %d\n", 
 			 tdb->transaction->nesting));
 		return 0;
 	}
 
-	if (tdb->num_locks != 0) {
+	if (tdb->num_locks != 0 || tdb->global_lock.count) {
 		/* 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_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: cannot start a transaction with locks held\n"));
 		tdb->ecode = TDB_ERR_LOCK;
 		return -1;
 	}
@@ -408,7 +408,7 @@ int tdb_transaction_start(struct tdb_context *tdb)
 		/* 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_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: cannot start a transaction within a traverse\n"));
 		tdb->ecode = TDB_ERR_LOCK;
 		return -1;
 	}
@@ -423,8 +423,8 @@ int tdb_transaction_start(struct tdb_context *tdb)
 	/* 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"));
+	if (tdb_brlock(tdb, TRANSACTION_LOCK, F_WRLCK, F_SETLKW, 0, 1) == -1) {
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: failed to get transaction lock\n"));
 		tdb->ecode = TDB_ERR_LOCK;
 		SAFE_FREE(tdb->transaction);
 		return -1;
@@ -432,23 +432,23 @@ int tdb_transaction_start(struct tdb_context *tdb)
 	
 	/* 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"));
+	if (tdb_brlock(tdb, FREELIST_TOP, F_RDLCK, F_SETLKW, 0, 0) == -1) {
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "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 = (unsigned int *)
-		calloc(tdb->header.hash_size+1, sizeof(tdb_off_t));
+	tdb->transaction->hash_heads = (u32 *)
+		calloc(tdb->header.hash_size+1, sizeof(u32));
 	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_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_start: failed to read hash heads\n"));
 		tdb->ecode = TDB_ERR_IO;
 		goto fail;
 	}
@@ -467,7 +467,7 @@ int tdb_transaction_start(struct tdb_context *tdb)
 	   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_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_start: failed to prime hash table\n"));
 		tdb->ecode = TDB_ERR_IO;
 		goto fail;
 	}
@@ -475,8 +475,8 @@ int tdb_transaction_start(struct tdb_context *tdb)
 	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);
+	tdb_brlock(tdb, FREELIST_TOP, F_UNLCK, F_SETLKW, 0, 0);
+	tdb_brlock(tdb, TRANSACTION_LOCK, F_UNLCK, F_SETLKW, 0, 1);
 	SAFE_FREE(tdb->transaction->hash_heads);
 	SAFE_FREE(tdb->transaction);
 	return -1;
@@ -489,7 +489,7 @@ fail:
 int tdb_transaction_cancel(struct tdb_context *tdb)
 {	
 	if (tdb->transaction == NULL) {
-		TDB_LOG((tdb, 0, "tdb_transaction_cancel: no transaction\n"));
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_cancel: no transaction\n"));
 		return -1;
 	}
 
@@ -509,12 +509,18 @@ int tdb_transaction_cancel(struct tdb_context *tdb)
 		free(el);
 	}
 
+	/* remove any global lock created during the transaction */
+	if (tdb->global_lock.count != 0) {
+		tdb_brlock(tdb, FREELIST_TOP, F_UNLCK, F_SETLKW, 0, 4*tdb->header.hash_size);
+		tdb->global_lock.count = 0;
+	}
+
 	/* 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_brlock(tdb,FREELIST_TOP+4*h,F_UNLCK,F_SETLKW, 0, 1);
 				tdb->locked[h].count = 0;
 			}
 		}
@@ -524,8 +530,8 @@ int tdb_transaction_cancel(struct tdb_context *tdb)
 	/* 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);
+	tdb_brlock(tdb, FREELIST_TOP, F_UNLCK, F_SETLKW, 0, 0);
+	tdb_brlock(tdb, TRANSACTION_LOCK, F_UNLCK, F_SETLKW, 0, 1);
 	SAFE_FREE(tdb->transaction->hash_heads);
 	SAFE_FREE(tdb->transaction);
 	
@@ -539,7 +545,7 @@ static int transaction_sync(struct tdb_context *tdb, tdb_off_t offset, tdb_len_t
 {	
 	if (fsync(tdb->fd) != 0) {
 		tdb->ecode = TDB_ERR_IO;
-		TDB_LOG((tdb, 0, "tdb_transaction: fsync failed\n"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction: fsync failed\n"));
 		return -1;
 	}
 #ifdef MS_SYNC
@@ -548,7 +554,7 @@ static int transaction_sync(struct tdb_context *tdb, tdb_off_t offset, tdb_len_t
 		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",
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction: msync failed - %s\n",
 				 strerror(errno)));
 			return -1;
 		}
@@ -591,7 +597,7 @@ static int tdb_recovery_allocate(struct tdb_context *tdb,
 	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"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to read recovery head\n"));
 		return -1;
 	}
 
@@ -599,7 +605,7 @@ static int tdb_recovery_allocate(struct tdb_context *tdb,
 
 	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"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to read recovery record\n"));
 		return -1;
 	}
 
@@ -619,7 +625,7 @@ static int tdb_recovery_allocate(struct tdb_context *tdb,
 	   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"));
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to free previous recovery area\n"));
 			return -1;
 		}
 	}
@@ -635,7 +641,7 @@ static int tdb_recovery_allocate(struct tdb_context *tdb,
 	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"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to create recovery area\n"));
 		return -1;
 	}
 
@@ -651,7 +657,7 @@ static int tdb_recovery_allocate(struct tdb_context *tdb,
 	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"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to write recovery head\n"));
 		return -1;
 	}
 
@@ -705,7 +711,7 @@ static int transaction_setup_recovery(struct tdb_context *tdb,
 			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"));
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: transaction data over new region boundary\n"));
 			free(data);
 			tdb->ecode = TDB_ERR_CORRUPT;
 			return -1;
@@ -733,7 +739,7 @@ static int transaction_setup_recovery(struct tdb_context *tdb,
 
 	/* 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"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: failed to write recovery data\n"));
 		free(data);
 		tdb->ecode = TDB_ERR_IO;
 		return -1;
@@ -755,7 +761,7 @@ static int transaction_setup_recovery(struct tdb_context *tdb,
 	*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_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: failed to write recovery magic\n"));
 		tdb->ecode = TDB_ERR_IO;
 		return -1;
 	}
@@ -778,14 +784,14 @@ int tdb_transaction_commit(struct tdb_context *tdb)
 	u32 zero = 0;
 
 	if (tdb->transaction == NULL) {
-		TDB_LOG((tdb, 0, "tdb_transaction_commit: no transaction\n"));
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "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"));
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_commit: transaction error pending\n"));
 		return -1;
 	}
 
@@ -804,16 +810,16 @@ int tdb_transaction_commit(struct tdb_context *tdb)
 	
 	/* if there are any locks pending then the caller has not
 	   nested their locks properly, so fail the transaction */
-	if (tdb->num_locks) {
+	if (tdb->num_locks || tdb->global_lock.count) {
 		tdb->ecode = TDB_ERR_LOCK;
-		TDB_LOG((tdb, 0, "tdb_transaction_commit: locks pending on commit\n"));
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "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_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: failed to upgrade hash locks\n"));
 		tdb->ecode = TDB_ERR_LOCK;
 		tdb_transaction_cancel(tdb);
 		return -1;
@@ -821,8 +827,8 @@ int tdb_transaction_commit(struct tdb_context *tdb)
 
 	/* 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"));
+	if (tdb_brlock(tdb, GLOBAL_LOCK, F_WRLCK, F_SETLKW, 0, 1) == -1) {
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_commit: failed to get global lock\n"));
 		tdb->ecode = TDB_ERR_LOCK;
 		tdb_transaction_cancel(tdb);
 		return -1;
@@ -831,8 +837,8 @@ int tdb_transaction_commit(struct tdb_context *tdb)
 	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_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_commit: failed to setup recovery data\n"));
+			tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
 			tdb_transaction_cancel(tdb);
 			return -1;
 		}
@@ -844,8 +850,8 @@ int tdb_transaction_commit(struct tdb_context *tdb)
 					     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_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_commit: expansion failed\n"));
+			tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
 			tdb_transaction_cancel(tdb);
 			return -1;
 		}
@@ -858,7 +864,7 @@ int tdb_transaction_commit(struct tdb_context *tdb)
 		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"));
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_commit: write failed during commit\n"));
 			
 			/* we've overwritten part of the data and
 			   possibly expanded the file, so we need to
@@ -867,9 +873,9 @@ int tdb_transaction_commit(struct tdb_context *tdb)
 			tdb_transaction_recover(tdb); 
 
 			tdb_transaction_cancel(tdb);
-			tdb_brlock_len(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
+			tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
 
-			TDB_LOG((tdb, 0, "tdb_transaction_commit: write failed\n"));
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_commit: write failed\n"));
 			return -1;
 		}
 		tdb->transaction->elements = el->next;
@@ -885,7 +891,7 @@ int tdb_transaction_commit(struct tdb_context *tdb)
 
 		/* 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"));
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_commit: failed to remove recovery magic\n"));
 			return -1;
 		}
 
@@ -895,7 +901,7 @@ int tdb_transaction_commit(struct tdb_context *tdb)
 		}
 	}
 
-	tdb_brlock_len(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
+	tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0, 1);
 
 	/*
 	  TODO: maybe write to some dummy hdr field, or write to magic
@@ -933,7 +939,7 @@ int tdb_transaction_recover(struct tdb_context *tdb)
 
 	/* 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_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to read recovery head\n"));
 		tdb->ecode = TDB_ERR_IO;
 		return -1;
 	}
@@ -946,7 +952,7 @@ int tdb_transaction_recover(struct tdb_context *tdb)
 	/* 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_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to read recovery record\n"));		
 		tdb->ecode = TDB_ERR_IO;
 		return -1;
 	}
@@ -957,7 +963,7 @@ int tdb_transaction_recover(struct tdb_context *tdb)
 	}
 
 	if (tdb->read_only) {
-		TDB_LOG((tdb, 0, "tdb_transaction_recover: attempt to recover read only database\n"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: attempt to recover read only database\n"));
 		tdb->ecode = TDB_ERR_CORRUPT;
 		return -1;
 	}
@@ -966,7 +972,7 @@ int tdb_transaction_recover(struct tdb_context *tdb)
 
 	data = (unsigned char *)malloc(rec.data_len);
 	if (data == NULL) {
-		TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to allocate recovery data\n"));		
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to allocate recovery data\n"));		
 		tdb->ecode = TDB_ERR_OOM;
 		return -1;
 	}
@@ -974,7 +980,7 @@ int tdb_transaction_recover(struct tdb_context *tdb)
 	/* 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_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to read recovery data\n"));		
 		tdb->ecode = TDB_ERR_IO;
 		return -1;
 	}
@@ -991,7 +997,7 @@ int tdb_transaction_recover(struct tdb_context *tdb)
 
 		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_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to recover %d bytes at offset %d\n", len, ofs));
 			tdb->ecode = TDB_ERR_IO;
 			return -1;
 		}
@@ -1001,7 +1007,7 @@ int tdb_transaction_recover(struct tdb_context *tdb)
 	free(data);
 
 	if (transaction_sync(tdb, 0, tdb->map_size) == -1) {
-		TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to sync recovery\n"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to sync recovery\n"));
 		tdb->ecode = TDB_ERR_IO;
 		return -1;
 	}
@@ -1009,7 +1015,7 @@ int tdb_transaction_recover(struct tdb_context *tdb)
 	/* 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_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to remove recovery head\n"));
 			tdb->ecode = TDB_ERR_IO;
 			return -1;			
 		}
@@ -1018,7 +1024,7 @@ int tdb_transaction_recover(struct tdb_context *tdb)
 	/* 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_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to remove recovery magic\n"));
 		tdb->ecode = TDB_ERR_IO;
 		return -1;			
 	}
@@ -1026,7 +1032,7 @@ int tdb_transaction_recover(struct tdb_context *tdb)
 	/* 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_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to reduce to recovery size\n"));
 		tdb->ecode = TDB_ERR_IO;
 		return -1;			
 	}
@@ -1034,12 +1040,12 @@ int tdb_transaction_recover(struct tdb_context *tdb)
 	tdb_mmap(tdb);
 
 	if (transaction_sync(tdb, 0, recovery_eof) == -1) {
-		TDB_LOG((tdb, 0, "tdb_transaction_recover: failed to sync2 recovery\n"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "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", 
+	TDB_LOG((tdb, TDB_DEBUG_TRACE, "tdb_transaction_recover: recovered %d byte database\n", 
 		 recovery_eof));
 
 	/* all done */
diff --git a/source3/tdb/common/traverse.c b/source3/tdb/common/traverse.c
index 712504a764..6d3b111479 100644
--- a/source3/tdb/common/traverse.c
+++ b/source3/tdb/common/traverse.c
@@ -100,7 +100,7 @@ static int tdb_next_lock(struct tdb_context *tdb, struct tdb_traverse_lock *tloc
 
 			/* Detect infinite loops. From "Shlomi Yaakobovich" <Shlomi@exanet.com>. */
 			if (tlock->off == rec->next) {
-				TDB_LOG((tdb, 0, "tdb_next_lock: loop detected.\n"));
+				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_next_lock: loop detected.\n"));
 				goto fail;
 			}
 
@@ -127,7 +127,7 @@ static int tdb_next_lock(struct tdb_context *tdb, struct tdb_traverse_lock *tloc
  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"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_next_lock: On error unlock failed!\n"));
 	return -1;
 }
 
@@ -163,7 +163,7 @@ static int tdb_traverse_internal(struct tdb_context *tdb,
 			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"));
+				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_traverse: key.dptr == NULL and unlock_record failed!\n"));
 			goto out;
 		}
 		key.dsize = rec.key_len;
@@ -180,7 +180,7 @@ static int tdb_traverse_internal(struct tdb_context *tdb,
 			/* 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"));;
+				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_traverse: unlock_record failed!\n"));;
 				ret = -1;
 			}
 			SAFE_FREE(key.dptr);
@@ -208,8 +208,8 @@ int tdb_traverse_read(struct tdb_context *tdb,
 	
 	/* 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"));
+	if (tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, F_RDLCK, F_SETLKW, 0, 1) == -1) {
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_traverse_read: failed to get transaction lock\n"));
 		tdb->ecode = TDB_ERR_LOCK;
 		return -1;
 	}
@@ -218,7 +218,7 @@ int tdb_traverse_read(struct tdb_context *tdb,
 	ret = tdb_traverse_internal(tdb, fn, private_data, &tl);
 	tdb->traverse_read--;
 
-	tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, F_UNLCK, F_SETLKW, 0);
+	tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, F_UNLCK, F_SETLKW, 0, 1);
 
 	return ret;
 }
@@ -237,15 +237,15 @@ int tdb_traverse(struct tdb_context *tdb,
 		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"));
+	if (tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, F_WRLCK, F_SETLKW, 0, 1) == -1) {
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "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);
+	tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, F_UNLCK, F_SETLKW, 0, 1);
 
 	return ret;
 }
@@ -269,7 +269,7 @@ TDB_DATA tdb_firstkey(struct tdb_context *tdb)
 	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"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_firstkey: error occurred while tdb_unlocking!\n"));
 	return key;
 }
 
@@ -311,7 +311,7 @@ TDB_DATA tdb_nextkey(struct tdb_context *tdb, TDB_DATA oldkey)
 			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)));
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_nextkey: lock_record failed (%s)!\n", strerror(errno)));
 			return tdb_null;
 		}
 	}
@@ -325,11 +325,11 @@ TDB_DATA tdb_nextkey(struct tdb_context *tdb, TDB_DATA oldkey)
 					  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"));
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "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"));
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_nextkey: WARNING tdb_unlock failed!\n"));
 	return key;
 }
 
diff --git a/source3/tdb/config.mk b/source3/tdb/config.mk
index b1e6e82d74..c5d1a33990 100644
--- a/source3/tdb/config.mk
+++ b/source3/tdb/config.mk
@@ -8,8 +8,7 @@ OBJ_FILES = \
 	common/tdb.o common/dump.o common/io.o common/lock.o \
 	common/open.o common/traverse.o common/freelist.o \
 	common/error.o common/transaction.o common/tdbutil.o
-PUBLIC_DEPENDENCIES = \
-		LIBREPLACE
+CFLAGS = -Ilib/tdb/include
 PUBLIC_HEADERS = include/tdb.h
 #
 # End SUBSYSTEM ldb
diff --git a/source3/tdb/include/tdb.h b/source3/tdb/include/tdb.h
index 82f8828c89..691c26af81 100644
--- a/source3/tdb/include/tdb.h
+++ b/source3/tdb/include/tdb.h
@@ -55,6 +55,10 @@ 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, TDB_ERR_EINVAL, TDB_ERR_RDONLY};
 
+/* debugging uses one of the following levels */
+enum tdb_debug_level {TDB_DEBUG_FATAL = 0, TDB_DEBUG_ERROR, 
+		      TDB_DEBUG_WARNING, TDB_DEBUG_TRACE};
+
 typedef struct TDB_DATA {
 	char *dptr;
 	size_t dsize;
@@ -76,19 +80,24 @@ typedef struct TDB_DATA {
 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 void (*tdb_log_func)(struct tdb_context *, enum tdb_debug_level, const char *, ...) PRINTF_ATTRIBUTE(3, 4);
 typedef unsigned int (*tdb_hash_func)(TDB_DATA *key);
 
+struct tdb_logging_context {
+        tdb_log_func log_fn;
+        void *log_private;
+};
+
 struct tdb_context *tdb_open(const char *name, int hash_size, int tdb_flags,
 		      int open_flags, mode_t mode);
 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,
+			 const struct tdb_logging_context *log_ctx,
 			 tdb_hash_func hash_fn);
 
 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);
+void tdb_set_logging_function(struct tdb_context *tdb, const struct tdb_logging_context *log);
 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);
@@ -102,16 +111,21 @@ 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);
+int tdb_unlockall(struct tdb_context *tdb);
+int tdb_lockall_read(struct tdb_context *tdb);
+int tdb_unlockall_read(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);
+void *tdb_get_logging_private(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);
+size_t tdb_map_size(struct tdb_context *tdb);
+int tdb_get_flags(struct tdb_context *tdb);
 
 /* Low level locking functions: use with care */
 int tdb_chainlock(struct tdb_context *tdb, TDB_DATA key);
diff --git a/source3/tdb/include/tdbutil.h b/source3/tdb/include/tdbutil.h
index 72d1505f4e..cb5d98fc52 100644
--- a/source3/tdb/include/tdbutil.h
+++ b/source3/tdb/include/tdbutil.h
@@ -67,7 +67,5 @@ int 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/source3/tdb/tools/tdbtest.c b/source3/tdb/tools/tdbtest.c
index 89295a3291..c7a09789fe 100644
--- a/source3/tdb/tools/tdbtest.c
+++ b/source3/tdb/tools/tdbtest.c
@@ -1,48 +1,43 @@
-#include <stdlib.h>
-#include <stdio.h>
-#include <fcntl.h>
-#include <unistd.h>
-#include <string.h>
-#include <fcntl.h>
-#include <stdarg.h>
-#include <sys/mman.h>
-#include <sys/stat.h>
-#include <sys/time.h>
-#include <signal.h>
-#include "tdb.h"
-#include <gdbm.h>
-
 /* a test program for tdb - the trivial database */
 
+#include "replace.h"
+#include "tdb.h"
+#include "system/filesys.h"
+#include "system/time.h"
+
+#include <gdbm.h>
 
 
 #define DELETE_PROB 7
 #define STORE_PROB 5
 
-static TDB_CONTEXT *db;
+static struct tdb_context *db;
 static GDBM_FILE gdbm;
 
 struct timeval tp1,tp2;
 
-static void start_timer(void)
+static void _start_timer(void)
 {
 	gettimeofday(&tp1,NULL);
 }
 
-static double end_timer(void)
+static double _end_timer(void)
 {
 	gettimeofday(&tp2,NULL);
 	return((tp2.tv_sec - tp1.tv_sec) + 
 	       (tp2.tv_usec - tp1.tv_usec)*1.0e-6);
 }
 
-static void fatal(char *why)
+static void fatal(const char *why)
 {
 	perror(why);
 	exit(1);
 }
 
-static void tdb_log(TDB_CONTEXT *tdb, int level, const char *format, ...)
+#ifdef PRINTF_ATTRIBUTE
+static void tdb_log(struct tdb_context *tdb, int level, const char *format, ...) PRINTF_ATTRIBUTE(3,4);
+#endif
+static void tdb_log(struct tdb_context *tdb, int level, const char *format, ...)
 {
 	va_list ap;
     
@@ -179,7 +174,7 @@ static void addrec_gdbm(void)
 	free(d);
 }
 
-static int traverse_fn(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA dbuf, void *state)
+static int traverse_fn(struct tdb_context *tdb, TDB_DATA key, TDB_DATA dbuf, void *state)
 {
 #if 0
 	printf("[%s] [%s]\n", key.dptr, dbuf.dptr);
@@ -192,14 +187,15 @@ static void merge_test(void)
 {
 	int i;
 	char keys[5][2];
+	char tdata[] = "test";
 	TDB_DATA key, data;
 	
 	for (i = 0; i < 5; i++) {
-		sprintf(keys[i], "%d", i);
+		snprintf(keys[i],2, "%d", i);
 		key.dptr = keys[i];
 		key.dsize = 2;
 		
-		data.dptr = "test";
+		data.dptr = tdata;
 		data.dsize = 4;
 		
 		if (tdb_store(db, key, data, TDB_REPLACE) != 0) {
@@ -219,16 +215,17 @@ static void merge_test(void)
 	tdb_delete(db, key);
 }
 	
-int main(int argc, char *argv[])
+ int main(int argc, const char *argv[])
 {
 	int i, seed=0;
 	int loops = 10000;
+	char test_gdbm[] = "test.gdbm";
 
 	unlink("test.gdbm");
 
 	db = tdb_open("test.tdb", 0, TDB_CLEAR_IF_FIRST, 
 		      O_RDWR | O_CREAT | O_TRUNC, 0600);
-	gdbm = gdbm_open("test.gdbm", 512, GDBM_WRITER|GDBM_NEWDB|GDBM_FAST, 
+	gdbm = gdbm_open(test_gdbm, 512, GDBM_WRITER|GDBM_NEWDB|GDBM_FAST, 
 			 0600, NULL);
 
 	if (!db || !gdbm) {
@@ -239,17 +236,17 @@ int main(int argc, char *argv[])
 	
 #if 1
 	srand(seed);
-	start_timer();
+	_start_timer();
 	for (i=0;i<loops;i++) addrec_gdbm();
-	printf("gdbm got %.2f ops/sec\n", i/end_timer());
+	printf("gdbm got %.2f ops/sec\n", i/_end_timer());
 #endif
 
 	merge_test();
 
 	srand(seed);
-	start_timer();
+	_start_timer();
 	for (i=0;i<loops;i++) addrec_db();
-	printf("tdb got %.2f ops/sec\n", i/end_timer());
+	printf("tdb got %.2f ops/sec\n", i/_end_timer());
 
 	compare_db();
 
diff --git a/source3/tdb/tools/tdbtorture.c b/source3/tdb/tools/tdbtorture.c
index 2d367b91e1..14a2b48cdc 100644
--- a/source3/tdb/tools/tdbtorture.c
+++ b/source3/tdb/tools/tdbtorture.c
@@ -1,41 +1,43 @@
-#include <stdlib.h>
-#include <time.h>
-#include <stdio.h>
-#include <fcntl.h>
-#include <unistd.h>
-#include <string.h>
-#include <fcntl.h>
-#include <signal.h>
-#include <stdarg.h>
-#include <sys/mman.h>
-#include <sys/stat.h>
-#include <sys/time.h>
-#include <sys/wait.h>
-#include "tdb.h"
-
 /* this tests tdb by doing lots of ops from several simultaneous
-   writers - that stresses the locking code. Build with TDB_DEBUG=1
-   for best effect */
+   writers - that stresses the locking code. 
+*/
 
+#include "replace.h"
+#include "tdb.h"
+#include "system/time.h"
+#include "system/wait.h"
+#include "system/filesys.h"
+
+#ifdef HAVE_GETOPT_H
+#include <getopt.h>
+#endif
 
 
 #define REOPEN_PROB 30
 #define DELETE_PROB 8
 #define STORE_PROB 4
 #define APPEND_PROB 6
-#define LOCKSTORE_PROB 0
+#define TRANSACTION_PROB 10
+#define LOCKSTORE_PROB 5
 #define TRAVERSE_PROB 20
+#define TRAVERSE_READ_PROB 20
 #define CULL_PROB 100
 #define KEYLEN 3
 #define DATALEN 100
-#define LOCKLEN 20
 
-static TDB_CONTEXT *db;
+static struct tdb_context *db;
+static int in_transaction;
+static int error_count;
 
-static void tdb_log(TDB_CONTEXT *tdb, int level, const char *format, ...)
+#ifdef PRINTF_ATTRIBUTE
+static void tdb_log(struct tdb_context *tdb, enum tdb_debug_level level, const char *format, ...) PRINTF_ATTRIBUTE(3,4);
+#endif
+static void tdb_log(struct tdb_context *tdb, enum tdb_debug_level level, const char *format, ...)
 {
 	va_list ap;
     
+	error_count++;
+
 	va_start(ap, format);
 	vfprintf(stdout, format, ap);
 	va_end(ap);
@@ -50,10 +52,10 @@ static void tdb_log(TDB_CONTEXT *tdb, int level, const char *format, ...)
 #endif	
 }
 
-static void fatal(char *why)
+static void fatal(const char *why)
 {
 	perror(why);
-	exit(1);
+	error_count++;
 }
 
 static char *randbuf(int len)
@@ -69,41 +71,62 @@ static char *randbuf(int len)
 	return buf;
 }
 
-static int cull_traverse(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA dbuf,
+static int cull_traverse(struct tdb_context *tdb, TDB_DATA key, TDB_DATA dbuf,
 			 void *state)
 {
+#if CULL_PROB
 	if (random() % CULL_PROB == 0) {
 		tdb_delete(tdb, key);
 	}
+#endif
 	return 0;
 }
 
 static void addrec_db(void)
 {
-	int klen, dlen, slen;
-	char *k, *d, *s;
-	TDB_DATA key, data, lockkey;
+	int klen, dlen;
+	char *k, *d;
+	TDB_DATA key, data;
 
 	klen = 1 + (rand() % KEYLEN);
 	dlen = 1 + (rand() % DATALEN);
-	slen = 1 + (rand() % LOCKLEN);
 
 	k = randbuf(klen);
 	d = randbuf(dlen);
-	s = randbuf(slen);
 
-	key.dptr = k;
+	key.dptr = (unsigned char *)k;
 	key.dsize = klen+1;
 
-	data.dptr = d;
+	data.dptr = (unsigned char *)d;
 	data.dsize = dlen+1;
 
-	lockkey.dptr = s;
-	lockkey.dsize = slen+1;
+#if TRANSACTION_PROB
+	if (in_transaction == 0 && random() % TRANSACTION_PROB == 0) {
+		if (tdb_transaction_start(db) != 0) {
+			fatal("tdb_transaction_start failed");
+		}
+		in_transaction++;
+		goto next;
+	}
+	if (in_transaction && random() % TRANSACTION_PROB == 0) {
+		if (tdb_transaction_commit(db) != 0) {
+			fatal("tdb_transaction_commit failed");
+		}
+		in_transaction--;
+		goto next;
+	}
+	if (in_transaction && random() % TRANSACTION_PROB == 0) {
+		if (tdb_transaction_cancel(db) != 0) {
+			fatal("tdb_transaction_cancel failed");
+		}
+		in_transaction--;
+		goto next;
+	}
+#endif
 
 #if REOPEN_PROB
-	if (random() % REOPEN_PROB == 0) {
-		tdb_reopen_all(1);
+	if (in_transaction == 0 && random() % REOPEN_PROB == 0) {
+		tdb_reopen_all(0);
 		goto next;
 	} 
 #endif
@@ -135,13 +158,13 @@ static void addrec_db(void)
 
 #if LOCKSTORE_PROB
 	if (random() % LOCKSTORE_PROB == 0) {
-		tdb_chainlock(db, lockkey);
+		tdb_chainlock(db, key);
 		data = tdb_fetch(db, key);
 		if (tdb_store(db, key, data, TDB_REPLACE) != 0) {
 			fatal("tdb_store failed");
 		}
 		if (data.dptr) free(data.dptr);
-		tdb_chainunlock(db, lockkey);
+		tdb_chainunlock(db, key);
 		goto next;
 	} 
 #endif
@@ -153,75 +176,143 @@ static void addrec_db(void)
 	}
 #endif
 
+#if TRAVERSE_READ_PROB
+	if (random() % TRAVERSE_READ_PROB == 0) {
+		tdb_traverse_read(db, NULL, NULL);
+		goto next;
+	}
+#endif
+
 	data = tdb_fetch(db, key);
 	if (data.dptr) free(data.dptr);
 
 next:
 	free(k);
 	free(d);
-	free(s);
 }
 
-static int traverse_fn(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA dbuf,
+static int traverse_fn(struct tdb_context *tdb, TDB_DATA key, TDB_DATA dbuf,
                        void *state)
 {
 	tdb_delete(tdb, key);
 	return 0;
 }
 
-#ifndef NPROC
-#define NPROC 6
-#endif
-
-#ifndef NLOOPS
-#define NLOOPS 200000
-#endif
+static void usage(void)
+{
+	printf("Usage: tdbtorture [-n NUM_PROCS] [-l NUM_LOOPS] [-s SEED] [-H HASH_SIZE]\n");
+	exit(0);
+}
 
-int main(int argc, char *argv[])
+ int main(int argc, char * const *argv)
 {
-	int i, seed=0;
-	int loops = NLOOPS;
-	pid_t pids[NPROC];
+	int i, seed = -1;
+	int num_procs = 3;
+	int num_loops = 5000;
+	int hash_size = 2;
+	int c;
+	extern char *optarg;
+	pid_t *pids;
+
+	struct tdb_logging_context log_ctx;
+	log_ctx.log_fn = tdb_log;
+
+	while ((c = getopt(argc, argv, "n:l:s:H:h")) != -1) {
+		switch (c) {
+		case 'n':
+			num_procs = strtol(optarg, NULL, 0);
+			break;
+		case 'l':
+			num_loops = strtol(optarg, NULL, 0);
+			break;
+		case 'H':
+			hash_size = strtol(optarg, NULL, 0);
+			break;
+		case 's':
+			seed = strtol(optarg, NULL, 0);
+			break;
+		default:
+			usage();
+		}
+	}
+
+	unlink("torture.tdb");
+
+	pids = calloc(sizeof(pid_t), num_procs);
+	pids[0] = getpid();
 
-	db = tdb_open("torture.tdb", 0, TDB_CLEAR_IF_FIRST, 
-		      O_RDWR | O_CREAT, 0600);
+	for (i=0;i<num_procs-1;i++) {
+		if ((pids[i+1]=fork()) == 0) break;
+	}
+
+	db = tdb_open_ex("torture.tdb", hash_size, TDB_CLEAR_IF_FIRST, 
+			 O_RDWR | O_CREAT, 0600, &log_ctx, NULL);
 	if (!db) {
 		fatal("db open failed");
 	}
 
-	for (i=0;i<NPROC;i++) {
-		pids[i] = fork();
-		if (pids[i] == 0) {
-			tdb_reopen_all(1);
+	if (seed == -1) {
+		seed = (getpid() + time(NULL)) & 0x7FFFFFFF;
+	}
 
-			tdb_logging_function(db, tdb_log);
+	if (i == 0) {
+		printf("testing with %d processes, %d loops, %d hash_size, seed=%d\n", 
+		       num_procs, num_loops, hash_size, seed);
+	}
 
-			srand(seed + getpid());
-			srandom(seed + getpid() + time(NULL));
-			for (i=0;i<loops;i++) addrec_db();
+	srand(seed + i);
+	srandom(seed + i);
 
-			tdb_traverse(db, NULL, NULL);
-			tdb_traverse(db, traverse_fn, NULL);
-			tdb_traverse(db, traverse_fn, NULL);
+	for (i=0;i<num_loops && error_count == 0;i++) {
+		addrec_db();
+	}
 
-			tdb_close(db);
-			exit(0);
-		}
+	if (error_count == 0) {
+		tdb_traverse_read(db, NULL, NULL);
+		tdb_traverse(db, traverse_fn, NULL);
+		tdb_traverse(db, traverse_fn, NULL);
 	}
 
-	for (i=0;i<NPROC;i++) {
-		int status;
-		if (waitpid(pids[i], &status, 0) != pids[i]) {
-			printf("failed to wait for %d\n",
-			       (int)pids[i]);
+	tdb_close(db);
+
+	if (getpid() != pids[0]) {
+		return error_count;
+	}
+
+	for (i=1;i<num_procs;i++) {
+		int status, j;
+		pid_t pid;
+		if (error_count != 0) {
+			/* try and stop the test on any failure */
+			for (j=1;j<num_procs;j++) {
+				if (pids[j] != 0) {
+					kill(pids[j], SIGTERM);
+				}
+			}
+		}
+		pid = waitpid(-1, &status, 0);
+		if (pid == -1) {
+			perror("failed to wait for child\n");
+			exit(1);
+		}
+		for (j=1;j<num_procs;j++) {
+			if (pids[j] == pid) break;
+		}
+		if (j == num_procs) {
+			printf("unknown child %d exited!?\n", (int)pid);
 			exit(1);
 		}
 		if (WEXITSTATUS(status) != 0) {
 			printf("child %d exited with status %d\n",
-			       (int)pids[i], WEXITSTATUS(status));
-			exit(1);
+			       (int)pid, WEXITSTATUS(status));
+			error_count++;
 		}
+		pids[j] = 0;
 	}
-	printf("OK\n");
-	return 0;
+
+	if (error_count == 0) {
+		printf("OK\n");
+	}
+
+	return error_count;
 }