Merge branch 'media_fixes' of git://git.kernel.org/pub/scm/linux/kernel/git/mchehab...
[cris-mirror.git] / fs / dlm / lockspace.c
blobf994a7dfda85c69e9d1d514329add11811bd9278
1 /******************************************************************************
2 *******************************************************************************
3 **
4 ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
5 ** Copyright (C) 2004-2008 Red Hat, Inc. All rights reserved.
6 **
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"
16 #include "member.h"
17 #include "recoverd.h"
18 #include "ast.h"
19 #include "dir.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"
28 static int ls_count;
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)
37 ssize_t ret = len;
38 int n = simple_strtol(buf, NULL, 0);
40 ls = dlm_find_lockspace_local(ls->ls_local_handle);
41 if (!ls)
42 return -EINVAL;
44 switch (n) {
45 case 0:
46 dlm_ls_stop(ls);
47 break;
48 case 1:
49 dlm_ls_start(ls);
50 break;
51 default:
52 ret = -EINVAL;
54 dlm_put_lockspace(ls);
55 return ret;
58 static ssize_t dlm_event_store(struct dlm_ls *ls, const char *buf, size_t len)
60 ls->ls_uevent_result = simple_strtol(buf, NULL, 0);
61 set_bit(LSFL_UEVENT_WAIT, &ls->ls_flags);
62 wake_up(&ls->ls_uevent_wait);
63 return len;
66 static ssize_t dlm_id_show(struct dlm_ls *ls, char *buf)
68 return snprintf(buf, PAGE_SIZE, "%u\n", ls->ls_global_id);
71 static ssize_t dlm_id_store(struct dlm_ls *ls, const char *buf, size_t len)
73 ls->ls_global_id = simple_strtoul(buf, NULL, 0);
74 return len;
77 static ssize_t dlm_recover_status_show(struct dlm_ls *ls, char *buf)
79 uint32_t status = dlm_recover_status(ls);
80 return snprintf(buf, PAGE_SIZE, "%x\n", status);
83 static ssize_t dlm_recover_nodeid_show(struct dlm_ls *ls, char *buf)
85 return snprintf(buf, PAGE_SIZE, "%d\n", ls->ls_recover_nodeid);
88 struct dlm_attr {
89 struct attribute attr;
90 ssize_t (*show)(struct dlm_ls *, char *);
91 ssize_t (*store)(struct dlm_ls *, const char *, size_t);
94 static struct dlm_attr dlm_attr_control = {
95 .attr = {.name = "control", .mode = S_IWUSR},
96 .store = dlm_control_store
99 static struct dlm_attr dlm_attr_event = {
100 .attr = {.name = "event_done", .mode = S_IWUSR},
101 .store = dlm_event_store
104 static struct dlm_attr dlm_attr_id = {
105 .attr = {.name = "id", .mode = S_IRUGO | S_IWUSR},
106 .show = dlm_id_show,
107 .store = dlm_id_store
110 static struct dlm_attr dlm_attr_recover_status = {
111 .attr = {.name = "recover_status", .mode = S_IRUGO},
112 .show = dlm_recover_status_show
115 static struct dlm_attr dlm_attr_recover_nodeid = {
116 .attr = {.name = "recover_nodeid", .mode = S_IRUGO},
117 .show = dlm_recover_nodeid_show
120 static struct attribute *dlm_attrs[] = {
121 &dlm_attr_control.attr,
122 &dlm_attr_event.attr,
123 &dlm_attr_id.attr,
124 &dlm_attr_recover_status.attr,
125 &dlm_attr_recover_nodeid.attr,
126 NULL,
129 static ssize_t dlm_attr_show(struct kobject *kobj, struct attribute *attr,
130 char *buf)
132 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
133 struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
134 return a->show ? a->show(ls, buf) : 0;
137 static ssize_t dlm_attr_store(struct kobject *kobj, struct attribute *attr,
138 const char *buf, size_t len)
140 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
141 struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
142 return a->store ? a->store(ls, buf, len) : len;
145 static void lockspace_kobj_release(struct kobject *k)
147 struct dlm_ls *ls = container_of(k, struct dlm_ls, ls_kobj);
148 kfree(ls);
151 static const struct sysfs_ops dlm_attr_ops = {
152 .show = dlm_attr_show,
153 .store = dlm_attr_store,
156 static struct kobj_type dlm_ktype = {
157 .default_attrs = dlm_attrs,
158 .sysfs_ops = &dlm_attr_ops,
159 .release = lockspace_kobj_release,
162 static struct kset *dlm_kset;
164 static int do_uevent(struct dlm_ls *ls, int in)
166 int error;
168 if (in)
169 kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE);
170 else
171 kobject_uevent(&ls->ls_kobj, KOBJ_OFFLINE);
173 log_debug(ls, "%s the lockspace group...", in ? "joining" : "leaving");
175 /* dlm_controld will see the uevent, do the necessary group management
176 and then write to sysfs to wake us */
178 error = wait_event_interruptible(ls->ls_uevent_wait,
179 test_and_clear_bit(LSFL_UEVENT_WAIT, &ls->ls_flags));
181 log_debug(ls, "group event done %d %d", error, ls->ls_uevent_result);
183 if (error)
184 goto out;
186 error = ls->ls_uevent_result;
187 out:
188 if (error)
189 log_error(ls, "group %s failed %d %d", in ? "join" : "leave",
190 error, ls->ls_uevent_result);
191 return error;
194 static int dlm_uevent(struct kset *kset, struct kobject *kobj,
195 struct kobj_uevent_env *env)
197 struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
199 add_uevent_var(env, "LOCKSPACE=%s", ls->ls_name);
200 return 0;
203 static struct kset_uevent_ops dlm_uevent_ops = {
204 .uevent = dlm_uevent,
207 int __init dlm_lockspace_init(void)
209 ls_count = 0;
210 mutex_init(&ls_lock);
211 INIT_LIST_HEAD(&lslist);
212 spin_lock_init(&lslist_lock);
214 dlm_kset = kset_create_and_add("dlm", &dlm_uevent_ops, kernel_kobj);
215 if (!dlm_kset) {
216 printk(KERN_WARNING "%s: can not create kset\n", __func__);
217 return -ENOMEM;
219 return 0;
222 void dlm_lockspace_exit(void)
224 kset_unregister(dlm_kset);
227 static struct dlm_ls *find_ls_to_scan(void)
229 struct dlm_ls *ls;
231 spin_lock(&lslist_lock);
232 list_for_each_entry(ls, &lslist, ls_list) {
233 if (time_after_eq(jiffies, ls->ls_scan_time +
234 dlm_config.ci_scan_secs * HZ)) {
235 spin_unlock(&lslist_lock);
236 return ls;
239 spin_unlock(&lslist_lock);
240 return NULL;
243 static int dlm_scand(void *data)
245 struct dlm_ls *ls;
246 int timeout_jiffies = dlm_config.ci_scan_secs * HZ;
248 while (!kthread_should_stop()) {
249 ls = find_ls_to_scan();
250 if (ls) {
251 if (dlm_lock_recovery_try(ls)) {
252 ls->ls_scan_time = jiffies;
253 dlm_scan_rsbs(ls);
254 dlm_scan_timeout(ls);
255 dlm_unlock_recovery(ls);
256 } else {
257 ls->ls_scan_time += HZ;
259 } else {
260 schedule_timeout_interruptible(timeout_jiffies);
263 return 0;
266 static int dlm_scand_start(void)
268 struct task_struct *p;
269 int error = 0;
271 p = kthread_run(dlm_scand, NULL, "dlm_scand");
272 if (IS_ERR(p))
273 error = PTR_ERR(p);
274 else
275 scand_task = p;
276 return error;
279 static void dlm_scand_stop(void)
281 kthread_stop(scand_task);
284 struct dlm_ls *dlm_find_lockspace_global(uint32_t id)
286 struct dlm_ls *ls;
288 spin_lock(&lslist_lock);
290 list_for_each_entry(ls, &lslist, ls_list) {
291 if (ls->ls_global_id == id) {
292 ls->ls_count++;
293 goto out;
296 ls = NULL;
297 out:
298 spin_unlock(&lslist_lock);
299 return ls;
302 struct dlm_ls *dlm_find_lockspace_local(dlm_lockspace_t *lockspace)
304 struct dlm_ls *ls;
306 spin_lock(&lslist_lock);
307 list_for_each_entry(ls, &lslist, ls_list) {
308 if (ls->ls_local_handle == lockspace) {
309 ls->ls_count++;
310 goto out;
313 ls = NULL;
314 out:
315 spin_unlock(&lslist_lock);
316 return ls;
319 struct dlm_ls *dlm_find_lockspace_device(int minor)
321 struct dlm_ls *ls;
323 spin_lock(&lslist_lock);
324 list_for_each_entry(ls, &lslist, ls_list) {
325 if (ls->ls_device.minor == minor) {
326 ls->ls_count++;
327 goto out;
330 ls = NULL;
331 out:
332 spin_unlock(&lslist_lock);
333 return ls;
336 void dlm_put_lockspace(struct dlm_ls *ls)
338 spin_lock(&lslist_lock);
339 ls->ls_count--;
340 spin_unlock(&lslist_lock);
343 static void remove_lockspace(struct dlm_ls *ls)
345 for (;;) {
346 spin_lock(&lslist_lock);
347 if (ls->ls_count == 0) {
348 WARN_ON(ls->ls_create_count != 0);
349 list_del(&ls->ls_list);
350 spin_unlock(&lslist_lock);
351 return;
353 spin_unlock(&lslist_lock);
354 ssleep(1);
358 static int threads_start(void)
360 int error;
362 /* Thread which process lock requests for all lockspace's */
363 error = dlm_astd_start();
364 if (error) {
365 log_print("cannot start dlm_astd thread %d", error);
366 goto fail;
369 error = dlm_scand_start();
370 if (error) {
371 log_print("cannot start dlm_scand thread %d", error);
372 goto astd_fail;
375 /* Thread for sending/receiving messages for all lockspace's */
376 error = dlm_lowcomms_start();
377 if (error) {
378 log_print("cannot start dlm lowcomms %d", error);
379 goto scand_fail;
382 return 0;
384 scand_fail:
385 dlm_scand_stop();
386 astd_fail:
387 dlm_astd_stop();
388 fail:
389 return error;
392 static void threads_stop(void)
394 dlm_scand_stop();
395 dlm_lowcomms_stop();
396 dlm_astd_stop();
399 static int new_lockspace(const char *name, int namelen, void **lockspace,
400 uint32_t flags, int lvblen)
402 struct dlm_ls *ls;
403 int i, size, error;
404 int do_unreg = 0;
406 if (namelen > DLM_LOCKSPACE_LEN)
407 return -EINVAL;
409 if (!lvblen || (lvblen % 8))
410 return -EINVAL;
412 if (!try_module_get(THIS_MODULE))
413 return -EINVAL;
415 if (!dlm_user_daemon_available()) {
416 module_put(THIS_MODULE);
417 return -EUNATCH;
420 error = 0;
422 spin_lock(&lslist_lock);
423 list_for_each_entry(ls, &lslist, ls_list) {
424 WARN_ON(ls->ls_create_count <= 0);
425 if (ls->ls_namelen != namelen)
426 continue;
427 if (memcmp(ls->ls_name, name, namelen))
428 continue;
429 if (flags & DLM_LSFL_NEWEXCL) {
430 error = -EEXIST;
431 break;
433 ls->ls_create_count++;
434 *lockspace = ls;
435 error = 1;
436 break;
438 spin_unlock(&lslist_lock);
440 if (error)
441 goto out;
443 error = -ENOMEM;
445 ls = kzalloc(sizeof(struct dlm_ls) + namelen, GFP_NOFS);
446 if (!ls)
447 goto out;
448 memcpy(ls->ls_name, name, namelen);
449 ls->ls_namelen = namelen;
450 ls->ls_lvblen = lvblen;
451 ls->ls_count = 0;
452 ls->ls_flags = 0;
453 ls->ls_scan_time = jiffies;
455 if (flags & DLM_LSFL_TIMEWARN)
456 set_bit(LSFL_TIMEWARN, &ls->ls_flags);
458 /* ls_exflags are forced to match among nodes, and we don't
459 need to require all nodes to have some flags set */
460 ls->ls_exflags = (flags & ~(DLM_LSFL_TIMEWARN | DLM_LSFL_FS |
461 DLM_LSFL_NEWEXCL));
463 size = dlm_config.ci_rsbtbl_size;
464 ls->ls_rsbtbl_size = size;
466 ls->ls_rsbtbl = kmalloc(sizeof(struct dlm_rsbtable) * size, GFP_NOFS);
467 if (!ls->ls_rsbtbl)
468 goto out_lsfree;
469 for (i = 0; i < size; i++) {
470 INIT_LIST_HEAD(&ls->ls_rsbtbl[i].list);
471 INIT_LIST_HEAD(&ls->ls_rsbtbl[i].toss);
472 spin_lock_init(&ls->ls_rsbtbl[i].lock);
475 size = dlm_config.ci_lkbtbl_size;
476 ls->ls_lkbtbl_size = size;
478 ls->ls_lkbtbl = kmalloc(sizeof(struct dlm_lkbtable) * size, GFP_NOFS);
479 if (!ls->ls_lkbtbl)
480 goto out_rsbfree;
481 for (i = 0; i < size; i++) {
482 INIT_LIST_HEAD(&ls->ls_lkbtbl[i].list);
483 rwlock_init(&ls->ls_lkbtbl[i].lock);
484 ls->ls_lkbtbl[i].counter = 1;
487 size = dlm_config.ci_dirtbl_size;
488 ls->ls_dirtbl_size = size;
490 ls->ls_dirtbl = kmalloc(sizeof(struct dlm_dirtable) * size, GFP_NOFS);
491 if (!ls->ls_dirtbl)
492 goto out_lkbfree;
493 for (i = 0; i < size; i++) {
494 INIT_LIST_HEAD(&ls->ls_dirtbl[i].list);
495 spin_lock_init(&ls->ls_dirtbl[i].lock);
498 INIT_LIST_HEAD(&ls->ls_waiters);
499 mutex_init(&ls->ls_waiters_mutex);
500 INIT_LIST_HEAD(&ls->ls_orphans);
501 mutex_init(&ls->ls_orphans_mutex);
502 INIT_LIST_HEAD(&ls->ls_timeout);
503 mutex_init(&ls->ls_timeout_mutex);
505 INIT_LIST_HEAD(&ls->ls_nodes);
506 INIT_LIST_HEAD(&ls->ls_nodes_gone);
507 ls->ls_num_nodes = 0;
508 ls->ls_low_nodeid = 0;
509 ls->ls_total_weight = 0;
510 ls->ls_node_array = NULL;
512 memset(&ls->ls_stub_rsb, 0, sizeof(struct dlm_rsb));
513 ls->ls_stub_rsb.res_ls = ls;
515 ls->ls_debug_rsb_dentry = NULL;
516 ls->ls_debug_waiters_dentry = NULL;
518 init_waitqueue_head(&ls->ls_uevent_wait);
519 ls->ls_uevent_result = 0;
520 init_completion(&ls->ls_members_done);
521 ls->ls_members_result = -1;
523 ls->ls_recoverd_task = NULL;
524 mutex_init(&ls->ls_recoverd_active);
525 spin_lock_init(&ls->ls_recover_lock);
526 spin_lock_init(&ls->ls_rcom_spin);
527 get_random_bytes(&ls->ls_rcom_seq, sizeof(uint64_t));
528 ls->ls_recover_status = 0;
529 ls->ls_recover_seq = 0;
530 ls->ls_recover_args = NULL;
531 init_rwsem(&ls->ls_in_recovery);
532 init_rwsem(&ls->ls_recv_active);
533 INIT_LIST_HEAD(&ls->ls_requestqueue);
534 mutex_init(&ls->ls_requestqueue_mutex);
535 mutex_init(&ls->ls_clear_proc_locks);
537 ls->ls_recover_buf = kmalloc(dlm_config.ci_buffer_size, GFP_NOFS);
538 if (!ls->ls_recover_buf)
539 goto out_dirfree;
541 INIT_LIST_HEAD(&ls->ls_recover_list);
542 spin_lock_init(&ls->ls_recover_list_lock);
543 ls->ls_recover_list_count = 0;
544 ls->ls_local_handle = ls;
545 init_waitqueue_head(&ls->ls_wait_general);
546 INIT_LIST_HEAD(&ls->ls_root_list);
547 init_rwsem(&ls->ls_root_sem);
549 down_write(&ls->ls_in_recovery);
551 spin_lock(&lslist_lock);
552 ls->ls_create_count = 1;
553 list_add(&ls->ls_list, &lslist);
554 spin_unlock(&lslist_lock);
556 /* needs to find ls in lslist */
557 error = dlm_recoverd_start(ls);
558 if (error) {
559 log_error(ls, "can't start dlm_recoverd %d", error);
560 goto out_delist;
563 ls->ls_kobj.kset = dlm_kset;
564 error = kobject_init_and_add(&ls->ls_kobj, &dlm_ktype, NULL,
565 "%s", ls->ls_name);
566 if (error)
567 goto out_stop;
568 kobject_uevent(&ls->ls_kobj, KOBJ_ADD);
570 /* let kobject handle freeing of ls if there's an error */
571 do_unreg = 1;
573 /* This uevent triggers dlm_controld in userspace to add us to the
574 group of nodes that are members of this lockspace (managed by the
575 cluster infrastructure.) Once it's done that, it tells us who the
576 current lockspace members are (via configfs) and then tells the
577 lockspace to start running (via sysfs) in dlm_ls_start(). */
579 error = do_uevent(ls, 1);
580 if (error)
581 goto out_stop;
583 wait_for_completion(&ls->ls_members_done);
584 error = ls->ls_members_result;
585 if (error)
586 goto out_members;
588 dlm_create_debug_file(ls);
590 log_debug(ls, "join complete");
591 *lockspace = ls;
592 return 0;
594 out_members:
595 do_uevent(ls, 0);
596 dlm_clear_members(ls);
597 kfree(ls->ls_node_array);
598 out_stop:
599 dlm_recoverd_stop(ls);
600 out_delist:
601 spin_lock(&lslist_lock);
602 list_del(&ls->ls_list);
603 spin_unlock(&lslist_lock);
604 kfree(ls->ls_recover_buf);
605 out_dirfree:
606 kfree(ls->ls_dirtbl);
607 out_lkbfree:
608 kfree(ls->ls_lkbtbl);
609 out_rsbfree:
610 kfree(ls->ls_rsbtbl);
611 out_lsfree:
612 if (do_unreg)
613 kobject_put(&ls->ls_kobj);
614 else
615 kfree(ls);
616 out:
617 module_put(THIS_MODULE);
618 return error;
621 int dlm_new_lockspace(const char *name, int namelen, void **lockspace,
622 uint32_t flags, int lvblen)
624 int error = 0;
626 mutex_lock(&ls_lock);
627 if (!ls_count)
628 error = threads_start();
629 if (error)
630 goto out;
632 error = new_lockspace(name, namelen, lockspace, flags, lvblen);
633 if (!error)
634 ls_count++;
635 if (error > 0)
636 error = 0;
637 if (!ls_count)
638 threads_stop();
639 out:
640 mutex_unlock(&ls_lock);
641 return error;
644 /* Return 1 if the lockspace still has active remote locks,
645 * 2 if the lockspace still has active local locks.
647 static int lockspace_busy(struct dlm_ls *ls)
649 int i, lkb_found = 0;
650 struct dlm_lkb *lkb;
652 /* NOTE: We check the lockidtbl here rather than the resource table.
653 This is because there may be LKBs queued as ASTs that have been
654 unlinked from their RSBs and are pending deletion once the AST has
655 been delivered */
657 for (i = 0; i < ls->ls_lkbtbl_size; i++) {
658 read_lock(&ls->ls_lkbtbl[i].lock);
659 if (!list_empty(&ls->ls_lkbtbl[i].list)) {
660 lkb_found = 1;
661 list_for_each_entry(lkb, &ls->ls_lkbtbl[i].list,
662 lkb_idtbl_list) {
663 if (!lkb->lkb_nodeid) {
664 read_unlock(&ls->ls_lkbtbl[i].lock);
665 return 2;
669 read_unlock(&ls->ls_lkbtbl[i].lock);
671 return lkb_found;
674 static int release_lockspace(struct dlm_ls *ls, int force)
676 struct dlm_lkb *lkb;
677 struct dlm_rsb *rsb;
678 struct list_head *head;
679 int i, busy, rv;
681 busy = lockspace_busy(ls);
683 spin_lock(&lslist_lock);
684 if (ls->ls_create_count == 1) {
685 if (busy > force)
686 rv = -EBUSY;
687 else {
688 /* remove_lockspace takes ls off lslist */
689 ls->ls_create_count = 0;
690 rv = 0;
692 } else if (ls->ls_create_count > 1) {
693 rv = --ls->ls_create_count;
694 } else {
695 rv = -EINVAL;
697 spin_unlock(&lslist_lock);
699 if (rv) {
700 log_debug(ls, "release_lockspace no remove %d", rv);
701 return rv;
704 dlm_device_deregister(ls);
706 if (force < 3 && dlm_user_daemon_available())
707 do_uevent(ls, 0);
709 dlm_recoverd_stop(ls);
711 remove_lockspace(ls);
713 dlm_delete_debug_file(ls);
715 dlm_astd_suspend();
717 kfree(ls->ls_recover_buf);
720 * Free direntry structs.
723 dlm_dir_clear(ls);
724 kfree(ls->ls_dirtbl);
727 * Free all lkb's on lkbtbl[] lists.
730 for (i = 0; i < ls->ls_lkbtbl_size; i++) {
731 head = &ls->ls_lkbtbl[i].list;
732 while (!list_empty(head)) {
733 lkb = list_entry(head->next, struct dlm_lkb,
734 lkb_idtbl_list);
736 list_del(&lkb->lkb_idtbl_list);
738 dlm_del_ast(lkb);
740 if (lkb->lkb_lvbptr && lkb->lkb_flags & DLM_IFL_MSTCPY)
741 dlm_free_lvb(lkb->lkb_lvbptr);
743 dlm_free_lkb(lkb);
746 dlm_astd_resume();
748 kfree(ls->ls_lkbtbl);
751 * Free all rsb's on rsbtbl[] lists
754 for (i = 0; i < ls->ls_rsbtbl_size; i++) {
755 head = &ls->ls_rsbtbl[i].list;
756 while (!list_empty(head)) {
757 rsb = list_entry(head->next, struct dlm_rsb,
758 res_hashchain);
760 list_del(&rsb->res_hashchain);
761 dlm_free_rsb(rsb);
764 head = &ls->ls_rsbtbl[i].toss;
765 while (!list_empty(head)) {
766 rsb = list_entry(head->next, struct dlm_rsb,
767 res_hashchain);
768 list_del(&rsb->res_hashchain);
769 dlm_free_rsb(rsb);
773 kfree(ls->ls_rsbtbl);
776 * Free structures on any other lists
779 dlm_purge_requestqueue(ls);
780 kfree(ls->ls_recover_args);
781 dlm_clear_free_entries(ls);
782 dlm_clear_members(ls);
783 dlm_clear_members_gone(ls);
784 kfree(ls->ls_node_array);
785 log_debug(ls, "release_lockspace final free");
786 kobject_put(&ls->ls_kobj);
787 /* The ls structure will be freed when the kobject is done with */
789 module_put(THIS_MODULE);
790 return 0;
794 * Called when a system has released all its locks and is not going to use the
795 * lockspace any longer. We free everything we're managing for this lockspace.
796 * Remaining nodes will go through the recovery process as if we'd died. The
797 * lockspace must continue to function as usual, participating in recoveries,
798 * until this returns.
800 * Force has 4 possible values:
801 * 0 - don't destroy locksapce if it has any LKBs
802 * 1 - destroy lockspace if it has remote LKBs but not if it has local LKBs
803 * 2 - destroy lockspace regardless of LKBs
804 * 3 - destroy lockspace as part of a forced shutdown
807 int dlm_release_lockspace(void *lockspace, int force)
809 struct dlm_ls *ls;
810 int error;
812 ls = dlm_find_lockspace_local(lockspace);
813 if (!ls)
814 return -EINVAL;
815 dlm_put_lockspace(ls);
817 mutex_lock(&ls_lock);
818 error = release_lockspace(ls, force);
819 if (!error)
820 ls_count--;
821 if (!ls_count)
822 threads_stop();
823 mutex_unlock(&ls_lock);
825 return error;
828 void dlm_stop_lockspaces(void)
830 struct dlm_ls *ls;
832 restart:
833 spin_lock(&lslist_lock);
834 list_for_each_entry(ls, &lslist, ls_list) {
835 if (!test_bit(LSFL_RUNNING, &ls->ls_flags))
836 continue;
837 spin_unlock(&lslist_lock);
838 log_error(ls, "no userland control daemon, stopping lockspace");
839 dlm_ls_stop(ls);
840 goto restart;
842 spin_unlock(&lslist_lock);