source3/lib/ctdb_conn.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3    Samba3 ctdb connection handling
   4    Copyright (C) Volker Lendecke 2012
   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 "includes.h"
  21 #include "lib/util/tevent_unix.h"
  22 #include "ctdb_conn.h"
  23
  24 #ifdef CLUSTER_SUPPORT
  25
  26 #include "lib/async_req/async_sock.h"
  27
  28 struct ctdb_conn {
  29         int fd;
  30         struct tevent_queue *outqueue;
  31 };
  32
  33 struct ctdb_conn_init_state {
  34         struct sockaddr_un addr;
  35         struct ctdb_conn *conn;
  36 };
  37
  38 /*
  39  * use the callbacks of async_connect_send to make sure
  40  * we are connecting to CTDB as root
  41  */
  42 static void before_connect_cb(void *private_data) {
  43         become_root();
  44 }
  45
  46 static void after_connect_cb(void *private_data) {
  47         unbecome_root();
  48 }
  49
  50 static void ctdb_conn_init_done(struct tevent_req *subreq);
  51 static int ctdb_conn_destructor(struct ctdb_conn *conn);
  52
  53 struct tevent_req *ctdb_conn_init_send(TALLOC_CTX *mem_ctx,
  54                                        struct tevent_context *ev,
  55                                        const char *sock)
  56 {
  57         struct tevent_req *req, *subreq;
  58         struct ctdb_conn_init_state *state;
  59
  60         req = tevent_req_create(mem_ctx, &state, struct ctdb_conn_init_state);
  61         if (req == NULL) {
  62                 return NULL;
  63         }
  64
  65         if (!lp_clustering()) {
  66                 tevent_req_error(req, ENOSYS);
  67                 return tevent_req_post(req, ev);
  68         }
  69
  70         if (strlen(sock) >= sizeof(state->addr.sun_path)) {
  71                 tevent_req_error(req, ENAMETOOLONG);
  72                 return tevent_req_post(req, ev);
  73         }
  74
  75         state->conn = talloc(state, struct ctdb_conn);
  76         if (tevent_req_nomem(state->conn, req)) {
  77                 return tevent_req_post(req, ev);
  78         }
  79
  80         state->conn->outqueue = tevent_queue_create(
  81                 state->conn, "ctdb outqueue");
  82         if (tevent_req_nomem(state->conn->outqueue, req)) {
  83                 return tevent_req_post(req, ev);
  84         }
  85
  86         state->conn->fd = socket(AF_UNIX, SOCK_STREAM, 0);
  87         if (state->conn->fd == -1) {
  88                 tevent_req_error(req, errno);
  89                 return tevent_req_post(req, ev);
  90         }
  91         talloc_set_destructor(state->conn, ctdb_conn_destructor);
  92
  93         state->addr.sun_family = AF_UNIX;
  94         strncpy(state->addr.sun_path, sock, sizeof(state->addr.sun_path));
  95
  96         subreq = async_connect_send(state, ev, state->conn->fd,
  97                                     (struct sockaddr *)&state->addr,
  98                                     sizeof(state->addr), before_connect_cb,
  99                                     after_connect_cb, NULL);
 100         if (tevent_req_nomem(subreq, req)) {
 101                 return tevent_req_post(req, ev);
 102         }
 103         tevent_req_set_callback(subreq, ctdb_conn_init_done, req);
 104         return req;
 105 }
 106
 107 static int ctdb_conn_destructor(struct ctdb_conn *c)
 108 {
 109         if (c->fd != -1) {
 110                 close(c->fd);
 111                 c->fd = -1;
 112         }
 113         return 0;
 114 }
 115
 116 static void ctdb_conn_init_done(struct tevent_req *subreq)
 117 {
 118         struct tevent_req *req = tevent_req_callback_data(
 119                 subreq, struct tevent_req);
 120         int ret, err;
 121
 122         ret = async_connect_recv(subreq, &err);
 123         TALLOC_FREE(subreq);
 124         if (ret == -1) {
 125                 tevent_req_error(req, err);
 126                 return;
 127         }
 128         tevent_req_done(req);
 129 }
 130
 131 int ctdb_conn_init_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 132                         struct ctdb_conn **pconn)
 133 {
 134         struct ctdb_conn_init_state *state = tevent_req_data(
 135                 req, struct ctdb_conn_init_state);
 136         int err;
 137
 138         if (tevent_req_is_unix_error(req, &err)) {
 139                 return err;
 140         }
 141         *pconn = talloc_move(mem_ctx, &state->conn);
 142
 143         return 0;
 144 }
 145
 146 struct ctdb_conn_control_state {
 147         struct tevent_context *ev;
 148         struct ctdb_conn *conn;
 149         struct ctdb_req_control req;
 150         struct iovec iov[2];
 151         struct ctdb_reply_control *reply;
 152 };
 153
 154 static void ctdb_conn_control_written(struct tevent_req *subreq);
 155 static void ctdb_conn_control_done(struct tevent_req *subreq);
 156 static ssize_t ctdb_packet_more(uint8_t *buf, size_t buflen, void *p);
 157
 158 struct tevent_req *ctdb_conn_control_send(TALLOC_CTX *mem_ctx,
 159                                           struct tevent_context *ev,
 160                                           struct ctdb_conn *conn,
 161                                           uint32_t vnn, uint32_t opcode,
 162                                           uint64_t srvid, uint32_t flags,
 163                                           uint8_t *data, size_t datalen)
 164 {
 165         struct tevent_req *req, *subreq;
 166         struct ctdb_conn_control_state *state;
 167         struct ctdb_req_header *hdr;
 168
 169         req = tevent_req_create(mem_ctx, &state,
 170                                 struct ctdb_conn_control_state);
 171         if (req == NULL) {
 172                 return NULL;
 173         }
 174         state->ev = ev;
 175         state->conn = conn;
 176
 177         hdr = &state->req.hdr;
 178         hdr->length = offsetof(struct ctdb_req_control, data) + datalen;
 179         hdr->ctdb_magic    = CTDB_MAGIC;
 180         hdr->ctdb_version  = CTDB_VERSION;
 181         hdr->operation     = CTDB_REQ_CONTROL;
 182         hdr->reqid         = 1; /* FIXME */
 183         hdr->destnode      = vnn;
 184         state->req.opcode  = opcode;
 185         state->req.srvid   = srvid;
 186         state->req.datalen = datalen;
 187         state->req.flags   = flags;
 188
 189         state->iov[0].iov_base = &state->req;
 190         state->iov[0].iov_len = offsetof(struct ctdb_req_control, data);
 191         state->iov[1].iov_base = data;
 192         state->iov[1].iov_len = datalen;
 193
 194         subreq = writev_send(state, ev, conn->outqueue, conn->fd, false,
 195                              state->iov, 2);
 196         if (tevent_req_nomem(subreq, req)) {
 197                 return tevent_req_post(req, ev);
 198         }
 199         tevent_req_set_callback(subreq, ctdb_conn_control_written, req);
 200         return req;
 201 }
 202
 203 static void ctdb_conn_control_written(struct tevent_req *subreq)
 204 {
 205         struct tevent_req *req = tevent_req_callback_data(
 206                 subreq, struct tevent_req);
 207         struct ctdb_conn_control_state *state = tevent_req_data(
 208                 req, struct ctdb_conn_control_state);
 209         ssize_t written;
 210         int err;
 211
 212         written = writev_recv(subreq, &err);
 213         TALLOC_FREE(subreq);
 214         if (written == -1) {
 215                 tevent_req_error(req, err);
 216                 return;
 217         }
 218         subreq = read_packet_send(
 219                 state, state->ev, state->conn->fd, sizeof(uint32_t),
 220                 ctdb_packet_more, NULL);
 221         if (tevent_req_nomem(subreq, req)) {
 222                 return;
 223         }
 224         tevent_req_set_callback(subreq, ctdb_conn_control_done, req);
 225 }
 226
 227 static ssize_t ctdb_packet_more(uint8_t *buf, size_t buflen, void *p)
 228 {
 229         uint32_t len;
 230
 231         if (buflen > sizeof(uint32_t)) {
 232                 /* Been here, done */
 233                 return 0;
 234         }
 235         memcpy(&len, buf, sizeof(len));
 236
 237         if (len < sizeof(uint32_t)) {
 238                 return -1;
 239         }
 240
 241         return (len - sizeof(uint32_t));
 242 }
 243
 244 static void ctdb_conn_control_done(struct tevent_req *subreq)
 245 {
 246         struct tevent_req *req = tevent_req_callback_data(
 247                 subreq, struct tevent_req);
 248         struct ctdb_conn_control_state *state = tevent_req_data(
 249                 req, struct ctdb_conn_control_state);
 250         ssize_t nread;
 251         uint8_t *buf;
 252         int err;
 253
 254         nread = read_packet_recv(subreq, state, &buf, &err);
 255         TALLOC_FREE(subreq);
 256         if (nread == -1) {
 257                 tevent_req_error(req, err);
 258                 return;
 259         }
 260         state->reply = (struct ctdb_reply_control *)buf;
 261         tevent_req_done(req);
 262 }
 263
 264 int ctdb_conn_control_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 265                            struct ctdb_reply_control **preply)
 266 {
 267         struct ctdb_conn_control_state *state = tevent_req_data(
 268                 req, struct ctdb_conn_control_state);
 269         int err;
 270
 271         if (tevent_req_is_unix_error(req, &err)) {
 272                 return err;
 273         }
 274         if (preply != NULL) {
 275                 *preply = talloc_move(mem_ctx, &state->reply);
 276         }
 277         return 0;
 278 }
 279
 280 struct ctdb_conn_msg_write_state {
 281         struct ctdb_req_message ctdb_msg;
 282         struct iovec iov[2];
 283 };
 284
 285 static void ctdb_conn_msg_write_done(struct tevent_req *subreq);
 286
 287 struct tevent_req *ctdb_conn_msg_write_send(TALLOC_CTX *mem_ctx,
 288                                             struct tevent_context *ev,
 289                                             struct ctdb_conn *conn,
 290                                             uint32_t vnn, uint64_t srvid,
 291                                             uint8_t *msg, size_t msg_len)
 292 {
 293         struct tevent_req *req, *subreq;
 294         struct ctdb_conn_msg_write_state *state;
 295         struct ctdb_req_header *h;
 296
 297         req = tevent_req_create(mem_ctx, &state,
 298                                 struct ctdb_conn_msg_write_state);
 299         if (req == NULL) {
 300                 return NULL;
 301         }
 302
 303         h = &state->ctdb_msg.hdr;
 304
 305         h->length = offsetof(struct ctdb_req_message, data) + msg_len;
 306         h->ctdb_magic = CTDB_MAGIC;
 307         h->ctdb_version = CTDB_VERSION;
 308         h->generation = 1;
 309         h->operation  = CTDB_REQ_MESSAGE;
 310         h->destnode   = vnn;
 311         h->srcnode    = CTDB_CURRENT_NODE;
 312         h->reqid      = 0;
 313         state->ctdb_msg.srvid   = srvid;
 314         state->ctdb_msg.datalen = msg_len;
 315
 316         state->iov[0].iov_base = &state->ctdb_msg;
 317         state->iov[0].iov_len = offsetof(struct ctdb_req_message, data);
 318         state->iov[1].iov_base = msg;
 319         state->iov[1].iov_len = msg_len;
 320
 321         subreq = writev_send(state, ev, conn->outqueue, conn->fd, false,
 322                              state->iov, 2);
 323         if (tevent_req_nomem(subreq, req)) {
 324                 return tevent_req_post(req, ev);
 325         }
 326         tevent_req_set_callback(subreq, ctdb_conn_msg_write_done, req);
 327         return req;
 328 }
 329
 330 static void ctdb_conn_msg_write_done(struct tevent_req *subreq)
 331 {
 332         struct tevent_req *req = tevent_req_callback_data(
 333                 subreq, struct tevent_req);
 334         ssize_t written;
 335         int err;
 336
 337         written = writev_recv(subreq, &err);
 338         TALLOC_FREE(subreq);
 339         if (written == -1) {
 340                 tevent_req_error(req, err);
 341                 return;
 342         }
 343         tevent_req_done(req);
 344 }
 345
 346 int ctdb_conn_msg_write_recv(struct tevent_req *req)
 347 {
 348         int err;
 349         if (tevent_req_is_unix_error(req, &err)) {
 350                 return err;
 351         }
 352         return 0;
 353 }
 354
 355 struct ctdb_msg_channel {
 356         struct ctdb_conn *conn;
 357 };
 358
 359 struct ctdb_msg_channel_init_state {
 360         struct tevent_context *ev;
 361         struct ctdb_conn *conn;
 362         uint64_t srvid;
 363         struct ctdb_msg_channel *channel;
 364 };
 365
 366 static void ctdb_msg_channel_init_connected(struct tevent_req *subreq);
 367 static void ctdb_msg_channel_init_registered_srvid(struct tevent_req *subreq);
 368
 369 struct tevent_req *ctdb_msg_channel_init_send(
 370         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 371         const char *sock, uint64_t srvid)
 372 {
 373         struct tevent_req *req, *subreq;
 374         struct ctdb_msg_channel_init_state *state;
 375
 376         req = tevent_req_create(mem_ctx, &state,
 377                                 struct ctdb_msg_channel_init_state);
 378         if (req == NULL) {
 379                 return NULL;
 380         }
 381         state->ev = ev;
 382         state->srvid = srvid;
 383
 384         subreq = ctdb_conn_init_send(state, ev, sock);
 385         if (tevent_req_nomem(subreq, req)) {
 386                 return tevent_req_post(req, ev);
 387         }
 388         tevent_req_set_callback(subreq, ctdb_msg_channel_init_connected, req);
 389         return req;
 390 }
 391
 392 static void ctdb_msg_channel_init_connected(struct tevent_req *subreq)
 393 {
 394         struct tevent_req *req = tevent_req_callback_data(
 395                 subreq, struct tevent_req);
 396         struct ctdb_msg_channel_init_state *state = tevent_req_data(
 397                 req, struct ctdb_msg_channel_init_state);
 398         int ret;
 399
 400         ret = ctdb_conn_init_recv(subreq, state, &state->conn);
 401         TALLOC_FREE(subreq);
 402         if (tevent_req_error(req, ret)) {
 403                 return;
 404         }
 405         subreq = ctdb_conn_control_send(state, state->ev, state->conn,
 406                                         CTDB_CURRENT_NODE,
 407                                         CTDB_CONTROL_REGISTER_SRVID,
 408                                         state->srvid, 0, NULL, 0);
 409         if (tevent_req_nomem(subreq, req)) {
 410                 return;
 411         }
 412         tevent_req_set_callback(
 413                 subreq, ctdb_msg_channel_init_registered_srvid, req);
 414 }
 415
 416 static void ctdb_msg_channel_init_registered_srvid(struct tevent_req *subreq)
 417 {
 418         struct tevent_req *req = tevent_req_callback_data(
 419                 subreq, struct tevent_req);
 420         struct ctdb_msg_channel_init_state *state = tevent_req_data(
 421                 req, struct ctdb_msg_channel_init_state);
 422         struct ctdb_reply_control *reply;
 423         int ret;
 424
 425         ret = ctdb_conn_control_recv(subreq, talloc_tos(), &reply);
 426         TALLOC_FREE(subreq);
 427         if (tevent_req_error(req, ret)) {
 428                 return;
 429         }
 430         if (reply->status != 0) {
 431                 tevent_req_error(req, EIO);
 432                 return;
 433         }
 434         state->channel = talloc(state, struct ctdb_msg_channel);
 435         if (tevent_req_nomem(state->channel, req)) {
 436                 return;
 437         }
 438         state->channel->conn = talloc_move(state->channel, &state->conn);
 439         tevent_req_done(req);
 440 }
 441
 442 int ctdb_msg_channel_init_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 443                                struct ctdb_msg_channel **pchannel)
 444 {
 445         struct ctdb_msg_channel_init_state *state = tevent_req_data(
 446                 req, struct ctdb_msg_channel_init_state);
 447         int err;
 448
 449         if (tevent_req_is_unix_error(req, &err)) {
 450                 return err;
 451         }
 452         *pchannel = talloc_move(mem_ctx, &state->channel);
 453         return 0;
 454 }
 455
 456 struct ctdb_msg_read_state {
 457         size_t buflen;
 458         uint8_t *buf;
 459 };
 460
 461 static void ctdb_msg_channel_got_msg(struct tevent_req *subreq);
 462
 463 struct tevent_req *ctdb_msg_read_send(TALLOC_CTX *mem_ctx,
 464                                       struct tevent_context *ev,
 465                                       struct ctdb_msg_channel *channel)
 466 {
 467         struct tevent_req *req, *subreq;
 468         struct ctdb_msg_read_state *state;
 469
 470         req = tevent_req_create(mem_ctx, &state,
 471                                 struct ctdb_msg_read_state);
 472         if (req == NULL) {
 473                 return NULL;
 474         }
 475         subreq = read_packet_send(state, ev, channel->conn->fd,
 476                 sizeof(uint32_t), ctdb_packet_more, NULL);
 477         if (tevent_req_nomem(subreq, req)) {
 478                 return tevent_req_post(req, ev);
 479         }
 480         tevent_req_set_callback(subreq, ctdb_msg_channel_got_msg, req);
 481         return req;
 482 }
 483
 484 static void ctdb_msg_channel_got_msg(struct tevent_req *subreq)
 485 {
 486         struct tevent_req *req = tevent_req_callback_data(
 487                 subreq, struct tevent_req);
 488         struct ctdb_msg_read_state *state = tevent_req_data(
 489                 req, struct ctdb_msg_read_state);
 490         ssize_t nread;
 491         uint8_t *buf;
 492         int err;
 493
 494         nread = read_packet_recv(subreq, state, &buf, &err);
 495         if (nread == -1) {
 496                 tevent_req_error(req, err);
 497                 return;
 498         }
 499         state->buflen = nread;
 500         state->buf = buf;
 501         tevent_req_done(req);
 502 }
 503
 504 int ctdb_msg_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 505                        uint8_t **pmsg, size_t *pmsg_len)
 506 {
 507         struct ctdb_msg_read_state *state = tevent_req_data(
 508                 req, struct ctdb_msg_read_state);
 509         struct ctdb_req_header *hdr;
 510         struct ctdb_req_message *msg;
 511         uint8_t *buf;
 512         int err;
 513
 514         if (tevent_req_is_unix_error(req, &err)) {
 515                 return err;
 516         }
 517
 518         hdr = (struct ctdb_req_header *)state->buf;
 519         if (hdr->length != state->buflen) {
 520                 DEBUG(10, ("Got invalid header length\n"));
 521                 return EIO;
 522         }
 523         if (hdr->operation != CTDB_REQ_MESSAGE) {
 524                 DEBUG(10, ("Expected %d (CTDB_REQ_MESSAGE), got %d\n",
 525                            CTDB_REQ_MESSAGE, (int)hdr->operation));
 526                 return EIO;
 527         }
 528         if (hdr->length < offsetof(struct ctdb_req_message, data)) {
 529                 DEBUG(10, ("Got short msg, len=%d\n", (int)hdr->length));
 530                 return EIO;
 531         }
 532
 533         msg = (struct ctdb_req_message *)hdr;
 534         if (msg->datalen >
 535             hdr->length - offsetof(struct ctdb_req_message, data)) {
 536                 DEBUG(10, ("Got invalid datalen %d\n", (int)msg->datalen));
 537                 return EIO;
 538         }
 539
 540         buf = (uint8_t *)talloc_memdup(mem_ctx, msg->data, msg->datalen);
 541         if (buf == NULL) {
 542                 return ENOMEM;
 543         }
 544         *pmsg = buf;
 545         *pmsg_len = msg->datalen;
 546         return 0;
 547 }
 548
 549 #else
 550
 551 struct dummy_state {
 552         uint8_t dummy;
 553 };
 554
 555 static struct tevent_req *dummy_send(TALLOC_CTX *mem_ctx,
 556                                      struct tevent_context *ev)
 557 {
 558         struct tevent_req *req;
 559         struct dummy_state *state;
 560         req = tevent_req_create(mem_ctx, &state, struct dummy_state);
 561         if (req == NULL) {
 562                 return NULL;
 563         }
 564         tevent_req_done(req);
 565         return tevent_req_post(req, ev);
 566 }
 567
 568 struct tevent_req *ctdb_conn_init_send(TALLOC_CTX *mem_ctx,
 569                                        struct tevent_context *ev,
 570                                        const char *sock)
 571 {
 572         return dummy_send(mem_ctx, ev);
 573 }
 574
 575 int ctdb_conn_init_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 576                         struct ctdb_conn **pconn)
 577 {
 578         return ENOSYS;
 579 }
 580
 581 struct tevent_req *ctdb_conn_msg_write_send(TALLOC_CTX *mem_ctx,
 582                                             struct tevent_context *ev,
 583                                             struct ctdb_conn *conn,
 584                                             uint32_t vnn, uint64_t srvid,
 585                                             uint8_t *msg, size_t msg_len)
 586 {
 587         return dummy_send(mem_ctx, ev);
 588 }
 589
 590 int ctdb_conn_msg_write_recv(struct tevent_req *req)
 591 {
 592         return ENOSYS;
 593 }
 594
 595 struct tevent_req *ctdb_msg_channel_init_send(
 596         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 597         const char *sock, uint64_t srvid)
 598 {
 599         return dummy_send(mem_ctx, ev);
 600 }
 601
 602 int ctdb_msg_channel_init_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 603                                struct ctdb_msg_channel **pchannel)
 604 {
 605         return ENOSYS;
 606 }
 607
 608 struct tevent_req *ctdb_msg_read_send(TALLOC_CTX *mem_ctx,
 609                                       struct tevent_context *ev,
 610                                       struct ctdb_msg_channel *channel)
 611 {
 612         return dummy_send(mem_ctx, ev);
 613 }
 614
 615 int ctdb_msg_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 616                        uint8_t **pmsg, size_t *pmsg_len)
 617 {
 618         return ENOSYS;
 619 }
 620
 621 #endif