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 "lib/tdb_wrap/tdb_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->request) {
 276                 lock_ctx->request->lctx = NULL;
 277         }
 278         if (lock_ctx->child > 0) {
 279                 ctdb_kill(lock_ctx->ctdb, lock_ctx->child, SIGKILL);
 280                 if (lock_ctx->type == LOCK_RECORD) {
 281                         DLIST_REMOVE(lock_ctx->ctdb_db->lock_current, lock_ctx);
 282                 } else {
 283                         DLIST_REMOVE(lock_ctx->ctdb->lock_current, lock_ctx);
 284                 }
 285                 if (lock_ctx->ctdb_db) {
 286                         lock_ctx->ctdb_db->lock_num_current--;
 287                 }
 288                 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_current);
 289                 if (lock_ctx->ctdb_db) {
 290                         CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
 291                 }
 292         } else {
 293                 if (lock_ctx->type == LOCK_RECORD) {
 294                         DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
 295                 } else {
 296                         DLIST_REMOVE(lock_ctx->ctdb->lock_pending, lock_ctx);
 297                 }
 298                 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
 299                 if (lock_ctx->ctdb_db) {
 300                         CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 301                 }
 302         }
 303
 304         ctdb_lock_schedule(lock_ctx->ctdb);
 305
 306         return 0;
 307 }
 308
 309
 310 /*
 311  * Destructor to remove lock request
 312  */
 313 static int ctdb_lock_request_destructor(struct lock_request *lock_request)
 314 {
 315         if (lock_request->lctx == NULL) {
 316                 return 0;
 317         }
 318
 319         lock_request->lctx->request = NULL;
 320         TALLOC_FREE(lock_request->lctx);
 321
 322         return 0;
 323 }
 324
 325 /*
 326  * Process all the callbacks waiting for lock
 327  *
 328  * If lock has failed, callback is executed with locked=false
 329  */
 330 static void process_callbacks(struct lock_context *lock_ctx, bool locked)
 331 {
 332         struct lock_request *request;
 333
 334         if (lock_ctx->auto_mark && locked) {
 335                 switch (lock_ctx->type) {
 336                 case LOCK_RECORD:
 337                         tdb_chainlock_mark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
 338                         break;
 339
 340                 case LOCK_DB:
 341                         tdb_lockall_mark(lock_ctx->ctdb_db->ltdb->tdb);
 342                         break;
 343
 344                 case LOCK_ALLDB_PRIO:
 345                         ctdb_lockall_mark_prio(lock_ctx->ctdb, lock_ctx->priority);
 346                         break;
 347
 348                 case LOCK_ALLDB:
 349                         ctdb_lockall_mark(lock_ctx->ctdb);
 350                         break;
 351                 }
 352         }
 353
 354         request = lock_ctx->request;
 355         if (lock_ctx->auto_mark) {
 356                 /* Reset the destructor, so request is not removed from the list */
 357                 talloc_set_destructor(request, NULL);
 358         }
 359
 360         /* Since request may be freed in the callback, unset the request */
 361         lock_ctx->request = NULL;
 362
 363         request->callback(request->private_data, locked);
 364
 365         if (lock_ctx->auto_mark && locked) {
 366                 switch (lock_ctx->type) {
 367                 case LOCK_RECORD:
 368                         tdb_chainlock_unmark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
 369                         break;
 370
 371                 case LOCK_DB:
 372                         tdb_lockall_unmark(lock_ctx->ctdb_db->ltdb->tdb);
 373                         break;
 374
 375                 case LOCK_ALLDB_PRIO:
 376                         ctdb_lockall_unmark_prio(lock_ctx->ctdb, lock_ctx->priority);
 377                         break;
 378
 379                 case LOCK_ALLDB:
 380                         ctdb_lockall_unmark(lock_ctx->ctdb);
 381                         break;
 382                 }
 383         }
 384 }
 385
 386
 387 static int lock_bucket_id(double t)
 388 {
 389         double ms = 1.e-3, s = 1;
 390         int id;
 391
 392         if (t < 1*ms) {
 393                 id = 0;
 394         } else if (t < 10*ms) {
 395                 id = 1;
 396         } else if (t < 100*ms) {
 397                 id = 2;
 398         } else if (t < 1*s) {
 399                 id = 3;
 400         } else if (t < 2*s) {
 401                 id = 4;
 402         } else if (t < 4*s) {
 403                 id = 5;
 404         } else if (t < 8*s) {
 405                 id = 6;
 406         } else if (t < 16*s) {
 407                 id = 7;
 408         } else if (t < 32*s) {
 409                 id = 8;
 410         } else if (t < 64*s) {
 411                 id = 9;
 412         } else {
 413                 id = 10;
 414         }
 415
 416         return id;
 417 }
 418
 419 /*
 420  * Callback routine when the required locks are obtained.
 421  * Called from parent context
 422  */
 423 static void ctdb_lock_handler(struct tevent_context *ev,
 424                             struct tevent_fd *tfd,
 425                             uint16_t flags,
 426                             void *private_data)
 427 {
 428         struct lock_context *lock_ctx;
 429         TALLOC_CTX *tmp_ctx = NULL;
 430         char c;
 431         bool locked;
 432         double t;
 433         int id;
 434
 435         lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
 436
 437         /* cancel the timeout event */
 438         TALLOC_FREE(lock_ctx->ttimer);
 439
 440         t = timeval_elapsed(&lock_ctx->start_time);
 441         id = lock_bucket_id(t);
 442
 443         if (lock_ctx->auto_mark) {
 444                 tmp_ctx = talloc_new(ev);
 445                 talloc_steal(tmp_ctx, lock_ctx);
 446         }
 447
 448         /* Read the status from the child process */
 449         if (sys_read(lock_ctx->fd[0], &c, 1) != 1) {
 450                 locked = false;
 451         } else {
 452                 locked = (c == 0 ? true : false);
 453         }
 454
 455         /* Update statistics */
 456         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_calls);
 457         if (lock_ctx->ctdb_db) {
 458                 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_calls);
 459         }
 460
 461         if (locked) {
 462                 if (lock_ctx->ctdb_db) {
 463                         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.buckets[id]);
 464                         CTDB_UPDATE_LATENCY(lock_ctx->ctdb, lock_ctx->ctdb_db,
 465                                             lock_type_str[lock_ctx->type], locks.latency,
 466                                             lock_ctx->start_time);
 467
 468                         CTDB_UPDATE_DB_LATENCY(lock_ctx->ctdb_db, lock_type_str[lock_ctx->type], locks.latency, t);
 469                         CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.buckets[id]);
 470                 }
 471         } else {
 472                 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_failed);
 473                 if (lock_ctx->ctdb_db) {
 474                         CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_failed);
 475                 }
 476         }
 477
 478         process_callbacks(lock_ctx, locked);
 479
 480         if (lock_ctx->auto_mark) {
 481                 talloc_free(tmp_ctx);
 482         }
 483 }
 484
 485
 486 /*
 487  * Callback routine when required locks are not obtained within timeout
 488  * Called from parent context
 489  */
 490 static void ctdb_lock_timeout_handler(struct tevent_context *ev,
 491                                     struct tevent_timer *ttimer,
 492                                     struct timeval current_time,
 493                                     void *private_data)
 494 {
 495         static char debug_locks[PATH_MAX+1] = "";
 496         struct lock_context *lock_ctx;
 497         struct ctdb_context *ctdb;
 498         pid_t pid;
 499         double elapsed_time;
 500         int new_timer;
 501
 502         lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
 503         ctdb = lock_ctx->ctdb;
 504
 505         /* If a node stopped/banned, don't spam the logs */
 506         if (ctdb->nodes[ctdb->pnn]->flags & NODE_FLAGS_INACTIVE) {
 507                 lock_ctx->ttimer = NULL;
 508                 return;
 509         }
 510
 511         elapsed_time = timeval_elapsed(&lock_ctx->start_time);
 512         if (lock_ctx->ctdb_db) {
 513                 DEBUG(DEBUG_WARNING,
 514                       ("Unable to get %s lock on database %s for %.0lf seconds\n",
 515                        (lock_ctx->type == LOCK_RECORD ? "RECORD" : "DB"),
 516                        lock_ctx->ctdb_db->db_name, elapsed_time));
 517         } else {
 518                 DEBUG(DEBUG_WARNING,
 519                       ("Unable to get ALLDB locks for %.0lf seconds\n",
 520                        elapsed_time));
 521         }
 522
 523         if (ctdb_set_helper("lock debugging helper",
 524                             debug_locks, sizeof(debug_locks),
 525                             "CTDB_DEBUG_LOCKS",
 526                             getenv("CTDB_BASE"), "debug_locks.sh")) {
 527                 pid = vfork();
 528                 if (pid == 0) {
 529                         execl(debug_locks, debug_locks, NULL);
 530                         _exit(0);
 531                 }
 532                 ctdb_track_child(ctdb, pid);
 533         } else {
 534                 DEBUG(DEBUG_WARNING,
 535                       (__location__
 536                        " Unable to setup lock debugging\n"));
 537         }
 538
 539         /* Back-off logging if lock is not obtained for a long time */
 540         if (elapsed_time < 100.0) {
 541                 new_timer = 10;
 542         } else if (elapsed_time < 1000.0) {
 543                 new_timer = 100;
 544         } else {
 545                 new_timer = 1000;
 546         }
 547
 548         /* reset the timeout timer */
 549         // talloc_free(lock_ctx->ttimer);
 550         lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
 551                                             lock_ctx,
 552                                             timeval_current_ofs(new_timer, 0),
 553                                             ctdb_lock_timeout_handler,
 554                                             (void *)lock_ctx);
 555 }
 556
 557
 558 static int db_count_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
 559                             void *private_data)
 560 {
 561         int *count = (int *)private_data;
 562
 563         (*count) += 2;
 564
 565         return 0;
 566 }
 567
 568 static int db_flags(struct ctdb_db_context *ctdb_db)
 569 {
 570         int tdb_flags = TDB_DEFAULT;
 571
 572 #ifdef TDB_MUTEX_LOCKING
 573         if (!ctdb_db->persistent && ctdb_db->ctdb->tunable.mutex_enabled) {
 574                 tdb_flags = (TDB_MUTEX_LOCKING | TDB_CLEAR_IF_FIRST);
 575         }
 576 #endif
 577         return tdb_flags;
 578 }
 579
 580 struct db_namelist {
 581         const char **names;
 582         int n;
 583 };
 584
 585 static int db_name_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
 586                            void *private_data)
 587 {
 588         struct db_namelist *list = (struct db_namelist *)private_data;
 589
 590         list->names[list->n] = talloc_strdup(list->names, ctdb_db->db_path);
 591         list->names[list->n+1] = talloc_asprintf(list->names, "0x%x",
 592                                                  db_flags(ctdb_db));
 593         list->n += 2;
 594
 595         return 0;
 596 }
 597
 598 static bool lock_helper_args(TALLOC_CTX *mem_ctx,
 599                              struct lock_context *lock_ctx, int fd,
 600                              int *argc, const char ***argv)
 601 {
 602         struct ctdb_context *ctdb = lock_ctx->ctdb;
 603         const char **args = NULL;
 604         int nargs, i;
 605         int priority;
 606         struct db_namelist list;
 607
 608         switch (lock_ctx->type) {
 609         case LOCK_RECORD:
 610                 nargs = 6;
 611                 break;
 612
 613         case LOCK_DB:
 614                 nargs = 5;
 615                 break;
 616
 617         case LOCK_ALLDB_PRIO:
 618                 nargs = 3;
 619                 ctdb_db_iterator(ctdb, lock_ctx->priority, db_count_handler, &nargs);
 620                 break;
 621
 622         case LOCK_ALLDB:
 623                 nargs = 3;
 624                 for (priority=1; priority<NUM_DB_PRIORITIES; priority++) {
 625                         ctdb_db_iterator(ctdb, priority, db_count_handler, &nargs);
 626                 }
 627                 break;
 628         }
 629
 630         /* Add extra argument for null termination */
 631         nargs++;
 632
 633         args = talloc_array(mem_ctx, const char *, nargs);
 634         if (args == NULL) {
 635                 return false;
 636         }
 637
 638         args[0] = talloc_asprintf(args, "%d", getpid());
 639         args[1] = talloc_asprintf(args, "%d", fd);
 640
 641         switch (lock_ctx->type) {
 642         case LOCK_RECORD:
 643                 args[2] = talloc_strdup(args, "RECORD");
 644                 args[3] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
 645                 args[4] = talloc_asprintf(args, "0x%x",
 646                                           db_flags(lock_ctx->ctdb_db));
 647                 if (lock_ctx->key.dsize == 0) {
 648                         args[5] = talloc_strdup(args, "NULL");
 649                 } else {
 650                         args[5] = hex_encode_talloc(args, lock_ctx->key.dptr, lock_ctx->key.dsize);
 651                 }
 652                 break;
 653
 654         case LOCK_DB:
 655                 args[2] = talloc_strdup(args, "DB");
 656                 args[3] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
 657                 args[4] = talloc_asprintf(args, "0x%x",
 658                                           db_flags(lock_ctx->ctdb_db));
 659                 break;
 660
 661         case LOCK_ALLDB_PRIO:
 662                 args[2] = talloc_strdup(args, "DB");
 663                 list.names = args;
 664                 list.n = 3;
 665                 ctdb_db_iterator(ctdb, lock_ctx->priority, db_name_handler, &list);
 666                 break;
 667
 668         case LOCK_ALLDB:
 669                 args[2] = talloc_strdup(args, "DB");
 670                 list.names = args;
 671                 list.n = 3;
 672                 for (priority=1; priority<NUM_DB_PRIORITIES; priority++) {
 673                         ctdb_db_iterator(ctdb, priority, db_name_handler, &list);
 674                 }
 675                 break;
 676         }
 677
 678         /* Make sure last argument is NULL */
 679         args[nargs-1] = NULL;
 680
 681         for (i=0; i<nargs-1; i++) {
 682                 if (args[i] == NULL) {
 683                         talloc_free(args);
 684                         return false;
 685                 }
 686         }
 687
 688         *argc = nargs;
 689         *argv = args;
 690         return true;
 691 }
 692
 693 /*
 694  * Find a lock request that can be scheduled
 695  */
 696 static struct lock_context *ctdb_find_lock_context(struct ctdb_context *ctdb)
 697 {
 698         struct lock_context *lock_ctx, *next_ctx;
 699         struct ctdb_db_context *ctdb_db;
 700
 701         /* First check if there are database lock requests */
 702
 703         for (lock_ctx = ctdb->lock_pending; lock_ctx != NULL;
 704              lock_ctx = next_ctx) {
 705
 706                 if (lock_ctx->request != NULL) {
 707                         /* Found a lock context with a request */
 708                         return lock_ctx;
 709                 }
 710
 711                 next_ctx = lock_ctx->next;
 712
 713                 DEBUG(DEBUG_INFO, ("Removing lock context without lock "
 714                                    "request\n"));
 715                 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
 716                 CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
 717                 if (lock_ctx->ctdb_db) {
 718                         CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db,
 719                                                locks.num_pending);
 720                 }
 721                 talloc_free(lock_ctx);
 722         }
 723
 724         /* Next check database queues */
 725         for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
 726                 if (ctdb_db->lock_num_current ==
 727                     ctdb->tunable.lock_processes_per_db) {
 728                         continue;
 729                 }
 730
 731                 for (lock_ctx = ctdb_db->lock_pending; lock_ctx != NULL;
 732                      lock_ctx = next_ctx) {
 733
 734                         next_ctx = lock_ctx->next;
 735
 736                         if (lock_ctx->request != NULL) {
 737                                 return lock_ctx;
 738                         }
 739
 740                         DEBUG(DEBUG_INFO, ("Removing lock context without "
 741                                            "lock request\n"));
 742                         DLIST_REMOVE(ctdb_db->lock_pending, lock_ctx);
 743                         CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
 744                         CTDB_DECREMENT_DB_STAT(ctdb_db, locks.num_pending);
 745                         talloc_free(lock_ctx);
 746                 }
 747         }
 748
 749         return NULL;
 750 }
 751
 752 /*
 753  * Schedule a new lock child process
 754  * Set up callback handler and timeout handler
 755  */
 756 static void ctdb_lock_schedule(struct ctdb_context *ctdb)
 757 {
 758         struct lock_context *lock_ctx;
 759         int ret, argc;
 760         TALLOC_CTX *tmp_ctx;
 761         static char prog[PATH_MAX+1] = "";
 762         const char **args;
 763
 764         if (!ctdb_set_helper("lock helper",
 765                              prog, sizeof(prog),
 766                              "CTDB_LOCK_HELPER",
 767                              CTDB_HELPER_BINDIR, "ctdb_lock_helper")) {
 768                 ctdb_die(ctdb, __location__
 769                          " Unable to set lock helper\n");
 770         }
 771
 772         /* Find a lock context with requests */
 773         lock_ctx = ctdb_find_lock_context(ctdb);
 774         if (lock_ctx == NULL) {
 775                 return;
 776         }
 777
 778         lock_ctx->child = -1;
 779         ret = pipe(lock_ctx->fd);
 780         if (ret != 0) {
 781                 DEBUG(DEBUG_ERR, ("Failed to create pipe in ctdb_lock_schedule\n"));
 782                 return;
 783         }
 784
 785         set_close_on_exec(lock_ctx->fd[0]);
 786
 787         /* Create data for child process */
 788         tmp_ctx = talloc_new(lock_ctx);
 789         if (tmp_ctx == NULL) {
 790                 DEBUG(DEBUG_ERR, ("Failed to allocate memory for helper args\n"));
 791                 close(lock_ctx->fd[0]);
 792                 close(lock_ctx->fd[1]);
 793                 return;
 794         }
 795
 796         /* Create arguments for lock helper */
 797         if (!lock_helper_args(tmp_ctx, lock_ctx, lock_ctx->fd[1],
 798                               &argc, &args)) {
 799                 DEBUG(DEBUG_ERR, ("Failed to create lock helper args\n"));
 800                 close(lock_ctx->fd[0]);
 801                 close(lock_ctx->fd[1]);
 802                 talloc_free(tmp_ctx);
 803                 return;
 804         }
 805
 806         if (!ctdb_vfork_with_logging(lock_ctx, ctdb, "lock_helper",
 807                                      prog, argc, (const char **)args,
 808                                      NULL, NULL, &lock_ctx->child)) {
 809                 DEBUG(DEBUG_ERR, ("Failed to create a child in ctdb_lock_schedule\n"));
 810                 close(lock_ctx->fd[0]);
 811                 close(lock_ctx->fd[1]);
 812                 talloc_free(tmp_ctx);
 813                 return;
 814         }
 815
 816         /* Parent process */
 817         close(lock_ctx->fd[1]);
 818
 819         talloc_free(tmp_ctx);
 820
 821         /* Set up timeout handler */
 822         lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
 823                                             lock_ctx,
 824                                             timeval_current_ofs(10, 0),
 825                                             ctdb_lock_timeout_handler,
 826                                             (void *)lock_ctx);
 827         if (lock_ctx->ttimer == NULL) {
 828                 ctdb_kill(ctdb, lock_ctx->child, SIGKILL);
 829                 lock_ctx->child = -1;
 830                 close(lock_ctx->fd[0]);
 831                 return;
 832         }
 833
 834         /* Set up callback */
 835         lock_ctx->tfd = tevent_add_fd(ctdb->ev,
 836                                       lock_ctx,
 837                                       lock_ctx->fd[0],
 838                                       EVENT_FD_READ,
 839                                       ctdb_lock_handler,
 840                                       (void *)lock_ctx);
 841         if (lock_ctx->tfd == NULL) {
 842                 TALLOC_FREE(lock_ctx->ttimer);
 843                 ctdb_kill(ctdb, lock_ctx->child, SIGKILL);
 844                 lock_ctx->child = -1;
 845                 close(lock_ctx->fd[0]);
 846                 return;
 847         }
 848         tevent_fd_set_auto_close(lock_ctx->tfd);
 849
 850         /* Move the context from pending to current */
 851         if (lock_ctx->type == LOCK_RECORD) {
 852                 DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
 853                 DLIST_ADD_END(lock_ctx->ctdb_db->lock_current, lock_ctx, NULL);
 854         } else {
 855                 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
 856                 DLIST_ADD_END(ctdb->lock_current, lock_ctx, NULL);
 857         }
 858         CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
 859         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_current);
 860         if (lock_ctx->ctdb_db) {
 861                 lock_ctx->ctdb_db->lock_num_current++;
 862                 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 863                 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
 864         }
 865 }
 866
 867
 868 /*
 869  * Lock record / db depending on type
 870  */
 871 static struct lock_request *ctdb_lock_internal(TALLOC_CTX *mem_ctx,
 872                                                struct ctdb_context *ctdb,
 873                                                struct ctdb_db_context *ctdb_db,
 874                                                TDB_DATA key,
 875                                                uint32_t priority,
 876                                                void (*callback)(void *, bool),
 877                                                void *private_data,
 878                                                enum lock_type type,
 879                                                bool auto_mark)
 880 {
 881         struct lock_context *lock_ctx = NULL;
 882         struct lock_request *request;
 883
 884         if (callback == NULL) {
 885                 DEBUG(DEBUG_WARNING, ("No callback function specified, not locking\n"));
 886                 return NULL;
 887         }
 888
 889         lock_ctx = talloc_zero(ctdb, struct lock_context);
 890         if (lock_ctx == NULL) {
 891                 DEBUG(DEBUG_ERR, ("Failed to create a new lock context\n"));
 892                 return NULL;
 893         }
 894
 895         if ((request = talloc_zero(mem_ctx, struct lock_request)) == NULL) {
 896                 talloc_free(lock_ctx);
 897                 return NULL;
 898         }
 899
 900         lock_ctx->type = type;
 901         lock_ctx->ctdb = ctdb;
 902         lock_ctx->ctdb_db = ctdb_db;
 903         lock_ctx->key.dsize = key.dsize;
 904         if (key.dsize > 0) {
 905                 lock_ctx->key.dptr = talloc_memdup(lock_ctx, key.dptr, key.dsize);
 906                 if (lock_ctx->key.dptr == NULL) {
 907                         DEBUG(DEBUG_ERR, (__location__ "Memory allocation error\n"));
 908                         talloc_free(lock_ctx);
 909                         return NULL;
 910                 }
 911                 lock_ctx->key_hash = ctdb_hash(&key);
 912         } else {
 913                 lock_ctx->key.dptr = NULL;
 914         }
 915         lock_ctx->priority = priority;
 916         lock_ctx->auto_mark = auto_mark;
 917
 918         lock_ctx->request = request;
 919         lock_ctx->child = -1;
 920
 921         /* Non-record locks are required by recovery and should be scheduled
 922          * immediately, so keep them at the head of the pending queue.
 923          */
 924         if (lock_ctx->type == LOCK_RECORD) {
 925                 DLIST_ADD_END(ctdb_db->lock_pending, lock_ctx, NULL);
 926         } else {
 927                 DLIST_ADD_END(ctdb->lock_pending, lock_ctx, NULL);
 928         }
 929         CTDB_INCREMENT_STAT(ctdb, locks.num_pending);
 930         if (ctdb_db) {
 931                 CTDB_INCREMENT_DB_STAT(ctdb_db, locks.num_pending);
 932         }
 933
 934         /* Start the timer when we activate the context */
 935         lock_ctx->start_time = timeval_current();
 936
 937         request->lctx = lock_ctx;
 938         request->callback = callback;
 939         request->private_data = private_data;
 940
 941         talloc_set_destructor(request, ctdb_lock_request_destructor);
 942         talloc_set_destructor(lock_ctx, ctdb_lock_context_destructor);
 943
 944         ctdb_lock_schedule(ctdb);
 945
 946         return request;
 947 }
 948
 949
 950 /*
 951  * obtain a lock on a record in a database
 952  */
 953 struct lock_request *ctdb_lock_record(TALLOC_CTX *mem_ctx,
 954                                       struct ctdb_db_context *ctdb_db,
 955                                       TDB_DATA key,
 956                                       bool auto_mark,
 957                                       void (*callback)(void *, bool),
 958                                       void *private_data)
 959 {
 960         return ctdb_lock_internal(mem_ctx,
 961                                   ctdb_db->ctdb,
 962                                   ctdb_db,
 963                                   key,
 964                                   0,
 965                                   callback,
 966                                   private_data,
 967                                   LOCK_RECORD,
 968                                   auto_mark);
 969 }
 970
 971
 972 /*
 973  * obtain a lock on a database
 974  */
 975 struct lock_request *ctdb_lock_db(TALLOC_CTX *mem_ctx,
 976                                   struct ctdb_db_context *ctdb_db,
 977                                   bool auto_mark,
 978                                   void (*callback)(void *, bool),
 979                                   void *private_data)
 980 {
 981         return ctdb_lock_internal(mem_ctx,
 982                                   ctdb_db->ctdb,
 983                                   ctdb_db,
 984                                   tdb_null,
 985                                   0,
 986                                   callback,
 987                                   private_data,
 988                                   LOCK_DB,
 989                                   auto_mark);
 990 }
 991
 992
 993 /*
 994  * obtain locks on all databases of specified priority
 995  */
 996 struct lock_request *ctdb_lock_alldb_prio(TALLOC_CTX *mem_ctx,
 997                                           struct ctdb_context *ctdb,
 998                                           uint32_t priority,
 999                                           bool auto_mark,
1000                                           void (*callback)(void *, bool),
1001                                           void *private_data)
1002 {
1003         if (priority < 1 || priority > NUM_DB_PRIORITIES) {
1004                 DEBUG(DEBUG_ERR, ("Invalid db priority: %u\n", priority));
1005                 return NULL;
1006         }
1007
1008         return ctdb_lock_internal(mem_ctx,
1009                                   ctdb,
1010                                   NULL,
1011                                   tdb_null,
1012                                   priority,
1013                                   callback,
1014                                   private_data,
1015                                   LOCK_ALLDB_PRIO,
1016                                   auto_mark);
1017 }
1018
1019
1020 /*
1021  * obtain locks on all databases
1022  */
1023 struct lock_request *ctdb_lock_alldb(TALLOC_CTX *mem_ctx,
1024                                      struct ctdb_context *ctdb,
1025                                      bool auto_mark,
1026                                      void (*callback)(void *, bool),
1027                                      void *private_data)
1028 {
1029         return ctdb_lock_internal(mem_ctx,
1030                                   ctdb,
1031                                   NULL,
1032                                   tdb_null,
1033                                   0,
1034                                   callback,
1035                                   private_data,
1036                                   LOCK_ALLDB,
1037                                   auto_mark);
1038 }
1039