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