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