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
53 static const char * const lock_type_str
[] = {
62 /* lock_context is the common part for a lock request */
64 struct lock_context
*next
, *prev
;
66 struct ctdb_context
*ctdb
;
67 struct ctdb_db_context
*ctdb_db
;
71 struct lock_request
*request
;
74 struct tevent_fd
*tfd
;
75 struct tevent_timer
*ttimer
;
76 struct timeval start_time
;
81 /* lock_request is the client specific part for a lock request */
83 struct lock_context
*lctx
;
84 void (*callback
)(void *, bool);
90 * Support samba 3.6.x (and older) versions which do not set db priority.
92 * By default, all databases are set to priority 1. So only when priority
93 * is set to 1, check for databases that need higher priority.
95 static bool later_db(struct ctdb_context
*ctdb
, const char *name
)
97 if (ctdb
->tunable
.samba3_hack
== 0) {
101 if (strstr(name
, "brlock") ||
102 strstr(name
, "g_lock") ||
103 strstr(name
, "notify_onelevel") ||
104 strstr(name
, "serverid") ||
105 strstr(name
, "xattr_tdb")) {
112 typedef int (*db_handler_t
)(struct ctdb_db_context
*ctdb_db
,
116 static int ctdb_db_iterator(struct ctdb_context
*ctdb
, uint32_t priority
,
117 db_handler_t handler
, void *private_data
)
119 struct ctdb_db_context
*ctdb_db
;
122 for (ctdb_db
= ctdb
->db_list
; ctdb_db
; ctdb_db
= ctdb_db
->next
) {
123 if (ctdb_db
->priority
!= priority
) {
126 if (later_db(ctdb
, ctdb_db
->db_name
)) {
129 ret
= handler(ctdb_db
, priority
, private_data
);
135 /* If priority != 1, later_db check is not required and can return */
140 for (ctdb_db
= ctdb
->db_list
; ctdb_db
; ctdb_db
= ctdb_db
->next
) {
141 if (!later_db(ctdb
, ctdb_db
->db_name
)) {
144 ret
= handler(ctdb_db
, priority
, private_data
);
155 * lock all databases - mark only
157 static int db_lock_mark_handler(struct ctdb_db_context
*ctdb_db
, uint32_t priority
,
160 int tdb_transaction_write_lock_mark(struct tdb_context
*);
162 DEBUG(DEBUG_INFO
, ("marking locked database %s, priority:%u\n",
163 ctdb_db
->db_name
, priority
));
165 if (tdb_transaction_write_lock_mark(ctdb_db
->ltdb
->tdb
) != 0) {
166 DEBUG(DEBUG_ERR
, ("Failed to mark (transaction lock) database %s\n",
171 if (tdb_lockall_mark(ctdb_db
->ltdb
->tdb
) != 0) {
172 DEBUG(DEBUG_ERR
, ("Failed to mark (all lock) database %s\n",
180 int ctdb_lockall_mark_prio(struct ctdb_context
*ctdb
, uint32_t priority
)
183 * This function is only used by the main dameon during recovery.
184 * At this stage, the databases have already been locked, by a
185 * dedicated child process. The freeze_mode variable is used to track
186 * whether the actual locks are held by the child process or not.
189 if (ctdb
->freeze_mode
[priority
] != CTDB_FREEZE_FROZEN
) {
190 DEBUG(DEBUG_ERR
, ("Attempt to mark all databases locked when not frozen\n"));
194 return ctdb_db_iterator(ctdb
, priority
, db_lock_mark_handler
, NULL
);
197 static int ctdb_lockall_mark(struct ctdb_context
*ctdb
)
201 for (priority
=1; priority
<=NUM_DB_PRIORITIES
; priority
++) {
202 if (ctdb_db_iterator(ctdb
, priority
, db_lock_mark_handler
, NULL
) != 0) {
212 * lock all databases - unmark only
214 static int db_lock_unmark_handler(struct ctdb_db_context
*ctdb_db
, uint32_t priority
,
217 int tdb_transaction_write_lock_unmark(struct tdb_context
*);
219 DEBUG(DEBUG_INFO
, ("unmarking locked database %s, priority:%u\n",
220 ctdb_db
->db_name
, priority
));
222 if (tdb_transaction_write_lock_unmark(ctdb_db
->ltdb
->tdb
) != 0) {
223 DEBUG(DEBUG_ERR
, ("Failed to unmark (transaction lock) database %s\n",
228 if (tdb_lockall_unmark(ctdb_db
->ltdb
->tdb
) != 0) {
229 DEBUG(DEBUG_ERR
, ("Failed to unmark (all lock) database %s\n",
237 int ctdb_lockall_unmark_prio(struct ctdb_context
*ctdb
, uint32_t priority
)
240 * This function is only used by the main daemon during recovery.
241 * At this stage, the databases have already been locked, by a
242 * dedicated child process. The freeze_mode variable is used to track
243 * whether the actual locks are held by the child process or not.
246 if (ctdb
->freeze_mode
[priority
] != CTDB_FREEZE_FROZEN
) {
247 DEBUG(DEBUG_ERR
, ("Attempt to unmark all databases locked when not frozen\n"));
251 return ctdb_db_iterator(ctdb
, priority
, db_lock_unmark_handler
, NULL
);
254 static int ctdb_lockall_unmark(struct ctdb_context
*ctdb
)
258 for (priority
=NUM_DB_PRIORITIES
; priority
>0; priority
--) {
259 if (ctdb_db_iterator(ctdb
, priority
, db_lock_unmark_handler
, NULL
) != 0) {
268 static void ctdb_lock_schedule(struct ctdb_context
*ctdb
);
271 * Destructor to kill the child locking process
273 static int ctdb_lock_context_destructor(struct lock_context
*lock_ctx
)
275 if (lock_ctx
->request
) {
276 lock_ctx
->request
->lctx
= NULL
;
278 if (lock_ctx
->child
> 0) {
279 ctdb_kill(lock_ctx
->ctdb
, lock_ctx
->child
, SIGKILL
);
280 if (lock_ctx
->type
== LOCK_RECORD
) {
281 DLIST_REMOVE(lock_ctx
->ctdb_db
->lock_current
, lock_ctx
);
283 DLIST_REMOVE(lock_ctx
->ctdb
->lock_current
, lock_ctx
);
285 if (lock_ctx
->ctdb_db
) {
286 lock_ctx
->ctdb_db
->lock_num_current
--;
288 CTDB_DECREMENT_STAT(lock_ctx
->ctdb
, locks
.num_current
);
289 if (lock_ctx
->ctdb_db
) {
290 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_current
);
293 if (lock_ctx
->type
== LOCK_RECORD
) {
294 DLIST_REMOVE(lock_ctx
->ctdb_db
->lock_pending
, lock_ctx
);
296 DLIST_REMOVE(lock_ctx
->ctdb
->lock_pending
, lock_ctx
);
298 CTDB_DECREMENT_STAT(lock_ctx
->ctdb
, locks
.num_pending
);
299 if (lock_ctx
->ctdb_db
) {
300 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_pending
);
304 ctdb_lock_schedule(lock_ctx
->ctdb
);
311 * Destructor to remove lock request
313 static int ctdb_lock_request_destructor(struct lock_request
*lock_request
)
315 TALLOC_FREE(lock_request
->lctx
);
320 * Process all the callbacks waiting for lock
322 * If lock has failed, callback is executed with locked=false
324 static void process_callbacks(struct lock_context
*lock_ctx
, bool locked
)
326 struct lock_request
*request
;
328 if (lock_ctx
->auto_mark
&& locked
) {
329 switch (lock_ctx
->type
) {
331 tdb_chainlock_mark(lock_ctx
->ctdb_db
->ltdb
->tdb
, lock_ctx
->key
);
335 tdb_lockall_mark(lock_ctx
->ctdb_db
->ltdb
->tdb
);
338 case LOCK_ALLDB_PRIO
:
339 ctdb_lockall_mark_prio(lock_ctx
->ctdb
, lock_ctx
->priority
);
343 ctdb_lockall_mark(lock_ctx
->ctdb
);
348 request
= lock_ctx
->request
;
349 if (lock_ctx
->auto_mark
) {
350 /* Reset the destructor, so request is not removed from the list */
351 talloc_set_destructor(request
, NULL
);
353 request
->callback(request
->private_data
, locked
);
355 if (lock_ctx
->auto_mark
&& locked
) {
356 switch (lock_ctx
->type
) {
358 tdb_chainlock_unmark(lock_ctx
->ctdb_db
->ltdb
->tdb
, lock_ctx
->key
);
362 tdb_lockall_unmark(lock_ctx
->ctdb_db
->ltdb
->tdb
);
365 case LOCK_ALLDB_PRIO
:
366 ctdb_lockall_unmark_prio(lock_ctx
->ctdb
, lock_ctx
->priority
);
370 ctdb_lockall_unmark(lock_ctx
->ctdb
);
377 static int lock_bucket_id(double t
)
379 double ms
= 1.e
-3, s
= 1;
384 } else if (t
< 10*ms
) {
386 } else if (t
< 100*ms
) {
388 } else if (t
< 1*s
) {
390 } else if (t
< 2*s
) {
392 } else if (t
< 4*s
) {
394 } else if (t
< 8*s
) {
396 } else if (t
< 16*s
) {
398 } else if (t
< 32*s
) {
400 } else if (t
< 64*s
) {
410 * Callback routine when the required locks are obtained.
411 * Called from parent context
413 static void ctdb_lock_handler(struct tevent_context
*ev
,
414 struct tevent_fd
*tfd
,
418 struct lock_context
*lock_ctx
;
419 TALLOC_CTX
*tmp_ctx
= NULL
;
425 lock_ctx
= talloc_get_type_abort(private_data
, struct lock_context
);
427 /* cancel the timeout event */
428 TALLOC_FREE(lock_ctx
->ttimer
);
430 t
= timeval_elapsed(&lock_ctx
->start_time
);
431 id
= lock_bucket_id(t
);
433 if (lock_ctx
->auto_mark
) {
434 tmp_ctx
= talloc_new(ev
);
435 talloc_steal(tmp_ctx
, lock_ctx
);
438 /* Read the status from the child process */
439 if (sys_read(lock_ctx
->fd
[0], &c
, 1) != 1) {
442 locked
= (c
== 0 ? true : false);
445 /* Update statistics */
446 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.num_calls
);
447 if (lock_ctx
->ctdb_db
) {
448 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_calls
);
452 if (lock_ctx
->ctdb_db
) {
453 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.buckets
[id
]);
454 CTDB_UPDATE_LATENCY(lock_ctx
->ctdb
, lock_ctx
->ctdb_db
,
455 lock_type_str
[lock_ctx
->type
], locks
.latency
,
456 lock_ctx
->start_time
);
458 CTDB_UPDATE_DB_LATENCY(lock_ctx
->ctdb_db
, lock_type_str
[lock_ctx
->type
], locks
.latency
, t
);
459 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.buckets
[id
]);
462 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.num_failed
);
463 if (lock_ctx
->ctdb_db
) {
464 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_failed
);
468 process_callbacks(lock_ctx
, locked
);
470 if (lock_ctx
->auto_mark
) {
471 talloc_free(tmp_ctx
);
477 * Callback routine when required locks are not obtained within timeout
478 * Called from parent context
480 static void ctdb_lock_timeout_handler(struct tevent_context
*ev
,
481 struct tevent_timer
*ttimer
,
482 struct timeval current_time
,
485 static const char * debug_locks
= NULL
;
486 struct lock_context
*lock_ctx
;
487 struct ctdb_context
*ctdb
;
490 lock_ctx
= talloc_get_type_abort(private_data
, struct lock_context
);
491 ctdb
= lock_ctx
->ctdb
;
493 /* If a node stopped/banned, don't spam the logs */
494 if (ctdb
->nodes
[ctdb
->pnn
]->flags
& NODE_FLAGS_INACTIVE
) {
495 lock_ctx
->ttimer
= NULL
;
498 if (lock_ctx
->ctdb_db
) {
500 ("Unable to get %s lock on database %s for %.0lf seconds\n",
501 (lock_ctx
->type
== LOCK_RECORD
? "RECORD" : "DB"),
502 lock_ctx
->ctdb_db
->db_name
,
503 timeval_elapsed(&lock_ctx
->start_time
)));
506 ("Unable to get ALLDB locks for %.0lf seconds\n",
507 timeval_elapsed(&lock_ctx
->start_time
)));
510 /* Fire a child process to find the blocking process. */
511 if (debug_locks
== NULL
) {
512 debug_locks
= getenv("CTDB_DEBUG_LOCKS");
513 if (debug_locks
== NULL
) {
514 debug_locks
= talloc_asprintf(ctdb
,
516 getenv("CTDB_BASE"));
519 if (debug_locks
!= NULL
) {
522 execl(debug_locks
, debug_locks
, NULL
);
525 ctdb_track_child(ctdb
, pid
);
529 " Unable to setup lock debugging - no memory?\n"));
532 /* reset the timeout timer */
533 // talloc_free(lock_ctx->ttimer);
534 lock_ctx
->ttimer
= tevent_add_timer(ctdb
->ev
,
536 timeval_current_ofs(10, 0),
537 ctdb_lock_timeout_handler
,
542 static int db_count_handler(struct ctdb_db_context
*ctdb_db
, uint32_t priority
,
545 int *count
= (int *)private_data
;
552 static int db_flags(struct ctdb_db_context
*ctdb_db
)
554 int tdb_flags
= TDB_DEFAULT
;
556 #ifdef TDB_MUTEX_LOCKING
557 if (!ctdb_db
->persistent
&& ctdb_db
->ctdb
->tunable
.mutex_enabled
) {
558 tdb_flags
= (TDB_MUTEX_LOCKING
| TDB_CLEAR_IF_FIRST
);
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
);
575 list
->names
[list
->n
+1] = talloc_asprintf(list
->names
, "0x%x",
582 static bool lock_helper_args(TALLOC_CTX
*mem_ctx
,
583 struct lock_context
*lock_ctx
, int fd
,
584 int *argc
, const char ***argv
)
586 struct ctdb_context
*ctdb
= lock_ctx
->ctdb
;
587 const char **args
= NULL
;
590 struct db_namelist list
;
592 switch (lock_ctx
->type
) {
601 case LOCK_ALLDB_PRIO
:
603 ctdb_db_iterator(ctdb
, lock_ctx
->priority
, db_count_handler
, &nargs
);
608 for (priority
=1; priority
<NUM_DB_PRIORITIES
; priority
++) {
609 ctdb_db_iterator(ctdb
, priority
, db_count_handler
, &nargs
);
614 /* Add extra argument for null termination */
617 args
= talloc_array(mem_ctx
, const char *, nargs
);
622 args
[0] = talloc_asprintf(args
, "%d", getpid());
623 args
[1] = talloc_asprintf(args
, "%d", fd
);
625 switch (lock_ctx
->type
) {
627 args
[2] = talloc_strdup(args
, "RECORD");
628 args
[3] = talloc_strdup(args
, lock_ctx
->ctdb_db
->db_path
);
629 args
[4] = talloc_asprintf(args
, "0x%x",
630 db_flags(lock_ctx
->ctdb_db
));
631 if (lock_ctx
->key
.dsize
== 0) {
632 args
[5] = talloc_strdup(args
, "NULL");
634 args
[5] = hex_encode_talloc(args
, lock_ctx
->key
.dptr
, lock_ctx
->key
.dsize
);
639 args
[2] = talloc_strdup(args
, "DB");
640 args
[3] = talloc_strdup(args
, lock_ctx
->ctdb_db
->db_path
);
641 args
[4] = talloc_asprintf(args
, "0x%x",
642 db_flags(lock_ctx
->ctdb_db
));
645 case LOCK_ALLDB_PRIO
:
646 args
[2] = talloc_strdup(args
, "DB");
649 ctdb_db_iterator(ctdb
, lock_ctx
->priority
, db_name_handler
, &list
);
653 args
[2] = talloc_strdup(args
, "DB");
656 for (priority
=1; priority
<NUM_DB_PRIORITIES
; priority
++) {
657 ctdb_db_iterator(ctdb
, priority
, db_name_handler
, &list
);
662 /* Make sure last argument is NULL */
663 args
[nargs
-1] = NULL
;
665 for (i
=0; i
<nargs
-1; i
++) {
666 if (args
[i
] == NULL
) {
678 * Find a lock request that can be scheduled
680 static struct lock_context
*ctdb_find_lock_context(struct ctdb_context
*ctdb
)
682 struct lock_context
*lock_ctx
, *next_ctx
;
683 struct ctdb_db_context
*ctdb_db
;
685 /* First check if there are database lock requests */
687 for (lock_ctx
= ctdb
->lock_pending
; lock_ctx
!= NULL
;
688 lock_ctx
= next_ctx
) {
690 if (lock_ctx
->request
!= NULL
) {
691 /* Found a lock context with a request */
695 next_ctx
= lock_ctx
->next
;
697 DEBUG(DEBUG_INFO
, ("Removing lock context without lock "
699 DLIST_REMOVE(ctdb
->lock_pending
, lock_ctx
);
700 CTDB_DECREMENT_STAT(ctdb
, locks
.num_pending
);
701 if (lock_ctx
->ctdb_db
) {
702 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
,
705 talloc_free(lock_ctx
);
708 /* Next check database queues */
709 for (ctdb_db
= ctdb
->db_list
; ctdb_db
; ctdb_db
= ctdb_db
->next
) {
710 if (ctdb_db
->lock_num_current
==
711 ctdb
->tunable
.lock_processes_per_db
) {
715 for (lock_ctx
= ctdb_db
->lock_pending
; lock_ctx
!= NULL
;
716 lock_ctx
= next_ctx
) {
718 next_ctx
= lock_ctx
->next
;
720 if (lock_ctx
->request
!= NULL
) {
724 DEBUG(DEBUG_INFO
, ("Removing lock context without "
726 DLIST_REMOVE(ctdb_db
->lock_pending
, lock_ctx
);
727 CTDB_DECREMENT_STAT(ctdb
, locks
.num_pending
);
728 CTDB_DECREMENT_DB_STAT(ctdb_db
, locks
.num_pending
);
729 talloc_free(lock_ctx
);
737 * Schedule a new lock child process
738 * Set up callback handler and timeout handler
740 static void ctdb_lock_schedule(struct ctdb_context
*ctdb
)
742 struct lock_context
*lock_ctx
;
745 const char *helper
= CTDB_HELPER_BINDIR
"/ctdb_lock_helper";
746 static const char *prog
= NULL
;
752 t
= getenv("CTDB_LOCK_HELPER");
754 prog
= talloc_strdup(ctdb
, t
);
756 prog
= talloc_strdup(ctdb
, helper
);
758 CTDB_NO_MEMORY_VOID(ctdb
, prog
);
761 /* Find a lock context with requests */
762 lock_ctx
= ctdb_find_lock_context(ctdb
);
763 if (lock_ctx
== NULL
) {
767 lock_ctx
->child
= -1;
768 ret
= pipe(lock_ctx
->fd
);
770 DEBUG(DEBUG_ERR
, ("Failed to create pipe in ctdb_lock_schedule\n"));
774 set_close_on_exec(lock_ctx
->fd
[0]);
776 /* Create data for child process */
777 tmp_ctx
= talloc_new(lock_ctx
);
778 if (tmp_ctx
== NULL
) {
779 DEBUG(DEBUG_ERR
, ("Failed to allocate memory for helper args\n"));
780 close(lock_ctx
->fd
[0]);
781 close(lock_ctx
->fd
[1]);
785 /* Create arguments for lock helper */
786 if (!lock_helper_args(tmp_ctx
, lock_ctx
, lock_ctx
->fd
[1],
788 DEBUG(DEBUG_ERR
, ("Failed to create lock helper args\n"));
789 close(lock_ctx
->fd
[0]);
790 close(lock_ctx
->fd
[1]);
791 talloc_free(tmp_ctx
);
795 if (!ctdb_vfork_with_logging(lock_ctx
, ctdb
, "lock_helper",
796 prog
, argc
, (const char **)args
,
797 NULL
, NULL
, &lock_ctx
->child
)) {
798 DEBUG(DEBUG_ERR
, ("Failed to create a child in ctdb_lock_schedule\n"));
799 close(lock_ctx
->fd
[0]);
800 close(lock_ctx
->fd
[1]);
801 talloc_free(tmp_ctx
);
806 close(lock_ctx
->fd
[1]);
808 talloc_set_destructor(lock_ctx
, ctdb_lock_context_destructor
);
810 talloc_free(tmp_ctx
);
812 /* Set up timeout handler */
813 lock_ctx
->ttimer
= tevent_add_timer(ctdb
->ev
,
815 timeval_current_ofs(10, 0),
816 ctdb_lock_timeout_handler
,
818 if (lock_ctx
->ttimer
== NULL
) {
819 ctdb_kill(ctdb
, lock_ctx
->child
, SIGKILL
);
820 lock_ctx
->child
= -1;
821 talloc_set_destructor(lock_ctx
, NULL
);
822 close(lock_ctx
->fd
[0]);
826 /* Set up callback */
827 lock_ctx
->tfd
= tevent_add_fd(ctdb
->ev
,
833 if (lock_ctx
->tfd
== NULL
) {
834 TALLOC_FREE(lock_ctx
->ttimer
);
835 ctdb_kill(ctdb
, lock_ctx
->child
, SIGKILL
);
836 lock_ctx
->child
= -1;
837 talloc_set_destructor(lock_ctx
, NULL
);
838 close(lock_ctx
->fd
[0]);
841 tevent_fd_set_auto_close(lock_ctx
->tfd
);
843 /* Move the context from pending to current */
844 if (lock_ctx
->type
== LOCK_RECORD
) {
845 DLIST_REMOVE(lock_ctx
->ctdb_db
->lock_pending
, lock_ctx
);
846 DLIST_ADD_END(lock_ctx
->ctdb_db
->lock_current
, lock_ctx
, NULL
);
848 DLIST_REMOVE(ctdb
->lock_pending
, lock_ctx
);
849 DLIST_ADD_END(ctdb
->lock_current
, lock_ctx
, NULL
);
851 CTDB_DECREMENT_STAT(lock_ctx
->ctdb
, locks
.num_pending
);
852 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.num_current
);
853 if (lock_ctx
->ctdb_db
) {
854 lock_ctx
->ctdb_db
->lock_num_current
++;
855 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_pending
);
856 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_current
);
862 * Lock record / db depending on type
864 static struct lock_request
*ctdb_lock_internal(TALLOC_CTX
*mem_ctx
,
865 struct ctdb_context
*ctdb
,
866 struct ctdb_db_context
*ctdb_db
,
869 void (*callback
)(void *, bool),
874 struct lock_context
*lock_ctx
= NULL
;
875 struct lock_request
*request
;
877 if (callback
== NULL
) {
878 DEBUG(DEBUG_WARNING
, ("No callback function specified, not locking\n"));
882 lock_ctx
= talloc_zero(ctdb
, struct lock_context
);
883 if (lock_ctx
== NULL
) {
884 DEBUG(DEBUG_ERR
, ("Failed to create a new lock context\n"));
888 if ((request
= talloc_zero(mem_ctx
, struct lock_request
)) == NULL
) {
889 talloc_free(lock_ctx
);
893 lock_ctx
->type
= type
;
894 lock_ctx
->ctdb
= ctdb
;
895 lock_ctx
->ctdb_db
= ctdb_db
;
896 lock_ctx
->key
.dsize
= key
.dsize
;
898 lock_ctx
->key
.dptr
= talloc_memdup(lock_ctx
, key
.dptr
, key
.dsize
);
899 if (lock_ctx
->key
.dptr
== NULL
) {
900 DEBUG(DEBUG_ERR
, (__location__
"Memory allocation error\n"));
901 talloc_free(lock_ctx
);
904 lock_ctx
->key_hash
= ctdb_hash(&key
);
906 lock_ctx
->key
.dptr
= NULL
;
908 lock_ctx
->priority
= priority
;
909 lock_ctx
->auto_mark
= auto_mark
;
911 lock_ctx
->request
= request
;
912 lock_ctx
->child
= -1;
914 /* Non-record locks are required by recovery and should be scheduled
915 * immediately, so keep them at the head of the pending queue.
917 if (lock_ctx
->type
== LOCK_RECORD
) {
918 DLIST_ADD_END(ctdb_db
->lock_pending
, lock_ctx
, NULL
);
920 DLIST_ADD_END(ctdb
->lock_pending
, lock_ctx
, NULL
);
922 CTDB_INCREMENT_STAT(ctdb
, locks
.num_pending
);
924 CTDB_INCREMENT_DB_STAT(ctdb_db
, locks
.num_pending
);
927 /* Start the timer when we activate the context */
928 lock_ctx
->start_time
= timeval_current();
930 request
->lctx
= lock_ctx
;
931 request
->callback
= callback
;
932 request
->private_data
= private_data
;
934 talloc_set_destructor(request
, ctdb_lock_request_destructor
);
936 ctdb_lock_schedule(ctdb
);
943 * obtain a lock on a record in a database
945 struct lock_request
*ctdb_lock_record(TALLOC_CTX
*mem_ctx
,
946 struct ctdb_db_context
*ctdb_db
,
949 void (*callback
)(void *, bool),
952 return ctdb_lock_internal(mem_ctx
,
965 * obtain a lock on a database
967 struct lock_request
*ctdb_lock_db(TALLOC_CTX
*mem_ctx
,
968 struct ctdb_db_context
*ctdb_db
,
970 void (*callback
)(void *, bool),
973 return ctdb_lock_internal(mem_ctx
,
986 * obtain locks on all databases of specified priority
988 struct lock_request
*ctdb_lock_alldb_prio(TALLOC_CTX
*mem_ctx
,
989 struct ctdb_context
*ctdb
,
992 void (*callback
)(void *, bool),
995 if (priority
< 1 || priority
> NUM_DB_PRIORITIES
) {
996 DEBUG(DEBUG_ERR
, ("Invalid db priority: %u\n", priority
));
1000 return ctdb_lock_internal(mem_ctx
,
1013 * obtain locks on all databases
1015 struct lock_request
*ctdb_lock_alldb(TALLOC_CTX
*mem_ctx
,
1016 struct ctdb_context
*ctdb
,
1018 void (*callback
)(void *, bool),
1021 return ctdb_lock_internal(mem_ctx
,