libctdb/ctdb.c

   1 /*
   2    core of libctdb
   3
   4    Copyright (C) Rusty Russell 2010
   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 #include <ctdb.h>
  20 #include <poll.h>
  21 #include <errno.h>
  22 #include <unistd.h>
  23 #include <fcntl.h>
  24 #include <stdlib.h>
  25 #include <sys/socket.h>
  26 #include <sys/un.h>
  27 #include "libctdb_private.h"
  28 #include "io_elem.h"
  29 #include "local_tdb.h"
  30 #include "messages.h"
  31 #include <dlinklist.h>
  32 #include <ctdb_protocol.h>
  33
  34 /* Remove type-safety macros. */
  35 #undef ctdb_attachdb_send
  36 #undef ctdb_readrecordlock_async
  37 #undef ctdb_connect
  38
  39 /* FIXME: Could be in shared util code with rest of ctdb */
  40 static void close_noerr(int fd)
  41 {
  42         int olderr = errno;
  43         close(fd);
  44         errno = olderr;
  45 }
  46
  47 /* FIXME: Could be in shared util code with rest of ctdb */
  48 static void free_noerr(void *p)
  49 {
  50         int olderr = errno;
  51         free(p);
  52         errno = olderr;
  53 }
  54
  55 /* FIXME: Could be in shared util code with rest of ctdb */
  56 static void set_nonblocking(int fd)
  57 {
  58         unsigned v;
  59         v = fcntl(fd, F_GETFL, 0);
  60         fcntl(fd, F_SETFL, v | O_NONBLOCK);
  61 }
  62
  63 /* FIXME: Could be in shared util code with rest of ctdb */
  64 static void set_close_on_exec(int fd)
  65 {
  66         unsigned v;
  67         v = fcntl(fd, F_GETFD, 0);
  68         fcntl(fd, F_SETFD, v | FD_CLOEXEC);
  69 }
  70
  71 static void set_pnn(struct ctdb_connection *ctdb,
  72                     struct ctdb_request *req,
  73                     void *unused)
  74 {
  75         if (ctdb_getpnn_recv(ctdb, req, &ctdb->pnn) != 0) {
  76                 /* FIXME: Report error. */
  77                 ctdb->broken = true;
  78         }
  79         ctdb_request_free(ctdb, req);
  80 }
  81
  82 struct ctdb_connection *ctdb_connect(const char *addr,
  83                                      ctdb_log_fn_t log_fn, void *log_priv)
  84 {
  85         struct ctdb_connection *ctdb;
  86         struct sockaddr_un sun;
  87
  88         ctdb = malloc(sizeof(*ctdb));
  89         if (!ctdb)
  90                 goto fail;
  91         ctdb->outq = NULL;
  92         ctdb->doneq = NULL;
  93         ctdb->in = NULL;
  94         ctdb->message_handlers = NULL;
  95         ctdb->next_id = 0;
  96         ctdb->broken = false;
  97         ctdb->log = log_fn;
  98         ctdb->log_priv = log_priv;
  99
 100         memset(&sun, 0, sizeof(sun));
 101         sun.sun_family = AF_UNIX;
 102         if (!addr)
 103                 addr = CTDB_PATH;
 104         strncpy(sun.sun_path, addr, sizeof(sun.sun_path));
 105         ctdb->fd = socket(AF_UNIX, SOCK_STREAM, 0);
 106         if (ctdb->fd < 0)
 107                 goto free_fail;
 108
 109         set_nonblocking(ctdb->fd);
 110         set_close_on_exec(ctdb->fd);
 111
 112         if (connect(ctdb->fd, (struct sockaddr *)&sun, sizeof(sun)) == -1)
 113                 goto close_fail;
 114
 115         /* Immediately queue a request to get our pnn. */
 116         if (!ctdb_getpnn_send(ctdb, CTDB_CURRENT_NODE, set_pnn, NULL))
 117                 goto close_fail;
 118
 119         return ctdb;
 120
 121 close_fail:
 122         close_noerr(ctdb->fd);
 123 free_fail:
 124         free_noerr(ctdb);
 125 fail:
 126         return NULL;
 127 }
 128
 129 int ctdb_get_fd(struct ctdb_connection *ctdb)
 130 {
 131         return ctdb->fd;
 132 }
 133
 134 int ctdb_which_events(struct ctdb_connection *ctdb)
 135 {
 136         int events = POLLIN;
 137
 138         if (ctdb->outq)
 139                 events |= POLLOUT;
 140         return events;
 141 }
 142
 143 struct ctdb_request *new_ctdb_request(size_t len,
 144                                       ctdb_callback_t cb, void *cbdata)
 145 {
 146         struct ctdb_request *req = malloc(sizeof(*req));
 147         if (!req)
 148                 return NULL;
 149         req->io = new_io_elem(len);
 150         if (!req->io) {
 151                 free(req);
 152                 return NULL;
 153         }
 154         req->hdr.hdr = io_elem_data(req->io, NULL);
 155         req->reply = NULL;
 156         req->callback = cb;
 157         req->priv_data = cbdata;
 158         req->extra = NULL;
 159         req->extra_destructor = NULL;
 160         return req;
 161 }
 162
 163 void ctdb_request_free(struct ctdb_connection *ctdb, struct ctdb_request *req)
 164 {
 165         if (req->extra_destructor) {
 166                 req->extra_destructor(ctdb, req);
 167         }
 168         if (req->reply) {
 169                 free_io_elem(req->reply);
 170         }
 171         free_io_elem(req->io);
 172         free(req);
 173 }
 174
 175 /* Sanity-checking wrapper for reply.
 176  * FIXME: logging! */
 177 static struct ctdb_reply_call *unpack_reply_call(struct ctdb_connection *ctdb,
 178                                                  struct ctdb_request *req,
 179                                                  uint32_t callid)
 180 {
 181         size_t len;
 182         struct ctdb_reply_call *inhdr = io_elem_data(req->reply, &len);
 183
 184         /* ctdbd or our error if this isn't a reply call. */
 185         if (len < sizeof(*inhdr) || inhdr->hdr.operation != CTDB_REPLY_CALL) {
 186                 errno = EIO;
 187                 return NULL;
 188         }
 189
 190         /* Library user error if this isn't a reply to a call. */
 191         if (req->hdr.hdr->operation != CTDB_REQ_CALL
 192             || req->hdr.call->callid != callid) {
 193                 errno = EINVAL;
 194                 return NULL;
 195         }
 196
 197         return inhdr;
 198 }
 199
 200 /* Sanity-checking wrapper for reply.
 201  * FIXME: logging! */
 202 struct ctdb_reply_control *unpack_reply_control(struct ctdb_connection *ctdb,
 203                                                 struct ctdb_request *req,
 204                                                 enum ctdb_controls control)
 205 {
 206         size_t len;
 207         struct ctdb_reply_control *inhdr = io_elem_data(req->reply, &len);
 208
 209         /* Library user error if this isn't a reply to a call. */
 210         if (len < sizeof(*inhdr)
 211             || req->hdr.hdr->operation != CTDB_REQ_CONTROL) {
 212                 errno = EINVAL;
 213                 return NULL;
 214         }
 215
 216         /* ... or if it was a different control from what we expected. */
 217         if (req->hdr.control->opcode != control) {
 218                 errno = EINVAL;
 219                 return NULL;
 220         }
 221
 222         /* ctdbd or our error if this isn't a reply call. */
 223         if (inhdr->hdr.operation != CTDB_REPLY_CONTROL) {
 224                 errno = EIO;
 225                 return NULL;
 226         }
 227
 228         return inhdr;
 229 }
 230
 231 static void handle_incoming(struct ctdb_connection *ctdb, struct io_elem *in)
 232 {
 233         struct ctdb_req_header *hdr;
 234         size_t len;
 235         struct ctdb_request *i;
 236
 237         hdr = io_elem_data(in, &len);
 238         /* FIXME: use len to check packet! */
 239
 240         if (hdr->operation == CTDB_REQ_MESSAGE) {
 241                 deliver_message(ctdb, hdr);
 242                 return;
 243         }
 244
 245         for (i = ctdb->doneq; i; i = i->next) {
 246                 if (i->hdr.hdr->reqid == hdr->reqid) {
 247                         DLIST_REMOVE(ctdb->doneq, i);
 248                         i->reply = in;
 249                         i->callback(ctdb, i, i->priv_data);
 250                         return;
 251                 }
 252         }
 253         /* FIXME: report this error. */
 254         free_io_elem(in);
 255 }
 256
 257 /* Remove "harmless" errors. */
 258 static ssize_t real_error(ssize_t ret)
 259 {
 260         if (ret < 0 && (errno == EINTR || errno == EWOULDBLOCK))
 261                 return 0;
 262         return ret;
 263 }
 264
 265 int ctdb_service(struct ctdb_connection *ctdb, int revents)
 266 {
 267         if (ctdb->broken) {
 268                 return -1;
 269         }
 270
 271         if (revents & POLLOUT) {
 272                 while (ctdb->outq) {
 273                         if (real_error(write_io_elem(ctdb->fd,
 274                                                      ctdb->outq->io)) < 0) {
 275                                 ctdb->broken = true;
 276                                 return -1;
 277                         }
 278                         if (io_elem_finished(ctdb->outq->io)) {
 279                                 struct ctdb_request *done = ctdb->outq;
 280                                 DLIST_REMOVE(ctdb->outq, done);
 281                                 /* We add at the head: any dead ones
 282                                  * sit and end. */
 283                                 DLIST_ADD(ctdb->doneq, done);
 284                         }
 285                 }
 286         }
 287
 288         while (revents & POLLIN) {
 289                 int ret;
 290
 291                 if (!ctdb->in) {
 292                         ctdb->in = new_io_elem(sizeof(struct ctdb_req_header));
 293                         if (!ctdb->in) {
 294                                 ctdb->broken = true;
 295                                 return -1;
 296                         }
 297                 }
 298
 299                 ret = read_io_elem(ctdb->fd, ctdb->in);
 300                 if (real_error(ret) < 0 || ret == 0) {
 301                         /* They closed fd? */
 302                         if (ret == 0)
 303                                 errno = EBADF;
 304                         ctdb->broken = true;
 305                         return -1;
 306                 } else if (ret < 0) {
 307                         /* No progress, stop loop. */
 308                         revents = 0;
 309                 } else if (io_elem_finished(ctdb->in)) {
 310                         handle_incoming(ctdb, ctdb->in);
 311                         ctdb->in = NULL;
 312                 }
 313         }
 314
 315         return 0;
 316 }
 317
 318 /* This is inefficient.  We could pull in idtree.c. */
 319 static bool reqid_used(const struct ctdb_connection *ctdb, uint32_t reqid)
 320 {
 321         struct ctdb_request *i;
 322
 323         for (i = ctdb->outq; i; i = i->next) {
 324                 if (i->hdr.hdr->reqid == reqid) {
 325                         return true;
 326                 }
 327         }
 328         for (i = ctdb->doneq; i; i = i->next) {
 329                 if (i->hdr.hdr->reqid == reqid) {
 330                         return true;
 331                 }
 332         }
 333         return false;
 334 }
 335
 336 uint32_t new_reqid(struct ctdb_connection *ctdb)
 337 {
 338         while (reqid_used(ctdb, ctdb->next_id)) {
 339                 ctdb->next_id++;
 340         }
 341         return ctdb->next_id++;
 342 }
 343
 344 struct ctdb_request *new_ctdb_control_request(struct ctdb_connection *ctdb,
 345                                               uint32_t opcode,
 346                                               uint32_t destnode,
 347                                               const void *extra_data,
 348                                               size_t extra,
 349                                               ctdb_callback_t callback,
 350                                               void *cbdata)
 351 {
 352         struct ctdb_request *req;
 353         struct ctdb_req_control *pkt;
 354
 355         req = new_ctdb_request(offsetof(struct ctdb_req_control, data) + extra, callback, cbdata);
 356         if (!req)
 357                 return NULL;
 358
 359         io_elem_init_req_header(req->io,
 360                                 CTDB_REQ_CONTROL, destnode, new_reqid(ctdb));
 361
 362         pkt = req->hdr.control;
 363         pkt->pad = 0;
 364         pkt->opcode = opcode;
 365         pkt->srvid = 0;
 366         pkt->client_id = 0;
 367         pkt->flags = 0;
 368         pkt->datalen = extra;
 369         memcpy(pkt->data, extra_data, extra);
 370         DLIST_ADD(ctdb->outq, req);
 371         return req;
 372 }
 373
 374 void ctdb_cancel_callback(struct ctdb_connection *ctdb,
 375                           struct ctdb_request *req,
 376                           void *unused)
 377 {
 378         ctdb_request_free(ctdb, req);
 379 }
 380
 381 int ctdb_cancel(struct ctdb_connection *ctdb, struct ctdb_request *req)
 382 {
 383         /* FIXME: If it's not sent, we could just free it right now. */
 384         req->callback = ctdb_cancel_callback;
 385         return 0;
 386 }
 387
 388 struct ctdb_db {
 389         struct ctdb_connection *ctdb;
 390         bool persistent;
 391         uint32_t tdb_flags;
 392         uint32_t id;
 393         struct tdb_context *tdb;
 394
 395         ctdb_callback_t callback;
 396         void *private_data;
 397 };
 398
 399 static void attachdb_getdbpath_done(struct ctdb_connection *ctdb,
 400                                     struct ctdb_request *req,
 401                                     void *_db)
 402 {
 403         struct ctdb_db *db = _db;
 404
 405         /* Do callback on original request. */
 406         db->callback(ctdb, req->extra, db->private_data);
 407 }
 408
 409 struct ctdb_db *ctdb_attachdb_recv(struct ctdb_connection *ctdb,
 410                                    struct ctdb_request *req)
 411 {
 412         struct ctdb_request *dbpath_req = req->extra;
 413         struct ctdb_reply_control *reply;
 414         struct ctdb_db *db = req->priv_data;
 415         uint32_t tdb_flags = db->tdb_flags;
 416
 417         /* Never sent the dbpath request?  We've failed. */
 418         if (!dbpath_req) {
 419                 /* FIXME: Save errno? */
 420                 errno = EINVAL;
 421                 return NULL;
 422         }
 423
 424         reply = unpack_reply_control(ctdb, dbpath_req, CTDB_CONTROL_GETDBPATH);
 425         if (!reply || reply->status != 0) {
 426                 return NULL;
 427         }
 428
 429         tdb_flags = db->persistent ? TDB_DEFAULT : TDB_NOSYNC;
 430         tdb_flags |= TDB_DISALLOW_NESTING;
 431
 432         db->tdb = tdb_open((char *)reply->data, 0, tdb_flags, O_RDWR, 0);
 433         if (db->tdb == NULL) {
 434                 return NULL;
 435         }
 436
 437         /* Finally, separate the db from the request (see destroy_req_db). */
 438         req->priv_data = NULL;
 439         return db;
 440 }
 441
 442 static void attachdb_done(struct ctdb_connection *ctdb,
 443                           struct ctdb_request *req,
 444                           void *_db)
 445 {
 446         struct ctdb_db *db = _db;
 447         struct ctdb_request *req2;
 448         struct ctdb_reply_control *reply;
 449         enum ctdb_controls control = CTDB_CONTROL_DB_ATTACH;
 450
 451         if (db->persistent) {
 452                 control = CTDB_CONTROL_DB_ATTACH_PERSISTENT;
 453         }
 454
 455         reply = unpack_reply_control(ctdb, req, control);
 456         if (!reply || reply->status != 0) {
 457                 /* We failed.  Hand request to user and have them discover it
 458                  * via ctdb_attachdb_recv. */
 459                 db->callback(ctdb, req, db->private_data);
 460                 return;
 461         }
 462         db->id = *(uint32_t *)reply->data;
 463
 464         /* Now we do another call, to get the dbpath. */
 465         req2 = new_ctdb_control_request(db->ctdb, CTDB_CONTROL_GETDBPATH,
 466                                         CTDB_CURRENT_NODE,
 467                                         &db->id, sizeof(db->id),
 468                                         attachdb_getdbpath_done, db);
 469         if (!req2) {
 470                 db->callback(ctdb, req, db->private_data);
 471                 return;
 472         }
 473         req->extra = req2;
 474         req2->extra = req;
 475 }
 476
 477 static void destroy_req_db(struct ctdb_connection *ctdb,
 478                            struct ctdb_request *req)
 479 {
 480         /* Incomplete db is in priv_data. */
 481         free(req->priv_data);
 482         /* second request is chained off this one. */
 483         if (req->extra) {
 484                 ctdb_request_free(ctdb, req->extra);
 485         }
 486 }
 487
 488 struct ctdb_request *
 489 ctdb_attachdb_send(struct ctdb_connection *ctdb,
 490                    const char *name, int persistent, uint32_t tdb_flags,
 491                    ctdb_callback_t callback, void *private_data)
 492 {
 493         struct ctdb_request *req;
 494         struct ctdb_db *db;
 495         uint32_t opcode;
 496
 497         /* FIXME: Search if db already open. */
 498         db = malloc(sizeof(*db));
 499         if (!db) {
 500                 return NULL;
 501         }
 502
 503         if (persistent) {
 504                 opcode = CTDB_CONTROL_DB_ATTACH_PERSISTENT;
 505         } else {
 506                 opcode = CTDB_CONTROL_DB_ATTACH;
 507         }
 508
 509         req = new_ctdb_control_request(ctdb, opcode, CTDB_CURRENT_NODE, name,
 510                                        strlen(name) + 1, attachdb_done, db);
 511         if (!req) {
 512                 free(db);
 513                 return NULL;
 514         }
 515
 516         db->ctdb = ctdb;
 517         db->tdb_flags = tdb_flags;
 518         db->persistent = persistent;
 519         db->callback = callback;
 520         db->private_data = private_data;
 521
 522         req->extra_destructor = destroy_req_db;
 523         /* This is set non-NULL when we succeed, see ctdb_attachdb_recv */
 524         req->extra = NULL;
 525
 526         /* Flags get overloaded into srvid. */
 527         req->hdr.control->srvid = tdb_flags;
 528         return req;
 529 }
 530
 531 struct ctdb_lock {
 532         struct ctdb_db *ctdb_db;
 533         TDB_DATA key;
 534
 535         /* This will always be true by the time user sees this. */
 536         bool held;
 537         struct ctdb_ltdb_header *hdr;
 538
 539         /* For convenience, we stash original callback here. */
 540         ctdb_rrl_callback_t callback;
 541 };
 542
 543 void ctdb_release_lock(struct ctdb_lock *lock)
 544 {
 545         if (lock->held) {
 546                 tdb_chainunlock(lock->ctdb_db->tdb, lock->key);
 547                 lock->held = false;
 548         }
 549 }
 550
 551 static void ctdb_free_lock(struct ctdb_lock *lock)
 552 {
 553         if (lock->held) {
 554                 /* FIXME: report error. Callback never released the lock */
 555                 ctdb_release_lock(lock);
 556         }
 557
 558         free(lock->hdr);
 559         free(lock);
 560 }
 561
 562 /* We keep the lock if local node is the dmaster. */
 563 static bool try_readrecordlock(struct ctdb_lock *lock, TDB_DATA *data)
 564 {
 565         struct ctdb_ltdb_header *hdr;
 566
 567         if (tdb_chainlock(lock->ctdb_db->tdb, lock->key) != 0) {
 568                 return NULL;
 569         }
 570
 571         hdr = ctdb_local_fetch(lock->ctdb_db->tdb, lock->key, data);
 572         if (hdr && hdr->dmaster == lock->ctdb_db->ctdb->pnn) {
 573                 lock->held = true;
 574                 lock->hdr = hdr;
 575                 return true;
 576         }
 577
 578         tdb_chainunlock(lock->ctdb_db->tdb, lock->key);
 579         free(hdr);
 580         return NULL;
 581 }
 582
 583 /* If they shutdown before we hand them the lock, we free it here. */
 584 static void destroy_lock(struct ctdb_connection *ctdb,
 585                          struct ctdb_request *req)
 586 {
 587         ctdb_release_lock(req->extra);
 588         ctdb_free_lock(req->extra);
 589 }
 590
 591 static void readrecordlock_retry(struct ctdb_connection *ctdb,
 592                                  struct ctdb_request *req, void *private)
 593 {
 594         struct ctdb_lock *lock = req->extra;
 595         struct ctdb_reply_call *reply;
 596         TDB_DATA data;
 597
 598         /* OK, we've received reply to noop migration */
 599         reply = unpack_reply_call(ctdb, req, CTDB_NULL_FUNC);
 600         if (!reply || reply->status != 0) {
 601                 lock->callback(lock->ctdb_db, NULL, tdb_null, private);
 602                 ctdb_request_free(ctdb, req); /* Also frees lock. */
 603                 ctdb_free_lock(lock);
 604                 return;
 605         }
 606
 607         /* Can we get lock now? */
 608         if (try_readrecordlock(lock, &data)) {
 609                 /* Now it's their responsibility to free lock & request! */
 610                 req->extra_destructor = NULL;
 611                 lock->callback(lock->ctdb_db, lock, data, private);
 612                 ctdb_free_lock(lock);
 613                 return;
 614         }
 615
 616         /* Retransmit the same request again (we lost race). */
 617         io_elem_reset(req->io);
 618         DLIST_ADD(ctdb->outq, req);
 619 }
 620
 621 bool
 622 ctdb_readrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
 623                           ctdb_rrl_callback_t callback, void *cbdata)
 624 {
 625         struct ctdb_request *req;
 626         struct ctdb_lock *lock;
 627         TDB_DATA data;
 628
 629         /* Setup lock */
 630         lock = malloc(sizeof(*lock) + key.dsize);
 631         if (!lock) {
 632                 return false;
 633         }
 634         lock->key.dptr = (void *)(lock + 1);
 635         memcpy(lock->key.dptr, key.dptr, key.dsize);
 636         lock->key.dsize = key.dsize;
 637         lock->ctdb_db = ctdb_db;
 638         lock->hdr = NULL;
 639         lock->held = false;
 640
 641         /* Fast path. */
 642         if (try_readrecordlock(lock, &data)) {
 643                 callback(ctdb_db, lock, data, cbdata);
 644                 ctdb_free_lock(lock);
 645                 return true;
 646         }
 647
 648         /* Slow path: create request. */
 649         req = new_ctdb_request(offsetof(struct ctdb_req_call, data)
 650                                + key.dsize, readrecordlock_retry, cbdata);
 651         if (!req) {
 652                 ctdb_release_lock(lock);
 653                 ctdb_free_lock(lock);
 654                 return NULL;
 655         }
 656         req->extra = lock;
 657         req->extra_destructor = destroy_lock;
 658         /* We store the original callback in the lock, and use our own. */
 659         lock->callback = callback;
 660
 661         io_elem_init_req_header(req->io, CTDB_REQ_CALL, CTDB_CURRENT_NODE,
 662                                 new_reqid(ctdb_db->ctdb));
 663
 664         req->hdr.call->flags = CTDB_IMMEDIATE_MIGRATION;
 665         req->hdr.call->db_id = ctdb_db->id;
 666         req->hdr.call->callid = CTDB_NULL_FUNC;
 667         req->hdr.call->hopcount = 0;
 668         req->hdr.call->keylen = key.dsize;
 669         req->hdr.call->calldatalen = 0;
 670         memcpy(req->hdr.call->data, key.dptr, key.dsize);
 671         DLIST_ADD(ctdb_db->ctdb->outq, req);
 672         return true;
 673 }
 674
 675 int ctdb_writerecord(struct ctdb_lock *lock, TDB_DATA data)
 676 {
 677         if (lock->ctdb_db->persistent) {
 678                 /* FIXME: Report error. */
 679                 return -1;
 680         }
 681
 682         if (!lock->held) {
 683                 /* FIXME: Report error. */
 684                 return -1;
 685         }
 686
 687         return ctdb_local_store(lock->ctdb_db->tdb, lock->key, lock->hdr,
 688                                 data);
 689 }