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 "include/ctdb_private.h"
22 #include "include/ctdb_protocol.h"
26 #include "system/filesys.h"
27 #include "lib/util/dlinklist.h"
30 * Non-blocking Locking API
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
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
43 * auto_mark - whether to mark/unmark DBs in before/after callback
46 /* FIXME: Add a tunable max_lock_processes_per_db */
47 #define MAX_LOCK_PROCESSES_PER_DB (100)
56 static const char * const lock_type_str[] = {
65 /* lock_context is the common part for a lock request */
67 struct lock_context *next, *prev;
69 struct ctdb_context *ctdb;
70 struct ctdb_db_context *ctdb_db;
74 struct lock_request *req_queue;
77 struct tevent_fd *tfd;
78 struct tevent_timer *ttimer;
81 struct timeval start_time;
85 /* lock_request is the client specific part for a lock request */
87 struct lock_request *next, *prev;
88 struct lock_context *lctx;
89 void (*callback)(void *, bool);
95 * Support samba 3.6.x (and older) versions which do not set db priority.
97 * By default, all databases are set to priority 1. So only when priority
98 * is set to 1, check for databases that need higher priority.
100 static bool later_db(struct ctdb_context *ctdb, const char *name)
102 if (ctdb->tunable.samba3_hack == 0) {
106 if (strstr(name, "brlock") ||
107 strstr(name, "g_lock") ||
108 strstr(name, "notify_onelevel") ||
109 strstr(name, "serverid") ||
110 strstr(name, "xattr_tdb")) {
117 typedef int (*db_handler_t)(struct ctdb_db_context *ctdb_db,
121 static int ctdb_db_iterator(struct ctdb_context *ctdb, uint32_t priority,
122 db_handler_t handler, void *private_data)
124 struct ctdb_db_context *ctdb_db;
127 for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
128 if (ctdb_db->priority != priority) {
131 if (later_db(ctdb, ctdb_db->db_name)) {
134 ret = handler(ctdb_db, priority, private_data);
140 /* If priority != 1, later_db check is not required and can return */
145 for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
146 if (!later_db(ctdb, ctdb_db->db_name)) {
149 ret = handler(ctdb_db, priority, private_data);
160 * lock all databases - mark only
162 static int db_lock_mark_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
165 int tdb_transaction_write_lock_mark(struct tdb_context *);
167 DEBUG(DEBUG_INFO, ("marking locked database %s, priority:%u\n",
168 ctdb_db->db_name, priority));
170 if (tdb_transaction_write_lock_mark(ctdb_db->ltdb->tdb) != 0) {
171 DEBUG(DEBUG_ERR, ("Failed to mark (transaction lock) database %s\n",
176 if (tdb_lockall_mark(ctdb_db->ltdb->tdb) != 0) {
177 DEBUG(DEBUG_ERR, ("Failed to mark (all lock) database %s\n",
185 int ctdb_lockall_mark_prio(struct ctdb_context *ctdb, uint32_t priority)
188 * This function is only used by the main dameon during recovery.
189 * At this stage, the databases have already been locked, by a
190 * dedicated child process. The freeze_mode variable is used to track
191 * whether the actual locks are held by the child process or not.
194 if (ctdb->freeze_mode[priority] != CTDB_FREEZE_FROZEN) {
195 DEBUG(DEBUG_ERR, ("Attempt to mark all databases locked when not frozen\n"));
199 return ctdb_db_iterator(ctdb, priority, db_lock_mark_handler, NULL);
202 static int ctdb_lockall_mark(struct ctdb_context *ctdb)
206 for (priority=1; priority<=NUM_DB_PRIORITIES; priority++) {
207 if (ctdb_db_iterator(ctdb, priority, db_lock_mark_handler, NULL) != 0) {
217 * lock all databases - unmark only
219 static int db_lock_unmark_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
222 int tdb_transaction_write_lock_unmark(struct tdb_context *);
224 DEBUG(DEBUG_INFO, ("unmarking locked database %s, priority:%u\n",
225 ctdb_db->db_name, priority));
227 if (tdb_transaction_write_lock_unmark(ctdb_db->ltdb->tdb) != 0) {
228 DEBUG(DEBUG_ERR, ("Failed to unmark (transaction lock) database %s\n",
233 if (tdb_lockall_unmark(ctdb_db->ltdb->tdb) != 0) {
234 DEBUG(DEBUG_ERR, ("Failed to unmark (all lock) database %s\n",
242 int ctdb_lockall_unmark_prio(struct ctdb_context *ctdb, uint32_t priority)
245 * This function is only used by the main dameon during recovery.
246 * At this stage, the databases have already been locked, by a
247 * dedicated child process. The freeze_mode variable is used to track
248 * whether the actual locks are held by the child process or not.
251 if (ctdb->freeze_mode[priority] != CTDB_FREEZE_FROZEN) {
252 DEBUG(DEBUG_ERR, ("Attempt to unmark all databases locked when not frozen\n"));
256 return ctdb_db_iterator(ctdb, priority, db_lock_unmark_handler, NULL);
259 static int ctdb_lockall_unmark(struct ctdb_context *ctdb)
263 for (priority=NUM_DB_PRIORITIES; priority>0; priority--) {
264 if (ctdb_db_iterator(ctdb, priority, db_lock_unmark_handler, NULL) != 0) {
273 static void ctdb_lock_schedule(struct ctdb_context *ctdb);
276 * Destructor to kill the child locking process
278 static int ctdb_lock_context_destructor(struct lock_context *lock_ctx)
280 if (lock_ctx->child > 0) {
281 ctdb_kill(lock_ctx->ctdb, lock_ctx->child, SIGKILL);
282 DLIST_REMOVE(lock_ctx->ctdb->lock_current, lock_ctx);
283 if (lock_ctx->ctdb_db) {
284 lock_ctx->ctdb_db->lock_num_current--;
286 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_current);
287 if (lock_ctx->type == LOCK_RECORD || lock_ctx->type == LOCK_DB) {
288 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
291 DLIST_REMOVE(lock_ctx->ctdb->lock_pending, lock_ctx);
292 lock_ctx->ctdb->lock_num_pending--;
293 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
294 if (lock_ctx->type == LOCK_RECORD || lock_ctx->type == LOCK_DB) {
295 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
299 ctdb_lock_schedule(lock_ctx->ctdb);
306 * Destructor to remove lock request
308 static int ctdb_lock_request_destructor(struct lock_request *lock_request)
310 DLIST_REMOVE(lock_request->lctx->req_queue, lock_request);
315 void ctdb_lock_free_request_context(struct lock_request *lock_req)
317 struct lock_context *lock_ctx;
319 lock_ctx = lock_req->lctx;
320 talloc_free(lock_req);
321 talloc_free(lock_ctx);
326 * Process all the callbacks waiting for lock
328 * If lock has failed, callback is executed with locked=false
330 static void process_callbacks(struct lock_context *lock_ctx, bool locked)
332 struct lock_request *request, *next;
334 if (lock_ctx->auto_mark && locked) {
335 switch (lock_ctx->type) {
337 tdb_chainlock_mark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
341 tdb_lockall_mark(lock_ctx->ctdb_db->ltdb->tdb);
344 case LOCK_ALLDB_PRIO:
345 ctdb_lockall_mark_prio(lock_ctx->ctdb, lock_ctx->priority);
349 ctdb_lockall_mark(lock_ctx->ctdb);
354 /* Iterate through all callbacks */
355 request = lock_ctx->req_queue;
357 if (lock_ctx->auto_mark) {
358 /* Reset the destructor, so request is not removed from the list */
359 talloc_set_destructor(request, NULL);
362 /* In case, callback frees the request, store next */
363 next = request->next;
364 request->callback(request->private_data, locked);
368 if (lock_ctx->auto_mark && locked) {
369 switch (lock_ctx->type) {
371 tdb_chainlock_unmark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
375 tdb_lockall_unmark(lock_ctx->ctdb_db->ltdb->tdb);
378 case LOCK_ALLDB_PRIO:
379 ctdb_lockall_unmark_prio(lock_ctx->ctdb, lock_ctx->priority);
383 ctdb_lockall_unmark(lock_ctx->ctdb);
390 static int lock_bucket_id(double t)
392 double ms = 1.e-3, s = 1;
397 } else if (t < 10*ms) {
399 } else if (t < 100*ms) {
401 } else if (t < 1*s) {
403 } else if (t < 2*s) {
405 } else if (t < 4*s) {
407 } else if (t < 8*s) {
409 } else if (t < 16*s) {
411 } else if (t < 32*s) {
413 } else if (t < 64*s) {
423 * Callback routine when the required locks are obtained.
424 * Called from parent context
426 static void ctdb_lock_handler(struct tevent_context *ev,
427 struct tevent_fd *tfd,
431 struct lock_context *lock_ctx;
432 TALLOC_CTX *tmp_ctx = NULL;
438 lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
440 /* cancel the timeout event */
441 if (lock_ctx->ttimer) {
442 TALLOC_FREE(lock_ctx->ttimer);
445 t = timeval_elapsed(&lock_ctx->start_time);
446 id = lock_bucket_id(t);
448 if (lock_ctx->auto_mark) {
449 tmp_ctx = talloc_new(ev);
450 talloc_steal(tmp_ctx, lock_ctx);
453 /* Read the status from the child process */
454 if (read(lock_ctx->fd[0], &c, 1) != 1) {
457 locked = (c == 0 ? true : false);
460 /* Update statistics */
461 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
462 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_calls);
463 if (lock_ctx->ctdb_db) {
464 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
465 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_calls);
469 if (lock_ctx->ctdb_db) {
470 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.buckets[id]);
471 CTDB_UPDATE_LATENCY(lock_ctx->ctdb, lock_ctx->ctdb_db,
472 lock_type_str[lock_ctx->type], locks.latency,
473 lock_ctx->start_time);
475 CTDB_UPDATE_DB_LATENCY(lock_ctx->ctdb_db, lock_type_str[lock_ctx->type], locks.latency, t);
476 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.buckets[id]);
479 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_failed);
480 if (lock_ctx->ctdb_db) {
481 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_failed);
485 process_callbacks(lock_ctx, locked);
487 if (lock_ctx->auto_mark) {
488 talloc_free(tmp_ctx);
494 * Callback routine when required locks are not obtained within timeout
495 * Called from parent context
497 static void ctdb_lock_timeout_handler(struct tevent_context *ev,
498 struct tevent_timer *ttimer,
499 struct timeval current_time,
502 static const char * debug_locks = NULL;
503 struct lock_context *lock_ctx;
504 struct ctdb_context *ctdb;
507 lock_ctx = talloc_get_type_abort(private_data, struct lock_context);
508 ctdb = lock_ctx->ctdb;
510 if (lock_ctx->type == LOCK_RECORD || lock_ctx->type == LOCK_DB) {
512 ("Unable to get %s lock on database %s for %.0lf seconds\n",
513 (lock_ctx->type == LOCK_RECORD ? "RECORD" : "DB"),
514 lock_ctx->ctdb_db->db_name,
515 timeval_elapsed(&lock_ctx->start_time)));
518 ("Unable to get ALLDB locks for %.0lf seconds\n",
519 timeval_elapsed(&lock_ctx->start_time)));
522 /* Fire a child process to find the blocking process. */
523 if (debug_locks == NULL) {
524 debug_locks = getenv("CTDB_DEBUG_LOCKS");
525 if (debug_locks == NULL) {
526 debug_locks = talloc_asprintf(ctdb,
528 getenv("CTDB_BASE"));
531 if (debug_locks != NULL) {
534 execl(debug_locks, debug_locks, NULL);
537 ctdb_track_child(ctdb, pid);
541 " Unable to setup lock debugging - no memory?\n"));
544 /* reset the timeout timer */
545 // talloc_free(lock_ctx->ttimer);
546 lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
548 timeval_current_ofs(10, 0),
549 ctdb_lock_timeout_handler,
554 static int db_count_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
557 int *count = (int *)private_data;
569 static int db_name_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
572 struct db_namelist *list = (struct db_namelist *)private_data;
574 list->names[list->n] = talloc_strdup(list->names, ctdb_db->db_path);
580 static char **lock_helper_args(TALLOC_CTX *mem_ctx, struct lock_context *lock_ctx, int fd)
582 struct ctdb_context *ctdb = lock_ctx->ctdb;
586 struct db_namelist list;
588 switch (lock_ctx->type) {
597 case LOCK_ALLDB_PRIO:
599 ctdb_db_iterator(ctdb, lock_ctx->priority, db_count_handler, &nargs);
604 for (priority=1; priority<NUM_DB_PRIORITIES; priority++) {
605 ctdb_db_iterator(ctdb, priority, db_count_handler, &nargs);
610 /* Add extra argument for null termination */
613 args = talloc_array(mem_ctx, char *, nargs);
618 args[0] = talloc_strdup(args, "ctdb_lock_helper");
619 args[1] = talloc_asprintf(args, "%d", getpid());
620 args[2] = talloc_asprintf(args, "%d", fd);
622 switch (lock_ctx->type) {
624 args[3] = talloc_strdup(args, "RECORD");
625 args[4] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
626 if (lock_ctx->key.dsize == 0) {
627 args[5] = talloc_strdup(args, "NULL");
629 args[5] = hex_encode_talloc(args, lock_ctx->key.dptr, lock_ctx->key.dsize);
634 args[3] = talloc_strdup(args, "DB");
635 args[4] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
638 case LOCK_ALLDB_PRIO:
639 args[3] = talloc_strdup(args, "DB");
642 ctdb_db_iterator(ctdb, lock_ctx->priority, db_name_handler, &list);
646 args[3] = talloc_strdup(args, "DB");
649 for (priority=1; priority<NUM_DB_PRIORITIES; priority++) {
650 ctdb_db_iterator(ctdb, priority, db_name_handler, &list);
655 /* Make sure last argument is NULL */
656 args[nargs-1] = NULL;
658 for (i=0; i<nargs-1; i++) {
659 if (args[i] == NULL) {
670 * Find the lock context of a given type
672 static struct lock_context *find_lock_context(struct lock_context *lock_list,
673 struct ctdb_db_context *ctdb_db,
679 struct lock_context *lock_ctx;
681 /* Search active locks */
682 for (lock_ctx=lock_list; lock_ctx; lock_ctx=lock_ctx->next) {
683 if (lock_ctx->type != type) {
687 switch (lock_ctx->type) {
689 if (ctdb_db == lock_ctx->ctdb_db &&
690 key_hash == lock_ctx->key_hash) {
696 if (ctdb_db == lock_ctx->ctdb_db) {
701 case LOCK_ALLDB_PRIO:
702 if (priority == lock_ctx->priority) {
713 /* Did not find the lock context we are searching for */
723 * Schedule a new lock child process
724 * Set up callback handler and timeout handler
726 static void ctdb_lock_schedule(struct ctdb_context *ctdb)
728 struct lock_context *lock_ctx, *next_ctx, *active_ctx;
731 const char *helper = BINDIR "/ctdb_lock_helper";
732 static const char *prog = NULL;
738 t = getenv("CTDB_LOCK_HELPER");
740 prog = talloc_strdup(ctdb, t);
742 prog = talloc_strdup(ctdb, helper);
744 CTDB_NO_MEMORY_VOID(ctdb, prog);
747 if (ctdb->lock_pending == NULL) {
751 /* Find a lock context with requests */
752 lock_ctx = ctdb->lock_pending;
753 while (lock_ctx != NULL) {
754 next_ctx = lock_ctx->next;
755 if (! lock_ctx->req_queue) {
756 DEBUG(DEBUG_INFO, ("Removing lock context without lock requests\n"));
757 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
758 ctdb->lock_num_pending--;
759 CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
760 if (lock_ctx->ctdb_db) {
761 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
763 talloc_free(lock_ctx);
765 active_ctx = find_lock_context(ctdb->lock_current, lock_ctx->ctdb_db,
766 lock_ctx->key, lock_ctx->priority,
767 lock_ctx->type, lock_ctx->key_hash);
768 if (active_ctx == NULL) {
769 if (lock_ctx->ctdb_db == NULL ||
770 lock_ctx->ctdb_db->lock_num_current < MAX_LOCK_PROCESSES_PER_DB) {
771 /* Found a lock context with lock requests */
776 /* There is already a child waiting for the
777 * same key. So don't schedule another child
784 if (lock_ctx == NULL) {
788 lock_ctx->child = -1;
789 ret = pipe(lock_ctx->fd);
791 DEBUG(DEBUG_ERR, ("Failed to create pipe in ctdb_lock_schedule\n"));
795 set_close_on_exec(lock_ctx->fd[0]);
797 /* Create data for child process */
798 tmp_ctx = talloc_new(lock_ctx);
799 if (tmp_ctx == NULL) {
800 DEBUG(DEBUG_ERR, ("Failed to allocate memory for helper args\n"));
801 close(lock_ctx->fd[0]);
802 close(lock_ctx->fd[1]);
806 /* Create arguments for lock helper */
807 args = lock_helper_args(tmp_ctx, lock_ctx, lock_ctx->fd[1]);
809 DEBUG(DEBUG_ERR, ("Failed to create lock helper args\n"));
810 close(lock_ctx->fd[0]);
811 close(lock_ctx->fd[1]);
812 talloc_free(tmp_ctx);
816 lock_ctx->child = vfork();
818 if (lock_ctx->child == (pid_t)-1) {
819 DEBUG(DEBUG_ERR, ("Failed to create a child in ctdb_lock_schedule\n"));
820 close(lock_ctx->fd[0]);
821 close(lock_ctx->fd[1]);
822 talloc_free(tmp_ctx);
828 if (lock_ctx->child == 0) {
829 ret = execv(prog, args);
831 DEBUG(DEBUG_ERR, ("Failed to execute helper %s (%d, %s)\n",
832 prog, errno, strerror(errno)));
838 ctdb_track_child(ctdb, lock_ctx->child);
839 close(lock_ctx->fd[1]);
841 talloc_set_destructor(lock_ctx, ctdb_lock_context_destructor);
843 talloc_free(tmp_ctx);
845 /* Set up timeout handler */
846 lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
848 timeval_current_ofs(10, 0),
849 ctdb_lock_timeout_handler,
851 if (lock_ctx->ttimer == NULL) {
852 ctdb_kill(ctdb, lock_ctx->child, SIGKILL);
853 lock_ctx->child = -1;
854 talloc_set_destructor(lock_ctx, NULL);
855 close(lock_ctx->fd[0]);
859 /* Set up callback */
860 lock_ctx->tfd = tevent_add_fd(ctdb->ev,
866 if (lock_ctx->tfd == NULL) {
867 TALLOC_FREE(lock_ctx->ttimer);
868 ctdb_kill(ctdb, lock_ctx->child, SIGKILL);
869 lock_ctx->child = -1;
870 talloc_set_destructor(lock_ctx, NULL);
871 close(lock_ctx->fd[0]);
874 tevent_fd_set_auto_close(lock_ctx->tfd);
876 /* Move the context from pending to current */
877 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
878 ctdb->lock_num_pending--;
879 DLIST_ADD_END(ctdb->lock_current, lock_ctx, NULL);
880 if (lock_ctx->ctdb_db) {
881 lock_ctx->ctdb_db->lock_num_current++;
882 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_current);
883 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
889 * Lock record / db depending on type
891 static struct lock_request *ctdb_lock_internal(struct ctdb_context *ctdb,
892 struct ctdb_db_context *ctdb_db,
895 void (*callback)(void *, bool),
900 struct lock_context *lock_ctx = NULL;
901 struct lock_request *request;
903 if (callback == NULL) {
904 DEBUG(DEBUG_WARNING, ("No callback function specified, not locking\n"));
909 /* Disable this optimization to ensure first-in-first-out fair
910 * scheduling of lock requests */
912 /* get a context for this key - search only the pending contexts,
913 * current contexts might in the middle of processing callbacks */
914 lock_ctx = find_lock_context(ctdb->lock_pending, ctdb_db, key, priority, type);
917 /* No existing context, create one */
918 if (lock_ctx == NULL) {
919 lock_ctx = talloc_zero(ctdb, struct lock_context);
920 if (lock_ctx == NULL) {
921 DEBUG(DEBUG_ERR, ("Failed to create a new lock context\n"));
925 lock_ctx->type = type;
926 lock_ctx->ctdb = ctdb;
927 lock_ctx->ctdb_db = ctdb_db;
928 lock_ctx->key.dsize = key.dsize;
930 lock_ctx->key.dptr = talloc_memdup(lock_ctx, key.dptr, key.dsize);
931 if (lock_ctx->key.dptr == NULL) {
932 DEBUG(DEBUG_ERR, (__location__ "Memory allocation error\n"));
933 talloc_free(lock_ctx);
936 lock_ctx->key_hash = ctdb_hash(&key);
938 lock_ctx->key.dptr = NULL;
940 lock_ctx->priority = priority;
941 lock_ctx->auto_mark = auto_mark;
943 lock_ctx->child = -1;
944 lock_ctx->block_child = -1;
946 DLIST_ADD_END(ctdb->lock_pending, lock_ctx, NULL);
947 ctdb->lock_num_pending++;
948 CTDB_INCREMENT_STAT(ctdb, locks.num_pending);
950 CTDB_INCREMENT_DB_STAT(ctdb_db, locks.num_pending);
953 /* Start the timer when we activate the context */
954 lock_ctx->start_time = timeval_current();
957 if ((request = talloc_zero(lock_ctx, struct lock_request)) == NULL) {
958 talloc_free(lock_ctx);
962 request->lctx = lock_ctx;
963 request->callback = callback;
964 request->private_data = private_data;
966 talloc_set_destructor(request, ctdb_lock_request_destructor);
967 DLIST_ADD_END(lock_ctx->req_queue, request, NULL);
969 ctdb_lock_schedule(ctdb);
976 * obtain a lock on a record in a database
978 struct lock_request *ctdb_lock_record(struct ctdb_db_context *ctdb_db,
981 void (*callback)(void *, bool),
984 return ctdb_lock_internal(ctdb_db->ctdb,
996 * obtain a lock on a database
998 struct lock_request *ctdb_lock_db(struct ctdb_db_context *ctdb_db,
1000 void (*callback)(void *, bool),
1003 return ctdb_lock_internal(ctdb_db->ctdb,
1015 * obtain locks on all databases of specified priority
1017 struct lock_request *ctdb_lock_alldb_prio(struct ctdb_context *ctdb,
1020 void (*callback)(void *, bool),
1023 if (priority < 1 || priority > NUM_DB_PRIORITIES) {
1024 DEBUG(DEBUG_ERR, ("Invalid db priority: %u\n", priority));
1028 return ctdb_lock_internal(ctdb,
1040 * obtain locks on all databases
1042 struct lock_request *ctdb_lock_alldb(struct ctdb_context *ctdb,
1044 void (*callback)(void *, bool),
1047 return ctdb_lock_internal(ctdb,