Merge branch 'for-linus' of git://git.kernel.org/pub/scm/linux/kernel/git/jikos/trivial
[linux-3.10.git] / fs / dlm / lockspace.c
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.
10 **
11 *******************************************************************************
12 ******************************************************************************/
13
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"
27
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;
33
34
35 static ssize_t dlm_control_store(struct dlm_ls *ls, const char *buf, size_t len)
36 {
37         ssize_t ret = len;
38         int n = simple_strtol(buf, NULL, 0);
39
40         ls = dlm_find_lockspace_local(ls->ls_local_handle);
41         if (!ls)
42                 return -EINVAL;
43
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;
53         }
54         dlm_put_lockspace(ls);
55         return ret;
56 }
57
58 static ssize_t dlm_event_store(struct dlm_ls *ls, const char *buf, size_t len)
59 {
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;
64 }
65
66 static ssize_t dlm_id_show(struct dlm_ls *ls, char *buf)
67 {
68         return snprintf(buf, PAGE_SIZE, "%u\n", ls->ls_global_id);
69 }
70
71 static ssize_t dlm_id_store(struct dlm_ls *ls, const char *buf, size_t len)
72 {
73         ls->ls_global_id = simple_strtoul(buf, NULL, 0);
74         return len;
75 }
76
77 static ssize_t dlm_recover_status_show(struct dlm_ls *ls, char *buf)
78 {
79         uint32_t status = dlm_recover_status(ls);
80         return snprintf(buf, PAGE_SIZE, "%x\n", status);
81 }
82
83 static ssize_t dlm_recover_nodeid_show(struct dlm_ls *ls, char *buf)
84 {
85         return snprintf(buf, PAGE_SIZE, "%d\n", ls->ls_recover_nodeid);
86 }
87
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);
92 };
93
94 static struct dlm_attr dlm_attr_control = {
95         .attr  = {.name = "control", .mode = S_IWUSR},
96         .store = dlm_control_store
97 };
98
99 static struct dlm_attr dlm_attr_event = {
100         .attr  = {.name = "event_done", .mode = S_IWUSR},
101         .store = dlm_event_store
102 };
103
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
108 };
109
110 static struct dlm_attr dlm_attr_recover_status = {
111         .attr  = {.name = "recover_status", .mode = S_IRUGO},
112         .show  = dlm_recover_status_show
113 };
114
115 static struct dlm_attr dlm_attr_recover_nodeid = {
116         .attr  = {.name = "recover_nodeid", .mode = S_IRUGO},
117         .show  = dlm_recover_nodeid_show
118 };
119
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,
127 };
128
129 static ssize_t dlm_attr_show(struct kobject *kobj, struct attribute *attr,
130                              char *buf)
131 {
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;
135 }
136
137 static ssize_t dlm_attr_store(struct kobject *kobj, struct attribute *attr,
138                               const char *buf, size_t len)
139 {
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;
143 }
144
145 static void lockspace_kobj_release(struct kobject *k)
146 {
147         struct dlm_ls *ls  = container_of(k, struct dlm_ls, ls_kobj);
148         kfree(ls);
149 }
150
151 static const struct sysfs_ops dlm_attr_ops = {
152         .show  = dlm_attr_show,
153         .store = dlm_attr_store,
154 };
155
156 static struct kobj_type dlm_ktype = {
157         .default_attrs = dlm_attrs,
158         .sysfs_ops     = &dlm_attr_ops,
159         .release       = lockspace_kobj_release,
160 };
161
162 static struct kset *dlm_kset;
163
164 static int do_uevent(struct dlm_ls *ls, int in)
165 {
166         int error;
167
168         if (in)
169                 kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE);
170         else
171                 kobject_uevent(&ls->ls_kobj, KOBJ_OFFLINE);
172
173         log_debug(ls, "%s the lockspace group...", in ? "joining" : "leaving");
174
175         /* dlm_controld will see the uevent, do the necessary group management
176            and then write to sysfs to wake us */
177
178         error = wait_event_interruptible(ls->ls_uevent_wait,
179                         test_and_clear_bit(LSFL_UEVENT_WAIT, &ls->ls_flags));
180
181         log_debug(ls, "group event done %d %d", error, ls->ls_uevent_result);
182
183         if (error)
184                 goto out;
185
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;
192 }
193
194 static int dlm_uevent(struct kset *kset, struct kobject *kobj,
195                       struct kobj_uevent_env *env)
196 {
197         struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
198
199         add_uevent_var(env, "LOCKSPACE=%s", ls->ls_name);
200         return 0;
201 }
202
203 static struct kset_uevent_ops dlm_uevent_ops = {
204         .uevent = dlm_uevent,
205 };
206
207 int __init dlm_lockspace_init(void)
208 {
209         ls_count = 0;
210         mutex_init(&ls_lock);
211         INIT_LIST_HEAD(&lslist);
212         spin_lock_init(&lslist_lock);
213
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;
218         }
219         return 0;
220 }
221
222 void dlm_lockspace_exit(void)
223 {
224         kset_unregister(dlm_kset);
225 }
226
227 static struct dlm_ls *find_ls_to_scan(void)
228 {
229         struct dlm_ls *ls;
230
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;
237                 }
238         }
239         spin_unlock(&lslist_lock);
240         return NULL;
241 }
242
243 static int dlm_scand(void *data)
244 {
245         struct dlm_ls *ls;
246         int timeout_jiffies = dlm_config.ci_scan_secs * HZ;
247
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;
258                         }
259                 } else {
260                         schedule_timeout_interruptible(timeout_jiffies);
261                 }
262         }
263         return 0;
264 }
265
266 static int dlm_scand_start(void)
267 {
268         struct task_struct *p;
269         int error = 0;
270
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;
277 }
278
279 static void dlm_scand_stop(void)
280 {
281         kthread_stop(scand_task);
282 }
283
284 struct dlm_ls *dlm_find_lockspace_global(uint32_t id)
285 {
286         struct dlm_ls *ls;
287
288         spin_lock(&lslist_lock);
289
290         list_for_each_entry(ls, &lslist, ls_list) {
291                 if (ls->ls_global_id == id) {
292                         ls->ls_count++;
293                         goto out;
294                 }
295         }
296         ls = NULL;
297  out:
298         spin_unlock(&lslist_lock);
299         return ls;
300 }
301
302 struct dlm_ls *dlm_find_lockspace_local(dlm_lockspace_t *lockspace)
303 {
304         struct dlm_ls *ls;
305
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;
311                 }
312         }
313         ls = NULL;
314  out:
315         spin_unlock(&lslist_lock);
316         return ls;
317 }
318
319 struct dlm_ls *dlm_find_lockspace_device(int minor)
320 {
321         struct dlm_ls *ls;
322
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;
328                 }
329         }
330         ls = NULL;
331  out:
332         spin_unlock(&lslist_lock);
333         return ls;
334 }
335
336 void dlm_put_lockspace(struct dlm_ls *ls)
337 {
338         spin_lock(&lslist_lock);
339         ls->ls_count--;
340         spin_unlock(&lslist_lock);
341 }
342
343 static void remove_lockspace(struct dlm_ls *ls)
344 {
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;
352                 }
353                 spin_unlock(&lslist_lock);
354                 ssleep(1);
355         }
356 }
357
358 static int threads_start(void)
359 {
360         int error;
361
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;
367         }
368
369         error = dlm_scand_start();
370         if (error) {
371                 log_print("cannot start dlm_scand thread %d", error);
372                 goto astd_fail;
373         }
374
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;
380         }
381
382         return 0;
383
384  scand_fail:
385         dlm_scand_stop();
386  astd_fail:
387         dlm_astd_stop();
388  fail:
389         return error;
390 }
391
392 static void threads_stop(void)
393 {
394         dlm_scand_stop();
395         dlm_lowcomms_stop();
396         dlm_astd_stop();
397 }
398
399 static int new_lockspace(const char *name, int namelen, void **lockspace,
400                          uint32_t flags, int lvblen)
401 {
402         struct dlm_ls *ls;
403         int i, size, error;
404         int do_unreg = 0;
405
406         if (namelen > DLM_LOCKSPACE_LEN)
407                 return -EINVAL;
408
409         if (!lvblen || (lvblen % 8))
410                 return -EINVAL;
411
412         if (!try_module_get(THIS_MODULE))
413                 return -EINVAL;
414
415         if (!dlm_user_daemon_available()) {
416                 module_put(THIS_MODULE);
417                 return -EUNATCH;
418         }
419
420         error = 0;
421
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;
432                 }
433                 ls->ls_create_count++;
434                 *lockspace = ls;
435                 error = 1;
436                 break;
437         }
438         spin_unlock(&lslist_lock);
439
440         if (error)
441                 goto out;
442
443         error = -ENOMEM;
444
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;
454
455         if (flags & DLM_LSFL_TIMEWARN)
456                 set_bit(LSFL_TIMEWARN, &ls->ls_flags);
457
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));
462
463         size = dlm_config.ci_rsbtbl_size;
464         ls->ls_rsbtbl_size = size;
465
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);
473         }
474
475         size = dlm_config.ci_lkbtbl_size;
476         ls->ls_lkbtbl_size = size;
477
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;
485         }
486
487         size = dlm_config.ci_dirtbl_size;
488         ls->ls_dirtbl_size = size;
489
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);
496         }
497
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);
504
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;
511
512         memset(&ls->ls_stub_rsb, 0, sizeof(struct dlm_rsb));
513         ls->ls_stub_rsb.res_ls = ls;
514
515         ls->ls_debug_rsb_dentry = NULL;
516         ls->ls_debug_waiters_dentry = NULL;
517
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;
522
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);
536
537         ls->ls_recover_buf = kmalloc(dlm_config.ci_buffer_size, GFP_NOFS);
538         if (!ls->ls_recover_buf)
539                 goto out_dirfree;
540
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);
548
549         down_write(&ls->ls_in_recovery);
550
551         spin_lock(&lslist_lock);
552         ls->ls_create_count = 1;
553         list_add(&ls->ls_list, &lslist);
554         spin_unlock(&lslist_lock);
555
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;
561         }
562
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);
569
570         /* let kobject handle freeing of ls if there's an error */
571         do_unreg = 1;
572
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(). */
578
579         error = do_uevent(ls, 1);
580         if (error)
581                 goto out_stop;
582
583         wait_for_completion(&ls->ls_members_done);
584         error = ls->ls_members_result;
585         if (error)
586                 goto out_members;
587
588         dlm_create_debug_file(ls);
589
590         log_debug(ls, "join complete");
591         *lockspace = ls;
592         return 0;
593
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;
619 }
620
621 int dlm_new_lockspace(const char *name, int namelen, void **lockspace,
622                       uint32_t flags, int lvblen)
623 {
624         int error = 0;
625
626         mutex_lock(&ls_lock);
627         if (!ls_count)
628                 error = threads_start();
629         if (error)
630                 goto out;
631
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;
642 }
643
644 /* Return 1 if the lockspace still has active remote locks,
645  *        2 if the lockspace still has active local locks.
646  */
647 static int lockspace_busy(struct dlm_ls *ls)
648 {
649         int i, lkb_found = 0;
650         struct dlm_lkb *lkb;
651
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 */
656
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;
666                                 }
667                         }
668                 }
669                 read_unlock(&ls->ls_lkbtbl[i].lock);
670         }
671         return lkb_found;
672 }
673
674 static int release_lockspace(struct dlm_ls *ls, int force)
675 {
676         struct dlm_lkb *lkb;
677         struct dlm_rsb *rsb;
678         struct list_head *head;
679         int i, busy, rv;
680
681         busy = lockspace_busy(ls);
682
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;
691                 }
692         } else if (ls->ls_create_count > 1) {
693                 rv = --ls->ls_create_count;
694         } else {
695                 rv = -EINVAL;
696         }
697         spin_unlock(&lslist_lock);
698
699         if (rv) {
700                 log_debug(ls, "release_lockspace no remove %d", rv);
701                 return rv;
702         }
703
704         dlm_device_deregister(ls);
705
706         if (force < 3 && dlm_user_daemon_available())
707                 do_uevent(ls, 0);
708
709         dlm_recoverd_stop(ls);
710
711         remove_lockspace(ls);
712
713         dlm_delete_debug_file(ls);
714
715         dlm_astd_suspend();
716
717         kfree(ls->ls_recover_buf);
718
719         /*
720          * Free direntry structs.
721          */
722
723         dlm_dir_clear(ls);
724         kfree(ls->ls_dirtbl);
725
726         /*
727          * Free all lkb's on lkbtbl[] lists.
728          */
729
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);
735
736                         list_del(&lkb->lkb_idtbl_list);
737
738                         dlm_del_ast(lkb);
739
740                         if (lkb->lkb_lvbptr && lkb->lkb_flags & DLM_IFL_MSTCPY)
741                                 dlm_free_lvb(lkb->lkb_lvbptr);
742
743                         dlm_free_lkb(lkb);
744                 }
745         }
746         dlm_astd_resume();
747
748         kfree(ls->ls_lkbtbl);
749
750         /*
751          * Free all rsb's on rsbtbl[] lists
752          */
753
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);
759
760                         list_del(&rsb->res_hashchain);
761                         dlm_free_rsb(rsb);
762                 }
763
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);
770                 }
771         }
772
773         kfree(ls->ls_rsbtbl);
774
775         /*
776          * Free structures on any other lists
777          */
778
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 */
788
789         module_put(THIS_MODULE);
790         return 0;
791 }
792
793 /*
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.
799  *
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
805  */
806
807 int dlm_release_lockspace(void *lockspace, int force)
808 {
809         struct dlm_ls *ls;
810         int error;
811
812         ls = dlm_find_lockspace_local(lockspace);
813         if (!ls)
814                 return -EINVAL;
815         dlm_put_lockspace(ls);
816
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);
824
825         return error;
826 }
827
828 void dlm_stop_lockspaces(void)
829 {
830         struct dlm_ls *ls;
831
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;
841         }
842         spin_unlock(&lslist_lock);
843 }
844