SLUB slab validation: Move tracking information alloc outside of lock
[linux-2.6] / fs / dlm / lockspace.c
1 /******************************************************************************
2 *******************************************************************************
3 **
4 **  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
5 **  Copyright (C) 2004-2007 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
27 #ifdef CONFIG_DLM_DEBUG
28 int dlm_create_debug_file(struct dlm_ls *ls);
29 void dlm_delete_debug_file(struct dlm_ls *ls);
30 #else
31 static inline int dlm_create_debug_file(struct dlm_ls *ls) { return 0; }
32 static inline void dlm_delete_debug_file(struct dlm_ls *ls) { }
33 #endif
34
35 static int                      ls_count;
36 static struct mutex             ls_lock;
37 static struct list_head         lslist;
38 static spinlock_t               lslist_lock;
39 static struct task_struct *     scand_task;
40
41
42 static ssize_t dlm_control_store(struct dlm_ls *ls, const char *buf, size_t len)
43 {
44         ssize_t ret = len;
45         int n = simple_strtol(buf, NULL, 0);
46
47         ls = dlm_find_lockspace_local(ls->ls_local_handle);
48         if (!ls)
49                 return -EINVAL;
50
51         switch (n) {
52         case 0:
53                 dlm_ls_stop(ls);
54                 break;
55         case 1:
56                 dlm_ls_start(ls);
57                 break;
58         default:
59                 ret = -EINVAL;
60         }
61         dlm_put_lockspace(ls);
62         return ret;
63 }
64
65 static ssize_t dlm_event_store(struct dlm_ls *ls, const char *buf, size_t len)
66 {
67         ls->ls_uevent_result = simple_strtol(buf, NULL, 0);
68         set_bit(LSFL_UEVENT_WAIT, &ls->ls_flags);
69         wake_up(&ls->ls_uevent_wait);
70         return len;
71 }
72
73 static 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
78 static ssize_t dlm_id_store(struct dlm_ls *ls, const char *buf, size_t len)
79 {
80         ls->ls_global_id = simple_strtoul(buf, NULL, 0);
81         return len;
82 }
83
84 static ssize_t dlm_recover_status_show(struct dlm_ls *ls, char *buf)
85 {
86         uint32_t status = dlm_recover_status(ls);
87         return snprintf(buf, PAGE_SIZE, "%x\n", status);
88 }
89
90 static ssize_t dlm_recover_nodeid_show(struct dlm_ls *ls, char *buf)
91 {
92         return snprintf(buf, PAGE_SIZE, "%d\n", ls->ls_recover_nodeid);
93 }
94
95 struct dlm_attr {
96         struct attribute attr;
97         ssize_t (*show)(struct dlm_ls *, char *);
98         ssize_t (*store)(struct dlm_ls *, const char *, size_t);
99 };
100
101 static struct dlm_attr dlm_attr_control = {
102         .attr  = {.name = "control", .mode = S_IWUSR},
103         .store = dlm_control_store
104 };
105
106 static struct dlm_attr dlm_attr_event = {
107         .attr  = {.name = "event_done", .mode = S_IWUSR},
108         .store = dlm_event_store
109 };
110
111 static struct dlm_attr dlm_attr_id = {
112         .attr  = {.name = "id", .mode = S_IRUGO | S_IWUSR},
113         .show  = dlm_id_show,
114         .store = dlm_id_store
115 };
116
117 static struct dlm_attr dlm_attr_recover_status = {
118         .attr  = {.name = "recover_status", .mode = S_IRUGO},
119         .show  = dlm_recover_status_show
120 };
121
122 static struct dlm_attr dlm_attr_recover_nodeid = {
123         .attr  = {.name = "recover_nodeid", .mode = S_IRUGO},
124         .show  = dlm_recover_nodeid_show
125 };
126
127 static struct attribute *dlm_attrs[] = {
128         &dlm_attr_control.attr,
129         &dlm_attr_event.attr,
130         &dlm_attr_id.attr,
131         &dlm_attr_recover_status.attr,
132         &dlm_attr_recover_nodeid.attr,
133         NULL,
134 };
135
136 static ssize_t dlm_attr_show(struct kobject *kobj, struct attribute *attr,
137                              char *buf)
138 {
139         struct dlm_ls *ls  = container_of(kobj, struct dlm_ls, ls_kobj);
140         struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
141         return a->show ? a->show(ls, buf) : 0;
142 }
143
144 static ssize_t dlm_attr_store(struct kobject *kobj, struct attribute *attr,
145                               const char *buf, size_t len)
146 {
147         struct dlm_ls *ls  = container_of(kobj, struct dlm_ls, ls_kobj);
148         struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
149         return a->store ? a->store(ls, buf, len) : len;
150 }
151
152 static void lockspace_kobj_release(struct kobject *k)
153 {
154         struct dlm_ls *ls  = container_of(k, struct dlm_ls, ls_kobj);
155         kfree(ls);
156 }
157
158 static struct sysfs_ops dlm_attr_ops = {
159         .show  = dlm_attr_show,
160         .store = dlm_attr_store,
161 };
162
163 static struct kobj_type dlm_ktype = {
164         .default_attrs = dlm_attrs,
165         .sysfs_ops     = &dlm_attr_ops,
166         .release       = lockspace_kobj_release,
167 };
168
169 static struct kset dlm_kset = {
170         .kobj   = {.name = "dlm",},
171         .ktype  = &dlm_ktype,
172 };
173
174 static int kobject_setup(struct dlm_ls *ls)
175 {
176         char lsname[DLM_LOCKSPACE_LEN];
177         int error;
178
179         memset(lsname, 0, DLM_LOCKSPACE_LEN);
180         snprintf(lsname, DLM_LOCKSPACE_LEN, "%s", ls->ls_name);
181
182         error = kobject_set_name(&ls->ls_kobj, "%s", lsname);
183         if (error)
184                 return error;
185
186         ls->ls_kobj.kset = &dlm_kset;
187         ls->ls_kobj.ktype = &dlm_ktype;
188         return 0;
189 }
190
191 static int do_uevent(struct dlm_ls *ls, int in)
192 {
193         int error;
194
195         if (in)
196                 kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE);
197         else
198                 kobject_uevent(&ls->ls_kobj, KOBJ_OFFLINE);
199
200         log_debug(ls, "%s the lockspace group...", in ? "joining" : "leaving");
201
202         /* dlm_controld will see the uevent, do the necessary group management
203            and then write to sysfs to wake us */
204
205         error = wait_event_interruptible(ls->ls_uevent_wait,
206                         test_and_clear_bit(LSFL_UEVENT_WAIT, &ls->ls_flags));
207
208         log_debug(ls, "group event done %d %d", error, ls->ls_uevent_result);
209
210         if (error)
211                 goto out;
212
213         error = ls->ls_uevent_result;
214  out:
215         if (error)
216                 log_error(ls, "group %s failed %d %d", in ? "join" : "leave",
217                           error, ls->ls_uevent_result);
218         return error;
219 }
220
221
222 int dlm_lockspace_init(void)
223 {
224         int error;
225
226         ls_count = 0;
227         mutex_init(&ls_lock);
228         INIT_LIST_HEAD(&lslist);
229         spin_lock_init(&lslist_lock);
230
231         kobj_set_kset_s(&dlm_kset, kernel_subsys);
232         error = kset_register(&dlm_kset);
233         if (error)
234                 printk("dlm_lockspace_init: cannot register kset %d\n", error);
235         return error;
236 }
237
238 void dlm_lockspace_exit(void)
239 {
240         kset_unregister(&dlm_kset);
241 }
242
243 static int dlm_scand(void *data)
244 {
245         struct dlm_ls *ls;
246
247         while (!kthread_should_stop()) {
248                 list_for_each_entry(ls, &lslist, ls_list) {
249                         if (dlm_lock_recovery_try(ls)) {
250                                 dlm_scan_rsbs(ls);
251                                 dlm_scan_timeout(ls);
252                                 dlm_unlock_recovery(ls);
253                         }
254                 }
255                 schedule_timeout_interruptible(dlm_config.ci_scan_secs * HZ);
256         }
257         return 0;
258 }
259
260 static int dlm_scand_start(void)
261 {
262         struct task_struct *p;
263         int error = 0;
264
265         p = kthread_run(dlm_scand, NULL, "dlm_scand");
266         if (IS_ERR(p))
267                 error = PTR_ERR(p);
268         else
269                 scand_task = p;
270         return error;
271 }
272
273 static void dlm_scand_stop(void)
274 {
275         kthread_stop(scand_task);
276 }
277
278 static struct dlm_ls *dlm_find_lockspace_name(char *name, int namelen)
279 {
280         struct dlm_ls *ls;
281
282         spin_lock(&lslist_lock);
283
284         list_for_each_entry(ls, &lslist, ls_list) {
285                 if (ls->ls_namelen == namelen &&
286                     memcmp(ls->ls_name, name, namelen) == 0)
287                         goto out;
288         }
289         ls = NULL;
290  out:
291         spin_unlock(&lslist_lock);
292         return ls;
293 }
294
295 struct dlm_ls *dlm_find_lockspace_global(uint32_t id)
296 {
297         struct dlm_ls *ls;
298
299         spin_lock(&lslist_lock);
300
301         list_for_each_entry(ls, &lslist, ls_list) {
302                 if (ls->ls_global_id == id) {
303                         ls->ls_count++;
304                         goto out;
305                 }
306         }
307         ls = NULL;
308  out:
309         spin_unlock(&lslist_lock);
310         return ls;
311 }
312
313 struct dlm_ls *dlm_find_lockspace_local(dlm_lockspace_t *lockspace)
314 {
315         struct dlm_ls *ls;
316
317         spin_lock(&lslist_lock);
318         list_for_each_entry(ls, &lslist, ls_list) {
319                 if (ls->ls_local_handle == lockspace) {
320                         ls->ls_count++;
321                         goto out;
322                 }
323         }
324         ls = NULL;
325  out:
326         spin_unlock(&lslist_lock);
327         return ls;
328 }
329
330 struct dlm_ls *dlm_find_lockspace_device(int minor)
331 {
332         struct dlm_ls *ls;
333
334         spin_lock(&lslist_lock);
335         list_for_each_entry(ls, &lslist, ls_list) {
336                 if (ls->ls_device.minor == minor) {
337                         ls->ls_count++;
338                         goto out;
339                 }
340         }
341         ls = NULL;
342  out:
343         spin_unlock(&lslist_lock);
344         return ls;
345 }
346
347 void dlm_put_lockspace(struct dlm_ls *ls)
348 {
349         spin_lock(&lslist_lock);
350         ls->ls_count--;
351         spin_unlock(&lslist_lock);
352 }
353
354 static void remove_lockspace(struct dlm_ls *ls)
355 {
356         for (;;) {
357                 spin_lock(&lslist_lock);
358                 if (ls->ls_count == 0) {
359                         list_del(&ls->ls_list);
360                         spin_unlock(&lslist_lock);
361                         return;
362                 }
363                 spin_unlock(&lslist_lock);
364                 ssleep(1);
365         }
366 }
367
368 static int threads_start(void)
369 {
370         int error;
371
372         /* Thread which process lock requests for all lockspace's */
373         error = dlm_astd_start();
374         if (error) {
375                 log_print("cannot start dlm_astd thread %d", error);
376                 goto fail;
377         }
378
379         error = dlm_scand_start();
380         if (error) {
381                 log_print("cannot start dlm_scand thread %d", error);
382                 goto astd_fail;
383         }
384
385         /* Thread for sending/receiving messages for all lockspace's */
386         error = dlm_lowcomms_start();
387         if (error) {
388                 log_print("cannot start dlm lowcomms %d", error);
389                 goto scand_fail;
390         }
391
392         return 0;
393
394  scand_fail:
395         dlm_scand_stop();
396  astd_fail:
397         dlm_astd_stop();
398  fail:
399         return error;
400 }
401
402 static void threads_stop(void)
403 {
404         dlm_scand_stop();
405         dlm_lowcomms_stop();
406         dlm_astd_stop();
407 }
408
409 static int new_lockspace(char *name, int namelen, void **lockspace,
410                          uint32_t flags, int lvblen)
411 {
412         struct dlm_ls *ls;
413         int i, size, error = -ENOMEM;
414         int do_unreg = 0;
415
416         if (namelen > DLM_LOCKSPACE_LEN)
417                 return -EINVAL;
418
419         if (!lvblen || (lvblen % 8))
420                 return -EINVAL;
421
422         if (!try_module_get(THIS_MODULE))
423                 return -EINVAL;
424
425         ls = dlm_find_lockspace_name(name, namelen);
426         if (ls) {
427                 *lockspace = ls;
428                 module_put(THIS_MODULE);
429                 return -EEXIST;
430         }
431
432         ls = kzalloc(sizeof(struct dlm_ls) + namelen, GFP_KERNEL);
433         if (!ls)
434                 goto out;
435         memcpy(ls->ls_name, name, namelen);
436         ls->ls_namelen = namelen;
437         ls->ls_lvblen = lvblen;
438         ls->ls_count = 0;
439         ls->ls_flags = 0;
440
441         if (flags & DLM_LSFL_TIMEWARN)
442                 set_bit(LSFL_TIMEWARN, &ls->ls_flags);
443
444         if (flags & DLM_LSFL_FS)
445                 ls->ls_allocation = GFP_NOFS;
446         else
447                 ls->ls_allocation = GFP_KERNEL;
448
449         /* ls_exflags are forced to match among nodes, and we don't
450            need to require all nodes to have TIMEWARN or FS set */
451         ls->ls_exflags = (flags & ~(DLM_LSFL_TIMEWARN | DLM_LSFL_FS));
452
453         size = dlm_config.ci_rsbtbl_size;
454         ls->ls_rsbtbl_size = size;
455
456         ls->ls_rsbtbl = kmalloc(sizeof(struct dlm_rsbtable) * size, GFP_KERNEL);
457         if (!ls->ls_rsbtbl)
458                 goto out_lsfree;
459         for (i = 0; i < size; i++) {
460                 INIT_LIST_HEAD(&ls->ls_rsbtbl[i].list);
461                 INIT_LIST_HEAD(&ls->ls_rsbtbl[i].toss);
462                 rwlock_init(&ls->ls_rsbtbl[i].lock);
463         }
464
465         size = dlm_config.ci_lkbtbl_size;
466         ls->ls_lkbtbl_size = size;
467
468         ls->ls_lkbtbl = kmalloc(sizeof(struct dlm_lkbtable) * size, GFP_KERNEL);
469         if (!ls->ls_lkbtbl)
470                 goto out_rsbfree;
471         for (i = 0; i < size; i++) {
472                 INIT_LIST_HEAD(&ls->ls_lkbtbl[i].list);
473                 rwlock_init(&ls->ls_lkbtbl[i].lock);
474                 ls->ls_lkbtbl[i].counter = 1;
475         }
476
477         size = dlm_config.ci_dirtbl_size;
478         ls->ls_dirtbl_size = size;
479
480         ls->ls_dirtbl = kmalloc(sizeof(struct dlm_dirtable) * size, GFP_KERNEL);
481         if (!ls->ls_dirtbl)
482                 goto out_lkbfree;
483         for (i = 0; i < size; i++) {
484                 INIT_LIST_HEAD(&ls->ls_dirtbl[i].list);
485                 rwlock_init(&ls->ls_dirtbl[i].lock);
486         }
487
488         INIT_LIST_HEAD(&ls->ls_waiters);
489         mutex_init(&ls->ls_waiters_mutex);
490         INIT_LIST_HEAD(&ls->ls_orphans);
491         mutex_init(&ls->ls_orphans_mutex);
492         INIT_LIST_HEAD(&ls->ls_timeout);
493         mutex_init(&ls->ls_timeout_mutex);
494
495         INIT_LIST_HEAD(&ls->ls_nodes);
496         INIT_LIST_HEAD(&ls->ls_nodes_gone);
497         ls->ls_num_nodes = 0;
498         ls->ls_low_nodeid = 0;
499         ls->ls_total_weight = 0;
500         ls->ls_node_array = NULL;
501
502         memset(&ls->ls_stub_rsb, 0, sizeof(struct dlm_rsb));
503         ls->ls_stub_rsb.res_ls = ls;
504
505         ls->ls_debug_rsb_dentry = NULL;
506         ls->ls_debug_waiters_dentry = NULL;
507
508         init_waitqueue_head(&ls->ls_uevent_wait);
509         ls->ls_uevent_result = 0;
510         init_completion(&ls->ls_members_done);
511         ls->ls_members_result = -1;
512
513         ls->ls_recoverd_task = NULL;
514         mutex_init(&ls->ls_recoverd_active);
515         spin_lock_init(&ls->ls_recover_lock);
516         spin_lock_init(&ls->ls_rcom_spin);
517         get_random_bytes(&ls->ls_rcom_seq, sizeof(uint64_t));
518         ls->ls_recover_status = 0;
519         ls->ls_recover_seq = 0;
520         ls->ls_recover_args = NULL;
521         init_rwsem(&ls->ls_in_recovery);
522         INIT_LIST_HEAD(&ls->ls_requestqueue);
523         mutex_init(&ls->ls_requestqueue_mutex);
524         mutex_init(&ls->ls_clear_proc_locks);
525
526         ls->ls_recover_buf = kmalloc(dlm_config.ci_buffer_size, GFP_KERNEL);
527         if (!ls->ls_recover_buf)
528                 goto out_dirfree;
529
530         INIT_LIST_HEAD(&ls->ls_recover_list);
531         spin_lock_init(&ls->ls_recover_list_lock);
532         ls->ls_recover_list_count = 0;
533         ls->ls_local_handle = ls;
534         init_waitqueue_head(&ls->ls_wait_general);
535         INIT_LIST_HEAD(&ls->ls_root_list);
536         init_rwsem(&ls->ls_root_sem);
537
538         down_write(&ls->ls_in_recovery);
539
540         spin_lock(&lslist_lock);
541         list_add(&ls->ls_list, &lslist);
542         spin_unlock(&lslist_lock);
543
544         /* needs to find ls in lslist */
545         error = dlm_recoverd_start(ls);
546         if (error) {
547                 log_error(ls, "can't start dlm_recoverd %d", error);
548                 goto out_delist;
549         }
550
551         error = kobject_setup(ls);
552         if (error)
553                 goto out_stop;
554
555         error = kobject_register(&ls->ls_kobj);
556         if (error)
557                 goto out_stop;
558
559         /* let kobject handle freeing of ls if there's an error */
560         do_unreg = 1;
561
562         /* This uevent triggers dlm_controld in userspace to add us to the
563            group of nodes that are members of this lockspace (managed by the
564            cluster infrastructure.)  Once it's done that, it tells us who the
565            current lockspace members are (via configfs) and then tells the
566            lockspace to start running (via sysfs) in dlm_ls_start(). */
567
568         error = do_uevent(ls, 1);
569         if (error)
570                 goto out_stop;
571
572         wait_for_completion(&ls->ls_members_done);
573         error = ls->ls_members_result;
574         if (error)
575                 goto out_members;
576
577         dlm_create_debug_file(ls);
578
579         log_debug(ls, "join complete");
580
581         *lockspace = ls;
582         return 0;
583
584  out_members:
585         do_uevent(ls, 0);
586         dlm_clear_members(ls);
587         kfree(ls->ls_node_array);
588  out_stop:
589         dlm_recoverd_stop(ls);
590  out_delist:
591         spin_lock(&lslist_lock);
592         list_del(&ls->ls_list);
593         spin_unlock(&lslist_lock);
594         kfree(ls->ls_recover_buf);
595  out_dirfree:
596         kfree(ls->ls_dirtbl);
597  out_lkbfree:
598         kfree(ls->ls_lkbtbl);
599  out_rsbfree:
600         kfree(ls->ls_rsbtbl);
601  out_lsfree:
602         if (do_unreg)
603                 kobject_unregister(&ls->ls_kobj);
604         else
605                 kfree(ls);
606  out:
607         module_put(THIS_MODULE);
608         return error;
609 }
610
611 int dlm_new_lockspace(char *name, int namelen, void **lockspace,
612                       uint32_t flags, int lvblen)
613 {
614         int error = 0;
615
616         mutex_lock(&ls_lock);
617         if (!ls_count)
618                 error = threads_start();
619         if (error)
620                 goto out;
621
622         error = new_lockspace(name, namelen, lockspace, flags, lvblen);
623         if (!error)
624                 ls_count++;
625         else if (!ls_count)
626                 threads_stop();
627  out:
628         mutex_unlock(&ls_lock);
629         return error;
630 }
631
632 /* Return 1 if the lockspace still has active remote locks,
633  *        2 if the lockspace still has active local locks.
634  */
635 static int lockspace_busy(struct dlm_ls *ls)
636 {
637         int i, lkb_found = 0;
638         struct dlm_lkb *lkb;
639
640         /* NOTE: We check the lockidtbl here rather than the resource table.
641            This is because there may be LKBs queued as ASTs that have been
642            unlinked from their RSBs and are pending deletion once the AST has
643            been delivered */
644
645         for (i = 0; i < ls->ls_lkbtbl_size; i++) {
646                 read_lock(&ls->ls_lkbtbl[i].lock);
647                 if (!list_empty(&ls->ls_lkbtbl[i].list)) {
648                         lkb_found = 1;
649                         list_for_each_entry(lkb, &ls->ls_lkbtbl[i].list,
650                                             lkb_idtbl_list) {
651                                 if (!lkb->lkb_nodeid) {
652                                         read_unlock(&ls->ls_lkbtbl[i].lock);
653                                         return 2;
654                                 }
655                         }
656                 }
657                 read_unlock(&ls->ls_lkbtbl[i].lock);
658         }
659         return lkb_found;
660 }
661
662 static int release_lockspace(struct dlm_ls *ls, int force)
663 {
664         struct dlm_lkb *lkb;
665         struct dlm_rsb *rsb;
666         struct list_head *head;
667         int i;
668         int busy = lockspace_busy(ls);
669
670         if (busy > force)
671                 return -EBUSY;
672
673         if (force < 3)
674                 do_uevent(ls, 0);
675
676         dlm_recoverd_stop(ls);
677
678         remove_lockspace(ls);
679
680         dlm_delete_debug_file(ls);
681
682         dlm_astd_suspend();
683
684         kfree(ls->ls_recover_buf);
685
686         /*
687          * Free direntry structs.
688          */
689
690         dlm_dir_clear(ls);
691         kfree(ls->ls_dirtbl);
692
693         /*
694          * Free all lkb's on lkbtbl[] lists.
695          */
696
697         for (i = 0; i < ls->ls_lkbtbl_size; i++) {
698                 head = &ls->ls_lkbtbl[i].list;
699                 while (!list_empty(head)) {
700                         lkb = list_entry(head->next, struct dlm_lkb,
701                                          lkb_idtbl_list);
702
703                         list_del(&lkb->lkb_idtbl_list);
704
705                         dlm_del_ast(lkb);
706
707                         if (lkb->lkb_lvbptr && lkb->lkb_flags & DLM_IFL_MSTCPY)
708                                 free_lvb(lkb->lkb_lvbptr);
709
710                         free_lkb(lkb);
711                 }
712         }
713         dlm_astd_resume();
714
715         kfree(ls->ls_lkbtbl);
716
717         /*
718          * Free all rsb's on rsbtbl[] lists
719          */
720
721         for (i = 0; i < ls->ls_rsbtbl_size; i++) {
722                 head = &ls->ls_rsbtbl[i].list;
723                 while (!list_empty(head)) {
724                         rsb = list_entry(head->next, struct dlm_rsb,
725                                          res_hashchain);
726
727                         list_del(&rsb->res_hashchain);
728                         free_rsb(rsb);
729                 }
730
731                 head = &ls->ls_rsbtbl[i].toss;
732                 while (!list_empty(head)) {
733                         rsb = list_entry(head->next, struct dlm_rsb,
734                                          res_hashchain);
735                         list_del(&rsb->res_hashchain);
736                         free_rsb(rsb);
737                 }
738         }
739
740         kfree(ls->ls_rsbtbl);
741
742         /*
743          * Free structures on any other lists
744          */
745
746         dlm_purge_requestqueue(ls);
747         kfree(ls->ls_recover_args);
748         dlm_clear_free_entries(ls);
749         dlm_clear_members(ls);
750         dlm_clear_members_gone(ls);
751         kfree(ls->ls_node_array);
752         kobject_unregister(&ls->ls_kobj);
753         /* The ls structure will be freed when the kobject is done with */
754
755         mutex_lock(&ls_lock);
756         ls_count--;
757         if (!ls_count)
758                 threads_stop();
759         mutex_unlock(&ls_lock);
760
761         module_put(THIS_MODULE);
762         return 0;
763 }
764
765 /*
766  * Called when a system has released all its locks and is not going to use the
767  * lockspace any longer.  We free everything we're managing for this lockspace.
768  * Remaining nodes will go through the recovery process as if we'd died.  The
769  * lockspace must continue to function as usual, participating in recoveries,
770  * until this returns.
771  *
772  * Force has 4 possible values:
773  * 0 - don't destroy locksapce if it has any LKBs
774  * 1 - destroy lockspace if it has remote LKBs but not if it has local LKBs
775  * 2 - destroy lockspace regardless of LKBs
776  * 3 - destroy lockspace as part of a forced shutdown
777  */
778
779 int dlm_release_lockspace(void *lockspace, int force)
780 {
781         struct dlm_ls *ls;
782
783         ls = dlm_find_lockspace_local(lockspace);
784         if (!ls)
785                 return -EINVAL;
786         dlm_put_lockspace(ls);
787         return release_lockspace(ls, force);
788 }
789