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"
31 #include "lib/util/sys_rw.h"
33 #include "ctdb_private.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
50 * auto_mark - whether to mark/unmark DBs in before/after callback
51 * = false is used for freezing databases for
52 * recovery since the recovery cannot start till
53 * databases are locked on all the nodes.
54 * = true is used for record locks.
62 static const char * const lock_type_str
[] = {
69 /* lock_context is the common part for a lock request */
71 struct lock_context
*next
, *prev
;
73 struct ctdb_context
*ctdb
;
74 struct ctdb_db_context
*ctdb_db
;
78 struct lock_request
*request
;
81 struct tevent_fd
*tfd
;
82 struct tevent_timer
*ttimer
;
83 struct timeval start_time
;
88 /* lock_request is the client specific part for a lock request */
90 struct lock_context
*lctx
;
91 void (*callback
)(void *, bool);
96 int ctdb_db_iterator(struct ctdb_context
*ctdb
, ctdb_db_handler_t handler
,
99 struct ctdb_db_context
*ctdb_db
;
102 for (ctdb_db
= ctdb
->db_list
; ctdb_db
; ctdb_db
= ctdb_db
->next
) {
103 ret
= handler(ctdb_db
, private_data
);
113 * lock all databases - mark only
115 static int db_lock_mark_handler(struct ctdb_db_context
*ctdb_db
,
118 int tdb_transaction_write_lock_mark(struct tdb_context
*);
120 DEBUG(DEBUG_INFO
, ("marking locked database %s\n", ctdb_db
->db_name
));
122 if (tdb_transaction_write_lock_mark(ctdb_db
->ltdb
->tdb
) != 0) {
123 DEBUG(DEBUG_ERR
, ("Failed to mark (transaction lock) database %s\n",
128 if (tdb_lockall_mark(ctdb_db
->ltdb
->tdb
) != 0) {
129 DEBUG(DEBUG_ERR
, ("Failed to mark (all lock) database %s\n",
137 int ctdb_lockdb_mark(struct ctdb_db_context
*ctdb_db
)
139 if (!ctdb_db_frozen(ctdb_db
)) {
141 ("Attempt to mark database locked when not frozen\n"));
145 return db_lock_mark_handler(ctdb_db
, NULL
);
149 * lock all databases - unmark only
151 static int db_lock_unmark_handler(struct ctdb_db_context
*ctdb_db
,
154 int tdb_transaction_write_lock_unmark(struct tdb_context
*);
156 DEBUG(DEBUG_INFO
, ("unmarking locked database %s\n", ctdb_db
->db_name
));
158 if (tdb_transaction_write_lock_unmark(ctdb_db
->ltdb
->tdb
) != 0) {
159 DEBUG(DEBUG_ERR
, ("Failed to unmark (transaction lock) database %s\n",
164 if (tdb_lockall_unmark(ctdb_db
->ltdb
->tdb
) != 0) {
165 DEBUG(DEBUG_ERR
, ("Failed to unmark (all lock) database %s\n",
173 int ctdb_lockdb_unmark(struct ctdb_db_context
*ctdb_db
)
175 if (!ctdb_db_frozen(ctdb_db
)) {
177 ("Attempt to unmark database locked when not frozen\n"));
181 return db_lock_unmark_handler(ctdb_db
, NULL
);
184 static void ctdb_lock_schedule(struct ctdb_context
*ctdb
);
187 * Destructor to kill the child locking process
189 static int ctdb_lock_context_destructor(struct lock_context
*lock_ctx
)
191 if (lock_ctx
->request
) {
192 lock_ctx
->request
->lctx
= NULL
;
194 if (lock_ctx
->child
> 0) {
195 ctdb_kill(lock_ctx
->ctdb
, lock_ctx
->child
, SIGTERM
);
196 if (lock_ctx
->type
== LOCK_RECORD
) {
197 DLIST_REMOVE(lock_ctx
->ctdb_db
->lock_current
, lock_ctx
);
199 DLIST_REMOVE(lock_ctx
->ctdb
->lock_current
, lock_ctx
);
201 if (lock_ctx
->ctdb_db
) {
202 lock_ctx
->ctdb_db
->lock_num_current
--;
204 CTDB_DECREMENT_STAT(lock_ctx
->ctdb
, locks
.num_current
);
205 if (lock_ctx
->ctdb_db
) {
206 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_current
);
209 if (lock_ctx
->type
== LOCK_RECORD
) {
210 DLIST_REMOVE(lock_ctx
->ctdb_db
->lock_pending
, lock_ctx
);
212 DLIST_REMOVE(lock_ctx
->ctdb
->lock_pending
, lock_ctx
);
214 CTDB_DECREMENT_STAT(lock_ctx
->ctdb
, locks
.num_pending
);
215 if (lock_ctx
->ctdb_db
) {
216 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_pending
);
220 ctdb_lock_schedule(lock_ctx
->ctdb
);
227 * Destructor to remove lock request
229 static int ctdb_lock_request_destructor(struct lock_request
*lock_request
)
231 if (lock_request
->lctx
== NULL
) {
235 lock_request
->lctx
->request
= NULL
;
236 TALLOC_FREE(lock_request
->lctx
);
242 * Process all the callbacks waiting for lock
244 * If lock has failed, callback is executed with locked=false
246 static void process_callbacks(struct lock_context
*lock_ctx
, bool locked
)
248 struct lock_request
*request
;
249 bool auto_mark
= lock_ctx
->auto_mark
;
251 if (auto_mark
&& locked
) {
252 switch (lock_ctx
->type
) {
254 tdb_chainlock_mark(lock_ctx
->ctdb_db
->ltdb
->tdb
, lock_ctx
->key
);
258 ctdb_lockdb_mark(lock_ctx
->ctdb_db
);
263 request
= lock_ctx
->request
;
265 /* Since request may be freed in the callback, unset the lock
266 * context, so request destructor will not free lock context.
268 request
->lctx
= NULL
;
271 /* Since request may be freed in the callback, unset the request */
272 lock_ctx
->request
= NULL
;
274 request
->callback(request
->private_data
, locked
);
281 switch (lock_ctx
->type
) {
283 tdb_chainlock_unmark(lock_ctx
->ctdb_db
->ltdb
->tdb
, lock_ctx
->key
);
287 ctdb_lockdb_unmark(lock_ctx
->ctdb_db
);
292 talloc_free(lock_ctx
);
296 static int lock_bucket_id(double t
)
298 double ms
= 1.e
-3, s
= 1;
303 } else if (t
< 10*ms
) {
305 } else if (t
< 100*ms
) {
307 } else if (t
< 1*s
) {
309 } else if (t
< 2*s
) {
311 } else if (t
< 4*s
) {
313 } else if (t
< 8*s
) {
315 } else if (t
< 16*s
) {
317 } else if (t
< 32*s
) {
319 } else if (t
< 64*s
) {
329 * Callback routine when the required locks are obtained.
330 * Called from parent context
332 static void ctdb_lock_handler(struct tevent_context
*ev
,
333 struct tevent_fd
*tfd
,
337 struct lock_context
*lock_ctx
;
343 lock_ctx
= talloc_get_type_abort(private_data
, struct lock_context
);
345 /* cancel the timeout event */
346 TALLOC_FREE(lock_ctx
->ttimer
);
348 t
= timeval_elapsed(&lock_ctx
->start_time
);
349 id
= lock_bucket_id(t
);
351 /* Read the status from the child process */
352 if (sys_read(lock_ctx
->fd
[0], &c
, 1) != 1) {
355 locked
= (c
== 0 ? true : false);
358 /* Update statistics */
359 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.num_calls
);
360 if (lock_ctx
->ctdb_db
) {
361 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_calls
);
365 if (lock_ctx
->ctdb_db
) {
366 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.buckets
[id
]);
367 CTDB_UPDATE_LATENCY(lock_ctx
->ctdb
, lock_ctx
->ctdb_db
,
368 lock_type_str
[lock_ctx
->type
], locks
.latency
,
369 lock_ctx
->start_time
);
371 CTDB_UPDATE_DB_LATENCY(lock_ctx
->ctdb_db
, lock_type_str
[lock_ctx
->type
], locks
.latency
, t
);
372 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.buckets
[id
]);
375 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.num_failed
);
376 if (lock_ctx
->ctdb_db
) {
377 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_failed
);
381 process_callbacks(lock_ctx
, locked
);
384 struct lock_log_entry
{
385 struct db_hash_context
*lock_log
;
387 unsigned long log_sec
;
388 struct tevent_timer
*timer
;
391 static int lock_log_fetch_parser(uint8_t *keybuf
, size_t keylen
,
392 uint8_t *databuf
, size_t datalen
,
395 struct lock_log_entry
**entry
=
396 (struct lock_log_entry
**)private_data
;
398 if (datalen
!= sizeof(struct lock_log_entry
*)) {
402 *entry
= talloc_get_type_abort(*(void **)databuf
,
403 struct lock_log_entry
);
407 static void lock_log_cleanup(struct tevent_context
*ev
,
408 struct tevent_timer
*ttimer
,
409 struct timeval current_time
,
412 struct lock_log_entry
*entry
= talloc_get_type_abort(
413 private_data
, struct lock_log_entry
);
418 ret
= db_hash_delete(entry
->lock_log
, entry
->key
.dptr
,
426 static bool lock_log_skip(struct tevent_context
*ev
,
427 struct db_hash_context
*lock_log
,
428 TDB_DATA key
, unsigned long elapsed_sec
)
430 struct lock_log_entry
*entry
= NULL
;
433 ret
= db_hash_fetch(lock_log
, key
.dptr
, key
.dsize
,
434 lock_log_fetch_parser
, &entry
);
437 entry
= talloc_zero(lock_log
, struct lock_log_entry
);
442 entry
->lock_log
= lock_log
;
444 entry
->key
.dptr
= talloc_memdup(entry
, key
.dptr
, key
.dsize
);
445 if (entry
->key
.dptr
== NULL
) {
449 entry
->key
.dsize
= key
.dsize
;
451 entry
->log_sec
= elapsed_sec
;
452 entry
->timer
= tevent_add_timer(ev
, entry
,
453 timeval_current_ofs(30, 0),
454 lock_log_cleanup
, entry
);
455 if (entry
->timer
== NULL
) {
460 ret
= db_hash_add(lock_log
, key
.dptr
, key
.dsize
,
462 sizeof(struct lock_log_entry
*));
470 } else if (ret
== EINVAL
) {
472 ret
= db_hash_delete(lock_log
, key
.dptr
, key
.dsize
);
479 } else if (ret
== 0) {
481 if (elapsed_sec
<= entry
->log_sec
) {
485 entry
->log_sec
= elapsed_sec
;
487 TALLOC_FREE(entry
->timer
);
488 entry
->timer
= tevent_add_timer(ev
, entry
,
489 timeval_current_ofs(30, 0),
490 lock_log_cleanup
, entry
);
491 if (entry
->timer
== NULL
) {
492 ret
= db_hash_delete(lock_log
, key
.dptr
, key
.dsize
);
509 * Callback routine when required locks are not obtained within timeout
510 * Called from parent context
512 static void ctdb_lock_timeout_handler(struct tevent_context
*ev
,
513 struct tevent_timer
*ttimer
,
514 struct timeval current_time
,
517 static char debug_locks
[PATH_MAX
+1] = "";
518 struct lock_context
*lock_ctx
;
519 struct ctdb_context
*ctdb
;
525 lock_ctx
= talloc_get_type_abort(private_data
, struct lock_context
);
526 ctdb
= lock_ctx
->ctdb
;
528 elapsed_time
= timeval_elapsed(&lock_ctx
->start_time
);
530 /* For database locks, always log */
531 if (lock_ctx
->type
== LOCK_DB
) {
533 ("Unable to get DB lock on database %s for "
535 lock_ctx
->ctdb_db
->db_name
, elapsed_time
));
539 /* For record locks, check if we have already logged */
540 skip
= lock_log_skip(ev
, lock_ctx
->ctdb_db
->lock_log
,
541 lock_ctx
->key
, (unsigned long)elapsed_time
);
543 goto skip_lock_debug
;
546 keystr
= hex_encode_talloc(lock_ctx
, lock_ctx
->key
.dptr
,
547 lock_ctx
->key
.dsize
);
549 ("Unable to get RECORD lock on database %s for %.0lf seconds"
551 lock_ctx
->ctdb_db
->db_name
, elapsed_time
,
552 keystr
? keystr
: ""));
555 /* If a node stopped/banned, don't spam the logs */
556 if (ctdb
->nodes
[ctdb
->pnn
]->flags
& NODE_FLAGS_INACTIVE
) {
557 goto skip_lock_debug
;
562 if (ctdb_set_helper("lock debugging helper",
563 debug_locks
, sizeof(debug_locks
),
565 getenv("CTDB_BASE"), "debug_locks.sh")) {
568 execl(debug_locks
, debug_locks
, NULL
);
571 ctdb_track_child(ctdb
, pid
);
575 " Unable to setup lock debugging\n"));
580 /* reset the timeout timer */
581 // talloc_free(lock_ctx->ttimer);
582 lock_ctx
->ttimer
= tevent_add_timer(ctdb
->ev
,
584 timeval_current_ofs(10, 0),
585 ctdb_lock_timeout_handler
,
589 static bool lock_helper_args(TALLOC_CTX
*mem_ctx
,
590 struct lock_context
*lock_ctx
, int fd
,
591 int *argc
, const char ***argv
)
593 const char **args
= NULL
;
596 switch (lock_ctx
->type
) {
606 /* Add extra argument for null termination */
609 args
= talloc_array(mem_ctx
, const char *, nargs
);
614 args
[0] = talloc_asprintf(args
, "%d", getpid());
615 args
[1] = talloc_asprintf(args
, "%d", fd
);
617 switch (lock_ctx
->type
) {
619 args
[2] = talloc_strdup(args
, "RECORD");
620 args
[3] = talloc_strdup(args
, lock_ctx
->ctdb_db
->db_path
);
621 args
[4] = talloc_asprintf(args
, "0x%x",
622 tdb_get_flags(lock_ctx
->ctdb_db
->ltdb
->tdb
));
623 if (lock_ctx
->key
.dsize
== 0) {
624 args
[5] = talloc_strdup(args
, "NULL");
626 args
[5] = hex_encode_talloc(args
, lock_ctx
->key
.dptr
, lock_ctx
->key
.dsize
);
631 args
[2] = talloc_strdup(args
, "DB");
632 args
[3] = talloc_strdup(args
, lock_ctx
->ctdb_db
->db_path
);
633 args
[4] = talloc_asprintf(args
, "0x%x",
634 tdb_get_flags(lock_ctx
->ctdb_db
->ltdb
->tdb
));
638 /* Make sure last argument is NULL */
639 args
[nargs
-1] = NULL
;
641 for (i
=0; i
<nargs
-1; i
++) {
642 if (args
[i
] == NULL
) {
654 * Find a lock request that can be scheduled
656 static struct lock_context
*ctdb_find_lock_context(struct ctdb_context
*ctdb
)
658 struct lock_context
*lock_ctx
, *next_ctx
;
659 struct ctdb_db_context
*ctdb_db
;
661 /* First check if there are database lock requests */
663 for (lock_ctx
= ctdb
->lock_pending
; lock_ctx
!= NULL
;
664 lock_ctx
= next_ctx
) {
666 if (lock_ctx
->request
!= NULL
) {
667 /* Found a lock context with a request */
671 next_ctx
= lock_ctx
->next
;
673 DEBUG(DEBUG_INFO
, ("Removing lock context without lock "
675 DLIST_REMOVE(ctdb
->lock_pending
, lock_ctx
);
676 CTDB_DECREMENT_STAT(ctdb
, locks
.num_pending
);
677 if (lock_ctx
->ctdb_db
) {
678 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
,
681 talloc_free(lock_ctx
);
684 /* Next check database queues */
685 for (ctdb_db
= ctdb
->db_list
; ctdb_db
; ctdb_db
= ctdb_db
->next
) {
686 if (ctdb_db
->lock_num_current
==
687 ctdb
->tunable
.lock_processes_per_db
) {
691 for (lock_ctx
= ctdb_db
->lock_pending
; lock_ctx
!= NULL
;
692 lock_ctx
= next_ctx
) {
694 next_ctx
= lock_ctx
->next
;
696 if (lock_ctx
->request
!= NULL
) {
700 DEBUG(DEBUG_INFO
, ("Removing lock context without "
702 DLIST_REMOVE(ctdb_db
->lock_pending
, lock_ctx
);
703 CTDB_DECREMENT_STAT(ctdb
, locks
.num_pending
);
704 CTDB_DECREMENT_DB_STAT(ctdb_db
, locks
.num_pending
);
705 talloc_free(lock_ctx
);
713 * Schedule a new lock child process
714 * Set up callback handler and timeout handler
716 static void ctdb_lock_schedule(struct ctdb_context
*ctdb
)
718 struct lock_context
*lock_ctx
;
721 static char prog
[PATH_MAX
+1] = "";
724 if (!ctdb_set_helper("lock helper",
727 CTDB_HELPER_BINDIR
, "ctdb_lock_helper")) {
728 ctdb_die(ctdb
, __location__
729 " Unable to set lock helper\n");
732 /* Find a lock context with requests */
733 lock_ctx
= ctdb_find_lock_context(ctdb
);
734 if (lock_ctx
== NULL
) {
738 lock_ctx
->child
= -1;
739 ret
= pipe(lock_ctx
->fd
);
741 DEBUG(DEBUG_ERR
, ("Failed to create pipe in ctdb_lock_schedule\n"));
745 set_close_on_exec(lock_ctx
->fd
[0]);
747 /* Create data for child process */
748 tmp_ctx
= talloc_new(lock_ctx
);
749 if (tmp_ctx
== NULL
) {
750 DEBUG(DEBUG_ERR
, ("Failed to allocate memory for helper args\n"));
751 close(lock_ctx
->fd
[0]);
752 close(lock_ctx
->fd
[1]);
756 if (! ctdb
->do_setsched
) {
757 ret
= setenv("CTDB_NOSETSCHED", "1", 1);
760 ("Failed to set CTDB_NOSETSCHED variable\n"));
764 /* Create arguments for lock helper */
765 if (!lock_helper_args(tmp_ctx
, lock_ctx
, lock_ctx
->fd
[1],
767 DEBUG(DEBUG_ERR
, ("Failed to create lock helper args\n"));
768 close(lock_ctx
->fd
[0]);
769 close(lock_ctx
->fd
[1]);
770 talloc_free(tmp_ctx
);
774 lock_ctx
->child
= ctdb_vfork_exec(lock_ctx
, ctdb
, prog
, argc
,
775 (const char **)args
);
776 if (lock_ctx
->child
== -1) {
777 DEBUG(DEBUG_ERR
, ("Failed to create a child in ctdb_lock_schedule\n"));
778 close(lock_ctx
->fd
[0]);
779 close(lock_ctx
->fd
[1]);
780 talloc_free(tmp_ctx
);
785 close(lock_ctx
->fd
[1]);
787 talloc_free(tmp_ctx
);
789 /* Set up timeout handler */
790 lock_ctx
->ttimer
= tevent_add_timer(ctdb
->ev
,
792 timeval_current_ofs(10, 0),
793 ctdb_lock_timeout_handler
,
795 if (lock_ctx
->ttimer
== NULL
) {
796 ctdb_kill(ctdb
, lock_ctx
->child
, SIGTERM
);
797 lock_ctx
->child
= -1;
798 close(lock_ctx
->fd
[0]);
802 /* Set up callback */
803 lock_ctx
->tfd
= tevent_add_fd(ctdb
->ev
,
809 if (lock_ctx
->tfd
== NULL
) {
810 TALLOC_FREE(lock_ctx
->ttimer
);
811 ctdb_kill(ctdb
, lock_ctx
->child
, SIGTERM
);
812 lock_ctx
->child
= -1;
813 close(lock_ctx
->fd
[0]);
816 tevent_fd_set_auto_close(lock_ctx
->tfd
);
818 /* Move the context from pending to current */
819 if (lock_ctx
->type
== LOCK_RECORD
) {
820 DLIST_REMOVE(lock_ctx
->ctdb_db
->lock_pending
, lock_ctx
);
821 DLIST_ADD_END(lock_ctx
->ctdb_db
->lock_current
, lock_ctx
);
823 DLIST_REMOVE(ctdb
->lock_pending
, lock_ctx
);
824 DLIST_ADD_END(ctdb
->lock_current
, lock_ctx
);
826 CTDB_DECREMENT_STAT(lock_ctx
->ctdb
, locks
.num_pending
);
827 CTDB_INCREMENT_STAT(lock_ctx
->ctdb
, locks
.num_current
);
828 if (lock_ctx
->ctdb_db
) {
829 lock_ctx
->ctdb_db
->lock_num_current
++;
830 CTDB_DECREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_pending
);
831 CTDB_INCREMENT_DB_STAT(lock_ctx
->ctdb_db
, locks
.num_current
);
837 * Lock record / db depending on type
839 static struct lock_request
*ctdb_lock_internal(TALLOC_CTX
*mem_ctx
,
840 struct ctdb_context
*ctdb
,
841 struct ctdb_db_context
*ctdb_db
,
844 void (*callback
)(void *, bool),
849 struct lock_context
*lock_ctx
= NULL
;
850 struct lock_request
*request
;
852 if (callback
== NULL
) {
853 DEBUG(DEBUG_WARNING
, ("No callback function specified, not locking\n"));
857 lock_ctx
= talloc_zero(ctdb
, struct lock_context
);
858 if (lock_ctx
== NULL
) {
859 DEBUG(DEBUG_ERR
, ("Failed to create a new lock context\n"));
863 if ((request
= talloc_zero(mem_ctx
, struct lock_request
)) == NULL
) {
864 talloc_free(lock_ctx
);
868 lock_ctx
->type
= type
;
869 lock_ctx
->ctdb
= ctdb
;
870 lock_ctx
->ctdb_db
= ctdb_db
;
871 lock_ctx
->key
.dsize
= key
.dsize
;
873 lock_ctx
->key
.dptr
= talloc_memdup(lock_ctx
, key
.dptr
, key
.dsize
);
874 if (lock_ctx
->key
.dptr
== NULL
) {
875 DEBUG(DEBUG_ERR
, (__location__
"Memory allocation error\n"));
876 talloc_free(lock_ctx
);
877 talloc_free(request
);
880 lock_ctx
->key_hash
= ctdb_hash(&key
);
882 lock_ctx
->key
.dptr
= NULL
;
884 lock_ctx
->priority
= priority
;
885 lock_ctx
->auto_mark
= auto_mark
;
887 lock_ctx
->request
= request
;
888 lock_ctx
->child
= -1;
890 /* Non-record locks are required by recovery and should be scheduled
891 * immediately, so keep them at the head of the pending queue.
893 if (lock_ctx
->type
== LOCK_RECORD
) {
894 DLIST_ADD_END(ctdb_db
->lock_pending
, lock_ctx
);
896 DLIST_ADD_END(ctdb
->lock_pending
, lock_ctx
);
898 CTDB_INCREMENT_STAT(ctdb
, locks
.num_pending
);
900 CTDB_INCREMENT_DB_STAT(ctdb_db
, locks
.num_pending
);
903 /* Start the timer when we activate the context */
904 lock_ctx
->start_time
= timeval_current();
906 request
->lctx
= lock_ctx
;
907 request
->callback
= callback
;
908 request
->private_data
= private_data
;
910 talloc_set_destructor(request
, ctdb_lock_request_destructor
);
911 talloc_set_destructor(lock_ctx
, ctdb_lock_context_destructor
);
913 ctdb_lock_schedule(ctdb
);
920 * obtain a lock on a record in a database
922 struct lock_request
*ctdb_lock_record(TALLOC_CTX
*mem_ctx
,
923 struct ctdb_db_context
*ctdb_db
,
926 void (*callback
)(void *, bool),
929 return ctdb_lock_internal(mem_ctx
,
942 * obtain a lock on a database
944 struct lock_request
*ctdb_lock_db(TALLOC_CTX
*mem_ctx
,
945 struct ctdb_db_context
*ctdb_db
,
947 void (*callback
)(void *, bool),
950 return ctdb_lock_internal(mem_ctx
,