libctdb: add synchronous message handling and unregister, with tests.

It turns out that we *do* want a separate private arg for the message
handler and the completion callback, so we change that.

We also fix the prototypes of the remove_message functions as we
implement them.

Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>

diff --git a/include/ctdb.h b/include/ctdb.h
index c5e6f935e42d50c24da0e951983ce52bd79f94eb..ef13b9f95f4e59c53b50d5d09f1229009a8cc266 100644 (file)
--- a/include/ctdb.h
+++ b/include/ctdb.h
@@ -307,8 +307,9 @@ typedef void (*ctdb_message_fn_t)(struct ctdb_connection *,
  * @ctdb: the ctdb_connection from ctdb_connect.
  * @srvid: the 64 bit identifier for our messages.
  * @handler: the callback when we receive such a message (typesafe)
+ * @handler_data: the argument to handler()
  * @callback: the callback when ctdb replies to our message (typesafe)
- * @cbdata: the argument to callback() and handler()
+ * @cbdata: the argument to callback()
  *
  * Note: our callback will always be called before handler.
  *
@@ -318,6 +319,7 @@ typedef void (*ctdb_message_fn_t)(struct ctdb_connection *,
 struct ctdb_request *
 ctdb_set_message_handler_send(struct ctdb_connection *ctdb, uint64_t srvid,
                              ctdb_message_fn_t handler,
+                             void *handler_data,
                              ctdb_callback_t callback,
                              void *cbdata);
 
@@ -336,6 +338,8 @@ bool ctdb_set_message_handler_recv(struct ctdb_connection *ctdb,
  * ctdb_remove_message_handler_send - unregister for messages to a srvid
  * @ctdb: the ctdb_connection from ctdb_connect.
  * @srvid: the 64 bit identifier for our messages.
+ * @handler: the callback when we receive such a message (typesafe)
+ * @handler_data: the argument to handler()
  * @callback: the callback when ctdb replies to our message (typesafe)
  * @cbdata: the argument to callback()
  *
@@ -344,6 +348,7 @@ bool ctdb_set_message_handler_recv(struct ctdb_connection *ctdb,
  */
 struct ctdb_request *
 ctdb_remove_message_handler_send(struct ctdb_connection *ctdb, uint64_t srvid,
+                                ctdb_message_fn_t handler, void *handler_data,
                                 ctdb_callback_t callback, void *cbdata);
 
 /**
@@ -354,7 +359,8 @@ ctdb_remove_message_handler_send(struct ctdb_connection *ctdb, uint64_t srvid,
  * After this returns true, the registered handler will no longer be called.
  * If this returns false, the de-registration failed.
  */
-bool ctdb_remove_message_handler_recv(struct ctdb_request *handle);
+bool ctdb_remove_message_handler_recv(struct ctdb_connection *ctdb,
+                                     struct ctdb_request *req);
 
 
 /**
@@ -501,18 +507,21 @@ struct ctdb_lock *ctdb_readrecordlock(struct ctdb_connection *ctdb,
  * failed.
  */
 bool ctdb_set_message_handler(struct ctdb_connection *ctdb, uint64_t srvid,
-                            ctdb_message_fn_t handler, void *cbdata);
+                             ctdb_message_fn_t handler, void *cbdata);
 
 
 /**
  * ctdb_remove_message_handler - deregister for messages (synchronous)
  * @ctdb: the ctdb_connection from ctdb_connect.
  * @srvid: the 64 bit identifier for our messages.
+ * @handler: the callback when we receive such a message (typesafe)
+ * @handler_data: the argument to handler()
  *
  * If this returns true, the message handler will no longer be called.
  * If this returns false, the deregistration failed.
  */
-bool ctdb_remove_message_handler(struct ctdb_connection *ctdb, uint64_t srvid);
+bool ctdb_remove_message_handler(struct ctdb_connection *ctdb, uint64_t srvid,
+                                ctdb_message_fn_t handler, void *handler_data);
 
 /**
  * ctdb_getpnn - read the pnn number of a node (synchronous)
@@ -552,12 +561,23 @@ bool ctdb_getrecmaster(struct ctdb_connection *ctdb,
         typesafe_cb_preargs(void, (cb), (cbdata),                      \
                             struct ctdb_connection *, struct ctdb_request *)
 
+#define ctdb_msgcb(cb, cbdata)                                         \
+       typesafe_cb_preargs(void, (cb), (cbdata),                       \
+                           struct ctdb_connection *, uint64_t, TDB_DATA)
+
 #define ctdb_connect(addr, log, logpriv)                               \
        ctdb_connect((addr),                                            \
                     typesafe_cb_postargs(void, (log), (logpriv),       \
                                          int, const char *, va_list),  \
                     (logpriv))
 
+#define ctdb_set_message_handler(ctdb, srvid, handler, hdata)          \
+       ctdb_set_message_handler((ctdb), (srvid),                       \
+                                ctdb_msgcb((handler), (hdata)), (hdata))
+
+#define ctdb_remove_message_handler(ctdb, srvid, handler, hdata)       \
+       ctdb_remove_message_handler((ctdb), (srvid),                    \
+                                   ctdb_msgcb((handler), (hdata)), (hdata))
 
 #define ctdb_attachdb_send(ctdb, name, persistent, tdb_flags, cb, cbdata) \
        ctdb_attachdb_send((ctdb), (name), (persistent), (tdb_flags),   \
@@ -569,12 +589,14 @@ bool ctdb_getrecmaster(struct ctdb_connection *ctdb,
                                    struct ctdb_db *, struct ctdb_lock *, \
                                    TDB_DATA), (cbdata))
 
-#define ctdb_set_message_handler_send(ctdb, srvid, handler, cb, cbdata)        \
-       ctdb_set_message_handler_send((ctdb), (srvid), (handler),       \
-             ctdb_sendcb((cb), (cbdata)), (cbdata))
+#define ctdb_set_message_handler_send(ctdb, srvid, handler, hdata, cb, cbdata) \
+       ctdb_set_message_handler_send((ctdb), (srvid),                  \
+                                     ctdb_msgcb((handler), (hdata)), (hdata), \
+                                     ctdb_sendcb((cb), (cbdata)), (cbdata))
 
-#define ctdb_remove_message_handler_send(ctdb, srvid, cb, cbdata)      \
+#define ctdb_remove_message_handler_send(ctdb, srvid, handler, hdata, cb, cbdata) \
        ctdb_remove_message_handler_send((ctdb), (srvid),               \
+             ctdb_msgcb((handler), (hdata)), (hdata),                  \
              ctdb_sendcb((cb), (cbdata)), (cbdata))
 
 #define ctdb_getpnn_send(ctdb, destnode, cb, cbdata)                   \

diff --git a/libctdb/messages.c b/libctdb/messages.c
index 7ad48d5dc834542fb6ce3edec2095057dbd229e6..b79857c52ff86f1ccbffaa7f6744160703ab5b67 100644 (file)
--- a/libctdb/messages.c
+++ b/libctdb/messages.c
@@ -6,6 +6,7 @@
 #include <ctdb_protocol.h>
 #include <stdlib.h>
 #include <string.h>
+#include <errno.h>
 
 /* Remove type-safety macros. */
 #undef ctdb_set_message_handler_send
@@ -17,7 +18,7 @@ struct message_handler_info {
 
        uint64_t srvid;
        ctdb_message_fn_t handler;
-       void *private_data;
+       void *handler_data;
 };
 
 void deliver_message(struct ctdb_connection *ctdb, struct ctdb_req_header *hdr)
@@ -30,9 +31,10 @@ void deliver_message(struct ctdb_connection *ctdb, struct ctdb_req_header *hdr)
        data.dptr = msg->data;
        data.dsize = msg->datalen;
 
+       /* Note: we want to call *every* handler: there may be more than one */
        for (i = ctdb->message_handlers; i; i = i->next) {
                if (i->srvid == msg->srvid) {
-                       i->handler(ctdb, msg->srvid, data, i->private_data);
+                       i->handler(ctdb, msg->srvid, data, i->handler_data);
                        found = true;
                }
        }
@@ -86,7 +88,7 @@ static void free_info(struct ctdb_connection *ctdb, struct ctdb_request *req)
 
 struct ctdb_request *
 ctdb_set_message_handler_send(struct ctdb_connection *ctdb, uint64_t srvid,
-                             ctdb_message_fn_t handler,
+                             ctdb_message_fn_t handler, void *handler_data,
                              ctdb_callback_t callback, void *private_data)
 {
        struct message_handler_info *info;
@@ -114,7 +116,7 @@ ctdb_set_message_handler_send(struct ctdb_connection *ctdb, uint64_t srvid,
 
        info->srvid = srvid;
        info->handler = handler;
-       info->private_data = private_data;
+       info->handler_data = handler_data;
 
        DEBUG(ctdb, LOG_DEBUG,
              "ctdb_set_message_handler_send: sending request %u for id %llu",
@@ -122,6 +124,69 @@ ctdb_set_message_handler_send(struct ctdb_connection *ctdb, uint64_t srvid,
        return req;
 }
 
+struct ctdb_request *
+ctdb_remove_message_handler_send(struct ctdb_connection *ctdb, uint64_t srvid,
+                                ctdb_message_fn_t handler, void *hdata,
+                                ctdb_callback_t callback, void *cbdata)
+{
+       struct message_handler_info *i;
+       struct ctdb_request *req;
+
+       for (i = ctdb->message_handlers; i; i = i->next) {
+               if (i->srvid == srvid
+                   && i->handler == handler && i->handler_data == hdata) {
+                       break;
+               }
+       }
+       if (!i) {
+               DEBUG(ctdb, LOG_ALERT,
+                     "ctdb_remove_message_handler_send: no such handler");
+               errno = ENOENT;
+               return NULL;
+       }
+
+       req = new_ctdb_control_request(ctdb, CTDB_CONTROL_DEREGISTER_SRVID,
+                                      CTDB_CURRENT_NODE, NULL, 0,
+                                      callback, cbdata);
+       if (!req) {
+               DEBUG(ctdb, LOG_ERR,
+                     "ctdb_remove_message_handler_send: allocating request");
+               return NULL;
+       }
+       req->hdr.control->srvid = srvid;
+       req->extra = i;
+
+       DEBUG(ctdb, LOG_DEBUG,
+             "ctdb_set_remove_handler_send: sending request %u for id %llu",
+             req->hdr.hdr->reqid, srvid);
+       return req;
+}
+
+bool ctdb_remove_message_handler_recv(struct ctdb_connection *ctdb,
+                                     struct ctdb_request *req)
+{
+       struct message_handler_info *handler = req->extra;
+       struct ctdb_reply_control *reply;
+
+       reply = unpack_reply_control(ctdb, req, CTDB_CONTROL_DEREGISTER_SRVID);
+       if (!reply) {
+               return false;
+       }
+       if (reply->status != 0) {
+               DEBUG(ctdb, LOG_ERR,
+                     "ctdb_remove_message_handler_recv: status %i",
+                     reply->status);
+               return false;
+       }
+
+       /* Remove ourselves from list of handlers. */
+       DLIST_REMOVE(ctdb->message_handlers, handler);
+       free(handler);
+       /* Crash if they call this again! */
+       req->extra = NULL;
+       return true;
+}
+
 bool ctdb_send_message(struct ctdb_connection *ctdb,
                      uint32_t pnn, uint64_t srvid,
                      TDB_DATA data)

diff --git a/libctdb/sync.c b/libctdb/sync.c
index 2e6ba9926fe5dae1b804a1e7c49a113082a7c483..2ec96d8737b444badace1b3b7d3d94c0a75dccca 100644 (file)
--- a/libctdb/sync.c
+++ b/libctdb/sync.c
@@ -23,6 +23,9 @@
 #include <stdlib.h>
 #include "libctdb_private.h"
 
+/* Remove type-safety macros. */
+#undef ctdb_set_message_handler
+
 /* On failure, frees req and returns NULL. */
 static struct ctdb_request *synchronous(struct ctdb_connection *ctdb,
                                        struct ctdb_request *req,
@@ -116,6 +119,24 @@ bool ctdb_getpnn(struct ctdb_connection *ctdb,
        return ret;
 }
 
+bool ctdb_set_message_handler(struct ctdb_connection *ctdb, uint64_t srvid,
+                             ctdb_message_fn_t handler, void *cbdata)
+{
+       struct ctdb_request *req;
+       bool done = false;
+       bool ret = false;
+
+       req = synchronous(ctdb,
+                         ctdb_set_message_handler_send(ctdb, srvid, handler,
+                                                       cbdata, set, &done),
+                         &done);
+       if (req != NULL) {
+               ret = ctdb_set_message_handler_recv(ctdb, req);
+               ctdb_request_free(ctdb, req);
+       }
+       return ret;
+}
+
 struct rrl_info {
        bool done;
        struct ctdb_lock *lock;