libctdb/ctdb.c

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