merge from tridge

(This used to be ctdb commit 268a89db5886955e290538fdc08d3e94172e3527)

6 files changed, 76 insertions, 59 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
diff --git a/ctdb/common/ctdb_client.c b/ctdb/common/ctdb_client.c
index 94dec19d56..ddc9af4cf3 100644
--- a/ctdb/common/ctdb_client.c
+++ b/ctdb/common/ctdb_client.c
@@ -245,22 +245,14 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db,
 		ux_socket_connect(ctdb);
 	}
 
-	ret = ctdb_ltdb_lock(ctdb_db, call->key);
-	if (ret != 0) {
-		printf("failed to lock ltdb record\n");
-		return NULL;
-	}
-
 	ret = ctdb_ltdb_fetch(ctdb_db, call->key, &header, ctdb_db, &data);
 	if (ret != 0) {
-		ctdb_ltdb_unlock(ctdb_db, call->key);
 		return NULL;
 	}
 
 #if 0
 	if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
 		state = ctdb_call_local_send(ctdb_db, call, &header, &data);
-		ctdb_ltdb_unlock(ctdb_db, call->key);
 		return state;
 	}
 #endif
@@ -268,7 +260,6 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db,
 	state = talloc_zero(ctdb_db, struct ctdb_call_state);
 	if (state == NULL) {
 		printf("failed to allocate state\n");
-		ctdb_ltdb_unlock(ctdb_db, call->key);
 		return NULL;
 	}
 
@@ -278,7 +269,6 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db,
 	state->c = ctdbd_allocate_pkt(ctdb, len);
 	if (state->c == NULL) {
 		printf("failed to allocate packet\n");
-		ctdb_ltdb_unlock(ctdb_db, call->key);
 		return NULL;
 	}
 	talloc_set_name_const(state->c, "ctdbd req_call packet");
@@ -318,7 +308,6 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db,
 			ctdb_call_timeout, state);
 */
 
-	ctdb_ltdb_unlock(ctdb_db, call->key);
 	return state;
 }
 
@@ -432,7 +421,8 @@ void ctdb_connect_wait(struct ctdb_context *ctdb)
 
 static struct ctdb_fetch_lock_state *ctdb_client_fetch_lock_send(struct ctdb_db_context *ctdb_db, 
 								 TALLOC_CTX *mem_ctx, 
-								 TDB_DATA key)
+								 TDB_DATA key, 
+								 struct ctdb_ltdb_header *header)
 {
 	struct ctdb_fetch_lock_state *state;
 	struct ctdb_context *ctdb = ctdb_db->ctdb;
@@ -468,6 +458,7 @@ static struct ctdb_fetch_lock_state *ctdb_client_fetch_lock_send(struct ctdb_db_
 	req->hdr.reqid       = idr_get_new(ctdb->idr, state, 0xFFFF);
 	req->db_id           = ctdb_db->db_id;
 	req->keylen          = key.dsize;
+	req->header          = *header;
 	memcpy(&req->key[0], key.dptr, key.dsize);
 	
 	res = ctdb_client_queue_pkt(ctdb, &req->hdr);
@@ -572,7 +563,7 @@ struct ctdb_record_handle *ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALL
 	}
 
 	/* we're not the dmaster - ask the ctdb daemon to make us dmaster */
-	state = ctdb_client_fetch_lock_send(ctdb_db, mem_ctx, key);
+	state = ctdb_client_fetch_lock_send(ctdb_db, mem_ctx, key, &h->header);
 	ret = ctdb_client_fetch_lock_recv(state, mem_ctx, key, &h->header, data);
 	if (ret != 0) {
 		talloc_free(h);
diff --git a/ctdb/common/ctdb_daemon.c b/ctdb/common/ctdb_daemon.c
index 67653202af..aa1f12474a 100644
--- a/ctdb/common/ctdb_daemon.c
+++ b/ctdb/common/ctdb_daemon.c
@@ -125,7 +125,8 @@ static void daemon_request_register_message_handler(struct ctdb_client *client,
 
 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)
+							   TDB_DATA key, struct ctdb_ltdb_header *header,
+							   TDB_DATA *data)
 {
 	struct ctdb_call *call;
 	struct ctdb_record_handle *rec;
@@ -141,13 +142,12 @@ static struct ctdb_call_state *ctdb_daemon_fetch_lock_send(struct ctdb_db_contex
 	call->key = key;
 	call->flags = CTDB_IMMEDIATE_MIGRATION;
 
-
 	rec->ctdb_db = ctdb_db;
 	rec->key = key;
 	rec->key.dptr = talloc_memdup(rec, key.dptr, key.dsize);
 	rec->data = data;
 
-	state = ctdb_daemon_call_send(ctdb_db, call);
+	state = ctdb_daemon_call_send_remote(ctdb_db, call, header);
 	state->fetch_private = rec;
 
 	return state;
@@ -192,7 +192,7 @@ static void daemon_fetch_lock_complete(struct ctdb_call_state *state)
   called when the daemon gets a fetch lock request from a client
  */
 static void daemon_request_fetch_lock(struct ctdb_client *client, 
-					struct ctdb_req_fetch_lock *f)
+				      struct ctdb_req_fetch_lock *f)
 {
 	struct ctdb_call_state *state;
 	TDB_DATA key, *data;
@@ -226,7 +226,7 @@ static void daemon_request_fetch_lock(struct ctdb_client *client,
 	data->dptr  = NULL;
 	data->dsize = 0;
 
-	state = ctdb_daemon_fetch_lock_send(ctdb_db, client, key, data);
+	state = ctdb_daemon_fetch_lock_send(ctdb_db, client, key, &f->header, data);
 	talloc_steal(state, data);
 
 	fl_data = talloc(state, struct client_fetch_lock_data);
diff --git a/ctdb/common/ctdb_ltdb.c b/ctdb/common/ctdb_ltdb.c
index e2aa47e4cd..a1e45efb02 100644
--- a/ctdb/common/ctdb_ltdb.c
+++ b/ctdb/common/ctdb_ltdb.c
@@ -82,6 +82,10 @@ struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb, const char *name,
 		}
 	}
 
+	/* add the node id to the database name, so when we run on loopback
+	   we don't conflict in the local filesystem */
+	name = talloc_asprintf(ctdb_db, "%s.%u", name, ctdb_get_vnn(ctdb));
+
 	/* 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 */
diff --git a/ctdb/include/ctdb_private.h b/ctdb/include/ctdb_private.h
index 181200e485..be0fbf85ae 100644
--- a/ctdb/include/ctdb_private.h
+++ b/ctdb/include/ctdb_private.h
@@ -306,6 +306,7 @@ struct ctdb_reply_connect_wait {
 
 struct ctdb_req_fetch_lock {
 	struct ctdb_req_header hdr;
+	struct ctdb_ltdb_header header;
 	uint32_t db_id;
 	uint32_t keylen;
 	uint8_t key[1]; /* key[] */
@@ -439,4 +440,8 @@ struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctdb_db,
 
 int ctdb_daemon_call_recv(struct ctdb_call_state *state, 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);
+
 #endif
diff --git a/ctdb/tests/ctdb_fetch.c b/ctdb/tests/ctdb_fetch.c
index 35ab989299..1ae8e8aafd 100644
--- a/ctdb/tests/ctdb_fetch.c
+++ b/ctdb/tests/ctdb_fetch.c
@@ -94,8 +94,6 @@ static void bench_fetch_1node(struct ctdb_context *ctdb)
 		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 */