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
;
84 /* lock_request is the client specific part for a lock request */
86 struct lock_request
*next
, *prev
;
87 struct lock_context
*lctx
;
88 void (*callback
)(void *, bool);
94 * Support samba 3.6.x (and older) versions which do not set db priority.
96 * By default, all databases are set to priority 1. So only when priority
97 * is set to 1, check for databases that need higher priority.
99 static bool later_db(struct ctdb_context
*ctdb
, const char *name
)
101 if (ctdb
->tunable
.samba3_hack
== 0) {
105 if (strstr(name
, "brlock") ||
106 strstr(name
, "g_lock") ||
107 strstr(name
, "notify_onelevel") ||
108 strstr(name
, "serverid") ||
109 strstr(name
, "xattr_tdb")) {
116 typedef int (*db_handler_t
)(struct ctdb_db_context
*ctdb_db
,
120 static int ctdb_db_iterator(struct ctdb_context
*ctdb
, uint32_t priority
,
121 db_handler_t handler
, void *private_data
)
123 struct ctdb_db_context
*ctdb_db
;
126 for (ctdb_db
= ctdb
->db_list
; ctdb_db
; ctdb_db
= ctdb_db
->next
) {
127 if (ctdb_db
->priority
!= priority
) {
130 if (later_db(ctdb
, ctdb_db
->db_name
)) {
133 ret
= handler(ctdb_db
, priority
, private_data
);
139 /* If priority != 1, later_db check is not required and can return */
144 for (ctdb_db
= ctdb
->db_list
; ctdb_db
; ctdb_db
= ctdb_db
->next
) {
145 if (!later_db(ctdb
, ctdb_db
->db_name
)) {
148 ret
= handler(ctdb_db
, priority
, private_data
);
159 * lock all databases - mark only
161 static int db_lock_mark_handler(struct ctdb_db_context
*ctdb_db
, uint32_t priority
,
164 int tdb_transaction_write_lock_mark(struct tdb_context
*);
166 DEBUG(DEBUG_INFO
, ("marking locked database %s, priority:%u\n",
167 ctdb_db
->db_name
, priority
));
169 if (tdb_transaction_write_lock_mark(ctdb_db
->ltdb
->tdb
) != 0) {
170 DEBUG(DEBUG_ERR
, ("Failed to mark (transaction lock) database %s\n",
175 if (tdb_lockall_mark(ctdb_db
->ltdb
->tdb
) != 0) {
176 DEBUG(DEBUG_ERR
, ("Failed to mark (all lock) database %s\n",
184 int ctdb_lockall_mark_prio(struct ctdb_context
*ctdb
, uint32_t priority
)
187 * This function is only used by the main dameon during recovery.
188 * At this stage, the databases have already been locked, by a
189 * dedicated child process. The freeze_mode variable is used to track
190 * whether the actual locks are held by the child process or not.
193 if (ctdb
->freeze_mode
[priority
] != CTDB_FREEZE_FROZEN
) {
194 DEBUG(DEBUG_ERR
, ("Attempt to mark all databases locked when not frozen\n"));
198 return ctdb_db_iterator(ctdb
, priority
, db_lock_mark_handler
, NULL
);
201 static int ctdb_lockall_mark(struct ctdb_context
*ctdb
)
205 for (priority
=1; priority
<=NUM_DB_PRIORITIES
; priority
++) {
206 if (ctdb_db_iterator(ctdb
, priority
, db_lock_mark_handler
, NULL
) != 0) {
216 * lock all databases - unmark only
218 static int db_lock_unmark_handler(struct ctdb_db_context
*ctdb_db
, uint32_t priority
,
221 int tdb_transaction_write_lock_unmark(struct tdb_context
*);
223 DEBUG(DEBUG_INFO
, ("unmarking locked database %s, priority:%u\n",
224 ctdb_db
->db_name
, priority
));
226 if (tdb_transaction_write_lock_unmark(ctdb_db
->ltdb
->tdb
) != 0) {
227 DEBUG(DEBUG_ERR
, ("Failed to unmark (transaction lock) database %s\n",
232 if (tdb_lockall_unmark(ctdb_db
->ltdb
->tdb
) != 0) {
233 DEBUG(DEBUG_ERR
, ("Failed to unmark (all lock) database %s\n",
241 int ctdb_lockall_unmark_prio(struct ctdb_context
*ctdb
, uint32_t priority
)
244 * This function is only used by the main dameon during recovery.
245 * At this stage, the databases have already been locked, by a
246 * dedicated child process. The freeze_mode variable is used to track
247 * whether the actual locks are held by the child process or not.
250 if (ctdb
->freeze_mode
[priority
] != CTDB_FREEZE_FROZEN
) {
251 DEBUG(DEBUG_ERR
, ("Attempt to unmark all databases locked when not frozen\n"));
255 return ctdb_db_iterator(ctdb
, priority
, db_lock_unmark_handler
, NULL
);
258 static int ctdb_lockall_unmark(struct ctdb_context
*ctdb
)
262 for (priority
=NUM_DB_PRIORITIES
; priority
>0; priority
--) {
263 if (ctdb_db_iterator(ctdb
, priority
, db_lock_unmark_handler
, NULL
) != 0) {
272 static void ctdb_lock_schedule(struct ctdb_context
*ctdb
);
275 * Destructor to kill the child locking process
277 static int ctdb_lock_context_destructor(struct lock_context
*lock_ctx
)
279 if (lock_ctx
->child
> 0) {
280 ctdb_kill(lock_ctx
->ctdb
, lock_ctx
->child
, SIGKILL
);
281 DLIST_REMOVE(lock_ctx
->ctdb
->lock_current
, lock_ctx
);
282 if (lock_ctx
->ctdb_db
) {
283 lock_ctx
->ctdb_db
->lock_num_current
--;
285 CTDB_DECREMENT_STAT(lock_ctx
->ctdb
, locks
.num_current
);
286 if (lock_ctx
->type
== LOCK_RECORD
|| lock_ctx
->type
== LOCK_DB
) {
287 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_current
);
290 DLIST_REMOVE(lock_ctx
->ctdb
->lock_pending
, lock_ctx
);
291 lock_ctx
->ctdb
->lock_num_pending
--;
292 CTDB_DECREMENT_STAT(lock_ctx
->ctdb
, locks
.num_pending
);
293 if (lock_ctx
->type
== LOCK_RECORD
|| lock_ctx
->type
== LOCK_DB
) {
294 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_pending
);
298 ctdb_lock_schedule(lock_ctx
->ctdb
);
305 * Destructor to remove lock request
307 static int ctdb_lock_request_destructor(struct lock_request
*lock_request
)
309 DLIST_REMOVE(lock_request
->lctx
->req_queue
, lock_request
);
314 void ctdb_lock_free_request_context(struct lock_request
*lock_req
)
316 struct lock_context
*lock_ctx
;
318 lock_ctx
= lock_req
->lctx
;
319 talloc_free(lock_req
);
320 talloc_free(lock_ctx
);
325 * Process all the callbacks waiting for lock
327 * If lock has failed, callback is executed with locked=false
329 static void process_callbacks(struct lock_context
*lock_ctx
, bool locked
)
331 struct lock_request
*request
, *next
;
333 if (lock_ctx
->auto_mark
&& locked
) {
334 switch (lock_ctx
->type
) {
336 tdb_chainlock_mark(lock_ctx
->ctdb_db
->ltdb
->tdb
, lock_ctx
->key
);
340 tdb_lockall_mark(lock_ctx
->ctdb_db
->ltdb
->tdb
);
343 case LOCK_ALLDB_PRIO
:
344 ctdb_lockall_mark_prio(lock_ctx
->ctdb
, lock_ctx
->priority
);
348 ctdb_lockall_mark(lock_ctx
->ctdb
);
353 /* Iterate through all callbacks */
354 request
= lock_ctx
->req_queue
;
356 if (lock_ctx
->auto_mark
) {
357 /* Reset the destructor, so request is not removed from the list */
358 talloc_set_destructor(request
, NULL
);
361 /* In case, callback frees the request, store next */
362 next
= request
->next
;
363 request
->callback(request
->private_data
, locked
);
367 if (lock_ctx
->auto_mark
&& locked
) {
368 switch (lock_ctx
->type
) {
370 tdb_chainlock_unmark(lock_ctx
->ctdb_db
->ltdb
->tdb
, lock_ctx
->key
);
374 tdb_lockall_unmark(lock_ctx
->ctdb_db
->ltdb
->tdb
);
377 case LOCK_ALLDB_PRIO
:
378 ctdb_lockall_unmark_prio(lock_ctx
->ctdb
, lock_ctx
->priority
);
382 ctdb_lockall_unmark(lock_ctx
->ctdb
);
389 static int lock_bucket_id(double t
)
391 double ms
= 1.e
-3, s
= 1;
396 } else if (t
< 10*ms
) {
398 } else if (t
< 100*ms
) {
400 } else if (t
< 1*s
) {
402 } else if (t
< 2*s
) {
404 } else if (t
< 4*s
) {
406 } else if (t
< 8*s
) {
408 } else if (t
< 16*s
) {
410 } else if (t
< 32*s
) {
412 } else if (t
< 64*s
) {
422 * Callback routine when the required locks are obtained.
423 * Called from parent context
425 static void ctdb_lock_handler(struct tevent_context
*ev
,
426 struct tevent_fd
*tfd
,
430 struct lock_context
*lock_ctx
;
431 TALLOC_CTX
*tmp_ctx
= NULL
;
437 lock_ctx
= talloc_get_type_abort(private_data
, struct lock_context
);
439 /* cancel the timeout event */
440 if (lock_ctx
->ttimer
) {
441 TALLOC_FREE(lock_ctx
->ttimer
);
444 t
= timeval_elapsed(&lock_ctx
->start_time
);
445 id
= lock_bucket_id(t
);
447 if (lock_ctx
->auto_mark
) {
448 tmp_ctx
= talloc_new(ev
);
449 talloc_steal(tmp_ctx
, lock_ctx
);
452 /* Read the status from the child process */
453 if (read(lock_ctx
->fd
[0], &c
, 1) != 1) {
456 locked
= (c
== 0 ? true : false);
459 /* Update statistics */
460 CTDB_DECREMENT_STAT(lock_ctx
->ctdb
, locks
.num_pending
);
461 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.num_calls
);
462 if (lock_ctx
->ctdb_db
) {
463 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_pending
);
464 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_calls
);
468 if (lock_ctx
->ctdb_db
) {
469 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.buckets
[id
]);
470 CTDB_UPDATE_LATENCY(lock_ctx
->ctdb
, lock_ctx
->ctdb_db
,
471 lock_type_str
[lock_ctx
->type
], locks
.latency
,
472 lock_ctx
->start_time
);
474 CTDB_UPDATE_DB_LATENCY(lock_ctx
->ctdb_db
, lock_type_str
[lock_ctx
->type
], locks
.latency
, t
);
475 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.buckets
[id
]);
478 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.num_failed
);
479 if (lock_ctx
->ctdb_db
) {
480 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_failed
);
484 process_callbacks(lock_ctx
, locked
);
486 if (lock_ctx
->auto_mark
) {
487 talloc_free(tmp_ctx
);
493 * Callback routine when required locks are not obtained within timeout
494 * Called from parent context
496 static void ctdb_lock_timeout_handler(struct tevent_context
*ev
,
497 struct tevent_timer
*ttimer
,
498 struct timeval current_time
,
501 static const char * debug_locks
= NULL
;
502 struct lock_context
*lock_ctx
;
503 struct ctdb_context
*ctdb
;
506 lock_ctx
= talloc_get_type_abort(private_data
, struct lock_context
);
507 ctdb
= lock_ctx
->ctdb
;
509 if (lock_ctx
->type
== LOCK_RECORD
|| lock_ctx
->type
== LOCK_DB
) {
511 ("Unable to get %s lock on database %s for %.0lf seconds\n",
512 (lock_ctx
->type
== LOCK_RECORD
? "RECORD" : "DB"),
513 lock_ctx
->ctdb_db
->db_name
,
514 timeval_elapsed(&lock_ctx
->start_time
)));
517 ("Unable to get ALLDB locks for %.0lf seconds\n",
518 timeval_elapsed(&lock_ctx
->start_time
)));
521 /* Fire a child process to find the blocking process. */
522 if (debug_locks
== NULL
) {
523 debug_locks
= getenv("CTDB_DEBUG_LOCKS");
524 if (debug_locks
== NULL
) {
525 debug_locks
= talloc_asprintf(ctdb
,
527 getenv("CTDB_BASE"));
530 if (debug_locks
!= NULL
) {
533 execl(debug_locks
, debug_locks
, NULL
);
536 ctdb_track_child(ctdb
, pid
);
540 " Unable to setup lock debugging - no memory?\n"));
543 /* reset the timeout timer */
544 // talloc_free(lock_ctx->ttimer);
545 lock_ctx
->ttimer
= tevent_add_timer(ctdb
->ev
,
547 timeval_current_ofs(10, 0),
548 ctdb_lock_timeout_handler
,
553 static int db_count_handler(struct ctdb_db_context
*ctdb_db
, uint32_t priority
,
556 int *count
= (int *)private_data
;
568 static int db_name_handler(struct ctdb_db_context
*ctdb_db
, uint32_t priority
,
571 struct db_namelist
*list
= (struct db_namelist
*)private_data
;
573 list
->names
[list
->n
] = talloc_strdup(list
->names
, ctdb_db
->db_path
);
579 static char **lock_helper_args(TALLOC_CTX
*mem_ctx
, struct lock_context
*lock_ctx
, int fd
)
581 struct ctdb_context
*ctdb
= lock_ctx
->ctdb
;
585 struct db_namelist list
;
587 switch (lock_ctx
->type
) {
596 case LOCK_ALLDB_PRIO
:
598 ctdb_db_iterator(ctdb
, lock_ctx
->priority
, db_count_handler
, &nargs
);
603 for (priority
=1; priority
<NUM_DB_PRIORITIES
; priority
++) {
604 ctdb_db_iterator(ctdb
, priority
, db_count_handler
, &nargs
);
609 /* Add extra argument for null termination */
612 args
= talloc_array(mem_ctx
, char *, nargs
);
617 args
[0] = talloc_strdup(args
, "ctdb_lock_helper");
618 args
[1] = talloc_asprintf(args
, "%d", getpid());
619 args
[2] = talloc_asprintf(args
, "%d", fd
);
621 switch (lock_ctx
->type
) {
623 args
[3] = talloc_strdup(args
, "RECORD");
624 args
[4] = talloc_strdup(args
, lock_ctx
->ctdb_db
->db_path
);
625 if (lock_ctx
->key
.dsize
== 0) {
626 args
[5] = talloc_strdup(args
, "NULL");
628 args
[5] = hex_encode_talloc(args
, lock_ctx
->key
.dptr
, lock_ctx
->key
.dsize
);
633 args
[3] = talloc_strdup(args
, "DB");
634 args
[4] = talloc_strdup(args
, lock_ctx
->ctdb_db
->db_path
);
637 case LOCK_ALLDB_PRIO
:
638 args
[3] = talloc_strdup(args
, "DB");
641 ctdb_db_iterator(ctdb
, lock_ctx
->priority
, db_name_handler
, &list
);
645 args
[3] = talloc_strdup(args
, "DB");
648 for (priority
=1; priority
<NUM_DB_PRIORITIES
; priority
++) {
649 ctdb_db_iterator(ctdb
, priority
, db_name_handler
, &list
);
654 /* Make sure last argument is NULL */
655 args
[nargs
-1] = NULL
;
657 for (i
=0; i
<nargs
-1; i
++) {
658 if (args
[i
] == NULL
) {
669 * Find the lock context of a given type
671 static struct lock_context
*find_lock_context(struct lock_context
*lock_list
,
672 struct ctdb_db_context
*ctdb_db
,
677 struct lock_context
*lock_ctx
;
679 /* Search active locks */
680 for (lock_ctx
=lock_list
; lock_ctx
; lock_ctx
=lock_ctx
->next
) {
681 if (lock_ctx
->type
!= type
) {
685 switch (lock_ctx
->type
) {
687 if (ctdb_db
== lock_ctx
->ctdb_db
&&
688 key
.dsize
== lock_ctx
->key
.dsize
&&
689 memcmp(key
.dptr
, lock_ctx
->key
.dptr
, key
.dsize
) == 0) {
695 if (ctdb_db
== lock_ctx
->ctdb_db
) {
700 case LOCK_ALLDB_PRIO
:
701 if (priority
== lock_ctx
->priority
) {
712 /* Did not find the lock context we are searching for */
722 * Schedule a new lock child process
723 * Set up callback handler and timeout handler
725 static void ctdb_lock_schedule(struct ctdb_context
*ctdb
)
727 struct lock_context
*lock_ctx
, *next_ctx
, *active_ctx
;
730 const char *helper
= BINDIR
"/ctdb_lock_helper";
731 static const char *prog
= NULL
;
737 t
= getenv("CTDB_LOCK_HELPER");
739 prog
= talloc_strdup(ctdb
, t
);
741 prog
= talloc_strdup(ctdb
, helper
);
743 CTDB_NO_MEMORY_VOID(ctdb
, prog
);
746 if (ctdb
->lock_pending
== NULL
) {
750 /* Find a lock context with requests */
751 lock_ctx
= ctdb
->lock_pending
;
752 while (lock_ctx
!= NULL
) {
753 next_ctx
= lock_ctx
->next
;
754 if (! lock_ctx
->req_queue
) {
755 DEBUG(DEBUG_INFO
, ("Removing lock context without lock requests\n"));
756 DLIST_REMOVE(ctdb
->lock_pending
, lock_ctx
);
757 ctdb
->lock_num_pending
--;
758 CTDB_DECREMENT_STAT(ctdb
, locks
.num_pending
);
759 if (lock_ctx
->ctdb_db
) {
760 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_pending
);
762 talloc_free(lock_ctx
);
764 active_ctx
= find_lock_context(ctdb
->lock_current
, lock_ctx
->ctdb_db
,
765 lock_ctx
->key
, lock_ctx
->priority
,
767 if (active_ctx
== NULL
) {
768 if (lock_ctx
->ctdb_db
== NULL
||
769 lock_ctx
->ctdb_db
->lock_num_current
< MAX_LOCK_PROCESSES_PER_DB
) {
770 /* Found a lock context with lock requests */
775 /* There is already a child waiting for the
776 * same key. So don't schedule another child
783 if (lock_ctx
== NULL
) {
787 lock_ctx
->child
= -1;
788 ret
= pipe(lock_ctx
->fd
);
790 DEBUG(DEBUG_ERR
, ("Failed to create pipe in ctdb_lock_schedule\n"));
794 set_close_on_exec(lock_ctx
->fd
[0]);
796 /* Create data for child process */
797 tmp_ctx
= talloc_new(lock_ctx
);
798 if (tmp_ctx
== NULL
) {
799 DEBUG(DEBUG_ERR
, ("Failed to allocate memory for helper args\n"));
800 close(lock_ctx
->fd
[0]);
801 close(lock_ctx
->fd
[1]);
805 /* Create arguments for lock helper */
806 args
= lock_helper_args(tmp_ctx
, lock_ctx
, lock_ctx
->fd
[1]);
808 DEBUG(DEBUG_ERR
, ("Failed to create lock helper args\n"));
809 close(lock_ctx
->fd
[0]);
810 close(lock_ctx
->fd
[1]);
811 talloc_free(tmp_ctx
);
815 lock_ctx
->child
= vfork();
817 if (lock_ctx
->child
== (pid_t
)-1) {
818 DEBUG(DEBUG_ERR
, ("Failed to create a child in ctdb_lock_schedule\n"));
819 close(lock_ctx
->fd
[0]);
820 close(lock_ctx
->fd
[1]);
821 talloc_free(tmp_ctx
);
827 if (lock_ctx
->child
== 0) {
828 ret
= execv(prog
, args
);
830 DEBUG(DEBUG_ERR
, ("Failed to execute helper %s (%d, %s)\n",
831 prog
, errno
, strerror(errno
)));
837 ctdb_track_child(ctdb
, lock_ctx
->child
);
838 close(lock_ctx
->fd
[1]);
840 talloc_set_destructor(lock_ctx
, ctdb_lock_context_destructor
);
842 talloc_free(tmp_ctx
);
844 /* Set up timeout handler */
845 lock_ctx
->ttimer
= tevent_add_timer(ctdb
->ev
,
847 timeval_current_ofs(10, 0),
848 ctdb_lock_timeout_handler
,
850 if (lock_ctx
->ttimer
== NULL
) {
851 ctdb_kill(ctdb
, lock_ctx
->child
, SIGKILL
);
852 lock_ctx
->child
= -1;
853 talloc_set_destructor(lock_ctx
, NULL
);
854 close(lock_ctx
->fd
[0]);
858 /* Set up callback */
859 lock_ctx
->tfd
= tevent_add_fd(ctdb
->ev
,
865 if (lock_ctx
->tfd
== NULL
) {
866 TALLOC_FREE(lock_ctx
->ttimer
);
867 ctdb_kill(ctdb
, lock_ctx
->child
, SIGKILL
);
868 lock_ctx
->child
= -1;
869 talloc_set_destructor(lock_ctx
, NULL
);
870 close(lock_ctx
->fd
[0]);
873 tevent_fd_set_auto_close(lock_ctx
->tfd
);
875 /* Move the context from pending to current */
876 DLIST_REMOVE(ctdb
->lock_pending
, lock_ctx
);
877 ctdb
->lock_num_pending
--;
878 DLIST_ADD_END(ctdb
->lock_current
, lock_ctx
, NULL
);
879 if (lock_ctx
->ctdb_db
) {
880 lock_ctx
->ctdb_db
->lock_num_current
++;
881 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.num_current
);
882 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_current
);
888 * Lock record / db depending on type
890 static struct lock_request
*ctdb_lock_internal(struct ctdb_context
*ctdb
,
891 struct ctdb_db_context
*ctdb_db
,
894 void (*callback
)(void *, bool),
899 struct lock_context
*lock_ctx
= NULL
;
900 struct lock_request
*request
;
902 if (callback
== NULL
) {
903 DEBUG(DEBUG_WARNING
, ("No callback function specified, not locking\n"));
908 /* Disable this optimization to ensure first-in-first-out fair
909 * scheduling of lock requests */
911 /* get a context for this key - search only the pending contexts,
912 * current contexts might in the middle of processing callbacks */
913 lock_ctx
= find_lock_context(ctdb
->lock_pending
, ctdb_db
, key
, priority
, type
);
916 /* No existing context, create one */
917 if (lock_ctx
== NULL
) {
918 lock_ctx
= talloc_zero(ctdb
, struct lock_context
);
919 if (lock_ctx
== NULL
) {
920 DEBUG(DEBUG_ERR
, ("Failed to create a new lock context\n"));
924 lock_ctx
->type
= type
;
925 lock_ctx
->ctdb
= ctdb
;
926 lock_ctx
->ctdb_db
= ctdb_db
;
927 lock_ctx
->key
.dsize
= key
.dsize
;
929 lock_ctx
->key
.dptr
= talloc_memdup(lock_ctx
, key
.dptr
, key
.dsize
);
931 lock_ctx
->key
.dptr
= NULL
;
933 lock_ctx
->priority
= priority
;
934 lock_ctx
->auto_mark
= auto_mark
;
936 lock_ctx
->child
= -1;
937 lock_ctx
->block_child
= -1;
939 DLIST_ADD_END(ctdb
->lock_pending
, lock_ctx
, NULL
);
940 ctdb
->lock_num_pending
++;
941 CTDB_INCREMENT_STAT(ctdb
, locks
.num_pending
);
943 CTDB_INCREMENT_DB_STAT(ctdb_db
, locks
.num_pending
);
946 /* Start the timer when we activate the context */
947 lock_ctx
->start_time
= timeval_current();
950 if ((request
= talloc_zero(lock_ctx
, struct lock_request
)) == NULL
) {
954 request
->lctx
= lock_ctx
;
955 request
->callback
= callback
;
956 request
->private_data
= private_data
;
958 talloc_set_destructor(request
, ctdb_lock_request_destructor
);
959 DLIST_ADD_END(lock_ctx
->req_queue
, request
, NULL
);
961 ctdb_lock_schedule(ctdb
);
968 * obtain a lock on a record in a database
970 struct lock_request
*ctdb_lock_record(struct ctdb_db_context
*ctdb_db
,
973 void (*callback
)(void *, bool),
976 return ctdb_lock_internal(ctdb_db
->ctdb
,
988 * obtain a lock on a database
990 struct lock_request
*ctdb_lock_db(struct ctdb_db_context
*ctdb_db
,
992 void (*callback
)(void *, bool),
995 return ctdb_lock_internal(ctdb_db
->ctdb
,
1007 * obtain locks on all databases of specified priority
1009 struct lock_request
*ctdb_lock_alldb_prio(struct ctdb_context
*ctdb
,
1012 void (*callback
)(void *, bool),
1015 if (priority
< 1 || priority
> NUM_DB_PRIORITIES
) {
1016 DEBUG(DEBUG_ERR
, ("Invalid db priority: %u\n", priority
));
1020 return ctdb_lock_internal(ctdb
,
1032 * obtain locks on all databases
1034 struct lock_request
*ctdb_lock_alldb(struct ctdb_context
*ctdb
,
1036 void (*callback
)(void *, bool),
1039 return ctdb_lock_internal(ctdb
,