ctdb-tests: Simplify and rename wait_until_ips_are_on_nodeglob()
[Samba.git] / ctdb / server / ctdb_lock.c
blobe6653102ac68291e5929a8583686ff51574834e6
1 /*
2 ctdb lock handling
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/>.
20 #include "includes.h"
21 #include "include/ctdb_private.h"
22 #include "include/ctdb_protocol.h"
23 #include "tevent.h"
24 #include "tdb.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
46 enum lock_type {
47 LOCK_RECORD,
48 LOCK_DB,
49 LOCK_ALLDB_PRIO,
50 LOCK_ALLDB,
53 static const char * const lock_type_str[] = {
54 "lock_record",
55 "lock_db",
56 "lock_alldb_prio",
57 "lock_alldb",
60 struct lock_request;
62 /* lock_context is the common part for a lock request */
63 struct lock_context {
64 struct lock_context *next, *prev;
65 enum lock_type type;
66 struct ctdb_context *ctdb;
67 struct ctdb_db_context *ctdb_db;
68 TDB_DATA key;
69 uint32_t priority;
70 bool auto_mark;
71 struct lock_request *request;
72 pid_t child;
73 int fd[2];
74 struct tevent_fd *tfd;
75 struct tevent_timer *ttimer;
76 struct timeval start_time;
77 uint32_t key_hash;
78 bool can_schedule;
81 /* lock_request is the client specific part for a lock request */
82 struct lock_request {
83 struct lock_context *lctx;
84 void (*callback)(void *, bool);
85 void *private_data;
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) {
98 return false;
101 if (strstr(name, "brlock") ||
102 strstr(name, "g_lock") ||
103 strstr(name, "notify_onelevel") ||
104 strstr(name, "serverid") ||
105 strstr(name, "xattr_tdb")) {
106 return true;
109 return false;
112 typedef int (*db_handler_t)(struct ctdb_db_context *ctdb_db,
113 uint32_t priority,
114 void *private_data);
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;
120 int ret;
122 for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
123 if (ctdb_db->priority != priority) {
124 continue;
126 if (later_db(ctdb, ctdb_db->db_name)) {
127 continue;
129 ret = handler(ctdb_db, priority, private_data);
130 if (ret != 0) {
131 return -1;
135 /* If priority != 1, later_db check is not required and can return */
136 if (priority != 1) {
137 return 0;
140 for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
141 if (!later_db(ctdb, ctdb_db->db_name)) {
142 continue;
144 ret = handler(ctdb_db, priority, private_data);
145 if (ret != 0) {
146 return -1;
150 return 0;
155 * lock all databases - mark only
157 static int db_lock_mark_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
158 void *private_data)
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",
167 ctdb_db->db_name));
168 return -1;
171 if (tdb_lockall_mark(ctdb_db->ltdb->tdb) != 0) {
172 DEBUG(DEBUG_ERR, ("Failed to mark (all lock) database %s\n",
173 ctdb_db->db_name));
174 return -1;
177 return 0;
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"));
191 return -1;
194 return ctdb_db_iterator(ctdb, priority, db_lock_mark_handler, NULL);
197 static int ctdb_lockall_mark(struct ctdb_context *ctdb)
199 uint32_t priority;
201 for (priority=1; priority<=NUM_DB_PRIORITIES; priority++) {
202 if (ctdb_db_iterator(ctdb, priority, db_lock_mark_handler, NULL) != 0) {
203 return -1;
207 return 0;
212 * lock all databases - unmark only
214 static int db_lock_unmark_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
215 void *private_data)
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",
224 ctdb_db->db_name));
225 return -1;
228 if (tdb_lockall_unmark(ctdb_db->ltdb->tdb) != 0) {
229 DEBUG(DEBUG_ERR, ("Failed to unmark (all lock) database %s\n",
230 ctdb_db->db_name));
231 return -1;
234 return 0;
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"));
248 return -1;
251 return ctdb_db_iterator(ctdb, priority, db_lock_unmark_handler, NULL);
254 static int ctdb_lockall_unmark(struct ctdb_context *ctdb)
256 uint32_t priority;
258 for (priority=NUM_DB_PRIORITIES; priority>0; priority--) {
259 if (ctdb_db_iterator(ctdb, priority, db_lock_unmark_handler, NULL) != 0) {
260 return -1;
264 return 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);
282 } else {
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);
292 } else {
293 if (lock_ctx->type == LOCK_RECORD) {
294 DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
295 } else {
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);
306 return 0;
311 * Destructor to remove lock request
313 static int ctdb_lock_request_destructor(struct lock_request *lock_request)
315 TALLOC_FREE(lock_request->lctx);
316 return 0;
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) {
330 case LOCK_RECORD:
331 tdb_chainlock_mark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
332 break;
334 case LOCK_DB:
335 tdb_lockall_mark(lock_ctx->ctdb_db->ltdb->tdb);
336 break;
338 case LOCK_ALLDB_PRIO:
339 ctdb_lockall_mark_prio(lock_ctx->ctdb, lock_ctx->priority);
340 break;
342 case LOCK_ALLDB:
343 ctdb_lockall_mark(lock_ctx->ctdb);
344 break;
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) {
357 case LOCK_RECORD:
358 tdb_chainlock_unmark(lock_ctx->ctdb_db->ltdb->tdb, lock_ctx->key);
359 break;
361 case LOCK_DB:
362 tdb_lockall_unmark(lock_ctx->ctdb_db->ltdb->tdb);
363 break;
365 case LOCK_ALLDB_PRIO:
366 ctdb_lockall_unmark_prio(lock_ctx->ctdb, lock_ctx->priority);
367 break;
369 case LOCK_ALLDB:
370 ctdb_lockall_unmark(lock_ctx->ctdb);
371 break;
377 static int lock_bucket_id(double t)
379 double ms = 1.e-3, s = 1;
380 int id;
382 if (t < 1*ms) {
383 id = 0;
384 } else if (t < 10*ms) {
385 id = 1;
386 } else if (t < 100*ms) {
387 id = 2;
388 } else if (t < 1*s) {
389 id = 3;
390 } else if (t < 2*s) {
391 id = 4;
392 } else if (t < 4*s) {
393 id = 5;
394 } else if (t < 8*s) {
395 id = 6;
396 } else if (t < 16*s) {
397 id = 7;
398 } else if (t < 32*s) {
399 id = 8;
400 } else if (t < 64*s) {
401 id = 9;
402 } else {
403 id = 10;
406 return id;
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,
415 uint16_t flags,
416 void *private_data)
418 struct lock_context *lock_ctx;
419 TALLOC_CTX *tmp_ctx = NULL;
420 char c;
421 bool locked;
422 double t;
423 int id;
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) {
440 locked = false;
441 } else {
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);
451 if (locked) {
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]);
461 } else {
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,
483 void *private_data)
485 static const char * debug_locks = NULL;
486 struct lock_context *lock_ctx;
487 struct ctdb_context *ctdb;
488 pid_t pid;
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 return;
497 if (lock_ctx->ctdb_db) {
498 DEBUG(DEBUG_WARNING,
499 ("Unable to get %s lock on database %s for %.0lf seconds\n",
500 (lock_ctx->type == LOCK_RECORD ? "RECORD" : "DB"),
501 lock_ctx->ctdb_db->db_name,
502 timeval_elapsed(&lock_ctx->start_time)));
503 } else {
504 DEBUG(DEBUG_WARNING,
505 ("Unable to get ALLDB locks for %.0lf seconds\n",
506 timeval_elapsed(&lock_ctx->start_time)));
509 /* Fire a child process to find the blocking process. */
510 if (debug_locks == NULL) {
511 debug_locks = getenv("CTDB_DEBUG_LOCKS");
512 if (debug_locks == NULL) {
513 debug_locks = talloc_asprintf(ctdb,
514 "%s/debug_locks.sh",
515 getenv("CTDB_BASE"));
518 if (debug_locks != NULL) {
519 pid = vfork();
520 if (pid == 0) {
521 execl(debug_locks, debug_locks, NULL);
522 _exit(0);
524 ctdb_track_child(ctdb, pid);
525 } else {
526 DEBUG(DEBUG_WARNING,
527 (__location__
528 " Unable to setup lock debugging - no memory?\n"));
531 /* reset the timeout timer */
532 // talloc_free(lock_ctx->ttimer);
533 lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
534 lock_ctx,
535 timeval_current_ofs(10, 0),
536 ctdb_lock_timeout_handler,
537 (void *)lock_ctx);
541 static int db_count_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
542 void *private_data)
544 int *count = (int *)private_data;
546 (*count)++;
548 return 0;
551 struct db_namelist {
552 const char **names;
553 int n;
556 static int db_name_handler(struct ctdb_db_context *ctdb_db, uint32_t priority,
557 void *private_data)
559 struct db_namelist *list = (struct db_namelist *)private_data;
561 list->names[list->n] = talloc_strdup(list->names, ctdb_db->db_path);
562 list->n++;
564 return 0;
567 static bool lock_helper_args(TALLOC_CTX *mem_ctx,
568 struct lock_context *lock_ctx, int fd,
569 int *argc, const char ***argv)
571 struct ctdb_context *ctdb = lock_ctx->ctdb;
572 const char **args = NULL;
573 int nargs, i;
574 int priority;
575 struct db_namelist list;
577 switch (lock_ctx->type) {
578 case LOCK_RECORD:
579 nargs = 5;
580 break;
582 case LOCK_DB:
583 nargs = 4;
584 break;
586 case LOCK_ALLDB_PRIO:
587 nargs = 3;
588 ctdb_db_iterator(ctdb, lock_ctx->priority, db_count_handler, &nargs);
589 break;
591 case LOCK_ALLDB:
592 nargs = 3;
593 for (priority=1; priority<NUM_DB_PRIORITIES; priority++) {
594 ctdb_db_iterator(ctdb, priority, db_count_handler, &nargs);
596 break;
599 /* Add extra argument for null termination */
600 nargs++;
602 args = talloc_array(mem_ctx, const char *, nargs);
603 if (args == NULL) {
604 return false;
607 args[0] = talloc_asprintf(args, "%d", getpid());
608 args[1] = talloc_asprintf(args, "%d", fd);
610 switch (lock_ctx->type) {
611 case LOCK_RECORD:
612 args[2] = talloc_strdup(args, "RECORD");
613 args[3] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
614 if (lock_ctx->key.dsize == 0) {
615 args[4] = talloc_strdup(args, "NULL");
616 } else {
617 args[4] = hex_encode_talloc(args, lock_ctx->key.dptr, lock_ctx->key.dsize);
619 break;
621 case LOCK_DB:
622 args[2] = talloc_strdup(args, "DB");
623 args[3] = talloc_strdup(args, lock_ctx->ctdb_db->db_path);
624 break;
626 case LOCK_ALLDB_PRIO:
627 args[2] = talloc_strdup(args, "DB");
628 list.names = args;
629 list.n = 3;
630 ctdb_db_iterator(ctdb, lock_ctx->priority, db_name_handler, &list);
631 break;
633 case LOCK_ALLDB:
634 args[2] = talloc_strdup(args, "DB");
635 list.names = args;
636 list.n = 3;
637 for (priority=1; priority<NUM_DB_PRIORITIES; priority++) {
638 ctdb_db_iterator(ctdb, priority, db_name_handler, &list);
640 break;
643 /* Make sure last argument is NULL */
644 args[nargs-1] = NULL;
646 for (i=0; i<nargs-1; i++) {
647 if (args[i] == NULL) {
648 talloc_free(args);
649 return false;
653 *argc = nargs;
654 *argv = args;
655 return true;
659 * Find a lock request that can be scheduled
661 static struct lock_context *ctdb_find_lock_context(struct ctdb_context *ctdb)
663 struct lock_context *lock_ctx, *next_ctx;
664 struct ctdb_db_context *ctdb_db;
666 /* First check if there are database lock requests */
668 for (lock_ctx = ctdb->lock_pending; lock_ctx != NULL;
669 lock_ctx = next_ctx) {
671 if (lock_ctx->request != NULL) {
672 /* Found a lock context with a request */
673 return lock_ctx;
676 next_ctx = lock_ctx->next;
678 DEBUG(DEBUG_INFO, ("Removing lock context without lock "
679 "request\n"));
680 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
681 CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
682 if (lock_ctx->ctdb_db) {
683 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db,
684 locks.num_pending);
686 talloc_free(lock_ctx);
689 /* Next check database queues */
690 for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
691 if (ctdb_db->lock_num_current ==
692 ctdb->tunable.lock_processes_per_db) {
693 continue;
696 for (lock_ctx = ctdb_db->lock_pending; lock_ctx != NULL;
697 lock_ctx = next_ctx) {
699 next_ctx = lock_ctx->next;
701 if (lock_ctx->request != NULL) {
702 return lock_ctx;
705 DEBUG(DEBUG_INFO, ("Removing lock context without "
706 "lock request\n"));
707 DLIST_REMOVE(ctdb_db->lock_pending, lock_ctx);
708 CTDB_DECREMENT_STAT(ctdb, locks.num_pending);
709 CTDB_DECREMENT_DB_STAT(ctdb_db, locks.num_pending);
710 talloc_free(lock_ctx);
714 return NULL;
718 * Schedule a new lock child process
719 * Set up callback handler and timeout handler
721 static void ctdb_lock_schedule(struct ctdb_context *ctdb)
723 struct lock_context *lock_ctx;
724 int ret, argc;
725 TALLOC_CTX *tmp_ctx;
726 const char *helper = BINDIR "/ctdb_lock_helper";
727 static const char *prog = NULL;
728 const char **args;
730 if (prog == NULL) {
731 const char *t;
733 t = getenv("CTDB_LOCK_HELPER");
734 if (t != NULL) {
735 prog = talloc_strdup(ctdb, t);
736 } else {
737 prog = talloc_strdup(ctdb, helper);
739 CTDB_NO_MEMORY_VOID(ctdb, prog);
742 /* Find a lock context with requests */
743 lock_ctx = ctdb_find_lock_context(ctdb);
744 if (lock_ctx == NULL) {
745 return;
748 lock_ctx->child = -1;
749 ret = pipe(lock_ctx->fd);
750 if (ret != 0) {
751 DEBUG(DEBUG_ERR, ("Failed to create pipe in ctdb_lock_schedule\n"));
752 return;
755 set_close_on_exec(lock_ctx->fd[0]);
757 /* Create data for child process */
758 tmp_ctx = talloc_new(lock_ctx);
759 if (tmp_ctx == NULL) {
760 DEBUG(DEBUG_ERR, ("Failed to allocate memory for helper args\n"));
761 close(lock_ctx->fd[0]);
762 close(lock_ctx->fd[1]);
763 return;
766 /* Create arguments for lock helper */
767 if (!lock_helper_args(tmp_ctx, lock_ctx, lock_ctx->fd[1],
768 &argc, &args)) {
769 DEBUG(DEBUG_ERR, ("Failed to create lock helper args\n"));
770 close(lock_ctx->fd[0]);
771 close(lock_ctx->fd[1]);
772 talloc_free(tmp_ctx);
773 return;
776 if (!ctdb_vfork_with_logging(lock_ctx, ctdb, "lock_helper",
777 prog, argc, (const char **)args,
778 NULL, NULL, &lock_ctx->child)) {
779 DEBUG(DEBUG_ERR, ("Failed to create a child in ctdb_lock_schedule\n"));
780 close(lock_ctx->fd[0]);
781 close(lock_ctx->fd[1]);
782 talloc_free(tmp_ctx);
783 return;
786 /* Parent process */
787 close(lock_ctx->fd[1]);
789 talloc_set_destructor(lock_ctx, ctdb_lock_context_destructor);
791 talloc_free(tmp_ctx);
793 /* Set up timeout handler */
794 lock_ctx->ttimer = tevent_add_timer(ctdb->ev,
795 lock_ctx,
796 timeval_current_ofs(10, 0),
797 ctdb_lock_timeout_handler,
798 (void *)lock_ctx);
799 if (lock_ctx->ttimer == NULL) {
800 ctdb_kill(ctdb, lock_ctx->child, SIGKILL);
801 lock_ctx->child = -1;
802 talloc_set_destructor(lock_ctx, NULL);
803 close(lock_ctx->fd[0]);
804 return;
807 /* Set up callback */
808 lock_ctx->tfd = tevent_add_fd(ctdb->ev,
809 lock_ctx,
810 lock_ctx->fd[0],
811 EVENT_FD_READ,
812 ctdb_lock_handler,
813 (void *)lock_ctx);
814 if (lock_ctx->tfd == NULL) {
815 TALLOC_FREE(lock_ctx->ttimer);
816 ctdb_kill(ctdb, lock_ctx->child, SIGKILL);
817 lock_ctx->child = -1;
818 talloc_set_destructor(lock_ctx, NULL);
819 close(lock_ctx->fd[0]);
820 return;
822 tevent_fd_set_auto_close(lock_ctx->tfd);
824 /* Move the context from pending to current */
825 if (lock_ctx->type == LOCK_RECORD) {
826 DLIST_REMOVE(lock_ctx->ctdb_db->lock_pending, lock_ctx);
827 DLIST_ADD_END(lock_ctx->ctdb_db->lock_current, lock_ctx, NULL);
828 } else {
829 DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
830 DLIST_ADD_END(ctdb->lock_current, lock_ctx, NULL);
832 CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_pending);
833 CTDB_INCREMENT_STAT(lock_ctx->ctdb, locks.num_current);
834 if (lock_ctx->ctdb_db) {
835 lock_ctx->ctdb_db->lock_num_current++;
836 CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_pending);
837 CTDB_INCREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
843 * Lock record / db depending on type
845 static struct lock_request *ctdb_lock_internal(TALLOC_CTX *mem_ctx,
846 struct ctdb_context *ctdb,
847 struct ctdb_db_context *ctdb_db,
848 TDB_DATA key,
849 uint32_t priority,
850 void (*callback)(void *, bool),
851 void *private_data,
852 enum lock_type type,
853 bool auto_mark)
855 struct lock_context *lock_ctx = NULL;
856 struct lock_request *request;
858 if (callback == NULL) {
859 DEBUG(DEBUG_WARNING, ("No callback function specified, not locking\n"));
860 return NULL;
863 lock_ctx = talloc_zero(ctdb, struct lock_context);
864 if (lock_ctx == NULL) {
865 DEBUG(DEBUG_ERR, ("Failed to create a new lock context\n"));
866 return NULL;
869 if ((request = talloc_zero(mem_ctx, struct lock_request)) == NULL) {
870 talloc_free(lock_ctx);
871 return NULL;
874 lock_ctx->type = type;
875 lock_ctx->ctdb = ctdb;
876 lock_ctx->ctdb_db = ctdb_db;
877 lock_ctx->key.dsize = key.dsize;
878 if (key.dsize > 0) {
879 lock_ctx->key.dptr = talloc_memdup(lock_ctx, key.dptr, key.dsize);
880 if (lock_ctx->key.dptr == NULL) {
881 DEBUG(DEBUG_ERR, (__location__ "Memory allocation error\n"));
882 talloc_free(lock_ctx);
883 return NULL;
885 lock_ctx->key_hash = ctdb_hash(&key);
886 } else {
887 lock_ctx->key.dptr = NULL;
889 lock_ctx->priority = priority;
890 lock_ctx->auto_mark = auto_mark;
892 lock_ctx->request = request;
893 lock_ctx->child = -1;
895 /* Non-record locks are required by recovery and should be scheduled
896 * immediately, so keep them at the head of the pending queue.
898 if (lock_ctx->type == LOCK_RECORD) {
899 DLIST_ADD_END(ctdb_db->lock_pending, lock_ctx, NULL);
900 } else {
901 DLIST_ADD_END(ctdb->lock_pending, lock_ctx, NULL);
903 CTDB_INCREMENT_STAT(ctdb, locks.num_pending);
904 if (ctdb_db) {
905 CTDB_INCREMENT_DB_STAT(ctdb_db, locks.num_pending);
908 /* Start the timer when we activate the context */
909 lock_ctx->start_time = timeval_current();
911 request->lctx = lock_ctx;
912 request->callback = callback;
913 request->private_data = private_data;
915 talloc_set_destructor(request, ctdb_lock_request_destructor);
917 ctdb_lock_schedule(ctdb);
919 return request;
924 * obtain a lock on a record in a database
926 struct lock_request *ctdb_lock_record(TALLOC_CTX *mem_ctx,
927 struct ctdb_db_context *ctdb_db,
928 TDB_DATA key,
929 bool auto_mark,
930 void (*callback)(void *, bool),
931 void *private_data)
933 return ctdb_lock_internal(mem_ctx,
934 ctdb_db->ctdb,
935 ctdb_db,
936 key,
938 callback,
939 private_data,
940 LOCK_RECORD,
941 auto_mark);
946 * obtain a lock on a database
948 struct lock_request *ctdb_lock_db(TALLOC_CTX *mem_ctx,
949 struct ctdb_db_context *ctdb_db,
950 bool auto_mark,
951 void (*callback)(void *, bool),
952 void *private_data)
954 return ctdb_lock_internal(mem_ctx,
955 ctdb_db->ctdb,
956 ctdb_db,
957 tdb_null,
959 callback,
960 private_data,
961 LOCK_DB,
962 auto_mark);
967 * obtain locks on all databases of specified priority
969 struct lock_request *ctdb_lock_alldb_prio(TALLOC_CTX *mem_ctx,
970 struct ctdb_context *ctdb,
971 uint32_t priority,
972 bool auto_mark,
973 void (*callback)(void *, bool),
974 void *private_data)
976 if (priority < 1 || priority > NUM_DB_PRIORITIES) {
977 DEBUG(DEBUG_ERR, ("Invalid db priority: %u\n", priority));
978 return NULL;
981 return ctdb_lock_internal(mem_ctx,
982 ctdb,
983 NULL,
984 tdb_null,
985 priority,
986 callback,
987 private_data,
988 LOCK_ALLDB_PRIO,
989 auto_mark);
994 * obtain locks on all databases
996 struct lock_request *ctdb_lock_alldb(TALLOC_CTX *mem_ctx,
997 struct ctdb_context *ctdb,
998 bool auto_mark,
999 void (*callback)(void *, bool),
1000 void *private_data)
1002 return ctdb_lock_internal(mem_ctx,
1003 ctdb,
1004 NULL,
1005 tdb_null,
1007 callback,
1008 private_data,
1009 LOCK_ALLDB,
1010 auto_mark);