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