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 "includes.h"
  21 #include "include/ctdb_private.h"
  22 #include "include/ctdb_protocol.h"
  23 #include "tevent.h"
  24 #include "tdb.h"
  25 #include "db_wrap.h"
  26 #include "system/filesys.h"
  27 #include "lib/util/dlinklist.h"
  28
  29 /*
  30  * Non-blocking Locking API
  31  *
  32  * 1. Create a child process to do blocking locks.
  33  * 2. Once the locks are obtained, signal parent process via fd.
  34  * 3. Invoke registered callback routine with locking status.
  35  * 4. If the child process cannot get locks within certain time,
  36  *    execute an external script to debug.
  37  *
  38  * ctdb_lock_record()      - get a lock on a record
  39  * ctdb_lock_db()          - get a lock on a DB
  40  * ctdb_lock_alldb_prio()  - get a lock on all DBs with given priority
  41  * ctdb_lock_alldb()       - get a lock on all DBs
  42  *
  43  *  auto_mark              - whether to mark/unmark DBs in before/after callback
  44  */
  45
  46 enum lock_type {
  47         LOCK_RECORD,
  48         LOCK_DB,
  49         LOCK_ALLDB_PRIO,
  50         LOCK_ALLDB,
  51 };
  52
  53 static const char * const lock_type_str[] = {
  54         "lock_record",
  55         "lock_db",
  56         "lock_alldb_prio",
  57         "lock_alldb",
  58 };
  59
  60 struct lock_request;
  61
  62 /* lock_context is the common part for a lock request */
  63 struct lock_context {
  64         struct lock_context *next, *prev;
  65         enum lock_type type;
  66         struct ctdb_context *ctdb;
  67         struct ctdb_db_context *ctdb_db;
  68         TDB_DATA key;
  69         uint32_t priority;
  70         bool auto_mark;
  71         struct lock_request *request;
  72         pid_t child;
  73         int fd[2];
  74         struct tevent_fd *tfd;
  75         struct tevent_timer *ttimer;
  76         struct timeval start_time;
  77         uint32_t key_hash;
  78         bool can_schedule;
  79 };
  80
  81 /* lock_request is the client specific part for a lock request */
  82 struct lock_request {
  83         struct lock_context *lctx;
  84         void (*callback)(void *, bool);
  85         void *private_data;
  86 };
  87
  88
  89 /*
  90  * Support samba 3.6.x (and older) versions which do not set db priority.
  91  *
  92  * By default, all databases are set to priority 1. So only when priority
  93  * is set to 1, check for databases that need higher priority.
  94  */
  95 static bool later_db(struct ctdb_context *ctdb, const char *name)
  96 {
  97         if (ctdb->tunable.samba3_hack == 0) {
  98                 return false;
  99         }
 100
 101         if (strstr(name, "brlock") ||
 102             strstr(name, "g_lock") ||
 103             strstr(name, "notify_onelevel") ||
 104             strstr(name, "serverid") ||
 105             strstr(name, "xattr_tdb")) {
 106                 return true;
 107         }
 108
 109         return false;
 110 }
 111
 112 typedef int (*db_handler_t)(struct ctdb_db_context *ctdb_db,
 113                             uint32_t priority,
 114                             void *private_data);
 115
 116 static int ctdb_db_iterator(struct ctdb_context *ctdb, uint32_t priority,
 117                             db_handler_t handler, void *private_data)
 118 {
 119         struct ctdb_db_context *ctdb_db;
 120         int ret;
 121
 122         for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
 123                 if (ctdb_db->priority != priority) {
 124                         continue;
 125                 }
 126                 if (later_db(ctdb, ctdb_db->db_name)) {
 127                         continue;
 128                 }
 129                 ret = handler(ctdb_db, priority, private_data);
 130                 if (ret != 0) {
 131                         return -1;
 132                 }
 133         }
 134
 135         /* If priority != 1, later_db check is not required and can return */
 136         if (priority != 1) {
 137                 return 0;
 138         }
 139
 140         for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
 141                 if (!later_db(ctdb, ctdb_db->db_name)) {
 142                         continue;
 143                 }
 144                 ret = handler(ctdb_db, priority, private_data);
 145                 if (ret != 0) {
 146                         return -1;
 147                 }
 148         }
 149
 150         return 0;
 151 }
 152
 153
 154 /*
 155  * lock all databases - mark only
 156  */
 157 static int db_lock_mark_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
 158                                 void *private_data)
 159 {
 160         int tdb_transaction_write_lock_mark(struct tdb_context *);
 161
 162         DEBUG(DEBUG_INFO, ("marking locked database %s, priority:%u\n",
 163                            ctdb_db->db_name, priority));
 164
 165         if (tdb_transaction_write_lock_mark(ctdb_db->ltdb->tdb) != 0) {
 166                 DEBUG(DEBUG_ERR, ("Failed to mark (transaction lock) database %s\n",
 167                                   ctdb_db->db_name));
 168                 return -1;
 169         }
 170
 171         if (tdb_lockall_mark(ctdb_db->ltdb->tdb) != 0) {
 172                 DEBUG(DEBUG_ERR, ("Failed to mark (all lock) database %s\n",
 173                                   ctdb_db->db_name));
 174                 return -1;
 175         }
 176
 177         return 0;
 178 }
 179
 180 int ctdb_lockall_mark_prio(struct ctdb_context *ctdb, uint32_t priority)
 181 {
 182         /*
 183          * This function is only used by the main dameon during recovery.
 184          * At this stage, the databases have already been locked, by a
 185          * dedicated child process. The freeze_mode variable is used to track
 186          * whether the actual locks are held by the child process or not.
 187          */
 188
 189         if (ctdb->freeze_mode[priority] != CTDB_FREEZE_FROZEN) {
 190                 DEBUG(DEBUG_ERR, ("Attempt to mark all databases locked when not frozen\n"));
 191                 return -1;
 192         }
 193
 194         return ctdb_db_iterator(ctdb, priority, db_lock_mark_handler, NULL);
 195 }
 196
 197 static int ctdb_lockall_mark(struct ctdb_context *ctdb)
 198 {
 199         uint32_t priority;
 200
 201         for (priority=1; priority<=NUM_DB_PRIORITIES; priority++) {
 202                 if (ctdb_db_iterator(ctdb, priority, db_lock_mark_handler, NULL) != 0) {
 203                         return -1;
 204                 }
 205         }
 206
 207         return 0;
 208 }
 209
 210
 211 /*
 212  * lock all databases - unmark only
 213  */
 214 static int db_lock_unmark_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
 215                                   void *private_data)
 216 {
 217         int tdb_transaction_write_lock_unmark(struct tdb_context *);
 218
 219         DEBUG(DEBUG_INFO, ("unmarking locked database %s, priority:%u\n",
 220                            ctdb_db->db_name, priority));
 221
 222         if (tdb_transaction_write_lock_unmark(ctdb_db->ltdb->tdb) != 0) {
 223                 DEBUG(DEBUG_ERR, ("Failed to unmark (transaction lock) database %s\n",
 224                                   ctdb_db->db_name));
 225                 return -1;
 226         }
 227
 228         if (tdb_lockall_unmark(ctdb_db->ltdb->tdb) != 0) {
 229                 DEBUG(DEBUG_ERR, ("Failed to unmark (all lock) database %s\n",
 230                                   ctdb_db->db_name));
 231                 return -1;
 232         }
 233
 234         return 0;
 235 }
 236
 237 int ctdb_lockall_unmark_prio(struct ctdb_context *ctdb, uint32_t priority)
 238 {
 239         /*
 240          * This function is only used by the main daemon during recovery.
 241          * At this stage, the databases have already been locked, by a
 242          * dedicated child process. The freeze_mode variable is used to track
 243          * whether the actual locks are held by the child process or not.
 244          */
 245
 246         if (ctdb->freeze_mode[priority] != CTDB_FREEZE_FROZEN) {
 247                 DEBUG(DEBUG_ERR, ("Attempt to unmark all databases locked when not frozen\n"));
 248                 return -1;
 249         }
 250
 251         return ctdb_db_iterator(ctdb, priority, db_lock_unmark_handler, NULL);
 252 }
 253
 254 static int ctdb_lockall_unmark(struct ctdb_context *ctdb)
 255 {
 256         uint32_t priority;
 257
 258         for (priority=NUM_DB_PRIORITIES; priority>0; priority--) {
 259                 if (ctdb_db_iterator(ctdb, priority, db_lock_unmark_handler, NULL) != 0) {
 260                         return -1;
 261                 }
 262         }
 263
 264         return 0;
 265 }
 266
 267
 268 static void ctdb_lock_schedule(struct ctdb_context *ctdb);
 269
 270 /*
 271  * Destructor to kill the child locking process
 272  */
 273 static int ctdb_lock_context_destructor(struct lock_context *lock_ctx)
 274 {
 275         if (lock_ctx->child > 0) {
 276                 ctdb_kill(lock_ctx->ctdb, lock_ctx->child, SIGKILL);
 277                 if (lock_ctx->type == LOCK_RECORD) {
 278                         DLIST_REMOVE(lock_ctx->ctdb_db->lock_current, lock_ctx);
 279                 } else {
 280                         DLIST_REMOVE(lock_ctx->ctdb->lock_current, lock_ctx);
 281                 }
 282                 if (lock_ctx->ctdb_db) {
 283                         lock_ctx->ctdb_db->lock_num_current--;
 284                 }
 285                 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_current);
 286                 if (lock_ctx->ctdb_db) {
 287                         CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
 288                 }
 289         } else {
 290                 if (lock_ctx->type == LOCK_RECORD) {
 291                         DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
 292                 } else {
 293                         DLIST_REMOVE(lock_ctx->ctdb->lock_pending, lock_ctx);
 294                 }
 295                 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
 296                 if (lock_ctx->ctdb_db) {
 297                         CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 298                 }
 299         }
 300
 301         ctdb_lock_schedule(lock_ctx->ctdb);
 302
 303         return 0;
 304 }
 305
 306
 307 /*
 308  * Destructor to remove lock request
 309  */
 310 static int ctdb_lock_request_destructor(struct lock_request *lock_request)
 311 {
 312         lock_request->lctx->request = NULL;
 313         return 0;
 314 }
 315
 316 void ctdb_lock_free_request_context(struct lock_request *lock_req)
 317 {
 318         struct lock_context *lock_ctx;
 319
 320         lock_ctx = lock_req->lctx;
 321         talloc_free(lock_req);
 322         talloc_free(lock_ctx);
 323 }
 324
 325
 326 /*
 327  * Process all the callbacks waiting for lock
 328  *
 329  * If lock has failed, callback is executed with locked=false
 330  */
 331 static void process_callbacks(struct lock_context *lock_ctx, bool locked)
 332 {
 333         struct lock_request *request;
 334
 335         if (lock_ctx->auto_mark && locked) {
 336                 switch (lock_ctx->type) {
 337                 case LOCK_RECORD:
 338                         tdb_chainlock_mark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
 339                         break;
 340
 341                 case LOCK_DB:
 342                         tdb_lockall_mark(lock_ctx->ctdb_db->ltdb->tdb);
 343                         break;
 344
 345                 case LOCK_ALLDB_PRIO:
 346                         ctdb_lockall_mark_prio(lock_ctx->ctdb, lock_ctx->priority);
 347                         break;
 348
 349                 case LOCK_ALLDB:
 350                         ctdb_lockall_mark(lock_ctx->ctdb);
 351                         break;
 352                 }
 353         }
 354
 355         request = lock_ctx->request;
 356         if (lock_ctx->auto_mark) {
 357                 /* Reset the destructor, so request is not removed from the list */
 358                 talloc_set_destructor(request, NULL);
 359         }
 360         request->callback(request->private_data, locked);
 361
 362         if (lock_ctx->auto_mark && locked) {
 363                 switch (lock_ctx->type) {
 364                 case LOCK_RECORD:
 365                         tdb_chainlock_unmark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
 366                         break;
 367
 368                 case LOCK_DB:
 369                         tdb_lockall_unmark(lock_ctx->ctdb_db->ltdb->tdb);
 370                         break;
 371
 372                 case LOCK_ALLDB_PRIO:
 373                         ctdb_lockall_unmark_prio(lock_ctx->ctdb, lock_ctx->priority);
 374                         break;
 375
 376                 case LOCK_ALLDB:
 377                         ctdb_lockall_unmark(lock_ctx->ctdb);
 378                         break;
 379                 }
 380         }
 381 }
 382
 383
 384 static int lock_bucket_id(double t)
 385 {
 386         double ms = 1.e-3, s = 1;
 387         int id;
 388
 389         if (t < 1*ms) {
 390                 id = 0;
 391         } else if (t < 10*ms) {
 392                 id = 1;
 393         } else if (t < 100*ms) {
 394                 id = 2;
 395         } else if (t < 1*s) {
 396                 id = 3;
 397         } else if (t < 2*s) {
 398                 id = 4;
 399         } else if (t < 4*s) {
 400                 id = 5;
 401         } else if (t < 8*s) {
 402                 id = 6;
 403         } else if (t < 16*s) {
 404                 id = 7;
 405         } else if (t < 32*s) {
 406                 id = 8;
 407         } else if (t < 64*s) {
 408                 id = 9;
 409         } else {
 410                 id = 10;
 411         }
 412
 413         return id;
 414 }
 415
 416 /*
 417  * Callback routine when the required locks are obtained.
 418  * Called from parent context
 419  */
 420 static void ctdb_lock_handler(struct tevent_context *ev,
 421                             struct tevent_fd *tfd,
 422                             uint16_t flags,
 423                             void *private_data)
 424 {
 425         struct lock_context *lock_ctx;
 426         TALLOC_CTX *tmp_ctx = NULL;
 427         char c;
 428         bool locked;
 429         double t;
 430         int id;
 431
 432         lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
 433
 434         /* cancel the timeout event */
 435         if (lock_ctx->ttimer) {
 436                 TALLOC_FREE(lock_ctx->ttimer);
 437         }
 438
 439         t = timeval_elapsed(&lock_ctx->start_time);
 440         id = lock_bucket_id(t);
 441
 442         if (lock_ctx->auto_mark) {
 443                 tmp_ctx = talloc_new(ev);
 444                 talloc_steal(tmp_ctx, lock_ctx);
 445         }
 446
 447         /* Read the status from the child process */
 448         if (read(lock_ctx->fd[0], &c, 1) != 1) {
 449                 locked = false;
 450         } else {
 451                 locked = (c == 0 ? true : false);
 452         }
 453
 454         /* Update statistics */
 455         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_calls);
 456         if (lock_ctx->ctdb_db) {
 457                 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_calls);
 458         }
 459
 460         if (locked) {
 461                 if (lock_ctx->ctdb_db) {
 462                         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.buckets[id]);
 463                         CTDB_UPDATE_LATENCY(lock_ctx->ctdb, lock_ctx->ctdb_db,
 464                                             lock_type_str[lock_ctx->type], locks.latency,
 465                                             lock_ctx->start_time);
 466
 467                         CTDB_UPDATE_DB_LATENCY(lock_ctx->ctdb_db, lock_type_str[lock_ctx->type], locks.latency, t);
 468                         CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.buckets[id]);
 469                 }
 470         } else {
 471                 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_failed);
 472                 if (lock_ctx->ctdb_db) {
 473                         CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_failed);
 474                 }
 475         }
 476
 477         process_callbacks(lock_ctx, locked);
 478
 479         if (lock_ctx->auto_mark) {
 480                 talloc_free(tmp_ctx);
 481         }
 482 }
 483
 484
 485 /*
 486  * Callback routine when required locks are not obtained within timeout
 487  * Called from parent context
 488  */
 489 static void ctdb_lock_timeout_handler(struct tevent_context *ev,
 490                                     struct tevent_timer *ttimer,
 491                                     struct timeval current_time,
 492                                     void *private_data)
 493 {
 494         static const char * debug_locks = NULL;
 495         struct lock_context *lock_ctx;
 496         struct ctdb_context *ctdb;
 497         pid_t pid;
 498
 499         lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
 500         ctdb = lock_ctx->ctdb;
 501
 502         if (lock_ctx->ctdb_db) {
 503                 DEBUG(DEBUG_WARNING,
 504                       ("Unable to get %s lock on database %s for %.0lf seconds\n",
 505                        (lock_ctx->type == LOCK_RECORD ? "RECORD" : "DB"),
 506                        lock_ctx->ctdb_db->db_name,
 507                        timeval_elapsed(&lock_ctx->start_time)));
 508         } else {
 509                 DEBUG(DEBUG_WARNING,
 510                       ("Unable to get ALLDB locks for %.0lf seconds\n",
 511                        timeval_elapsed(&lock_ctx->start_time)));
 512         }
 513
 514         /* Fire a child process to find the blocking process. */
 515         if (debug_locks == NULL) {
 516                 debug_locks = getenv("CTDB_DEBUG_LOCKS");
 517                 if (debug_locks == NULL) {
 518                         debug_locks = talloc_asprintf(ctdb,
 519                                                       "%s/debug_locks.sh",
 520                                                       getenv("CTDB_BASE"));
 521                 }
 522         }
 523         if (debug_locks != NULL) {
 524                 pid = vfork();
 525                 if (pid == 0) {
 526                         execl(debug_locks, debug_locks, NULL);
 527                         _exit(0);
 528                 }
 529                 ctdb_track_child(ctdb, pid);
 530         } else {
 531                 DEBUG(DEBUG_WARNING,
 532                       (__location__
 533                        " Unable to setup lock debugging - no memory?\n"));
 534         }
 535
 536         /* reset the timeout timer */
 537         // talloc_free(lock_ctx->ttimer);
 538         lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
 539                                             lock_ctx,
 540                                             timeval_current_ofs(10, 0),
 541                                             ctdb_lock_timeout_handler,
 542                                             (void *)lock_ctx);
 543 }
 544
 545
 546 static int db_count_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
 547                             void *private_data)
 548 {
 549         int *count = (int *)private_data;
 550
 551         (*count)++;
 552
 553         return 0;
 554 }
 555
 556 struct db_namelist {
 557         char **names;
 558         int n;
 559 };
 560
 561 static int db_name_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
 562                            void *private_data)
 563 {
 564         struct db_namelist *list = (struct db_namelist *)private_data;
 565
 566         list->names[list->n] = talloc_strdup(list->names, ctdb_db->db_path);
 567         list->n++;
 568
 569         return 0;
 570 }
 571
 572 static char **lock_helper_args(TALLOC_CTX *mem_ctx, struct lock_context *lock_ctx, int fd)
 573 {
 574         struct ctdb_context *ctdb = lock_ctx->ctdb;
 575         char **args = NULL;
 576         int nargs, i;
 577         int priority;
 578         struct db_namelist list;
 579
 580         switch (lock_ctx->type) {
 581         case LOCK_RECORD:
 582                 nargs = 6;
 583                 break;
 584
 585         case LOCK_DB:
 586                 nargs = 5;
 587                 break;
 588
 589         case LOCK_ALLDB_PRIO:
 590                 nargs = 4;
 591                 ctdb_db_iterator(ctdb, lock_ctx->priority, db_count_handler, &nargs);
 592                 break;
 593
 594         case LOCK_ALLDB:
 595                 nargs = 4;
 596                 for (priority=1; priority<NUM_DB_PRIORITIES; priority++) {
 597                         ctdb_db_iterator(ctdb, priority, db_count_handler, &nargs);
 598                 }
 599                 break;
 600         }
 601
 602         /* Add extra argument for null termination */
 603         nargs++;
 604
 605         args = talloc_array(mem_ctx, char *, nargs);
 606         if (args == NULL) {
 607                 return NULL;
 608         }
 609
 610         args[0] = talloc_strdup(args, "ctdb_lock_helper");
 611         args[1] = talloc_asprintf(args, "%d", getpid());
 612         args[2] = talloc_asprintf(args, "%d", fd);
 613
 614         switch (lock_ctx->type) {
 615         case LOCK_RECORD:
 616                 args[3] = talloc_strdup(args, "RECORD");
 617                 args[4] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
 618                 if (lock_ctx->key.dsize == 0) {
 619                         args[5] = talloc_strdup(args, "NULL");
 620                 } else {
 621                         args[5] = hex_encode_talloc(args, lock_ctx->key.dptr, lock_ctx->key.dsize);
 622                 }
 623                 break;
 624
 625         case LOCK_DB:
 626                 args[3] = talloc_strdup(args, "DB");
 627                 args[4] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
 628                 break;
 629
 630         case LOCK_ALLDB_PRIO:
 631                 args[3] = talloc_strdup(args, "DB");
 632                 list.names = args;
 633                 list.n = 4;
 634                 ctdb_db_iterator(ctdb, lock_ctx->priority, db_name_handler, &list);
 635                 break;
 636
 637         case LOCK_ALLDB:
 638                 args[3] = talloc_strdup(args, "DB");
 639                 list.names = args;
 640                 list.n = 4;
 641                 for (priority=1; priority<NUM_DB_PRIORITIES; priority++) {
 642                         ctdb_db_iterator(ctdb, priority, db_name_handler, &list);
 643                 }
 644                 break;
 645         }
 646
 647         /* Make sure last argument is NULL */
 648         args[nargs-1] = NULL;
 649
 650         for (i=0; i<nargs-1; i++) {
 651                 if (args[i] == NULL) {
 652                         talloc_free(args);
 653                         return NULL;
 654                 }
 655         }
 656
 657         return args;
 658 }
 659
 660 /*
 661  * Find a lock request that can be scheduled
 662  */
 663 struct lock_context *ctdb_find_lock_context(struct ctdb_context *ctdb)
 664 {
 665         struct lock_context *lock_ctx, *next_ctx;
 666         struct ctdb_db_context *ctdb_db;
 667
 668         /* First check if there are database lock requests */
 669         lock_ctx = ctdb->lock_pending;
 670         while (lock_ctx != NULL) {
 671                 next_ctx = lock_ctx->next;
 672                 if (! lock_ctx->request) {
 673                         DEBUG(DEBUG_INFO, ("Removing lock context without lock request\n"));
 674                         DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
 675                         CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
 676                         if (lock_ctx->ctdb_db) {
 677                                 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 678                         }
 679                         talloc_free(lock_ctx);
 680                 } else {
 681                         /* Found a lock context with lock requests */
 682                         break;
 683                 }
 684                 lock_ctx = next_ctx;
 685         }
 686
 687         if (lock_ctx) {
 688                 return lock_ctx;
 689         }
 690
 691         /* Next check database queues */
 692         for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
 693                 if (ctdb_db->lock_num_current == ctdb->tunable.lock_processes_per_db) {
 694                         continue;
 695                 }
 696                 lock_ctx = ctdb_db->lock_pending;
 697                 while (lock_ctx != NULL) {
 698                         next_ctx = lock_ctx->next;
 699                         if (! lock_ctx->request) {
 700                                 DEBUG(DEBUG_INFO, ("Removing lock context without lock request\n"));
 701                                 DLIST_REMOVE(ctdb_db->lock_pending, lock_ctx);
 702                                 CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
 703                                 CTDB_DECREMENT_DB_STAT(ctdb_db, locks.num_pending);
 704                                 talloc_free(lock_ctx);
 705                         } else {
 706                                 break;
 707                         }
 708
 709                         lock_ctx = next_ctx;
 710                 }
 711
 712                 if (lock_ctx) {
 713                         return lock_ctx;
 714                 }
 715         }
 716
 717         return NULL;
 718 }
 719
 720 /*
 721  * Schedule a new lock child process
 722  * Set up callback handler and timeout handler
 723  */
 724 static void ctdb_lock_schedule(struct ctdb_context *ctdb)
 725 {
 726         struct lock_context *lock_ctx;
 727         int ret;
 728         TALLOC_CTX *tmp_ctx;
 729         const char *helper = BINDIR "/ctdb_lock_helper";
 730         static const char *prog = NULL;
 731         char **args;
 732
 733         if (prog == NULL) {
 734                 const char *t;
 735
 736                 t = getenv("CTDB_LOCK_HELPER");
 737                 if (t != NULL) {
 738                         prog = talloc_strdup(ctdb, t);
 739                 } else {
 740                         prog = talloc_strdup(ctdb, helper);
 741                 }
 742                 CTDB_NO_MEMORY_VOID(ctdb, prog);
 743         }
 744
 745         /* Find a lock context with requests */
 746         lock_ctx = ctdb_find_lock_context(ctdb);
 747         if (lock_ctx == NULL) {
 748                 return;
 749         }
 750
 751         lock_ctx->child = -1;
 752         ret = pipe(lock_ctx->fd);
 753         if (ret != 0) {
 754                 DEBUG(DEBUG_ERR, ("Failed to create pipe in ctdb_lock_schedule\n"));
 755                 return;
 756         }
 757
 758         set_close_on_exec(lock_ctx->fd[0]);
 759
 760         /* Create data for child process */
 761         tmp_ctx = talloc_new(lock_ctx);
 762         if (tmp_ctx == NULL) {
 763                 DEBUG(DEBUG_ERR, ("Failed to allocate memory for helper args\n"));
 764                 close(lock_ctx->fd[0]);
 765                 close(lock_ctx->fd[1]);
 766                 return;
 767         }
 768
 769         /* Create arguments for lock helper */
 770         args = lock_helper_args(tmp_ctx, lock_ctx, lock_ctx->fd[1]);
 771         if (args == NULL) {
 772                 DEBUG(DEBUG_ERR, ("Failed to create lock helper args\n"));
 773                 close(lock_ctx->fd[0]);
 774                 close(lock_ctx->fd[1]);
 775                 talloc_free(tmp_ctx);
 776                 return;
 777         }
 778
 779         lock_ctx->child = vfork();
 780
 781         if (lock_ctx->child == (pid_t)-1) {
 782                 DEBUG(DEBUG_ERR, ("Failed to create a child in ctdb_lock_schedule\n"));
 783                 close(lock_ctx->fd[0]);
 784                 close(lock_ctx->fd[1]);
 785                 talloc_free(tmp_ctx);
 786                 return;
 787         }
 788
 789
 790         /* Child process */
 791         if (lock_ctx->child == 0) {
 792                 ret = execv(prog, args);
 793                 if (ret < 0) {
 794                         DEBUG(DEBUG_ERR, ("Failed to execute helper %s (%d, %s)\n",
 795                                           prog, errno, strerror(errno)));
 796                 }
 797                 _exit(1);
 798         }
 799
 800         /* Parent process */
 801         ctdb_track_child(ctdb, lock_ctx->child);
 802         close(lock_ctx->fd[1]);
 803
 804         talloc_set_destructor(lock_ctx, ctdb_lock_context_destructor);
 805
 806         talloc_free(tmp_ctx);
 807
 808         /* Set up timeout handler */
 809         lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
 810                                             lock_ctx,
 811                                             timeval_current_ofs(10, 0),
 812                                             ctdb_lock_timeout_handler,
 813                                             (void *)lock_ctx);
 814         if (lock_ctx->ttimer == NULL) {
 815                 ctdb_kill(ctdb, lock_ctx->child, SIGKILL);
 816                 lock_ctx->child = -1;
 817                 talloc_set_destructor(lock_ctx, NULL);
 818                 close(lock_ctx->fd[0]);
 819                 return;
 820         }
 821
 822         /* Set up callback */
 823         lock_ctx->tfd = tevent_add_fd(ctdb->ev,
 824                                       lock_ctx,
 825                                       lock_ctx->fd[0],
 826                                       EVENT_FD_READ,
 827                                       ctdb_lock_handler,
 828                                       (void *)lock_ctx);
 829         if (lock_ctx->tfd == NULL) {
 830                 TALLOC_FREE(lock_ctx->ttimer);
 831                 ctdb_kill(ctdb, lock_ctx->child, SIGKILL);
 832                 lock_ctx->child = -1;
 833                 talloc_set_destructor(lock_ctx, NULL);
 834                 close(lock_ctx->fd[0]);
 835                 return;
 836         }
 837         tevent_fd_set_auto_close(lock_ctx->tfd);
 838
 839         /* Move the context from pending to current */
 840         if (lock_ctx->type == LOCK_RECORD) {
 841                 DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
 842                 DLIST_ADD_END(lock_ctx->ctdb_db->lock_current, lock_ctx, NULL);
 843         } else {
 844                 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
 845                 DLIST_ADD_END(ctdb->lock_current, lock_ctx, NULL);
 846         }
 847         CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
 848         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_current);
 849         if (lock_ctx->ctdb_db) {
 850                 lock_ctx->ctdb_db->lock_num_current++;
 851                 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 852                 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
 853         }
 854 }
 855
 856
 857 /*
 858  * Lock record / db depending on type
 859  */
 860 static struct lock_request *ctdb_lock_internal(struct ctdb_context *ctdb,
 861                                                struct ctdb_db_context *ctdb_db,
 862                                                TDB_DATA key,
 863                                                uint32_t priority,
 864                                                void (*callback)(void *, bool),
 865                                                void *private_data,
 866                                                enum lock_type type,
 867                                                bool auto_mark)
 868 {
 869         struct lock_context *lock_ctx = NULL;
 870         struct lock_request *request;
 871
 872         if (callback == NULL) {
 873                 DEBUG(DEBUG_WARNING, ("No callback function specified, not locking\n"));
 874                 return NULL;
 875         }
 876
 877         lock_ctx = talloc_zero(ctdb, struct lock_context);
 878         if (lock_ctx == NULL) {
 879                 DEBUG(DEBUG_ERR, ("Failed to create a new lock context\n"));
 880                 return NULL;
 881         }
 882
 883         if ((request = talloc_zero(lock_ctx, struct lock_request)) == NULL) {
 884                 talloc_free(lock_ctx);
 885                 return NULL;
 886         }
 887
 888         lock_ctx->type = type;
 889         lock_ctx->ctdb = ctdb;
 890         lock_ctx->ctdb_db = ctdb_db;
 891         lock_ctx->key.dsize = key.dsize;
 892         if (key.dsize > 0) {
 893                 lock_ctx->key.dptr = talloc_memdup(lock_ctx, key.dptr, key.dsize);
 894                 if (lock_ctx->key.dptr == NULL) {
 895                         DEBUG(DEBUG_ERR, (__location__ "Memory allocation error\n"));
 896                         talloc_free(lock_ctx);
 897                         return NULL;
 898                 }
 899                 lock_ctx->key_hash = ctdb_hash(&key);
 900         } else {
 901                 lock_ctx->key.dptr = NULL;
 902         }
 903         lock_ctx->priority = priority;
 904         lock_ctx->auto_mark = auto_mark;
 905
 906         lock_ctx->request = request;
 907         lock_ctx->child = -1;
 908
 909         /* Non-record locks are required by recovery and should be scheduled
 910          * immediately, so keep them at the head of the pending queue.
 911          */
 912         if (lock_ctx->type == LOCK_RECORD) {
 913                 DLIST_ADD_END(ctdb_db->lock_pending, lock_ctx, NULL);
 914         } else {
 915                 DLIST_ADD_END(ctdb->lock_pending, lock_ctx, NULL);
 916         }
 917         CTDB_INCREMENT_STAT(ctdb, locks.num_pending);
 918         if (ctdb_db) {
 919                 CTDB_INCREMENT_DB_STAT(ctdb_db, locks.num_pending);
 920         }
 921
 922         /* Start the timer when we activate the context */
 923         lock_ctx->start_time = timeval_current();
 924
 925         request->lctx = lock_ctx;
 926         request->callback = callback;
 927         request->private_data = private_data;
 928
 929         talloc_set_destructor(request, ctdb_lock_request_destructor);
 930
 931         ctdb_lock_schedule(ctdb);
 932
 933         return request;
 934 }
 935
 936
 937 /*
 938  * obtain a lock on a record in a database
 939  */
 940 struct lock_request *ctdb_lock_record(struct ctdb_db_context *ctdb_db,
 941                                       TDB_DATA key,
 942                                       bool auto_mark,
 943                                       void (*callback)(void *, bool),
 944                                       void *private_data)
 945 {
 946         return ctdb_lock_internal(ctdb_db->ctdb,
 947                                   ctdb_db,
 948                                   key,
 949                                   0,
 950                                   callback,
 951                                   private_data,
 952                                   LOCK_RECORD,
 953                                   auto_mark);
 954 }
 955
 956
 957 /*
 958  * obtain a lock on a database
 959  */
 960 struct lock_request *ctdb_lock_db(struct ctdb_db_context *ctdb_db,
 961                                   bool auto_mark,
 962                                   void (*callback)(void *, bool),
 963                                   void *private_data)
 964 {
 965         return ctdb_lock_internal(ctdb_db->ctdb,
 966                                   ctdb_db,
 967                                   tdb_null,
 968                                   0,
 969                                   callback,
 970                                   private_data,
 971                                   LOCK_DB,
 972                                   auto_mark);
 973 }
 974
 975
 976 /*
 977  * obtain locks on all databases of specified priority
 978  */
 979 struct lock_request *ctdb_lock_alldb_prio(struct ctdb_context *ctdb,
 980                                           uint32_t priority,
 981                                           bool auto_mark,
 982                                           void (*callback)(void *, bool),
 983                                           void *private_data)
 984 {
 985         if (priority < 1 || priority > NUM_DB_PRIORITIES) {
 986                 DEBUG(DEBUG_ERR, ("Invalid db priority: %u\n", priority));
 987                 return NULL;
 988         }
 989
 990         return ctdb_lock_internal(ctdb,
 991                                   NULL,
 992                                   tdb_null,
 993                                   priority,
 994                                   callback,
 995                                   private_data,
 996                                   LOCK_ALLDB_PRIO,
 997                                   auto_mark);
 998 }
 999
1000
1001 /*
1002  * obtain locks on all databases
1003  */
1004 struct lock_request *ctdb_lock_alldb(struct ctdb_context *ctdb,
1005                                      bool auto_mark,
1006                                      void (*callback)(void *, bool),
1007                                      void *private_data)
1008 {
1009         return ctdb_lock_internal(ctdb,
1010                                   NULL,
1011                                   tdb_null,
1012                                   0,
1013                                   callback,
1014                                   private_data,
1015                                   LOCK_ALLDB,
1016                                   auto_mark);
1017 }
1018