1 // SPDX-License-Identifier: GPL-2.0-only
2 /******************************************************************************
3 *******************************************************************************
5 ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
6 ** Copyright (C) 2004-2011 Red Hat, Inc. All rights reserved.
9 *******************************************************************************
10 ******************************************************************************/
12 #include <linux/module.h>
14 #include "dlm_internal.h"
15 #include "lockspace.h"
24 #include "requestqueue.h"
29 static struct mutex ls_lock;
30 static struct list_head lslist;
31 static spinlock_t lslist_lock;
32 static struct task_struct * scand_task;
35 static ssize_t dlm_control_store(struct dlm_ls *ls, const char *buf, size_t len)
39 int rc = kstrtoint(buf, 0, &n);
43 ls = dlm_find_lockspace_local(ls->ls_local_handle);
57 dlm_put_lockspace(ls);
61 static ssize_t dlm_event_store(struct dlm_ls *ls, const char *buf, size_t len)
63 int rc = kstrtoint(buf, 0, &ls->ls_uevent_result);
67 set_bit(LSFL_UEVENT_WAIT, &ls->ls_flags);
68 wake_up(&ls->ls_uevent_wait);
72 static ssize_t dlm_id_show(struct dlm_ls *ls, char *buf)
74 return snprintf(buf, PAGE_SIZE, "%u\n", ls->ls_global_id);
77 static ssize_t dlm_id_store(struct dlm_ls *ls, const char *buf, size_t len)
79 int rc = kstrtouint(buf, 0, &ls->ls_global_id);
86 static ssize_t dlm_nodir_show(struct dlm_ls *ls, char *buf)
88 return snprintf(buf, PAGE_SIZE, "%u\n", dlm_no_directory(ls));
91 static ssize_t dlm_nodir_store(struct dlm_ls *ls, const char *buf, size_t len)
94 int rc = kstrtoint(buf, 0, &val);
99 set_bit(LSFL_NODIR, &ls->ls_flags);
103 static ssize_t dlm_recover_status_show(struct dlm_ls *ls, char *buf)
105 uint32_t status = dlm_recover_status(ls);
106 return snprintf(buf, PAGE_SIZE, "%x\n", status);
109 static ssize_t dlm_recover_nodeid_show(struct dlm_ls *ls, char *buf)
111 return snprintf(buf, PAGE_SIZE, "%d\n", ls->ls_recover_nodeid);
115 struct attribute attr;
116 ssize_t (*show)(struct dlm_ls *, char *);
117 ssize_t (*store)(struct dlm_ls *, const char *, size_t);
120 static struct dlm_attr dlm_attr_control = {
121 .attr = {.name = "control", .mode = S_IWUSR},
122 .store = dlm_control_store
125 static struct dlm_attr dlm_attr_event = {
126 .attr = {.name = "event_done", .mode = S_IWUSR},
127 .store = dlm_event_store
130 static struct dlm_attr dlm_attr_id = {
131 .attr = {.name = "id", .mode = S_IRUGO | S_IWUSR},
133 .store = dlm_id_store
136 static struct dlm_attr dlm_attr_nodir = {
137 .attr = {.name = "nodir", .mode = S_IRUGO | S_IWUSR},
138 .show = dlm_nodir_show,
139 .store = dlm_nodir_store
142 static struct dlm_attr dlm_attr_recover_status = {
143 .attr = {.name = "recover_status", .mode = S_IRUGO},
144 .show = dlm_recover_status_show
147 static struct dlm_attr dlm_attr_recover_nodeid = {
148 .attr = {.name = "recover_nodeid", .mode = S_IRUGO},
149 .show = dlm_recover_nodeid_show
152 static struct attribute *dlm_attrs[] = {
153 &dlm_attr_control.attr,
154 &dlm_attr_event.attr,
156 &dlm_attr_nodir.attr,
157 &dlm_attr_recover_status.attr,
158 &dlm_attr_recover_nodeid.attr,
161 ATTRIBUTE_GROUPS(dlm);
163 static ssize_t dlm_attr_show(struct kobject *kobj, struct attribute *attr,
166 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
167 struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
168 return a->show ? a->show(ls, buf) : 0;
171 static ssize_t dlm_attr_store(struct kobject *kobj, struct attribute *attr,
172 const char *buf, size_t len)
174 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
175 struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
176 return a->store ? a->store(ls, buf, len) : len;
179 static void lockspace_kobj_release(struct kobject *k)
181 struct dlm_ls *ls = container_of(k, struct dlm_ls, ls_kobj);
185 static const struct sysfs_ops dlm_attr_ops = {
186 .show = dlm_attr_show,
187 .store = dlm_attr_store,
190 static struct kobj_type dlm_ktype = {
191 .default_groups = dlm_groups,
192 .sysfs_ops = &dlm_attr_ops,
193 .release = lockspace_kobj_release,
196 static struct kset *dlm_kset;
198 static int do_uevent(struct dlm_ls *ls, int in)
201 kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE);
203 kobject_uevent(&ls->ls_kobj, KOBJ_OFFLINE);
205 log_rinfo(ls, "%s the lockspace group...", in ? "joining" : "leaving");
207 /* dlm_controld will see the uevent, do the necessary group management
208 and then write to sysfs to wake us */
210 wait_event(ls->ls_uevent_wait,
211 test_and_clear_bit(LSFL_UEVENT_WAIT, &ls->ls_flags));
213 log_rinfo(ls, "group event done %d", ls->ls_uevent_result);
215 return ls->ls_uevent_result;
218 static int dlm_uevent(const struct kobject *kobj, struct kobj_uevent_env *env)
220 const struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
222 add_uevent_var(env, "LOCKSPACE=%s", ls->ls_name);
226 static const struct kset_uevent_ops dlm_uevent_ops = {
227 .uevent = dlm_uevent,
230 int __init dlm_lockspace_init(void)
233 mutex_init(&ls_lock);
234 INIT_LIST_HEAD(&lslist);
235 spin_lock_init(&lslist_lock);
237 dlm_kset = kset_create_and_add("dlm", &dlm_uevent_ops, kernel_kobj);
239 printk(KERN_WARNING "%s: can not create kset\n", __func__);
245 void dlm_lockspace_exit(void)
247 kset_unregister(dlm_kset);
250 static struct dlm_ls *find_ls_to_scan(void)
254 spin_lock(&lslist_lock);
255 list_for_each_entry(ls, &lslist, ls_list) {
256 if (time_after_eq(jiffies, ls->ls_scan_time +
257 dlm_config.ci_scan_secs * HZ)) {
258 spin_unlock(&lslist_lock);
262 spin_unlock(&lslist_lock);
266 static int dlm_scand(void *data)
270 while (!kthread_should_stop()) {
271 ls = find_ls_to_scan();
273 if (dlm_lock_recovery_try(ls)) {
274 ls->ls_scan_time = jiffies;
276 dlm_unlock_recovery(ls);
278 ls->ls_scan_time += HZ;
282 schedule_timeout_interruptible(dlm_config.ci_scan_secs * HZ);
287 static int dlm_scand_start(void)
289 struct task_struct *p;
292 p = kthread_run(dlm_scand, NULL, "dlm_scand");
300 static void dlm_scand_stop(void)
302 kthread_stop(scand_task);
305 struct dlm_ls *dlm_find_lockspace_global(uint32_t id)
309 spin_lock(&lslist_lock);
311 list_for_each_entry(ls, &lslist, ls_list) {
312 if (ls->ls_global_id == id) {
313 atomic_inc(&ls->ls_count);
319 spin_unlock(&lslist_lock);
323 struct dlm_ls *dlm_find_lockspace_local(dlm_lockspace_t *lockspace)
327 spin_lock(&lslist_lock);
328 list_for_each_entry(ls, &lslist, ls_list) {
329 if (ls->ls_local_handle == lockspace) {
330 atomic_inc(&ls->ls_count);
336 spin_unlock(&lslist_lock);
340 struct dlm_ls *dlm_find_lockspace_device(int minor)
344 spin_lock(&lslist_lock);
345 list_for_each_entry(ls, &lslist, ls_list) {
346 if (ls->ls_device.minor == minor) {
347 atomic_inc(&ls->ls_count);
353 spin_unlock(&lslist_lock);
357 void dlm_put_lockspace(struct dlm_ls *ls)
359 if (atomic_dec_and_test(&ls->ls_count))
360 wake_up(&ls->ls_count_wait);
363 static void remove_lockspace(struct dlm_ls *ls)
366 wait_event(ls->ls_count_wait, atomic_read(&ls->ls_count) == 0);
368 spin_lock(&lslist_lock);
369 if (atomic_read(&ls->ls_count) != 0) {
370 spin_unlock(&lslist_lock);
374 WARN_ON(ls->ls_create_count != 0);
375 list_del(&ls->ls_list);
376 spin_unlock(&lslist_lock);
379 static int threads_start(void)
383 /* Thread for sending/receiving messages for all lockspace's */
384 error = dlm_midcomms_start();
386 log_print("cannot start dlm midcomms %d", error);
390 error = dlm_scand_start();
392 log_print("cannot start dlm_scand thread %d", error);
404 static int new_lockspace(const char *name, const char *cluster,
405 uint32_t flags, int lvblen,
406 const struct dlm_lockspace_ops *ops, void *ops_arg,
407 int *ops_result, dlm_lockspace_t **lockspace)
412 int namelen = strlen(name);
414 if (namelen > DLM_LOCKSPACE_LEN || namelen == 0)
420 if (!try_module_get(THIS_MODULE))
423 if (!dlm_user_daemon_available()) {
424 log_print("dlm user daemon not available");
429 if (ops && ops_result) {
430 if (!dlm_config.ci_recover_callbacks)
431 *ops_result = -EOPNOTSUPP;
437 log_print("dlm cluster name '%s' is being used without an application provided cluster name",
438 dlm_config.ci_cluster_name);
440 if (dlm_config.ci_recover_callbacks && cluster &&
441 strncmp(cluster, dlm_config.ci_cluster_name, DLM_LOCKSPACE_LEN)) {
442 log_print("dlm cluster name '%s' does not match "
443 "the application cluster name '%s'",
444 dlm_config.ci_cluster_name, cluster);
451 spin_lock(&lslist_lock);
452 list_for_each_entry(ls, &lslist, ls_list) {
453 WARN_ON(ls->ls_create_count <= 0);
454 if (ls->ls_namelen != namelen)
456 if (memcmp(ls->ls_name, name, namelen))
458 if (flags & DLM_LSFL_NEWEXCL) {
462 ls->ls_create_count++;
467 spin_unlock(&lslist_lock);
474 ls = kzalloc(sizeof(*ls), GFP_NOFS);
477 memcpy(ls->ls_name, name, namelen);
478 ls->ls_namelen = namelen;
479 ls->ls_lvblen = lvblen;
480 atomic_set(&ls->ls_count, 0);
481 init_waitqueue_head(&ls->ls_count_wait);
483 ls->ls_scan_time = jiffies;
485 if (ops && dlm_config.ci_recover_callbacks) {
487 ls->ls_ops_arg = ops_arg;
490 /* ls_exflags are forced to match among nodes, and we don't
491 * need to require all nodes to have some flags set
493 ls->ls_exflags = (flags & ~(DLM_LSFL_FS | DLM_LSFL_NEWEXCL));
495 size = READ_ONCE(dlm_config.ci_rsbtbl_size);
496 ls->ls_rsbtbl_size = size;
498 ls->ls_rsbtbl = vmalloc(array_size(size, sizeof(struct dlm_rsbtable)));
501 for (i = 0; i < size; i++) {
502 ls->ls_rsbtbl[i].keep.rb_node = NULL;
503 ls->ls_rsbtbl[i].toss.rb_node = NULL;
504 spin_lock_init(&ls->ls_rsbtbl[i].lock);
507 for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++) {
508 ls->ls_remove_names[i] = kzalloc(DLM_RESNAME_MAXLEN+1,
510 if (!ls->ls_remove_names[i])
514 idr_init(&ls->ls_lkbidr);
515 spin_lock_init(&ls->ls_lkbidr_spin);
517 INIT_LIST_HEAD(&ls->ls_waiters);
518 mutex_init(&ls->ls_waiters_mutex);
519 INIT_LIST_HEAD(&ls->ls_orphans);
520 mutex_init(&ls->ls_orphans_mutex);
522 INIT_LIST_HEAD(&ls->ls_new_rsb);
523 spin_lock_init(&ls->ls_new_rsb_spin);
525 INIT_LIST_HEAD(&ls->ls_nodes);
526 INIT_LIST_HEAD(&ls->ls_nodes_gone);
527 ls->ls_num_nodes = 0;
528 ls->ls_low_nodeid = 0;
529 ls->ls_total_weight = 0;
530 ls->ls_node_array = NULL;
532 memset(&ls->ls_local_rsb, 0, sizeof(struct dlm_rsb));
533 ls->ls_local_rsb.res_ls = ls;
535 ls->ls_debug_rsb_dentry = NULL;
536 ls->ls_debug_waiters_dentry = NULL;
538 init_waitqueue_head(&ls->ls_uevent_wait);
539 ls->ls_uevent_result = 0;
540 init_completion(&ls->ls_recovery_done);
541 ls->ls_recovery_result = -1;
543 spin_lock_init(&ls->ls_cb_lock);
544 INIT_LIST_HEAD(&ls->ls_cb_delay);
546 ls->ls_recoverd_task = NULL;
547 mutex_init(&ls->ls_recoverd_active);
548 spin_lock_init(&ls->ls_recover_lock);
549 spin_lock_init(&ls->ls_rcom_spin);
550 get_random_bytes(&ls->ls_rcom_seq, sizeof(uint64_t));
551 ls->ls_recover_status = 0;
552 ls->ls_recover_seq = get_random_u64();
553 ls->ls_recover_args = NULL;
554 init_rwsem(&ls->ls_in_recovery);
555 init_rwsem(&ls->ls_recv_active);
556 INIT_LIST_HEAD(&ls->ls_requestqueue);
557 atomic_set(&ls->ls_requestqueue_cnt, 0);
558 init_waitqueue_head(&ls->ls_requestqueue_wait);
559 mutex_init(&ls->ls_requestqueue_mutex);
560 spin_lock_init(&ls->ls_clear_proc_locks);
562 /* Due backwards compatibility with 3.1 we need to use maximum
563 * possible dlm message size to be sure the message will fit and
564 * not having out of bounds issues. However on sending side 3.2
567 ls->ls_recover_buf = kmalloc(DLM_MAX_SOCKET_BUFSIZE, GFP_NOFS);
568 if (!ls->ls_recover_buf)
572 ls->ls_num_slots = 0;
573 ls->ls_slots_size = 0;
576 INIT_LIST_HEAD(&ls->ls_recover_list);
577 spin_lock_init(&ls->ls_recover_list_lock);
578 idr_init(&ls->ls_recover_idr);
579 spin_lock_init(&ls->ls_recover_idr_lock);
580 ls->ls_recover_list_count = 0;
581 ls->ls_local_handle = ls;
582 init_waitqueue_head(&ls->ls_wait_general);
583 INIT_LIST_HEAD(&ls->ls_root_list);
584 init_rwsem(&ls->ls_root_sem);
586 spin_lock(&lslist_lock);
587 ls->ls_create_count = 1;
588 list_add(&ls->ls_list, &lslist);
589 spin_unlock(&lslist_lock);
591 if (flags & DLM_LSFL_FS) {
592 error = dlm_callback_start(ls);
594 log_error(ls, "can't start dlm_callback %d", error);
599 init_waitqueue_head(&ls->ls_recover_lock_wait);
602 * Once started, dlm_recoverd first looks for ls in lslist, then
603 * initializes ls_in_recovery as locked in "down" mode. We need
604 * to wait for the wakeup from dlm_recoverd because in_recovery
605 * has to start out in down mode.
608 error = dlm_recoverd_start(ls);
610 log_error(ls, "can't start dlm_recoverd %d", error);
614 wait_event(ls->ls_recover_lock_wait,
615 test_bit(LSFL_RECOVER_LOCK, &ls->ls_flags));
617 /* let kobject handle freeing of ls if there's an error */
620 ls->ls_kobj.kset = dlm_kset;
621 error = kobject_init_and_add(&ls->ls_kobj, &dlm_ktype, NULL,
625 kobject_uevent(&ls->ls_kobj, KOBJ_ADD);
627 /* This uevent triggers dlm_controld in userspace to add us to the
628 group of nodes that are members of this lockspace (managed by the
629 cluster infrastructure.) Once it's done that, it tells us who the
630 current lockspace members are (via configfs) and then tells the
631 lockspace to start running (via sysfs) in dlm_ls_start(). */
633 error = do_uevent(ls, 1);
637 /* wait until recovery is successful or failed */
638 wait_for_completion(&ls->ls_recovery_done);
639 error = ls->ls_recovery_result;
643 dlm_create_debug_file(ls);
645 log_rinfo(ls, "join complete");
651 dlm_clear_members(ls);
652 kfree(ls->ls_node_array);
654 dlm_recoverd_stop(ls);
656 dlm_callback_stop(ls);
658 spin_lock(&lslist_lock);
659 list_del(&ls->ls_list);
660 spin_unlock(&lslist_lock);
661 idr_destroy(&ls->ls_recover_idr);
662 kfree(ls->ls_recover_buf);
664 idr_destroy(&ls->ls_lkbidr);
666 for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++)
667 kfree(ls->ls_remove_names[i]);
668 vfree(ls->ls_rsbtbl);
671 kobject_put(&ls->ls_kobj);
675 module_put(THIS_MODULE);
679 static int __dlm_new_lockspace(const char *name, const char *cluster,
680 uint32_t flags, int lvblen,
681 const struct dlm_lockspace_ops *ops,
682 void *ops_arg, int *ops_result,
683 dlm_lockspace_t **lockspace)
687 mutex_lock(&ls_lock);
689 error = threads_start();
693 error = new_lockspace(name, cluster, flags, lvblen, ops, ops_arg,
694 ops_result, lockspace);
701 dlm_midcomms_shutdown();
705 mutex_unlock(&ls_lock);
709 int dlm_new_lockspace(const char *name, const char *cluster, uint32_t flags,
710 int lvblen, const struct dlm_lockspace_ops *ops,
711 void *ops_arg, int *ops_result,
712 dlm_lockspace_t **lockspace)
714 return __dlm_new_lockspace(name, cluster, flags | DLM_LSFL_FS, lvblen,
715 ops, ops_arg, ops_result, lockspace);
718 int dlm_new_user_lockspace(const char *name, const char *cluster,
719 uint32_t flags, int lvblen,
720 const struct dlm_lockspace_ops *ops,
721 void *ops_arg, int *ops_result,
722 dlm_lockspace_t **lockspace)
724 return __dlm_new_lockspace(name, cluster, flags, lvblen, ops,
725 ops_arg, ops_result, lockspace);
728 static int lkb_idr_is_local(int id, void *p, void *data)
730 struct dlm_lkb *lkb = p;
732 return lkb->lkb_nodeid == 0 && lkb->lkb_grmode != DLM_LOCK_IV;
735 static int lkb_idr_is_any(int id, void *p, void *data)
740 static int lkb_idr_free(int id, void *p, void *data)
742 struct dlm_lkb *lkb = p;
744 if (lkb->lkb_lvbptr && test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags))
745 dlm_free_lvb(lkb->lkb_lvbptr);
751 /* NOTE: We check the lkbidr here rather than the resource table.
752 This is because there may be LKBs queued as ASTs that have been unlinked
753 from their RSBs and are pending deletion once the AST has been delivered */
755 static int lockspace_busy(struct dlm_ls *ls, int force)
759 spin_lock(&ls->ls_lkbidr_spin);
761 rv = idr_for_each(&ls->ls_lkbidr, lkb_idr_is_any, ls);
762 } else if (force == 1) {
763 rv = idr_for_each(&ls->ls_lkbidr, lkb_idr_is_local, ls);
767 spin_unlock(&ls->ls_lkbidr_spin);
771 static int release_lockspace(struct dlm_ls *ls, int force)
777 busy = lockspace_busy(ls, force);
779 spin_lock(&lslist_lock);
780 if (ls->ls_create_count == 1) {
784 /* remove_lockspace takes ls off lslist */
785 ls->ls_create_count = 0;
788 } else if (ls->ls_create_count > 1) {
789 rv = --ls->ls_create_count;
793 spin_unlock(&lslist_lock);
796 log_debug(ls, "release_lockspace no remove %d", rv);
801 dlm_midcomms_version_wait();
803 dlm_device_deregister(ls);
805 if (force < 3 && dlm_user_daemon_available())
808 dlm_recoverd_stop(ls);
812 dlm_clear_members(ls);
813 dlm_midcomms_shutdown();
816 dlm_callback_stop(ls);
818 remove_lockspace(ls);
820 dlm_delete_debug_file(ls);
822 idr_destroy(&ls->ls_recover_idr);
823 kfree(ls->ls_recover_buf);
826 * Free all lkb's in idr
829 idr_for_each(&ls->ls_lkbidr, lkb_idr_free, ls);
830 idr_destroy(&ls->ls_lkbidr);
833 * Free all rsb's on rsbtbl[] lists
836 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
837 while ((n = rb_first(&ls->ls_rsbtbl[i].keep))) {
838 rsb = rb_entry(n, struct dlm_rsb, res_hashnode);
839 rb_erase(n, &ls->ls_rsbtbl[i].keep);
843 while ((n = rb_first(&ls->ls_rsbtbl[i].toss))) {
844 rsb = rb_entry(n, struct dlm_rsb, res_hashnode);
845 rb_erase(n, &ls->ls_rsbtbl[i].toss);
850 vfree(ls->ls_rsbtbl);
852 for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++)
853 kfree(ls->ls_remove_names[i]);
855 while (!list_empty(&ls->ls_new_rsb)) {
856 rsb = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb,
858 list_del(&rsb->res_hashchain);
863 * Free structures on any other lists
866 dlm_purge_requestqueue(ls);
867 kfree(ls->ls_recover_args);
868 dlm_clear_members(ls);
869 dlm_clear_members_gone(ls);
870 kfree(ls->ls_node_array);
871 log_rinfo(ls, "release_lockspace final free");
872 kobject_put(&ls->ls_kobj);
873 /* The ls structure will be freed when the kobject is done with */
875 module_put(THIS_MODULE);
880 * Called when a system has released all its locks and is not going to use the
881 * lockspace any longer. We free everything we're managing for this lockspace.
882 * Remaining nodes will go through the recovery process as if we'd died. The
883 * lockspace must continue to function as usual, participating in recoveries,
884 * until this returns.
886 * Force has 4 possible values:
887 * 0 - don't destroy lockspace if it has any LKBs
888 * 1 - destroy lockspace if it has remote LKBs but not if it has local LKBs
889 * 2 - destroy lockspace regardless of LKBs
890 * 3 - destroy lockspace as part of a forced shutdown
893 int dlm_release_lockspace(void *lockspace, int force)
898 ls = dlm_find_lockspace_local(lockspace);
901 dlm_put_lockspace(ls);
903 mutex_lock(&ls_lock);
904 error = release_lockspace(ls, force);
909 mutex_unlock(&ls_lock);
914 void dlm_stop_lockspaces(void)
921 spin_lock(&lslist_lock);
922 list_for_each_entry(ls, &lslist, ls_list) {
923 if (!test_bit(LSFL_RUNNING, &ls->ls_flags)) {
927 spin_unlock(&lslist_lock);
928 log_error(ls, "no userland control daemon, stopping lockspace");
932 spin_unlock(&lslist_lock);
935 log_print("dlm user daemon left %d lockspaces", count);