3 provide API to do non-blocking locks for single or all databases
5 Copyright (C) Amitay Isaacs 2012
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.
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.
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/>.
21 #include "system/filesys.h"
22 #include "system/network.h"
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"
32 #include "ctdb_private.h"
34 #include "common/system.h"
35 #include "common/common.h"
36 #include "common/logging.h"
39 * Non-blocking Locking API
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.
47 * ctdb_lock_record() - get a lock on a record
48 * ctdb_lock_db() - get a lock on a DB
49 * ctdb_lock_alldb_prio() - get a lock on all DBs with given priority
50 * ctdb_lock_alldb() - get a lock on all DBs
52 * auto_mark - whether to mark/unmark DBs in before/after callback
53 * = false is used for freezing databases for
54 * recovery since the recovery cannot start till
55 * databases are locked on all the nodes.
56 * = true is used for record locks.
66 static const char * const lock_type_str[] = {
75 /* lock_context is the common part for a lock request */
77 struct lock_context *next, *prev;
79 struct ctdb_context *ctdb;
80 struct ctdb_db_context *ctdb_db;
84 struct lock_request *request;
87 struct tevent_fd *tfd;
88 struct tevent_timer *ttimer;
89 struct timeval start_time;
94 /* lock_request is the client specific part for a lock request */
96 struct lock_context *lctx;
97 void (*callback)(void *, bool);
103 * Support samba 3.6.x (and older) versions which do not set db priority.
105 * By default, all databases are set to priority 1. So only when priority
106 * is set to 1, check for databases that need higher priority.
108 static bool later_db(struct ctdb_context *ctdb, const char *name)
110 if (ctdb->tunable.samba3_hack == 0) {
114 if (strstr(name, "brlock") ||
115 strstr(name, "g_lock") ||
116 strstr(name, "notify_onelevel") ||
117 strstr(name, "serverid") ||
118 strstr(name, "xattr_tdb")) {
125 int ctdb_db_prio_iterator(struct ctdb_context *ctdb, uint32_t priority,
126 ctdb_db_handler_t handler, void *private_data)
128 struct ctdb_db_context *ctdb_db;
131 for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
132 if (ctdb_db->priority != priority) {
135 if (later_db(ctdb, ctdb_db->db_name)) {
138 ret = handler(ctdb_db, private_data);
144 /* If priority != 1, later_db check is not required and can return */
149 for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
150 if (!later_db(ctdb, ctdb_db->db_name)) {
153 ret = handler(ctdb_db, private_data);
162 int ctdb_db_iterator(struct ctdb_context *ctdb, ctdb_db_handler_t handler,
165 struct ctdb_db_context *ctdb_db;
168 for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
169 ret = handler(ctdb_db, private_data);
179 * lock all databases - mark only
181 static int db_lock_mark_handler(struct ctdb_db_context *ctdb_db,
184 int tdb_transaction_write_lock_mark(struct tdb_context *);
186 DEBUG(DEBUG_INFO, ("marking locked database %s\n", ctdb_db->db_name));
188 if (tdb_transaction_write_lock_mark(ctdb_db->ltdb->tdb) != 0) {
189 DEBUG(DEBUG_ERR, ("Failed to mark (transaction lock) database %s\n",
194 if (tdb_lockall_mark(ctdb_db->ltdb->tdb) != 0) {
195 DEBUG(DEBUG_ERR, ("Failed to mark (all lock) database %s\n",
203 int ctdb_lockdb_mark(struct ctdb_db_context *ctdb_db)
205 if (!ctdb_db_frozen(ctdb_db)) {
207 ("Attempt to mark database locked when not frozen\n"));
211 return db_lock_mark_handler(ctdb_db, NULL);
214 int ctdb_lockall_mark_prio(struct ctdb_context *ctdb, uint32_t priority)
217 * This function is only used by the main dameon during recovery.
218 * At this stage, the databases have already been locked, by a
219 * dedicated child process.
222 if (!ctdb_db_prio_frozen(ctdb, priority)) {
223 DEBUG(DEBUG_ERR, ("Attempt to mark all databases locked when not frozen\n"));
227 return ctdb_db_prio_iterator(ctdb, priority, db_lock_mark_handler, NULL);
230 static int ctdb_lockall_mark(struct ctdb_context *ctdb)
234 for (priority=1; priority<=NUM_DB_PRIORITIES; priority++) {
237 ret = ctdb_db_prio_iterator(ctdb, priority,
238 db_lock_mark_handler, NULL);
249 * lock all databases - unmark only
251 static int db_lock_unmark_handler(struct ctdb_db_context *ctdb_db,
254 int tdb_transaction_write_lock_unmark(struct tdb_context *);
256 DEBUG(DEBUG_INFO, ("unmarking locked database %s\n", ctdb_db->db_name));
258 if (tdb_transaction_write_lock_unmark(ctdb_db->ltdb->tdb) != 0) {
259 DEBUG(DEBUG_ERR, ("Failed to unmark (transaction lock) database %s\n",
264 if (tdb_lockall_unmark(ctdb_db->ltdb->tdb) != 0) {
265 DEBUG(DEBUG_ERR, ("Failed to unmark (all lock) database %s\n",
273 int ctdb_lockdb_unmark(struct ctdb_db_context *ctdb_db)
275 if (!ctdb_db_frozen(ctdb_db)) {
277 ("Attempt to unmark database locked when not frozen\n"));
281 return db_lock_unmark_handler(ctdb_db, NULL);
284 int ctdb_lockall_unmark_prio(struct ctdb_context *ctdb, uint32_t priority)
287 * This function is only used by the main daemon during recovery.
288 * At this stage, the databases have already been locked, by a
289 * dedicated child process.
292 if (!ctdb_db_prio_frozen(ctdb, priority)) {
293 DEBUG(DEBUG_ERR, ("Attempt to unmark all databases locked when not frozen\n"));
297 return ctdb_db_prio_iterator(ctdb, priority, db_lock_unmark_handler,
301 static int ctdb_lockall_unmark(struct ctdb_context *ctdb)
305 for (priority=NUM_DB_PRIORITIES; priority>0; priority--) {
308 ret = ctdb_db_prio_iterator(ctdb, priority,
309 db_lock_unmark_handler, NULL);
319 static void ctdb_lock_schedule(struct ctdb_context *ctdb);
322 * Destructor to kill the child locking process
324 static int ctdb_lock_context_destructor(struct lock_context *lock_ctx)
326 if (lock_ctx->request) {
327 lock_ctx->request->lctx = NULL;
329 if (lock_ctx->child > 0) {
330 ctdb_kill(lock_ctx->ctdb, lock_ctx->child, SIGKILL);
331 if (lock_ctx->type == LOCK_RECORD) {
332 DLIST_REMOVE(lock_ctx->ctdb_db->lock_current, lock_ctx);
334 DLIST_REMOVE(lock_ctx->ctdb->lock_current, lock_ctx);
336 if (lock_ctx->ctdb_db) {
337 lock_ctx->ctdb_db->lock_num_current--;
339 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_current);
340 if (lock_ctx->ctdb_db) {
341 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
344 if (lock_ctx->type == LOCK_RECORD) {
345 DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
347 DLIST_REMOVE(lock_ctx->ctdb->lock_pending, lock_ctx);
349 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
350 if (lock_ctx->ctdb_db) {
351 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
355 ctdb_lock_schedule(lock_ctx->ctdb);
362 * Destructor to remove lock request
364 static int ctdb_lock_request_destructor(struct lock_request *lock_request)
366 if (lock_request->lctx == NULL) {
370 lock_request->lctx->request = NULL;
371 TALLOC_FREE(lock_request->lctx);
377 * Process all the callbacks waiting for lock
379 * If lock has failed, callback is executed with locked=false
381 static void process_callbacks(struct lock_context *lock_ctx, bool locked)
383 struct lock_request *request;
384 bool auto_mark = lock_ctx->auto_mark;
386 if (auto_mark && locked) {
387 switch (lock_ctx->type) {
389 tdb_chainlock_mark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
393 ctdb_lockdb_mark(lock_ctx->ctdb_db);
396 case LOCK_ALLDB_PRIO:
397 ctdb_lockall_mark_prio(lock_ctx->ctdb, lock_ctx->priority);
401 ctdb_lockall_mark(lock_ctx->ctdb);
406 request = lock_ctx->request;
408 /* Since request may be freed in the callback, unset the lock
409 * context, so request destructor will not free lock context.
411 request->lctx = NULL;
414 /* Since request may be freed in the callback, unset the request */
415 lock_ctx->request = NULL;
417 request->callback(request->private_data, locked);
424 switch (lock_ctx->type) {
426 tdb_chainlock_unmark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
430 ctdb_lockdb_unmark(lock_ctx->ctdb_db);
433 case LOCK_ALLDB_PRIO:
434 ctdb_lockall_unmark_prio(lock_ctx->ctdb, lock_ctx->priority);
438 ctdb_lockall_unmark(lock_ctx->ctdb);
443 talloc_free(lock_ctx);
447 static int lock_bucket_id(double t)
449 double ms = 1.e-3, s = 1;
454 } else if (t < 10*ms) {
456 } else if (t < 100*ms) {
458 } else if (t < 1*s) {
460 } else if (t < 2*s) {
462 } else if (t < 4*s) {
464 } else if (t < 8*s) {
466 } else if (t < 16*s) {
468 } else if (t < 32*s) {
470 } else if (t < 64*s) {
480 * Callback routine when the required locks are obtained.
481 * Called from parent context
483 static void ctdb_lock_handler(struct tevent_context *ev,
484 struct tevent_fd *tfd,
488 struct lock_context *lock_ctx;
494 lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
496 /* cancel the timeout event */
497 TALLOC_FREE(lock_ctx->ttimer);
499 t = timeval_elapsed(&lock_ctx->start_time);
500 id = lock_bucket_id(t);
502 /* Read the status from the child process */
503 if (sys_read(lock_ctx->fd[0], &c, 1) != 1) {
506 locked = (c == 0 ? true : false);
509 /* Update statistics */
510 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_calls);
511 if (lock_ctx->ctdb_db) {
512 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_calls);
516 if (lock_ctx->ctdb_db) {
517 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.buckets[id]);
518 CTDB_UPDATE_LATENCY(lock_ctx->ctdb, lock_ctx->ctdb_db,
519 lock_type_str[lock_ctx->type], locks.latency,
520 lock_ctx->start_time);
522 CTDB_UPDATE_DB_LATENCY(lock_ctx->ctdb_db, lock_type_str[lock_ctx->type], locks.latency, t);
523 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.buckets[id]);
526 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_failed);
527 if (lock_ctx->ctdb_db) {
528 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_failed);
532 process_callbacks(lock_ctx, locked);
537 * Callback routine when required locks are not obtained within timeout
538 * Called from parent context
540 static void ctdb_lock_timeout_handler(struct tevent_context *ev,
541 struct tevent_timer *ttimer,
542 struct timeval current_time,
545 static char debug_locks[PATH_MAX+1] = "";
546 struct lock_context *lock_ctx;
547 struct ctdb_context *ctdb;
552 lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
553 ctdb = lock_ctx->ctdb;
555 /* If a node stopped/banned, don't spam the logs */
556 if (ctdb->nodes[ctdb->pnn]->flags & NODE_FLAGS_INACTIVE) {
557 lock_ctx->ttimer = NULL;
561 elapsed_time = timeval_elapsed(&lock_ctx->start_time);
562 if (lock_ctx->ctdb_db) {
564 ("Unable to get %s lock on database %s for %.0lf seconds\n",
565 (lock_ctx->type == LOCK_RECORD ? "RECORD" : "DB"),
566 lock_ctx->ctdb_db->db_name, elapsed_time));
569 ("Unable to get ALLDB locks for %.0lf seconds\n",
573 if (ctdb_set_helper("lock debugging helper",
574 debug_locks, sizeof(debug_locks),
576 getenv("CTDB_BASE"), "debug_locks.sh")) {
579 execl(debug_locks, debug_locks, NULL);
582 ctdb_track_child(ctdb, pid);
586 " Unable to setup lock debugging\n"));
589 /* Back-off logging if lock is not obtained for a long time */
590 if (elapsed_time < 100.0) {
592 } else if (elapsed_time < 1000.0) {
598 /* reset the timeout timer */
599 // talloc_free(lock_ctx->ttimer);
600 lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
602 timeval_current_ofs(new_timer, 0),
603 ctdb_lock_timeout_handler,
608 static int db_count_handler(struct ctdb_db_context *ctdb_db, void *private_data)
610 int *count = (int *)private_data;
617 static int db_flags(struct ctdb_db_context *ctdb_db)
619 int tdb_flags = TDB_DEFAULT;
621 #ifdef TDB_MUTEX_LOCKING
622 if (!ctdb_db->persistent && ctdb_db->ctdb->tunable.mutex_enabled) {
623 tdb_flags = (TDB_MUTEX_LOCKING | TDB_CLEAR_IF_FIRST);
634 static int db_name_handler(struct ctdb_db_context *ctdb_db, void *private_data)
636 struct db_namelist *list = (struct db_namelist *)private_data;
638 list->names[list->n] = talloc_strdup(list->names, ctdb_db->db_path);
639 list->names[list->n+1] = talloc_asprintf(list->names, "0x%x",
646 static bool lock_helper_args(TALLOC_CTX *mem_ctx,
647 struct lock_context *lock_ctx, int fd,
648 int *argc, const char ***argv)
650 struct ctdb_context *ctdb = lock_ctx->ctdb;
651 const char **args = NULL;
654 struct db_namelist list;
656 switch (lock_ctx->type) {
665 case LOCK_ALLDB_PRIO:
667 ctdb_db_prio_iterator(ctdb, lock_ctx->priority,
668 db_count_handler, &nargs);
673 for (priority=1; priority<NUM_DB_PRIORITIES; priority++) {
674 ctdb_db_prio_iterator(ctdb, priority,
675 db_count_handler, &nargs);
680 /* Add extra argument for null termination */
683 args = talloc_array(mem_ctx, const char *, nargs);
688 args[0] = talloc_asprintf(args, "%d", getpid());
689 args[1] = talloc_asprintf(args, "%d", fd);
691 switch (lock_ctx->type) {
693 args[2] = talloc_strdup(args, "RECORD");
694 args[3] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
695 args[4] = talloc_asprintf(args, "0x%x",
696 db_flags(lock_ctx->ctdb_db));
697 if (lock_ctx->key.dsize == 0) {
698 args[5] = talloc_strdup(args, "NULL");
700 args[5] = hex_encode_talloc(args, lock_ctx->key.dptr, lock_ctx->key.dsize);
705 args[2] = talloc_strdup(args, "DB");
706 args[3] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
707 args[4] = talloc_asprintf(args, "0x%x",
708 db_flags(lock_ctx->ctdb_db));
711 case LOCK_ALLDB_PRIO:
712 args[2] = talloc_strdup(args, "DB");
715 ctdb_db_prio_iterator(ctdb, lock_ctx->priority,
716 db_name_handler, &list);
720 args[2] = talloc_strdup(args, "DB");
723 for (priority=1; priority<NUM_DB_PRIORITIES; priority++) {
724 ctdb_db_prio_iterator(ctdb, priority,
725 db_name_handler, &list);
730 /* Make sure last argument is NULL */
731 args[nargs-1] = NULL;
733 for (i=0; i<nargs-1; i++) {
734 if (args[i] == NULL) {
746 * Find a lock request that can be scheduled
748 static struct lock_context *ctdb_find_lock_context(struct ctdb_context *ctdb)
750 struct lock_context *lock_ctx, *next_ctx;
751 struct ctdb_db_context *ctdb_db;
753 /* First check if there are database lock requests */
755 for (lock_ctx = ctdb->lock_pending; lock_ctx != NULL;
756 lock_ctx = next_ctx) {
758 if (lock_ctx->request != NULL) {
759 /* Found a lock context with a request */
763 next_ctx = lock_ctx->next;
765 DEBUG(DEBUG_INFO, ("Removing lock context without lock "
767 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
768 CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
769 if (lock_ctx->ctdb_db) {
770 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db,
773 talloc_free(lock_ctx);
776 /* Next check database queues */
777 for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
778 if (ctdb_db->lock_num_current ==
779 ctdb->tunable.lock_processes_per_db) {
783 for (lock_ctx = ctdb_db->lock_pending; lock_ctx != NULL;
784 lock_ctx = next_ctx) {
786 next_ctx = lock_ctx->next;
788 if (lock_ctx->request != NULL) {
792 DEBUG(DEBUG_INFO, ("Removing lock context without "
794 DLIST_REMOVE(ctdb_db->lock_pending, lock_ctx);
795 CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
796 CTDB_DECREMENT_DB_STAT(ctdb_db, locks.num_pending);
797 talloc_free(lock_ctx);
805 * Schedule a new lock child process
806 * Set up callback handler and timeout handler
808 static void ctdb_lock_schedule(struct ctdb_context *ctdb)
810 struct lock_context *lock_ctx;
813 static char prog[PATH_MAX+1] = "";
816 if (!ctdb_set_helper("lock helper",
819 CTDB_HELPER_BINDIR, "ctdb_lock_helper")) {
820 ctdb_die(ctdb, __location__
821 " Unable to set lock helper\n");
824 /* Find a lock context with requests */
825 lock_ctx = ctdb_find_lock_context(ctdb);
826 if (lock_ctx == NULL) {
830 lock_ctx->child = -1;
831 ret = pipe(lock_ctx->fd);
833 DEBUG(DEBUG_ERR, ("Failed to create pipe in ctdb_lock_schedule\n"));
837 set_close_on_exec(lock_ctx->fd[0]);
839 /* Create data for child process */
840 tmp_ctx = talloc_new(lock_ctx);
841 if (tmp_ctx == NULL) {
842 DEBUG(DEBUG_ERR, ("Failed to allocate memory for helper args\n"));
843 close(lock_ctx->fd[0]);
844 close(lock_ctx->fd[1]);
848 if (! ctdb->do_setsched) {
849 ret = setenv("CTDB_NOSETSCHED", "1", 1);
852 ("Failed to set CTDB_NOSETSCHED variable\n"));
856 /* Create arguments for lock helper */
857 if (!lock_helper_args(tmp_ctx, lock_ctx, lock_ctx->fd[1],
859 DEBUG(DEBUG_ERR, ("Failed to create lock helper args\n"));
860 close(lock_ctx->fd[0]);
861 close(lock_ctx->fd[1]);
862 talloc_free(tmp_ctx);
866 if (!ctdb_vfork_with_logging(lock_ctx, ctdb, "lock_helper",
867 prog, argc, (const char **)args,
868 NULL, NULL, &lock_ctx->child)) {
869 DEBUG(DEBUG_ERR, ("Failed to create a child in ctdb_lock_schedule\n"));
870 close(lock_ctx->fd[0]);
871 close(lock_ctx->fd[1]);
872 talloc_free(tmp_ctx);
877 close(lock_ctx->fd[1]);
879 talloc_free(tmp_ctx);
881 /* Set up timeout handler */
882 lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
884 timeval_current_ofs(10, 0),
885 ctdb_lock_timeout_handler,
887 if (lock_ctx->ttimer == NULL) {
888 ctdb_kill(ctdb, lock_ctx->child, SIGKILL);
889 lock_ctx->child = -1;
890 close(lock_ctx->fd[0]);
894 /* Set up callback */
895 lock_ctx->tfd = tevent_add_fd(ctdb->ev,
901 if (lock_ctx->tfd == NULL) {
902 TALLOC_FREE(lock_ctx->ttimer);
903 ctdb_kill(ctdb, lock_ctx->child, SIGKILL);
904 lock_ctx->child = -1;
905 close(lock_ctx->fd[0]);
908 tevent_fd_set_auto_close(lock_ctx->tfd);
910 /* Move the context from pending to current */
911 if (lock_ctx->type == LOCK_RECORD) {
912 DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
913 DLIST_ADD_END(lock_ctx->ctdb_db->lock_current, lock_ctx);
915 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
916 DLIST_ADD_END(ctdb->lock_current, lock_ctx);
918 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
919 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_current);
920 if (lock_ctx->ctdb_db) {
921 lock_ctx->ctdb_db->lock_num_current++;
922 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
923 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
929 * Lock record / db depending on type
931 static struct lock_request *ctdb_lock_internal(TALLOC_CTX *mem_ctx,
932 struct ctdb_context *ctdb,
933 struct ctdb_db_context *ctdb_db,
936 void (*callback)(void *, bool),
941 struct lock_context *lock_ctx = NULL;
942 struct lock_request *request;
944 if (callback == NULL) {
945 DEBUG(DEBUG_WARNING, ("No callback function specified, not locking\n"));
949 lock_ctx = talloc_zero(ctdb, struct lock_context);
950 if (lock_ctx == NULL) {
951 DEBUG(DEBUG_ERR, ("Failed to create a new lock context\n"));
955 if ((request = talloc_zero(mem_ctx, struct lock_request)) == NULL) {
956 talloc_free(lock_ctx);
960 lock_ctx->type = type;
961 lock_ctx->ctdb = ctdb;
962 lock_ctx->ctdb_db = ctdb_db;
963 lock_ctx->key.dsize = key.dsize;
965 lock_ctx->key.dptr = talloc_memdup(lock_ctx, key.dptr, key.dsize);
966 if (lock_ctx->key.dptr == NULL) {
967 DEBUG(DEBUG_ERR, (__location__ "Memory allocation error\n"));
968 talloc_free(lock_ctx);
969 talloc_free(request);
972 lock_ctx->key_hash = ctdb_hash(&key);
974 lock_ctx->key.dptr = NULL;
976 lock_ctx->priority = priority;
977 lock_ctx->auto_mark = auto_mark;
979 lock_ctx->request = request;
980 lock_ctx->child = -1;
982 /* Non-record locks are required by recovery and should be scheduled
983 * immediately, so keep them at the head of the pending queue.
985 if (lock_ctx->type == LOCK_RECORD) {
986 DLIST_ADD_END(ctdb_db->lock_pending, lock_ctx);
988 DLIST_ADD_END(ctdb->lock_pending, lock_ctx);
990 CTDB_INCREMENT_STAT(ctdb, locks.num_pending);
992 CTDB_INCREMENT_DB_STAT(ctdb_db, locks.num_pending);
995 /* Start the timer when we activate the context */
996 lock_ctx->start_time = timeval_current();
998 request->lctx = lock_ctx;
999 request->callback = callback;
1000 request->private_data = private_data;
1002 talloc_set_destructor(request, ctdb_lock_request_destructor);
1003 talloc_set_destructor(lock_ctx, ctdb_lock_context_destructor);
1005 ctdb_lock_schedule(ctdb);
1012 * obtain a lock on a record in a database
1014 struct lock_request *ctdb_lock_record(TALLOC_CTX *mem_ctx,
1015 struct ctdb_db_context *ctdb_db,
1018 void (*callback)(void *, bool),
1021 return ctdb_lock_internal(mem_ctx,
1034 * obtain a lock on a database
1036 struct lock_request *ctdb_lock_db(TALLOC_CTX *mem_ctx,
1037 struct ctdb_db_context *ctdb_db,
1039 void (*callback)(void *, bool),
1042 return ctdb_lock_internal(mem_ctx,
1055 * obtain locks on all databases of specified priority
1057 struct lock_request *ctdb_lock_alldb_prio(TALLOC_CTX *mem_ctx,
1058 struct ctdb_context *ctdb,
1061 void (*callback)(void *, bool),
1064 if (priority < 1 || priority > NUM_DB_PRIORITIES) {
1065 DEBUG(DEBUG_ERR, ("Invalid db priority: %u\n", priority));
1069 return ctdb_lock_internal(mem_ctx,
1082 * obtain locks on all databases
1084 struct lock_request *ctdb_lock_alldb(TALLOC_CTX *mem_ctx,
1085 struct ctdb_context *ctdb,
1087 void (*callback)(void *, bool),
1090 return ctdb_lock_internal(mem_ctx,