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)

9 files changed, 292 insertions, 261 deletions

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;
 }