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