libctdb/messages.c

   1 /*
   2    core of libctdb
   3
   4    Copyright (C) Rusty Russell 2010
   5
   6    This program is free software; you can redistribute it and/or modify
   7    it under the terms of the GNU General Public License as published by
   8    the Free Software Foundation; either version 3 of the License, or
   9    (at your option) any later version.
  10
  11    This program is distributed in the hope that it will be useful,
  12    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14    GNU General Public License for more details.
  15
  16    You should have received a copy of the GNU General Public License
  17    along with this program; if not, see <http://www.gnu.org/licenses/>.
  18 */
  19
  20 #include <sys/socket.h>
  21 #include "libctdb_private.h"
  22 #include "messages.h"
  23 #include "io_elem.h"
  24 #include <ctdb.h>
  25 #include <tdb.h>
  26 #include <ctdb_protocol.h>
  27 #include <stdlib.h>
  28 #include <string.h>
  29 #include <errno.h>
  30
  31 /* Remove type-safety macros. */
  32 #undef ctdb_set_message_handler_send
  33 #undef ctdb_set_message_handler_recv
  34 #undef ctdb_remove_message_handler_send
  35
  36 struct message_handler_info {
  37         struct message_handler_info *next, *prev;
  38
  39         uint64_t srvid;
  40         ctdb_message_fn_t handler;
  41         void *handler_data;
  42 };
  43
  44 void deliver_message(struct ctdb_connection *ctdb, struct ctdb_req_header *hdr)
  45 {
  46         struct message_handler_info *i;
  47         struct ctdb_req_message *msg = (struct ctdb_req_message *)hdr;
  48         TDB_DATA data;
  49         bool found;
  50
  51         data.dptr = msg->data;
  52         data.dsize = msg->datalen;
  53
  54         /* Note: we want to call *every* handler: there may be more than one */
  55         for (i = ctdb->message_handlers; i; i = i->next) {
  56                 if (i->srvid == msg->srvid) {
  57                         i->handler(ctdb, msg->srvid, data, i->handler_data);
  58                         found = true;
  59                 }
  60         }
  61         if (!found) {
  62                 DEBUG(ctdb, LOG_WARNING,
  63                       "ctdb_service: messsage for unregistered srvid %llu",
  64                       (unsigned long long)msg->srvid);
  65         }
  66 }
  67
  68 void remove_message_handlers(struct ctdb_connection *ctdb)
  69 {
  70         struct message_handler_info *i;
  71
  72         /* ctdbd should unregister automatically when we close fd, so we don't
  73            need to do that here. */
  74         while ((i = ctdb->message_handlers) != NULL) {
  75                 DLIST_REMOVE(ctdb->message_handlers, i);
  76                 free(i);
  77         }
  78 }
  79
  80 static void free_info(struct ctdb_connection *ctdb, struct ctdb_request *req);
  81
  82 struct ctdb_request *
  83 ctdb_set_message_handler_send(struct ctdb_connection *ctdb, uint64_t srvid,
  84                               ctdb_message_fn_t handler, void *handler_data,
  85                               ctdb_callback_t callback, void *private_data)
  86 {
  87         struct message_handler_info *info;
  88         struct ctdb_request *req;
  89
  90         info = malloc(sizeof(*info));
  91         if (!info) {
  92                 DEBUG(ctdb, LOG_ERR,
  93                       "ctdb_set_message_handler_send: allocating info");
  94                 return NULL;
  95         }
  96
  97         req = new_ctdb_control_request(ctdb, CTDB_CONTROL_REGISTER_SRVID,
  98                                        CTDB_CURRENT_NODE, NULL, 0,
  99                                        callback, private_data);
 100         if (!req) {
 101                 DEBUG(ctdb, LOG_ERR,
 102                       "ctdb_set_message_handler_send: allocating request");
 103                 free(info);
 104                 return NULL;
 105         }
 106         req->extra = info;
 107         req->extra_destructor = free_info;
 108         req->hdr.control->srvid = srvid;
 109
 110         info->srvid = srvid;
 111         info->handler = handler;
 112         info->handler_data = handler_data;
 113
 114         DEBUG(ctdb, LOG_DEBUG,
 115               "ctdb_set_message_handler_send: sending request %u for id %llx",
 116               req->hdr.hdr->reqid, (unsigned long long)srvid);
 117         return req;
 118 }
 119
 120 static void free_info(struct ctdb_connection *ctdb, struct ctdb_request *req)
 121 {
 122         free(req->extra);
 123 }
 124
 125 bool ctdb_set_message_handler_recv(struct ctdb_connection *ctdb,
 126                                    struct ctdb_request *req)
 127 {
 128         struct message_handler_info *info = req->extra;
 129         struct ctdb_reply_control *reply;
 130
 131         reply = unpack_reply_control(req, CTDB_CONTROL_REGISTER_SRVID);
 132         if (!reply) {
 133                 return false;
 134         }
 135         if (reply->status != 0) {
 136                 DEBUG(ctdb, LOG_ERR,
 137                       "ctdb_set_message_handler_recv: status %i",
 138                       reply->status);
 139                 return false;
 140         }
 141
 142         /* Put ourselves in list of handlers. */
 143         DLIST_ADD(ctdb->message_handlers, info);
 144         /* Keep safe from destructor */
 145         req->extra = NULL;
 146         return true;
 147 }
 148
 149 struct ctdb_request *
 150 ctdb_remove_message_handler_send(struct ctdb_connection *ctdb, uint64_t srvid,
 151                                  ctdb_message_fn_t handler, void *hdata,
 152                                  ctdb_callback_t callback, void *cbdata)
 153 {
 154         struct message_handler_info *i;
 155         struct ctdb_request *req;
 156
 157         for (i = ctdb->message_handlers; i; i = i->next) {
 158                 if (i->srvid == srvid
 159                     && i->handler == handler && i->handler_data == hdata) {
 160                         break;
 161                 }
 162         }
 163         if (!i) {
 164                 DEBUG(ctdb, LOG_ALERT,
 165                       "ctdb_remove_message_handler_send: no such handler");
 166                 errno = ENOENT;
 167                 return NULL;
 168         }
 169
 170         req = new_ctdb_control_request(ctdb, CTDB_CONTROL_DEREGISTER_SRVID,
 171                                        CTDB_CURRENT_NODE, NULL, 0,
 172                                        callback, cbdata);
 173         if (!req) {
 174                 DEBUG(ctdb, LOG_ERR,
 175                       "ctdb_remove_message_handler_send: allocating request");
 176                 return NULL;
 177         }
 178         req->hdr.control->srvid = srvid;
 179         req->extra = i;
 180
 181         DEBUG(ctdb, LOG_DEBUG,
 182               "ctdb_set_remove_handler_send: sending request %u for id %llu",
 183               req->hdr.hdr->reqid, (unsigned long long)srvid);
 184         return req;
 185 }
 186
 187 bool ctdb_remove_message_handler_recv(struct ctdb_connection *ctdb,
 188                                       struct ctdb_request *req)
 189 {
 190         struct message_handler_info *handler = req->extra;
 191         struct ctdb_reply_control *reply;
 192
 193         reply = unpack_reply_control(req, CTDB_CONTROL_DEREGISTER_SRVID);
 194         if (!reply) {
 195                 return false;
 196         }
 197         if (reply->status != 0) {
 198                 DEBUG(ctdb, LOG_ERR,
 199                       "ctdb_remove_message_handler_recv: status %i",
 200                       reply->status);
 201                 return false;
 202         }
 203
 204         /* Remove ourselves from list of handlers. */
 205         DLIST_REMOVE(ctdb->message_handlers, handler);
 206         free(handler);
 207         /* Crash if they call this again! */
 208         req->extra = NULL;
 209         return true;
 210 }
 211
 212 bool ctdb_send_message(struct ctdb_connection *ctdb,
 213                       uint32_t pnn, uint64_t srvid,
 214                       TDB_DATA data)
 215 {
 216         struct ctdb_request *req;
 217         struct ctdb_req_message *pkt;
 218
 219         /* We just discard it once it's finished: no reply. */
 220         req = new_ctdb_request(
 221                 ctdb, offsetof(struct ctdb_req_message, data) + data.dsize,
 222                 ctdb_cancel_callback, NULL);
 223         if (!req) {
 224                 DEBUG(ctdb, LOG_ERR, "ctdb_set_message: allocating message");
 225                 return false;
 226         }
 227
 228         io_elem_init_req_header(req->io,
 229                                 CTDB_REQ_MESSAGE, pnn, new_reqid(ctdb));
 230
 231         pkt = req->hdr.message;
 232         pkt->srvid = srvid;
 233         pkt->datalen = data.dsize;
 234         memcpy(pkt->data, data.dptr, data.dsize);
 235         DLIST_ADD_END(ctdb->outq, req, struct ctdb_request);
 236         return true;
 237 }