ctdb/client/client_tunnel.c

   1 /*
   2    CTDB client code
   3
   4    Copyright (C) Amitay Isaacs  2016
   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
  23 #include <talloc.h>
  24 #include <tevent.h>
  25 #include <tdb.h>
  26
  27 #include "lib/util/tevent_unix.h"
  28
  29 #include "common/reqid.h"
  30 #include "common/srvid.h"
  31 #include "common/comm.h"
  32
  33 #include "protocol/protocol.h"
  34 #include "protocol/protocol_api.h"
  35
  36 #include "client/client_private.h"
  37 #include "client/client.h"
  38
  39
  40 struct ctdb_tunnel_data {
  41         struct ctdb_req_header hdr;
  42         struct ctdb_req_tunnel *tunnel;
  43         uint32_t reqid;
  44 };
  45
  46 /*
  47  * Tunnel setup and destroy
  48  */
  49
  50 struct ctdb_tunnel_setup_state {
  51         struct ctdb_client_context *client;
  52         struct ctdb_tunnel_context *tctx;
  53         uint64_t tunnel_id;
  54 };
  55
  56 static void ctdb_tunnel_setup_register_done(struct tevent_req *subreq);
  57 static void ctdb_tunnel_handler(uint64_t tunnel_id, TDB_DATA data,
  58                                 void *private_data);
  59
  60 struct tevent_req *ctdb_tunnel_setup_send(TALLOC_CTX *mem_ctx,
  61                                           struct tevent_context *ev,
  62                                           struct ctdb_client_context *client,
  63                                           uint64_t tunnel_id,
  64                                           ctdb_tunnel_callback_func_t callback,
  65                                           void *private_data)
  66 {
  67         struct tevent_req *req, *subreq;
  68         struct ctdb_tunnel_setup_state *state;
  69         struct ctdb_tunnel_context *tctx;
  70         struct ctdb_req_control request;
  71         int ret;
  72
  73         req = tevent_req_create(mem_ctx, &state,
  74                                 struct ctdb_tunnel_setup_state);
  75         if (req == NULL) {
  76                 return NULL;
  77         }
  78
  79         tctx = talloc_zero(client, struct ctdb_tunnel_context);
  80         if (tevent_req_nomem(tctx, req)) {
  81                 return tevent_req_post(req, ev);
  82         }
  83
  84         tctx->client = client;
  85         tctx->tunnel_id = tunnel_id;
  86         tctx->callback = callback;
  87         tctx->private_data = private_data;
  88
  89         state->client = client;
  90         state->tunnel_id = tunnel_id;
  91         state->tctx = tctx;
  92
  93         ret = srvid_exists(client->tunnels, tunnel_id, NULL);
  94         if (ret == 0) {
  95                 tevent_req_error(req, EEXIST);
  96                 return tevent_req_post(req, ev);
  97         }
  98
  99         ctdb_req_control_tunnel_register(&request, tunnel_id);
 100         subreq = ctdb_client_control_send(state, ev, client,
 101                                           ctdb_client_pnn(client),
 102                                           tevent_timeval_zero(),
 103                                           &request);
 104         if (tevent_req_nomem(subreq, req)) {
 105                 return tevent_req_post(req, ev);
 106         }
 107         tevent_req_set_callback(subreq, ctdb_tunnel_setup_register_done, req);
 108
 109         return req;
 110 }
 111
 112 static void ctdb_tunnel_setup_register_done(struct tevent_req *subreq)
 113 {
 114         struct tevent_req *req = tevent_req_callback_data(
 115                 subreq, struct tevent_req);
 116         struct ctdb_tunnel_setup_state *state = tevent_req_data(
 117                 req, struct ctdb_tunnel_setup_state);
 118         struct ctdb_reply_control *reply;
 119         bool status;
 120         int ret;
 121
 122         status = ctdb_client_control_recv(subreq, &ret, state, &reply);
 123         TALLOC_FREE(subreq);
 124         if (! status) {
 125                 tevent_req_error(req, ret);
 126                 return;
 127         }
 128
 129         ret = ctdb_reply_control_tunnel_register(reply);
 130         talloc_free(reply);
 131         if (ret != 0) {
 132                 tevent_req_error(req, ret);
 133                 return;
 134         }
 135
 136         ret = srvid_register(state->client->tunnels, state->client,
 137                              state->tunnel_id,
 138                              ctdb_tunnel_handler, state->tctx);
 139         if (ret != 0) {
 140                 tevent_req_error(req, ret);
 141                 return;
 142         }
 143
 144         tevent_req_done(req);
 145 }
 146
 147 static void ctdb_tunnel_handler(uint64_t tunnel_id, TDB_DATA data,
 148                                 void *private_data)
 149 {
 150         struct ctdb_tunnel_context *tctx = talloc_get_type_abort(
 151                 private_data, struct ctdb_tunnel_context);
 152         struct ctdb_tunnel_data *tunnel_data;
 153
 154         if (tctx->tunnel_id != tunnel_id) {
 155                 return;
 156         }
 157
 158         if (data.dsize != sizeof(struct ctdb_tunnel_data)) {
 159                 return;
 160         }
 161
 162         tunnel_data = (struct ctdb_tunnel_data *)data.dptr;
 163
 164         tctx->callback(tctx, tunnel_data->hdr.srcnode, tunnel_data->reqid,
 165                        tunnel_data->tunnel->data.dptr,
 166                        tunnel_data->tunnel->data.dsize, tctx->private_data);
 167 }
 168
 169 bool ctdb_tunnel_setup_recv(struct tevent_req *req, int *perr,
 170                             struct ctdb_tunnel_context **result)
 171 {
 172         struct ctdb_tunnel_setup_state *state = tevent_req_data(
 173                 req, struct ctdb_tunnel_setup_state);
 174         int ret;
 175
 176         if (tevent_req_is_unix_error(req, &ret)) {
 177                 if (perr != NULL) {
 178                         *perr = ret;
 179                 }
 180                 return false;
 181         }
 182
 183         *result = state->tctx;
 184         return true;
 185 }
 186
 187 int ctdb_tunnel_setup(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 188                       struct ctdb_client_context *client, uint64_t tunnel_id,
 189                       ctdb_tunnel_callback_func_t callback, void *private_data,
 190                       struct ctdb_tunnel_context **result)
 191 {
 192         struct tevent_req *req;
 193         int ret;
 194         bool status;
 195
 196         req = ctdb_tunnel_setup_send(mem_ctx, ev, client, tunnel_id,
 197                                      callback, private_data);
 198         if (req == NULL) {
 199                 return ENOMEM;
 200         }
 201
 202         tevent_req_poll(req, ev);
 203
 204         status = ctdb_tunnel_setup_recv(req, &ret, result);
 205         talloc_free(req);
 206         if (! status) {
 207                 return ret;
 208         }
 209
 210         return 0;
 211 }
 212
 213 struct ctdb_tunnel_destroy_state {
 214         struct ctdb_tunnel_context *tctx;
 215 };
 216
 217 static void ctdb_tunnel_destroy_deregister_done(struct tevent_req *subreq);
 218
 219 struct tevent_req *ctdb_tunnel_destroy_send(TALLOC_CTX *mem_ctx,
 220                                             struct tevent_context *ev,
 221                                             struct ctdb_tunnel_context *tctx)
 222 {
 223         struct tevent_req *req, *subreq;
 224         struct ctdb_tunnel_destroy_state *state;
 225         struct ctdb_req_control request;
 226
 227         req = tevent_req_create(mem_ctx, &state,
 228                                 struct ctdb_tunnel_destroy_state);
 229         if (req == NULL) {
 230                 return NULL;
 231         }
 232
 233         state->tctx = tctx;
 234
 235         ctdb_req_control_tunnel_deregister(&request, tctx->tunnel_id);
 236         subreq = ctdb_client_control_send(state, ev, tctx->client,
 237                                           ctdb_client_pnn(tctx->client),
 238                                           tevent_timeval_zero(),
 239                                           &request);
 240         if (tevent_req_nomem(subreq, req)) {
 241                 return tevent_req_post(req, ev);
 242         }
 243         tevent_req_set_callback(subreq, ctdb_tunnel_destroy_deregister_done,
 244                                 req);
 245
 246         return req;
 247 }
 248
 249 static void ctdb_tunnel_destroy_deregister_done(struct tevent_req *subreq)
 250 {
 251         struct tevent_req *req = tevent_req_callback_data(
 252                 subreq, struct tevent_req);
 253         struct ctdb_tunnel_destroy_state *state = tevent_req_data(
 254                 req, struct ctdb_tunnel_destroy_state);
 255         struct ctdb_client_context *client = state->tctx->client;
 256         struct ctdb_reply_control *reply;
 257         bool status;
 258         int ret;
 259
 260         status = ctdb_client_control_recv(subreq, &ret, state, &reply);
 261         TALLOC_FREE(subreq);
 262         if (! status) {
 263                 tevent_req_error(req, ret);
 264                 return;
 265         }
 266
 267         ret = ctdb_reply_control_tunnel_deregister(reply);
 268         talloc_free(reply);
 269         if (ret != 0) {
 270                 tevent_req_error(req, ret);
 271                 return;
 272         }
 273
 274         ret = srvid_deregister(client->tunnels, state->tctx->tunnel_id,
 275                                state->tctx);
 276         if (ret != 0) {
 277                 tevent_req_error(req, ret);
 278                 return;
 279         }
 280
 281         tevent_req_done(req);
 282 }
 283
 284 bool ctdb_tunnel_destroy_recv(struct tevent_req *req, int *perr)
 285 {
 286         int ret;
 287
 288         if (tevent_req_is_unix_error(req, &ret)) {
 289                 if (perr != NULL) {
 290                         *perr = ret;
 291                 }
 292                 return false;
 293         }
 294         return true;
 295 }
 296
 297
 298 int ctdb_tunnel_destroy(struct tevent_context *ev,
 299                         struct ctdb_tunnel_context *tctx)
 300 {
 301         struct tevent_req *req;
 302         int ret;
 303         bool status;
 304
 305         req = ctdb_tunnel_destroy_send(ev, ev, tctx);
 306         if (req == NULL) {
 307                 return ENOMEM;
 308         }
 309
 310         tevent_req_poll(req, ev);
 311
 312         status = ctdb_tunnel_destroy_recv(req, &ret);
 313         talloc_free(req);
 314         if (! status) {
 315                 return ret;
 316         }
 317
 318         return 0;
 319 }
 320
 321 /*
 322  * Callback when REQ_TUNNEL packet is received
 323  */
 324
 325 static void ctdb_tunnel_request_reply(struct tevent_req *req,
 326                                       struct ctdb_tunnel_data *tunnel_data);
 327
 328 void ctdb_client_req_tunnel(struct ctdb_client_context *client,
 329                             uint8_t *buf, size_t buflen, uint32_t reqid)
 330 {
 331         TALLOC_CTX *tmp_ctx = talloc_new(client);
 332         struct ctdb_req_header h;
 333         struct ctdb_req_tunnel *tunnel;
 334         struct tevent_req *req;
 335         struct ctdb_tunnel_data tunnel_data;
 336         int ret;
 337
 338         tunnel = talloc_zero(tmp_ctx, struct ctdb_req_tunnel);
 339         if (tunnel == NULL) {
 340                 goto fail;
 341         }
 342
 343         ret = ctdb_req_tunnel_pull(buf, buflen, &h, tmp_ctx, tunnel);
 344         if (ret != 0) {
 345                 goto fail;
 346         }
 347
 348         tunnel_data = (struct ctdb_tunnel_data) {
 349                 .hdr = h,
 350                 .tunnel = tunnel,
 351                 .reqid = reqid,
 352         };
 353
 354         if (tunnel->flags & CTDB_TUNNEL_FLAG_REPLY) {
 355                 req = reqid_find(client->idr, reqid, struct tevent_req);
 356                 if (req == NULL) {
 357                         goto fail;
 358                 }
 359
 360                 ctdb_tunnel_request_reply(req, &tunnel_data);
 361
 362         } else if (tunnel->flags & CTDB_TUNNEL_FLAG_REQUEST) {
 363
 364                 TDB_DATA data = {
 365                         .dsize = sizeof(struct ctdb_tunnel_data),
 366                         .dptr = (uint8_t *)&tunnel_data,
 367                 };
 368
 369                 srvid_dispatch(client->tunnels, tunnel->tunnel_id, 0, data);
 370         }
 371
 372 fail:
 373         TALLOC_FREE(tmp_ctx);
 374 }
 375
 376
 377 /*
 378  * Send messages using tunnel
 379  */
 380
 381 struct ctdb_tunnel_request_state {
 382         struct ctdb_tunnel_context *tctx;
 383         bool wait_for_reply;
 384         uint32_t reqid;
 385         struct ctdb_req_tunnel *tunnel;
 386 };
 387
 388 static int ctdb_tunnel_request_state_destructor(
 389                         struct ctdb_tunnel_request_state *state);
 390 static void ctdb_tunnel_request_done(struct tevent_req *subreq);
 391
 392 struct tevent_req *ctdb_tunnel_request_send(TALLOC_CTX *mem_ctx,
 393                                             struct tevent_context *ev,
 394                                             struct ctdb_tunnel_context *tctx,
 395                                             int destnode,
 396                                             struct timeval timeout,
 397                                             uint8_t *buf, size_t buflen,
 398                                             bool wait_for_reply)
 399 {
 400         struct tevent_req *req, *subreq;
 401         struct ctdb_tunnel_request_state *state;
 402         struct ctdb_req_tunnel tunnel;
 403         struct ctdb_req_header h;
 404         uint8_t *pkt;
 405         size_t datalen, pkt_len;
 406         int ret;
 407
 408         req = tevent_req_create(mem_ctx, &state,
 409                                 struct ctdb_tunnel_request_state);
 410         if (req == NULL) {
 411                 return NULL;
 412         }
 413
 414         state->tctx = tctx;
 415         state->wait_for_reply = wait_for_reply;
 416         state->reqid = reqid_new(tctx->client->idr, req);
 417         if (state->reqid == REQID_INVALID) {
 418                 talloc_free(req);
 419                 return NULL;
 420         }
 421
 422         talloc_set_destructor(state, ctdb_tunnel_request_state_destructor);
 423
 424         tunnel = (struct ctdb_req_tunnel) {
 425                 .tunnel_id = state->tctx->tunnel_id,
 426                 .flags = CTDB_TUNNEL_FLAG_REQUEST,
 427                 .data = (TDB_DATA) {
 428                         .dptr = buf,
 429                         .dsize = buflen,
 430                 },
 431         };
 432
 433         if (destnode == CTDB_BROADCAST_ALL ||
 434             destnode == CTDB_BROADCAST_ACTIVE ||
 435             destnode == CTDB_BROADCAST_CONNECTED) {
 436                 state->wait_for_reply = false;
 437         }
 438         if (! state->wait_for_reply) {
 439                 tunnel.flags |= CTDB_TUNNEL_FLAG_NOREPLY;
 440         }
 441
 442         ctdb_req_header_fill(&h, 0, CTDB_REQ_TUNNEL, destnode,
 443                              ctdb_client_pnn(state->tctx->client),
 444                              state->reqid);
 445
 446         datalen = ctdb_req_tunnel_len(&h, &tunnel);
 447         ret = ctdb_allocate_pkt(state, datalen, &pkt, &pkt_len);
 448         if (ret != 0) {
 449                 tevent_req_error(req, ret);
 450                 return tevent_req_post(req, ev);
 451         }
 452
 453         ret = ctdb_req_tunnel_push(&h, &tunnel, pkt, &pkt_len);
 454         if (ret != 0) {
 455                 tevent_req_error(req, ret);
 456                 return tevent_req_post(req, ev);
 457         }
 458
 459         if (!tevent_timeval_is_zero(&timeout)) {
 460                 if (!tevent_req_set_endtime(req, ev, timeout)) {
 461                         return tevent_req_post(req, ev);
 462                 }
 463         }
 464
 465         subreq = comm_write_send(state, ev, tctx->client->comm,
 466                                  pkt, pkt_len);
 467         if (tevent_req_nomem(subreq, req)) {
 468                 return tevent_req_post(req, ev);
 469         }
 470         tevent_req_set_callback(subreq, ctdb_tunnel_request_done, req);
 471
 472         return req;
 473 }
 474
 475 static int ctdb_tunnel_request_state_destructor(
 476                         struct ctdb_tunnel_request_state *state)
 477 {
 478         reqid_remove(state->tctx->client->idr, state->reqid);
 479         return 0;
 480 }
 481
 482 static void ctdb_tunnel_request_done(struct tevent_req *subreq)
 483 {
 484         struct tevent_req *req = tevent_req_callback_data(
 485                 subreq, struct tevent_req);
 486         struct ctdb_tunnel_request_state *state = tevent_req_data(
 487                 req, struct ctdb_tunnel_request_state);
 488         int ret;
 489         bool status;
 490
 491         status = comm_write_recv(subreq, &ret);
 492         TALLOC_FREE(subreq);
 493         if (! status) {
 494                 tevent_req_error(req, ret);
 495                 return;
 496         }
 497
 498         if (! state->wait_for_reply) {
 499                 tevent_req_done(req);
 500         }
 501
 502         /* Wait for the reply or timeout */
 503 }
 504
 505 static void ctdb_tunnel_request_reply(struct tevent_req *req,
 506                                       struct ctdb_tunnel_data *tunnel_data)
 507 {
 508         struct ctdb_tunnel_request_state *state = tevent_req_data(
 509                 req, struct ctdb_tunnel_request_state);
 510
 511         if (tunnel_data->reqid != state->reqid) {
 512                 return;
 513         }
 514
 515         state->tunnel = talloc_steal(state, tunnel_data->tunnel);
 516         tevent_req_done(req);
 517 }
 518
 519 bool ctdb_tunnel_request_recv(struct tevent_req *req, int *perr,
 520                               TALLOC_CTX *mem_ctx, uint8_t **buf,
 521                               size_t *buflen)
 522 {
 523         struct ctdb_tunnel_request_state *state = tevent_req_data(
 524                 req, struct ctdb_tunnel_request_state);
 525         int ret;
 526
 527         if (tevent_req_is_unix_error(req, &ret)) {
 528                 if (perr != NULL) {
 529                         *perr = ret;
 530                 }
 531                 return false;
 532         }
 533
 534         if (state->wait_for_reply) {
 535                 if (buf != NULL) {
 536                         *buf = talloc_steal(mem_ctx, state->tunnel->data.dptr);
 537                 }
 538                 if (buflen != NULL) {
 539                         *buflen = state->tunnel->data.dsize;
 540                 }
 541         }
 542
 543         return true;
 544 }
 545
 546 int ctdb_tunnel_request(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 547                         struct ctdb_tunnel_context *tctx, int destnode,
 548                         struct timeval timeout, uint8_t *buf, size_t buflen,
 549                         bool wait_for_reply)
 550 {
 551         struct tevent_req *req;
 552         int ret;
 553         bool status;
 554
 555         req = ctdb_tunnel_request_send(mem_ctx, ev, tctx, destnode,
 556                                        timeout, buf, buflen, wait_for_reply);
 557         if (req == NULL) {
 558                 return ENOMEM;
 559         }
 560
 561         tevent_req_poll(req, ev);
 562
 563         status = ctdb_tunnel_request_recv(req, &ret, NULL, NULL, NULL);
 564         talloc_free(req);
 565         if (! status) {
 566                 return ret;
 567         }
 568
 569         return 0;
 570 }
 571
 572 struct ctdb_tunnel_reply_state {
 573 };
 574
 575 static void ctdb_tunnel_reply_done(struct tevent_req *subreq);
 576
 577 struct tevent_req *ctdb_tunnel_reply_send(TALLOC_CTX *mem_ctx,
 578                                           struct tevent_context *ev,
 579                                           struct ctdb_tunnel_context *tctx,
 580                                           int destnode, uint32_t reqid,
 581                                           struct timeval timeout,
 582                                           uint8_t *buf, size_t buflen)
 583 {
 584         struct tevent_req *req, *subreq;
 585         struct ctdb_tunnel_reply_state *state;
 586         struct ctdb_req_tunnel tunnel;
 587         struct ctdb_req_header h;
 588         uint8_t *pkt;
 589         size_t datalen, pkt_len;
 590         int ret;
 591
 592         req = tevent_req_create(mem_ctx, &state,
 593                                 struct ctdb_tunnel_reply_state);
 594         if (req == NULL) {
 595                 return NULL;
 596         }
 597
 598         tunnel = (struct ctdb_req_tunnel) {
 599                 .tunnel_id = tctx->tunnel_id,
 600                 .flags = CTDB_TUNNEL_FLAG_REPLY,
 601                 .data = (TDB_DATA) {
 602                         .dptr = buf,
 603                         .dsize = buflen,
 604                 },
 605         };
 606
 607         ctdb_req_header_fill(&h, 0, CTDB_REQ_TUNNEL, destnode,
 608                              ctdb_client_pnn(tctx->client), reqid);
 609
 610         datalen = ctdb_req_tunnel_len(&h, &tunnel);
 611         ret = ctdb_allocate_pkt(state, datalen, &pkt, &pkt_len);
 612         if (ret != 0) {
 613                 tevent_req_error(req, ret);
 614                 return tevent_req_post(req, ev);
 615         }
 616
 617         ret = ctdb_req_tunnel_push(&h, &tunnel, pkt, &pkt_len);
 618         if (ret != 0) {
 619                 tevent_req_error(req, ret);
 620                 return tevent_req_post(req, ev);
 621         }
 622
 623         if (!tevent_timeval_is_zero(&timeout)) {
 624                 if (!tevent_req_set_endtime(req, ev, timeout)) {
 625                         return tevent_req_post(req, ev);
 626                 }
 627         }
 628
 629         subreq = comm_write_send(state, ev, tctx->client->comm, pkt, pkt_len);
 630         if (tevent_req_nomem(subreq, req)) {
 631                 return tevent_req_post(req, ev);
 632         }
 633         tevent_req_set_callback(subreq, ctdb_tunnel_reply_done, req);
 634
 635         return req;
 636 }
 637
 638 static void ctdb_tunnel_reply_done(struct tevent_req *subreq)
 639 {
 640         struct tevent_req *req = tevent_req_callback_data(
 641                 subreq, struct tevent_req);
 642         int ret;
 643         bool status;
 644
 645         status = comm_write_recv(subreq, &ret);
 646         TALLOC_FREE(subreq);
 647         if (! status) {
 648                 tevent_req_error(req, ret);
 649                 return;
 650         }
 651
 652         tevent_req_done(req);
 653 }
 654
 655 bool ctdb_tunnel_reply_recv(struct tevent_req *req, int *perr)
 656 {
 657         int ret;
 658
 659         if (tevent_req_is_unix_error(req, &ret)) {
 660                 if (perr != NULL) {
 661                         *perr = ret;
 662                 }
 663                 return false;
 664         }
 665
 666         return true;
 667 }
 668
 669 int ctdb_tunnel_reply(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 670                       struct ctdb_tunnel_context *tctx, int destnode,
 671                       uint32_t reqid, struct timeval timeout,
 672                       uint8_t *buf, size_t buflen)
 673 {
 674         struct tevent_req *req;
 675         int ret;
 676         bool status;
 677
 678         req = ctdb_tunnel_reply_send(mem_ctx, ev, tctx, destnode, reqid,
 679                                      timeout, buf, buflen);
 680         if (req == NULL) {
 681                 return ENOMEM;
 682         }
 683
 684         tevent_req_poll(req, ev);
 685
 686         status = ctdb_tunnel_reply_recv(req, &ret);
 687         talloc_free(req);
 688         if (! status) {
 689                 return ret;
 690         }
 691
 692         return 0;
 693 }