r22070: merge in the changes from the bzr ctdb tree, and convert the brlock

ctdb backend to use the updated multi-database API
(This used to be commit 44dcac9e4d81bfc078512248967b6240db9d1bd8)

8 files changed, 211 insertions, 67 deletions

diff --git a/source4/cluster/ctdb/brlock_ctdb.c b/source4/cluster/ctdb/brlock_ctdb.c
index 04f617beb9..b1e8e64d40 100644
--- a/source4/cluster/ctdb/brlock_ctdb.c
+++ b/source4/cluster/ctdb/brlock_ctdb.c
@@ -46,6 +46,7 @@ enum my_functions {FUNC_BRL_LOCK=1, FUNC_BRL_UNLOCK=2,
 /* this struct is typically attached to tcon */
 struct brl_context {
 	struct ctdb_context *ctdb;
+	struct ctdb_db_context *ctdb_db;
 	struct server_id server;
 	struct messaging_context *messaging_ctx;
 };
@@ -115,6 +116,12 @@ static struct brl_context *brl_ctdb_init(TALLOC_CTX *mem_ctx, struct server_id s
 	}
 
 	brl->ctdb = ctdb;
+	brl->ctdb_db = ctdb_db_handle(ctdb, "brlock");
+	if (brl->ctdb_db == NULL) {
+		DEBUG(0,("Failed to get attached ctdb db handle for brlock\n"));
+		talloc_free(brl);
+		return NULL;
+	}
 	brl->server = server;
 	brl->messaging_ctx = messaging_ctx;
 
@@ -393,7 +400,7 @@ static NTSTATUS brl_ctdb_lock(struct brl_context *brl,
 	req.brl = brl;
 	req.ntvfs = brlh->ntvfs;
 
-	ret = ctdb_call(brl->ctdb, &call);
+	ret = ctdb_call(brl->ctdb_db, &call);
 	if (ret == -1) {
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
 	}
@@ -617,7 +624,7 @@ static NTSTATUS brl_ctdb_unlock(struct brl_context *brl,
 	req.brl = brl;
 	req.ntvfs = brlh->ntvfs;
 		
-	ret = ctdb_call(brl->ctdb, &call);
+	ret = ctdb_call(brl->ctdb_db, &call);
 	if (ret == -1) {
 		DEBUG(0,("ctdb_call failed - %s\n", __location__));
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
@@ -705,7 +712,7 @@ static NTSTATUS brl_ctdb_remove_pending(struct brl_context *brl,
 	req.notify_ptr = notify_ptr;
 	req.server = brl->server;
 		
-	ret = ctdb_call(brl->ctdb, &call);
+	ret = ctdb_call(brl->ctdb_db, &call);
 	if (ret == -1) {
 		DEBUG(0,("ctdb_call failed - %s\n", __location__));
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
@@ -792,7 +799,7 @@ static NTSTATUS brl_ctdb_locktest(struct brl_context *brl,
 	req.brl = brl;
 	req.ntvfs = brlh->ntvfs;
 
-	ret = ctdb_call(brl->ctdb, &call);
+	ret = ctdb_call(brl->ctdb_db, &call);
 	if (ret == -1) {
 		DEBUG(0,("ctdb_call failed - %s\n", __location__));
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
@@ -887,7 +894,7 @@ static NTSTATUS brl_ctdb_close(struct brl_context *brl,
 	req.server = brl->server;
 	req.ntvfs = brlh->ntvfs;
 
-	ret = ctdb_call(brl->ctdb, &call);
+	ret = ctdb_call(brl->ctdb_db, &call);
 	if (ret == -1) {
 		DEBUG(0,("ctdb_call failed - %s\n", __location__));
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
@@ -914,12 +921,19 @@ void brl_ctdb_init_ops(void)
 {
 	struct ctdb_context *ctdb = talloc_get_type(cluster_backend_handle(), 
 						    struct ctdb_context);
+	struct ctdb_db_context *ctdb_db;
 
 	brl_set_ops(&brlock_tdb_ops);
 
-	ctdb_set_call(ctdb, brl_ctdb_lock_func,  FUNC_BRL_LOCK);
-	ctdb_set_call(ctdb, brl_ctdb_unlock_func,  FUNC_BRL_UNLOCK);
-	ctdb_set_call(ctdb, brl_ctdb_remove_pending_func,  FUNC_BRL_REMOVE_PENDING);
-	ctdb_set_call(ctdb, brl_ctdb_locktest_func,  FUNC_BRL_LOCKTEST);
-	ctdb_set_call(ctdb, brl_ctdb_close_func,  FUNC_BRL_CLOSE);
+	ctdb_db = ctdb_db_handle(ctdb, "brlock");
+	if (ctdb_db == NULL) {
+		DEBUG(0,("Failed to get attached ctdb db handle for brlock\n"));
+		return;
+	}
+
+	ctdb_set_call(ctdb_db, brl_ctdb_lock_func,  FUNC_BRL_LOCK);
+	ctdb_set_call(ctdb_db, brl_ctdb_unlock_func,  FUNC_BRL_UNLOCK);
+	ctdb_set_call(ctdb_db, brl_ctdb_remove_pending_func,  FUNC_BRL_REMOVE_PENDING);
+	ctdb_set_call(ctdb_db, brl_ctdb_locktest_func,  FUNC_BRL_LOCKTEST);
+	ctdb_set_call(ctdb_db, brl_ctdb_close_func,  FUNC_BRL_CLOSE);
 }
diff --git a/source4/cluster/ctdb/common/ctdb.c b/source4/cluster/ctdb/common/ctdb.c
index 59a14d4dbd..e4f7289550 100644
--- a/source4/cluster/ctdb/common/ctdb.c
+++ b/source4/cluster/ctdb/common/ctdb.c
@@ -32,10 +32,19 @@
 int ctdb_set_transport(struct ctdb_context *ctdb, const char *transport)
 {
 	int ctdb_tcp_init(struct ctdb_context *ctdb);
+#ifdef USE_INFINIBAND
+	int ctdb_ibw_init(struct ctdb_context *ctdb);
+#endif /* USE_INFINIBAND */
 
 	if (strcmp(transport, "tcp") == 0) {
 		return ctdb_tcp_init(ctdb);
 	}
+#ifdef USE_INFINIBAND
+	if (strcmp(transport, "ib") == 0) {
+		return ctdb_ibw_init(ctdb);
+	}
+#endif /* USE_INFINIBAND */
+
 	ctdb_set_error(ctdb, "Unknown transport '%s'\n", transport);
 	return -1;
 }
@@ -141,15 +150,15 @@ int ctdb_set_address(struct ctdb_context *ctdb, const char *address)
 /*
   add a node to the list of active nodes
 */
-int ctdb_set_call(struct ctdb_context *ctdb, ctdb_fn_t fn, int id)
+int ctdb_set_call(struct ctdb_db_context *ctdb_db, ctdb_fn_t fn, int id)
 {
 	struct ctdb_registered_call *call;
 
-	call = talloc(ctdb, struct ctdb_registered_call);
+	call = talloc(ctdb_db, struct ctdb_registered_call);
 	call->fn = fn;
 	call->id = id;
 
-	DLIST_ADD(ctdb->calls, call);	
+	DLIST_ADD(ctdb_db->calls, call);	
 	return 0;
 }
 
diff --git a/source4/cluster/ctdb/common/ctdb_call.c b/source4/cluster/ctdb/common/ctdb_call.c
index fb29aad9ac..75355f7ae7 100644
--- a/source4/cluster/ctdb/common/ctdb_call.c
+++ b/source4/cluster/ctdb/common/ctdb_call.c
@@ -31,12 +31,13 @@
 /*
   local version of ctdb_call
 */
-static int ctdb_call_local(struct ctdb_context *ctdb, struct ctdb_call *call,
+static int ctdb_call_local(struct ctdb_db_context *ctdb_db, struct ctdb_call *call,
 			   struct ctdb_ltdb_header *header, TDB_DATA *data,
 			   uint32_t caller)
 {
 	struct ctdb_call_info *c;
 	struct ctdb_registered_call *fn;
+	struct ctdb_context *ctdb = ctdb_db->ctdb;
 
 	c = talloc(ctdb, struct ctdb_call_info);
 	CTDB_NO_MEMORY(ctdb, c);
@@ -50,7 +51,7 @@ static int ctdb_call_local(struct ctdb_context *ctdb, struct ctdb_call *call,
 	c->reply_data = NULL;
 	c->status = 0;
 
-	for (fn=ctdb->calls;fn;fn=fn->next) {
+	for (fn=ctdb_db->calls;fn;fn=fn->next) {
 		if (fn->id == call->call_id) break;
 	}
 	if (fn == NULL) {
@@ -76,7 +77,7 @@ static int ctdb_call_local(struct ctdb_context *ctdb, struct ctdb_call *call,
 	}
 
 	if (c->new_data) {
-		if (ctdb_ltdb_store(ctdb, call->key, header, *c->new_data) != 0) {
+		if (ctdb_ltdb_store(ctdb_db, call->key, header, *c->new_data) != 0) {
 			ctdb_set_error(ctdb, "ctdb_call tdb_store failed\n");
 			return -1;
 		}
@@ -170,12 +171,13 @@ static void ctdb_call_send_redirect(struct ctdb_context *ctdb,
   always knows who the dmaster is. The lmaster will then send a
   CTDB_REPLY_DMASTER to the new dmaster
 */
-static void ctdb_call_send_dmaster(struct ctdb_context *ctdb, 
+static void ctdb_call_send_dmaster(struct ctdb_db_context *ctdb_db, 
 				   struct ctdb_req_call *c, 
 				   struct ctdb_ltdb_header *header,
 				   TDB_DATA *key, TDB_DATA *data)
 {
 	struct ctdb_req_dmaster *r;
+	struct ctdb_context *ctdb = ctdb_db->ctdb;
 	int len;
 	
 	len = offsetof(struct ctdb_req_dmaster, data) + key->dsize + data->dsize;
@@ -186,6 +188,7 @@ static void ctdb_call_send_dmaster(struct ctdb_context *ctdb,
 	r->hdr.destnode  = ctdb_lmaster(ctdb, key);
 	r->hdr.srcnode   = ctdb->vnn;
 	r->hdr.reqid     = c->hdr.reqid;
+	r->db_id         = c->db_id;
 	r->dmaster       = header->laccessor;
 	r->keylen        = key->dsize;
 	r->datalen       = data->dsize;
@@ -200,7 +203,7 @@ static void ctdb_call_send_dmaster(struct ctdb_context *ctdb,
 
 		/* update the ltdb to record the new dmaster */
 		header->dmaster = r->hdr.destnode;
-		ctdb_ltdb_store(ctdb, *key, header, *data);
+		ctdb_ltdb_store(ctdb_db, *key, header, *data);
 	}
 
 	talloc_free(r);
@@ -219,6 +222,7 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
 	struct ctdb_reply_dmaster *r;
 	TDB_DATA key, data, data2;
 	struct ctdb_ltdb_header header;
+	struct ctdb_db_context *ctdb_db;
 	int ret, len;
 
 	key.dptr = c->data;
@@ -226,8 +230,18 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
 	data.dptr = c->data + c->keylen;
 	data.dsize = c->datalen;
 
+	for (ctdb_db=ctdb->db_list; ctdb_db; ctdb_db=ctdb_db->next) {
+		if (ctdb_db->db_id == c->db_id) {
+			break;
+		}
+	}
+	if (!ctdb_db) {
+		ctdb_send_error(ctdb, hdr, ret, "Unknown database in request. db_id==0x%08x",c->db_id);
+		return;
+	}
+
 	/* fetch the current record */
-	ret = ctdb_ltdb_fetch(ctdb, key, &header, &data2);
+	ret = ctdb_ltdb_fetch(ctdb_db, key, &header, &data2);
 	if (ret != 0) {
 		ctdb_fatal(ctdb, "ctdb_req_dmaster failed to fetch record");
 		return;
@@ -240,7 +254,7 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
 	}
 
 	header.dmaster = c->dmaster;
-	if (ctdb_ltdb_store(ctdb, key, &header, data) != 0) {
+	if (ctdb_ltdb_store(ctdb_db, key, &header, data) != 0) {
 		ctdb_fatal(ctdb, "ctdb_req_dmaster unable to update dmaster");
 		return;
 	}
@@ -278,6 +292,17 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 	int ret, len;
 	struct ctdb_ltdb_header header;
 	struct ctdb_call call;
+	struct ctdb_db_context *ctdb_db;
+
+	for (ctdb_db=ctdb->db_list; ctdb_db; ctdb_db=ctdb_db->next) {
+		if (ctdb_db->db_id == c->db_id) {
+			break;
+		}
+	}
+	if (!ctdb_db) {
+		ctdb_send_error(ctdb, hdr, ret, "Unknown database in request. db_id==0x%08x",c->db_id);
+		return;
+	}
 
 	call.call_id  = c->callid;
 	call.key.dptr = c->data;
@@ -289,7 +314,7 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 	   fetches the record data (if any), thus avoiding a 2nd fetch of the data 
 	   if the call will be answered locally */
 
-	ret = ctdb_ltdb_fetch(ctdb, call.key, &header, &data);
+	ret = ctdb_ltdb_fetch(ctdb_db, call.key, &header, &data);
 	if (ret != 0) {
 		ctdb_send_error(ctdb, hdr, ret, "ltdb fetch failed in ctdb_request_call");
 		return;
@@ -307,12 +332,12 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 	   then give them the record */
 	if (header.laccessor == c->hdr.srcnode &&
 	    header.lacount >= ctdb->max_lacount) {
-		ctdb_call_send_dmaster(ctdb, c, &header, &call.key, &data);
+		ctdb_call_send_dmaster(ctdb_db, c, &header, &call.key, &data);
 		talloc_free(data.dptr);
 		return;
 	}
 
-	ctdb_call_local(ctdb, &call, &header, &data, c->hdr.srcnode);
+	ctdb_call_local(ctdb_db, &call, &header, &data, c->hdr.srcnode);
 
 	len = offsetof(struct ctdb_reply_call, data) + call.reply_data.dsize;
 	r = ctdb->methods->allocate_pkt(ctdb, len);
@@ -342,6 +367,7 @@ enum call_state {CTDB_CALL_WAIT, CTDB_CALL_DONE, CTDB_CALL_ERROR};
 struct ctdb_call_state {
 	enum call_state state;
 	struct ctdb_req_call *c;
+	struct ctdb_db_context *ctdb_db;
 	struct ctdb_node *node;
 	const char *errmsg;
 	struct ctdb_call call;
@@ -384,10 +410,15 @@ void ctdb_reply_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 {
 	struct ctdb_reply_dmaster *c = (struct ctdb_reply_dmaster *)hdr;
 	struct ctdb_call_state *state;
+	struct ctdb_db_context *ctdb_db;
 	TDB_DATA data;
 
 	state = idr_find(ctdb->idr, hdr->reqid);
-	if (state == NULL) return;
+	if (state == NULL) {
+		return;
+	}
+	ctdb_db = state->ctdb_db;
+
 
 	data.dptr = c->data;
 	data.dsize = c->datalen;
@@ -398,12 +429,12 @@ void ctdb_reply_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 	   and data */
 	state->header.dmaster = ctdb->vnn;
 
-	if (ctdb_ltdb_store(ctdb, state->call.key, &state->header, data) != 0) {
+	if (ctdb_ltdb_store(ctdb_db, state->call.key, &state->header, data) != 0) {
 		ctdb_fatal(ctdb, "ctdb_reply_dmaster store failed\n");
 		return;
 	}
 
-	ctdb_call_local(ctdb, &state->call, &state->header, &data, ctdb->vnn);
+	ctdb_call_local(ctdb_db, &state->call, &state->header, &data, ctdb->vnn);
 
 	state->state = CTDB_CALL_DONE;
 }
@@ -483,22 +514,24 @@ void ctdb_call_timeout(struct event_context *ev, struct timed_event *te,
   this is used so that locally processed ctdb_call requests are processed
   in an event driven manner
 */
-struct ctdb_call_state *ctdb_call_local_send(struct ctdb_context *ctdb, 
+struct ctdb_call_state *ctdb_call_local_send(struct ctdb_db_context *ctdb_db, 
 					     struct ctdb_call *call,
 					     struct ctdb_ltdb_header *header,
 					     TDB_DATA *data)
 {
 	struct ctdb_call_state *state;
+	struct ctdb_context *ctdb = ctdb_db->ctdb;
 	int ret;
 
-	state = talloc_zero(ctdb, struct ctdb_call_state);
+	state = talloc_zero(ctdb_db, struct ctdb_call_state);
 	CTDB_NO_MEMORY_NULL(ctdb, state);
 
 	state->state = CTDB_CALL_DONE;
 	state->node = ctdb->nodes[ctdb->vnn];
 	state->call = *call;
+	state->ctdb_db = ctdb_db;
 
-	ret = ctdb_call_local(ctdb, &state->call, header, data, ctdb->vnn);
+	ret = ctdb_call_local(ctdb_db, &state->call, header, data, ctdb->vnn);
 
 	return state;
 }
@@ -510,13 +543,14 @@ struct ctdb_call_state *ctdb_call_local_send(struct ctdb_context *ctdb,
   This constructs a ctdb_call request and queues it for processing. 
   This call never blocks.
 */
-struct ctdb_call_state *ctdb_call_send(struct ctdb_context *ctdb, struct ctdb_call *call)
+struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, struct ctdb_call *call)
 {
 	uint32_t len;
 	struct ctdb_call_state *state;
 	int ret;
 	struct ctdb_ltdb_header header;
 	TDB_DATA data;
+	struct ctdb_context *ctdb = ctdb_db->ctdb;
 
 	/*
 	  if we are the dmaster for this key then we don't need to
@@ -524,14 +558,14 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_context *ctdb, struct ctdb_ca
 	  locally. To find out if we are the dmaster we need to look
 	  in our ltdb
 	*/
-	ret = ctdb_ltdb_fetch(ctdb, call->key, &header, &data);
+	ret = ctdb_ltdb_fetch(ctdb_db, call->key, &header, &data);
 	if (ret != 0) return NULL;
 
 	if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
-		return ctdb_call_local_send(ctdb, call, &header, &data);
+		return ctdb_call_local_send(ctdb_db, call, &header, &data);
 	}
 
-	state = talloc_zero(ctdb, struct ctdb_call_state);
+	state = talloc_zero(ctdb_db, struct ctdb_call_state);
 	CTDB_NO_MEMORY_NULL(ctdb, state);
 
 	len = offsetof(struct ctdb_req_call, data) + call->key.dsize + call->call_data.dsize;
@@ -544,6 +578,7 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_context *ctdb, struct ctdb_ca
 	state->c->hdr.srcnode   = ctdb->vnn;
 	/* this limits us to 16k outstanding messages - not unreasonable */
 	state->c->hdr.reqid     = idr_get_new(ctdb->idr, state, 0xFFFF);
+	state->c->db_id         = ctdb_db->db_id;
 	state->c->callid        = call->call_id;
 	state->c->keylen        = call->key.dsize;
 	state->c->calldatalen   = call->call_data.dsize;
@@ -557,6 +592,7 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_context *ctdb, struct ctdb_ca
 	state->node   = ctdb->nodes[header.dmaster];
 	state->state  = CTDB_CALL_WAIT;
 	state->header = header;
+	state->ctdb_db = ctdb_db;
 
 	talloc_set_destructor(state, ctdb_call_destructor);
 
@@ -601,9 +637,9 @@ int ctdb_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
 /*
   full ctdb_call. Equivalent to a ctdb_call_send() followed by a ctdb_call_recv()
 */
-int ctdb_call(struct ctdb_context *ctdb, struct ctdb_call *call)
+int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call)
 {
 	struct ctdb_call_state *state;
-	state = ctdb_call_send(ctdb, call);
+	state = ctdb_call_send(ctdb_db, call);
 	return ctdb_call_recv(state, call);
 }
diff --git a/source4/cluster/ctdb/common/ctdb_ltdb.c b/source4/cluster/ctdb/common/ctdb_ltdb.c
index 10bcde43b5..189816229f 100644
--- a/source4/cluster/ctdb/common/ctdb_ltdb.c
+++ b/source4/cluster/ctdb/common/ctdb_ltdb.c
@@ -25,23 +25,63 @@
 #include "system/filesys.h"
 #include "../include/ctdb_private.h"
 #include "db_wrap.h"
+#include "lib/util/dlinklist.h"
 
+/*
+  find an attached ctdb_db handle given a name
+ */
+struct ctdb_db_context *ctdb_db_handle(struct ctdb_context *ctdb, const char *name)
+{
+	struct ctdb_db_context *tmp_db;
+	for (tmp_db=ctdb->db_list;tmp_db;tmp_db=tmp_db->next) {
+		if (strcmp(name, tmp_db->db_name) == 0) {
+			return tmp_db;
+		}
+	}
+	return NULL;
+}
 
 /*
   attach to a specific database
 */
-int ctdb_attach(struct ctdb_context *ctdb, const char *name, int tdb_flags, 
-		int open_flags, mode_t mode)
+struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb, const char *name, int tdb_flags, 
+				    int open_flags, mode_t mode)
 {
+	struct ctdb_db_context *ctdb_db, *tmp_db;
+	TDB_DATA data;
+
+	ctdb_db = talloc_zero(ctdb, struct ctdb_db_context);
+	CTDB_NO_MEMORY_NULL(ctdb, ctdb_db);
+
+	ctdb_db->ctdb = ctdb;
+	ctdb_db->db_name = talloc_strdup(ctdb_db, name);
+	CTDB_NO_MEMORY_NULL(ctdb, ctdb_db->db_name);
+
+	data.dptr = discard_const(name);
+	data.dsize = strlen(name);
+	ctdb_db->db_id = ctdb_hash(&data);
+
+	for (tmp_db=ctdb->db_list;tmp_db;tmp_db=tmp_db->next) {
+		if (tmp_db->db_id == ctdb_db->db_id) {
+			ctdb_set_error(ctdb, "CTDB database hash collission '%s' : '%s'",
+					name, tmp_db->db_name);
+			talloc_free(ctdb_db);
+			return NULL;
+		}
+	}
+
 	/* when we have a separate daemon this will need to be a real
 	   file, not a TDB_INTERNAL, so the parent can access it to
 	   for ltdb bypass */
-	ctdb->ltdb = tdb_wrap_open(ctdb, name, 0, TDB_INTERNAL, open_flags, mode);
-	if (ctdb->ltdb == NULL) {
+	ctdb_db->ltdb = tdb_wrap_open(ctdb, name, 0, TDB_INTERNAL, open_flags, mode);
+	if (ctdb_db->ltdb == NULL) {
 		ctdb_set_error(ctdb, "Failed to open tdb %s\n", name);
-		return -1;
+		talloc_free(ctdb_db);
+		return NULL;
 	}
-	return 0;
+
+	DLIST_ADD(ctdb->db_list, ctdb_db);
+	return ctdb_db;
 }
 
 /*
@@ -56,13 +96,13 @@ uint32_t ctdb_lmaster(struct ctdb_context *ctdb, const TDB_DATA *key)
 /*
   construct an initial header for a record with no ltdb header yet
 */
-static void ltdb_initial_header(struct ctdb_context *ctdb, 
+static void ltdb_initial_header(struct ctdb_db_context *ctdb_db, 
 				TDB_DATA key,
 				struct ctdb_ltdb_header *header)
 {
 	header->rsn = 0;
 	/* initial dmaster is the lmaster */
-	header->dmaster = ctdb_lmaster(ctdb, &key);
+	header->dmaster = ctdb_lmaster(ctdb_db->ctdb, &key);
 	header->laccessor = header->dmaster;
 	header->lacount = 0;
 }
@@ -73,16 +113,17 @@ static void ltdb_initial_header(struct ctdb_context *ctdb,
   and returning the body of the record. A valid (initial) header is
   returned if the record is not present
 */
-int ctdb_ltdb_fetch(struct ctdb_context *ctdb, 
+int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db, 
 		    TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA *data)
 {
 	TDB_DATA rec;
+	struct ctdb_context *ctdb = ctdb_db->ctdb;
 
-	rec = tdb_fetch(ctdb->ltdb->tdb, key);
+	rec = tdb_fetch(ctdb_db->ltdb->tdb, key);
 	if (rec.dsize < sizeof(*header)) {
 		/* return an initial header */
 		free(rec.dptr);
-		ltdb_initial_header(ctdb, key, header);
+		ltdb_initial_header(ctdb_db, key, header);
 		data->dptr = NULL;
 		data->dsize = 0;
 		return 0;
@@ -91,7 +132,7 @@ int ctdb_ltdb_fetch(struct ctdb_context *ctdb,
 	*header = *(struct ctdb_ltdb_header *)rec.dptr;
 
 	data->dsize = rec.dsize - sizeof(struct ctdb_ltdb_header);
-	data->dptr = talloc_memdup(ctdb, sizeof(struct ctdb_ltdb_header)+rec.dptr,
+	data->dptr = talloc_memdup(ctdb_db, sizeof(struct ctdb_ltdb_header)+rec.dptr,
 				   data->dsize);
 	free(rec.dptr);
 	CTDB_NO_MEMORY(ctdb, data->dptr);
@@ -105,9 +146,10 @@ int ctdb_ltdb_fetch(struct ctdb_context *ctdb,
   and returning the body of the record. A valid (initial) header is
   returned if the record is not present
 */
-int ctdb_ltdb_store(struct ctdb_context *ctdb, TDB_DATA key, 
+int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key, 
 		    struct ctdb_ltdb_header *header, TDB_DATA data)
 {
+	struct ctdb_context *ctdb = ctdb_db->ctdb;
 	TDB_DATA rec;
 	int ret;
 
@@ -118,7 +160,7 @@ int ctdb_ltdb_store(struct ctdb_context *ctdb, TDB_DATA key,
 	memcpy(rec.dptr, header, sizeof(*header));
 	memcpy(rec.dptr + sizeof(*header), data.dptr, data.dsize);
 
-	ret = tdb_store(ctdb->ltdb->tdb, key, rec, TDB_REPLACE);
+	ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
 	talloc_free(rec.dptr);
 
 	return ret;
diff --git a/source4/cluster/ctdb/ctdb_cluster.c b/source4/cluster/ctdb/ctdb_cluster.c
index 917a56d2b8..4892f3e173 100644
--- a/source4/cluster/ctdb/ctdb_cluster.c
+++ b/source4/cluster/ctdb/ctdb_cluster.c
@@ -193,7 +193,8 @@ void cluster_ctdb_init(struct event_context *ev)
 	const char *address;
 	const char *transport;
 	struct cluster_state *state;
-	int ret, lacount;
+	int ret, lacount, i;
+	const char *db_list[] = { "brlock", "opendb" };
 
 	nlist = lp_parm_string(-1, "ctdb", "nlist");
 	if (nlist == NULL) return;
@@ -255,10 +256,12 @@ void cluster_ctdb_init(struct event_context *ev)
 		goto failed;
         }
 
-	ret = ctdb_attach(state->ctdb, "cluster.tdb", TDB_DEFAULT, O_RDWR|O_CREAT|O_TRUNC, 0666);
-	if (ret == -1) {
-		DEBUG(0,("ctdb_attach failed - %s\n", ctdb_errstr(state->ctdb)));
-		goto failed;
+	/* attach all the databases we will need */
+	for (i=0;i<ARRAY_SIZE(db_list);i++) {
+		struct ctdb_db_context *ctdb_db;
+		ctdb_db = ctdb_attach(state->ctdb, db_list[i], TDB_DEFAULT, 
+				      O_RDWR|O_CREAT|O_TRUNC, 0666);
+		if (ctdb_db == NULL) goto failed;
 	}
 
 	/* start the protocol running */
diff --git a/source4/cluster/ctdb/include/ctdb.h b/source4/cluster/ctdb/include/ctdb.h
index efb12c5daa..c5b6dbc782 100644
--- a/source4/cluster/ctdb/include/ctdb.h
+++ b/source4/cluster/ctdb/include/ctdb.h
@@ -89,6 +89,17 @@ int ctdb_set_nlist(struct ctdb_context *ctdb, const char *nlist);
 int ctdb_start(struct ctdb_context *ctdb);
 
 /*
+  attach to a ctdb database
+*/
+struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb, const char *name, int tdb_flags, 
+				    int open_flags, mode_t mode);
+
+/*
+  find an attached ctdb_db handle given a name
+ */
+struct ctdb_db_context *ctdb_db_handle(struct ctdb_context *ctdb, const char *name);
+
+/*
   error string for last ctdb error
 */
 const char *ctdb_errstr(struct ctdb_context *);
@@ -99,20 +110,15 @@ typedef int (*ctdb_fn_t)(struct ctdb_call_info *);
 /*
   setup a ctdb call function
 */
-int ctdb_set_call(struct ctdb_context *ctdb, ctdb_fn_t fn, int id);
+int ctdb_set_call(struct ctdb_db_context *ctdb_db, ctdb_fn_t fn, int id);
 
-/*
-  attach to a ctdb database
-*/
-int ctdb_attach(struct ctdb_context *ctdb, const char *name, int tdb_flags, 
-		int open_flags, mode_t mode);
 
 
 /*
   make a ctdb call. The associated ctdb call function will be called on the DMASTER
   for the given record
 */
-int ctdb_call(struct ctdb_context *ctdb, struct ctdb_call *call);
+int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call);
 
 /*
   wait for all nodes to be connected - useful for test code
diff --git a/source4/cluster/ctdb/include/ctdb_private.h b/source4/cluster/ctdb/include/ctdb_private.h
index c863781c50..1e2244ff7d 100644
--- a/source4/cluster/ctdb/include/ctdb_private.h
+++ b/source4/cluster/ctdb/include/ctdb_private.h
@@ -76,6 +76,7 @@ struct ctdb_upcalls {
 	void (*node_connected)(struct ctdb_node *);
 };
 
+
 /* main state of the ctdb daemon */
 struct ctdb_context {
 	struct event_context *ev;
@@ -87,15 +88,23 @@ struct ctdb_context {
 	unsigned flags;
 	struct idr_context *idr;
 	struct ctdb_node **nodes; /* array of nodes in the cluster - indexed by vnn */
-	struct ctdb_registered_call *calls; /* list of registered calls */
 	char *err_msg;
-	struct tdb_wrap *ltdb;
 	const struct ctdb_methods *methods; /* transport methods */
 	const struct ctdb_upcalls *upcalls; /* transport upcalls */
 	void *private; /* private to transport */
 	unsigned max_lacount;
 	ctdb_message_fn_t message_handler;
 	void *message_private;
+	struct ctdb_db_context *db_list;
+};
+
+struct ctdb_db_context {
+	struct ctdb_db_context *next, *prev;
+	struct ctdb_context *ctdb;
+	uint32_t db_id;
+	const char *db_name;
+	struct tdb_wrap *ltdb;
+	struct ctdb_registered_call *calls; /* list of registered calls */
 };
 
 #define CTDB_NO_MEMORY(ctdb, p) do { if (!(p)) { \
@@ -157,6 +166,7 @@ struct ctdb_req_header {
 
 struct ctdb_req_call {
 	struct ctdb_req_header hdr;
+	uint32_t db_id;
 	uint32_t callid;
 	uint32_t keylen;
 	uint32_t calldatalen;
@@ -184,6 +194,7 @@ struct ctdb_reply_redirect {
 
 struct ctdb_req_dmaster {
 	struct ctdb_req_header hdr;
+	uint32_t db_id;
 	uint32_t dmaster;
 	uint32_t keylen;
 	uint32_t datalen;
@@ -220,9 +231,9 @@ void ctdb_reply_error(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
 void ctdb_reply_redirect(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
 
 uint32_t ctdb_lmaster(struct ctdb_context *ctdb, const TDB_DATA *key);
-int ctdb_ltdb_fetch(struct ctdb_context *ctdb, 
+int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db, 
 		    TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA *data);
-int ctdb_ltdb_store(struct ctdb_context *ctdb, TDB_DATA key, 
+int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key, 
 		    struct ctdb_ltdb_header *header, TDB_DATA data);
 void ctdb_queue_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
 
diff --git a/source4/cluster/ctdb/tcp/tcp_connect.c b/source4/cluster/ctdb/tcp/tcp_connect.c
index e828bb7cbb..fe0fc210ba 100644
--- a/source4/cluster/ctdb/tcp/tcp_connect.c
+++ b/source4/cluster/ctdb/tcp/tcp_connect.c
@@ -46,6 +46,7 @@ static void ctdb_node_connect_write(struct event_context *ev, struct fd_event *f
 	struct ctdb_context *ctdb = node->ctdb;
 	int error = 0;
 	socklen_t len = sizeof(error);
+	int one = 1;
 
 	if (getsockopt(tnode->fd, SOL_SOCKET, SO_ERROR, &error, &len) != 0 ||
 	    error != 0) {
@@ -64,11 +65,29 @@ static void ctdb_node_connect_write(struct event_context *ev, struct fd_event *f
 	/* tell the ctdb layer we are connected */
 	node->ctdb->upcalls->node_connected(node);
 
+        setsockopt(tnode->fd,IPPROTO_TCP,TCP_NODELAY,(char *)&one,sizeof(one));
+
 	if (tnode->queue) {
 		EVENT_FD_WRITEABLE(tnode->fde);		
 	}
 }
 
+
+static int ctdb_tcp_get_address(struct ctdb_context *ctdb,
+				const char *address, struct in_addr *addr)
+{
+	if (inet_pton(AF_INET, address, addr) <= 0) {
+		struct hostent *he = gethostbyname(address);
+		if (he == NULL || he->h_length > sizeof(*addr)) {
+			ctdb_set_error(ctdb, "invalid nework address '%s'\n", 
+				       address);
+			return -1;
+		}
+		memcpy(addr, he->h_addr, he->h_length);
+	}
+	return 0;
+}
+
 /*
   called when we should try and establish a tcp connection to a node
 */
@@ -85,7 +104,9 @@ void ctdb_tcp_node_connect(struct event_context *ev, struct timed_event *te,
 
 	set_nonblocking(tnode->fd);
 
-	inet_pton(AF_INET, node->address.address, &sock_out.sin_addr);
+	if (ctdb_tcp_get_address(ctdb, node->address.address, &sock_out.sin_addr) != 0) {
+		return;
+	}
 	sock_out.sin_port = htons(node->address.port);
 	sock_out.sin_family = PF_INET;
 	
@@ -159,7 +180,9 @@ int ctdb_tcp_listen(struct ctdb_context *ctdb)
 
         sock.sin_port = htons(ctdb->address.port);
         sock.sin_family = PF_INET;
-	inet_pton(AF_INET, ctdb->address.address, &sock.sin_addr);
+	if (ctdb_tcp_get_address(ctdb, ctdb->address.address, &sock.sin_addr) != 0) {
+		return -1;
+	}
 
         ctcp->listen_fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
         if (ctcp->listen_fd == -1) {