1 files changed, 58 insertions, 39 deletions

diff --git a/ctdb/common/ctdb_call.c b/ctdb/common/ctdb_call.c
index 8cc97c8269..9e12581000 100644
--- a/ctdb/common/ctdb_call.c
+++ b/ctdb/common/ctdb_call.c
@@ -266,23 +266,27 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
 		return;
 	}
 	
-	/* fetch the current record */
-	ret = ctdb_ltdb_fetch(ctdb_db, key, &header, hdr, &data2);
-	if (ret != 0) {
-		ctdb_fatal(ctdb, "ctdb_req_dmaster failed to fetch record");
-		return;
-	}
-
-	/* its a protocol error if the sending node is not the current dmaster */
-	if (header.dmaster != hdr->srcnode) {
-		ctdb_fatal(ctdb, "dmaster request from non-master");
-		return;
-	}
+	/* if the new dmaster and the lmaster are the same node, then
+	   we don't need to update the record header now */
+	if (c->dmaster != ctdb->vnn) {
+		/* fetch the current record */
+		ret = ctdb_ltdb_fetch(ctdb_db, key, &header, hdr, &data2);
+		if (ret != 0) {
+			ctdb_fatal(ctdb, "ctdb_req_dmaster failed to fetch record");
+			return;
+		}
+		
+		/* its a protocol error if the sending node is not the current dmaster */
+		if (header.dmaster != hdr->srcnode) {
+			ctdb_fatal(ctdb, "dmaster request from non-master");
+			return;
+		}
 
-	header.dmaster = c->dmaster;
-	if (ctdb_ltdb_store(ctdb_db, key, &header, data) != 0) {
-		ctdb_fatal(ctdb, "ctdb_req_dmaster unable to update dmaster");
-		return;
+		header.dmaster = c->dmaster;
+		if (ctdb_ltdb_store(ctdb_db, key, &header, data) != 0) {
+			ctdb_fatal(ctdb, "ctdb_req_dmaster unable to update dmaster");
+			return;
+		}
 	}
 
 	/* send the CTDB_REPLY_DMASTER */
@@ -592,34 +596,17 @@ 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.
 */
-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_remote(struct ctdb_db_context *ctdb_db, 
+						     struct ctdb_call *call, 
+						     struct ctdb_ltdb_header *header)
 {
 	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
-	  send it off at all, we can bypass the network and handle it
-	  locally. To find out if we are the dmaster we need to look
-	  in our ltdb
-	*/
-	ret = ctdb_ltdb_fetch(ctdb_db, call->key, &header, ctdb_db, &data);
-	if (ret != 0) return NULL;
-
-	if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
-		return ctdb_call_local_send(ctdb_db, call, &header, &data);
-	}
-
 	state = talloc_zero(ctdb_db, struct ctdb_call_state);
 	CTDB_NO_MEMORY_NULL(ctdb, state);
 
-	talloc_steal(state, data.dptr);
-
 	len = offsetof(struct ctdb_req_call, data) + call->key.dsize + call->call_data.dsize;
 	state->c = ctdb->methods->allocate_pkt(ctdb, len);
 	CTDB_NO_MEMORY_NULL(ctdb, state->c);
@@ -630,7 +617,7 @@ struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctdb_db,
 	state->c->hdr.ctdb_magic = CTDB_MAGIC;
 	state->c->hdr.ctdb_version = CTDB_VERSION;
 	state->c->hdr.operation = CTDB_REQ_CALL;
-	state->c->hdr.destnode  = header.dmaster;
+	state->c->hdr.destnode  = header->dmaster;
 	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);
@@ -646,9 +633,9 @@ struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctdb_db,
 	state->call.call_data.dptr = &state->c->data[call->key.dsize];
 	state->call.key.dptr       = &state->c->data[0];
 
-	state->node   = ctdb->nodes[header.dmaster];
+	state->node   = ctdb->nodes[header->dmaster];
 	state->state  = CTDB_CALL_WAIT;
-	state->header = header;
+	state->header = *header;
 	state->ctdb_db = ctdb_db;
 
 	talloc_set_destructor(state, ctdb_call_destructor);
@@ -660,6 +647,38 @@ struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctdb_db,
 	return state;
 }
 
+/*
+  make a remote ctdb call - async send. Called in daemon context.
+
+  This constructs a ctdb_call request and queues it for processing. 
+  This call never blocks.
+*/
+struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctdb_db, 
+					      struct ctdb_call *call)
+{
+	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
+	  send it off at all, we can bypass the network and handle it
+	  locally. To find out if we are the dmaster we need to look
+	  in our ltdb
+	*/
+	ret = ctdb_ltdb_fetch(ctdb_db, call->key, &header, ctdb_db, &data);
+	if (ret != 0) return NULL;
+
+	if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
+		return ctdb_call_local_send(ctdb_db, call, &header, &data);
+	}
+
+	talloc_free(data.dptr);
+
+	return ctdb_daemon_call_send_remote(ctdb_db, call, &header);
+}
+
 
 /*
   make a remote ctdb call - async recv - called in daemon context