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
,