1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
5 ** Copyright (C) 2004-2007 Red Hat, Inc. All rights reserved.
7 ** This copyrighted material is made available to anyone wishing to use,
8 ** modify, copy, or redistribute it subject to the terms and conditions
9 ** of the GNU General Public License v.2.
11 *******************************************************************************
12 ******************************************************************************/
14 #include "dlm_internal.h"
15 #include "lockspace.h"
25 #include "requestqueue.h"
27 #ifdef CONFIG_DLM_DEBUG
28 int dlm_create_debug_file(struct dlm_ls *ls);
29 void dlm_delete_debug_file(struct dlm_ls *ls);
31 static inline int dlm_create_debug_file(struct dlm_ls *ls) { return 0; }
32 static inline void dlm_delete_debug_file(struct dlm_ls *ls) { }
36 static struct mutex ls_lock;
37 static struct list_head lslist;
38 static spinlock_t lslist_lock;
39 static struct task_struct * scand_task;
42 static ssize_t dlm_control_store(struct dlm_ls *ls, const char *buf, size_t len)
45 int n = simple_strtol(buf, NULL, 0);
47 ls = dlm_find_lockspace_local(ls->ls_local_handle);
61 dlm_put_lockspace(ls);
65 static ssize_t dlm_event_store(struct dlm_ls *ls, const char *buf, size_t len)
67 ls->ls_uevent_result = simple_strtol(buf, NULL, 0);
68 set_bit(LSFL_UEVENT_WAIT, &ls->ls_flags);
69 wake_up(&ls->ls_uevent_wait);
73 static ssize_t dlm_id_show(struct dlm_ls *ls, char *buf)
75 return snprintf(buf, PAGE_SIZE, "%u\n", ls->ls_global_id);
78 static ssize_t dlm_id_store(struct dlm_ls *ls, const char *buf, size_t len)
80 ls->ls_global_id = simple_strtoul(buf, NULL, 0);
84 static ssize_t dlm_recover_status_show(struct dlm_ls *ls, char *buf)
86 uint32_t status = dlm_recover_status(ls);
87 return snprintf(buf, PAGE_SIZE, "%x\n", status);
90 static ssize_t dlm_recover_nodeid_show(struct dlm_ls *ls, char *buf)
92 return snprintf(buf, PAGE_SIZE, "%d\n", ls->ls_recover_nodeid);
96 struct attribute attr;
97 ssize_t (*show)(struct dlm_ls *, char *);
98 ssize_t (*store)(struct dlm_ls *, const char *, size_t);
101 static struct dlm_attr dlm_attr_control = {
102 .attr = {.name = "control", .mode = S_IWUSR},
103 .store = dlm_control_store
106 static struct dlm_attr dlm_attr_event = {
107 .attr = {.name = "event_done", .mode = S_IWUSR},
108 .store = dlm_event_store
111 static struct dlm_attr dlm_attr_id = {
112 .attr = {.name = "id", .mode = S_IRUGO | S_IWUSR},
114 .store = dlm_id_store
117 static struct dlm_attr dlm_attr_recover_status = {
118 .attr = {.name = "recover_status", .mode = S_IRUGO},
119 .show = dlm_recover_status_show
122 static struct dlm_attr dlm_attr_recover_nodeid = {
123 .attr = {.name = "recover_nodeid", .mode = S_IRUGO},
124 .show = dlm_recover_nodeid_show
127 static struct attribute *dlm_attrs[] = {
128 &dlm_attr_control.attr,
129 &dlm_attr_event.attr,
131 &dlm_attr_recover_status.attr,
132 &dlm_attr_recover_nodeid.attr,
136 static ssize_t dlm_attr_show(struct kobject *kobj, struct attribute *attr,
139 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
140 struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
141 return a->show ? a->show(ls, buf) : 0;
144 static ssize_t dlm_attr_store(struct kobject *kobj, struct attribute *attr,
145 const char *buf, size_t len)
147 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
148 struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
149 return a->store ? a->store(ls, buf, len) : len;
152 static void lockspace_kobj_release(struct kobject *k)
154 struct dlm_ls *ls = container_of(k, struct dlm_ls, ls_kobj);
158 static struct sysfs_ops dlm_attr_ops = {
159 .show = dlm_attr_show,
160 .store = dlm_attr_store,
163 static struct kobj_type dlm_ktype = {
164 .default_attrs = dlm_attrs,
165 .sysfs_ops = &dlm_attr_ops,
166 .release = lockspace_kobj_release,
169 static struct kset dlm_kset = {
170 .kobj = {.name = "dlm",},
174 static int kobject_setup(struct dlm_ls *ls)
176 char lsname[DLM_LOCKSPACE_LEN];
179 memset(lsname, 0, DLM_LOCKSPACE_LEN);
180 snprintf(lsname, DLM_LOCKSPACE_LEN, "%s", ls->ls_name);
182 error = kobject_set_name(&ls->ls_kobj, "%s", lsname);
186 ls->ls_kobj.kset = &dlm_kset;
187 ls->ls_kobj.ktype = &dlm_ktype;
191 static int do_uevent(struct dlm_ls *ls, int in)
196 kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE);
198 kobject_uevent(&ls->ls_kobj, KOBJ_OFFLINE);
200 log_debug(ls, "%s the lockspace group...", in ? "joining" : "leaving");
202 /* dlm_controld will see the uevent, do the necessary group management
203 and then write to sysfs to wake us */
205 error = wait_event_interruptible(ls->ls_uevent_wait,
206 test_and_clear_bit(LSFL_UEVENT_WAIT, &ls->ls_flags));
208 log_debug(ls, "group event done %d %d", error, ls->ls_uevent_result);
213 error = ls->ls_uevent_result;
216 log_error(ls, "group %s failed %d %d", in ? "join" : "leave",
217 error, ls->ls_uevent_result);
222 int dlm_lockspace_init(void)
227 mutex_init(&ls_lock);
228 INIT_LIST_HEAD(&lslist);
229 spin_lock_init(&lslist_lock);
231 kobj_set_kset_s(&dlm_kset, kernel_subsys);
232 error = kset_register(&dlm_kset);
234 printk("dlm_lockspace_init: cannot register kset %d\n", error);
238 void dlm_lockspace_exit(void)
240 kset_unregister(&dlm_kset);
243 static int dlm_scand(void *data)
247 while (!kthread_should_stop()) {
248 list_for_each_entry(ls, &lslist, ls_list) {
249 if (dlm_lock_recovery_try(ls)) {
251 dlm_scan_timeout(ls);
252 dlm_unlock_recovery(ls);
255 schedule_timeout_interruptible(dlm_config.ci_scan_secs * HZ);
260 static int dlm_scand_start(void)
262 struct task_struct *p;
265 p = kthread_run(dlm_scand, NULL, "dlm_scand");
273 static void dlm_scand_stop(void)
275 kthread_stop(scand_task);
278 static struct dlm_ls *dlm_find_lockspace_name(char *name, int namelen)
282 spin_lock(&lslist_lock);
284 list_for_each_entry(ls, &lslist, ls_list) {
285 if (ls->ls_namelen == namelen &&
286 memcmp(ls->ls_name, name, namelen) == 0)
291 spin_unlock(&lslist_lock);
295 struct dlm_ls *dlm_find_lockspace_global(uint32_t id)
299 spin_lock(&lslist_lock);
301 list_for_each_entry(ls, &lslist, ls_list) {
302 if (ls->ls_global_id == id) {
309 spin_unlock(&lslist_lock);
313 struct dlm_ls *dlm_find_lockspace_local(dlm_lockspace_t *lockspace)
317 spin_lock(&lslist_lock);
318 list_for_each_entry(ls, &lslist, ls_list) {
319 if (ls->ls_local_handle == lockspace) {
326 spin_unlock(&lslist_lock);
330 struct dlm_ls *dlm_find_lockspace_device(int minor)
334 spin_lock(&lslist_lock);
335 list_for_each_entry(ls, &lslist, ls_list) {
336 if (ls->ls_device.minor == minor) {
343 spin_unlock(&lslist_lock);
347 void dlm_put_lockspace(struct dlm_ls *ls)
349 spin_lock(&lslist_lock);
351 spin_unlock(&lslist_lock);
354 static void remove_lockspace(struct dlm_ls *ls)
357 spin_lock(&lslist_lock);
358 if (ls->ls_count == 0) {
359 list_del(&ls->ls_list);
360 spin_unlock(&lslist_lock);
363 spin_unlock(&lslist_lock);
368 static int threads_start(void)
372 /* Thread which process lock requests for all lockspace's */
373 error = dlm_astd_start();
375 log_print("cannot start dlm_astd thread %d", error);
379 error = dlm_scand_start();
381 log_print("cannot start dlm_scand thread %d", error);
385 /* Thread for sending/receiving messages for all lockspace's */
386 error = dlm_lowcomms_start();
388 log_print("cannot start dlm lowcomms %d", error);
402 static void threads_stop(void)
409 static int new_lockspace(char *name, int namelen, void **lockspace,
410 uint32_t flags, int lvblen)
413 int i, size, error = -ENOMEM;
416 if (namelen > DLM_LOCKSPACE_LEN)
419 if (!lvblen || (lvblen % 8))
422 if (!try_module_get(THIS_MODULE))
425 ls = dlm_find_lockspace_name(name, namelen);
428 module_put(THIS_MODULE);
432 ls = kzalloc(sizeof(struct dlm_ls) + namelen, GFP_KERNEL);
435 memcpy(ls->ls_name, name, namelen);
436 ls->ls_namelen = namelen;
437 ls->ls_lvblen = lvblen;
441 /* ls_exflags are forced to match among nodes, and we don't
442 need to require all nodes to have TIMEWARN active */
443 if (flags & DLM_LSFL_TIMEWARN)
444 set_bit(LSFL_TIMEWARN, &ls->ls_flags);
445 ls->ls_exflags = (flags & ~DLM_LSFL_TIMEWARN);
447 size = dlm_config.ci_rsbtbl_size;
448 ls->ls_rsbtbl_size = size;
450 ls->ls_rsbtbl = kmalloc(sizeof(struct dlm_rsbtable) * size, GFP_KERNEL);
453 for (i = 0; i < size; i++) {
454 INIT_LIST_HEAD(&ls->ls_rsbtbl[i].list);
455 INIT_LIST_HEAD(&ls->ls_rsbtbl[i].toss);
456 rwlock_init(&ls->ls_rsbtbl[i].lock);
459 size = dlm_config.ci_lkbtbl_size;
460 ls->ls_lkbtbl_size = size;
462 ls->ls_lkbtbl = kmalloc(sizeof(struct dlm_lkbtable) * size, GFP_KERNEL);
465 for (i = 0; i < size; i++) {
466 INIT_LIST_HEAD(&ls->ls_lkbtbl[i].list);
467 rwlock_init(&ls->ls_lkbtbl[i].lock);
468 ls->ls_lkbtbl[i].counter = 1;
471 size = dlm_config.ci_dirtbl_size;
472 ls->ls_dirtbl_size = size;
474 ls->ls_dirtbl = kmalloc(sizeof(struct dlm_dirtable) * size, GFP_KERNEL);
477 for (i = 0; i < size; i++) {
478 INIT_LIST_HEAD(&ls->ls_dirtbl[i].list);
479 rwlock_init(&ls->ls_dirtbl[i].lock);
482 INIT_LIST_HEAD(&ls->ls_waiters);
483 mutex_init(&ls->ls_waiters_mutex);
484 INIT_LIST_HEAD(&ls->ls_orphans);
485 mutex_init(&ls->ls_orphans_mutex);
486 INIT_LIST_HEAD(&ls->ls_timeout);
487 mutex_init(&ls->ls_timeout_mutex);
489 INIT_LIST_HEAD(&ls->ls_nodes);
490 INIT_LIST_HEAD(&ls->ls_nodes_gone);
491 ls->ls_num_nodes = 0;
492 ls->ls_low_nodeid = 0;
493 ls->ls_total_weight = 0;
494 ls->ls_node_array = NULL;
496 memset(&ls->ls_stub_rsb, 0, sizeof(struct dlm_rsb));
497 ls->ls_stub_rsb.res_ls = ls;
499 ls->ls_debug_rsb_dentry = NULL;
500 ls->ls_debug_waiters_dentry = NULL;
502 init_waitqueue_head(&ls->ls_uevent_wait);
503 ls->ls_uevent_result = 0;
504 init_completion(&ls->ls_members_done);
505 ls->ls_members_result = -1;
507 ls->ls_recoverd_task = NULL;
508 mutex_init(&ls->ls_recoverd_active);
509 spin_lock_init(&ls->ls_recover_lock);
510 spin_lock_init(&ls->ls_rcom_spin);
511 get_random_bytes(&ls->ls_rcom_seq, sizeof(uint64_t));
512 ls->ls_recover_status = 0;
513 ls->ls_recover_seq = 0;
514 ls->ls_recover_args = NULL;
515 init_rwsem(&ls->ls_in_recovery);
516 INIT_LIST_HEAD(&ls->ls_requestqueue);
517 mutex_init(&ls->ls_requestqueue_mutex);
518 mutex_init(&ls->ls_clear_proc_locks);
520 ls->ls_recover_buf = kmalloc(dlm_config.ci_buffer_size, GFP_KERNEL);
521 if (!ls->ls_recover_buf)
524 INIT_LIST_HEAD(&ls->ls_recover_list);
525 spin_lock_init(&ls->ls_recover_list_lock);
526 ls->ls_recover_list_count = 0;
527 ls->ls_local_handle = ls;
528 init_waitqueue_head(&ls->ls_wait_general);
529 INIT_LIST_HEAD(&ls->ls_root_list);
530 init_rwsem(&ls->ls_root_sem);
532 down_write(&ls->ls_in_recovery);
534 spin_lock(&lslist_lock);
535 list_add(&ls->ls_list, &lslist);
536 spin_unlock(&lslist_lock);
538 /* needs to find ls in lslist */
539 error = dlm_recoverd_start(ls);
541 log_error(ls, "can't start dlm_recoverd %d", error);
545 error = kobject_setup(ls);
549 error = kobject_register(&ls->ls_kobj);
553 /* let kobject handle freeing of ls if there's an error */
556 /* This uevent triggers dlm_controld in userspace to add us to the
557 group of nodes that are members of this lockspace (managed by the
558 cluster infrastructure.) Once it's done that, it tells us who the
559 current lockspace members are (via configfs) and then tells the
560 lockspace to start running (via sysfs) in dlm_ls_start(). */
562 error = do_uevent(ls, 1);
566 wait_for_completion(&ls->ls_members_done);
567 error = ls->ls_members_result;
571 dlm_create_debug_file(ls);
573 log_debug(ls, "join complete");
580 dlm_clear_members(ls);
581 kfree(ls->ls_node_array);
583 dlm_recoverd_stop(ls);
585 spin_lock(&lslist_lock);
586 list_del(&ls->ls_list);
587 spin_unlock(&lslist_lock);
588 kfree(ls->ls_recover_buf);
590 kfree(ls->ls_dirtbl);
592 kfree(ls->ls_lkbtbl);
594 kfree(ls->ls_rsbtbl);
597 kobject_unregister(&ls->ls_kobj);
601 module_put(THIS_MODULE);
605 int dlm_new_lockspace(char *name, int namelen, void **lockspace,
606 uint32_t flags, int lvblen)
610 mutex_lock(&ls_lock);
612 error = threads_start();
616 error = new_lockspace(name, namelen, lockspace, flags, lvblen);
622 mutex_unlock(&ls_lock);
626 /* Return 1 if the lockspace still has active remote locks,
627 * 2 if the lockspace still has active local locks.
629 static int lockspace_busy(struct dlm_ls *ls)
631 int i, lkb_found = 0;
634 /* NOTE: We check the lockidtbl here rather than the resource table.
635 This is because there may be LKBs queued as ASTs that have been
636 unlinked from their RSBs and are pending deletion once the AST has
639 for (i = 0; i < ls->ls_lkbtbl_size; i++) {
640 read_lock(&ls->ls_lkbtbl[i].lock);
641 if (!list_empty(&ls->ls_lkbtbl[i].list)) {
643 list_for_each_entry(lkb, &ls->ls_lkbtbl[i].list,
645 if (!lkb->lkb_nodeid) {
646 read_unlock(&ls->ls_lkbtbl[i].lock);
651 read_unlock(&ls->ls_lkbtbl[i].lock);
656 static int release_lockspace(struct dlm_ls *ls, int force)
660 struct list_head *head;
662 int busy = lockspace_busy(ls);
670 dlm_recoverd_stop(ls);
672 remove_lockspace(ls);
674 dlm_delete_debug_file(ls);
678 kfree(ls->ls_recover_buf);
681 * Free direntry structs.
685 kfree(ls->ls_dirtbl);
688 * Free all lkb's on lkbtbl[] lists.
691 for (i = 0; i < ls->ls_lkbtbl_size; i++) {
692 head = &ls->ls_lkbtbl[i].list;
693 while (!list_empty(head)) {
694 lkb = list_entry(head->next, struct dlm_lkb,
697 list_del(&lkb->lkb_idtbl_list);
701 if (lkb->lkb_lvbptr && lkb->lkb_flags & DLM_IFL_MSTCPY)
702 free_lvb(lkb->lkb_lvbptr);
709 kfree(ls->ls_lkbtbl);
712 * Free all rsb's on rsbtbl[] lists
715 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
716 head = &ls->ls_rsbtbl[i].list;
717 while (!list_empty(head)) {
718 rsb = list_entry(head->next, struct dlm_rsb,
721 list_del(&rsb->res_hashchain);
725 head = &ls->ls_rsbtbl[i].toss;
726 while (!list_empty(head)) {
727 rsb = list_entry(head->next, struct dlm_rsb,
729 list_del(&rsb->res_hashchain);
734 kfree(ls->ls_rsbtbl);
737 * Free structures on any other lists
740 dlm_purge_requestqueue(ls);
741 kfree(ls->ls_recover_args);
742 dlm_clear_free_entries(ls);
743 dlm_clear_members(ls);
744 dlm_clear_members_gone(ls);
745 kfree(ls->ls_node_array);
746 kobject_unregister(&ls->ls_kobj);
747 /* The ls structure will be freed when the kobject is done with */
749 mutex_lock(&ls_lock);
753 mutex_unlock(&ls_lock);
755 module_put(THIS_MODULE);
760 * Called when a system has released all its locks and is not going to use the
761 * lockspace any longer. We free everything we're managing for this lockspace.
762 * Remaining nodes will go through the recovery process as if we'd died. The
763 * lockspace must continue to function as usual, participating in recoveries,
764 * until this returns.
766 * Force has 4 possible values:
767 * 0 - don't destroy locksapce if it has any LKBs
768 * 1 - destroy lockspace if it has remote LKBs but not if it has local LKBs
769 * 2 - destroy lockspace regardless of LKBs
770 * 3 - destroy lockspace as part of a forced shutdown
773 int dlm_release_lockspace(void *lockspace, int force)
777 ls = dlm_find_lockspace_local(lockspace);
780 dlm_put_lockspace(ls);
781 return release_lockspace(ls, force);