- removed the non-daemon mode from ctdb, in order to simplify the

  code. It may be added back later once everything is working nicely,
  or simulated using a in-process pipe instead of a unix domain socket

- rewrote the ctdb_fetch_lock() code to follow the new design

(This used to be ctdb commit 5024dd1f305fe1ecc262db2240c56f773b4f28f0)

14 files changed, 193 insertions, 279 deletions

diff --git a/ctdb/common/ctdb.c b/ctdb/common/ctdb.c
index 559b0ed9a63..0ccbecd905e 100644
--- a/ctdb/common/ctdb.c
+++ b/ctdb/common/ctdb.c
@@ -338,11 +338,3 @@ struct ctdb_context *ctdb_init(struct event_context *ev)
 	return ctdb;
 }
 
-int ctdb_start(struct ctdb_context *ctdb)
-{
-	if (ctdb->flags&CTDB_FLAG_DAEMON_MODE) {
-		return ctdbd_start(ctdb);
-	}
-
-	return ctdb->methods->start(ctdb);
-}
diff --git a/ctdb/common/ctdb_call.c b/ctdb/common/ctdb_call.c
index 224798aecb1..8cc97c8269e 100644
--- a/ctdb/common/ctdb_call.c
+++ b/ctdb/common/ctdb_call.c
@@ -451,7 +451,8 @@ 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_db, state->call.key, &state->header, data) != 0) {
+	if (!state->fetch_private &&
+	    ctdb_ltdb_store(ctdb_db, state->call.key, &state->header, data) != 0) {
 		ctdb_fatal(ctdb, "ctdb_reply_dmaster store failed\n");
 		return;
 	}
@@ -591,8 +592,8 @@ struct ctdb_call_state *ctdb_call_local_send(struct ctdb_db_context *ctdb_db,
   This constructs a ctdb_call request and queues it for processing. 
   This call never blocks.
 */
-static struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctdb_db, 
-						     struct ctdb_call *call)
+struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctdb_db, 
+					      struct ctdb_call *call)
 {
 	uint32_t len;
 	struct ctdb_call_state *state;
@@ -659,19 +660,6 @@ static struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctd
 	return state;
 }
 
-/*
-  make a remote ctdb call - async send
-
-  This constructs a ctdb_call request and queues it for processing. 
-  This call never blocks.
-*/
-struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, struct ctdb_call *call)
-{
-	if (ctdb_db->ctdb->flags & CTDB_FLAG_DAEMON_MODE) {
-		return ctdb_client_call_send(ctdb_db, call);
-	}
-	return ctdb_daemon_call_send(ctdb_db, call);
-}
 
 /*
   make a remote ctdb call - async recv - called in daemon context
@@ -679,7 +667,7 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, struct c
   This is called when the program wants to wait for a ctdb_call to complete and get the 
   results. This call will block unless the call has already completed.
 */
-static int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
+int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
 {
 	struct ctdb_record_handle *rec;
 
@@ -696,6 +684,7 @@ static int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call
 
 	/* ugly hack to manage forced migration */
 	if (rec != NULL) {
+		rec->header = state->header;
 		rec->data->dptr = talloc_steal(rec, state->call.reply_data.dptr);
 		rec->data->dsize = state->call.reply_data.dsize;
 		talloc_free(state);
@@ -717,68 +706,3 @@ static int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call
 }
 
 
-/*
-  make a remote ctdb call - async recv. 
-
-  This is called when the program wants to wait for a ctdb_call to complete and get the 
-  results. This call will block unless the call has already completed.
-*/
-int ctdb_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
-{
-	if (state->ctdb_db->ctdb->flags & CTDB_FLAG_DAEMON_MODE) {
-		return ctdb_client_call_recv(state, call);
-	}
-	return ctdb_daemon_call_recv(state, call);
-}
-
-/*
-  full ctdb_call. Equivalent to a ctdb_call_send() followed by a ctdb_call_recv()
-*/
-int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call)
-{
-	struct ctdb_call_state *state;
-
-	state = ctdb_call_send(ctdb_db, call);
-	return ctdb_call_recv(state, call);
-}
-
-
-
-int ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, 
-					   TDB_DATA key, TDB_DATA *data)
-{
-	struct ctdb_call call;
-	struct ctdb_record_handle *rec;
-	struct ctdb_call_state *state;
-	int ret;
-
-	if (ctdb_db->ctdb->flags & CTDB_FLAG_DAEMON_MODE) {
-		return ctdb_client_fetch_lock(ctdb_db, mem_ctx, key, data);
-	}
-
-	ZERO_STRUCT(call);
-	call.call_id = CTDB_FETCH_FUNC;
-	call.key = key;
-	call.flags = CTDB_IMMEDIATE_MIGRATION;
-
-	rec = talloc(mem_ctx, struct ctdb_record_handle);
-	CTDB_NO_MEMORY_NULL(ctdb_db->ctdb, rec);
-
-	rec->ctdb_db = ctdb_db;
-	rec->key = key;
-	rec->key.dptr = talloc_memdup(rec, key.dptr, key.dsize);
-	rec->data = data;
-
-	state = ctdb_call_send(ctdb_db, &call);
-	state->fetch_private = rec;
-
-	ret = ctdb_call_recv(state, &call);
-	if (ret != 0) {
-		talloc_free(rec);
-		return FETCH_LOCK_DMASTERFAILED;
-	}
-
-	return FETCH_LOCK_SUCCESS;
-}
-
-
diff --git a/ctdb/common/ctdb_client.c b/ctdb/common/ctdb_client.c
index 4f63da5e6cd..94dec19d56d 100644
--- a/ctdb/common/ctdb_client.c
+++ b/ctdb/common/ctdb_client.c
@@ -48,6 +48,20 @@ static void ctdb_reply_connect_wait(struct ctdb_context *ctdb,
 	ctdb->num_connected = r->num_connected;
 }
 
+enum fetch_lock_state { CTDB_FETCH_LOCK_WAIT, CTDB_FETCH_LOCK_DONE, CTDB_FETCH_LOCK_ERROR };
+
+/*
+  state of a in-progress ctdb call
+*/
+struct ctdb_fetch_lock_state {
+	enum fetch_lock_state state;
+	struct ctdb_db_context *ctdb_db;
+	struct ctdb_reply_fetch_lock *r;
+	struct ctdb_ltdb_header header;
+};
+
+
+
 /*
   called in the client when we receive a CTDB_REPLY_FETCH_LOCK from the daemon
 
@@ -56,26 +70,19 @@ static void ctdb_reply_connect_wait(struct ctdb_context *ctdb,
 */
 void ctdb_reply_fetch_lock(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 {
-	struct ctdb_reply_fetch_lock *c = (struct ctdb_reply_fetch_lock *)hdr;
-	struct ctdb_call_state *state;
+	struct ctdb_reply_fetch_lock *r = (struct ctdb_reply_fetch_lock *)hdr;
+	struct ctdb_fetch_lock_state *state;
 
 	state = idr_find(ctdb->idr, hdr->reqid);
 	if (state == NULL) return;
 
-	state->call.reply_data.dptr = c->data;
-	state->call.reply_data.dsize = c->datalen;
-	state->call.status = c->state;
-
-	talloc_steal(state, c);
+	state->r = talloc_steal(state, r);
 
 	/* get an extra reference here - this prevents the free in ctdb_recv_pkt()
 	   from freeing the data */
-	(void)talloc_reference(state, c);
+	(void)talloc_reference(state, r);
 
-	state->state = CTDB_CALL_DONE;
-	if (state->async.fn) {
-		state->async.fn(state);
-	}
+	state->state = CTDB_FETCH_LOCK_DONE;
 }
 
 /*
@@ -165,7 +172,7 @@ static int ux_socket_connect(struct ctdb_context *ctdb)
   This is called when the program wants to wait for a ctdb_call to complete and get the 
   results. This call will block unless the call has already completed.
 */
-int ctdb_client_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
+int ctdb_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
 {
 	struct ctdb_record_handle *rec;
 
@@ -223,8 +230,8 @@ static int ctdb_client_call_destructor(struct ctdb_call_state *state)
   This constructs a ctdb_call request and queues it for processing. 
   This call never blocks.
 */
-struct ctdb_call_state *ctdb_client_call_send(struct ctdb_db_context *ctdb_db, 
-					      struct ctdb_call *call)
+struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, 
+				       struct ctdb_call *call)
 {
 	struct ctdb_call_state *state;
 	struct ctdb_context *ctdb = ctdb_db->ctdb;
@@ -316,14 +323,25 @@ struct ctdb_call_state *ctdb_client_call_send(struct ctdb_db_context *ctdb_db,
 }
 
 
+/*
+  full ctdb_call. Equivalent to a ctdb_call_send() followed by a ctdb_call_recv()
+*/
+int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call)
+{
+	struct ctdb_call_state *state;
+
+	state = ctdb_call_send(ctdb_db, call);
+	return ctdb_call_recv(state, call);
+}
+
 
 /*
   tell the daemon what messaging srvid we will use, and register the message
   handler function in the client
 */
-int ctdb_client_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid, 
-				    ctdb_message_fn_t handler,
-				    void *private_data)
+int ctdb_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid, 
+			     ctdb_message_fn_t handler,
+			     void *private_data)
 				    
 {
 	struct ctdb_req_register c;
@@ -352,26 +370,10 @@ int ctdb_client_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid,
 }
 
 
-
-/*
-  setup handler for receipt of ctdb messages from ctdb_send_message()
-*/
-int ctdb_set_message_handler(struct ctdb_context *ctdb, 
-			     uint32_t srvid, 
-			     ctdb_message_fn_t handler,
-			     void *private_data)
-{
-	if (ctdb->flags & CTDB_FLAG_DAEMON_MODE) {
-		return ctdb_client_set_message_handler(ctdb, srvid, handler, private_data);
-	}
-	return ctdb_daemon_set_message_handler(ctdb, srvid, handler, private_data);
-}
-
-
 /*
   send a message - from client context
  */
-int ctdb_client_send_message(struct ctdb_context *ctdb, uint32_t vnn,
+int ctdb_send_message(struct ctdb_context *ctdb, uint32_t vnn,
 		      uint32_t srvid, TDB_DATA data)
 {
 	struct ctdb_req_message *r;
@@ -405,7 +407,7 @@ int ctdb_client_send_message(struct ctdb_context *ctdb, uint32_t vnn,
 /*
   wait for all nodes to be connected - from client
  */
-static void ctdb_client_connect_wait(struct ctdb_context *ctdb)
+void ctdb_connect_wait(struct ctdb_context *ctdb)
 {
 	struct ctdb_req_connect_wait r;
 	int res;
@@ -428,25 +430,11 @@ static void ctdb_client_connect_wait(struct ctdb_context *ctdb)
 	ctdb_daemon_connect_wait(ctdb);
 }
 
-/*
-  wait for all nodes to be connected
-*/
-void ctdb_connect_wait(struct ctdb_context *ctdb)
-{
-	if (!(ctdb->flags & CTDB_FLAG_DAEMON_MODE)) {
-		ctdb_daemon_connect_wait(ctdb);
-		return;
-	}
-
-	ctdb_client_connect_wait(ctdb);
-}
-
-
-struct ctdb_call_state *ctdb_client_fetch_lock_send(struct ctdb_db_context *ctdb_db, 
-						  TALLOC_CTX *mem_ctx, 
-						  TDB_DATA key)
+static struct ctdb_fetch_lock_state *ctdb_client_fetch_lock_send(struct ctdb_db_context *ctdb_db, 
+								 TALLOC_CTX *mem_ctx, 
+								 TDB_DATA key)
 {
-	struct ctdb_call_state *state;
+	struct ctdb_fetch_lock_state *state;
 	struct ctdb_context *ctdb = ctdb_db->ctdb;
 	struct ctdb_req_fetch_lock *req;
 	int len, res;
@@ -456,24 +444,23 @@ struct ctdb_call_state *ctdb_client_fetch_lock_send(struct ctdb_db_context *ctdb
 		ux_socket_connect(ctdb);
 	}
 
-	state = talloc_zero(ctdb_db, struct ctdb_call_state);
+	state = talloc_zero(ctdb_db, struct ctdb_fetch_lock_state);
 	if (state == NULL) {
 		printf("failed to allocate state\n");
 		return NULL;
 	}
-	state->state   = CTDB_CALL_WAIT;
+	state->state   = CTDB_FETCH_LOCK_WAIT;
 	state->ctdb_db = ctdb_db;
 	len = offsetof(struct ctdb_req_fetch_lock, key) + key.dsize;
-	state->c = ctdbd_allocate_pkt(ctdb, len);
-	if (state->c == NULL) {
+	req = ctdbd_allocate_pkt(ctdb, len);
+	if (req == NULL) {
 		printf("failed to allocate packet\n");
 		return NULL;
 	}
-	bzero(state->c, len);
-	talloc_set_name_const(state->c, "ctdbd req_fetch_lock packet");
-	talloc_steal(state, state->c);
+	ZERO_STRUCT(*req);
+	talloc_set_name_const(req, "ctdbd req_fetch_lock packet");
+	talloc_steal(state, req);
 
-	req = (struct ctdb_req_fetch_lock *)state->c;
 	req->hdr.length      = len;
 	req->hdr.ctdb_magic  = CTDB_MAGIC;
 	req->hdr.ctdb_version = CTDB_VERSION;
@@ -488,8 +475,6 @@ struct ctdb_call_state *ctdb_client_fetch_lock_send(struct ctdb_db_context *ctdb
 		return NULL;
 	}
 
-	talloc_free(req);
-
 	return state;
 }
 
@@ -500,77 +485,119 @@ struct ctdb_call_state *ctdb_client_fetch_lock_send(struct ctdb_db_context *ctdb
   This is called when the program wants to wait for a ctdb_fetch_lock to complete and get the 
   results. This call will block unless the call has already completed.
 */
-int ctdb_client_fetch_lock_recv(struct ctdb_call_state *state, TALLOC_CTX *mem_ctx, TDB_DATA key, TDB_DATA *data)
+int ctdb_client_fetch_lock_recv(struct ctdb_fetch_lock_state *state, TALLOC_CTX *mem_ctx, 
+				TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA *data)
 {
-	while (state->state < CTDB_CALL_DONE) {
+	while (state->state < CTDB_FETCH_LOCK_DONE) {
 		event_loop_once(state->ctdb_db->ctdb->ev);
 	}
-	if (state->state != CTDB_CALL_DONE) {
-		ctdb_set_error(state->node->ctdb, "%s", state->errmsg);
+	if (state->state != CTDB_FETCH_LOCK_DONE) {
 		talloc_free(state);
 		return -1;
 	}
 
-	data->dsize = state->call.reply_data.dsize;
-	data->dptr  = talloc_memdup(mem_ctx, state->call.reply_data.dptr, data->dsize);
+	*header = state->r->header;
+	data->dsize = state->r->datalen;
+	data->dptr  = talloc_memdup(mem_ctx, state->r->data, data->dsize);
+
+	talloc_free(state);
 
 	return 0;
 }
 
-int ctdb_client_fetch_lock(struct ctdb_db_context *ctdb_db, 
-			   TALLOC_CTX *mem_ctx, 
-			   TDB_DATA key,
-			   TDB_DATA *data)
+/*
+  cancel a ctdb_fetch_lock operation, releasing the lock
+ */
+static int fetch_lock_destructor(struct ctdb_record_handle *h)
+{
+	ctdb_ltdb_unlock(h->ctdb_db, h->key);
+	return 0;
+}
+
+/*
+  get a lock on a record, and return the records data. Blocks until it gets the lock
+ */
+struct ctdb_record_handle *ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, 
+					   TDB_DATA key, TDB_DATA *data)
 {
-	struct ctdb_ltdb_header header;
 	int ret;
+	struct ctdb_record_handle *h;
+	struct ctdb_fetch_lock_state *state;
+
+	/*
+	  procedure is as follows:
+
+	  1) get the chain lock. 
+	  2) check if we are dmaster
+	  3) if we are the dmaster then return handle 
+	  4) if not dmaster then ask ctdb daemon to make us dmaster, and wait for
+	     reply from ctdbd
+	  5) when we get the reply, we are now dmaster, update vnn in header
+	  6) return handle
+	 */
+
+	h = talloc_zero(mem_ctx, struct ctdb_record_handle);
+	if (h == NULL) {
+		return NULL;
+	}
 
+	h->ctdb_db = ctdb_db;
+	h->key     = key;
+	h->key.dptr = talloc_memdup(h, key.dptr, key.dsize);
+	if (h->key.dptr == NULL) {
+		talloc_free(h);
+		return NULL;
+	}
+	h->data    = data;
+
+	/* step 1 - get the chain lock */
 	ret = ctdb_ltdb_lock(ctdb_db, key);
 	if (ret != 0) {
 		printf("failed to lock ltdb record\n");
-		return FETCH_LOCK_LOCKFAILED;
+		talloc_free(h);
+		return NULL;
 	}
 
-	ret = ctdb_ltdb_fetch(ctdb_db, key, &header, ctdb_db, data);
+	talloc_set_destructor(h, fetch_lock_destructor);
+
+	ret = ctdb_ltdb_fetch(ctdb_db, key, &h->header, ctdb_db, data);
 	if (ret != 0) {
-		ctdb_ltdb_unlock(ctdb_db, key);
-		return FETCH_LOCK_FETCHFAILED;
+		talloc_free(h);
+		return NULL;
 	}
 
+	/* step 2 - check if we are the dmaster */
+	if (h->header.dmaster == ctdb_db->ctdb->vnn) {
+		return h;
+	}
 
-	if (header.dmaster != ctdb_db->ctdb->vnn) {
-		struct ctdb_call_state *state;
+	/* we're not the dmaster - ask the ctdb daemon to make us dmaster */
+	state = ctdb_client_fetch_lock_send(ctdb_db, mem_ctx, key);
+	ret = ctdb_client_fetch_lock_recv(state, mem_ctx, key, &h->header, data);
+	if (ret != 0) {
+		talloc_free(h);
+		return NULL;
+	}
 
-		state = ctdb_client_fetch_lock_send(ctdb_db, mem_ctx, key);
-		ret = ctdb_client_fetch_lock_recv(state, mem_ctx, key, data);
-		if (ret != 0) {
-			ctdb_ltdb_unlock(ctdb_db, key);
-			return FETCH_LOCK_DMASTERFAILED;
-		}
+	/* the record is now local, and locked. update the record on disk
+	   to mark us as the dmaster*/
+	h->header.dmaster = ctdb_db->ctdb->vnn;
+	ret = ctdb_ltdb_store(ctdb_db, key, &h->header, *data);
+	if (ret != 0) {
+		printf("bugger - we're in real trouble now! can't update record to mark us as dmasterx\n");
+		talloc_free(h);
+		return NULL;
 	}
 
-	return 0;
+	/* give the caller a handle to be used for ctdb_record_store() or a cancel via
+	   a talloc_free() */
+	return h;
 }
 
 /*
-   a helper function for the client that will store the new data for the
-   record and release the tdb chainlock
+  store some data to the record that was locked with ctdb_fetch_lock()
 */
-int ctdb_client_store_unlock(struct ctdb_db_context *ctdb_db, TDB_DATA key, TDB_DATA data)
+int ctdb_record_store(struct ctdb_record_handle *h, TDB_DATA data)
 {
-	int ret;
-	struct ctdb_ltdb_header header;
-
-	/* should be avoided if possible    hang header off rec ? */
-	ret = ctdb_ltdb_fetch(ctdb_db, key, &header, NULL, NULL);
-	if (ret) {
-		ctdb_set_error(ctdb_db->ctdb, "Fetch of locally held record failed");
-		return ret;
-	}
-
-	ret = ctdb_ltdb_store(ctdb_db, key, &header, data);
-
-	ctdb_ltdb_unlock(ctdb_db, key);
-	
-	return ret;
+	return ctdb_ltdb_store(h->ctdb_db, h->key, &h->header, data);
 }
diff --git a/ctdb/common/ctdb_daemon.c b/ctdb/common/ctdb_daemon.c
index 1813b8083b5..5a90003bc0d 100644
--- a/ctdb/common/ctdb_daemon.c
+++ b/ctdb/common/ctdb_daemon.c
@@ -110,9 +110,9 @@ static void daemon_request_register_message_handler(struct ctdb_client *client,
 }
 
 
-static struct ctdb_call_state *ctdb_fetch_lock_send(struct ctdb_db_context *ctdb_db, 
-					 	    TALLOC_CTX *mem_ctx, 
-						    TDB_DATA key, TDB_DATA *data)
+static struct ctdb_call_state *ctdb_daemon_fetch_lock_send(struct ctdb_db_context *ctdb_db, 
+							   TALLOC_CTX *mem_ctx, 
+							   TDB_DATA key, TDB_DATA *data)
 {
 	struct ctdb_call *call;
 	struct ctdb_record_handle *rec;
@@ -134,7 +134,7 @@ static struct ctdb_call_state *ctdb_fetch_lock_send(struct ctdb_db_context *ctdb
 	rec->key.dptr = talloc_memdup(rec, key.dptr, key.dsize);
 	rec->data = data;
 
-	state = ctdb_call_send(ctdb_db, call);
+	state = ctdb_daemon_call_send(ctdb_db, call);
 	state->fetch_private = rec;
 
 	return state;
@@ -144,6 +144,7 @@ struct client_fetch_lock_data {
 	struct ctdb_client *client;
 	uint32_t reqid;
 };
+
 static void daemon_fetch_lock_complete(struct ctdb_call_state *state)
 {
 	struct ctdb_reply_fetch_lock *r;
@@ -212,7 +213,7 @@ static void daemon_request_fetch_lock(struct ctdb_client *client,
 	data->dptr  = NULL;
 	data->dsize = 0;
 
-	state = ctdb_fetch_lock_send(ctdb_db, client, key, data);
+	state = ctdb_daemon_fetch_lock_send(ctdb_db, client, key, data);
 	talloc_steal(state, data);
 
 	fl_data = talloc(state, struct client_fetch_lock_data);
@@ -316,10 +317,12 @@ static void daemon_request_call_from_client(struct ctdb_client *client,
 	call.call_data.dptr = c->data + c->keylen;
 	call.call_data.dsize = c->calldatalen;
 
-	state = ctdb_call_send(ctdb_db, &call);
+	state = ctdb_daemon_call_send(ctdb_db, &call);
+//	state->async.fn = daemon_call_from_client_callback;
+//	state->async.private_data = state;
 
 /* XXX this must be converted to fully async */
-	res = ctdb_call_recv(state, &call);
+	res = ctdb_daemon_call_recv(state, &call);
 	if (res != 0) {
 		printf("ctdbd_call_recv() returned error\n");
 		exit(1);
@@ -340,7 +343,7 @@ static void daemon_request_call_from_client(struct ctdb_client *client,
 	r->datalen          = call.reply_data.dsize;
 	memcpy(&r->data[0], call.reply_data.dptr, r->datalen);
 
-	res = ctdb_queue_send(client->queue, (uint8_t *)&r, r->hdr.length);
+	res = ctdb_queue_send(client->queue, (uint8_t *)&r->hdr, r->hdr.length);
 	if (res != 0) {
 		printf("Failed to queue packet from daemon to client\n");
 	}
@@ -349,7 +352,7 @@ static void daemon_request_call_from_client(struct ctdb_client *client,
 
 
 /* data contains a packet from the client */
-static void client_incoming_packet(struct ctdb_client *client, void *data, size_t nread)
+static void daemon_incoming_packet(struct ctdb_client *client, void *data, size_t nread)
 {
 	struct ctdb_req_header *hdr = data;
 
@@ -391,7 +394,7 @@ done:
 }
 
 
-static void ctdb_client_read_cb(uint8_t *data, size_t cnt, void *args)
+static void ctdb_daemon_read_cb(uint8_t *data, size_t cnt, void *args)
 {
 	struct ctdb_client *client = talloc_get_type(args, struct ctdb_client);
 	struct ctdb_req_header *hdr;
@@ -423,7 +426,7 @@ static void ctdb_client_read_cb(uint8_t *data, size_t cnt, void *args)
 	}
 
 	/* it is the responsibility of the incoming packet function to free 'data' */
-	client_incoming_packet(client, data, cnt);
+	daemon_incoming_packet(client, data, cnt);
 }
 
 static void ctdb_accept_client(struct event_context *ev, struct fd_event *fde, 
@@ -448,7 +451,7 @@ static void ctdb_accept_client(struct event_context *ev, struct fd_event *fde,
 	client->fd = fd;
 
 	client->queue = ctdb_queue_setup(ctdb, client, fd, CTDB_DS_ALIGNMENT, 
-					 ctdb_client_read_cb, client);
+					 ctdb_daemon_read_cb, client);
 
 	talloc_set_destructor(client, ctdb_client_destructor);
 }
@@ -517,7 +520,7 @@ static int unlink_destructor(const char *name)
 /*
   start the protocol going
 */
-int ctdbd_start(struct ctdb_context *ctdb)
+int ctdb_start(struct ctdb_context *ctdb)
 {
 	pid_t pid;
 	static int fd[2];
@@ -560,7 +563,7 @@ int ctdbd_start(struct ctdb_context *ctdb)
 	talloc_set_destructor(domain_socket_name, unlink_destructor);	
 	
 	close(fd[1]);
-	ctdb_clear_flags(ctdb, CTDB_FLAG_DAEMON_MODE);
+
 	ctdb->ev = event_context_init(NULL);
 	fde = event_add_fd(ctdb->ev, ctdb, fd[0], EVENT_FD_READ, ctdb_read_from_parent, &fd[0]);
 	fde = event_add_fd(ctdb->ev, ctdb, ctdb->daemon.sd, EVENT_FD_READ, ctdb_accept_client, ctdb);
@@ -580,10 +583,3 @@ void *ctdbd_allocate_pkt(struct ctdb_context *ctdb, size_t len)
 	return talloc_size(ctdb, size);
 }
 
-int ctdb_daemon_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid, 
-			     ctdb_message_fn_t handler,
-			     void *private_data)
-{
-	return ctdb_register_message_handler(ctdb, ctdb, srvid, handler, private_data);
-}
-
diff --git a/ctdb/common/ctdb_message.c b/ctdb/common/ctdb_message.c
index ad88ec22d26..88fe04df503 100644
--- a/ctdb/common/ctdb_message.c
+++ b/ctdb/common/ctdb_message.c
@@ -147,18 +147,6 @@ int ctdb_daemon_send_message(struct ctdb_context *ctdb, uint32_t vnn,
 	return 0;
 }
 
-/*
-  send a ctdb message
-*/
-int ctdb_send_message(struct ctdb_context *ctdb, uint32_t vnn,
-		      uint32_t srvid, TDB_DATA data)
-{
-	if (ctdb->flags & CTDB_FLAG_DAEMON_MODE) {
-		return ctdb_client_send_message(ctdb, vnn, srvid, data);
-	}
-	return ctdb_daemon_send_message(ctdb, vnn, srvid, data);
-}
-
 
 /*
   when a client goes away, we need to remove its srvid handler from the list
diff --git a/ctdb/direct/ctdbd.c b/ctdb/direct/ctdbd.c
index 700416e5e95..9e13c4c825a 100644
--- a/ctdb/direct/ctdbd.c
+++ b/ctdb/direct/ctdbd.c
@@ -105,9 +105,6 @@ int main(int argc, const char *argv[])
 	if (self_connect) {
 		ctdb_set_flags(ctdb, CTDB_FLAG_SELF_CONNECT);
 	}
-	if (daemon_mode) {
-		ctdb_set_flags(ctdb, CTDB_FLAG_DAEMON_MODE);
-	}
 
 	ret = ctdb_set_transport(ctdb, transport);
 	if (ret == -1) {
diff --git a/ctdb/include/ctdb.h b/ctdb/include/ctdb.h
index a41e5152075..9203cdc0746 100644
--- a/ctdb/include/ctdb.h
+++ b/ctdb/include/ctdb.h
@@ -50,8 +50,6 @@ struct ctdb_call_info {
   ctdb flags
 */
 #define CTDB_FLAG_SELF_CONNECT (1<<0)
-/* fork off a separate ctdb daemon */
-#define CTDB_FLAG_DAEMON_MODE  (1<<1)
 /* for test code only: make ctdb_start() block until all nodes are connected */
 #define CTDB_FLAG_CONNECT_WAIT (1<<2)
 
@@ -177,7 +175,9 @@ int ctdb_send_message(struct ctdb_context *ctdb, uint32_t vnn,
    dmaster for the record to be moved to the local node. 
 
 */
-int ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, TDB_DATA key, TDB_DATA *data);
+struct ctdb_record_handle *ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, 
+					   TDB_DATA key, TDB_DATA *data);
+
 
 /*
   do a fetch lock from a client to the local daemon
@@ -192,8 +192,7 @@ int ctdb_client_fetch_lock(struct ctdb_db_context *ctdb_db,
 						  TDB_DATA key, TDB_DATA *data);
 
 
-int ctdb_client_store_unlock(struct ctdb_db_context *ctdb_db, TDB_DATA key, TDB_DATA data);
-
+int ctdb_record_store(struct ctdb_record_handle *h, TDB_DATA data);
 
 int ctdb_register_message_handler(struct ctdb_context *ctdb, 
 				  TALLOC_CTX *mem_ctx,
diff --git a/ctdb/include/ctdb_private.h b/ctdb/include/ctdb_private.h
index a11ebf5ad4c..181200e485c 100644
--- a/ctdb/include/ctdb_private.h
+++ b/ctdb/include/ctdb_private.h
@@ -199,9 +199,9 @@ struct ctdb_record_handle {
 	struct ctdb_db_context *ctdb_db;
 	TDB_DATA key;
 	TDB_DATA *data;
+	struct ctdb_ltdb_header header;
 };
 
-
 /*
   operation IDs
 */
@@ -314,6 +314,7 @@ struct ctdb_req_fetch_lock {
 struct ctdb_reply_fetch_lock {
 	struct ctdb_req_header hdr;
 	uint32_t state;
+	struct ctdb_ltdb_header header;
 	uint32_t datalen;
 	uint8_t data[1]; /* data[] */
 };
@@ -433,4 +434,9 @@ struct lockwait_handle *ctdb_lockwait(struct ctdb_db_context *ctdb_db,
 				      TDB_DATA key,
 				      void (*callback)(void *), void *private_data);
 
+struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctdb_db, 
+					      struct ctdb_call *call);
+
+int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call *call);
+
 #endif
diff --git a/ctdb/tests/cmdline.c b/ctdb/tests/cmdline.c
index 57829b9e9bd..94d488a084a 100644
--- a/ctdb/tests/cmdline.c
+++ b/ctdb/tests/cmdline.c
@@ -31,13 +31,11 @@ static struct {
 	const char *transport;
 	const char *myaddress;
 	int self_connect;
-	int daemon_mode;
 } ctdb_cmdline = {
 	.nlist = NULL,
 	.transport = "tcp",
 	.myaddress = NULL,
 	.self_connect = 0,
-	.daemon_mode = 0
 };
 
 
@@ -46,7 +44,6 @@ struct poptOption popt_ctdb_cmdline[] = {
 	{ "listen", 0, POPT_ARG_STRING, &ctdb_cmdline.myaddress, 0, "address to listen on", "address" },
 	{ "transport", 0, POPT_ARG_STRING, &ctdb_cmdline.transport, 0, "protocol transport", NULL },
 	{ "self-connect", 0, POPT_ARG_NONE, &ctdb_cmdline.self_connect, 0, "enable self connect", "boolean" },
-	{ "daemon", 0, POPT_ARG_NONE, &ctdb_cmdline.daemon_mode, 0, "spawn a ctdb daemon", "boolean" },
 	{ NULL }
 };
 
@@ -74,9 +71,6 @@ struct ctdb_context *ctdb_cmdline_init(struct event_context *ev)
 	if (ctdb_cmdline.self_connect) {
 		ctdb_set_flags(ctdb, CTDB_FLAG_SELF_CONNECT);
 	}
-	if (ctdb_cmdline.daemon_mode) {
-		ctdb_set_flags(ctdb, CTDB_FLAG_DAEMON_MODE);
-	}
 
 	ret = ctdb_set_transport(ctdb, ctdb_cmdline.transport);
 	if (ret == -1) {
diff --git a/ctdb/tests/ctdb_fetch.c b/ctdb/tests/ctdb_fetch.c
index ebb7292c73b..a292dfb4432 100644
--- a/ctdb/tests/ctdb_fetch.c
+++ b/ctdb/tests/ctdb_fetch.c
@@ -61,14 +61,15 @@ static void bench_fetch_1node(struct ctdb_context *ctdb)
 	struct ctdb_db_context *ctdb_db;
 	TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
 	int dest, ret;
+	struct ctdb_record_handle *h;
 
 	key.dptr = discard_const("testkey");
 	key.dsize = strlen((const char *)key.dptr);
 
 	ctdb_db = ctdb_db_handle(ctdb, "test.tdb");
 
-	ret = ctdb_client_fetch_lock(ctdb_db, tmp_ctx, key, &data);
-	if (ret != 0) {
+	h = ctdb_fetch_lock(ctdb_db, tmp_ctx, key, &data);
+	if (h == NULL) {
 		printf("Failed to fetch record '%s' on node %d\n", 
 		       (const char *)key.dptr, ctdb_get_vnn(ctdb));
 		talloc_free(tmp_ctx);
@@ -87,11 +88,13 @@ static void bench_fetch_1node(struct ctdb_context *ctdb)
 						      msg_count, ctdb_get_vnn(ctdb));
 	data.dsize = strlen((const char *)data.dptr)+1;
 
-	ret = ctdb_client_store_unlock(ctdb_db, key, data);
+	ret = ctdb_record_store(h, data);
 	if (ret != 0) {
 		printf("Failed to store record\n");
 	}
 
+	printf("DATA IS NOW:%s\n", (const char *)data.dptr);
+
 	talloc_free(tmp_ctx);
 
 	/* tell the next node to do the same */
diff --git a/ctdb/tests/ctdb_fetch1.c b/ctdb/tests/ctdb_fetch1.c
index afae8c53cf1..635311c9980 100644
--- a/ctdb/tests/ctdb_fetch1.c
+++ b/ctdb/tests/ctdb_fetch1.c
@@ -47,6 +47,7 @@ void test1(struct ctdb_db_context *ctdb_db)
 {
 	TDB_DATA key, data, data2, store_data;
 	int ret;
+	struct ctdb_record_handle *h;
  
 	/* 
 	   test 1 : write data and read it back.   should all be the same
@@ -54,23 +55,24 @@ void test1(struct ctdb_db_context *ctdb_db)
 	printf("Test1: write and verify we can read it back: ");
 	key.dptr  = discard_const("Record");
 	key.dsize = strlen((const char *)key.dptr)+1;
-	ret = ctdb_client_fetch_lock(ctdb_db, ctdb_db, key, &data);
-	if (ret!=0) {
-		printf("test1: ctdb_client_fetch_lock() failed\n");
+	h = ctdb_fetch_lock(ctdb_db, ctdb_db, key, &data);
+	if (h == NULL) {
+		printf("test1: ctdb_fetch_lock() failed\n");
 		exit(1);
 	}
 
 	store_data.dptr  = discard_const("data to store");
 	store_data.dsize = strlen((const char *)store_data.dptr)+1;
-	ret = ctdb_client_store_unlock(ctdb_db, key, store_data);
+	ret = ctdb_record_store(h, store_data);
+	talloc_free(h);
 	if (ret!=0) {
-		printf("test1: ctdb_client_store_unlock() failed\n");
+		printf("test1: ctdb_record_store() failed\n");
 		exit(1);
 	}
 
-	ret = ctdb_client_fetch_lock(ctdb_db, ctdb_db, key, &data2);
-	if (ret!=0) {
-		printf("test1: ctdb_client_fetch_lock() failed\n");
+	h = ctdb_fetch_lock(ctdb_db, ctdb_db, key, &data2);
+	if (h == NULL) {
+		printf("test1: ctdb_fetch_lock() failed\n");
 		exit(1);
 	}
 
@@ -83,9 +85,10 @@ void test1(struct ctdb_db_context *ctdb_db)
 	}
 	
 	/* just write it back to unlock it */
-	ret = ctdb_client_store_unlock(ctdb_db, key, store_data);
+	ret = ctdb_record_store(h, store_data);
+	talloc_free(h);
 	if (ret!=0) {
-		printf("test1: ctdb_client_store_unlock() failed\n");
+		printf("test1: ctdb_record_store() failed\n");
 		exit(1);
 	}
 }
@@ -94,7 +97,7 @@ void child(int srvid, struct event_context *ev, struct ctdb_context *ctdb, struc
 {
 	TDB_DATA data;
 	TDB_DATA key, data2;
-	int ret;
+	struct ctdb_record_handle *h;
 
 	data.dptr=discard_const("dummy message");
 	data.dsize=strlen((const char *)data.dptr)+1;
@@ -110,13 +113,13 @@ void child(int srvid, struct event_context *ev, struct ctdb_context *ctdb, struc
 	/* fetch and lock the record */
 	key.dptr  = discard_const("Record");
 	key.dsize = strlen((const char *)key.dptr)+1;
-	ret = ctdb_client_fetch_lock(ctdb_db, ctdb_db, key, &data2);
-	if (ret!=0) {
-		printf("client: ctdb_client_fetch_lock() failed\n");
+	h = ctdb_fetch_lock(ctdb_db, ctdb_db, key, &data2);
+	if (h == NULL) {
+		printf("client: ctdb_fetch_lock() failed\n");
 		exit(1);
 	}
 	ctdb_send_message(ctdb, ctdb_get_vnn(ctdb), PARENT_SRVID, data);
-
+	talloc_free(h);
 
 	while (1) {
 		event_loop_once(ev);
diff --git a/ctdb/tests/fetch1.sh b/ctdb/tests/fetch1.sh
index 4331a57d3e0..3562ca867e6 100755
--- a/ctdb/tests/fetch1.sh
+++ b/ctdb/tests/fetch1.sh
@@ -3,6 +3,6 @@
 killall -q ctdb_fetch1
 
 echo "Trying node"
-bin/ctdb_fetch1 --nlist tests/1node.txt --listen 127.0.0.1:9001 --daemon $* 
+bin/ctdb_fetch1 --nlist tests/1node.txt --listen 127.0.0.1:9001 $* 
 
 killall -q ctdb_fetch1
diff --git a/ctdb/tests/test.sh b/ctdb/tests/test.sh
index 6197ce9616d..de7d8a342fb 100755
--- a/ctdb/tests/test.sh
+++ b/ctdb/tests/test.sh
@@ -19,17 +19,4 @@ sleep 3
 
 killall ctdb_test
 
-echo "Trying 2 nodes in daemon mode"
-bin/ctdb_test --nlist tests/nodes.txt --listen 127.0.0.1:9001 --daemon &
-bin/ctdb_test --nlist tests/nodes.txt --listen 127.0.0.2:9001 --daemon &
 
-sleep 3
-killall ctdb_test
-
-echo "Trying 4 nodes in daemon mode"
-bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.1:9001 --daemon &
-bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.2:9001 --daemon &
-bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.3:9001 --daemon &
-bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.4:9001 --daemon &
-sleep 3
-killall ctdb_test
diff --git a/ctdb/tests/test1.sh b/ctdb/tests/test1.sh
index 64d5c9fc873..f13799b45c7 100755
--- a/ctdb/tests/test1.sh
+++ b/ctdb/tests/test1.sh
@@ -1,8 +1,6 @@
 #!/bin/sh
 
-echo "Testing local send"
-bin/ctdb_test --nlist tests/1node.txt --listen 127.0.0.1:9001 
 echo "Testing daemon mode"
-bin/ctdb_test --nlist tests/1node.txt --listen 127.0.0.1:9001 --daemon
+bin/ctdb_test --nlist tests/1node.txt --listen 127.0.0.1:9001
 echo "Testing self connect"
 bin/ctdb_test --nlist tests/1node.txt --listen 127.0.0.1:9001 --self-connect