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