ctdb/client/client_message.c

   1 /*
   2    CTDB client code
   3
   4    Copyright (C) Amitay Isaacs  2015
   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 "replace.h"
  21 #include "system/network.h"
  22 #include "system/filesys.h"
  23
  24 #include <talloc.h>
  25 #include <tevent.h>
  26 #include <tdb.h>
  27
  28 #include "lib/util/tevent_unix.h"
  29
  30 #include "common/reqid.h"
  31 #include "common/srvid.h"
  32 #include "common/comm.h"
  33
  34 #include "protocol/protocol.h"
  35 #include "protocol/protocol_api.h"
  36
  37 #include "client/client_private.h"
  38 #include "client/client.h"
  39
  40
  41 /*
  42  * Handle REQ_MESSAGE
  43  */
  44
  45 struct ctdb_client_message_state {
  46         struct ctdb_client_context *client;
  47         uint32_t reqid;
  48 };
  49
  50 static int ctdb_client_message_state_destructor(
  51         struct ctdb_client_message_state *state);
  52 static void ctdb_client_message_done(struct tevent_req *subreq);
  53
  54 struct tevent_req *ctdb_client_message_send(TALLOC_CTX *mem_ctx,
  55                                             struct tevent_context *ev,
  56                                             struct ctdb_client_context *client,
  57                                             uint32_t destnode,
  58                                             struct ctdb_req_message *message)
  59 {
  60         struct tevent_req *req, *subreq;
  61         struct ctdb_client_message_state *state;
  62         struct ctdb_req_header h;
  63         uint32_t reqid;
  64         uint8_t *buf;
  65         size_t buflen;
  66         int ret;
  67
  68         req = tevent_req_create(mem_ctx, &state,
  69                                 struct ctdb_client_message_state);
  70         if (req == NULL) {
  71                 return NULL;
  72         }
  73
  74         reqid = reqid_new(client->idr, state);
  75         if (reqid == REQID_INVALID) {
  76                 talloc_free(req);
  77                 return NULL;
  78         }
  79
  80         state->client = client;
  81         state->reqid = reqid;
  82
  83         talloc_set_destructor(state, ctdb_client_message_state_destructor);
  84
  85         ctdb_req_header_fill(&h, 0, CTDB_REQ_MESSAGE, destnode,
  86                              client->pnn, reqid);
  87
  88         ret = ctdb_req_message_push(&h, message, state, &buf, &buflen);
  89         if (ret != 0) {
  90                 tevent_req_error(req, ret);
  91                 return tevent_req_post(req, ev);
  92         }
  93
  94         subreq = comm_write_send(state, ev, client->comm, buf, buflen);
  95         if (tevent_req_nomem(subreq, req)) {
  96                 return tevent_req_post(req, ev);
  97         }
  98         tevent_req_set_callback(subreq, ctdb_client_message_done, req);
  99
 100         return req;
 101 }
 102
 103 static int ctdb_client_message_state_destructor(
 104         struct ctdb_client_message_state *state)
 105 {
 106         reqid_remove(state->client->idr, state->reqid);
 107         return 0;
 108 }
 109
 110 static void ctdb_client_message_done(struct tevent_req *subreq)
 111 {
 112         struct tevent_req *req = tevent_req_callback_data(
 113                 subreq, struct tevent_req);
 114         int ret;
 115         bool status;
 116
 117         status = comm_write_recv(subreq, &ret);
 118         TALLOC_FREE(subreq);
 119         if (! status) {
 120                 tevent_req_error(req, ret);
 121                 return;
 122         }
 123
 124         tevent_req_done(req);
 125 }
 126
 127 bool ctdb_client_message_recv(struct tevent_req *req, int *perr)
 128 {
 129         int err;
 130
 131         if (tevent_req_is_unix_error(req, &err)) {
 132                 if (perr != NULL) {
 133                         *perr = err;
 134                 }
 135                 return false;
 136         }
 137
 138         return true;
 139 }
 140
 141 void ctdb_client_req_message(struct ctdb_client_context *client,
 142                              uint8_t *buf, size_t buflen, uint32_t reqid)
 143 {
 144         struct ctdb_req_header h;
 145         struct ctdb_req_message_data message;
 146         TALLOC_CTX *tmp_ctx = talloc_new(client);
 147         int ret;
 148
 149         ret = ctdb_req_message_data_pull(buf, buflen, &h, tmp_ctx, &message);
 150         if (ret != 0) {
 151                 return;
 152         }
 153
 154         srvid_dispatch(client->srv, message.srvid, CTDB_SRVID_ALL,
 155                        message.data);
 156         talloc_free(tmp_ctx);
 157 }
 158
 159 /*
 160  * Handle multiple nodes
 161  */
 162
 163 struct ctdb_client_message_multi_state {
 164         uint32_t *pnn_list;
 165         int count;
 166         int done;
 167         int err;
 168         int *err_list;
 169 };
 170
 171 struct message_index_state {
 172         struct tevent_req *req;
 173         int index;
 174 };
 175
 176 static void ctdb_client_message_multi_done(struct tevent_req *subreq);
 177
 178 struct tevent_req *ctdb_client_message_multi_send(
 179                                 TALLOC_CTX *mem_ctx,
 180                                 struct tevent_context *ev,
 181                                 struct ctdb_client_context *client,
 182                                 uint32_t *pnn_list, int count,
 183                                 struct ctdb_req_message *message)
 184 {
 185         struct tevent_req *req, *subreq;
 186         struct ctdb_client_message_multi_state *state;
 187         int i;
 188
 189         if (pnn_list == NULL || count == 0) {
 190                 return NULL;
 191         }
 192
 193         req = tevent_req_create(mem_ctx, &state,
 194                                 struct ctdb_client_message_multi_state);
 195         if (req == NULL) {
 196                 return NULL;
 197         }
 198
 199         state->pnn_list = pnn_list;
 200         state->count = count;
 201         state->done = 0;
 202         state->err = 0;
 203         state->err_list = talloc_zero_array(state, int, count);
 204         if (tevent_req_nomem(state->err_list, req)) {
 205                 return tevent_req_post(req, ev);
 206         }
 207
 208         for (i=0; i<count; i++) {
 209                 struct message_index_state *substate;
 210
 211                 subreq = ctdb_client_message_send(state, ev, client,
 212                                                   pnn_list[i], message);
 213                 if (tevent_req_nomem(subreq, req)) {
 214                         return tevent_req_post(req, ev);
 215                 }
 216
 217                 substate = talloc(subreq, struct message_index_state);
 218                 if (tevent_req_nomem(substate, req)) {
 219                         return tevent_req_post(req, ev);
 220                 }
 221
 222                 substate->req = req;
 223                 substate->index = i;
 224
 225                 tevent_req_set_callback(subreq, ctdb_client_message_multi_done,
 226                                         substate);
 227         }
 228
 229         return req;
 230 }
 231
 232 static void ctdb_client_message_multi_done(struct tevent_req *subreq)
 233 {
 234         struct message_index_state *substate = tevent_req_callback_data(
 235                 subreq, struct message_index_state);
 236         struct tevent_req *req = substate->req;
 237         int idx = substate->index;
 238         struct ctdb_client_message_multi_state *state = tevent_req_data(
 239                 req, struct ctdb_client_message_multi_state);
 240         bool status;
 241         int ret;
 242
 243         status = ctdb_client_message_recv(subreq, &ret);
 244         TALLOC_FREE(subreq);
 245         if (! status) {
 246                 if (state->err == 0) {
 247                         state->err = ret;
 248                         state->err_list[idx] = state->err;
 249                 }
 250         }
 251
 252         state->done += 1;
 253
 254         if (state->done == state->count) {
 255                 tevent_req_done(req);
 256         }
 257 }
 258
 259 bool ctdb_client_message_multi_recv(struct tevent_req *req, int *perr,
 260                                     TALLOC_CTX *mem_ctx, int **perr_list)
 261 {
 262         struct ctdb_client_message_multi_state *state = tevent_req_data(
 263                 req, struct ctdb_client_message_multi_state);
 264         int err;
 265
 266         if (tevent_req_is_unix_error(req, &err)) {
 267                 if (perr != NULL) {
 268                         *perr = err;
 269                 }
 270                 if (perr_list != NULL) {
 271                         *perr_list = talloc_steal(mem_ctx, state->err_list);
 272                 }
 273                 return false;
 274         }
 275
 276         if (perr != NULL) {
 277                 *perr = state->err;
 278         }
 279
 280         if (perr_list != NULL) {
 281                 *perr_list = talloc_steal(mem_ctx, state->err_list);
 282         }
 283
 284         if (state->err != 0) {
 285                 return false;
 286         }
 287
 288         return true;
 289 }
 290
 291 /*
 292  * sync version of message send
 293  */
 294
 295 int ctdb_client_message(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 296                         struct ctdb_client_context *client,
 297                         uint32_t destnode, struct ctdb_req_message *message)
 298 {
 299         TALLOC_CTX *tmp_ctx;
 300         struct tevent_req *req;
 301         int ret;
 302         bool status;
 303
 304         tmp_ctx = talloc_new(client);
 305         if (tmp_ctx == NULL) {
 306                 return ENOMEM;
 307         }
 308
 309         req = ctdb_client_message_send(tmp_ctx, ev, client, destnode, message);
 310         if (req == NULL) {
 311                 talloc_free(tmp_ctx);
 312                 return ENOMEM;
 313         }
 314
 315         tevent_req_poll(req, ev);
 316
 317         status = ctdb_client_message_recv(req, &ret);
 318         if (! status) {
 319                 talloc_free(tmp_ctx);
 320                 return ret;
 321         }
 322
 323         talloc_free(tmp_ctx);
 324         return 0;
 325 }
 326
 327 struct ctdb_client_set_message_handler_state {
 328         struct ctdb_client_context *client;
 329         uint64_t srvid;
 330         srvid_handler_fn handler;
 331         void *private_data;
 332 };
 333
 334 static void ctdb_client_set_message_handler_done(struct tevent_req *subreq);
 335
 336 struct tevent_req *ctdb_client_set_message_handler_send(
 337                                         TALLOC_CTX *mem_ctx,
 338                                         struct tevent_context *ev,
 339                                         struct ctdb_client_context *client,
 340                                         uint64_t srvid,
 341                                         srvid_handler_fn handler,
 342                                         void *private_data)
 343 {
 344         struct tevent_req *req, *subreq;
 345         struct ctdb_client_set_message_handler_state *state;
 346         struct ctdb_req_control request;
 347
 348         req = tevent_req_create(mem_ctx, &state,
 349                                 struct ctdb_client_set_message_handler_state);
 350         if (req == NULL) {
 351                 return NULL;
 352         }
 353
 354         state->client = client;
 355         state->srvid = srvid;
 356         state->handler = handler;
 357         state->private_data = private_data;
 358
 359         ctdb_req_control_register_srvid(&request, srvid);
 360         subreq = ctdb_client_control_send(state, ev, client, client->pnn,
 361                                           tevent_timeval_zero(), &request);
 362         if (tevent_req_nomem(subreq, req)) {
 363                 return tevent_req_post(req, ev);
 364         }
 365         tevent_req_set_callback(subreq, ctdb_client_set_message_handler_done,
 366                                 req);
 367
 368         return req;
 369 }
 370
 371 static void ctdb_client_set_message_handler_done(struct tevent_req *subreq)
 372 {
 373         struct tevent_req *req = tevent_req_callback_data(
 374                 subreq, struct tevent_req);
 375         struct ctdb_client_set_message_handler_state *state = tevent_req_data(
 376                 req, struct ctdb_client_set_message_handler_state);
 377         struct ctdb_reply_control *reply;
 378         bool status;
 379         int ret;
 380
 381         status = ctdb_client_control_recv(subreq, &ret, state, &reply);
 382         TALLOC_FREE(subreq);
 383         if (! status) {
 384                 tevent_req_error(req, ret);
 385                 return;
 386         }
 387
 388         ret = ctdb_reply_control_register_srvid(reply);
 389         talloc_free(reply);
 390         if (ret != 0) {
 391                 tevent_req_error(req, ret);
 392                 return;
 393         }
 394
 395         ret = srvid_register(state->client->srv, state->client, state->srvid,
 396                              state->handler, state->private_data);
 397         if (ret != 0) {
 398                 tevent_req_error(req, ret);
 399                 return;
 400         }
 401
 402         tevent_req_done(req);
 403 }
 404
 405 bool ctdb_client_set_message_handler_recv(struct tevent_req *req, int *perr)
 406 {
 407         int err;
 408
 409         if (tevent_req_is_unix_error(req, &err)) {
 410                 if (perr != NULL) {
 411                         *perr = err;
 412                 }
 413                 return false;
 414         }
 415         return true;
 416 }
 417
 418 struct ctdb_client_remove_message_handler_state {
 419         struct ctdb_client_context *client;
 420         uint64_t srvid;
 421         void *private_data;
 422 };
 423
 424 static void ctdb_client_remove_message_handler_done(struct tevent_req *subreq);
 425
 426 struct tevent_req *ctdb_client_remove_message_handler_send(
 427                                         TALLOC_CTX *mem_ctx,
 428                                         struct tevent_context *ev,
 429                                         struct ctdb_client_context *client,
 430                                         uint64_t srvid,
 431                                         void *private_data)
 432 {
 433         struct tevent_req *req, *subreq;
 434         struct ctdb_client_remove_message_handler_state *state;
 435         struct ctdb_req_control request;
 436
 437         req = tevent_req_create(mem_ctx, &state,
 438                                 struct ctdb_client_remove_message_handler_state);
 439         if (req == NULL) {
 440                 return NULL;
 441         }
 442
 443         state->client = client;
 444         state->srvid = srvid;
 445         state->private_data = private_data;
 446
 447         ctdb_req_control_deregister_srvid(&request, srvid);
 448         subreq = ctdb_client_control_send(state, ev, client, client->pnn,
 449                                           tevent_timeval_zero(), &request);
 450         if (tevent_req_nomem(subreq, req)) {
 451                 return tevent_req_post(req, ev);
 452         }
 453         tevent_req_set_callback(subreq,
 454                                 ctdb_client_remove_message_handler_done, req);
 455
 456         return req;
 457 }
 458
 459 static void ctdb_client_remove_message_handler_done(struct tevent_req *subreq)
 460 {
 461         struct tevent_req *req = tevent_req_callback_data(
 462                 subreq, struct tevent_req);
 463         struct ctdb_client_remove_message_handler_state *state = tevent_req_data(
 464                 req, struct ctdb_client_remove_message_handler_state);
 465         struct ctdb_reply_control *reply;
 466         bool status;
 467         int ret;
 468
 469         status = ctdb_client_control_recv(subreq, &ret, state, &reply);
 470         TALLOC_FREE(subreq);
 471         if (! status) {
 472                 tevent_req_error(req, ret);
 473                 return;
 474         }
 475
 476         ret = ctdb_reply_control_deregister_srvid(reply);
 477         talloc_free(reply);
 478         if (ret != 0) {
 479                 tevent_req_error(req, ret);
 480                 return;
 481         }
 482
 483         ret = srvid_deregister(state->client->srv, state->srvid,
 484                                state->private_data);
 485         if (ret != 0) {
 486                 tevent_req_error(req, ret);
 487                 return;
 488         }
 489
 490         tevent_req_done(req);
 491 }
 492
 493 bool ctdb_client_remove_message_handler_recv(struct tevent_req *req, int *perr)
 494 {
 495         int err;
 496
 497         if (tevent_req_is_unix_error(req, &err)) {
 498                 if (perr != NULL) {
 499                         *perr = err;
 500                 }
 501                 return false;
 502         }
 503         return true;
 504 }
 505
 506 int ctdb_client_set_message_handler(TALLOC_CTX *mem_ctx,
 507                                     struct tevent_context *ev,
 508                                     struct ctdb_client_context *client,
 509                                     uint64_t srvid, srvid_handler_fn handler,
 510                                     void *private_data)
 511 {
 512         int ret;
 513
 514         ret = ctdb_ctrl_register_srvid(mem_ctx, ev, client, client->pnn,
 515                                        tevent_timeval_zero(), srvid);
 516         if (ret != 0) {
 517                 return ret;
 518         }
 519
 520         return srvid_register(client->srv, client, srvid,
 521                               handler, private_data);
 522 }
 523
 524 int ctdb_client_remove_message_handler(TALLOC_CTX *mem_ctx,
 525                                        struct tevent_context *ev,
 526                                        struct ctdb_client_context *client,
 527                                        uint64_t srvid, void *private_data)
 528 {
 529         int ret;
 530
 531         ret = ctdb_ctrl_deregister_srvid(mem_ctx, ev, client, client->pnn,
 532                                          tevent_timeval_zero(), srvid);
 533         if (ret != 0) {
 534                 return ret;
 535         }
 536
 537         return srvid_deregister(client->srv, srvid, private_data);
 538 }