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 #include "lib/util/sys_rw.h"
  32
  33 #include "ctdb_private.h"
  34
  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  *
  50  *  auto_mark              - whether to mark/unmark DBs in before/after callback
  51  *                           = false is used for freezing databases for
  52  *                           recovery since the recovery cannot start till
  53  *                           databases are locked on all the nodes.
  54  *                           = true is used for record locks.
  55  */
  56
  57 enum lock_type {
  58         LOCK_RECORD,
  59         LOCK_DB,
  60 };
  61
  62 static const char * const lock_type_str[] = {
  63         "lock_record",
  64         "lock_db",
  65 };
  66
  67 struct lock_request;
  68
  69 /* lock_context is the common part for a lock request */
  70 struct lock_context {
  71         struct lock_context *next, *prev;
  72         enum lock_type type;
  73         struct ctdb_context *ctdb;
  74         struct ctdb_db_context *ctdb_db;
  75         TDB_DATA key;
  76         uint32_t priority;
  77         bool auto_mark;
  78         struct lock_request *request;
  79         pid_t child;
  80         int fd[2];
  81         struct tevent_fd *tfd;
  82         struct tevent_timer *ttimer;
  83         struct timeval start_time;
  84         uint32_t key_hash;
  85         bool can_schedule;
  86 };
  87
  88 /* lock_request is the client specific part for a lock request */
  89 struct lock_request {
  90         struct lock_context *lctx;
  91         void (*callback)(void *, bool);
  92         void *private_data;
  93 };
  94
  95
  96 int ctdb_db_iterator(struct ctdb_context *ctdb, ctdb_db_handler_t handler,
  97                      void *private_data)
  98 {
  99         struct ctdb_db_context *ctdb_db;
 100         int ret;
 101
 102         for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
 103                 ret = handler(ctdb_db, private_data);
 104                 if (ret != 0) {
 105                         return -1;
 106                 }
 107         }
 108
 109         return 0;
 110 }
 111
 112 /*
 113  * lock all databases - mark only
 114  */
 115 static int db_lock_mark_handler(struct ctdb_db_context *ctdb_db,
 116                                 void *private_data)
 117 {
 118         int tdb_transaction_write_lock_mark(struct tdb_context *);
 119
 120         DEBUG(DEBUG_INFO, ("marking locked database %s\n", ctdb_db->db_name));
 121
 122         if (tdb_transaction_write_lock_mark(ctdb_db->ltdb->tdb) != 0) {
 123                 DEBUG(DEBUG_ERR, ("Failed to mark (transaction lock) database %s\n",
 124                                   ctdb_db->db_name));
 125                 return -1;
 126         }
 127
 128         if (tdb_lockall_mark(ctdb_db->ltdb->tdb) != 0) {
 129                 DEBUG(DEBUG_ERR, ("Failed to mark (all lock) database %s\n",
 130                                   ctdb_db->db_name));
 131                 return -1;
 132         }
 133
 134         return 0;
 135 }
 136
 137 int ctdb_lockdb_mark(struct ctdb_db_context *ctdb_db)
 138 {
 139         if (!ctdb_db_frozen(ctdb_db)) {
 140                 DEBUG(DEBUG_ERR,
 141                       ("Attempt to mark database locked when not frozen\n"));
 142                 return -1;
 143         }
 144
 145         return db_lock_mark_handler(ctdb_db, NULL);
 146 }
 147
 148 /*
 149  * lock all databases - unmark only
 150  */
 151 static int db_lock_unmark_handler(struct ctdb_db_context *ctdb_db,
 152                                   void *private_data)
 153 {
 154         int tdb_transaction_write_lock_unmark(struct tdb_context *);
 155
 156         DEBUG(DEBUG_INFO, ("unmarking locked database %s\n", ctdb_db->db_name));
 157
 158         if (tdb_transaction_write_lock_unmark(ctdb_db->ltdb->tdb) != 0) {
 159                 DEBUG(DEBUG_ERR, ("Failed to unmark (transaction lock) database %s\n",
 160                                   ctdb_db->db_name));
 161                 return -1;
 162         }
 163
 164         if (tdb_lockall_unmark(ctdb_db->ltdb->tdb) != 0) {
 165                 DEBUG(DEBUG_ERR, ("Failed to unmark (all lock) database %s\n",
 166                                   ctdb_db->db_name));
 167                 return -1;
 168         }
 169
 170         return 0;
 171 }
 172
 173 int ctdb_lockdb_unmark(struct ctdb_db_context *ctdb_db)
 174 {
 175         if (!ctdb_db_frozen(ctdb_db)) {
 176                 DEBUG(DEBUG_ERR,
 177                       ("Attempt to unmark database locked when not frozen\n"));
 178                 return -1;
 179         }
 180
 181         return db_lock_unmark_handler(ctdb_db, NULL);
 182 }
 183
 184 static void ctdb_lock_schedule(struct ctdb_context *ctdb);
 185
 186 /*
 187  * Destructor to kill the child locking process
 188  */
 189 static int ctdb_lock_context_destructor(struct lock_context *lock_ctx)
 190 {
 191         if (lock_ctx->request) {
 192                 lock_ctx->request->lctx = NULL;
 193         }
 194         if (lock_ctx->child > 0) {
 195                 ctdb_kill(lock_ctx->ctdb, lock_ctx->child, SIGTERM);
 196                 if (lock_ctx->type == LOCK_RECORD) {
 197                         DLIST_REMOVE(lock_ctx->ctdb_db->lock_current, lock_ctx);
 198                 } else {
 199                         DLIST_REMOVE(lock_ctx->ctdb->lock_current, lock_ctx);
 200                 }
 201                 if (lock_ctx->ctdb_db) {
 202                         lock_ctx->ctdb_db->lock_num_current--;
 203                 }
 204                 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_current);
 205                 if (lock_ctx->ctdb_db) {
 206                         CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
 207                 }
 208         } else {
 209                 if (lock_ctx->type == LOCK_RECORD) {
 210                         DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
 211                 } else {
 212                         DLIST_REMOVE(lock_ctx->ctdb->lock_pending, lock_ctx);
 213                 }
 214                 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
 215                 if (lock_ctx->ctdb_db) {
 216                         CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 217                 }
 218         }
 219
 220         ctdb_lock_schedule(lock_ctx->ctdb);
 221
 222         return 0;
 223 }
 224
 225
 226 /*
 227  * Destructor to remove lock request
 228  */
 229 static int ctdb_lock_request_destructor(struct lock_request *lock_request)
 230 {
 231         if (lock_request->lctx == NULL) {
 232                 return 0;
 233         }
 234
 235         lock_request->lctx->request = NULL;
 236         TALLOC_FREE(lock_request->lctx);
 237
 238         return 0;
 239 }
 240
 241 /*
 242  * Process all the callbacks waiting for lock
 243  *
 244  * If lock has failed, callback is executed with locked=false
 245  */
 246 static void process_callbacks(struct lock_context *lock_ctx, bool locked)
 247 {
 248         struct lock_request *request;
 249         bool auto_mark = lock_ctx->auto_mark;
 250
 251         if (auto_mark && locked) {
 252                 switch (lock_ctx->type) {
 253                 case LOCK_RECORD:
 254                         tdb_chainlock_mark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
 255                         break;
 256
 257                 case LOCK_DB:
 258                         ctdb_lockdb_mark(lock_ctx->ctdb_db);
 259                         break;
 260                 }
 261         }
 262
 263         request = lock_ctx->request;
 264         if (auto_mark) {
 265                 /* Since request may be freed in the callback, unset the lock
 266                  * context, so request destructor will not free lock context.
 267                  */
 268                 request->lctx = NULL;
 269         }
 270
 271         /* Since request may be freed in the callback, unset the request */
 272         lock_ctx->request = NULL;
 273
 274         request->callback(request->private_data, locked);
 275
 276         if (!auto_mark) {
 277                 return;
 278         }
 279
 280         if (locked) {
 281                 switch (lock_ctx->type) {
 282                 case LOCK_RECORD:
 283                         tdb_chainlock_unmark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
 284                         break;
 285
 286                 case LOCK_DB:
 287                         ctdb_lockdb_unmark(lock_ctx->ctdb_db);
 288                         break;
 289                 }
 290         }
 291
 292         talloc_free(lock_ctx);
 293 }
 294
 295
 296 static int lock_bucket_id(double t)
 297 {
 298         double ms = 1.e-3, s = 1;
 299         int id;
 300
 301         if (t < 1*ms) {
 302                 id = 0;
 303         } else if (t < 10*ms) {
 304                 id = 1;
 305         } else if (t < 100*ms) {
 306                 id = 2;
 307         } else if (t < 1*s) {
 308                 id = 3;
 309         } else if (t < 2*s) {
 310                 id = 4;
 311         } else if (t < 4*s) {
 312                 id = 5;
 313         } else if (t < 8*s) {
 314                 id = 6;
 315         } else if (t < 16*s) {
 316                 id = 7;
 317         } else if (t < 32*s) {
 318                 id = 8;
 319         } else if (t < 64*s) {
 320                 id = 9;
 321         } else {
 322                 id = 10;
 323         }
 324
 325         return id;
 326 }
 327
 328 /*
 329  * Callback routine when the required locks are obtained.
 330  * Called from parent context
 331  */
 332 static void ctdb_lock_handler(struct tevent_context *ev,
 333                             struct tevent_fd *tfd,
 334                             uint16_t flags,
 335                             void *private_data)
 336 {
 337         struct lock_context *lock_ctx;
 338         char c;
 339         bool locked;
 340         double t;
 341         int id;
 342
 343         lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
 344
 345         /* cancel the timeout event */
 346         TALLOC_FREE(lock_ctx->ttimer);
 347
 348         t = timeval_elapsed(&lock_ctx->start_time);
 349         id = lock_bucket_id(t);
 350
 351         /* Read the status from the child process */
 352         if (sys_read(lock_ctx->fd[0], &c, 1) != 1) {
 353                 locked = false;
 354         } else {
 355                 locked = (c == 0 ? true : false);
 356         }
 357
 358         /* Update statistics */
 359         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_calls);
 360         if (lock_ctx->ctdb_db) {
 361                 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_calls);
 362         }
 363
 364         if (locked) {
 365                 if (lock_ctx->ctdb_db) {
 366                         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.buckets[id]);
 367                         CTDB_UPDATE_LATENCY(lock_ctx->ctdb, lock_ctx->ctdb_db,
 368                                             lock_type_str[lock_ctx->type], locks.latency,
 369                                             lock_ctx->start_time);
 370
 371                         CTDB_UPDATE_DB_LATENCY(lock_ctx->ctdb_db, lock_type_str[lock_ctx->type], locks.latency, t);
 372                         CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.buckets[id]);
 373                 }
 374         } else {
 375                 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_failed);
 376                 if (lock_ctx->ctdb_db) {
 377                         CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_failed);
 378                 }
 379         }
 380
 381         process_callbacks(lock_ctx, locked);
 382 }
 383
 384
 385 /*
 386  * Callback routine when required locks are not obtained within timeout
 387  * Called from parent context
 388  */
 389 static void ctdb_lock_timeout_handler(struct tevent_context *ev,
 390                                     struct tevent_timer *ttimer,
 391                                     struct timeval current_time,
 392                                     void *private_data)
 393 {
 394         static char debug_locks[PATH_MAX+1] = "";
 395         static struct timeval last_debug_time;
 396         struct lock_context *lock_ctx;
 397         struct ctdb_context *ctdb;
 398         struct timeval now;
 399         pid_t pid;
 400         double elapsed_time;
 401         int new_timer;
 402
 403         lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
 404         ctdb = lock_ctx->ctdb;
 405
 406         elapsed_time = timeval_elapsed(&lock_ctx->start_time);
 407         if (lock_ctx->ctdb_db) {
 408                 DEBUG(DEBUG_WARNING,
 409                       ("Unable to get %s lock on database %s for %.0lf seconds\n",
 410                        (lock_ctx->type == LOCK_RECORD ? "RECORD" : "DB"),
 411                        lock_ctx->ctdb_db->db_name, elapsed_time));
 412         } else {
 413                 DEBUG(DEBUG_WARNING,
 414                       ("Unable to get ALLDB locks for %.0lf seconds\n",
 415                        elapsed_time));
 416         }
 417
 418         /* If a node stopped/banned, don't spam the logs */
 419         if (ctdb->nodes[ctdb->pnn]->flags & NODE_FLAGS_INACTIVE) {
 420                 goto skip_lock_debug;
 421         }
 422
 423         /* Restrict log debugging to once per second */
 424         now = timeval_current();
 425         if (last_debug_time.tv_sec == now.tv_sec) {
 426                 goto skip_lock_debug;
 427         }
 428
 429         last_debug_time.tv_sec = now.tv_sec;
 430
 431         if (ctdb_set_helper("lock debugging helper",
 432                             debug_locks, sizeof(debug_locks),
 433                             "CTDB_DEBUG_LOCKS",
 434                             getenv("CTDB_BASE"), "debug_locks.sh")) {
 435                 pid = vfork();
 436                 if (pid == 0) {
 437                         execl(debug_locks, debug_locks, NULL);
 438                         _exit(0);
 439                 }
 440                 ctdb_track_child(ctdb, pid);
 441         } else {
 442                 DEBUG(DEBUG_WARNING,
 443                       (__location__
 444                        " Unable to setup lock debugging\n"));
 445         }
 446
 447 skip_lock_debug:
 448
 449         /* Back-off logging if lock is not obtained for a long time */
 450         if (elapsed_time < 100.0) {
 451                 new_timer = 10;
 452         } else if (elapsed_time < 1000.0) {
 453                 new_timer = 100;
 454         } else {
 455                 new_timer = 1000;
 456         }
 457
 458         /* reset the timeout timer */
 459         // talloc_free(lock_ctx->ttimer);
 460         lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
 461                                             lock_ctx,
 462                                             timeval_current_ofs(new_timer, 0),
 463                                             ctdb_lock_timeout_handler,
 464                                             (void *)lock_ctx);
 465 }
 466
 467 static int db_flags(struct ctdb_db_context *ctdb_db)
 468 {
 469         int tdb_flags = TDB_DEFAULT;
 470
 471 #ifdef TDB_MUTEX_LOCKING
 472         if (!ctdb_db->persistent && ctdb_db->ctdb->tunable.mutex_enabled) {
 473                 tdb_flags = (TDB_MUTEX_LOCKING | TDB_CLEAR_IF_FIRST);
 474         }
 475 #endif
 476         return tdb_flags;
 477 }
 478
 479 static bool lock_helper_args(TALLOC_CTX *mem_ctx,
 480                              struct lock_context *lock_ctx, int fd,
 481                              int *argc, const char ***argv)
 482 {
 483         const char **args = NULL;
 484         int nargs = 0, i;
 485
 486         switch (lock_ctx->type) {
 487         case LOCK_RECORD:
 488                 nargs = 6;
 489                 break;
 490
 491         case LOCK_DB:
 492                 nargs = 5;
 493                 break;
 494         }
 495
 496         /* Add extra argument for null termination */
 497         nargs++;
 498
 499         args = talloc_array(mem_ctx, const char *, nargs);
 500         if (args == NULL) {
 501                 return false;
 502         }
 503
 504         args[0] = talloc_asprintf(args, "%d", getpid());
 505         args[1] = talloc_asprintf(args, "%d", fd);
 506
 507         switch (lock_ctx->type) {
 508         case LOCK_RECORD:
 509                 args[2] = talloc_strdup(args, "RECORD");
 510                 args[3] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
 511                 args[4] = talloc_asprintf(args, "0x%x",
 512                                           db_flags(lock_ctx->ctdb_db));
 513                 if (lock_ctx->key.dsize == 0) {
 514                         args[5] = talloc_strdup(args, "NULL");
 515                 } else {
 516                         args[5] = hex_encode_talloc(args, lock_ctx->key.dptr, lock_ctx->key.dsize);
 517                 }
 518                 break;
 519
 520         case LOCK_DB:
 521                 args[2] = talloc_strdup(args, "DB");
 522                 args[3] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
 523                 args[4] = talloc_asprintf(args, "0x%x",
 524                                           db_flags(lock_ctx->ctdb_db));
 525                 break;
 526         }
 527
 528         /* Make sure last argument is NULL */
 529         args[nargs-1] = NULL;
 530
 531         for (i=0; i<nargs-1; i++) {
 532                 if (args[i] == NULL) {
 533                         talloc_free(args);
 534                         return false;
 535                 }
 536         }
 537
 538         *argc = nargs;
 539         *argv = args;
 540         return true;
 541 }
 542
 543 /*
 544  * Find a lock request that can be scheduled
 545  */
 546 static struct lock_context *ctdb_find_lock_context(struct ctdb_context *ctdb)
 547 {
 548         struct lock_context *lock_ctx, *next_ctx;
 549         struct ctdb_db_context *ctdb_db;
 550
 551         /* First check if there are database lock requests */
 552
 553         for (lock_ctx = ctdb->lock_pending; lock_ctx != NULL;
 554              lock_ctx = next_ctx) {
 555
 556                 if (lock_ctx->request != NULL) {
 557                         /* Found a lock context with a request */
 558                         return lock_ctx;
 559                 }
 560
 561                 next_ctx = lock_ctx->next;
 562
 563                 DEBUG(DEBUG_INFO, ("Removing lock context without lock "
 564                                    "request\n"));
 565                 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
 566                 CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
 567                 if (lock_ctx->ctdb_db) {
 568                         CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db,
 569                                                locks.num_pending);
 570                 }
 571                 talloc_free(lock_ctx);
 572         }
 573
 574         /* Next check database queues */
 575         for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
 576                 if (ctdb_db->lock_num_current ==
 577                     ctdb->tunable.lock_processes_per_db) {
 578                         continue;
 579                 }
 580
 581                 for (lock_ctx = ctdb_db->lock_pending; lock_ctx != NULL;
 582                      lock_ctx = next_ctx) {
 583
 584                         next_ctx = lock_ctx->next;
 585
 586                         if (lock_ctx->request != NULL) {
 587                                 return lock_ctx;
 588                         }
 589
 590                         DEBUG(DEBUG_INFO, ("Removing lock context without "
 591                                            "lock request\n"));
 592                         DLIST_REMOVE(ctdb_db->lock_pending, lock_ctx);
 593                         CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
 594                         CTDB_DECREMENT_DB_STAT(ctdb_db, locks.num_pending);
 595                         talloc_free(lock_ctx);
 596                 }
 597         }
 598
 599         return NULL;
 600 }
 601
 602 /*
 603  * Schedule a new lock child process
 604  * Set up callback handler and timeout handler
 605  */
 606 static void ctdb_lock_schedule(struct ctdb_context *ctdb)
 607 {
 608         struct lock_context *lock_ctx;
 609         int ret, argc;
 610         TALLOC_CTX *tmp_ctx;
 611         static char prog[PATH_MAX+1] = "";
 612         const char **args;
 613
 614         if (!ctdb_set_helper("lock helper",
 615                              prog, sizeof(prog),
 616                              "CTDB_LOCK_HELPER",
 617                              CTDB_HELPER_BINDIR, "ctdb_lock_helper")) {
 618                 ctdb_die(ctdb, __location__
 619                          " Unable to set lock helper\n");
 620         }
 621
 622         /* Find a lock context with requests */
 623         lock_ctx = ctdb_find_lock_context(ctdb);
 624         if (lock_ctx == NULL) {
 625                 return;
 626         }
 627
 628         lock_ctx->child = -1;
 629         ret = pipe(lock_ctx->fd);
 630         if (ret != 0) {
 631                 DEBUG(DEBUG_ERR, ("Failed to create pipe in ctdb_lock_schedule\n"));
 632                 return;
 633         }
 634
 635         set_close_on_exec(lock_ctx->fd[0]);
 636
 637         /* Create data for child process */
 638         tmp_ctx = talloc_new(lock_ctx);
 639         if (tmp_ctx == NULL) {
 640                 DEBUG(DEBUG_ERR, ("Failed to allocate memory for helper args\n"));
 641                 close(lock_ctx->fd[0]);
 642                 close(lock_ctx->fd[1]);
 643                 return;
 644         }
 645
 646         if (! ctdb->do_setsched) {
 647                 ret = setenv("CTDB_NOSETSCHED", "1", 1);
 648                 if (ret != 0) {
 649                         DEBUG(DEBUG_WARNING,
 650                               ("Failed to set CTDB_NOSETSCHED variable\n"));
 651                 }
 652         }
 653
 654         /* Create arguments for lock helper */
 655         if (!lock_helper_args(tmp_ctx, lock_ctx, lock_ctx->fd[1],
 656                               &argc, &args)) {
 657                 DEBUG(DEBUG_ERR, ("Failed to create lock helper args\n"));
 658                 close(lock_ctx->fd[0]);
 659                 close(lock_ctx->fd[1]);
 660                 talloc_free(tmp_ctx);
 661                 return;
 662         }
 663
 664         lock_ctx->child = ctdb_vfork_exec(lock_ctx, ctdb, prog, argc,
 665                                           (const char **)args);
 666         if (lock_ctx->child == -1) {
 667                 DEBUG(DEBUG_ERR, ("Failed to create a child in ctdb_lock_schedule\n"));
 668                 close(lock_ctx->fd[0]);
 669                 close(lock_ctx->fd[1]);
 670                 talloc_free(tmp_ctx);
 671                 return;
 672         }
 673
 674         /* Parent process */
 675         close(lock_ctx->fd[1]);
 676
 677         talloc_free(tmp_ctx);
 678
 679         /* Set up timeout handler */
 680         lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
 681                                             lock_ctx,
 682                                             timeval_current_ofs(10, 0),
 683                                             ctdb_lock_timeout_handler,
 684                                             (void *)lock_ctx);
 685         if (lock_ctx->ttimer == NULL) {
 686                 ctdb_kill(ctdb, lock_ctx->child, SIGTERM);
 687                 lock_ctx->child = -1;
 688                 close(lock_ctx->fd[0]);
 689                 return;
 690         }
 691
 692         /* Set up callback */
 693         lock_ctx->tfd = tevent_add_fd(ctdb->ev,
 694                                       lock_ctx,
 695                                       lock_ctx->fd[0],
 696                                       TEVENT_FD_READ,
 697                                       ctdb_lock_handler,
 698                                       (void *)lock_ctx);
 699         if (lock_ctx->tfd == NULL) {
 700                 TALLOC_FREE(lock_ctx->ttimer);
 701                 ctdb_kill(ctdb, lock_ctx->child, SIGTERM);
 702                 lock_ctx->child = -1;
 703                 close(lock_ctx->fd[0]);
 704                 return;
 705         }
 706         tevent_fd_set_auto_close(lock_ctx->tfd);
 707
 708         /* Move the context from pending to current */
 709         if (lock_ctx->type == LOCK_RECORD) {
 710                 DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
 711                 DLIST_ADD_END(lock_ctx->ctdb_db->lock_current, lock_ctx);
 712         } else {
 713                 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
 714                 DLIST_ADD_END(ctdb->lock_current, lock_ctx);
 715         }
 716         CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
 717         CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_current);
 718         if (lock_ctx->ctdb_db) {
 719                 lock_ctx->ctdb_db->lock_num_current++;
 720                 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
 721                 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
 722         }
 723 }
 724
 725
 726 /*
 727  * Lock record / db depending on type
 728  */
 729 static struct lock_request *ctdb_lock_internal(TALLOC_CTX *mem_ctx,
 730                                                struct ctdb_context *ctdb,
 731                                                struct ctdb_db_context *ctdb_db,
 732                                                TDB_DATA key,
 733                                                uint32_t priority,
 734                                                void (*callback)(void *, bool),
 735                                                void *private_data,
 736                                                enum lock_type type,
 737                                                bool auto_mark)
 738 {
 739         struct lock_context *lock_ctx = NULL;
 740         struct lock_request *request;
 741
 742         if (callback == NULL) {
 743                 DEBUG(DEBUG_WARNING, ("No callback function specified, not locking\n"));
 744                 return NULL;
 745         }
 746
 747         lock_ctx = talloc_zero(ctdb, struct lock_context);
 748         if (lock_ctx == NULL) {
 749                 DEBUG(DEBUG_ERR, ("Failed to create a new lock context\n"));
 750                 return NULL;
 751         }
 752
 753         if ((request = talloc_zero(mem_ctx, struct lock_request)) == NULL) {
 754                 talloc_free(lock_ctx);
 755                 return NULL;
 756         }
 757
 758         lock_ctx->type = type;
 759         lock_ctx->ctdb = ctdb;
 760         lock_ctx->ctdb_db = ctdb_db;
 761         lock_ctx->key.dsize = key.dsize;
 762         if (key.dsize > 0) {
 763                 lock_ctx->key.dptr = talloc_memdup(lock_ctx, key.dptr, key.dsize);
 764                 if (lock_ctx->key.dptr == NULL) {
 765                         DEBUG(DEBUG_ERR, (__location__ "Memory allocation error\n"));
 766                         talloc_free(lock_ctx);
 767                         talloc_free(request);
 768                         return NULL;
 769                 }
 770                 lock_ctx->key_hash = ctdb_hash(&key);
 771         } else {
 772                 lock_ctx->key.dptr = NULL;
 773         }
 774         lock_ctx->priority = priority;
 775         lock_ctx->auto_mark = auto_mark;
 776
 777         lock_ctx->request = request;
 778         lock_ctx->child = -1;
 779
 780         /* Non-record locks are required by recovery and should be scheduled
 781          * immediately, so keep them at the head of the pending queue.
 782          */
 783         if (lock_ctx->type == LOCK_RECORD) {
 784                 DLIST_ADD_END(ctdb_db->lock_pending, lock_ctx);
 785         } else {
 786                 DLIST_ADD_END(ctdb->lock_pending, lock_ctx);
 787         }
 788         CTDB_INCREMENT_STAT(ctdb, locks.num_pending);
 789         if (ctdb_db) {
 790                 CTDB_INCREMENT_DB_STAT(ctdb_db, locks.num_pending);
 791         }
 792
 793         /* Start the timer when we activate the context */
 794         lock_ctx->start_time = timeval_current();
 795
 796         request->lctx = lock_ctx;
 797         request->callback = callback;
 798         request->private_data = private_data;
 799
 800         talloc_set_destructor(request, ctdb_lock_request_destructor);
 801         talloc_set_destructor(lock_ctx, ctdb_lock_context_destructor);
 802
 803         ctdb_lock_schedule(ctdb);
 804
 805         return request;
 806 }
 807
 808
 809 /*
 810  * obtain a lock on a record in a database
 811  */
 812 struct lock_request *ctdb_lock_record(TALLOC_CTX *mem_ctx,
 813                                       struct ctdb_db_context *ctdb_db,
 814                                       TDB_DATA key,
 815                                       bool auto_mark,
 816                                       void (*callback)(void *, bool),
 817                                       void *private_data)
 818 {
 819         return ctdb_lock_internal(mem_ctx,
 820                                   ctdb_db->ctdb,
 821                                   ctdb_db,
 822                                   key,
 823                                   0,
 824                                   callback,
 825                                   private_data,
 826                                   LOCK_RECORD,
 827                                   auto_mark);
 828 }
 829
 830
 831 /*
 832  * obtain a lock on a database
 833  */
 834 struct lock_request *ctdb_lock_db(TALLOC_CTX *mem_ctx,
 835                                   struct ctdb_db_context *ctdb_db,
 836                                   bool auto_mark,
 837                                   void (*callback)(void *, bool),
 838                                   void *private_data)
 839 {
 840         return ctdb_lock_internal(mem_ctx,
 841                                   ctdb_db->ctdb,
 842                                   ctdb_db,
 843                                   tdb_null,
 844                                   0,
 845                                   callback,
 846                                   private_data,
 847                                   LOCK_DB,
 848                                   auto_mark);
 849 }