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