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