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