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                 lock_ctx->ctdb_db->lock_num_current--;
 202                 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_current);
 203                 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
 204         } else {
 205                 if (lock_ctx->type == LOCK_RECORD) {
 206                         DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
 207                 } else {
 208                         DLIST_REMOVE(lock_ctx->ctdb->lock_pending, lock_ctx);
 209                 }
 210                 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
 211                 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 212         }
 213
 214         ctdb_lock_schedule(lock_ctx->ctdb);
 215
 216         return 0;
 217 }
 218
 219
 220 /*
 221  * Destructor to remove lock request
 222  */
 223 static int ctdb_lock_request_destructor(struct lock_request *lock_request)
 224 {
 225         if (lock_request->lctx == NULL) {
 226                 return 0;
 227         }
 228
 229         lock_request->lctx->request = NULL;
 230         TALLOC_FREE(lock_request->lctx);
 231
 232         return 0;
 233 }
 234
 235 /*
 236  * Process all the callbacks waiting for lock
 237  *
 238  * If lock has failed, callback is executed with locked=false
 239  */
 240 static void process_callbacks(struct lock_context *lock_ctx, bool locked)
 241 {
 242         struct lock_request *request;
 243         bool auto_mark = lock_ctx->auto_mark;
 244
 245         if (auto_mark && locked) {
 246                 switch (lock_ctx->type) {
 247                 case LOCK_RECORD:
 248                         tdb_chainlock_mark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
 249                         break;
 250
 251                 case LOCK_DB:
 252                         (void)ctdb_lockdb_mark(lock_ctx->ctdb_db);
 253                         break;
 254                 }
 255         }
 256
 257         request = lock_ctx->request;
 258         if (auto_mark) {
 259                 /* Since request may be freed in the callback, unset the lock
 260                  * context, so request destructor will not free lock context.
 261                  */
 262                 request->lctx = NULL;
 263         }
 264
 265         /* Since request may be freed in the callback, unset the request */
 266         lock_ctx->request = NULL;
 267
 268         request->callback(request->private_data, locked);
 269
 270         if (!auto_mark) {
 271                 return;
 272         }
 273
 274         if (locked) {
 275                 switch (lock_ctx->type) {
 276                 case LOCK_RECORD:
 277                         tdb_chainlock_unmark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
 278                         break;
 279
 280                 case LOCK_DB:
 281                         ctdb_lockdb_unmark(lock_ctx->ctdb_db);
 282                         break;
 283                 }
 284         }
 285
 286         talloc_free(lock_ctx);
 287 }
 288
 289
 290 static int lock_bucket_id(double t)
 291 {
 292         double ms = 1.e-3, s = 1;
 293         int id;
 294
 295         if (t < 1*ms) {
 296                 id = 0;
 297         } else if (t < 10*ms) {
 298                 id = 1;
 299         } else if (t < 100*ms) {
 300                 id = 2;
 301         } else if (t < 1*s) {
 302                 id = 3;
 303         } else if (t < 2*s) {
 304                 id = 4;
 305         } else if (t < 4*s) {
 306                 id = 5;
 307         } else if (t < 8*s) {
 308                 id = 6;
 309         } else if (t < 16*s) {
 310                 id = 7;
 311         } else if (t < 32*s) {
 312                 id = 8;
 313         } else if (t < 64*s) {
 314                 id = 9;
 315         } else {
 316                 id = 10;
 317         }
 318
 319         return id;
 320 }
 321
 322 /*
 323  * Callback routine when the required locks are obtained.
 324  * Called from parent context
 325  */
 326 static void ctdb_lock_handler(struct tevent_context *ev,
 327                             struct tevent_fd *tfd,
 328                             uint16_t flags,
 329                             void *private_data)
 330 {
 331         struct lock_context *lock_ctx;
 332         char c;
 333         bool locked;
 334         double t;
 335         int id;
 336
 337         lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
 338
 339         /* cancel the timeout event */
 340         TALLOC_FREE(lock_ctx->ttimer);
 341
 342         t = timeval_elapsed(&lock_ctx->start_time);
 343         id = lock_bucket_id(t);
 344
 345         /* Read the status from the child process */
 346         if (sys_read(lock_ctx->fd[0], &c, 1) != 1) {
 347                 locked = false;
 348         } else {
 349                 locked = (c == 0 ? true : false);
 350         }
 351
 352         /* Update statistics */
 353         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_calls);
 354         CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_calls);
 355
 356         if (locked) {
 357                 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.buckets[id]);
 358                 CTDB_UPDATE_LATENCY(lock_ctx->ctdb, lock_ctx->ctdb_db,
 359                                     lock_type_str[lock_ctx->type], locks.latency,
 360                                     lock_ctx->start_time);
 361
 362                 CTDB_UPDATE_DB_LATENCY(lock_ctx->ctdb_db, lock_type_str[lock_ctx->type], locks.latency, t);
 363                 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.buckets[id]);
 364         } else {
 365                 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_failed);
 366                 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_failed);
 367         }
 368
 369         process_callbacks(lock_ctx, locked);
 370 }
 371
 372 struct lock_log_entry {
 373         struct db_hash_context *lock_log;
 374         TDB_DATA key;
 375         unsigned long log_sec;
 376         struct tevent_timer *timer;
 377 };
 378
 379 static int lock_log_fetch_parser(uint8_t *keybuf, size_t keylen,
 380                                  uint8_t *databuf, size_t datalen,
 381                                  void *private_data)
 382 {
 383         struct lock_log_entry **entry =
 384                 (struct lock_log_entry **)private_data;
 385
 386         if (datalen != sizeof(struct lock_log_entry *)) {
 387                 return EINVAL;
 388         }
 389
 390         *entry = talloc_get_type_abort(*(void **)databuf,
 391                                        struct lock_log_entry);
 392         return 0;
 393 }
 394
 395 static void lock_log_cleanup(struct tevent_context *ev,
 396                              struct tevent_timer *ttimer,
 397                              struct timeval current_time,
 398                              void *private_data)
 399 {
 400         struct lock_log_entry *entry = talloc_get_type_abort(
 401                 private_data, struct lock_log_entry);
 402         int ret;
 403
 404         entry->timer = NULL;
 405
 406         ret = db_hash_delete(entry->lock_log, entry->key.dptr,
 407                              entry->key.dsize);
 408         if (ret != 0) {
 409                 return;
 410         }
 411         talloc_free(entry);
 412 }
 413
 414 static bool lock_log_skip(struct tevent_context *ev,
 415                           struct db_hash_context *lock_log,
 416                           TDB_DATA key, unsigned long elapsed_sec)
 417 {
 418         struct lock_log_entry *entry = NULL;
 419         int ret;
 420
 421         ret = db_hash_fetch(lock_log, key.dptr, key.dsize,
 422                             lock_log_fetch_parser, &entry);
 423         if (ret == ENOENT) {
 424
 425                 entry = talloc_zero(lock_log, struct lock_log_entry);
 426                 if (entry == NULL) {
 427                         goto fail;
 428                 }
 429
 430                 entry->lock_log = lock_log;
 431
 432                 entry->key.dptr = talloc_memdup(entry, key.dptr, key.dsize);
 433                 if (entry->key.dptr == NULL) {
 434                         talloc_free(entry);
 435                         goto fail;
 436                 }
 437                 entry->key.dsize = key.dsize;
 438
 439                 entry->log_sec = elapsed_sec;
 440                 entry->timer = tevent_add_timer(ev, entry,
 441                                                 timeval_current_ofs(30, 0),
 442                                                 lock_log_cleanup, entry);
 443                 if (entry->timer == NULL) {
 444                         talloc_free(entry);
 445                         goto fail;
 446                 }
 447
 448                 ret = db_hash_add(lock_log, key.dptr, key.dsize,
 449                                   (uint8_t *)&entry,
 450                                   sizeof(struct lock_log_entry *));
 451                 if (ret != 0) {
 452                         talloc_free(entry);
 453                         goto fail;
 454                 }
 455
 456                 return false;
 457
 458         } else if (ret == EINVAL) {
 459
 460                 ret = db_hash_delete(lock_log, key.dptr, key.dsize);
 461                 if (ret != 0) {
 462                         goto fail;
 463                 }
 464
 465                 return false;
 466
 467         } else if (ret == 0) {
 468
 469                 if (elapsed_sec <= entry->log_sec) {
 470                         return true;
 471                 }
 472
 473                 entry->log_sec = elapsed_sec;
 474
 475                 TALLOC_FREE(entry->timer);
 476                 entry->timer = tevent_add_timer(ev, entry,
 477                                                 timeval_current_ofs(30, 0),
 478                                                 lock_log_cleanup, entry);
 479                 if (entry->timer == NULL) {
 480                         ret = db_hash_delete(lock_log, key.dptr, key.dsize);
 481                         if (ret != 0) {
 482                                 goto fail;
 483                         }
 484                         talloc_free(entry);
 485                 }
 486
 487                 return false;
 488         }
 489
 490
 491 fail:
 492         return false;
 493
 494 }
 495
 496 /*
 497  * Callback routine when required locks are not obtained within timeout
 498  * Called from parent context
 499  */
 500 static void ctdb_lock_timeout_handler(struct tevent_context *ev,
 501                                     struct tevent_timer *ttimer,
 502                                     struct timeval current_time,
 503                                     void *private_data)
 504 {
 505         static char debug_locks[PATH_MAX+1] = "";
 506         struct lock_context *lock_ctx;
 507         struct ctdb_context *ctdb;
 508         pid_t pid;
 509         double elapsed_time;
 510         bool skip;
 511         char *keystr;
 512
 513         lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
 514         ctdb = lock_ctx->ctdb;
 515
 516         elapsed_time = timeval_elapsed(&lock_ctx->start_time);
 517
 518         /* For database locks, always log */
 519         if (lock_ctx->type == LOCK_DB) {
 520                 DEBUG(DEBUG_WARNING,
 521                       ("Unable to get DB lock on database %s for "
 522                        "%.0lf seconds\n",
 523                        lock_ctx->ctdb_db->db_name, elapsed_time));
 524                 goto lock_debug;
 525         }
 526
 527         /* For record locks, check if we have already logged */
 528         skip = lock_log_skip(ev, lock_ctx->ctdb_db->lock_log,
 529                              lock_ctx->key, (unsigned long)elapsed_time);
 530         if (skip) {
 531                 goto skip_lock_debug;
 532         }
 533
 534         keystr = hex_encode_talloc(lock_ctx, lock_ctx->key.dptr,
 535                                    lock_ctx->key.dsize);
 536         DEBUG(DEBUG_WARNING,
 537               ("Unable to get RECORD lock on database %s for %.0lf seconds"
 538                " (key %s)\n",
 539                lock_ctx->ctdb_db->db_name, elapsed_time,
 540                keystr ? keystr : ""));
 541         TALLOC_FREE(keystr);
 542
 543         /* If a node stopped/banned, don't spam the logs */
 544         if (ctdb->nodes[ctdb->pnn]->flags & NODE_FLAGS_INACTIVE) {
 545                 goto skip_lock_debug;
 546         }
 547
 548 lock_debug:
 549
 550         if (ctdb_set_helper("lock debugging helper",
 551                             debug_locks, sizeof(debug_locks),
 552                             "CTDB_DEBUG_LOCKS",
 553                             getenv("CTDB_BASE"), "debug_locks.sh")) {
 554                 pid = vfork();
 555                 if (pid == 0) {
 556                         execl(debug_locks, debug_locks, NULL);
 557                         _exit(0);
 558                 }
 559                 ctdb_track_child(ctdb, pid);
 560         } else {
 561                 DEBUG(DEBUG_WARNING,
 562                       (__location__
 563                        " Unable to setup lock debugging\n"));
 564         }
 565
 566 skip_lock_debug:
 567
 568         /* reset the timeout timer */
 569         // talloc_free(lock_ctx->ttimer);
 570         lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
 571                                             lock_ctx,
 572                                             timeval_current_ofs(10, 0),
 573                                             ctdb_lock_timeout_handler,
 574                                             (void *)lock_ctx);
 575 }
 576
 577 static bool lock_helper_args(TALLOC_CTX *mem_ctx,
 578                              struct lock_context *lock_ctx, int fd,
 579                              int *argc, const char ***argv)
 580 {
 581         const char **args = NULL;
 582         int nargs = 0, i;
 583
 584         switch (lock_ctx->type) {
 585         case LOCK_RECORD:
 586                 nargs = 6;
 587                 break;
 588
 589         case LOCK_DB:
 590                 nargs = 5;
 591                 break;
 592         }
 593
 594         /* Add extra argument for null termination */
 595         nargs++;
 596
 597         args = talloc_array(mem_ctx, const char *, nargs);
 598         if (args == NULL) {
 599                 return false;
 600         }
 601
 602         args[0] = talloc_asprintf(args, "%d", getpid());
 603         args[1] = talloc_asprintf(args, "%d", fd);
 604
 605         switch (lock_ctx->type) {
 606         case LOCK_RECORD:
 607                 args[2] = talloc_strdup(args, "RECORD");
 608                 args[3] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
 609                 args[4] = talloc_asprintf(args, "0x%x",
 610                                 tdb_get_flags(lock_ctx->ctdb_db->ltdb->tdb));
 611                 if (lock_ctx->key.dsize == 0) {
 612                         args[5] = talloc_strdup(args, "NULL");
 613                 } else {
 614                         args[5] = hex_encode_talloc(args, lock_ctx->key.dptr, lock_ctx->key.dsize);
 615                 }
 616                 break;
 617
 618         case LOCK_DB:
 619                 args[2] = talloc_strdup(args, "DB");
 620                 args[3] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
 621                 args[4] = talloc_asprintf(args, "0x%x",
 622                                 tdb_get_flags(lock_ctx->ctdb_db->ltdb->tdb));
 623                 break;
 624         }
 625
 626         /* Make sure last argument is NULL */
 627         args[nargs-1] = NULL;
 628
 629         for (i=0; i<nargs-1; i++) {
 630                 if (args[i] == NULL) {
 631                         talloc_free(args);
 632                         return false;
 633                 }
 634         }
 635
 636         *argc = nargs;
 637         *argv = args;
 638         return true;
 639 }
 640
 641 /*
 642  * Find a lock request that can be scheduled
 643  */
 644 static struct lock_context *ctdb_find_lock_context(struct ctdb_context *ctdb)
 645 {
 646         struct lock_context *lock_ctx, *next_ctx;
 647         struct ctdb_db_context *ctdb_db;
 648
 649         /* First check if there are database lock requests */
 650
 651         for (lock_ctx = ctdb->lock_pending; lock_ctx != NULL;
 652              lock_ctx = next_ctx) {
 653
 654                 if (lock_ctx->request != NULL) {
 655                         /* Found a lock context with a request */
 656                         return lock_ctx;
 657                 }
 658
 659                 next_ctx = lock_ctx->next;
 660
 661                 DEBUG(DEBUG_INFO, ("Removing lock context without lock "
 662                                    "request\n"));
 663                 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
 664                 CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
 665                 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 666                 talloc_free(lock_ctx);
 667         }
 668
 669         /* Next check database queues */
 670         for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
 671                 if (ctdb_db->lock_num_current ==
 672                     ctdb->tunable.lock_processes_per_db) {
 673                         continue;
 674                 }
 675
 676                 for (lock_ctx = ctdb_db->lock_pending; lock_ctx != NULL;
 677                      lock_ctx = next_ctx) {
 678
 679                         next_ctx = lock_ctx->next;
 680
 681                         if (lock_ctx->request != NULL) {
 682                                 return lock_ctx;
 683                         }
 684
 685                         DEBUG(DEBUG_INFO, ("Removing lock context without "
 686                                            "lock request\n"));
 687                         DLIST_REMOVE(ctdb_db->lock_pending, lock_ctx);
 688                         CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
 689                         CTDB_DECREMENT_DB_STAT(ctdb_db, locks.num_pending);
 690                         talloc_free(lock_ctx);
 691                 }
 692         }
 693
 694         return NULL;
 695 }
 696
 697 /*
 698  * Schedule a new lock child process
 699  * Set up callback handler and timeout handler
 700  */
 701 static void ctdb_lock_schedule(struct ctdb_context *ctdb)
 702 {
 703         struct lock_context *lock_ctx;
 704         int ret, argc;
 705         TALLOC_CTX *tmp_ctx;
 706         static char prog[PATH_MAX+1] = "";
 707         const char **args;
 708
 709         if (!ctdb_set_helper("lock helper",
 710                              prog, sizeof(prog),
 711                              "CTDB_LOCK_HELPER",
 712                              CTDB_HELPER_BINDIR, "ctdb_lock_helper")) {
 713                 ctdb_die(ctdb, __location__
 714                          " Unable to set lock helper\n");
 715         }
 716
 717         /* Find a lock context with requests */
 718         lock_ctx = ctdb_find_lock_context(ctdb);
 719         if (lock_ctx == NULL) {
 720                 return;
 721         }
 722
 723         lock_ctx->child = -1;
 724         ret = pipe(lock_ctx->fd);
 725         if (ret != 0) {
 726                 DEBUG(DEBUG_ERR, ("Failed to create pipe in ctdb_lock_schedule\n"));
 727                 return;
 728         }
 729
 730         set_close_on_exec(lock_ctx->fd[0]);
 731
 732         /* Create data for child process */
 733         tmp_ctx = talloc_new(lock_ctx);
 734         if (tmp_ctx == NULL) {
 735                 DEBUG(DEBUG_ERR, ("Failed to allocate memory for helper args\n"));
 736                 close(lock_ctx->fd[0]);
 737                 close(lock_ctx->fd[1]);
 738                 return;
 739         }
 740
 741         if (! ctdb->do_setsched) {
 742                 ret = setenv("CTDB_NOSETSCHED", "1", 1);
 743                 if (ret != 0) {
 744                         DEBUG(DEBUG_WARNING,
 745                               ("Failed to set CTDB_NOSETSCHED variable\n"));
 746                 }
 747         }
 748
 749         /* Create arguments for lock helper */
 750         if (!lock_helper_args(tmp_ctx, lock_ctx, lock_ctx->fd[1],
 751                               &argc, &args)) {
 752                 DEBUG(DEBUG_ERR, ("Failed to create lock helper args\n"));
 753                 close(lock_ctx->fd[0]);
 754                 close(lock_ctx->fd[1]);
 755                 talloc_free(tmp_ctx);
 756                 return;
 757         }
 758
 759         lock_ctx->child = ctdb_vfork_exec(lock_ctx, ctdb, prog, argc,
 760                                           (const char **)args);
 761         if (lock_ctx->child == -1) {
 762                 DEBUG(DEBUG_ERR, ("Failed to create a child in ctdb_lock_schedule\n"));
 763                 close(lock_ctx->fd[0]);
 764                 close(lock_ctx->fd[1]);
 765                 talloc_free(tmp_ctx);
 766                 return;
 767         }
 768
 769         /* Parent process */
 770         close(lock_ctx->fd[1]);
 771
 772         talloc_free(tmp_ctx);
 773
 774         /* Set up timeout handler */
 775         lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
 776                                             lock_ctx,
 777                                             timeval_current_ofs(10, 0),
 778                                             ctdb_lock_timeout_handler,
 779                                             (void *)lock_ctx);
 780         if (lock_ctx->ttimer == NULL) {
 781                 ctdb_kill(ctdb, lock_ctx->child, SIGTERM);
 782                 lock_ctx->child = -1;
 783                 close(lock_ctx->fd[0]);
 784                 return;
 785         }
 786
 787         /* Set up callback */
 788         lock_ctx->tfd = tevent_add_fd(ctdb->ev,
 789                                       lock_ctx,
 790                                       lock_ctx->fd[0],
 791                                       TEVENT_FD_READ,
 792                                       ctdb_lock_handler,
 793                                       (void *)lock_ctx);
 794         if (lock_ctx->tfd == NULL) {
 795                 TALLOC_FREE(lock_ctx->ttimer);
 796                 ctdb_kill(ctdb, lock_ctx->child, SIGTERM);
 797                 lock_ctx->child = -1;
 798                 close(lock_ctx->fd[0]);
 799                 return;
 800         }
 801         tevent_fd_set_auto_close(lock_ctx->tfd);
 802
 803         /* Move the context from pending to current */
 804         if (lock_ctx->type == LOCK_RECORD) {
 805                 DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
 806                 DLIST_ADD_END(lock_ctx->ctdb_db->lock_current, lock_ctx);
 807         } else {
 808                 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
 809                 DLIST_ADD_END(ctdb->lock_current, lock_ctx);
 810         }
 811         CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
 812         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_current);
 813         lock_ctx->ctdb_db->lock_num_current++;
 814         CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 815         CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
 816 }
 817
 818
 819 /*
 820  * Lock record / db depending on type
 821  */
 822 static struct lock_request *ctdb_lock_internal(TALLOC_CTX *mem_ctx,
 823                                                struct ctdb_context *ctdb,
 824                                                struct ctdb_db_context *ctdb_db,
 825                                                TDB_DATA key,
 826                                                uint32_t priority,
 827                                                void (*callback)(void *, bool),
 828                                                void *private_data,
 829                                                enum lock_type type,
 830                                                bool auto_mark)
 831 {
 832         struct lock_context *lock_ctx = NULL;
 833         struct lock_request *request;
 834
 835         if (callback == NULL) {
 836                 DEBUG(DEBUG_WARNING, ("No callback function specified, not locking\n"));
 837                 return NULL;
 838         }
 839
 840         lock_ctx = talloc_zero(ctdb, struct lock_context);
 841         if (lock_ctx == NULL) {
 842                 DEBUG(DEBUG_ERR, ("Failed to create a new lock context\n"));
 843                 return NULL;
 844         }
 845
 846         if ((request = talloc_zero(mem_ctx, struct lock_request)) == NULL) {
 847                 talloc_free(lock_ctx);
 848                 return NULL;
 849         }
 850
 851         lock_ctx->type = type;
 852         lock_ctx->ctdb = ctdb;
 853         lock_ctx->ctdb_db = ctdb_db;
 854         lock_ctx->key.dsize = key.dsize;
 855         if (key.dsize > 0) {
 856                 lock_ctx->key.dptr = talloc_memdup(lock_ctx, key.dptr, key.dsize);
 857                 if (lock_ctx->key.dptr == NULL) {
 858                         DEBUG(DEBUG_ERR, (__location__ "Memory allocation error\n"));
 859                         talloc_free(lock_ctx);
 860                         talloc_free(request);
 861                         return NULL;
 862                 }
 863                 lock_ctx->key_hash = ctdb_hash(&key);
 864         } else {
 865                 lock_ctx->key.dptr = NULL;
 866         }
 867         lock_ctx->priority = priority;
 868         lock_ctx->auto_mark = auto_mark;
 869
 870         lock_ctx->request = request;
 871         lock_ctx->child = -1;
 872
 873         /* Non-record locks are required by recovery and should be scheduled
 874          * immediately, so keep them at the head of the pending queue.
 875          */
 876         if (lock_ctx->type == LOCK_RECORD) {
 877                 DLIST_ADD_END(ctdb_db->lock_pending, lock_ctx);
 878         } else {
 879                 DLIST_ADD_END(ctdb->lock_pending, lock_ctx);
 880         }
 881         CTDB_INCREMENT_STAT(ctdb, locks.num_pending);
 882         if (ctdb_db) {
 883                 CTDB_INCREMENT_DB_STAT(ctdb_db, locks.num_pending);
 884         }
 885
 886         /* Start the timer when we activate the context */
 887         lock_ctx->start_time = timeval_current();
 888
 889         request->lctx = lock_ctx;
 890         request->callback = callback;
 891         request->private_data = private_data;
 892
 893         talloc_set_destructor(request, ctdb_lock_request_destructor);
 894         talloc_set_destructor(lock_ctx, ctdb_lock_context_destructor);
 895
 896         ctdb_lock_schedule(ctdb);
 897
 898         return request;
 899 }
 900
 901
 902 /*
 903  * obtain a lock on a record in a database
 904  */
 905 struct lock_request *ctdb_lock_record(TALLOC_CTX *mem_ctx,
 906                                       struct ctdb_db_context *ctdb_db,
 907                                       TDB_DATA key,
 908                                       bool auto_mark,
 909                                       void (*callback)(void *, bool),
 910                                       void *private_data)
 911 {
 912         return ctdb_lock_internal(mem_ctx,
 913                                   ctdb_db->ctdb,
 914                                   ctdb_db,
 915                                   key,
 916                                   0,
 917                                   callback,
 918                                   private_data,
 919                                   LOCK_RECORD,
 920                                   auto_mark);
 921 }
 922
 923
 924 /*
 925  * obtain a lock on a database
 926  */
 927 struct lock_request *ctdb_lock_db(TALLOC_CTX *mem_ctx,
 928                                   struct ctdb_db_context *ctdb_db,
 929                                   bool auto_mark,
 930                                   void (*callback)(void *, bool),
 931                                   void *private_data)
 932 {
 933         return ctdb_lock_internal(mem_ctx,
 934                                   ctdb_db->ctdb,
 935                                   ctdb_db,
 936                                   tdb_null,
 937                                   0,
 938                                   callback,
 939                                   private_data,
 940                                   LOCK_DB,
 941                                   auto_mark);
 942 }