Loading...
1// SPDX-License-Identifier: GPL-2.0-only
2/******************************************************************************
3*******************************************************************************
4**
5** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
6** Copyright (C) 2004-2011 Red Hat, Inc. All rights reserved.
7**
8**
9*******************************************************************************
10******************************************************************************/
11
12#include <linux/module.h>
13
14#include "dlm_internal.h"
15#include "lockspace.h"
16#include "member.h"
17#include "recoverd.h"
18#include "dir.h"
19#include "midcomms.h"
20#include "lowcomms.h"
21#include "config.h"
22#include "memory.h"
23#include "lock.h"
24#include "recover.h"
25#include "requestqueue.h"
26#include "user.h"
27#include "ast.h"
28
29static int ls_count;
30static struct mutex ls_lock;
31static struct list_head lslist;
32static spinlock_t lslist_lock;
33static struct task_struct * scand_task;
34
35
36static ssize_t dlm_control_store(struct dlm_ls *ls, const char *buf, size_t len)
37{
38 ssize_t ret = len;
39 int n;
40 int rc = kstrtoint(buf, 0, &n);
41
42 if (rc)
43 return rc;
44 ls = dlm_find_lockspace_local(ls->ls_local_handle);
45 if (!ls)
46 return -EINVAL;
47
48 switch (n) {
49 case 0:
50 dlm_ls_stop(ls);
51 break;
52 case 1:
53 dlm_ls_start(ls);
54 break;
55 default:
56 ret = -EINVAL;
57 }
58 dlm_put_lockspace(ls);
59 return ret;
60}
61
62static ssize_t dlm_event_store(struct dlm_ls *ls, const char *buf, size_t len)
63{
64 int rc = kstrtoint(buf, 0, &ls->ls_uevent_result);
65
66 if (rc)
67 return rc;
68 set_bit(LSFL_UEVENT_WAIT, &ls->ls_flags);
69 wake_up(&ls->ls_uevent_wait);
70 return len;
71}
72
73static ssize_t dlm_id_show(struct dlm_ls *ls, char *buf)
74{
75 return snprintf(buf, PAGE_SIZE, "%u\n", ls->ls_global_id);
76}
77
78static ssize_t dlm_id_store(struct dlm_ls *ls, const char *buf, size_t len)
79{
80 int rc = kstrtouint(buf, 0, &ls->ls_global_id);
81
82 if (rc)
83 return rc;
84 return len;
85}
86
87static ssize_t dlm_nodir_show(struct dlm_ls *ls, char *buf)
88{
89 return snprintf(buf, PAGE_SIZE, "%u\n", dlm_no_directory(ls));
90}
91
92static ssize_t dlm_nodir_store(struct dlm_ls *ls, const char *buf, size_t len)
93{
94 int val;
95 int rc = kstrtoint(buf, 0, &val);
96
97 if (rc)
98 return rc;
99 if (val == 1)
100 set_bit(LSFL_NODIR, &ls->ls_flags);
101 return len;
102}
103
104static ssize_t dlm_recover_status_show(struct dlm_ls *ls, char *buf)
105{
106 uint32_t status = dlm_recover_status(ls);
107 return snprintf(buf, PAGE_SIZE, "%x\n", status);
108}
109
110static ssize_t dlm_recover_nodeid_show(struct dlm_ls *ls, char *buf)
111{
112 return snprintf(buf, PAGE_SIZE, "%d\n", ls->ls_recover_nodeid);
113}
114
115struct dlm_attr {
116 struct attribute attr;
117 ssize_t (*show)(struct dlm_ls *, char *);
118 ssize_t (*store)(struct dlm_ls *, const char *, size_t);
119};
120
121static struct dlm_attr dlm_attr_control = {
122 .attr = {.name = "control", .mode = S_IWUSR},
123 .store = dlm_control_store
124};
125
126static struct dlm_attr dlm_attr_event = {
127 .attr = {.name = "event_done", .mode = S_IWUSR},
128 .store = dlm_event_store
129};
130
131static struct dlm_attr dlm_attr_id = {
132 .attr = {.name = "id", .mode = S_IRUGO | S_IWUSR},
133 .show = dlm_id_show,
134 .store = dlm_id_store
135};
136
137static struct dlm_attr dlm_attr_nodir = {
138 .attr = {.name = "nodir", .mode = S_IRUGO | S_IWUSR},
139 .show = dlm_nodir_show,
140 .store = dlm_nodir_store
141};
142
143static struct dlm_attr dlm_attr_recover_status = {
144 .attr = {.name = "recover_status", .mode = S_IRUGO},
145 .show = dlm_recover_status_show
146};
147
148static struct dlm_attr dlm_attr_recover_nodeid = {
149 .attr = {.name = "recover_nodeid", .mode = S_IRUGO},
150 .show = dlm_recover_nodeid_show
151};
152
153static struct attribute *dlm_attrs[] = {
154 &dlm_attr_control.attr,
155 &dlm_attr_event.attr,
156 &dlm_attr_id.attr,
157 &dlm_attr_nodir.attr,
158 &dlm_attr_recover_status.attr,
159 &dlm_attr_recover_nodeid.attr,
160 NULL,
161};
162ATTRIBUTE_GROUPS(dlm);
163
164static ssize_t dlm_attr_show(struct kobject *kobj, struct attribute *attr,
165 char *buf)
166{
167 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
168 struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
169 return a->show ? a->show(ls, buf) : 0;
170}
171
172static ssize_t dlm_attr_store(struct kobject *kobj, struct attribute *attr,
173 const char *buf, size_t len)
174{
175 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
176 struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
177 return a->store ? a->store(ls, buf, len) : len;
178}
179
180static void lockspace_kobj_release(struct kobject *k)
181{
182 struct dlm_ls *ls = container_of(k, struct dlm_ls, ls_kobj);
183 kfree(ls);
184}
185
186static const struct sysfs_ops dlm_attr_ops = {
187 .show = dlm_attr_show,
188 .store = dlm_attr_store,
189};
190
191static struct kobj_type dlm_ktype = {
192 .default_groups = dlm_groups,
193 .sysfs_ops = &dlm_attr_ops,
194 .release = lockspace_kobj_release,
195};
196
197static struct kset *dlm_kset;
198
199static int do_uevent(struct dlm_ls *ls, int in)
200{
201 if (in)
202 kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE);
203 else
204 kobject_uevent(&ls->ls_kobj, KOBJ_OFFLINE);
205
206 log_rinfo(ls, "%s the lockspace group...", in ? "joining" : "leaving");
207
208 /* dlm_controld will see the uevent, do the necessary group management
209 and then write to sysfs to wake us */
210
211 wait_event(ls->ls_uevent_wait,
212 test_and_clear_bit(LSFL_UEVENT_WAIT, &ls->ls_flags));
213
214 log_rinfo(ls, "group event done %d", ls->ls_uevent_result);
215
216 return ls->ls_uevent_result;
217}
218
219static int dlm_uevent(struct kset *kset, struct kobject *kobj,
220 struct kobj_uevent_env *env)
221{
222 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
223
224 add_uevent_var(env, "LOCKSPACE=%s", ls->ls_name);
225 return 0;
226}
227
228static const struct kset_uevent_ops dlm_uevent_ops = {
229 .uevent = dlm_uevent,
230};
231
232int __init dlm_lockspace_init(void)
233{
234 ls_count = 0;
235 mutex_init(&ls_lock);
236 INIT_LIST_HEAD(&lslist);
237 spin_lock_init(&lslist_lock);
238
239 dlm_kset = kset_create_and_add("dlm", &dlm_uevent_ops, kernel_kobj);
240 if (!dlm_kset) {
241 printk(KERN_WARNING "%s: can not create kset\n", __func__);
242 return -ENOMEM;
243 }
244 return 0;
245}
246
247void dlm_lockspace_exit(void)
248{
249 kset_unregister(dlm_kset);
250}
251
252static struct dlm_ls *find_ls_to_scan(void)
253{
254 struct dlm_ls *ls;
255
256 spin_lock(&lslist_lock);
257 list_for_each_entry(ls, &lslist, ls_list) {
258 if (time_after_eq(jiffies, ls->ls_scan_time +
259 dlm_config.ci_scan_secs * HZ)) {
260 spin_unlock(&lslist_lock);
261 return ls;
262 }
263 }
264 spin_unlock(&lslist_lock);
265 return NULL;
266}
267
268static int dlm_scand(void *data)
269{
270 struct dlm_ls *ls;
271
272 while (!kthread_should_stop()) {
273 ls = find_ls_to_scan();
274 if (ls) {
275 if (dlm_lock_recovery_try(ls)) {
276 ls->ls_scan_time = jiffies;
277 dlm_scan_rsbs(ls);
278 dlm_scan_timeout(ls);
279 dlm_scan_waiters(ls);
280 dlm_unlock_recovery(ls);
281 } else {
282 ls->ls_scan_time += HZ;
283 }
284 continue;
285 }
286 schedule_timeout_interruptible(dlm_config.ci_scan_secs * HZ);
287 }
288 return 0;
289}
290
291static int dlm_scand_start(void)
292{
293 struct task_struct *p;
294 int error = 0;
295
296 p = kthread_run(dlm_scand, NULL, "dlm_scand");
297 if (IS_ERR(p))
298 error = PTR_ERR(p);
299 else
300 scand_task = p;
301 return error;
302}
303
304static void dlm_scand_stop(void)
305{
306 kthread_stop(scand_task);
307}
308
309struct dlm_ls *dlm_find_lockspace_global(uint32_t id)
310{
311 struct dlm_ls *ls;
312
313 spin_lock(&lslist_lock);
314
315 list_for_each_entry(ls, &lslist, ls_list) {
316 if (ls->ls_global_id == id) {
317 ls->ls_count++;
318 goto out;
319 }
320 }
321 ls = NULL;
322 out:
323 spin_unlock(&lslist_lock);
324 return ls;
325}
326
327struct dlm_ls *dlm_find_lockspace_local(dlm_lockspace_t *lockspace)
328{
329 struct dlm_ls *ls;
330
331 spin_lock(&lslist_lock);
332 list_for_each_entry(ls, &lslist, ls_list) {
333 if (ls->ls_local_handle == lockspace) {
334 ls->ls_count++;
335 goto out;
336 }
337 }
338 ls = NULL;
339 out:
340 spin_unlock(&lslist_lock);
341 return ls;
342}
343
344struct dlm_ls *dlm_find_lockspace_device(int minor)
345{
346 struct dlm_ls *ls;
347
348 spin_lock(&lslist_lock);
349 list_for_each_entry(ls, &lslist, ls_list) {
350 if (ls->ls_device.minor == minor) {
351 ls->ls_count++;
352 goto out;
353 }
354 }
355 ls = NULL;
356 out:
357 spin_unlock(&lslist_lock);
358 return ls;
359}
360
361void dlm_put_lockspace(struct dlm_ls *ls)
362{
363 spin_lock(&lslist_lock);
364 ls->ls_count--;
365 spin_unlock(&lslist_lock);
366}
367
368static void remove_lockspace(struct dlm_ls *ls)
369{
370 for (;;) {
371 spin_lock(&lslist_lock);
372 if (ls->ls_count == 0) {
373 WARN_ON(ls->ls_create_count != 0);
374 list_del(&ls->ls_list);
375 spin_unlock(&lslist_lock);
376 return;
377 }
378 spin_unlock(&lslist_lock);
379 ssleep(1);
380 }
381}
382
383static int threads_start(void)
384{
385 int error;
386
387 error = dlm_scand_start();
388 if (error) {
389 log_print("cannot start dlm_scand thread %d", error);
390 goto fail;
391 }
392
393 /* Thread for sending/receiving messages for all lockspace's */
394 error = dlm_midcomms_start();
395 if (error) {
396 log_print("cannot start dlm lowcomms %d", error);
397 goto scand_fail;
398 }
399
400 return 0;
401
402 scand_fail:
403 dlm_scand_stop();
404 fail:
405 return error;
406}
407
408static int new_lockspace(const char *name, const char *cluster,
409 uint32_t flags, int lvblen,
410 const struct dlm_lockspace_ops *ops, void *ops_arg,
411 int *ops_result, dlm_lockspace_t **lockspace)
412{
413 struct dlm_ls *ls;
414 int i, size, error;
415 int do_unreg = 0;
416 int namelen = strlen(name);
417
418 if (namelen > DLM_LOCKSPACE_LEN || namelen == 0)
419 return -EINVAL;
420
421 if (!lvblen || (lvblen % 8))
422 return -EINVAL;
423
424 if (!try_module_get(THIS_MODULE))
425 return -EINVAL;
426
427 if (!dlm_user_daemon_available()) {
428 log_print("dlm user daemon not available");
429 error = -EUNATCH;
430 goto out;
431 }
432
433 if (ops && ops_result) {
434 if (!dlm_config.ci_recover_callbacks)
435 *ops_result = -EOPNOTSUPP;
436 else
437 *ops_result = 0;
438 }
439
440 if (!cluster)
441 log_print("dlm cluster name '%s' is being used without an application provided cluster name",
442 dlm_config.ci_cluster_name);
443
444 if (dlm_config.ci_recover_callbacks && cluster &&
445 strncmp(cluster, dlm_config.ci_cluster_name, DLM_LOCKSPACE_LEN)) {
446 log_print("dlm cluster name '%s' does not match "
447 "the application cluster name '%s'",
448 dlm_config.ci_cluster_name, cluster);
449 error = -EBADR;
450 goto out;
451 }
452
453 error = 0;
454
455 spin_lock(&lslist_lock);
456 list_for_each_entry(ls, &lslist, ls_list) {
457 WARN_ON(ls->ls_create_count <= 0);
458 if (ls->ls_namelen != namelen)
459 continue;
460 if (memcmp(ls->ls_name, name, namelen))
461 continue;
462 if (flags & DLM_LSFL_NEWEXCL) {
463 error = -EEXIST;
464 break;
465 }
466 ls->ls_create_count++;
467 *lockspace = ls;
468 error = 1;
469 break;
470 }
471 spin_unlock(&lslist_lock);
472
473 if (error)
474 goto out;
475
476 error = -ENOMEM;
477
478 ls = kzalloc(sizeof(struct dlm_ls) + namelen, GFP_NOFS);
479 if (!ls)
480 goto out;
481 memcpy(ls->ls_name, name, namelen);
482 ls->ls_namelen = namelen;
483 ls->ls_lvblen = lvblen;
484 ls->ls_count = 0;
485 ls->ls_flags = 0;
486 ls->ls_scan_time = jiffies;
487
488 if (ops && dlm_config.ci_recover_callbacks) {
489 ls->ls_ops = ops;
490 ls->ls_ops_arg = ops_arg;
491 }
492
493 if (flags & DLM_LSFL_TIMEWARN)
494 set_bit(LSFL_TIMEWARN, &ls->ls_flags);
495
496 /* ls_exflags are forced to match among nodes, and we don't
497 need to require all nodes to have some flags set */
498 ls->ls_exflags = (flags & ~(DLM_LSFL_TIMEWARN | DLM_LSFL_FS |
499 DLM_LSFL_NEWEXCL));
500
501 size = dlm_config.ci_rsbtbl_size;
502 ls->ls_rsbtbl_size = size;
503
504 ls->ls_rsbtbl = vmalloc(array_size(size, sizeof(struct dlm_rsbtable)));
505 if (!ls->ls_rsbtbl)
506 goto out_lsfree;
507 for (i = 0; i < size; i++) {
508 ls->ls_rsbtbl[i].keep.rb_node = NULL;
509 ls->ls_rsbtbl[i].toss.rb_node = NULL;
510 spin_lock_init(&ls->ls_rsbtbl[i].lock);
511 }
512
513 spin_lock_init(&ls->ls_remove_spin);
514
515 for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++) {
516 ls->ls_remove_names[i] = kzalloc(DLM_RESNAME_MAXLEN+1,
517 GFP_KERNEL);
518 if (!ls->ls_remove_names[i])
519 goto out_rsbtbl;
520 }
521
522 idr_init(&ls->ls_lkbidr);
523 spin_lock_init(&ls->ls_lkbidr_spin);
524
525 INIT_LIST_HEAD(&ls->ls_waiters);
526 mutex_init(&ls->ls_waiters_mutex);
527 INIT_LIST_HEAD(&ls->ls_orphans);
528 mutex_init(&ls->ls_orphans_mutex);
529 INIT_LIST_HEAD(&ls->ls_timeout);
530 mutex_init(&ls->ls_timeout_mutex);
531
532 INIT_LIST_HEAD(&ls->ls_new_rsb);
533 spin_lock_init(&ls->ls_new_rsb_spin);
534
535 INIT_LIST_HEAD(&ls->ls_nodes);
536 INIT_LIST_HEAD(&ls->ls_nodes_gone);
537 ls->ls_num_nodes = 0;
538 ls->ls_low_nodeid = 0;
539 ls->ls_total_weight = 0;
540 ls->ls_node_array = NULL;
541
542 memset(&ls->ls_stub_rsb, 0, sizeof(struct dlm_rsb));
543 ls->ls_stub_rsb.res_ls = ls;
544
545 ls->ls_debug_rsb_dentry = NULL;
546 ls->ls_debug_waiters_dentry = NULL;
547
548 init_waitqueue_head(&ls->ls_uevent_wait);
549 ls->ls_uevent_result = 0;
550 init_completion(&ls->ls_members_done);
551 ls->ls_members_result = -1;
552
553 mutex_init(&ls->ls_cb_mutex);
554 INIT_LIST_HEAD(&ls->ls_cb_delay);
555
556 ls->ls_recoverd_task = NULL;
557 mutex_init(&ls->ls_recoverd_active);
558 spin_lock_init(&ls->ls_recover_lock);
559 spin_lock_init(&ls->ls_rcom_spin);
560 get_random_bytes(&ls->ls_rcom_seq, sizeof(uint64_t));
561 ls->ls_recover_status = 0;
562 ls->ls_recover_seq = 0;
563 ls->ls_recover_args = NULL;
564 init_rwsem(&ls->ls_in_recovery);
565 init_rwsem(&ls->ls_recv_active);
566 INIT_LIST_HEAD(&ls->ls_requestqueue);
567 mutex_init(&ls->ls_requestqueue_mutex);
568 mutex_init(&ls->ls_clear_proc_locks);
569
570 /* Due backwards compatibility with 3.1 we need to use maximum
571 * possible dlm message size to be sure the message will fit and
572 * not having out of bounds issues. However on sending side 3.2
573 * might send less.
574 */
575 ls->ls_recover_buf = kmalloc(DLM_MAX_SOCKET_BUFSIZE, GFP_NOFS);
576 if (!ls->ls_recover_buf)
577 goto out_lkbidr;
578
579 ls->ls_slot = 0;
580 ls->ls_num_slots = 0;
581 ls->ls_slots_size = 0;
582 ls->ls_slots = NULL;
583
584 INIT_LIST_HEAD(&ls->ls_recover_list);
585 spin_lock_init(&ls->ls_recover_list_lock);
586 idr_init(&ls->ls_recover_idr);
587 spin_lock_init(&ls->ls_recover_idr_lock);
588 ls->ls_recover_list_count = 0;
589 ls->ls_local_handle = ls;
590 init_waitqueue_head(&ls->ls_wait_general);
591 INIT_LIST_HEAD(&ls->ls_root_list);
592 init_rwsem(&ls->ls_root_sem);
593
594 spin_lock(&lslist_lock);
595 ls->ls_create_count = 1;
596 list_add(&ls->ls_list, &lslist);
597 spin_unlock(&lslist_lock);
598
599 if (flags & DLM_LSFL_FS) {
600 error = dlm_callback_start(ls);
601 if (error) {
602 log_error(ls, "can't start dlm_callback %d", error);
603 goto out_delist;
604 }
605 }
606
607 init_waitqueue_head(&ls->ls_recover_lock_wait);
608
609 /*
610 * Once started, dlm_recoverd first looks for ls in lslist, then
611 * initializes ls_in_recovery as locked in "down" mode. We need
612 * to wait for the wakeup from dlm_recoverd because in_recovery
613 * has to start out in down mode.
614 */
615
616 error = dlm_recoverd_start(ls);
617 if (error) {
618 log_error(ls, "can't start dlm_recoverd %d", error);
619 goto out_callback;
620 }
621
622 wait_event(ls->ls_recover_lock_wait,
623 test_bit(LSFL_RECOVER_LOCK, &ls->ls_flags));
624
625 /* let kobject handle freeing of ls if there's an error */
626 do_unreg = 1;
627
628 ls->ls_kobj.kset = dlm_kset;
629 error = kobject_init_and_add(&ls->ls_kobj, &dlm_ktype, NULL,
630 "%s", ls->ls_name);
631 if (error)
632 goto out_recoverd;
633 kobject_uevent(&ls->ls_kobj, KOBJ_ADD);
634
635 /* This uevent triggers dlm_controld in userspace to add us to the
636 group of nodes that are members of this lockspace (managed by the
637 cluster infrastructure.) Once it's done that, it tells us who the
638 current lockspace members are (via configfs) and then tells the
639 lockspace to start running (via sysfs) in dlm_ls_start(). */
640
641 error = do_uevent(ls, 1);
642 if (error)
643 goto out_recoverd;
644
645 wait_for_completion(&ls->ls_members_done);
646 error = ls->ls_members_result;
647 if (error)
648 goto out_members;
649
650 dlm_create_debug_file(ls);
651
652 log_rinfo(ls, "join complete");
653 *lockspace = ls;
654 return 0;
655
656 out_members:
657 do_uevent(ls, 0);
658 dlm_clear_members(ls);
659 kfree(ls->ls_node_array);
660 out_recoverd:
661 dlm_recoverd_stop(ls);
662 out_callback:
663 dlm_callback_stop(ls);
664 out_delist:
665 spin_lock(&lslist_lock);
666 list_del(&ls->ls_list);
667 spin_unlock(&lslist_lock);
668 idr_destroy(&ls->ls_recover_idr);
669 kfree(ls->ls_recover_buf);
670 out_lkbidr:
671 idr_destroy(&ls->ls_lkbidr);
672 out_rsbtbl:
673 for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++)
674 kfree(ls->ls_remove_names[i]);
675 vfree(ls->ls_rsbtbl);
676 out_lsfree:
677 if (do_unreg)
678 kobject_put(&ls->ls_kobj);
679 else
680 kfree(ls);
681 out:
682 module_put(THIS_MODULE);
683 return error;
684}
685
686int dlm_new_lockspace(const char *name, const char *cluster,
687 uint32_t flags, int lvblen,
688 const struct dlm_lockspace_ops *ops, void *ops_arg,
689 int *ops_result, dlm_lockspace_t **lockspace)
690{
691 int error = 0;
692
693 mutex_lock(&ls_lock);
694 if (!ls_count)
695 error = threads_start();
696 if (error)
697 goto out;
698
699 error = new_lockspace(name, cluster, flags, lvblen, ops, ops_arg,
700 ops_result, lockspace);
701 if (!error)
702 ls_count++;
703 if (error > 0)
704 error = 0;
705 if (!ls_count) {
706 dlm_scand_stop();
707 dlm_midcomms_shutdown();
708 dlm_lowcomms_stop();
709 }
710 out:
711 mutex_unlock(&ls_lock);
712 return error;
713}
714
715static int lkb_idr_is_local(int id, void *p, void *data)
716{
717 struct dlm_lkb *lkb = p;
718
719 return lkb->lkb_nodeid == 0 && lkb->lkb_grmode != DLM_LOCK_IV;
720}
721
722static int lkb_idr_is_any(int id, void *p, void *data)
723{
724 return 1;
725}
726
727static int lkb_idr_free(int id, void *p, void *data)
728{
729 struct dlm_lkb *lkb = p;
730
731 if (lkb->lkb_lvbptr && lkb->lkb_flags & DLM_IFL_MSTCPY)
732 dlm_free_lvb(lkb->lkb_lvbptr);
733
734 dlm_free_lkb(lkb);
735 return 0;
736}
737
738/* NOTE: We check the lkbidr here rather than the resource table.
739 This is because there may be LKBs queued as ASTs that have been unlinked
740 from their RSBs and are pending deletion once the AST has been delivered */
741
742static int lockspace_busy(struct dlm_ls *ls, int force)
743{
744 int rv;
745
746 spin_lock(&ls->ls_lkbidr_spin);
747 if (force == 0) {
748 rv = idr_for_each(&ls->ls_lkbidr, lkb_idr_is_any, ls);
749 } else if (force == 1) {
750 rv = idr_for_each(&ls->ls_lkbidr, lkb_idr_is_local, ls);
751 } else {
752 rv = 0;
753 }
754 spin_unlock(&ls->ls_lkbidr_spin);
755 return rv;
756}
757
758static int release_lockspace(struct dlm_ls *ls, int force)
759{
760 struct dlm_rsb *rsb;
761 struct rb_node *n;
762 int i, busy, rv;
763
764 busy = lockspace_busy(ls, force);
765
766 spin_lock(&lslist_lock);
767 if (ls->ls_create_count == 1) {
768 if (busy) {
769 rv = -EBUSY;
770 } else {
771 /* remove_lockspace takes ls off lslist */
772 ls->ls_create_count = 0;
773 rv = 0;
774 }
775 } else if (ls->ls_create_count > 1) {
776 rv = --ls->ls_create_count;
777 } else {
778 rv = -EINVAL;
779 }
780 spin_unlock(&lslist_lock);
781
782 if (rv) {
783 log_debug(ls, "release_lockspace no remove %d", rv);
784 return rv;
785 }
786
787 dlm_device_deregister(ls);
788
789 if (force < 3 && dlm_user_daemon_available())
790 do_uevent(ls, 0);
791
792 dlm_recoverd_stop(ls);
793
794 if (ls_count == 1) {
795 dlm_scand_stop();
796 dlm_midcomms_shutdown();
797 }
798
799 dlm_callback_stop(ls);
800
801 remove_lockspace(ls);
802
803 dlm_delete_debug_file(ls);
804
805 idr_destroy(&ls->ls_recover_idr);
806 kfree(ls->ls_recover_buf);
807
808 /*
809 * Free all lkb's in idr
810 */
811
812 idr_for_each(&ls->ls_lkbidr, lkb_idr_free, ls);
813 idr_destroy(&ls->ls_lkbidr);
814
815 /*
816 * Free all rsb's on rsbtbl[] lists
817 */
818
819 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
820 while ((n = rb_first(&ls->ls_rsbtbl[i].keep))) {
821 rsb = rb_entry(n, struct dlm_rsb, res_hashnode);
822 rb_erase(n, &ls->ls_rsbtbl[i].keep);
823 dlm_free_rsb(rsb);
824 }
825
826 while ((n = rb_first(&ls->ls_rsbtbl[i].toss))) {
827 rsb = rb_entry(n, struct dlm_rsb, res_hashnode);
828 rb_erase(n, &ls->ls_rsbtbl[i].toss);
829 dlm_free_rsb(rsb);
830 }
831 }
832
833 vfree(ls->ls_rsbtbl);
834
835 for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++)
836 kfree(ls->ls_remove_names[i]);
837
838 while (!list_empty(&ls->ls_new_rsb)) {
839 rsb = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb,
840 res_hashchain);
841 list_del(&rsb->res_hashchain);
842 dlm_free_rsb(rsb);
843 }
844
845 /*
846 * Free structures on any other lists
847 */
848
849 dlm_purge_requestqueue(ls);
850 kfree(ls->ls_recover_args);
851 dlm_clear_members(ls);
852 dlm_clear_members_gone(ls);
853 kfree(ls->ls_node_array);
854 log_rinfo(ls, "release_lockspace final free");
855 kobject_put(&ls->ls_kobj);
856 /* The ls structure will be freed when the kobject is done with */
857
858 module_put(THIS_MODULE);
859 return 0;
860}
861
862/*
863 * Called when a system has released all its locks and is not going to use the
864 * lockspace any longer. We free everything we're managing for this lockspace.
865 * Remaining nodes will go through the recovery process as if we'd died. The
866 * lockspace must continue to function as usual, participating in recoveries,
867 * until this returns.
868 *
869 * Force has 4 possible values:
870 * 0 - don't destroy locksapce if it has any LKBs
871 * 1 - destroy lockspace if it has remote LKBs but not if it has local LKBs
872 * 2 - destroy lockspace regardless of LKBs
873 * 3 - destroy lockspace as part of a forced shutdown
874 */
875
876int dlm_release_lockspace(void *lockspace, int force)
877{
878 struct dlm_ls *ls;
879 int error;
880
881 ls = dlm_find_lockspace_local(lockspace);
882 if (!ls)
883 return -EINVAL;
884 dlm_put_lockspace(ls);
885
886 mutex_lock(&ls_lock);
887 error = release_lockspace(ls, force);
888 if (!error)
889 ls_count--;
890 if (!ls_count)
891 dlm_lowcomms_stop();
892 mutex_unlock(&ls_lock);
893
894 return error;
895}
896
897void dlm_stop_lockspaces(void)
898{
899 struct dlm_ls *ls;
900 int count;
901
902 restart:
903 count = 0;
904 spin_lock(&lslist_lock);
905 list_for_each_entry(ls, &lslist, ls_list) {
906 if (!test_bit(LSFL_RUNNING, &ls->ls_flags)) {
907 count++;
908 continue;
909 }
910 spin_unlock(&lslist_lock);
911 log_error(ls, "no userland control daemon, stopping lockspace");
912 dlm_ls_stop(ls);
913 goto restart;
914 }
915 spin_unlock(&lslist_lock);
916
917 if (count)
918 log_print("dlm user daemon left %d lockspaces", count);
919}
920
1// SPDX-License-Identifier: GPL-2.0-only
2/******************************************************************************
3*******************************************************************************
4**
5** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
6** Copyright (C) 2004-2011 Red Hat, Inc. All rights reserved.
7**
8**
9*******************************************************************************
10******************************************************************************/
11
12#include <linux/module.h>
13
14#include "dlm_internal.h"
15#include "lockspace.h"
16#include "member.h"
17#include "recoverd.h"
18#include "dir.h"
19#include "midcomms.h"
20#include "config.h"
21#include "memory.h"
22#include "lock.h"
23#include "recover.h"
24#include "requestqueue.h"
25#include "user.h"
26#include "ast.h"
27
28static int ls_count;
29static struct mutex ls_lock;
30static struct list_head lslist;
31static spinlock_t lslist_lock;
32static struct task_struct * scand_task;
33
34
35static ssize_t dlm_control_store(struct dlm_ls *ls, const char *buf, size_t len)
36{
37 ssize_t ret = len;
38 int n;
39 int rc = kstrtoint(buf, 0, &n);
40
41 if (rc)
42 return rc;
43 ls = dlm_find_lockspace_local(ls->ls_local_handle);
44 if (!ls)
45 return -EINVAL;
46
47 switch (n) {
48 case 0:
49 dlm_ls_stop(ls);
50 break;
51 case 1:
52 dlm_ls_start(ls);
53 break;
54 default:
55 ret = -EINVAL;
56 }
57 dlm_put_lockspace(ls);
58 return ret;
59}
60
61static ssize_t dlm_event_store(struct dlm_ls *ls, const char *buf, size_t len)
62{
63 int rc = kstrtoint(buf, 0, &ls->ls_uevent_result);
64
65 if (rc)
66 return rc;
67 set_bit(LSFL_UEVENT_WAIT, &ls->ls_flags);
68 wake_up(&ls->ls_uevent_wait);
69 return len;
70}
71
72static ssize_t dlm_id_show(struct dlm_ls *ls, char *buf)
73{
74 return snprintf(buf, PAGE_SIZE, "%u\n", ls->ls_global_id);
75}
76
77static ssize_t dlm_id_store(struct dlm_ls *ls, const char *buf, size_t len)
78{
79 int rc = kstrtouint(buf, 0, &ls->ls_global_id);
80
81 if (rc)
82 return rc;
83 return len;
84}
85
86static ssize_t dlm_nodir_show(struct dlm_ls *ls, char *buf)
87{
88 return snprintf(buf, PAGE_SIZE, "%u\n", dlm_no_directory(ls));
89}
90
91static ssize_t dlm_nodir_store(struct dlm_ls *ls, const char *buf, size_t len)
92{
93 int val;
94 int rc = kstrtoint(buf, 0, &val);
95
96 if (rc)
97 return rc;
98 if (val == 1)
99 set_bit(LSFL_NODIR, &ls->ls_flags);
100 return len;
101}
102
103static ssize_t dlm_recover_status_show(struct dlm_ls *ls, char *buf)
104{
105 uint32_t status = dlm_recover_status(ls);
106 return snprintf(buf, PAGE_SIZE, "%x\n", status);
107}
108
109static ssize_t dlm_recover_nodeid_show(struct dlm_ls *ls, char *buf)
110{
111 return snprintf(buf, PAGE_SIZE, "%d\n", ls->ls_recover_nodeid);
112}
113
114struct dlm_attr {
115 struct attribute attr;
116 ssize_t (*show)(struct dlm_ls *, char *);
117 ssize_t (*store)(struct dlm_ls *, const char *, size_t);
118};
119
120static struct dlm_attr dlm_attr_control = {
121 .attr = {.name = "control", .mode = S_IWUSR},
122 .store = dlm_control_store
123};
124
125static struct dlm_attr dlm_attr_event = {
126 .attr = {.name = "event_done", .mode = S_IWUSR},
127 .store = dlm_event_store
128};
129
130static struct dlm_attr dlm_attr_id = {
131 .attr = {.name = "id", .mode = S_IRUGO | S_IWUSR},
132 .show = dlm_id_show,
133 .store = dlm_id_store
134};
135
136static 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
140};
141
142static struct dlm_attr dlm_attr_recover_status = {
143 .attr = {.name = "recover_status", .mode = S_IRUGO},
144 .show = dlm_recover_status_show
145};
146
147static struct dlm_attr dlm_attr_recover_nodeid = {
148 .attr = {.name = "recover_nodeid", .mode = S_IRUGO},
149 .show = dlm_recover_nodeid_show
150};
151
152static struct attribute *dlm_attrs[] = {
153 &dlm_attr_control.attr,
154 &dlm_attr_event.attr,
155 &dlm_attr_id.attr,
156 &dlm_attr_nodir.attr,
157 &dlm_attr_recover_status.attr,
158 &dlm_attr_recover_nodeid.attr,
159 NULL,
160};
161ATTRIBUTE_GROUPS(dlm);
162
163static ssize_t dlm_attr_show(struct kobject *kobj, struct attribute *attr,
164 char *buf)
165{
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;
169}
170
171static ssize_t dlm_attr_store(struct kobject *kobj, struct attribute *attr,
172 const char *buf, size_t len)
173{
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;
177}
178
179static void lockspace_kobj_release(struct kobject *k)
180{
181 struct dlm_ls *ls = container_of(k, struct dlm_ls, ls_kobj);
182 kfree(ls);
183}
184
185static const struct sysfs_ops dlm_attr_ops = {
186 .show = dlm_attr_show,
187 .store = dlm_attr_store,
188};
189
190static struct kobj_type dlm_ktype = {
191 .default_groups = dlm_groups,
192 .sysfs_ops = &dlm_attr_ops,
193 .release = lockspace_kobj_release,
194};
195
196static struct kset *dlm_kset;
197
198static int do_uevent(struct dlm_ls *ls, int in)
199{
200 if (in)
201 kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE);
202 else
203 kobject_uevent(&ls->ls_kobj, KOBJ_OFFLINE);
204
205 log_rinfo(ls, "%s the lockspace group...", in ? "joining" : "leaving");
206
207 /* dlm_controld will see the uevent, do the necessary group management
208 and then write to sysfs to wake us */
209
210 wait_event(ls->ls_uevent_wait,
211 test_and_clear_bit(LSFL_UEVENT_WAIT, &ls->ls_flags));
212
213 log_rinfo(ls, "group event done %d", ls->ls_uevent_result);
214
215 return ls->ls_uevent_result;
216}
217
218static int dlm_uevent(struct kobject *kobj, struct kobj_uevent_env *env)
219{
220 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
221
222 add_uevent_var(env, "LOCKSPACE=%s", ls->ls_name);
223 return 0;
224}
225
226static const struct kset_uevent_ops dlm_uevent_ops = {
227 .uevent = dlm_uevent,
228};
229
230int __init dlm_lockspace_init(void)
231{
232 ls_count = 0;
233 mutex_init(&ls_lock);
234 INIT_LIST_HEAD(&lslist);
235 spin_lock_init(&lslist_lock);
236
237 dlm_kset = kset_create_and_add("dlm", &dlm_uevent_ops, kernel_kobj);
238 if (!dlm_kset) {
239 printk(KERN_WARNING "%s: can not create kset\n", __func__);
240 return -ENOMEM;
241 }
242 return 0;
243}
244
245void dlm_lockspace_exit(void)
246{
247 kset_unregister(dlm_kset);
248}
249
250static struct dlm_ls *find_ls_to_scan(void)
251{
252 struct dlm_ls *ls;
253
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);
259 return ls;
260 }
261 }
262 spin_unlock(&lslist_lock);
263 return NULL;
264}
265
266static int dlm_scand(void *data)
267{
268 struct dlm_ls *ls;
269
270 while (!kthread_should_stop()) {
271 ls = find_ls_to_scan();
272 if (ls) {
273 if (dlm_lock_recovery_try(ls)) {
274 ls->ls_scan_time = jiffies;
275 dlm_scan_rsbs(ls);
276 dlm_scan_timeout(ls);
277 dlm_unlock_recovery(ls);
278 } else {
279 ls->ls_scan_time += HZ;
280 }
281 continue;
282 }
283 schedule_timeout_interruptible(dlm_config.ci_scan_secs * HZ);
284 }
285 return 0;
286}
287
288static int dlm_scand_start(void)
289{
290 struct task_struct *p;
291 int error = 0;
292
293 p = kthread_run(dlm_scand, NULL, "dlm_scand");
294 if (IS_ERR(p))
295 error = PTR_ERR(p);
296 else
297 scand_task = p;
298 return error;
299}
300
301static void dlm_scand_stop(void)
302{
303 kthread_stop(scand_task);
304}
305
306struct dlm_ls *dlm_find_lockspace_global(uint32_t id)
307{
308 struct dlm_ls *ls;
309
310 spin_lock(&lslist_lock);
311
312 list_for_each_entry(ls, &lslist, ls_list) {
313 if (ls->ls_global_id == id) {
314 atomic_inc(&ls->ls_count);
315 goto out;
316 }
317 }
318 ls = NULL;
319 out:
320 spin_unlock(&lslist_lock);
321 return ls;
322}
323
324struct dlm_ls *dlm_find_lockspace_local(dlm_lockspace_t *lockspace)
325{
326 struct dlm_ls *ls;
327
328 spin_lock(&lslist_lock);
329 list_for_each_entry(ls, &lslist, ls_list) {
330 if (ls->ls_local_handle == lockspace) {
331 atomic_inc(&ls->ls_count);
332 goto out;
333 }
334 }
335 ls = NULL;
336 out:
337 spin_unlock(&lslist_lock);
338 return ls;
339}
340
341struct dlm_ls *dlm_find_lockspace_device(int minor)
342{
343 struct dlm_ls *ls;
344
345 spin_lock(&lslist_lock);
346 list_for_each_entry(ls, &lslist, ls_list) {
347 if (ls->ls_device.minor == minor) {
348 atomic_inc(&ls->ls_count);
349 goto out;
350 }
351 }
352 ls = NULL;
353 out:
354 spin_unlock(&lslist_lock);
355 return ls;
356}
357
358void dlm_put_lockspace(struct dlm_ls *ls)
359{
360 if (atomic_dec_and_test(&ls->ls_count))
361 wake_up(&ls->ls_count_wait);
362}
363
364static void remove_lockspace(struct dlm_ls *ls)
365{
366retry:
367 wait_event(ls->ls_count_wait, atomic_read(&ls->ls_count) == 0);
368
369 spin_lock(&lslist_lock);
370 if (atomic_read(&ls->ls_count) != 0) {
371 spin_unlock(&lslist_lock);
372 goto retry;
373 }
374
375 WARN_ON(ls->ls_create_count != 0);
376 list_del(&ls->ls_list);
377 spin_unlock(&lslist_lock);
378}
379
380static int threads_start(void)
381{
382 int error;
383
384 error = dlm_scand_start();
385 if (error) {
386 log_print("cannot start dlm_scand thread %d", error);
387 goto fail;
388 }
389
390 /* Thread for sending/receiving messages for all lockspace's */
391 error = dlm_midcomms_start();
392 if (error) {
393 log_print("cannot start dlm midcomms %d", error);
394 goto scand_fail;
395 }
396
397 return 0;
398
399 scand_fail:
400 dlm_scand_stop();
401 fail:
402 return error;
403}
404
405static int new_lockspace(const char *name, const char *cluster,
406 uint32_t flags, int lvblen,
407 const struct dlm_lockspace_ops *ops, void *ops_arg,
408 int *ops_result, dlm_lockspace_t **lockspace)
409{
410 struct dlm_ls *ls;
411 int i, size, error;
412 int do_unreg = 0;
413 int namelen = strlen(name);
414
415 if (namelen > DLM_LOCKSPACE_LEN || namelen == 0)
416 return -EINVAL;
417
418 if (lvblen % 8)
419 return -EINVAL;
420
421 if (!try_module_get(THIS_MODULE))
422 return -EINVAL;
423
424 if (!dlm_user_daemon_available()) {
425 log_print("dlm user daemon not available");
426 error = -EUNATCH;
427 goto out;
428 }
429
430 if (ops && ops_result) {
431 if (!dlm_config.ci_recover_callbacks)
432 *ops_result = -EOPNOTSUPP;
433 else
434 *ops_result = 0;
435 }
436
437 if (!cluster)
438 log_print("dlm cluster name '%s' is being used without an application provided cluster name",
439 dlm_config.ci_cluster_name);
440
441 if (dlm_config.ci_recover_callbacks && cluster &&
442 strncmp(cluster, dlm_config.ci_cluster_name, DLM_LOCKSPACE_LEN)) {
443 log_print("dlm cluster name '%s' does not match "
444 "the application cluster name '%s'",
445 dlm_config.ci_cluster_name, cluster);
446 error = -EBADR;
447 goto out;
448 }
449
450 error = 0;
451
452 spin_lock(&lslist_lock);
453 list_for_each_entry(ls, &lslist, ls_list) {
454 WARN_ON(ls->ls_create_count <= 0);
455 if (ls->ls_namelen != namelen)
456 continue;
457 if (memcmp(ls->ls_name, name, namelen))
458 continue;
459 if (flags & DLM_LSFL_NEWEXCL) {
460 error = -EEXIST;
461 break;
462 }
463 ls->ls_create_count++;
464 *lockspace = ls;
465 error = 1;
466 break;
467 }
468 spin_unlock(&lslist_lock);
469
470 if (error)
471 goto out;
472
473 error = -ENOMEM;
474
475 ls = kzalloc(sizeof(*ls), GFP_NOFS);
476 if (!ls)
477 goto out;
478 memcpy(ls->ls_name, name, namelen);
479 ls->ls_namelen = namelen;
480 ls->ls_lvblen = lvblen;
481 atomic_set(&ls->ls_count, 0);
482 init_waitqueue_head(&ls->ls_count_wait);
483 ls->ls_flags = 0;
484 ls->ls_scan_time = jiffies;
485
486 if (ops && dlm_config.ci_recover_callbacks) {
487 ls->ls_ops = ops;
488 ls->ls_ops_arg = ops_arg;
489 }
490
491#ifdef CONFIG_DLM_DEPRECATED_API
492 if (flags & DLM_LSFL_TIMEWARN) {
493 pr_warn_once("===============================================================\n"
494 "WARNING: the dlm DLM_LSFL_TIMEWARN flag is being deprecated and\n"
495 " will be removed in v6.2!\n"
496 " Inclusive DLM_LSFL_TIMEWARN define in UAPI header!\n"
497 "===============================================================\n");
498
499 set_bit(LSFL_TIMEWARN, &ls->ls_flags);
500 }
501
502 /* ls_exflags are forced to match among nodes, and we don't
503 * need to require all nodes to have some flags set
504 */
505 ls->ls_exflags = (flags & ~(DLM_LSFL_TIMEWARN | DLM_LSFL_FS |
506 DLM_LSFL_NEWEXCL));
507#else
508 /* ls_exflags are forced to match among nodes, and we don't
509 * need to require all nodes to have some flags set
510 */
511 ls->ls_exflags = (flags & ~(DLM_LSFL_FS | DLM_LSFL_NEWEXCL));
512#endif
513
514 size = READ_ONCE(dlm_config.ci_rsbtbl_size);
515 ls->ls_rsbtbl_size = size;
516
517 ls->ls_rsbtbl = vmalloc(array_size(size, sizeof(struct dlm_rsbtable)));
518 if (!ls->ls_rsbtbl)
519 goto out_lsfree;
520 for (i = 0; i < size; i++) {
521 ls->ls_rsbtbl[i].keep.rb_node = NULL;
522 ls->ls_rsbtbl[i].toss.rb_node = NULL;
523 spin_lock_init(&ls->ls_rsbtbl[i].lock);
524 }
525
526 for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++) {
527 ls->ls_remove_names[i] = kzalloc(DLM_RESNAME_MAXLEN+1,
528 GFP_KERNEL);
529 if (!ls->ls_remove_names[i])
530 goto out_rsbtbl;
531 }
532
533 idr_init(&ls->ls_lkbidr);
534 spin_lock_init(&ls->ls_lkbidr_spin);
535
536 INIT_LIST_HEAD(&ls->ls_waiters);
537 mutex_init(&ls->ls_waiters_mutex);
538 INIT_LIST_HEAD(&ls->ls_orphans);
539 mutex_init(&ls->ls_orphans_mutex);
540#ifdef CONFIG_DLM_DEPRECATED_API
541 INIT_LIST_HEAD(&ls->ls_timeout);
542 mutex_init(&ls->ls_timeout_mutex);
543#endif
544
545 INIT_LIST_HEAD(&ls->ls_new_rsb);
546 spin_lock_init(&ls->ls_new_rsb_spin);
547
548 INIT_LIST_HEAD(&ls->ls_nodes);
549 INIT_LIST_HEAD(&ls->ls_nodes_gone);
550 ls->ls_num_nodes = 0;
551 ls->ls_low_nodeid = 0;
552 ls->ls_total_weight = 0;
553 ls->ls_node_array = NULL;
554
555 memset(&ls->ls_stub_rsb, 0, sizeof(struct dlm_rsb));
556 ls->ls_stub_rsb.res_ls = ls;
557
558 ls->ls_debug_rsb_dentry = NULL;
559 ls->ls_debug_waiters_dentry = NULL;
560
561 init_waitqueue_head(&ls->ls_uevent_wait);
562 ls->ls_uevent_result = 0;
563 init_completion(&ls->ls_recovery_done);
564 ls->ls_recovery_result = -1;
565
566 spin_lock_init(&ls->ls_cb_lock);
567 INIT_LIST_HEAD(&ls->ls_cb_delay);
568
569 ls->ls_recoverd_task = NULL;
570 mutex_init(&ls->ls_recoverd_active);
571 spin_lock_init(&ls->ls_recover_lock);
572 spin_lock_init(&ls->ls_rcom_spin);
573 get_random_bytes(&ls->ls_rcom_seq, sizeof(uint64_t));
574 ls->ls_recover_status = 0;
575 ls->ls_recover_seq = 0;
576 ls->ls_recover_args = NULL;
577 init_rwsem(&ls->ls_in_recovery);
578 init_rwsem(&ls->ls_recv_active);
579 INIT_LIST_HEAD(&ls->ls_requestqueue);
580 atomic_set(&ls->ls_requestqueue_cnt, 0);
581 init_waitqueue_head(&ls->ls_requestqueue_wait);
582 mutex_init(&ls->ls_requestqueue_mutex);
583 spin_lock_init(&ls->ls_clear_proc_locks);
584
585 /* Due backwards compatibility with 3.1 we need to use maximum
586 * possible dlm message size to be sure the message will fit and
587 * not having out of bounds issues. However on sending side 3.2
588 * might send less.
589 */
590 ls->ls_recover_buf = kmalloc(DLM_MAX_SOCKET_BUFSIZE, GFP_NOFS);
591 if (!ls->ls_recover_buf)
592 goto out_lkbidr;
593
594 ls->ls_slot = 0;
595 ls->ls_num_slots = 0;
596 ls->ls_slots_size = 0;
597 ls->ls_slots = NULL;
598
599 INIT_LIST_HEAD(&ls->ls_recover_list);
600 spin_lock_init(&ls->ls_recover_list_lock);
601 idr_init(&ls->ls_recover_idr);
602 spin_lock_init(&ls->ls_recover_idr_lock);
603 ls->ls_recover_list_count = 0;
604 ls->ls_local_handle = ls;
605 init_waitqueue_head(&ls->ls_wait_general);
606 INIT_LIST_HEAD(&ls->ls_root_list);
607 init_rwsem(&ls->ls_root_sem);
608
609 spin_lock(&lslist_lock);
610 ls->ls_create_count = 1;
611 list_add(&ls->ls_list, &lslist);
612 spin_unlock(&lslist_lock);
613
614 if (flags & DLM_LSFL_FS) {
615 error = dlm_callback_start(ls);
616 if (error) {
617 log_error(ls, "can't start dlm_callback %d", error);
618 goto out_delist;
619 }
620 }
621
622 init_waitqueue_head(&ls->ls_recover_lock_wait);
623
624 /*
625 * Once started, dlm_recoverd first looks for ls in lslist, then
626 * initializes ls_in_recovery as locked in "down" mode. We need
627 * to wait for the wakeup from dlm_recoverd because in_recovery
628 * has to start out in down mode.
629 */
630
631 error = dlm_recoverd_start(ls);
632 if (error) {
633 log_error(ls, "can't start dlm_recoverd %d", error);
634 goto out_callback;
635 }
636
637 wait_event(ls->ls_recover_lock_wait,
638 test_bit(LSFL_RECOVER_LOCK, &ls->ls_flags));
639
640 /* let kobject handle freeing of ls if there's an error */
641 do_unreg = 1;
642
643 ls->ls_kobj.kset = dlm_kset;
644 error = kobject_init_and_add(&ls->ls_kobj, &dlm_ktype, NULL,
645 "%s", ls->ls_name);
646 if (error)
647 goto out_recoverd;
648 kobject_uevent(&ls->ls_kobj, KOBJ_ADD);
649
650 /* This uevent triggers dlm_controld in userspace to add us to the
651 group of nodes that are members of this lockspace (managed by the
652 cluster infrastructure.) Once it's done that, it tells us who the
653 current lockspace members are (via configfs) and then tells the
654 lockspace to start running (via sysfs) in dlm_ls_start(). */
655
656 error = do_uevent(ls, 1);
657 if (error)
658 goto out_recoverd;
659
660 /* wait until recovery is successful or failed */
661 wait_for_completion(&ls->ls_recovery_done);
662 error = ls->ls_recovery_result;
663 if (error)
664 goto out_members;
665
666 dlm_create_debug_file(ls);
667
668 log_rinfo(ls, "join complete");
669 *lockspace = ls;
670 return 0;
671
672 out_members:
673 do_uevent(ls, 0);
674 dlm_clear_members(ls);
675 kfree(ls->ls_node_array);
676 out_recoverd:
677 dlm_recoverd_stop(ls);
678 out_callback:
679 dlm_callback_stop(ls);
680 out_delist:
681 spin_lock(&lslist_lock);
682 list_del(&ls->ls_list);
683 spin_unlock(&lslist_lock);
684 idr_destroy(&ls->ls_recover_idr);
685 kfree(ls->ls_recover_buf);
686 out_lkbidr:
687 idr_destroy(&ls->ls_lkbidr);
688 out_rsbtbl:
689 for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++)
690 kfree(ls->ls_remove_names[i]);
691 vfree(ls->ls_rsbtbl);
692 out_lsfree:
693 if (do_unreg)
694 kobject_put(&ls->ls_kobj);
695 else
696 kfree(ls);
697 out:
698 module_put(THIS_MODULE);
699 return error;
700}
701
702static int __dlm_new_lockspace(const char *name, const char *cluster,
703 uint32_t flags, int lvblen,
704 const struct dlm_lockspace_ops *ops,
705 void *ops_arg, int *ops_result,
706 dlm_lockspace_t **lockspace)
707{
708 int error = 0;
709
710 mutex_lock(&ls_lock);
711 if (!ls_count)
712 error = threads_start();
713 if (error)
714 goto out;
715
716 error = new_lockspace(name, cluster, flags, lvblen, ops, ops_arg,
717 ops_result, lockspace);
718 if (!error)
719 ls_count++;
720 if (error > 0)
721 error = 0;
722 if (!ls_count) {
723 dlm_scand_stop();
724 dlm_midcomms_shutdown();
725 dlm_midcomms_stop();
726 }
727 out:
728 mutex_unlock(&ls_lock);
729 return error;
730}
731
732int dlm_new_lockspace(const char *name, const char *cluster, uint32_t flags,
733 int lvblen, const struct dlm_lockspace_ops *ops,
734 void *ops_arg, int *ops_result,
735 dlm_lockspace_t **lockspace)
736{
737 return __dlm_new_lockspace(name, cluster, flags | DLM_LSFL_FS, lvblen,
738 ops, ops_arg, ops_result, lockspace);
739}
740
741int dlm_new_user_lockspace(const char *name, const char *cluster,
742 uint32_t flags, int lvblen,
743 const struct dlm_lockspace_ops *ops,
744 void *ops_arg, int *ops_result,
745 dlm_lockspace_t **lockspace)
746{
747 return __dlm_new_lockspace(name, cluster, flags, lvblen, ops,
748 ops_arg, ops_result, lockspace);
749}
750
751static int lkb_idr_is_local(int id, void *p, void *data)
752{
753 struct dlm_lkb *lkb = p;
754
755 return lkb->lkb_nodeid == 0 && lkb->lkb_grmode != DLM_LOCK_IV;
756}
757
758static int lkb_idr_is_any(int id, void *p, void *data)
759{
760 return 1;
761}
762
763static int lkb_idr_free(int id, void *p, void *data)
764{
765 struct dlm_lkb *lkb = p;
766
767 if (lkb->lkb_lvbptr && lkb->lkb_flags & DLM_IFL_MSTCPY)
768 dlm_free_lvb(lkb->lkb_lvbptr);
769
770 dlm_free_lkb(lkb);
771 return 0;
772}
773
774/* NOTE: We check the lkbidr here rather than the resource table.
775 This is because there may be LKBs queued as ASTs that have been unlinked
776 from their RSBs and are pending deletion once the AST has been delivered */
777
778static int lockspace_busy(struct dlm_ls *ls, int force)
779{
780 int rv;
781
782 spin_lock(&ls->ls_lkbidr_spin);
783 if (force == 0) {
784 rv = idr_for_each(&ls->ls_lkbidr, lkb_idr_is_any, ls);
785 } else if (force == 1) {
786 rv = idr_for_each(&ls->ls_lkbidr, lkb_idr_is_local, ls);
787 } else {
788 rv = 0;
789 }
790 spin_unlock(&ls->ls_lkbidr_spin);
791 return rv;
792}
793
794static int release_lockspace(struct dlm_ls *ls, int force)
795{
796 struct dlm_rsb *rsb;
797 struct rb_node *n;
798 int i, busy, rv;
799
800 busy = lockspace_busy(ls, force);
801
802 spin_lock(&lslist_lock);
803 if (ls->ls_create_count == 1) {
804 if (busy) {
805 rv = -EBUSY;
806 } else {
807 /* remove_lockspace takes ls off lslist */
808 ls->ls_create_count = 0;
809 rv = 0;
810 }
811 } else if (ls->ls_create_count > 1) {
812 rv = --ls->ls_create_count;
813 } else {
814 rv = -EINVAL;
815 }
816 spin_unlock(&lslist_lock);
817
818 if (rv) {
819 log_debug(ls, "release_lockspace no remove %d", rv);
820 return rv;
821 }
822
823 dlm_device_deregister(ls);
824
825 if (force < 3 && dlm_user_daemon_available())
826 do_uevent(ls, 0);
827
828 dlm_recoverd_stop(ls);
829
830 if (ls_count == 1) {
831 dlm_scand_stop();
832 dlm_clear_members(ls);
833 dlm_midcomms_shutdown();
834 }
835
836 dlm_callback_stop(ls);
837
838 remove_lockspace(ls);
839
840 dlm_delete_debug_file(ls);
841
842 idr_destroy(&ls->ls_recover_idr);
843 kfree(ls->ls_recover_buf);
844
845 /*
846 * Free all lkb's in idr
847 */
848
849 idr_for_each(&ls->ls_lkbidr, lkb_idr_free, ls);
850 idr_destroy(&ls->ls_lkbidr);
851
852 /*
853 * Free all rsb's on rsbtbl[] lists
854 */
855
856 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
857 while ((n = rb_first(&ls->ls_rsbtbl[i].keep))) {
858 rsb = rb_entry(n, struct dlm_rsb, res_hashnode);
859 rb_erase(n, &ls->ls_rsbtbl[i].keep);
860 dlm_free_rsb(rsb);
861 }
862
863 while ((n = rb_first(&ls->ls_rsbtbl[i].toss))) {
864 rsb = rb_entry(n, struct dlm_rsb, res_hashnode);
865 rb_erase(n, &ls->ls_rsbtbl[i].toss);
866 dlm_free_rsb(rsb);
867 }
868 }
869
870 vfree(ls->ls_rsbtbl);
871
872 for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++)
873 kfree(ls->ls_remove_names[i]);
874
875 while (!list_empty(&ls->ls_new_rsb)) {
876 rsb = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb,
877 res_hashchain);
878 list_del(&rsb->res_hashchain);
879 dlm_free_rsb(rsb);
880 }
881
882 /*
883 * Free structures on any other lists
884 */
885
886 dlm_purge_requestqueue(ls);
887 kfree(ls->ls_recover_args);
888 dlm_clear_members(ls);
889 dlm_clear_members_gone(ls);
890 kfree(ls->ls_node_array);
891 log_rinfo(ls, "release_lockspace final free");
892 kobject_put(&ls->ls_kobj);
893 /* The ls structure will be freed when the kobject is done with */
894
895 module_put(THIS_MODULE);
896 return 0;
897}
898
899/*
900 * Called when a system has released all its locks and is not going to use the
901 * lockspace any longer. We free everything we're managing for this lockspace.
902 * Remaining nodes will go through the recovery process as if we'd died. The
903 * lockspace must continue to function as usual, participating in recoveries,
904 * until this returns.
905 *
906 * Force has 4 possible values:
907 * 0 - don't destroy lockspace if it has any LKBs
908 * 1 - destroy lockspace if it has remote LKBs but not if it has local LKBs
909 * 2 - destroy lockspace regardless of LKBs
910 * 3 - destroy lockspace as part of a forced shutdown
911 */
912
913int dlm_release_lockspace(void *lockspace, int force)
914{
915 struct dlm_ls *ls;
916 int error;
917
918 ls = dlm_find_lockspace_local(lockspace);
919 if (!ls)
920 return -EINVAL;
921 dlm_put_lockspace(ls);
922
923 mutex_lock(&ls_lock);
924 error = release_lockspace(ls, force);
925 if (!error)
926 ls_count--;
927 if (!ls_count)
928 dlm_midcomms_stop();
929 mutex_unlock(&ls_lock);
930
931 return error;
932}
933
934void dlm_stop_lockspaces(void)
935{
936 struct dlm_ls *ls;
937 int count;
938
939 restart:
940 count = 0;
941 spin_lock(&lslist_lock);
942 list_for_each_entry(ls, &lslist, ls_list) {
943 if (!test_bit(LSFL_RUNNING, &ls->ls_flags)) {
944 count++;
945 continue;
946 }
947 spin_unlock(&lslist_lock);
948 log_error(ls, "no userland control daemon, stopping lockspace");
949 dlm_ls_stop(ls);
950 goto restart;
951 }
952 spin_unlock(&lslist_lock);
953
954 if (count)
955 log_print("dlm user daemon left %d lockspaces", count);
956}
957
958void dlm_stop_lockspaces_check(void)
959{
960 struct dlm_ls *ls;
961
962 spin_lock(&lslist_lock);
963 list_for_each_entry(ls, &lslist, ls_list) {
964 if (WARN_ON(!rwsem_is_locked(&ls->ls_in_recovery) ||
965 !dlm_locking_stopped(ls)))
966 break;
967 }
968 spin_unlock(&lslist_lock);
969}