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                 tevent_req_set_endtime(req, ev, timeout);
 461         }
 462
 463         subreq = comm_write_send(state, ev, tctx->client->comm,
 464                                  pkt, pkt_len);
 465         if (tevent_req_nomem(subreq, req)) {
 466                 return tevent_req_post(req, ev);
 467         }
 468         tevent_req_set_callback(subreq, ctdb_tunnel_request_done, req);
 469
 470         return req;
 471 }
 472
 473 static int ctdb_tunnel_request_state_destructor(
 474                         struct ctdb_tunnel_request_state *state)
 475 {
 476         reqid_remove(state->tctx->client->idr, state->reqid);
 477         return 0;
 478 }
 479
 480 static void ctdb_tunnel_request_done(struct tevent_req *subreq)
 481 {
 482         struct tevent_req *req = tevent_req_callback_data(
 483                 subreq, struct tevent_req);
 484         struct ctdb_tunnel_request_state *state = tevent_req_data(
 485                 req, struct ctdb_tunnel_request_state);
 486         int ret;
 487         bool status;
 488
 489         status = comm_write_recv(subreq, &ret);
 490         TALLOC_FREE(subreq);
 491         if (! status) {
 492                 tevent_req_error(req, ret);
 493                 return;
 494         }
 495
 496         if (! state->wait_for_reply) {
 497                 tevent_req_done(req);
 498         }
 499
 500         /* Wait for the reply or timeout */
 501 }
 502
 503 static void ctdb_tunnel_request_reply(struct tevent_req *req,
 504                                       struct ctdb_tunnel_data *tunnel_data)
 505 {
 506         struct ctdb_tunnel_request_state *state = tevent_req_data(
 507                 req, struct ctdb_tunnel_request_state);
 508
 509         if (tunnel_data->reqid != state->reqid) {
 510                 return;
 511         }
 512
 513         state->tunnel = talloc_steal(state, tunnel_data->tunnel);
 514         tevent_req_done(req);
 515 }
 516
 517 bool ctdb_tunnel_request_recv(struct tevent_req *req, int *perr,
 518                               TALLOC_CTX *mem_ctx, uint8_t **buf,
 519                               size_t *buflen)
 520 {
 521         struct ctdb_tunnel_request_state *state = tevent_req_data(
 522                 req, struct ctdb_tunnel_request_state);
 523         int ret;
 524
 525         if (tevent_req_is_unix_error(req, &ret)) {
 526                 if (perr != NULL) {
 527                         *perr = ret;
 528                 }
 529                 return false;
 530         }
 531
 532         if (state->wait_for_reply) {
 533                 if (buf != NULL) {
 534                         *buf = talloc_steal(mem_ctx, state->tunnel->data.dptr);
 535                 }
 536                 if (buflen != NULL) {
 537                         *buflen = state->tunnel->data.dsize;
 538                 }
 539         }
 540
 541         return true;
 542 }
 543
 544 int ctdb_tunnel_request(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 545                         struct ctdb_tunnel_context *tctx, int destnode,
 546                         struct timeval timeout, uint8_t *buf, size_t buflen,
 547                         bool wait_for_reply)
 548 {
 549         struct tevent_req *req;
 550         int ret;
 551         bool status;
 552
 553         req = ctdb_tunnel_request_send(mem_ctx, ev, tctx, destnode,
 554                                        timeout, buf, buflen, wait_for_reply);
 555         if (req == NULL) {
 556                 return ENOMEM;
 557         }
 558
 559         tevent_req_poll(req, ev);
 560
 561         status = ctdb_tunnel_request_recv(req, &ret, NULL, NULL, NULL);
 562         talloc_free(req);
 563         if (! status) {
 564                 return ret;
 565         }
 566
 567         return 0;
 568 }
 569
 570 struct ctdb_tunnel_reply_state {
 571 };
 572
 573 static void ctdb_tunnel_reply_done(struct tevent_req *subreq);
 574
 575 struct tevent_req *ctdb_tunnel_reply_send(TALLOC_CTX *mem_ctx,
 576                                           struct tevent_context *ev,
 577                                           struct ctdb_tunnel_context *tctx,
 578                                           int destnode, uint32_t reqid,
 579                                           struct timeval timeout,
 580                                           uint8_t *buf, size_t buflen)
 581 {
 582         struct tevent_req *req, *subreq;
 583         struct ctdb_tunnel_reply_state *state;
 584         struct ctdb_req_tunnel tunnel;
 585         struct ctdb_req_header h;
 586         uint8_t *pkt;
 587         size_t datalen, pkt_len;
 588         int ret;
 589
 590         req = tevent_req_create(mem_ctx, &state,
 591                                 struct ctdb_tunnel_reply_state);
 592         if (req == NULL) {
 593                 return NULL;
 594         }
 595
 596         tunnel = (struct ctdb_req_tunnel) {
 597                 .tunnel_id = tctx->tunnel_id,
 598                 .flags = CTDB_TUNNEL_FLAG_REPLY,
 599                 .data = (TDB_DATA) {
 600                         .dptr = buf,
 601                         .dsize = buflen,
 602                 },
 603         };
 604
 605         ctdb_req_header_fill(&h, 0, CTDB_REQ_TUNNEL, destnode,
 606                              ctdb_client_pnn(tctx->client), reqid);
 607
 608         datalen = ctdb_req_tunnel_len(&h, &tunnel);
 609         ret = ctdb_allocate_pkt(state, datalen, &pkt, &pkt_len);
 610         if (ret != 0) {
 611                 tevent_req_error(req, ret);
 612                 return tevent_req_post(req, ev);
 613         }
 614
 615         ret = ctdb_req_tunnel_push(&h, &tunnel, pkt, &pkt_len);
 616         if (ret != 0) {
 617                 tevent_req_error(req, ret);
 618                 return tevent_req_post(req, ev);
 619         }
 620
 621         if (!tevent_timeval_is_zero(&timeout)) {
 622                 tevent_req_set_endtime(req, ev, timeout);
 623         }
 624
 625         subreq = comm_write_send(state, ev, tctx->client->comm, pkt, pkt_len);
 626         if (tevent_req_nomem(subreq, req)) {
 627                 return tevent_req_post(req, ev);
 628         }
 629         tevent_req_set_callback(subreq, ctdb_tunnel_reply_done, req);
 630
 631         return req;
 632 }
 633
 634 static void ctdb_tunnel_reply_done(struct tevent_req *subreq)
 635 {
 636         struct tevent_req *req = tevent_req_callback_data(
 637                 subreq, struct tevent_req);
 638         int ret;
 639         bool status;
 640
 641         status = comm_write_recv(subreq, &ret);
 642         TALLOC_FREE(subreq);
 643         if (! status) {
 644                 tevent_req_error(req, ret);
 645                 return;
 646         }
 647
 648         tevent_req_done(req);
 649 }
 650
 651 bool ctdb_tunnel_reply_recv(struct tevent_req *req, int *perr)
 652 {
 653         int ret;
 654
 655         if (tevent_req_is_unix_error(req, &ret)) {
 656                 if (perr != NULL) {
 657                         *perr = ret;
 658                 }
 659                 return false;
 660         }
 661
 662         return true;
 663 }
 664
 665 int ctdb_tunnel_reply(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 666                       struct ctdb_tunnel_context *tctx, int destnode,
 667                       uint32_t reqid, struct timeval timeout,
 668                       uint8_t *buf, size_t buflen)
 669 {
 670         struct tevent_req *req;
 671         int ret;
 672         bool status;
 673
 674         req = ctdb_tunnel_reply_send(mem_ctx, ev, tctx, destnode, reqid,
 675                                      timeout, buf, buflen);
 676         if (req == NULL) {
 677                 return ENOMEM;
 678         }
 679
 680         tevent_req_poll(req, ev);
 681
 682         status = ctdb_tunnel_reply_recv(req, &ret);
 683         talloc_free(req);
 684         if (! status) {
 685                 return ret;
 686         }
 687
 688         return 0;
 689 }