partially completed work towards full messaging system which will work in both daemon and standalone mode. Does not compile\! committing so ronnie can continue while I'm out

(This used to be ctdb commit 1b5e65a700e2bd0a5c913d7866024b25600a14c9)

6 files changed, 153 insertions, 22 deletions

diff --git a/ctdb/common/ctdb_call.c b/ctdb/common/ctdb_call.c
index f86fac05869..f0fef644560 100644
--- a/ctdb/common/ctdb_call.c
+++ b/ctdb/common/ctdb_call.c
@@ -630,9 +630,8 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, struct c
 {
 	if (ctdb_db->ctdb->flags & CTDB_FLAG_DAEMON_MODE) {
 		return ctdb_client_call_send(ctdb_db, call);
-	} else {
-		return ctdb_daemon_call_send(ctdb_db, call);
 	}
+	return ctdb_daemon_call_send(ctdb_db, call);
 }
 
 /*
@@ -689,9 +688,8 @@ 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);
-	} else {
-		return ctdb_daemon_call_recv(state, call);
 	}
+	return ctdb_daemon_call_recv(state, call);
 }
 
 /*
diff --git a/ctdb/common/ctdb_client.c b/ctdb/common/ctdb_client.c
index cbf21367291..e837a1af879 100644
--- a/ctdb/common/ctdb_client.c
+++ b/ctdb/common/ctdb_client.c
@@ -67,7 +67,15 @@ static void ctdb_client_read_cb(uint8_t *data, size_t cnt, void *args)
 		return;
 	}
 
-	ctdb_reply_call(ctdb, hdr);
+	switch (hdr->operation) {
+	case CTDB_REPLY_CALL:
+		ctdb_reply_call(ctdb, hdr);
+		break;
+
+	case CTDB_REQ_MESSAGE:
+		ctdb_request_message(ctdb, hdr);
+		break;
+	}
 }
 
 /*
@@ -255,3 +263,51 @@ struct ctdb_call_state *ctdb_client_call_send(struct ctdb_db_context *ctdb_db,
 	ctdb_ltdb_unlock(ctdb_db, call->key);
 	return state;
 }
+
+
+
+/*
+  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)
+				    
+{
+	struct ctdb_register_call c;
+	int res;
+
+	ZERO_STRUCT(c);
+
+	c.hdr.length       = sizeof(c);
+	c.hdr.ctdb_magic   = CTDB_MAGIC;
+	c.hdr.ctdb_version = CTDB_VERSION;
+	c.hdr.operation    = CTDB_REQ_REGISTER;
+	c.srvid            = srvid;
+
+	res = ctdb_client_queue_pkt(ctdb, &c.hdr);
+	if (res != 0) {
+		return res;
+	}
+
+	/* also need to register the handler with our ctdb structure */
+	return ctdb_register_message_handler(ctdb, ctdb, srvid, handler, private);
+}
+
+
+
+/*
+  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)
+{
+	if (ctdb->flags & CTDB_FLAG_DAEMON_MODE) {
+		return ctdb_client_set_message_handler(ctdb, srvid, handler, private);
+	}
+	return ctdb_daemon_set_message_handler(ctdb, srvid, handler, private);
+}
+
diff --git a/ctdb/common/ctdb_daemon.c b/ctdb/common/ctdb_daemon.c
index a7cd7217280..b520dc7f0d8 100644
--- a/ctdb/common/ctdb_daemon.c
+++ b/ctdb/common/ctdb_daemon.c
@@ -59,6 +59,56 @@ struct ctdb_client {
 
 
 /*
+  message handler for when we are in daemon mode. This redirects the message
+  to the right client
+ */
+static void daemon_message_handler(struct ctdb_context *ctdb, uint32_t srvid, 
+				    TDB_DATA data, void *private)
+{
+	struct ctdb_client *client = talloc_get_type(private, struct ctdb_client);
+	struct ctdb_req_message *r;
+	int len;
+
+	/* construct a message to send to the client containing the data */
+	len = offsetof(struct ctdb_req_message, data) + data.dsize;
+	r = ctdbd_allocate_pkt(ctdb, len);
+	CTDB_NO_MEMORY(ctdb, r);
+	talloc_set_name_const(r, "req_message packet");
+
+	r->hdr.length    = len;
+	r->hdr.ctdb_magic = CTDB_MAGIC;
+	r->hdr.ctdb_version = CTDB_VERSION;
+	r->hdr.operation = CTDB_REQ_MESSAGE;
+	r->srvid         = srvid;
+	r->datalen       = data.dsize;
+	memcpy(&r->data[0], data.dptr, data.dsize);
+	
+	ctdb_queue_send(client->queue, (uint8_t *)&r->hdr, len);
+
+	talloc_free(r);
+	return 0;
+}
+					   
+
+/*
+  this is called when the ctdb daemon received a ctdb request to 
+  set the srvid from the client
+ */
+static void daemon_request_register_message_handler(struct ctdb_client *client, 
+						    struct ctdb_req_register *c)
+{
+	int res;
+	printf("XXX registering messaging handler %u in daemon\n", c->srvid);
+	res = ctdb_register_message_handler(client->ctdb, client, 
+					    c->srvid, daemon_message_handler, 
+					    client);
+	if (res != 0) {
+		printf("Failed to register handler %u in daemon\n", c->srvid);
+	}
+}
+
+
+/*
   destroy a ctdb_client
 */
 static int ctdb_client_destructor(struct ctdb_client *client)
@@ -152,6 +202,10 @@ static void client_incoming_packet(struct ctdb_client *client, void *data, size_
 		daemon_request_call_from_client(client, (struct ctdb_req_call *)hdr);
 		break;
 
+	case CTDB_REQ_REGISTER:
+		daemon_request_register_message_handler(client, 
+							(struct ctdb_req_register *)hdr);
+		break;
 	}
 
 	talloc_free(data);
diff --git a/ctdb/common/ctdb_message.c b/ctdb/common/ctdb_message.c
index ec39525942d..27c5c64bc0e 100644
--- a/ctdb/common/ctdb_message.c
+++ b/ctdb/common/ctdb_message.c
@@ -81,13 +81,35 @@ int ctdb_send_message(struct ctdb_context *ctdb, uint32_t vnn,
 }
 
 /*
+  when a client goes away, we need to remove its srvid handler from the list
+ */
+static int message_handler_destructor(struct ctdb_message_list *m)
+{
+	DLIST_REMOVE(m->ctdb->message_list, m);
+}
+
+/*
   setup handler for receipt of ctdb messages from ctdb_send_message()
 */
-int ctdb_set_message_handler(struct ctdb_context *ctdb, ctdb_message_fn_t handler,
-			     uint32_t srvid, void *private)
+int ctdb_register_message_handler(struct ctdb_context *ctdb, 
+				  TALLOC_CTX *mem_ctx,
+				  uint32_t srvid,
+				  ctdb_message_fn_t handler,
+				  void *private)
 {
-	ctdb->message_handler = handler;
-	ctdb->message_private = private;
+	struct ctdb_message_list *m;
+
+	m = talloc(mem_ctx, struct ctdb_message_list);
+	CTDB_NO_MEMORY(ctdb, m);
+
+	m->ctdb            = ctdb;
+	m->srvid           = srvid;
+	m->message_handler = handler;
+	m->message_private = private;
+	
+	DLIST_ADD(ctdb->message_list, m);
+
+	talloc_set_destructor(m, message_handler_destructor);
+
 	return 0;
 }
-
diff --git a/ctdb/include/ctdb.h b/ctdb/include/ctdb.h
index 28c6f21f082..f77b34abbda 100644
--- a/ctdb/include/ctdb.h
+++ b/ctdb/include/ctdb.h
@@ -180,11 +180,4 @@ struct ctdb_record_handle *ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALL
  */
 int ctdb_record_store(struct ctdb_record_handle *rec, TDB_DATA data);
 
-/* when running in daemon mode this function is used by a client to tell 
-   ctdb daemon what its local identifier is.
-   when in non-daemon mode this is a noop.
- */
-int ctdb_register_message_local_id(struct ctdb_context *ctdb, uint32_t messenger_id);
- 
-
 #endif
diff --git a/ctdb/include/ctdb_private.h b/ctdb/include/ctdb_private.h
index 5d2e36b1048..416156482e6 100644
--- a/ctdb/include/ctdb_private.h
+++ b/ctdb/include/ctdb_private.h
@@ -97,6 +97,16 @@ struct ctdb_upcalls {
 	void (*node_connected)(struct ctdb_node *);
 };
 
+/* list of message handlers - needs to be changed to a more efficient data
+   structure so we can find a message handler given a srvid quickly */
+struct ctdb_message_list {
+	struct ctdb_context *ctdb;
+	struct ctdb_message_list *next, *prev;
+	uint32_t srvid;
+	ctdb_message_fn_t message_handler;
+	void *message_private;
+};
+
 /* additional data required for the daemon mode */
 struct ctdb_daemon_data {
 	int sd;
@@ -120,9 +130,8 @@ struct ctdb_context {
 	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_message_list *message_list;
 	struct ctdb_daemon_data daemon;
 };
 
@@ -196,7 +205,7 @@ enum ctdb_operation {
 	CTDB_REQ_DMASTER      = 3,
 	CTDB_REPLY_DMASTER    = 4,
 	CTDB_REPLY_ERROR      = 5,
-	CTDB_REGISTER_CALL    = 6,
+	CTDB_REQ_REGISTER     = 6,
 	CTDB_REQ_MESSAGE      = 7
 };
 
@@ -262,8 +271,7 @@ struct ctdb_reply_dmaster {
 
 struct ctdb_register_call {
 	struct ctdb_req_header hdr;
-	uint32_t datalen;
-	uint8_t data[4];
+	uint32_t srvid;
 };
 
 struct ctdb_req_message {