r24284: change brlock_tdb.c to use the dbwrap API. This actually makes the

backend abstraction for brlock pointless, but I have left it in place
for now. It would be useful for other clustering systems that can't
map to dbwrap, and would also be useful if we wanted to keep the
remote function call capabilities in ctdb instead of the less
efficient fetch_locked() call in dbwrap
(This used to be commit 912c014b7c131ab051ff6eb2db4e68cb6fbbeb14)

3 files changed, 67 insertions, 100 deletions

diff --git a/source4/ntvfs/common/brlock.c b/source4/ntvfs/common/brlock.c
index 56d3ff80f6b..27d7437f4ff 100644
--- a/source4/ntvfs/common/brlock.c
+++ b/source4/ntvfs/common/brlock.c
@@ -52,11 +52,7 @@ struct brl_context *brl_init(TALLOC_CTX *mem_ctx, struct server_id server,
 			     struct messaging_context *messaging_ctx)
 {
 	if (ops == NULL) {
-		if (lp_parm_bool(-1, "ctdb", "brlock", False)) {
-			brl_ctdb_init_ops();
-		} else {
-			brl_tdb_init_ops();
-		}
+		brl_tdb_init_ops();
 	}
 	return ops->brl_init(mem_ctx, server, messaging_ctx);
 }
diff --git a/source4/ntvfs/common/brlock_tdb.c b/source4/ntvfs/common/brlock_tdb.c
index 6e75a6fb0ac..25a0a040a58 100644
--- a/source4/ntvfs/common/brlock_tdb.c
+++ b/source4/ntvfs/common/brlock_tdb.c
@@ -28,7 +28,7 @@
 #include "system/filesys.h"
 #include "lib/tdb/include/tdb.h"
 #include "messaging/messaging.h"
-#include "db_wrap.h"
+#include "lib/dbwrap/dbwrap.h"
 #include "lib/messaging/irpc.h"
 #include "libcli/libcli.h"
 #include "cluster/cluster.h"
@@ -45,7 +45,7 @@
 
 /* this struct is typicaly attached to tcon */
 struct brl_context {
-	struct tdb_wrap *w;
+	struct db_context *db;
 	struct server_id server;
 	struct messaging_context *messaging_ctx;
 };
@@ -94,8 +94,8 @@ static struct brl_context *brl_tdb_init(TALLOC_CTX *mem_ctx, struct server_id se
 		return NULL;
 	}
 
-	brl->w = cluster_tdb_tmp_open(brl, "brlock.tdb", TDB_DEFAULT);
-	if (brl->w == NULL) {
+	brl->db = db_tmp_open(brl, "brlock.tdb", TDB_DEFAULT);
+	if (brl->db == NULL) {
 		talloc_free(brl);
 		return NULL;
 	}
@@ -281,11 +281,13 @@ static NTSTATUS brl_tdb_lock(struct brl_context *brl,
 	int count=0, i;
 	struct lock_struct lock, *locks=NULL;
 	NTSTATUS status;
+	struct db_record *rec = NULL;
 
 	kbuf.dptr = brlh->key.data;
 	kbuf.dsize = brlh->key.length;
 
-	if (tdb_chainlock(brl->w->tdb, kbuf) != 0) {
+	rec = brl->db->fetch_locked(brl->db, brl, kbuf);
+	if (rec == NULL) {
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
 	}
 
@@ -303,12 +305,12 @@ static NTSTATUS brl_tdb_lock(struct brl_context *brl,
 		brlh->last_lock = lock;
 
 		if (NT_STATUS_IS_OK(status)) {
-			tdb_chainunlock(brl->w->tdb, kbuf);
+			talloc_free(rec);
 			return NT_STATUS_OK;
 		}
 	}
 
-	dbuf = tdb_fetch(brl->w->tdb, kbuf);
+	dbuf = rec->value;
 
 	lock.context.smbpid = smbpid;
 	lock.context.server = brl->server;
@@ -333,7 +335,7 @@ static NTSTATUS brl_tdb_lock(struct brl_context *brl,
 	}
 
 	/* no conflicts - add it to the list of locks */
-	locks = realloc_p(locks, struct lock_struct, count+1);
+	locks = talloc_realloc(rec, locks, struct lock_struct, count+1);
 	if (!locks) {
 		status = NT_STATUS_NO_MEMORY;
 		goto fail;
@@ -343,13 +345,12 @@ static NTSTATUS brl_tdb_lock(struct brl_context *brl,
 	locks[count] = lock;
 	dbuf.dsize += sizeof(lock);
 
-	if (tdb_store(brl->w->tdb, kbuf, dbuf, TDB_REPLACE) != 0) {
-		status = NT_STATUS_INTERNAL_DB_CORRUPTION;
+	status = rec->store(rec, dbuf, TDB_REPLACE);
+	if (!NT_STATUS_IS_OK(status)) {
 		goto fail;
 	}
 
-	free(dbuf.dptr);
-	tdb_chainunlock(brl->w->tdb, kbuf);
+	talloc_free(rec);
 
 	/* the caller needs to know if the real lock was granted. If
 	   we have reached here then it must be a pending lock that
@@ -361,9 +362,7 @@ static NTSTATUS brl_tdb_lock(struct brl_context *brl,
 	return NT_STATUS_OK;
 
  fail:
-
-	free(dbuf.dptr);
-	tdb_chainunlock(brl->w->tdb, kbuf);
+	talloc_free(rec);
 	return status;
 }
 
@@ -431,20 +430,23 @@ static NTSTATUS brl_tdb_unlock(struct brl_context *brl,
 	struct lock_struct *locks, *lock;
 	struct lock_context context;
 	NTSTATUS status;
+	struct db_record *rec = NULL;
 
 	kbuf.dptr = brlh->key.data;
 	kbuf.dsize = brlh->key.length;
 
-	if (tdb_chainlock(brl->w->tdb, kbuf) != 0) {
+	rec = brl->db->fetch_locked(brl->db, brl, kbuf);
+	if (rec == NULL) {
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
 	}
 
-	dbuf = tdb_fetch(brl->w->tdb, kbuf);
-	if (!dbuf.dptr) {
-		tdb_chainunlock(brl->w->tdb, kbuf);
+	if (!rec->value.dptr) {
+		talloc_free(rec);
 		return NT_STATUS_RANGE_NOT_LOCKED;
 	}
 
+	dbuf = rec->value;
+
 	context.smbpid = smbpid;
 	context.server = brl->server;
 	context.ctx = brl;
@@ -477,43 +479,27 @@ static NTSTATUS brl_tdb_unlock(struct brl_context *brl,
 	}
 
 found:
-	if (i < count) {
-		/* found it - delete it */
-		if (count == 1) {
-			if (tdb_delete(brl->w->tdb, kbuf) != 0) {
-				status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-				goto fail;
-			}
-		} else {
-			struct lock_struct removed_lock = *lock;
-			if (i < count-1) {
-				memmove(&locks[i], &locks[i+1], 
-					sizeof(*locks)*((count-1) - i));
-			}
-			count--;
-			
-			/* send notifications for any relevant pending locks */
-			brl_tdb_notify_unlock(brl, locks, count, &removed_lock);
-			
-			dbuf.dsize = count * sizeof(*locks);
-			
-			if (tdb_store(brl->w->tdb, kbuf, dbuf, TDB_REPLACE) != 0) {
-				status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-				goto fail;
-			}
+	if (i == count) {
+		status = NT_STATUS_RANGE_NOT_LOCKED;
+	} else if (count == 1) {
+		status = rec->delete_rec(rec);
+	} else {
+		struct lock_struct removed_lock = *lock;
+		if (i < count-1) {
+			memmove(&locks[i], &locks[i+1], 
+				sizeof(*locks)*((count-1) - i));
 		}
+		count--;
 		
-		free(dbuf.dptr);
-		tdb_chainunlock(brl->w->tdb, kbuf);
-		return NT_STATUS_OK;
+		/* send notifications for any relevant pending locks */
+		brl_tdb_notify_unlock(brl, locks, count, &removed_lock);
+		
+		dbuf.dsize = count * sizeof(*locks);
+		
+		status = rec->store(rec, dbuf, TDB_REPLACE);
 	}
-	
-	/* we didn't find it */
-	status = NT_STATUS_RANGE_NOT_LOCKED;
 
- fail:
-	free(dbuf.dptr);
-	tdb_chainunlock(brl->w->tdb, kbuf);
+ 	talloc_free(rec);
 	return status;
 }
 
@@ -531,24 +517,24 @@ static NTSTATUS brl_tdb_remove_pending(struct brl_context *brl,
 	int count, i;
 	struct lock_struct *locks;
 	NTSTATUS status;
+	struct db_record *rec = NULL;
 
 	kbuf.dptr = brlh->key.data;
 	kbuf.dsize = brlh->key.length;
 
-	if (tdb_chainlock(brl->w->tdb, kbuf) != 0) {
+	rec = brl->db->fetch_locked(brl->db, brl, kbuf);
+	if (rec == NULL) {
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
 	}
 
-	dbuf = tdb_fetch(brl->w->tdb, kbuf);
-	if (!dbuf.dptr) {
-		tdb_chainunlock(brl->w->tdb, kbuf);
-		return NT_STATUS_RANGE_NOT_LOCKED;
-	}
+	dbuf = rec->value;
 
 	/* there are existing locks - find a match */
 	locks = (struct lock_struct *)dbuf.dptr;
 	count = dbuf.dsize / sizeof(*locks);
 
+	status = NT_STATUS_RANGE_NOT_LOCKED;
+
 	for (i=0; i<count; i++) {
 		struct lock_struct *lock = &locks[i];
 		
@@ -557,10 +543,7 @@ static NTSTATUS brl_tdb_remove_pending(struct brl_context *brl,
 		    cluster_id_equal(&lock->context.server, &brl->server)) {
 			/* found it - delete it */
 			if (count == 1) {
-				if (tdb_delete(brl->w->tdb, kbuf) != 0) {
-					status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-					goto fail;
-				}
+				status = rec->delete_rec(rec);
 			} else {
 				if (i < count-1) {
 					memmove(&locks[i], &locks[i+1], 
@@ -568,24 +551,13 @@ static NTSTATUS brl_tdb_remove_pending(struct brl_context *brl,
 				}
 				count--;
 				dbuf.dsize = count * sizeof(*locks);
-				if (tdb_store(brl->w->tdb, kbuf, dbuf, TDB_REPLACE) != 0) {
-					status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-					goto fail;
-				}
-			}
-			
-			free(dbuf.dptr);
-			tdb_chainunlock(brl->w->tdb, kbuf);
-			return NT_STATUS_OK;
+				status = rec->store(rec, dbuf, TDB_REPLACE);
+			}			
+			break;
 		}
 	}
-	
-	/* we didn't find it */
-	status = NT_STATUS_RANGE_NOT_LOCKED;
 
- fail:
-	free(dbuf.dptr);
-	tdb_chainunlock(brl->w->tdb, kbuf);
+	talloc_free(rec);
 	return status;
 }
 
@@ -602,12 +574,12 @@ static NTSTATUS brl_tdb_locktest(struct brl_context *brl,
 	TDB_DATA kbuf, dbuf;
 	int count, i;
 	struct lock_struct lock, *locks;
+	NTSTATUS status;
 
 	kbuf.dptr = brlh->key.data;
 	kbuf.dsize = brlh->key.length;
 
-	dbuf = tdb_fetch(brl->w->tdb, kbuf);
-	if (dbuf.dptr == NULL) {
+	if (brl->db->fetch(brl->db, brl, kbuf, &dbuf) != 0) {
 		return NT_STATUS_OK;
 	}
 
@@ -623,15 +595,17 @@ static NTSTATUS brl_tdb_locktest(struct brl_context *brl,
 	locks = (struct lock_struct *)dbuf.dptr;
 	count = dbuf.dsize / sizeof(*locks);
 
+	status = NT_STATUS_OK;
+
 	for (i=0; i<count; i++) {
 		if (brl_tdb_conflict_other(&locks[i], &lock)) {
-			free(dbuf.dptr);
-			return NT_STATUS_FILE_LOCK_CONFLICT;
+			status = NT_STATUS_FILE_LOCK_CONFLICT;
+			break;
 		}
 	}
 
-	free(dbuf.dptr);
-	return NT_STATUS_OK;
+	talloc_free(dbuf.dptr);
+	return status;
 }
 
 
@@ -645,17 +619,19 @@ static NTSTATUS brl_tdb_close(struct brl_context *brl,
 	int count, i, dcount=0;
 	struct lock_struct *locks;
 	NTSTATUS status;
+	struct db_record *rec = NULL;
 
 	kbuf.dptr = brlh->key.data;
 	kbuf.dsize = brlh->key.length;
 
-	if (tdb_chainlock(brl->w->tdb, kbuf) != 0) {
+	rec = brl->db->fetch_locked(brl->db, brl, kbuf);
+	if (rec == NULL) {
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
 	}
 
-	dbuf = tdb_fetch(brl->w->tdb, kbuf);
+	dbuf = rec->value;
 	if (!dbuf.dptr) {
-		tdb_chainunlock(brl->w->tdb, kbuf);
+		talloc_free(rec);
 		return NT_STATUS_OK;
 	}
 
@@ -683,9 +659,7 @@ static NTSTATUS brl_tdb_close(struct brl_context *brl,
 	status = NT_STATUS_OK;
 
 	if (count == 0) {
-		if (tdb_delete(brl->w->tdb, kbuf) != 0) {
-			status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-		}
+		status = rec->delete_rec(rec);
 	} else if (dcount != 0) {
 		/* tell all pending lock holders for this file that
 		   they have a chance now. This is a bit indiscriminant,
@@ -694,13 +668,10 @@ static NTSTATUS brl_tdb_close(struct brl_context *brl,
 
 		dbuf.dsize = count * sizeof(*locks);
 
-		if (tdb_store(brl->w->tdb, kbuf, dbuf, TDB_REPLACE) != 0) {
-			status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-		}
+		status = rec->store(rec, dbuf, TDB_REPLACE);
 	}
 
-	free(dbuf.dptr);
-	tdb_chainunlock(brl->w->tdb, kbuf);
+	talloc_free(rec);
 
 	return status;
 }
diff --git a/source4/ntvfs/common/config.mk b/source4/ntvfs/common/config.mk
index 5c744f9c903..2fc0942ed40 100644
--- a/source4/ntvfs/common/config.mk
+++ b/source4/ntvfs/common/config.mk
@@ -9,7 +9,7 @@ OBJ_FILES = \
 		opendb.o \
 		opendb_tdb.o \
 		notify.o
-PUBLIC_DEPENDENCIES = NDR_OPENDB NDR_NOTIFY sys_notify share
+PUBLIC_DEPENDENCIES = NDR_OPENDB NDR_NOTIFY sys_notify share LIBDBWRAP
 PRIVATE_DEPENDENCIES = brlock_ctdb opendb_ctdb
 # End LIBRARY ntvfs_common
 ################################################