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"
25 #include "lib/tdb_wrap/tdb_wrap.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 * execute an external script to debug.
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
44 * = false is used for freezing databases for
45 * recovery since the recovery cannot start till
46 * databases are locked on all the nodes.
47 * = true is used for record locks.
57 static const char * const lock_type_str
[] = {
66 /* lock_context is the common part for a lock request */
68 struct lock_context
*next
, *prev
;
70 struct ctdb_context
*ctdb
;
71 struct ctdb_db_context
*ctdb_db
;
75 struct lock_request
*request
;
78 struct tevent_fd
*tfd
;
79 struct tevent_timer
*ttimer
;
80 struct timeval start_time
;
85 /* lock_request is the client specific part for a lock request */
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 daemon 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
->request
) {
280 lock_ctx
->request
->lctx
= NULL
;
282 if (lock_ctx
->child
> 0) {
283 ctdb_kill(lock_ctx
->ctdb
, lock_ctx
->child
, SIGKILL
);
284 if (lock_ctx
->type
== LOCK_RECORD
) {
285 DLIST_REMOVE(lock_ctx
->ctdb_db
->lock_current
, lock_ctx
);
287 DLIST_REMOVE(lock_ctx
->ctdb
->lock_current
, lock_ctx
);
289 if (lock_ctx
->ctdb_db
) {
290 lock_ctx
->ctdb_db
->lock_num_current
--;
292 CTDB_DECREMENT_STAT(lock_ctx
->ctdb
, locks
.num_current
);
293 if (lock_ctx
->ctdb_db
) {
294 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_current
);
297 if (lock_ctx
->type
== LOCK_RECORD
) {
298 DLIST_REMOVE(lock_ctx
->ctdb_db
->lock_pending
, lock_ctx
);
300 DLIST_REMOVE(lock_ctx
->ctdb
->lock_pending
, lock_ctx
);
302 CTDB_DECREMENT_STAT(lock_ctx
->ctdb
, locks
.num_pending
);
303 if (lock_ctx
->ctdb_db
) {
304 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_pending
);
308 ctdb_lock_schedule(lock_ctx
->ctdb
);
315 * Destructor to remove lock request
317 static int ctdb_lock_request_destructor(struct lock_request
*lock_request
)
319 if (lock_request
->lctx
== NULL
) {
323 lock_request
->lctx
->request
= NULL
;
324 TALLOC_FREE(lock_request
->lctx
);
330 * Process all the callbacks waiting for lock
332 * If lock has failed, callback is executed with locked=false
334 static void process_callbacks(struct lock_context
*lock_ctx
, bool locked
)
336 struct lock_request
*request
;
337 bool auto_mark
= lock_ctx
->auto_mark
;
339 if (auto_mark
&& locked
) {
340 switch (lock_ctx
->type
) {
342 tdb_chainlock_mark(lock_ctx
->ctdb_db
->ltdb
->tdb
, lock_ctx
->key
);
346 tdb_lockall_mark(lock_ctx
->ctdb_db
->ltdb
->tdb
);
349 case LOCK_ALLDB_PRIO
:
350 ctdb_lockall_mark_prio(lock_ctx
->ctdb
, lock_ctx
->priority
);
354 ctdb_lockall_mark(lock_ctx
->ctdb
);
359 request
= lock_ctx
->request
;
361 /* Since request may be freed in the callback, unset the lock
362 * context, so request destructor will not free lock context.
364 request
->lctx
= NULL
;
367 /* Since request may be freed in the callback, unset the request */
368 lock_ctx
->request
= NULL
;
370 request
->callback(request
->private_data
, locked
);
377 switch (lock_ctx
->type
) {
379 tdb_chainlock_unmark(lock_ctx
->ctdb_db
->ltdb
->tdb
, lock_ctx
->key
);
383 tdb_lockall_unmark(lock_ctx
->ctdb_db
->ltdb
->tdb
);
386 case LOCK_ALLDB_PRIO
:
387 ctdb_lockall_unmark_prio(lock_ctx
->ctdb
, lock_ctx
->priority
);
391 ctdb_lockall_unmark(lock_ctx
->ctdb
);
398 static int lock_bucket_id(double t
)
400 double ms
= 1.e
-3, s
= 1;
405 } else if (t
< 10*ms
) {
407 } else if (t
< 100*ms
) {
409 } else if (t
< 1*s
) {
411 } else if (t
< 2*s
) {
413 } else if (t
< 4*s
) {
415 } else if (t
< 8*s
) {
417 } else if (t
< 16*s
) {
419 } else if (t
< 32*s
) {
421 } else if (t
< 64*s
) {
431 * Callback routine when the required locks are obtained.
432 * Called from parent context
434 static void ctdb_lock_handler(struct tevent_context
*ev
,
435 struct tevent_fd
*tfd
,
439 struct lock_context
*lock_ctx
;
440 TALLOC_CTX
*tmp_ctx
= NULL
;
446 lock_ctx
= talloc_get_type_abort(private_data
, struct lock_context
);
448 /* cancel the timeout event */
449 TALLOC_FREE(lock_ctx
->ttimer
);
451 t
= timeval_elapsed(&lock_ctx
->start_time
);
452 id
= lock_bucket_id(t
);
454 if (lock_ctx
->auto_mark
) {
455 tmp_ctx
= talloc_new(ev
);
456 talloc_steal(tmp_ctx
, lock_ctx
);
459 /* Read the status from the child process */
460 if (sys_read(lock_ctx
->fd
[0], &c
, 1) != 1) {
463 locked
= (c
== 0 ? true : false);
466 /* Update statistics */
467 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.num_calls
);
468 if (lock_ctx
->ctdb_db
) {
469 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_calls
);
473 if (lock_ctx
->ctdb_db
) {
474 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.buckets
[id
]);
475 CTDB_UPDATE_LATENCY(lock_ctx
->ctdb
, lock_ctx
->ctdb_db
,
476 lock_type_str
[lock_ctx
->type
], locks
.latency
,
477 lock_ctx
->start_time
);
479 CTDB_UPDATE_DB_LATENCY(lock_ctx
->ctdb_db
, lock_type_str
[lock_ctx
->type
], locks
.latency
, t
);
480 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.buckets
[id
]);
483 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.num_failed
);
484 if (lock_ctx
->ctdb_db
) {
485 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_failed
);
489 process_callbacks(lock_ctx
, locked
);
491 if (lock_ctx
->auto_mark
) {
492 talloc_free(tmp_ctx
);
498 * Callback routine when required locks are not obtained within timeout
499 * Called from parent context
501 static void ctdb_lock_timeout_handler(struct tevent_context
*ev
,
502 struct tevent_timer
*ttimer
,
503 struct timeval current_time
,
506 static char debug_locks
[PATH_MAX
+1] = "";
507 struct lock_context
*lock_ctx
;
508 struct ctdb_context
*ctdb
;
513 lock_ctx
= talloc_get_type_abort(private_data
, struct lock_context
);
514 ctdb
= lock_ctx
->ctdb
;
516 /* If a node stopped/banned, don't spam the logs */
517 if (ctdb
->nodes
[ctdb
->pnn
]->flags
& NODE_FLAGS_INACTIVE
) {
518 lock_ctx
->ttimer
= NULL
;
522 elapsed_time
= timeval_elapsed(&lock_ctx
->start_time
);
523 if (lock_ctx
->ctdb_db
) {
525 ("Unable to get %s lock on database %s for %.0lf seconds\n",
526 (lock_ctx
->type
== LOCK_RECORD
? "RECORD" : "DB"),
527 lock_ctx
->ctdb_db
->db_name
, elapsed_time
));
530 ("Unable to get ALLDB locks for %.0lf seconds\n",
534 if (ctdb_set_helper("lock debugging helper",
535 debug_locks
, sizeof(debug_locks
),
537 getenv("CTDB_BASE"), "debug_locks.sh")) {
540 execl(debug_locks
, debug_locks
, NULL
);
543 ctdb_track_child(ctdb
, pid
);
547 " Unable to setup lock debugging\n"));
550 /* Back-off logging if lock is not obtained for a long time */
551 if (elapsed_time
< 100.0) {
553 } else if (elapsed_time
< 1000.0) {
559 /* reset the timeout timer */
560 // talloc_free(lock_ctx->ttimer);
561 lock_ctx
->ttimer
= tevent_add_timer(ctdb
->ev
,
563 timeval_current_ofs(new_timer
, 0),
564 ctdb_lock_timeout_handler
,
569 static int db_count_handler(struct ctdb_db_context
*ctdb_db
, uint32_t priority
,
572 int *count
= (int *)private_data
;
579 static int db_flags(struct ctdb_db_context
*ctdb_db
)
581 int tdb_flags
= TDB_DEFAULT
;
583 #ifdef TDB_MUTEX_LOCKING
584 if (!ctdb_db
->persistent
&& ctdb_db
->ctdb
->tunable
.mutex_enabled
) {
585 tdb_flags
= (TDB_MUTEX_LOCKING
| TDB_CLEAR_IF_FIRST
);
596 static int db_name_handler(struct ctdb_db_context
*ctdb_db
, uint32_t priority
,
599 struct db_namelist
*list
= (struct db_namelist
*)private_data
;
601 list
->names
[list
->n
] = talloc_strdup(list
->names
, ctdb_db
->db_path
);
602 list
->names
[list
->n
+1] = talloc_asprintf(list
->names
, "0x%x",
609 static bool lock_helper_args(TALLOC_CTX
*mem_ctx
,
610 struct lock_context
*lock_ctx
, int fd
,
611 int *argc
, const char ***argv
)
613 struct ctdb_context
*ctdb
= lock_ctx
->ctdb
;
614 const char **args
= NULL
;
617 struct db_namelist list
;
619 switch (lock_ctx
->type
) {
628 case LOCK_ALLDB_PRIO
:
630 ctdb_db_iterator(ctdb
, lock_ctx
->priority
, db_count_handler
, &nargs
);
635 for (priority
=1; priority
<NUM_DB_PRIORITIES
; priority
++) {
636 ctdb_db_iterator(ctdb
, priority
, db_count_handler
, &nargs
);
641 /* Add extra argument for null termination */
644 args
= talloc_array(mem_ctx
, const char *, nargs
);
649 args
[0] = talloc_asprintf(args
, "%d", getpid());
650 args
[1] = talloc_asprintf(args
, "%d", fd
);
652 switch (lock_ctx
->type
) {
654 args
[2] = talloc_strdup(args
, "RECORD");
655 args
[3] = talloc_strdup(args
, lock_ctx
->ctdb_db
->db_path
);
656 args
[4] = talloc_asprintf(args
, "0x%x",
657 db_flags(lock_ctx
->ctdb_db
));
658 if (lock_ctx
->key
.dsize
== 0) {
659 args
[5] = talloc_strdup(args
, "NULL");
661 args
[5] = hex_encode_talloc(args
, lock_ctx
->key
.dptr
, lock_ctx
->key
.dsize
);
666 args
[2] = talloc_strdup(args
, "DB");
667 args
[3] = talloc_strdup(args
, lock_ctx
->ctdb_db
->db_path
);
668 args
[4] = talloc_asprintf(args
, "0x%x",
669 db_flags(lock_ctx
->ctdb_db
));
672 case LOCK_ALLDB_PRIO
:
673 args
[2] = talloc_strdup(args
, "DB");
676 ctdb_db_iterator(ctdb
, lock_ctx
->priority
, db_name_handler
, &list
);
680 args
[2] = talloc_strdup(args
, "DB");
683 for (priority
=1; priority
<NUM_DB_PRIORITIES
; priority
++) {
684 ctdb_db_iterator(ctdb
, priority
, db_name_handler
, &list
);
689 /* Make sure last argument is NULL */
690 args
[nargs
-1] = NULL
;
692 for (i
=0; i
<nargs
-1; i
++) {
693 if (args
[i
] == NULL
) {
705 * Find a lock request that can be scheduled
707 static struct lock_context
*ctdb_find_lock_context(struct ctdb_context
*ctdb
)
709 struct lock_context
*lock_ctx
, *next_ctx
;
710 struct ctdb_db_context
*ctdb_db
;
712 /* First check if there are database lock requests */
714 for (lock_ctx
= ctdb
->lock_pending
; lock_ctx
!= NULL
;
715 lock_ctx
= next_ctx
) {
717 if (lock_ctx
->request
!= NULL
) {
718 /* Found a lock context with a request */
722 next_ctx
= lock_ctx
->next
;
724 DEBUG(DEBUG_INFO
, ("Removing lock context without lock "
726 DLIST_REMOVE(ctdb
->lock_pending
, lock_ctx
);
727 CTDB_DECREMENT_STAT(ctdb
, locks
.num_pending
);
728 if (lock_ctx
->ctdb_db
) {
729 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
,
732 talloc_free(lock_ctx
);
735 /* Next check database queues */
736 for (ctdb_db
= ctdb
->db_list
; ctdb_db
; ctdb_db
= ctdb_db
->next
) {
737 if (ctdb_db
->lock_num_current
==
738 ctdb
->tunable
.lock_processes_per_db
) {
742 for (lock_ctx
= ctdb_db
->lock_pending
; lock_ctx
!= NULL
;
743 lock_ctx
= next_ctx
) {
745 next_ctx
= lock_ctx
->next
;
747 if (lock_ctx
->request
!= NULL
) {
751 DEBUG(DEBUG_INFO
, ("Removing lock context without "
753 DLIST_REMOVE(ctdb_db
->lock_pending
, lock_ctx
);
754 CTDB_DECREMENT_STAT(ctdb
, locks
.num_pending
);
755 CTDB_DECREMENT_DB_STAT(ctdb_db
, locks
.num_pending
);
756 talloc_free(lock_ctx
);
764 * Schedule a new lock child process
765 * Set up callback handler and timeout handler
767 static void ctdb_lock_schedule(struct ctdb_context
*ctdb
)
769 struct lock_context
*lock_ctx
;
772 static char prog
[PATH_MAX
+1] = "";
775 if (!ctdb_set_helper("lock helper",
778 CTDB_HELPER_BINDIR
, "ctdb_lock_helper")) {
779 ctdb_die(ctdb
, __location__
780 " Unable to set lock helper\n");
783 /* Find a lock context with requests */
784 lock_ctx
= ctdb_find_lock_context(ctdb
);
785 if (lock_ctx
== NULL
) {
789 lock_ctx
->child
= -1;
790 ret
= pipe(lock_ctx
->fd
);
792 DEBUG(DEBUG_ERR
, ("Failed to create pipe in ctdb_lock_schedule\n"));
796 set_close_on_exec(lock_ctx
->fd
[0]);
798 /* Create data for child process */
799 tmp_ctx
= talloc_new(lock_ctx
);
800 if (tmp_ctx
== NULL
) {
801 DEBUG(DEBUG_ERR
, ("Failed to allocate memory for helper args\n"));
802 close(lock_ctx
->fd
[0]);
803 close(lock_ctx
->fd
[1]);
807 /* Create arguments for lock helper */
808 if (!lock_helper_args(tmp_ctx
, lock_ctx
, lock_ctx
->fd
[1],
810 DEBUG(DEBUG_ERR
, ("Failed to create lock helper args\n"));
811 close(lock_ctx
->fd
[0]);
812 close(lock_ctx
->fd
[1]);
813 talloc_free(tmp_ctx
);
817 if (!ctdb_vfork_with_logging(lock_ctx
, ctdb
, "lock_helper",
818 prog
, argc
, (const char **)args
,
819 NULL
, NULL
, &lock_ctx
->child
)) {
820 DEBUG(DEBUG_ERR
, ("Failed to create a child in ctdb_lock_schedule\n"));
821 close(lock_ctx
->fd
[0]);
822 close(lock_ctx
->fd
[1]);
823 talloc_free(tmp_ctx
);
828 close(lock_ctx
->fd
[1]);
830 talloc_free(tmp_ctx
);
832 /* Set up timeout handler */
833 lock_ctx
->ttimer
= tevent_add_timer(ctdb
->ev
,
835 timeval_current_ofs(10, 0),
836 ctdb_lock_timeout_handler
,
838 if (lock_ctx
->ttimer
== NULL
) {
839 ctdb_kill(ctdb
, lock_ctx
->child
, SIGKILL
);
840 lock_ctx
->child
= -1;
841 close(lock_ctx
->fd
[0]);
845 /* Set up callback */
846 lock_ctx
->tfd
= tevent_add_fd(ctdb
->ev
,
852 if (lock_ctx
->tfd
== NULL
) {
853 TALLOC_FREE(lock_ctx
->ttimer
);
854 ctdb_kill(ctdb
, lock_ctx
->child
, SIGKILL
);
855 lock_ctx
->child
= -1;
856 close(lock_ctx
->fd
[0]);
859 tevent_fd_set_auto_close(lock_ctx
->tfd
);
861 /* Move the context from pending to current */
862 if (lock_ctx
->type
== LOCK_RECORD
) {
863 DLIST_REMOVE(lock_ctx
->ctdb_db
->lock_pending
, lock_ctx
);
864 DLIST_ADD_END(lock_ctx
->ctdb_db
->lock_current
, lock_ctx
, NULL
);
866 DLIST_REMOVE(ctdb
->lock_pending
, lock_ctx
);
867 DLIST_ADD_END(ctdb
->lock_current
, lock_ctx
, NULL
);
869 CTDB_DECREMENT_STAT(lock_ctx
->ctdb
, locks
.num_pending
);
870 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.num_current
);
871 if (lock_ctx
->ctdb_db
) {
872 lock_ctx
->ctdb_db
->lock_num_current
++;
873 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_pending
);
874 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_current
);
880 * Lock record / db depending on type
882 static struct lock_request
*ctdb_lock_internal(TALLOC_CTX
*mem_ctx
,
883 struct ctdb_context
*ctdb
,
884 struct ctdb_db_context
*ctdb_db
,
887 void (*callback
)(void *, bool),
892 struct lock_context
*lock_ctx
= NULL
;
893 struct lock_request
*request
;
895 if (callback
== NULL
) {
896 DEBUG(DEBUG_WARNING
, ("No callback function specified, not locking\n"));
900 lock_ctx
= talloc_zero(ctdb
, struct lock_context
);
901 if (lock_ctx
== NULL
) {
902 DEBUG(DEBUG_ERR
, ("Failed to create a new lock context\n"));
906 if ((request
= talloc_zero(mem_ctx
, struct lock_request
)) == NULL
) {
907 talloc_free(lock_ctx
);
911 lock_ctx
->type
= type
;
912 lock_ctx
->ctdb
= ctdb
;
913 lock_ctx
->ctdb_db
= ctdb_db
;
914 lock_ctx
->key
.dsize
= key
.dsize
;
916 lock_ctx
->key
.dptr
= talloc_memdup(lock_ctx
, key
.dptr
, key
.dsize
);
917 if (lock_ctx
->key
.dptr
== NULL
) {
918 DEBUG(DEBUG_ERR
, (__location__
"Memory allocation error\n"));
919 talloc_free(lock_ctx
);
920 talloc_free(request
);
923 lock_ctx
->key_hash
= ctdb_hash(&key
);
925 lock_ctx
->key
.dptr
= NULL
;
927 lock_ctx
->priority
= priority
;
928 lock_ctx
->auto_mark
= auto_mark
;
930 lock_ctx
->request
= request
;
931 lock_ctx
->child
= -1;
933 /* Non-record locks are required by recovery and should be scheduled
934 * immediately, so keep them at the head of the pending queue.
936 if (lock_ctx
->type
== LOCK_RECORD
) {
937 DLIST_ADD_END(ctdb_db
->lock_pending
, lock_ctx
, NULL
);
939 DLIST_ADD_END(ctdb
->lock_pending
, lock_ctx
, NULL
);
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();
949 request
->lctx
= lock_ctx
;
950 request
->callback
= callback
;
951 request
->private_data
= private_data
;
953 talloc_set_destructor(request
, ctdb_lock_request_destructor
);
954 talloc_set_destructor(lock_ctx
, ctdb_lock_context_destructor
);
956 ctdb_lock_schedule(ctdb
);
963 * obtain a lock on a record in a database
965 struct lock_request
*ctdb_lock_record(TALLOC_CTX
*mem_ctx
,
966 struct ctdb_db_context
*ctdb_db
,
969 void (*callback
)(void *, bool),
972 return ctdb_lock_internal(mem_ctx
,
985 * obtain a lock on a database
987 struct lock_request
*ctdb_lock_db(TALLOC_CTX
*mem_ctx
,
988 struct ctdb_db_context
*ctdb_db
,
990 void (*callback
)(void *, bool),
993 return ctdb_lock_internal(mem_ctx
,
1006 * obtain locks on all databases of specified priority
1008 struct lock_request
*ctdb_lock_alldb_prio(TALLOC_CTX
*mem_ctx
,
1009 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(mem_ctx
,
1033 * obtain locks on all databases
1035 struct lock_request
*ctdb_lock_alldb(TALLOC_CTX
*mem_ctx
,
1036 struct ctdb_context
*ctdb
,
1038 void (*callback
)(void *, bool),
1041 return ctdb_lock_internal(mem_ctx
,