ctdb/server/ctdb_lock.c

   1 /*
   2    ctdb lock handling
   3    provide API to do non-blocking locks for single or all databases
   4
   5    Copyright (C) Amitay Isaacs  2012
   6
   7    This program is free software; you can redistribute it and/or modify
   8    it under the terms of the GNU General Public License as published by
   9    the Free Software Foundation; either version 3 of the License, or
  10    (at your option) any later version.
  11
  12    This program is distributed in the hope that it will be useful,
  13    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15    GNU General Public License for more details.
  16
  17    You should have received a copy of the GNU General Public License
  18    along with this program; if not, see <http://www.gnu.org/licenses/>.
  19 */
  20 #include "replace.h"
  21 #include "system/filesys.h"
  22 #include "system/network.h"
  23
  24 #include <talloc.h>
  25 #include <tevent.h>
  26
  27 #include "lib/tdb_wrap/tdb_wrap.h"
  28 #include "lib/util/dlinklist.h"
  29 #include "lib/util/debug.h"
  30 #include "lib/util/samba_util.h"
  31 #include "lib/util/sys_rw.h"
  32
  33 #include "ctdb_private.h"
  34
  35 #include "common/common.h"
  36 #include "common/logging.h"
  37
  38 /*
  39  * Non-blocking Locking API
  40  *
  41  * 1. Create a child process to do blocking locks.
  42  * 2. Once the locks are obtained, signal parent process via fd.
  43  * 3. Invoke registered callback routine with locking status.
  44  * 4. If the child process cannot get locks within certain time,
  45  *    execute an external script to debug.
  46  *
  47  * ctdb_lock_record()      - get a lock on a record
  48  * ctdb_lock_db()          - get a lock on a DB
  49  *
  50  *  auto_mark              - whether to mark/unmark DBs in before/after callback
  51  *                           = false is used for freezing databases for
  52  *                           recovery since the recovery cannot start till
  53  *                           databases are locked on all the nodes.
  54  *                           = true is used for record locks.
  55  */
  56
  57 enum lock_type {
  58         LOCK_RECORD,
  59         LOCK_DB,
  60 };
  61
  62 static const char * const lock_type_str[] = {
  63         "lock_record",
  64         "lock_db",
  65 };
  66
  67 struct lock_request;
  68
  69 /* lock_context is the common part for a lock request */
  70 struct lock_context {
  71         struct lock_context *next, *prev;
  72         enum lock_type type;
  73         struct ctdb_context *ctdb;
  74         struct ctdb_db_context *ctdb_db;
  75         TDB_DATA key;
  76         uint32_t priority;
  77         bool auto_mark;
  78         struct lock_request *request;
  79         pid_t child;
  80         int fd[2];
  81         struct tevent_fd *tfd;
  82         struct tevent_timer *ttimer;
  83         struct timeval start_time;
  84         uint32_t key_hash;
  85         bool can_schedule;
  86 };
  87
  88 /* lock_request is the client specific part for a lock request */
  89 struct lock_request {
  90         struct lock_context *lctx;
  91         void (*callback)(void *, bool);
  92         void *private_data;
  93 };
  94
  95
  96 int ctdb_db_iterator(struct ctdb_context *ctdb, ctdb_db_handler_t handler,
  97                      void *private_data)
  98 {
  99         struct ctdb_db_context *ctdb_db;
 100         int ret;
 101
 102         for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
 103                 ret = handler(ctdb_db, private_data);
 104                 if (ret != 0) {
 105                         return -1;
 106                 }
 107         }
 108
 109         return 0;
 110 }
 111
 112 /*
 113  * lock all databases - mark only
 114  */
 115 static int db_lock_mark_handler(struct ctdb_db_context *ctdb_db,
 116                                 void *private_data)
 117 {
 118         int tdb_transaction_write_lock_mark(struct tdb_context *);
 119
 120         DEBUG(DEBUG_INFO, ("marking locked database %s\n", ctdb_db->db_name));
 121
 122         if (tdb_transaction_write_lock_mark(ctdb_db->ltdb->tdb) != 0) {
 123                 DEBUG(DEBUG_ERR, ("Failed to mark (transaction lock) database %s\n",
 124                                   ctdb_db->db_name));
 125                 return -1;
 126         }
 127
 128         if (tdb_lockall_mark(ctdb_db->ltdb->tdb) != 0) {
 129                 DEBUG(DEBUG_ERR, ("Failed to mark (all lock) database %s\n",
 130                                   ctdb_db->db_name));
 131                 return -1;
 132         }
 133
 134         return 0;
 135 }
 136
 137 int ctdb_lockdb_mark(struct ctdb_db_context *ctdb_db)
 138 {
 139         if (!ctdb_db_frozen(ctdb_db)) {
 140                 DEBUG(DEBUG_ERR,
 141                       ("Attempt to mark database locked when not frozen\n"));
 142                 return -1;
 143         }
 144
 145         return db_lock_mark_handler(ctdb_db, NULL);
 146 }
 147
 148 /*
 149  * lock all databases - unmark only
 150  */
 151 static int db_lock_unmark_handler(struct ctdb_db_context *ctdb_db,
 152                                   void *private_data)
 153 {
 154         int tdb_transaction_write_lock_unmark(struct tdb_context *);
 155
 156         DEBUG(DEBUG_INFO, ("unmarking locked database %s\n", ctdb_db->db_name));
 157
 158         if (tdb_transaction_write_lock_unmark(ctdb_db->ltdb->tdb) != 0) {
 159                 DEBUG(DEBUG_ERR, ("Failed to unmark (transaction lock) database %s\n",
 160                                   ctdb_db->db_name));
 161                 return -1;
 162         }
 163
 164         if (tdb_lockall_unmark(ctdb_db->ltdb->tdb) != 0) {
 165                 DEBUG(DEBUG_ERR, ("Failed to unmark (all lock) database %s\n",
 166                                   ctdb_db->db_name));
 167                 return -1;
 168         }
 169
 170         return 0;
 171 }
 172
 173 int ctdb_lockdb_unmark(struct ctdb_db_context *ctdb_db)
 174 {
 175         if (!ctdb_db_frozen(ctdb_db)) {
 176                 DEBUG(DEBUG_ERR,
 177                       ("Attempt to unmark database locked when not frozen\n"));
 178                 return -1;
 179         }
 180
 181         return db_lock_unmark_handler(ctdb_db, NULL);
 182 }
 183
 184 static void ctdb_lock_schedule(struct ctdb_context *ctdb);
 185
 186 /*
 187  * Destructor to kill the child locking process
 188  */
 189 static int ctdb_lock_context_destructor(struct lock_context *lock_ctx)
 190 {
 191         if (lock_ctx->request) {
 192                 lock_ctx->request->lctx = NULL;
 193         }
 194         if (lock_ctx->child > 0) {
 195                 ctdb_kill(lock_ctx->ctdb, lock_ctx->child, SIGTERM);
 196                 if (lock_ctx->type == LOCK_RECORD) {
 197                         DLIST_REMOVE(lock_ctx->ctdb_db->lock_current, lock_ctx);
 198                 } else {
 199                         DLIST_REMOVE(lock_ctx->ctdb->lock_current, lock_ctx);
 200                 }
 201                 if (lock_ctx->ctdb_db->lock_num_current == 0) {
 202                         ctdb_fatal(NULL, "Lock count is 0 before decrement\n");
 203                 }
 204                 lock_ctx->ctdb_db->lock_num_current--;
 205                 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_current);
 206                 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
 207         } else {
 208                 if (lock_ctx->type == LOCK_RECORD) {
 209                         DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
 210                 } else {
 211                         DLIST_REMOVE(lock_ctx->ctdb->lock_pending, lock_ctx);
 212                 }
 213                 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
 214                 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 215         }
 216
 217         ctdb_lock_schedule(lock_ctx->ctdb);
 218
 219         return 0;
 220 }
 221
 222
 223 /*
 224  * Destructor to remove lock request
 225  */
 226 static int ctdb_lock_request_destructor(struct lock_request *lock_request)
 227 {
 228         if (lock_request->lctx == NULL) {
 229                 return 0;
 230         }
 231
 232         lock_request->lctx->request = NULL;
 233         TALLOC_FREE(lock_request->lctx);
 234
 235         return 0;
 236 }
 237
 238 /*
 239  * Process all the callbacks waiting for lock
 240  *
 241  * If lock has failed, callback is executed with locked=false
 242  */
 243 static void process_callbacks(struct lock_context *lock_ctx, bool locked)
 244 {
 245         struct lock_request *request;
 246         bool auto_mark = lock_ctx->auto_mark;
 247
 248         if (auto_mark && locked) {
 249                 switch (lock_ctx->type) {
 250                 case LOCK_RECORD:
 251                         tdb_chainlock_mark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
 252                         break;
 253
 254                 case LOCK_DB:
 255                         (void)ctdb_lockdb_mark(lock_ctx->ctdb_db);
 256                         break;
 257                 }
 258         }
 259
 260         request = lock_ctx->request;
 261         if (auto_mark) {
 262                 /* Since request may be freed in the callback, unset the lock
 263                  * context, so request destructor will not free lock context.
 264                  */
 265                 request->lctx = NULL;
 266         }
 267
 268         /* Since request may be freed in the callback, unset the request */
 269         lock_ctx->request = NULL;
 270
 271         request->callback(request->private_data, locked);
 272
 273         if (!auto_mark) {
 274                 return;
 275         }
 276
 277         if (locked) {
 278                 switch (lock_ctx->type) {
 279                 case LOCK_RECORD:
 280                         tdb_chainlock_unmark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
 281                         break;
 282
 283                 case LOCK_DB:
 284                         ctdb_lockdb_unmark(lock_ctx->ctdb_db);
 285                         break;
 286                 }
 287         }
 288
 289         talloc_free(lock_ctx);
 290 }
 291
 292
 293 static int lock_bucket_id(double t)
 294 {
 295         double ms = 1.e-3, s = 1;
 296         int id;
 297
 298         if (t < 1*ms) {
 299                 id = 0;
 300         } else if (t < 10*ms) {
 301                 id = 1;
 302         } else if (t < 100*ms) {
 303                 id = 2;
 304         } else if (t < 1*s) {
 305                 id = 3;
 306         } else if (t < 2*s) {
 307                 id = 4;
 308         } else if (t < 4*s) {
 309                 id = 5;
 310         } else if (t < 8*s) {
 311                 id = 6;
 312         } else if (t < 16*s) {
 313                 id = 7;
 314         } else if (t < 32*s) {
 315                 id = 8;
 316         } else if (t < 64*s) {
 317                 id = 9;
 318         } else {
 319                 id = 10;
 320         }
 321
 322         return id;
 323 }
 324
 325 /*
 326  * Callback routine when the required locks are obtained.
 327  * Called from parent context
 328  */
 329 static void ctdb_lock_handler(struct tevent_context *ev,
 330                             struct tevent_fd *tfd,
 331                             uint16_t flags,
 332                             void *private_data)
 333 {
 334         struct lock_context *lock_ctx;
 335         char c;
 336         bool locked;
 337         double t;
 338         int id;
 339
 340         lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
 341
 342         /* cancel the timeout event */
 343         TALLOC_FREE(lock_ctx->ttimer);
 344
 345         t = timeval_elapsed(&lock_ctx->start_time);
 346         id = lock_bucket_id(t);
 347
 348         /* Read the status from the child process */
 349         if (sys_read(lock_ctx->fd[0], &c, 1) != 1) {
 350                 locked = false;
 351         } else {
 352                 locked = (c == 0 ? true : false);
 353         }
 354
 355         /* Update statistics */
 356         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_calls);
 357         CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_calls);
 358
 359         if (locked) {
 360                 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.buckets[id]);
 361                 CTDB_UPDATE_LATENCY(lock_ctx->ctdb, lock_ctx->ctdb_db,
 362                                     lock_type_str[lock_ctx->type], locks.latency,
 363                                     lock_ctx->start_time);
 364
 365                 CTDB_UPDATE_DB_LATENCY(lock_ctx->ctdb_db, lock_type_str[lock_ctx->type], locks.latency, t);
 366                 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.buckets[id]);
 367         } else {
 368                 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_failed);
 369                 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_failed);
 370         }
 371
 372         process_callbacks(lock_ctx, locked);
 373 }
 374
 375 struct lock_log_entry {
 376         struct db_hash_context *lock_log;
 377         TDB_DATA key;
 378         unsigned long log_sec;
 379         struct tevent_timer *timer;
 380 };
 381
 382 static int lock_log_fetch_parser(uint8_t *keybuf, size_t keylen,
 383                                  uint8_t *databuf, size_t datalen,
 384                                  void *private_data)
 385 {
 386         struct lock_log_entry **entry =
 387                 (struct lock_log_entry **)private_data;
 388
 389         if (datalen != sizeof(struct lock_log_entry *)) {
 390                 return EINVAL;
 391         }
 392
 393         *entry = talloc_get_type_abort(*(void **)databuf,
 394                                        struct lock_log_entry);
 395         return 0;
 396 }
 397
 398 static void lock_log_cleanup(struct tevent_context *ev,
 399                              struct tevent_timer *ttimer,
 400                              struct timeval current_time,
 401                              void *private_data)
 402 {
 403         struct lock_log_entry *entry = talloc_get_type_abort(
 404                 private_data, struct lock_log_entry);
 405         int ret;
 406
 407         entry->timer = NULL;
 408
 409         ret = db_hash_delete(entry->lock_log, entry->key.dptr,
 410                              entry->key.dsize);
 411         if (ret != 0) {
 412                 return;
 413         }
 414         talloc_free(entry);
 415 }
 416
 417 static bool lock_log_skip(struct tevent_context *ev,
 418                           struct db_hash_context *lock_log,
 419                           TDB_DATA key, unsigned long elapsed_sec)
 420 {
 421         struct lock_log_entry *entry = NULL;
 422         int ret;
 423
 424         ret = db_hash_fetch(lock_log, key.dptr, key.dsize,
 425                             lock_log_fetch_parser, &entry);
 426         if (ret == ENOENT) {
 427
 428                 entry = talloc_zero(lock_log, struct lock_log_entry);
 429                 if (entry == NULL) {
 430                         goto fail;
 431                 }
 432
 433                 entry->lock_log = lock_log;
 434
 435                 entry->key.dptr = talloc_memdup(entry, key.dptr, key.dsize);
 436                 if (entry->key.dptr == NULL) {
 437                         talloc_free(entry);
 438                         goto fail;
 439                 }
 440                 entry->key.dsize = key.dsize;
 441
 442                 entry->log_sec = elapsed_sec;
 443                 entry->timer = tevent_add_timer(ev, entry,
 444                                                 timeval_current_ofs(30, 0),
 445                                                 lock_log_cleanup, entry);
 446                 if (entry->timer == NULL) {
 447                         talloc_free(entry);
 448                         goto fail;
 449                 }
 450
 451                 ret = db_hash_add(lock_log, key.dptr, key.dsize,
 452                                   (uint8_t *)&entry,
 453                                   sizeof(struct lock_log_entry *));
 454                 if (ret != 0) {
 455                         talloc_free(entry);
 456                         goto fail;
 457                 }
 458
 459                 return false;
 460
 461         } else if (ret == EINVAL) {
 462
 463                 ret = db_hash_delete(lock_log, key.dptr, key.dsize);
 464                 if (ret != 0) {
 465                         goto fail;
 466                 }
 467
 468                 return false;
 469
 470         } else if (ret == 0) {
 471
 472                 if (elapsed_sec <= entry->log_sec) {
 473                         return true;
 474                 }
 475
 476                 entry->log_sec = elapsed_sec;
 477
 478                 TALLOC_FREE(entry->timer);
 479                 entry->timer = tevent_add_timer(ev, entry,
 480                                                 timeval_current_ofs(30, 0),
 481                                                 lock_log_cleanup, entry);
 482                 if (entry->timer == NULL) {
 483                         ret = db_hash_delete(lock_log, key.dptr, key.dsize);
 484                         if (ret != 0) {
 485                                 goto fail;
 486                         }
 487                         talloc_free(entry);
 488                 }
 489
 490                 return false;
 491         }
 492
 493
 494 fail:
 495         return false;
 496
 497 }
 498
 499 static char **debug_locks_args(TALLOC_CTX *mem_ctx,
 500                                struct lock_context *lock_ctx)
 501 {
 502         int tdb_flags = tdb_get_flags(lock_ctx->ctdb_db->ltdb->tdb);
 503         char **args = str_list_make_empty(mem_ctx);
 504
 505         str_list_add_printf(&args, "debug_locks");
 506         str_list_add_printf(&args, "%d", lock_ctx->child);
 507         str_list_add_printf(&args,
 508                             "%s",
 509                             (lock_ctx->type == LOCK_RECORD) ? "RECORD" : "DB");
 510         str_list_add_printf(&args, "%s", lock_ctx->ctdb_db->db_path);
 511
 512         str_list_add_printf(&args,
 513                             (tdb_flags & TDB_MUTEX_LOCKING) ? "MUTEX"
 514                                                             : "FCNTL");
 515         return args;
 516 }
 517
 518 /*
 519  * Callback routine when required locks are not obtained within timeout
 520  * Called from parent context
 521  */
 522 static void ctdb_lock_timeout_handler(struct tevent_context *ev,
 523                                     struct tevent_timer *ttimer,
 524                                     struct timeval current_time,
 525                                     void *private_data)
 526 {
 527         static char debug_locks[PATH_MAX+1] = "";
 528         struct lock_context *lock_ctx;
 529         struct ctdb_context *ctdb;
 530         pid_t pid;
 531         double elapsed_time;
 532         bool skip;
 533         char *keystr;
 534         char **args;
 535         bool ok;
 536
 537         lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
 538         ctdb = lock_ctx->ctdb;
 539
 540         elapsed_time = timeval_elapsed(&lock_ctx->start_time);
 541
 542         /* For database locks, always log */
 543         if (lock_ctx->type == LOCK_DB) {
 544                 DEBUG(DEBUG_WARNING,
 545                       ("Unable to get DB lock on database %s for "
 546                        "%.0lf seconds\n",
 547                        lock_ctx->ctdb_db->db_name, elapsed_time));
 548                 goto lock_debug;
 549         }
 550
 551         /* For record locks, check if we have already logged */
 552         skip = lock_log_skip(ev, lock_ctx->ctdb_db->lock_log,
 553                              lock_ctx->key, (unsigned long)elapsed_time);
 554         if (skip) {
 555                 goto skip_lock_debug;
 556         }
 557
 558         keystr = hex_encode_talloc(lock_ctx, lock_ctx->key.dptr,
 559                                    lock_ctx->key.dsize);
 560         DEBUG(DEBUG_WARNING,
 561               ("Unable to get RECORD lock on database %s for %.0lf seconds"
 562                " (key %s)\n",
 563                lock_ctx->ctdb_db->db_name, elapsed_time,
 564                keystr ? keystr : ""));
 565         TALLOC_FREE(keystr);
 566
 567         /* If a node stopped/banned, don't spam the logs */
 568         if (ctdb->nodes[ctdb->pnn]->flags & NODE_FLAGS_INACTIVE) {
 569                 goto skip_lock_debug;
 570         }
 571
 572 lock_debug:
 573
 574         ok = ctdb_set_helper("lock debugging helper",
 575                              debug_locks,
 576                              sizeof(debug_locks),
 577                              "CTDB_DEBUG_LOCKS",
 578                              getenv("CTDB_BASE"),
 579                              "debug_locks.sh");
 580         if (!ok) {
 581                 DBG_WARNING("Unable to setup lock debugging\n");
 582                 goto skip_lock_debug;
 583         }
 584
 585         args = debug_locks_args(lock_ctx, lock_ctx);
 586         if (args != NULL) {
 587                 pid = vfork();
 588                 if (pid == 0) {
 589                         execvp(debug_locks, args);
 590                         _exit(0);
 591                 }
 592                 talloc_free(args);
 593                 ctdb_track_child(ctdb, pid);
 594         } else {
 595                 D_WARNING("No memory for debug locks args\n");
 596         }
 597
 598 skip_lock_debug:
 599
 600         /* reset the timeout timer */
 601         // talloc_free(lock_ctx->ttimer);
 602         lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
 603                                             lock_ctx,
 604                                             timeval_current_ofs(10, 0),
 605                                             ctdb_lock_timeout_handler,
 606                                             (void *)lock_ctx);
 607 }
 608
 609 static char **lock_helper_args(TALLOC_CTX *mem_ctx,
 610                                struct lock_context *lock_ctx,
 611                                int fd)
 612 {
 613         int tdb_flags = tdb_get_flags(lock_ctx->ctdb_db->ltdb->tdb);
 614         char **args = str_list_make_empty(mem_ctx);
 615
 616         str_list_add_printf(&args, "%d", getpid());
 617         str_list_add_printf(&args, "%d", fd);
 618         str_list_add_printf(&args,
 619                             "%s",
 620                             (lock_ctx->type == LOCK_RECORD) ? "RECORD" : "DB");
 621
 622         str_list_add_printf(&args, "%s", lock_ctx->ctdb_db->db_path);
 623         str_list_add_printf(&args, "0x%x", tdb_flags);
 624
 625         if (lock_ctx->type == LOCK_RECORD) {
 626                 if (lock_ctx->key.dsize == 0) {
 627                         str_list_add_printf(&args, "NULL");
 628                 } else {
 629                         char *hex = hex_encode_talloc(mem_ctx,
 630                                                       lock_ctx->key.dptr,
 631                                                       lock_ctx->key.dsize);
 632                         if (hex == NULL) {
 633                                 TALLOC_FREE(args);
 634                                 return NULL;
 635                         }
 636                         str_list_add_printf(&args, "%s", hex);
 637                         TALLOC_FREE(hex);
 638                 }
 639         }
 640
 641         return args;
 642 }
 643
 644 /*
 645  * Find a lock request that can be scheduled
 646  */
 647 static struct lock_context *ctdb_find_lock_context(struct ctdb_context *ctdb)
 648 {
 649         struct lock_context *lock_ctx, *next_ctx;
 650         struct ctdb_db_context *ctdb_db;
 651
 652         /* First check if there are database lock requests */
 653
 654         for (lock_ctx = ctdb->lock_pending; lock_ctx != NULL;
 655              lock_ctx = next_ctx) {
 656
 657                 if (lock_ctx->request != NULL) {
 658                         /* Found a lock context with a request */
 659                         return lock_ctx;
 660                 }
 661
 662                 next_ctx = lock_ctx->next;
 663
 664                 DEBUG(DEBUG_INFO, ("Removing lock context without lock "
 665                                    "request\n"));
 666                 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
 667                 CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
 668                 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 669                 talloc_free(lock_ctx);
 670         }
 671
 672         /* Next check database queues */
 673         for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
 674                 if (ctdb_db->lock_num_current ==
 675                     ctdb->tunable.lock_processes_per_db) {
 676                         continue;
 677                 }
 678
 679                 for (lock_ctx = ctdb_db->lock_pending; lock_ctx != NULL;
 680                      lock_ctx = next_ctx) {
 681
 682                         next_ctx = lock_ctx->next;
 683
 684                         if (lock_ctx->request != NULL) {
 685                                 return lock_ctx;
 686                         }
 687
 688                         DEBUG(DEBUG_INFO, ("Removing lock context without "
 689                                            "lock request\n"));
 690                         DLIST_REMOVE(ctdb_db->lock_pending, lock_ctx);
 691                         CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
 692                         CTDB_DECREMENT_DB_STAT(ctdb_db, locks.num_pending);
 693                         talloc_free(lock_ctx);
 694                 }
 695         }
 696
 697         return NULL;
 698 }
 699
 700 /*
 701  * Schedule a new lock child process
 702  * Set up callback handler and timeout handler
 703  */
 704 static void ctdb_lock_schedule(struct ctdb_context *ctdb)
 705 {
 706         struct lock_context *lock_ctx;
 707         int ret;
 708         TALLOC_CTX *tmp_ctx;
 709         static char prog[PATH_MAX+1] = "";
 710         char **args = NULL;
 711
 712         if (!ctdb_set_helper("lock helper",
 713                              prog, sizeof(prog),
 714                              "CTDB_LOCK_HELPER",
 715                              CTDB_HELPER_BINDIR, "ctdb_lock_helper")) {
 716                 ctdb_die(ctdb, __location__
 717                          " Unable to set lock helper\n");
 718         }
 719
 720         /* Find a lock context with requests */
 721         lock_ctx = ctdb_find_lock_context(ctdb);
 722         if (lock_ctx == NULL) {
 723                 return;
 724         }
 725
 726         lock_ctx->child = -1;
 727         ret = pipe(lock_ctx->fd);
 728         if (ret != 0) {
 729                 DEBUG(DEBUG_ERR, ("Failed to create pipe in ctdb_lock_schedule\n"));
 730                 return;
 731         }
 732
 733         set_close_on_exec(lock_ctx->fd[0]);
 734
 735         /* Create data for child process */
 736         tmp_ctx = talloc_new(lock_ctx);
 737         if (tmp_ctx == NULL) {
 738                 DEBUG(DEBUG_ERR, ("Failed to allocate memory for helper args\n"));
 739                 close(lock_ctx->fd[0]);
 740                 close(lock_ctx->fd[1]);
 741                 return;
 742         }
 743
 744         if (! ctdb->do_setsched) {
 745                 ret = setenv("CTDB_NOSETSCHED", "1", 1);
 746                 if (ret != 0) {
 747                         DEBUG(DEBUG_WARNING,
 748                               ("Failed to set CTDB_NOSETSCHED variable\n"));
 749                 }
 750         }
 751
 752         /* Create arguments for lock helper */
 753         args = lock_helper_args(tmp_ctx, lock_ctx, lock_ctx->fd[1]);
 754         if (args == NULL) {
 755                 DEBUG(DEBUG_ERR, ("Failed to create lock helper args\n"));
 756                 close(lock_ctx->fd[0]);
 757                 close(lock_ctx->fd[1]);
 758                 talloc_free(tmp_ctx);
 759                 return;
 760         }
 761
 762         lock_ctx->child = ctdb_vfork_exec(lock_ctx,
 763                                           ctdb,
 764                                           prog,
 765                                           talloc_array_length(args),
 766                                           (const char *const *)args);
 767         if (lock_ctx->child == -1) {
 768                 DEBUG(DEBUG_ERR, ("Failed to create a child in ctdb_lock_schedule\n"));
 769                 close(lock_ctx->fd[0]);
 770                 close(lock_ctx->fd[1]);
 771                 talloc_free(tmp_ctx);
 772                 return;
 773         }
 774
 775         /* Parent process */
 776         close(lock_ctx->fd[1]);
 777
 778         talloc_free(tmp_ctx);
 779
 780         /* Set up timeout handler */
 781         lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
 782                                             lock_ctx,
 783                                             timeval_current_ofs(10, 0),
 784                                             ctdb_lock_timeout_handler,
 785                                             (void *)lock_ctx);
 786         if (lock_ctx->ttimer == NULL) {
 787                 ctdb_kill(ctdb, lock_ctx->child, SIGTERM);
 788                 lock_ctx->child = -1;
 789                 close(lock_ctx->fd[0]);
 790                 return;
 791         }
 792
 793         /* Set up callback */
 794         lock_ctx->tfd = tevent_add_fd(ctdb->ev,
 795                                       lock_ctx,
 796                                       lock_ctx->fd[0],
 797                                       TEVENT_FD_READ,
 798                                       ctdb_lock_handler,
 799                                       (void *)lock_ctx);
 800         if (lock_ctx->tfd == NULL) {
 801                 TALLOC_FREE(lock_ctx->ttimer);
 802                 ctdb_kill(ctdb, lock_ctx->child, SIGTERM);
 803                 lock_ctx->child = -1;
 804                 close(lock_ctx->fd[0]);
 805                 return;
 806         }
 807         tevent_fd_set_auto_close(lock_ctx->tfd);
 808
 809         /* Move the context from pending to current */
 810         if (lock_ctx->type == LOCK_RECORD) {
 811                 DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
 812                 DLIST_ADD_END(lock_ctx->ctdb_db->lock_current, lock_ctx);
 813         } else {
 814                 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
 815                 DLIST_ADD_END(ctdb->lock_current, lock_ctx);
 816         }
 817         CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
 818         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_current);
 819         lock_ctx->ctdb_db->lock_num_current++;
 820         CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 821         CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
 822 }
 823
 824
 825 /*
 826  * Lock record / db depending on type
 827  */
 828 static struct lock_request *ctdb_lock_internal(TALLOC_CTX *mem_ctx,
 829                                                struct ctdb_context *ctdb,
 830                                                struct ctdb_db_context *ctdb_db,
 831                                                TDB_DATA key,
 832                                                uint32_t priority,
 833                                                void (*callback)(void *, bool),
 834                                                void *private_data,
 835                                                enum lock_type type,
 836                                                bool auto_mark)
 837 {
 838         struct lock_context *lock_ctx = NULL;
 839         struct lock_request *request;
 840
 841         if (callback == NULL) {
 842                 DEBUG(DEBUG_WARNING, ("No callback function specified, not locking\n"));
 843                 return NULL;
 844         }
 845
 846         lock_ctx = talloc_zero(ctdb, struct lock_context);
 847         if (lock_ctx == NULL) {
 848                 DEBUG(DEBUG_ERR, ("Failed to create a new lock context\n"));
 849                 return NULL;
 850         }
 851
 852         if ((request = talloc_zero(mem_ctx, struct lock_request)) == NULL) {
 853                 talloc_free(lock_ctx);
 854                 return NULL;
 855         }
 856
 857         lock_ctx->type = type;
 858         lock_ctx->ctdb = ctdb;
 859         lock_ctx->ctdb_db = ctdb_db;
 860         lock_ctx->key.dsize = key.dsize;
 861         if (key.dsize > 0) {
 862                 lock_ctx->key.dptr = talloc_memdup(lock_ctx, key.dptr, key.dsize);
 863                 if (lock_ctx->key.dptr == NULL) {
 864                         DEBUG(DEBUG_ERR, (__location__ "Memory allocation error\n"));
 865                         talloc_free(lock_ctx);
 866                         talloc_free(request);
 867                         return NULL;
 868                 }
 869                 lock_ctx->key_hash = ctdb_hash(&key);
 870         } else {
 871                 lock_ctx->key.dptr = NULL;
 872         }
 873         lock_ctx->priority = priority;
 874         lock_ctx->auto_mark = auto_mark;
 875
 876         lock_ctx->request = request;
 877         lock_ctx->child = -1;
 878
 879         /* Non-record locks are required by recovery and should be scheduled
 880          * immediately, so keep them at the head of the pending queue.
 881          */
 882         if (lock_ctx->type == LOCK_RECORD) {
 883                 DLIST_ADD_END(ctdb_db->lock_pending, lock_ctx);
 884         } else {
 885                 DLIST_ADD_END(ctdb->lock_pending, lock_ctx);
 886         }
 887         CTDB_INCREMENT_STAT(ctdb, locks.num_pending);
 888         if (ctdb_db) {
 889                 CTDB_INCREMENT_DB_STAT(ctdb_db, locks.num_pending);
 890         }
 891
 892         /* Start the timer when we activate the context */
 893         lock_ctx->start_time = timeval_current();
 894
 895         request->lctx = lock_ctx;
 896         request->callback = callback;
 897         request->private_data = private_data;
 898
 899         talloc_set_destructor(request, ctdb_lock_request_destructor);
 900         talloc_set_destructor(lock_ctx, ctdb_lock_context_destructor);
 901
 902         ctdb_lock_schedule(ctdb);
 903
 904         return request;
 905 }
 906
 907
 908 /*
 909  * obtain a lock on a record in a database
 910  */
 911 struct lock_request *ctdb_lock_record(TALLOC_CTX *mem_ctx,
 912                                       struct ctdb_db_context *ctdb_db,
 913                                       TDB_DATA key,
 914                                       bool auto_mark,
 915                                       void (*callback)(void *, bool),
 916                                       void *private_data)
 917 {
 918         return ctdb_lock_internal(mem_ctx,
 919                                   ctdb_db->ctdb,
 920                                   ctdb_db,
 921                                   key,
 922                                   0,
 923                                   callback,
 924                                   private_data,
 925                                   LOCK_RECORD,
 926                                   auto_mark);
 927 }
 928
 929
 930 /*
 931  * obtain a lock on a database
 932  */
 933 struct lock_request *ctdb_lock_db(TALLOC_CTX *mem_ctx,
 934                                   struct ctdb_db_context *ctdb_db,
 935                                   bool auto_mark,
 936                                   void (*callback)(void *, bool),
 937                                   void *private_data)
 938 {
 939         return ctdb_lock_internal(mem_ctx,
 940                                   ctdb_db->ctdb,
 941                                   ctdb_db,
 942                                   tdb_null,
 943                                   0,
 944                                   callback,
 945                                   private_data,
 946                                   LOCK_DB,
 947                                   auto_mark);
 948 }