Pull trivial into test branch
[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-2005 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         .subsys = &kernel_subsys,
171         .kobj   = {.name = "dlm",},
172         .ktype  = &dlm_ktype,
173 };
174
175 static int kobject_setup(struct dlm_ls *ls)
176 {
177         char lsname[DLM_LOCKSPACE_LEN];
178         int error;
179
180         memset(lsname, 0, DLM_LOCKSPACE_LEN);
181         snprintf(lsname, DLM_LOCKSPACE_LEN, "%s", ls->ls_name);
182
183         error = kobject_set_name(&ls->ls_kobj, "%s", lsname);
184         if (error)
185                 return error;
186
187         ls->ls_kobj.kset = &dlm_kset;
188         ls->ls_kobj.ktype = &dlm_ktype;
189         return 0;
190 }
191
192 static int do_uevent(struct dlm_ls *ls, int in)
193 {
194         int error;
195
196         if (in)
197                 kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE);
198         else
199                 kobject_uevent(&ls->ls_kobj, KOBJ_OFFLINE);
200
201         error = wait_event_interruptible(ls->ls_uevent_wait,
202                         test_and_clear_bit(LSFL_UEVENT_WAIT, &ls->ls_flags));
203         if (error)
204                 goto out;
205
206         error = ls->ls_uevent_result;
207  out:
208         return error;
209 }
210
211
212 int dlm_lockspace_init(void)
213 {
214         int error;
215
216         ls_count = 0;
217         mutex_init(&ls_lock);
218         INIT_LIST_HEAD(&lslist);
219         spin_lock_init(&lslist_lock);
220
221         error = kset_register(&dlm_kset);
222         if (error)
223                 printk("dlm_lockspace_init: cannot register kset %d\n", error);
224         return error;
225 }
226
227 void dlm_lockspace_exit(void)
228 {
229         kset_unregister(&dlm_kset);
230 }
231
232 static int dlm_scand(void *data)
233 {
234         struct dlm_ls *ls;
235
236         while (!kthread_should_stop()) {
237                 list_for_each_entry(ls, &lslist, ls_list)
238                         dlm_scan_rsbs(ls);
239                 schedule_timeout_interruptible(dlm_config.scan_secs * HZ);
240         }
241         return 0;
242 }
243
244 static int dlm_scand_start(void)
245 {
246         struct task_struct *p;
247         int error = 0;
248
249         p = kthread_run(dlm_scand, NULL, "dlm_scand");
250         if (IS_ERR(p))
251                 error = PTR_ERR(p);
252         else
253                 scand_task = p;
254         return error;
255 }
256
257 static void dlm_scand_stop(void)
258 {
259         kthread_stop(scand_task);
260 }
261
262 static struct dlm_ls *dlm_find_lockspace_name(char *name, int namelen)
263 {
264         struct dlm_ls *ls;
265
266         spin_lock(&lslist_lock);
267
268         list_for_each_entry(ls, &lslist, ls_list) {
269                 if (ls->ls_namelen == namelen &&
270                     memcmp(ls->ls_name, name, namelen) == 0)
271                         goto out;
272         }
273         ls = NULL;
274  out:
275         spin_unlock(&lslist_lock);
276         return ls;
277 }
278
279 struct dlm_ls *dlm_find_lockspace_global(uint32_t id)
280 {
281         struct dlm_ls *ls;
282
283         spin_lock(&lslist_lock);
284
285         list_for_each_entry(ls, &lslist, ls_list) {
286                 if (ls->ls_global_id == id) {
287                         ls->ls_count++;
288                         goto out;
289                 }
290         }
291         ls = NULL;
292  out:
293         spin_unlock(&lslist_lock);
294         return ls;
295 }
296
297 struct dlm_ls *dlm_find_lockspace_local(dlm_lockspace_t *lockspace)
298 {
299         struct dlm_ls *ls;
300
301         spin_lock(&lslist_lock);
302         list_for_each_entry(ls, &lslist, ls_list) {
303                 if (ls->ls_local_handle == lockspace) {
304                         ls->ls_count++;
305                         goto out;
306                 }
307         }
308         ls = NULL;
309  out:
310         spin_unlock(&lslist_lock);
311         return ls;
312 }
313
314 struct dlm_ls *dlm_find_lockspace_device(int minor)
315 {
316         struct dlm_ls *ls;
317
318         spin_lock(&lslist_lock);
319         list_for_each_entry(ls, &lslist, ls_list) {
320                 if (ls->ls_device.minor == minor) {
321                         ls->ls_count++;
322                         goto out;
323                 }
324         }
325         ls = NULL;
326  out:
327         spin_unlock(&lslist_lock);
328         return ls;
329 }
330
331 void dlm_put_lockspace(struct dlm_ls *ls)
332 {
333         spin_lock(&lslist_lock);
334         ls->ls_count--;
335         spin_unlock(&lslist_lock);
336 }
337
338 static void remove_lockspace(struct dlm_ls *ls)
339 {
340         for (;;) {
341                 spin_lock(&lslist_lock);
342                 if (ls->ls_count == 0) {
343                         list_del(&ls->ls_list);
344                         spin_unlock(&lslist_lock);
345                         return;
346                 }
347                 spin_unlock(&lslist_lock);
348                 ssleep(1);
349         }
350 }
351
352 static int threads_start(void)
353 {
354         int error;
355
356         /* Thread which process lock requests for all lockspace's */
357         error = dlm_astd_start();
358         if (error) {
359                 log_print("cannot start dlm_astd thread %d", error);
360                 goto fail;
361         }
362
363         error = dlm_scand_start();
364         if (error) {
365                 log_print("cannot start dlm_scand thread %d", error);
366                 goto astd_fail;
367         }
368
369         /* Thread for sending/receiving messages for all lockspace's */
370         error = dlm_lowcomms_start();
371         if (error) {
372                 log_print("cannot start dlm lowcomms %d", error);
373                 goto scand_fail;
374         }
375
376         return 0;
377
378  scand_fail:
379         dlm_scand_stop();
380  astd_fail:
381         dlm_astd_stop();
382  fail:
383         return error;
384 }
385
386 static void threads_stop(void)
387 {
388         dlm_scand_stop();
389         dlm_lowcomms_stop();
390         dlm_astd_stop();
391 }
392
393 static int new_lockspace(char *name, int namelen, void **lockspace,
394                          uint32_t flags, int lvblen)
395 {
396         struct dlm_ls *ls;
397         int i, size, error = -ENOMEM;
398
399         if (namelen > DLM_LOCKSPACE_LEN)
400                 return -EINVAL;
401
402         if (!lvblen || (lvblen % 8))
403                 return -EINVAL;
404
405         if (!try_module_get(THIS_MODULE))
406                 return -EINVAL;
407
408         ls = dlm_find_lockspace_name(name, namelen);
409         if (ls) {
410                 *lockspace = ls;
411                 module_put(THIS_MODULE);
412                 return -EEXIST;
413         }
414
415         ls = kzalloc(sizeof(struct dlm_ls) + namelen, GFP_KERNEL);
416         if (!ls)
417                 goto out;
418         memcpy(ls->ls_name, name, namelen);
419         ls->ls_namelen = namelen;
420         ls->ls_exflags = flags;
421         ls->ls_lvblen = lvblen;
422         ls->ls_count = 0;
423         ls->ls_flags = 0;
424
425         size = dlm_config.rsbtbl_size;
426         ls->ls_rsbtbl_size = size;
427
428         ls->ls_rsbtbl = kmalloc(sizeof(struct dlm_rsbtable) * size, GFP_KERNEL);
429         if (!ls->ls_rsbtbl)
430                 goto out_lsfree;
431         for (i = 0; i < size; i++) {
432                 INIT_LIST_HEAD(&ls->ls_rsbtbl[i].list);
433                 INIT_LIST_HEAD(&ls->ls_rsbtbl[i].toss);
434                 rwlock_init(&ls->ls_rsbtbl[i].lock);
435         }
436
437         size = dlm_config.lkbtbl_size;
438         ls->ls_lkbtbl_size = size;
439
440         ls->ls_lkbtbl = kmalloc(sizeof(struct dlm_lkbtable) * size, GFP_KERNEL);
441         if (!ls->ls_lkbtbl)
442                 goto out_rsbfree;
443         for (i = 0; i < size; i++) {
444                 INIT_LIST_HEAD(&ls->ls_lkbtbl[i].list);
445                 rwlock_init(&ls->ls_lkbtbl[i].lock);
446                 ls->ls_lkbtbl[i].counter = 1;
447         }
448
449         size = dlm_config.dirtbl_size;
450         ls->ls_dirtbl_size = size;
451
452         ls->ls_dirtbl = kmalloc(sizeof(struct dlm_dirtable) * size, GFP_KERNEL);
453         if (!ls->ls_dirtbl)
454                 goto out_lkbfree;
455         for (i = 0; i < size; i++) {
456                 INIT_LIST_HEAD(&ls->ls_dirtbl[i].list);
457                 rwlock_init(&ls->ls_dirtbl[i].lock);
458         }
459
460         INIT_LIST_HEAD(&ls->ls_waiters);
461         mutex_init(&ls->ls_waiters_mutex);
462
463         INIT_LIST_HEAD(&ls->ls_nodes);
464         INIT_LIST_HEAD(&ls->ls_nodes_gone);
465         ls->ls_num_nodes = 0;
466         ls->ls_low_nodeid = 0;
467         ls->ls_total_weight = 0;
468         ls->ls_node_array = NULL;
469
470         memset(&ls->ls_stub_rsb, 0, sizeof(struct dlm_rsb));
471         ls->ls_stub_rsb.res_ls = ls;
472
473         ls->ls_debug_rsb_dentry = NULL;
474         ls->ls_debug_waiters_dentry = NULL;
475
476         init_waitqueue_head(&ls->ls_uevent_wait);
477         ls->ls_uevent_result = 0;
478
479         ls->ls_recoverd_task = NULL;
480         mutex_init(&ls->ls_recoverd_active);
481         spin_lock_init(&ls->ls_recover_lock);
482         spin_lock_init(&ls->ls_rcom_spin);
483         get_random_bytes(&ls->ls_rcom_seq, sizeof(uint64_t));
484         ls->ls_recover_status = 0;
485         ls->ls_recover_seq = 0;
486         ls->ls_recover_args = NULL;
487         init_rwsem(&ls->ls_in_recovery);
488         INIT_LIST_HEAD(&ls->ls_requestqueue);
489         mutex_init(&ls->ls_requestqueue_mutex);
490         mutex_init(&ls->ls_clear_proc_locks);
491
492         ls->ls_recover_buf = kmalloc(dlm_config.buffer_size, GFP_KERNEL);
493         if (!ls->ls_recover_buf)
494                 goto out_dirfree;
495
496         INIT_LIST_HEAD(&ls->ls_recover_list);
497         spin_lock_init(&ls->ls_recover_list_lock);
498         ls->ls_recover_list_count = 0;
499         ls->ls_local_handle = ls;
500         init_waitqueue_head(&ls->ls_wait_general);
501         INIT_LIST_HEAD(&ls->ls_root_list);
502         init_rwsem(&ls->ls_root_sem);
503
504         down_write(&ls->ls_in_recovery);
505
506         spin_lock(&lslist_lock);
507         list_add(&ls->ls_list, &lslist);
508         spin_unlock(&lslist_lock);
509
510         /* needs to find ls in lslist */
511         error = dlm_recoverd_start(ls);
512         if (error) {
513                 log_error(ls, "can't start dlm_recoverd %d", error);
514                 goto out_rcomfree;
515         }
516
517         dlm_create_debug_file(ls);
518
519         error = kobject_setup(ls);
520         if (error)
521                 goto out_del;
522
523         error = kobject_register(&ls->ls_kobj);
524         if (error)
525                 goto out_del;
526
527         error = do_uevent(ls, 1);
528         if (error)
529                 goto out_unreg;
530
531         *lockspace = ls;
532         return 0;
533
534  out_unreg:
535         kobject_unregister(&ls->ls_kobj);
536  out_del:
537         dlm_delete_debug_file(ls);
538         dlm_recoverd_stop(ls);
539  out_rcomfree:
540         spin_lock(&lslist_lock);
541         list_del(&ls->ls_list);
542         spin_unlock(&lslist_lock);
543         kfree(ls->ls_recover_buf);
544  out_dirfree:
545         kfree(ls->ls_dirtbl);
546  out_lkbfree:
547         kfree(ls->ls_lkbtbl);
548  out_rsbfree:
549         kfree(ls->ls_rsbtbl);
550  out_lsfree:
551         kfree(ls);
552  out:
553         module_put(THIS_MODULE);
554         return error;
555 }
556
557 int dlm_new_lockspace(char *name, int namelen, void **lockspace,
558                       uint32_t flags, int lvblen)
559 {
560         int error = 0;
561
562         mutex_lock(&ls_lock);
563         if (!ls_count)
564                 error = threads_start();
565         if (error)
566                 goto out;
567
568         error = new_lockspace(name, namelen, lockspace, flags, lvblen);
569         if (!error)
570                 ls_count++;
571  out:
572         mutex_unlock(&ls_lock);
573         return error;
574 }
575
576 /* Return 1 if the lockspace still has active remote locks,
577  *        2 if the lockspace still has active local locks.
578  */
579 static int lockspace_busy(struct dlm_ls *ls)
580 {
581         int i, lkb_found = 0;
582         struct dlm_lkb *lkb;
583
584         /* NOTE: We check the lockidtbl here rather than the resource table.
585            This is because there may be LKBs queued as ASTs that have been
586            unlinked from their RSBs and are pending deletion once the AST has
587            been delivered */
588
589         for (i = 0; i < ls->ls_lkbtbl_size; i++) {
590                 read_lock(&ls->ls_lkbtbl[i].lock);
591                 if (!list_empty(&ls->ls_lkbtbl[i].list)) {
592                         lkb_found = 1;
593                         list_for_each_entry(lkb, &ls->ls_lkbtbl[i].list,
594                                             lkb_idtbl_list) {
595                                 if (!lkb->lkb_nodeid) {
596                                         read_unlock(&ls->ls_lkbtbl[i].lock);
597                                         return 2;
598                                 }
599                         }
600                 }
601                 read_unlock(&ls->ls_lkbtbl[i].lock);
602         }
603         return lkb_found;
604 }
605
606 static int release_lockspace(struct dlm_ls *ls, int force)
607 {
608         struct dlm_lkb *lkb;
609         struct dlm_rsb *rsb;
610         struct list_head *head;
611         int i;
612         int busy = lockspace_busy(ls);
613
614         if (busy > force)
615                 return -EBUSY;
616
617         if (force < 3)
618                 do_uevent(ls, 0);
619
620         dlm_recoverd_stop(ls);
621
622         remove_lockspace(ls);
623
624         dlm_delete_debug_file(ls);
625
626         dlm_astd_suspend();
627
628         kfree(ls->ls_recover_buf);
629
630         /*
631          * Free direntry structs.
632          */
633
634         dlm_dir_clear(ls);
635         kfree(ls->ls_dirtbl);
636
637         /*
638          * Free all lkb's on lkbtbl[] lists.
639          */
640
641         for (i = 0; i < ls->ls_lkbtbl_size; i++) {
642                 head = &ls->ls_lkbtbl[i].list;
643                 while (!list_empty(head)) {
644                         lkb = list_entry(head->next, struct dlm_lkb,
645                                          lkb_idtbl_list);
646
647                         list_del(&lkb->lkb_idtbl_list);
648
649                         dlm_del_ast(lkb);
650
651                         if (lkb->lkb_lvbptr && lkb->lkb_flags & DLM_IFL_MSTCPY)
652                                 free_lvb(lkb->lkb_lvbptr);
653
654                         free_lkb(lkb);
655                 }
656         }
657         dlm_astd_resume();
658
659         kfree(ls->ls_lkbtbl);
660
661         /*
662          * Free all rsb's on rsbtbl[] lists
663          */
664
665         for (i = 0; i < ls->ls_rsbtbl_size; i++) {
666                 head = &ls->ls_rsbtbl[i].list;
667                 while (!list_empty(head)) {
668                         rsb = list_entry(head->next, struct dlm_rsb,
669                                          res_hashchain);
670
671                         list_del(&rsb->res_hashchain);
672                         free_rsb(rsb);
673                 }
674
675                 head = &ls->ls_rsbtbl[i].toss;
676                 while (!list_empty(head)) {
677                         rsb = list_entry(head->next, struct dlm_rsb,
678                                          res_hashchain);
679                         list_del(&rsb->res_hashchain);
680                         free_rsb(rsb);
681                 }
682         }
683
684         kfree(ls->ls_rsbtbl);
685
686         /*
687          * Free structures on any other lists
688          */
689
690         dlm_purge_requestqueue(ls);
691         kfree(ls->ls_recover_args);
692         dlm_clear_free_entries(ls);
693         dlm_clear_members(ls);
694         dlm_clear_members_gone(ls);
695         kfree(ls->ls_node_array);
696         kobject_unregister(&ls->ls_kobj);
697         /* The ls structure will be freed when the kobject is done with */
698
699         mutex_lock(&ls_lock);
700         ls_count--;
701         if (!ls_count)
702                 threads_stop();
703         mutex_unlock(&ls_lock);
704
705         module_put(THIS_MODULE);
706         return 0;
707 }
708
709 /*
710  * Called when a system has released all its locks and is not going to use the
711  * lockspace any longer.  We free everything we're managing for this lockspace.
712  * Remaining nodes will go through the recovery process as if we'd died.  The
713  * lockspace must continue to function as usual, participating in recoveries,
714  * until this returns.
715  *
716  * Force has 4 possible values:
717  * 0 - don't destroy locksapce if it has any LKBs
718  * 1 - destroy lockspace if it has remote LKBs but not if it has local LKBs
719  * 2 - destroy lockspace regardless of LKBs
720  * 3 - destroy lockspace as part of a forced shutdown
721  */
722
723 int dlm_release_lockspace(void *lockspace, int force)
724 {
725         struct dlm_ls *ls;
726
727         ls = dlm_find_lockspace_local(lockspace);
728         if (!ls)
729                 return -EINVAL;
730         dlm_put_lockspace(ls);
731         return release_lockspace(ls, force);
732 }
733