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