projects/linux·blob master · refs

841 lines · 19.6 KiB

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;
32
33static ssize_t dlm_control_store(struct dlm_ls *ls, const char *buf, size_t len)
34{
35 ssize_t ret = len;
36 int n;
37 int rc = kstrtoint(buf, 0, &n);
38
39 if (rc)
40 return rc;
41 ls = dlm_find_lockspace_local(ls);
42 if (!ls)
43 return -EINVAL;
44
45 switch (n) {
46 case 0:
47 dlm_ls_stop(ls);
48 break;
49 case 1:
50 dlm_ls_start(ls);
51 break;
52 default:
53 ret = -EINVAL;
54 }
55 dlm_put_lockspace(ls);
56 return ret;
57}
58
59static ssize_t dlm_event_store(struct dlm_ls *ls, const char *buf, size_t len)
60{
61 int rc = kstrtoint(buf, 0, &ls->ls_uevent_result);
62
63 if (rc)
64 return rc;
65 set_bit(LSFL_UEVENT_WAIT, &ls->ls_flags);
66 wake_up(&ls->ls_uevent_wait);
67 return len;
68}
69
70static ssize_t dlm_id_show(struct dlm_ls *ls, char *buf)
71{
72 return snprintf(buf, PAGE_SIZE, "%u\n", ls->ls_global_id);
73}
74
75static ssize_t dlm_id_store(struct dlm_ls *ls, const char *buf, size_t len)
76{
77 int rc = kstrtouint(buf, 0, &ls->ls_global_id);
78
79 if (rc)
80 return rc;
81 return len;
82}
83
84static ssize_t dlm_nodir_show(struct dlm_ls *ls, char *buf)
85{
86 return snprintf(buf, PAGE_SIZE, "%u\n", dlm_no_directory(ls));
87}
88
89static ssize_t dlm_nodir_store(struct dlm_ls *ls, const char *buf, size_t len)
90{
91 int val;
92 int rc = kstrtoint(buf, 0, &val);
93
94 if (rc)
95 return rc;
96 if (val == 1)
97 set_bit(LSFL_NODIR, &ls->ls_flags);
98 return len;
99}
100
101static ssize_t dlm_recover_status_show(struct dlm_ls *ls, char *buf)
102{
103 uint32_t status = dlm_recover_status(ls);
104 return snprintf(buf, PAGE_SIZE, "%x\n", status);
105}
106
107static ssize_t dlm_recover_nodeid_show(struct dlm_ls *ls, char *buf)
108{
109 return snprintf(buf, PAGE_SIZE, "%d\n", ls->ls_recover_nodeid);
110}
111
112struct dlm_attr {
113 struct attribute attr;
114 ssize_t (*show)(struct dlm_ls *, char *);
115 ssize_t (*store)(struct dlm_ls *, const char *, size_t);
116};
117
118static struct dlm_attr dlm_attr_control = {
119 .attr = {.name = "control", .mode = S_IWUSR},
120 .store = dlm_control_store
121};
122
123static struct dlm_attr dlm_attr_event = {
124 .attr = {.name = "event_done", .mode = S_IWUSR},
125 .store = dlm_event_store
126};
127
128static struct dlm_attr dlm_attr_id = {
129 .attr = {.name = "id", .mode = S_IRUGO | S_IWUSR},
130 .show = dlm_id_show,
131 .store = dlm_id_store
132};
133
134static struct dlm_attr dlm_attr_nodir = {
135 .attr = {.name = "nodir", .mode = S_IRUGO | S_IWUSR},
136 .show = dlm_nodir_show,
137 .store = dlm_nodir_store
138};
139
140static struct dlm_attr dlm_attr_recover_status = {
141 .attr = {.name = "recover_status", .mode = S_IRUGO},
142 .show = dlm_recover_status_show
143};
144
145static struct dlm_attr dlm_attr_recover_nodeid = {
146 .attr = {.name = "recover_nodeid", .mode = S_IRUGO},
147 .show = dlm_recover_nodeid_show
148};
149
150static struct attribute *dlm_attrs[] = {
151 &dlm_attr_control.attr,
152 &dlm_attr_event.attr,
153 &dlm_attr_id.attr,
154 &dlm_attr_nodir.attr,
155 &dlm_attr_recover_status.attr,
156 &dlm_attr_recover_nodeid.attr,
157 NULL,
158};
159ATTRIBUTE_GROUPS(dlm);
160
161static ssize_t dlm_attr_show(struct kobject *kobj, struct attribute *attr,
162 char *buf)
163{
164 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
165 struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
166 return a->show ? a->show(ls, buf) : 0;
167}
168
169static ssize_t dlm_attr_store(struct kobject *kobj, struct attribute *attr,
170 const char *buf, size_t len)
171{
172 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
173 struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
174 return a->store ? a->store(ls, buf, len) : len;
175}
176
177static const struct sysfs_ops dlm_attr_ops = {
178 .show = dlm_attr_show,
179 .store = dlm_attr_store,
180};
181
182static struct kobj_type dlm_ktype = {
183 .default_groups = dlm_groups,
184 .sysfs_ops = &dlm_attr_ops,
185};
186
187static struct kset *dlm_kset;
188
189static int do_uevent(struct dlm_ls *ls, int in, unsigned int release_recover)
190{
191 char message[512] = {};
192 char *envp[] = { message, NULL };
193
194 if (in) {
195 kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE);
196 } else {
197 snprintf(message, 511, "RELEASE_RECOVER=%u", release_recover);
198 kobject_uevent_env(&ls->ls_kobj, KOBJ_OFFLINE, envp);
199 }
200
201 log_rinfo(ls, "%s the lockspace group...", in ? "joining" : "leaving");
202
203 /* dlm_controld will see the uevent, do the necessary group management
204 and then write to sysfs to wake us */
205
206 wait_event(ls->ls_uevent_wait,
207 test_and_clear_bit(LSFL_UEVENT_WAIT, &ls->ls_flags));
208
209 log_rinfo(ls, "group event done %d", ls->ls_uevent_result);
210
211 return ls->ls_uevent_result;
212}
213
214static int dlm_uevent(const struct kobject *kobj, struct kobj_uevent_env *env)
215{
216 const struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
217
218 add_uevent_var(env, "LOCKSPACE=%s", ls->ls_name);
219 return 0;
220}
221
222static const struct kset_uevent_ops dlm_uevent_ops = {
223 .uevent = dlm_uevent,
224};
225
226int __init dlm_lockspace_init(void)
227{
228 ls_count = 0;
229 mutex_init(&ls_lock);
230 INIT_LIST_HEAD(&lslist);
231 spin_lock_init(&lslist_lock);
232
233 dlm_kset = kset_create_and_add("dlm", &dlm_uevent_ops, kernel_kobj);
234 if (!dlm_kset) {
235 printk(KERN_WARNING "%s: can not create kset\n", __func__);
236 return -ENOMEM;
237 }
238 return 0;
239}
240
241void dlm_lockspace_exit(void)
242{
243 kset_unregister(dlm_kset);
244}
245
246struct dlm_ls *dlm_find_lockspace_global(uint32_t id)
247{
248 struct dlm_ls *ls;
249
250 spin_lock_bh(&lslist_lock);
251
252 list_for_each_entry(ls, &lslist, ls_list) {
253 if (ls->ls_global_id == id) {
254 atomic_inc(&ls->ls_count);
255 goto out;
256 }
257 }
258 ls = NULL;
259 out:
260 spin_unlock_bh(&lslist_lock);
261 return ls;
262}
263
264struct dlm_ls *dlm_find_lockspace_local(dlm_lockspace_t *lockspace)
265{
266 struct dlm_ls *ls = lockspace;
267
268 atomic_inc(&ls->ls_count);
269 return ls;
270}
271
272struct dlm_ls *dlm_find_lockspace_device(int minor)
273{
274 struct dlm_ls *ls;
275
276 spin_lock_bh(&lslist_lock);
277 list_for_each_entry(ls, &lslist, ls_list) {
278 if (ls->ls_device.minor == minor) {
279 atomic_inc(&ls->ls_count);
280 goto out;
281 }
282 }
283 ls = NULL;
284 out:
285 spin_unlock_bh(&lslist_lock);
286 return ls;
287}
288
289void dlm_put_lockspace(struct dlm_ls *ls)
290{
291 if (atomic_dec_and_test(&ls->ls_count))
292 wake_up(&ls->ls_count_wait);
293}
294
295static void remove_lockspace(struct dlm_ls *ls)
296{
297retry:
298 wait_event(ls->ls_count_wait, atomic_read(&ls->ls_count) == 0);
299
300 spin_lock_bh(&lslist_lock);
301 if (atomic_read(&ls->ls_count) != 0) {
302 spin_unlock_bh(&lslist_lock);
303 goto retry;
304 }
305
306 WARN_ON(ls->ls_create_count != 0);
307 list_del(&ls->ls_list);
308 spin_unlock_bh(&lslist_lock);
309}
310
311static int threads_start(void)
312{
313 int error;
314
315 /* Thread for sending/receiving messages for all lockspace's */
316 error = dlm_midcomms_start();
317 if (error)
318 log_print("cannot start dlm midcomms %d", error);
319
320 return error;
321}
322
323static int lkb_idr_free(struct dlm_lkb *lkb)
324{
325 if (lkb->lkb_lvbptr && test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags))
326 dlm_free_lvb(lkb->lkb_lvbptr);
327
328 dlm_free_lkb(lkb);
329 return 0;
330}
331
332static void rhash_free_rsb(void *ptr, void *arg)
333{
334 struct dlm_rsb *rsb = ptr;
335
336 dlm_free_rsb(rsb);
337}
338
339static void free_lockspace(struct work_struct *work)
340{
341 struct dlm_ls *ls = container_of(work, struct dlm_ls, ls_free_work);
342 struct dlm_lkb *lkb;
343 unsigned long id;
344
345 /*
346 * Free all lkb's in xa
347 */
348 xa_for_each(&ls->ls_lkbxa, id, lkb) {
349 lkb_idr_free(lkb);
350 }
351 xa_destroy(&ls->ls_lkbxa);
352
353 /*
354 * Free all rsb's on rsbtbl
355 */
356 rhashtable_free_and_destroy(&ls->ls_rsbtbl, rhash_free_rsb, NULL);
357
358 kfree(ls);
359}
360
361static int new_lockspace(const char *name, const char *cluster,
362 uint32_t flags, int lvblen,
363 const struct dlm_lockspace_ops *ops, void *ops_arg,
364 int *ops_result, dlm_lockspace_t **lockspace)
365{
366 struct dlm_ls *ls;
367 int namelen = strlen(name);
368 int error;
369
370 if (namelen > DLM_LOCKSPACE_LEN || namelen == 0)
371 return -EINVAL;
372
373 if (lvblen % 8)
374 return -EINVAL;
375
376 if (!try_module_get(THIS_MODULE))
377 return -EINVAL;
378
379 if (!dlm_user_daemon_available()) {
380 log_print("dlm user daemon not available");
381 error = -EUNATCH;
382 goto out;
383 }
384
385 if (ops && ops_result) {
386 if (!dlm_config.ci_recover_callbacks)
387 *ops_result = -EOPNOTSUPP;
388 else
389 *ops_result = 0;
390 }
391
392 if (!cluster)
393 log_print("dlm cluster name '%s' is being used without an application provided cluster name",
394 dlm_config.ci_cluster_name);
395
396 if (dlm_config.ci_recover_callbacks && cluster &&
397 strncmp(cluster, dlm_config.ci_cluster_name, DLM_LOCKSPACE_LEN)) {
398 log_print("dlm cluster name '%s' does not match "
399 "the application cluster name '%s'",
400 dlm_config.ci_cluster_name, cluster);
401 error = -EBADR;
402 goto out;
403 }
404
405 error = 0;
406
407 spin_lock_bh(&lslist_lock);
408 list_for_each_entry(ls, &lslist, ls_list) {
409 WARN_ON(ls->ls_create_count <= 0);
410 if (ls->ls_namelen != namelen)
411 continue;
412 if (memcmp(ls->ls_name, name, namelen))
413 continue;
414 if (flags & DLM_LSFL_NEWEXCL) {
415 error = -EEXIST;
416 break;
417 }
418 ls->ls_create_count++;
419 *lockspace = ls;
420 error = 1;
421 break;
422 }
423 spin_unlock_bh(&lslist_lock);
424
425 if (error)
426 goto out;
427
428 error = -ENOMEM;
429
430 ls = kzalloc_obj(*ls, GFP_NOFS);
431 if (!ls)
432 goto out;
433 memcpy(ls->ls_name, name, namelen);
434 ls->ls_namelen = namelen;
435 ls->ls_lvblen = lvblen;
436 atomic_set(&ls->ls_count, 0);
437 init_waitqueue_head(&ls->ls_count_wait);
438 ls->ls_flags = 0;
439
440 if (ops && dlm_config.ci_recover_callbacks) {
441 ls->ls_ops = ops;
442 ls->ls_ops_arg = ops_arg;
443 }
444
445 if (flags & DLM_LSFL_SOFTIRQ)
446 set_bit(LSFL_SOFTIRQ, &ls->ls_flags);
447
448 /* ls_exflags are forced to match among nodes, and we don't
449 * need to require all nodes to have some flags set
450 */
451 ls->ls_exflags = (flags & ~(DLM_LSFL_FS | DLM_LSFL_NEWEXCL |
452 DLM_LSFL_SOFTIRQ));
453
454 INIT_LIST_HEAD(&ls->ls_slow_inactive);
455 INIT_LIST_HEAD(&ls->ls_slow_active);
456 rwlock_init(&ls->ls_rsbtbl_lock);
457
458 error = rhashtable_init(&ls->ls_rsbtbl, &dlm_rhash_rsb_params);
459 if (error)
460 goto out_lsfree;
461
462 xa_init_flags(&ls->ls_lkbxa, XA_FLAGS_ALLOC | XA_FLAGS_LOCK_BH);
463 rwlock_init(&ls->ls_lkbxa_lock);
464
465 INIT_LIST_HEAD(&ls->ls_waiters);
466 spin_lock_init(&ls->ls_waiters_lock);
467 INIT_LIST_HEAD(&ls->ls_orphans);
468 spin_lock_init(&ls->ls_orphans_lock);
469
470 INIT_LIST_HEAD(&ls->ls_nodes);
471 INIT_LIST_HEAD(&ls->ls_nodes_gone);
472 ls->ls_num_nodes = 0;
473 ls->ls_low_nodeid = 0;
474 ls->ls_total_weight = 0;
475 ls->ls_node_array = NULL;
476
477 memset(&ls->ls_local_rsb, 0, sizeof(struct dlm_rsb));
478 ls->ls_local_rsb.res_ls = ls;
479
480 ls->ls_debug_rsb_dentry = NULL;
481 ls->ls_debug_waiters_dentry = NULL;
482
483 init_waitqueue_head(&ls->ls_uevent_wait);
484 ls->ls_uevent_result = 0;
485 init_completion(&ls->ls_recovery_done);
486 ls->ls_recovery_result = -1;
487
488 spin_lock_init(&ls->ls_cb_lock);
489 INIT_LIST_HEAD(&ls->ls_cb_delay);
490
491 INIT_WORK(&ls->ls_free_work, free_lockspace);
492
493 ls->ls_recoverd_task = NULL;
494 mutex_init(&ls->ls_recoverd_active);
495 spin_lock_init(&ls->ls_recover_lock);
496 spin_lock_init(&ls->ls_rcom_spin);
497 get_random_bytes(&ls->ls_rcom_seq, sizeof(uint64_t));
498 ls->ls_recover_status = 0;
499 ls->ls_recover_seq = get_random_u64();
500 ls->ls_recover_args = NULL;
501 init_rwsem(&ls->ls_in_recovery);
502 rwlock_init(&ls->ls_recv_active);
503 INIT_LIST_HEAD(&ls->ls_requestqueue);
504 rwlock_init(&ls->ls_requestqueue_lock);
505 spin_lock_init(&ls->ls_clear_proc_locks);
506
507 /* Due backwards compatibility with 3.1 we need to use maximum
508 * possible dlm message size to be sure the message will fit and
509 * not having out of bounds issues. However on sending side 3.2
510 * might send less.
511 */
512 ls->ls_recover_buf = kmalloc(DLM_MAX_SOCKET_BUFSIZE, GFP_NOFS);
513 if (!ls->ls_recover_buf) {
514 error = -ENOMEM;
515 goto out_lkbxa;
516 }
517
518 ls->ls_slot = 0;
519 ls->ls_num_slots = 0;
520 ls->ls_slots_size = 0;
521 ls->ls_slots = NULL;
522
523 INIT_LIST_HEAD(&ls->ls_recover_list);
524 spin_lock_init(&ls->ls_recover_list_lock);
525 xa_init_flags(&ls->ls_recover_xa, XA_FLAGS_ALLOC | XA_FLAGS_LOCK_BH);
526 spin_lock_init(&ls->ls_recover_xa_lock);
527 ls->ls_recover_list_count = 0;
528 init_waitqueue_head(&ls->ls_wait_general);
529 INIT_LIST_HEAD(&ls->ls_masters_list);
530 rwlock_init(&ls->ls_masters_lock);
531 INIT_LIST_HEAD(&ls->ls_dir_dump_list);
532 rwlock_init(&ls->ls_dir_dump_lock);
533
534 INIT_LIST_HEAD(&ls->ls_scan_list);
535 spin_lock_init(&ls->ls_scan_lock);
536 timer_setup(&ls->ls_scan_timer, dlm_rsb_scan, TIMER_DEFERRABLE);
537
538 spin_lock_bh(&lslist_lock);
539 ls->ls_create_count = 1;
540 list_add(&ls->ls_list, &lslist);
541 spin_unlock_bh(&lslist_lock);
542
543 if (flags & DLM_LSFL_FS)
544 set_bit(LSFL_FS, &ls->ls_flags);
545
546 error = dlm_callback_start(ls);
547 if (error) {
548 log_error(ls, "can't start dlm_callback %d", error);
549 goto out_delist;
550 }
551
552 init_waitqueue_head(&ls->ls_recover_lock_wait);
553
554 /*
555 * Once started, dlm_recoverd first looks for ls in lslist, then
556 * initializes ls_in_recovery as locked in "down" mode. We need
557 * to wait for the wakeup from dlm_recoverd because in_recovery
558 * has to start out in down mode.
559 */
560
561 error = dlm_recoverd_start(ls);
562 if (error) {
563 log_error(ls, "can't start dlm_recoverd %d", error);
564 goto out_callback;
565 }
566
567 wait_event(ls->ls_recover_lock_wait,
568 test_bit(LSFL_RECOVER_LOCK, &ls->ls_flags));
569
570 ls->ls_kobj.kset = dlm_kset;
571 error = kobject_init_and_add(&ls->ls_kobj, &dlm_ktype, NULL,
572 "%s", ls->ls_name);
573 if (error)
574 goto out_recoverd;
575 kobject_uevent(&ls->ls_kobj, KOBJ_ADD);
576
577 /* This uevent triggers dlm_controld in userspace to add us to the
578 group of nodes that are members of this lockspace (managed by the
579 cluster infrastructure.) Once it's done that, it tells us who the
580 current lockspace members are (via configfs) and then tells the
581 lockspace to start running (via sysfs) in dlm_ls_start(). */
582
583 error = do_uevent(ls, 1, 0);
584 if (error < 0)
585 goto out_recoverd;
586
587 /* wait until recovery is successful or failed */
588 wait_for_completion(&ls->ls_recovery_done);
589 error = ls->ls_recovery_result;
590 if (error)
591 goto out_members;
592
593 dlm_create_debug_file(ls);
594
595 log_rinfo(ls, "join complete");
596 *lockspace = ls;
597 return 0;
598
599 out_members:
600 do_uevent(ls, 0, 0);
601 dlm_clear_members(ls);
602 kfree(ls->ls_node_array);
603 out_recoverd:
604 dlm_recoverd_stop(ls);
605 out_callback:
606 dlm_callback_stop(ls);
607 out_delist:
608 spin_lock_bh(&lslist_lock);
609 list_del(&ls->ls_list);
610 spin_unlock_bh(&lslist_lock);
611 xa_destroy(&ls->ls_recover_xa);
612 kfree(ls->ls_recover_buf);
613 out_lkbxa:
614 xa_destroy(&ls->ls_lkbxa);
615 rhashtable_destroy(&ls->ls_rsbtbl);
616 out_lsfree:
617 kobject_put(&ls->ls_kobj);
618 kfree(ls);
619 out:
620 module_put(THIS_MODULE);
621 return error;
622}
623
624static int __dlm_new_lockspace(const char *name, const char *cluster,
625 uint32_t flags, int lvblen,
626 const struct dlm_lockspace_ops *ops,
627 void *ops_arg, int *ops_result,
628 dlm_lockspace_t **lockspace)
629{
630 int error = 0;
631
632 mutex_lock(&ls_lock);
633 if (!ls_count)
634 error = threads_start();
635 if (error)
636 goto out;
637
638 error = new_lockspace(name, cluster, flags, lvblen, ops, ops_arg,
639 ops_result, lockspace);
640 if (!error)
641 ls_count++;
642 if (error > 0)
643 error = 0;
644 if (!ls_count) {
645 dlm_midcomms_shutdown();
646 dlm_midcomms_stop();
647 }
648 out:
649 mutex_unlock(&ls_lock);
650 return error;
651}
652
653int dlm_new_lockspace(const char *name, const char *cluster, uint32_t flags,
654 int lvblen, const struct dlm_lockspace_ops *ops,
655 void *ops_arg, int *ops_result,
656 dlm_lockspace_t **lockspace)
657{
658 return __dlm_new_lockspace(name, cluster, flags | DLM_LSFL_FS, lvblen,
659 ops, ops_arg, ops_result, lockspace);
660}
661
662int dlm_new_user_lockspace(const char *name, const char *cluster,
663 uint32_t flags, int lvblen,
664 const struct dlm_lockspace_ops *ops,
665 void *ops_arg, int *ops_result,
666 dlm_lockspace_t **lockspace)
667{
668 if (flags & DLM_LSFL_SOFTIRQ)
669 return -EINVAL;
670
671 return __dlm_new_lockspace(name, cluster, flags, lvblen, ops,
672 ops_arg, ops_result, lockspace);
673}
674
675/* NOTE: We check the lkbxa here rather than the resource table.
676 This is because there may be LKBs queued as ASTs that have been unlinked
677 from their RSBs and are pending deletion once the AST has been delivered */
678
679static int lockspace_busy(struct dlm_ls *ls, unsigned int release_option)
680{
681 struct dlm_lkb *lkb;
682 unsigned long id;
683 int rv = 0;
684
685 read_lock_bh(&ls->ls_lkbxa_lock);
686 if (release_option == DLM_RELEASE_NO_LOCKS) {
687 xa_for_each(&ls->ls_lkbxa, id, lkb) {
688 rv = 1;
689 break;
690 }
691 } else if (release_option == DLM_RELEASE_UNUSED) {
692 /* TODO: handle this UNUSED option as NO_LOCKS in later patch */
693 xa_for_each(&ls->ls_lkbxa, id, lkb) {
694 if (lkb->lkb_nodeid == 0 &&
695 lkb->lkb_grmode != DLM_LOCK_IV) {
696 rv = 1;
697 break;
698 }
699 }
700 } else {
701 rv = 0;
702 }
703 read_unlock_bh(&ls->ls_lkbxa_lock);
704 return rv;
705}
706
707static int release_lockspace(struct dlm_ls *ls, unsigned int release_option)
708{
709 int busy, rv;
710
711 busy = lockspace_busy(ls, release_option);
712
713 spin_lock_bh(&lslist_lock);
714 if (ls->ls_create_count == 1) {
715 if (busy) {
716 rv = -EBUSY;
717 } else {
718 /* remove_lockspace takes ls off lslist */
719 ls->ls_create_count = 0;
720 rv = 0;
721 }
722 } else if (ls->ls_create_count > 1) {
723 rv = --ls->ls_create_count;
724 } else {
725 rv = -EINVAL;
726 }
727 spin_unlock_bh(&lslist_lock);
728
729 if (rv) {
730 log_debug(ls, "release_lockspace no remove %d", rv);
731 return rv;
732 }
733
734 if (ls_count == 1)
735 dlm_midcomms_version_wait();
736
737 dlm_device_deregister(ls);
738
739 if (release_option != DLM_RELEASE_NO_EVENT &&
740 dlm_user_daemon_available())
741 do_uevent(ls, 0, (release_option == DLM_RELEASE_RECOVER));
742
743 dlm_recoverd_stop(ls);
744
745 /* clear the LSFL_RUNNING flag to fast up
746 * time_shutdown_sync(), we don't care anymore
747 */
748 clear_bit(LSFL_RUNNING, &ls->ls_flags);
749 timer_shutdown_sync(&ls->ls_scan_timer);
750
751 if (ls_count == 1) {
752 dlm_clear_members(ls);
753 dlm_midcomms_shutdown();
754 }
755
756 dlm_callback_stop(ls);
757
758 remove_lockspace(ls);
759
760 dlm_delete_debug_file(ls);
761
762 kobject_put(&ls->ls_kobj);
763
764 xa_destroy(&ls->ls_recover_xa);
765 kfree(ls->ls_recover_buf);
766
767 /*
768 * Free structures on any other lists
769 */
770
771 dlm_purge_requestqueue(ls);
772 kfree(ls->ls_recover_args);
773 dlm_clear_members(ls);
774 dlm_clear_members_gone(ls);
775 kfree(ls->ls_node_array);
776
777 log_rinfo(ls, "%s final free", __func__);
778
779 /* delayed free of data structures see free_lockspace() */
780 queue_work(dlm_wq, &ls->ls_free_work);
781 module_put(THIS_MODULE);
782 return 0;
783}
784
785/*
786 * Called when a system has released all its locks and is not going to use the
787 * lockspace any longer. We free everything we're managing for this lockspace.
788 * Remaining nodes will go through the recovery process as if we'd died. The
789 * lockspace must continue to function as usual, participating in recoveries,
790 * until this returns.
791 *
792 * See DLM_RELEASE defines for release_option values and their meaning.
793 */
794
795int dlm_release_lockspace(void *lockspace, unsigned int release_option)
796{
797 struct dlm_ls *ls;
798 int error;
799
800 if (release_option > __DLM_RELEASE_MAX)
801 return -EINVAL;
802
803 ls = dlm_find_lockspace_local(lockspace);
804 if (!ls)
805 return -EINVAL;
806 dlm_put_lockspace(ls);
807
808 mutex_lock(&ls_lock);
809 error = release_lockspace(ls, release_option);
810 if (!error)
811 ls_count--;
812 if (!ls_count)
813 dlm_midcomms_stop();
814 mutex_unlock(&ls_lock);
815
816 return error;
817}
818
819void dlm_stop_lockspaces(void)
820{
821 struct dlm_ls *ls;
822 int count;
823
824 restart:
825 count = 0;
826 spin_lock_bh(&lslist_lock);
827 list_for_each_entry(ls, &lslist, ls_list) {
828 if (!test_bit(LSFL_RUNNING, &ls->ls_flags)) {
829 count++;
830 continue;
831 }
832 spin_unlock_bh(&lslist_lock);
833 log_error(ls, "no userland control daemon, stopping lockspace");
834 dlm_ls_stop(ls);
835 goto restart;
836 }
837 spin_unlock_bh(&lslist_lock);
838
839 if (count)
840 log_print("dlm user daemon left %d lockspaces", count);
841}