~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

TOMOYO Linux Cross Reference
Linux/fs/dlm/lockspace.c

Version: ~ [ linux-5.3-rc5 ] ~ [ linux-5.2.9 ] ~ [ linux-5.1.21 ] ~ [ linux-5.0.21 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.67 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.139 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.189 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.189 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.19.8 ] ~ [ linux-3.18.140 ] ~ [ linux-3.17.8 ] ~ [ linux-3.16.72 ] ~ [ linux-3.15.10 ] ~ [ linux-3.14.79 ] ~ [ linux-3.13.11 ] ~ [ linux-3.12.74 ] ~ [ linux-3.11.10 ] ~ [ linux-3.10.108 ] ~ [ linux-3.9.11 ] ~ [ linux-3.8.13 ] ~ [ linux-3.7.10 ] ~ [ linux-3.6.11 ] ~ [ linux-3.5.7 ] ~ [ linux-3.4.113 ] ~ [ linux-3.3.8 ] ~ [ linux-3.2.102 ] ~ [ linux-3.1.10 ] ~ [ linux-3.0.101 ] ~ [ linux-2.6.32.71 ] ~ [ linux-2.6.0 ] ~ [ linux-2.4.37.11 ] ~ [ unix-v6-master ] ~ [ ccs-tools-1.8.5 ] ~ [ policy-sample ] ~
Architecture: ~ [ i386 ] ~ [ alpha ] ~ [ m68k ] ~ [ mips ] ~ [ ppc ] ~ [ sparc ] ~ [ sparc64 ] ~

  1 /******************************************************************************
  2 *******************************************************************************
  3 **
  4 **  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
  5 **  Copyright (C) 2004-2011 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 <linux/module.h>
 15 
 16 #include "dlm_internal.h"
 17 #include "lockspace.h"
 18 #include "member.h"
 19 #include "recoverd.h"
 20 #include "dir.h"
 21 #include "lowcomms.h"
 22 #include "config.h"
 23 #include "memory.h"
 24 #include "lock.h"
 25 #include "recover.h"
 26 #include "requestqueue.h"
 27 #include "user.h"
 28 #include "ast.h"
 29 
 30 static int                      ls_count;
 31 static struct mutex             ls_lock;
 32 static struct list_head         lslist;
 33 static spinlock_t               lslist_lock;
 34 static struct task_struct *     scand_task;
 35 
 36 
 37 static ssize_t dlm_control_store(struct dlm_ls *ls, const char *buf, size_t len)
 38 {
 39         ssize_t ret = len;
 40         int n;
 41         int rc = kstrtoint(buf, 0, &n);
 42 
 43         if (rc)
 44                 return rc;
 45         ls = dlm_find_lockspace_local(ls->ls_local_handle);
 46         if (!ls)
 47                 return -EINVAL;
 48 
 49         switch (n) {
 50         case 0:
 51                 dlm_ls_stop(ls);
 52                 break;
 53         case 1:
 54                 dlm_ls_start(ls);
 55                 break;
 56         default:
 57                 ret = -EINVAL;
 58         }
 59         dlm_put_lockspace(ls);
 60         return ret;
 61 }
 62 
 63 static ssize_t dlm_event_store(struct dlm_ls *ls, const char *buf, size_t len)
 64 {
 65         int rc = kstrtoint(buf, 0, &ls->ls_uevent_result);
 66 
 67         if (rc)
 68                 return rc;
 69         set_bit(LSFL_UEVENT_WAIT, &ls->ls_flags);
 70         wake_up(&ls->ls_uevent_wait);
 71         return len;
 72 }
 73 
 74 static ssize_t dlm_id_show(struct dlm_ls *ls, char *buf)
 75 {
 76         return snprintf(buf, PAGE_SIZE, "%u\n", ls->ls_global_id);
 77 }
 78 
 79 static ssize_t dlm_id_store(struct dlm_ls *ls, const char *buf, size_t len)
 80 {
 81         int rc = kstrtouint(buf, 0, &ls->ls_global_id);
 82 
 83         if (rc)
 84                 return rc;
 85         return len;
 86 }
 87 
 88 static ssize_t dlm_nodir_show(struct dlm_ls *ls, char *buf)
 89 {
 90         return snprintf(buf, PAGE_SIZE, "%u\n", dlm_no_directory(ls));
 91 }
 92 
 93 static ssize_t dlm_nodir_store(struct dlm_ls *ls, const char *buf, size_t len)
 94 {
 95         int val;
 96         int rc = kstrtoint(buf, 0, &val);
 97 
 98         if (rc)
 99                 return rc;
100         if (val == 1)
101                 set_bit(LSFL_NODIR, &ls->ls_flags);
102         return len;
103 }
104 
105 static ssize_t dlm_recover_status_show(struct dlm_ls *ls, char *buf)
106 {
107         uint32_t status = dlm_recover_status(ls);
108         return snprintf(buf, PAGE_SIZE, "%x\n", status);
109 }
110 
111 static ssize_t dlm_recover_nodeid_show(struct dlm_ls *ls, char *buf)
112 {
113         return snprintf(buf, PAGE_SIZE, "%d\n", ls->ls_recover_nodeid);
114 }
115 
116 struct dlm_attr {
117         struct attribute attr;
118         ssize_t (*show)(struct dlm_ls *, char *);
119         ssize_t (*store)(struct dlm_ls *, const char *, size_t);
120 };
121 
122 static struct dlm_attr dlm_attr_control = {
123         .attr  = {.name = "control", .mode = S_IWUSR},
124         .store = dlm_control_store
125 };
126 
127 static struct dlm_attr dlm_attr_event = {
128         .attr  = {.name = "event_done", .mode = S_IWUSR},
129         .store = dlm_event_store
130 };
131 
132 static struct dlm_attr dlm_attr_id = {
133         .attr  = {.name = "id", .mode = S_IRUGO | S_IWUSR},
134         .show  = dlm_id_show,
135         .store = dlm_id_store
136 };
137 
138 static struct dlm_attr dlm_attr_nodir = {
139         .attr  = {.name = "nodir", .mode = S_IRUGO | S_IWUSR},
140         .show  = dlm_nodir_show,
141         .store = dlm_nodir_store
142 };
143 
144 static struct dlm_attr dlm_attr_recover_status = {
145         .attr  = {.name = "recover_status", .mode = S_IRUGO},
146         .show  = dlm_recover_status_show
147 };
148 
149 static struct dlm_attr dlm_attr_recover_nodeid = {
150         .attr  = {.name = "recover_nodeid", .mode = S_IRUGO},
151         .show  = dlm_recover_nodeid_show
152 };
153 
154 static struct attribute *dlm_attrs[] = {
155         &dlm_attr_control.attr,
156         &dlm_attr_event.attr,
157         &dlm_attr_id.attr,
158         &dlm_attr_nodir.attr,
159         &dlm_attr_recover_status.attr,
160         &dlm_attr_recover_nodeid.attr,
161         NULL,
162 };
163 
164 static ssize_t dlm_attr_show(struct kobject *kobj, struct attribute *attr,
165                              char *buf)
166 {
167         struct dlm_ls *ls  = container_of(kobj, struct dlm_ls, ls_kobj);
168         struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
169         return a->show ? a->show(ls, buf) : 0;
170 }
171 
172 static ssize_t dlm_attr_store(struct kobject *kobj, struct attribute *attr,
173                               const char *buf, size_t len)
174 {
175         struct dlm_ls *ls  = container_of(kobj, struct dlm_ls, ls_kobj);
176         struct dlm_attr *a = container_of(attr, struct dlm_attr, attr);
177         return a->store ? a->store(ls, buf, len) : len;
178 }
179 
180 static void lockspace_kobj_release(struct kobject *k)
181 {
182         struct dlm_ls *ls  = container_of(k, struct dlm_ls, ls_kobj);
183         kfree(ls);
184 }
185 
186 static const struct sysfs_ops dlm_attr_ops = {
187         .show  = dlm_attr_show,
188         .store = dlm_attr_store,
189 };
190 
191 static struct kobj_type dlm_ktype = {
192         .default_attrs = dlm_attrs,
193         .sysfs_ops     = &dlm_attr_ops,
194         .release       = lockspace_kobj_release,
195 };
196 
197 static struct kset *dlm_kset;
198 
199 static int do_uevent(struct dlm_ls *ls, int in)
200 {
201         int error;
202 
203         if (in)
204                 kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE);
205         else
206                 kobject_uevent(&ls->ls_kobj, KOBJ_OFFLINE);
207 
208         log_rinfo(ls, "%s the lockspace group...", in ? "joining" : "leaving");
209 
210         /* dlm_controld will see the uevent, do the necessary group management
211            and then write to sysfs to wake us */
212 
213         error = wait_event_interruptible(ls->ls_uevent_wait,
214                         test_and_clear_bit(LSFL_UEVENT_WAIT, &ls->ls_flags));
215 
216         log_rinfo(ls, "group event done %d %d", error, ls->ls_uevent_result);
217 
218         if (error)
219                 goto out;
220 
221         error = ls->ls_uevent_result;
222  out:
223         if (error)
224                 log_error(ls, "group %s failed %d %d", in ? "join" : "leave",
225                           error, ls->ls_uevent_result);
226         return error;
227 }
228 
229 static int dlm_uevent(struct kset *kset, struct kobject *kobj,
230                       struct kobj_uevent_env *env)
231 {
232         struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
233 
234         add_uevent_var(env, "LOCKSPACE=%s", ls->ls_name);
235         return 0;
236 }
237 
238 static struct kset_uevent_ops dlm_uevent_ops = {
239         .uevent = dlm_uevent,
240 };
241 
242 int __init dlm_lockspace_init(void)
243 {
244         ls_count = 0;
245         mutex_init(&ls_lock);
246         INIT_LIST_HEAD(&lslist);
247         spin_lock_init(&lslist_lock);
248 
249         dlm_kset = kset_create_and_add("dlm", &dlm_uevent_ops, kernel_kobj);
250         if (!dlm_kset) {
251                 printk(KERN_WARNING "%s: can not create kset\n", __func__);
252                 return -ENOMEM;
253         }
254         return 0;
255 }
256 
257 void dlm_lockspace_exit(void)
258 {
259         kset_unregister(dlm_kset);
260 }
261 
262 static struct dlm_ls *find_ls_to_scan(void)
263 {
264         struct dlm_ls *ls;
265 
266         spin_lock(&lslist_lock);
267         list_for_each_entry(ls, &lslist, ls_list) {
268                 if (time_after_eq(jiffies, ls->ls_scan_time +
269                                             dlm_config.ci_scan_secs * HZ)) {
270                         spin_unlock(&lslist_lock);
271                         return ls;
272                 }
273         }
274         spin_unlock(&lslist_lock);
275         return NULL;
276 }
277 
278 static int dlm_scand(void *data)
279 {
280         struct dlm_ls *ls;
281 
282         while (!kthread_should_stop()) {
283                 ls = find_ls_to_scan();
284                 if (ls) {
285                         if (dlm_lock_recovery_try(ls)) {
286                                 ls->ls_scan_time = jiffies;
287                                 dlm_scan_rsbs(ls);
288                                 dlm_scan_timeout(ls);
289                                 dlm_scan_waiters(ls);
290                                 dlm_unlock_recovery(ls);
291                         } else {
292                                 ls->ls_scan_time += HZ;
293                         }
294                         continue;
295                 }
296                 schedule_timeout_interruptible(dlm_config.ci_scan_secs * HZ);
297         }
298         return 0;
299 }
300 
301 static int dlm_scand_start(void)
302 {
303         struct task_struct *p;
304         int error = 0;
305 
306         p = kthread_run(dlm_scand, NULL, "dlm_scand");
307         if (IS_ERR(p))
308                 error = PTR_ERR(p);
309         else
310                 scand_task = p;
311         return error;
312 }
313 
314 static void dlm_scand_stop(void)
315 {
316         kthread_stop(scand_task);
317 }
318 
319 struct dlm_ls *dlm_find_lockspace_global(uint32_t id)
320 {
321         struct dlm_ls *ls;
322 
323         spin_lock(&lslist_lock);
324 
325         list_for_each_entry(ls, &lslist, ls_list) {
326                 if (ls->ls_global_id == id) {
327                         ls->ls_count++;
328                         goto out;
329                 }
330         }
331         ls = NULL;
332  out:
333         spin_unlock(&lslist_lock);
334         return ls;
335 }
336 
337 struct dlm_ls *dlm_find_lockspace_local(dlm_lockspace_t *lockspace)
338 {
339         struct dlm_ls *ls;
340 
341         spin_lock(&lslist_lock);
342         list_for_each_entry(ls, &lslist, ls_list) {
343                 if (ls->ls_local_handle == lockspace) {
344                         ls->ls_count++;
345                         goto out;
346                 }
347         }
348         ls = NULL;
349  out:
350         spin_unlock(&lslist_lock);
351         return ls;
352 }
353 
354 struct dlm_ls *dlm_find_lockspace_device(int minor)
355 {
356         struct dlm_ls *ls;
357 
358         spin_lock(&lslist_lock);
359         list_for_each_entry(ls, &lslist, ls_list) {
360                 if (ls->ls_device.minor == minor) {
361                         ls->ls_count++;
362                         goto out;
363                 }
364         }
365         ls = NULL;
366  out:
367         spin_unlock(&lslist_lock);
368         return ls;
369 }
370 
371 void dlm_put_lockspace(struct dlm_ls *ls)
372 {
373         spin_lock(&lslist_lock);
374         ls->ls_count--;
375         spin_unlock(&lslist_lock);
376 }
377 
378 static void remove_lockspace(struct dlm_ls *ls)
379 {
380         for (;;) {
381                 spin_lock(&lslist_lock);
382                 if (ls->ls_count == 0) {
383                         WARN_ON(ls->ls_create_count != 0);
384                         list_del(&ls->ls_list);
385                         spin_unlock(&lslist_lock);
386                         return;
387                 }
388                 spin_unlock(&lslist_lock);
389                 ssleep(1);
390         }
391 }
392 
393 static int threads_start(void)
394 {
395         int error;
396 
397         error = dlm_scand_start();
398         if (error) {
399                 log_print("cannot start dlm_scand thread %d", error);
400                 goto fail;
401         }
402 
403         /* Thread for sending/receiving messages for all lockspace's */
404         error = dlm_lowcomms_start();
405         if (error) {
406                 log_print("cannot start dlm lowcomms %d", error);
407                 goto scand_fail;
408         }
409 
410         return 0;
411 
412  scand_fail:
413         dlm_scand_stop();
414  fail:
415         return error;
416 }
417 
418 static void threads_stop(void)
419 {
420         dlm_scand_stop();
421         dlm_lowcomms_stop();
422 }
423 
424 static int new_lockspace(const char *name, const char *cluster,
425                          uint32_t flags, int lvblen,
426                          const struct dlm_lockspace_ops *ops, void *ops_arg,
427                          int *ops_result, dlm_lockspace_t **lockspace)
428 {
429         struct dlm_ls *ls;
430         int i, size, error;
431         int do_unreg = 0;
432         int namelen = strlen(name);
433 
434         if (namelen > DLM_LOCKSPACE_LEN)
435                 return -EINVAL;
436 
437         if (!lvblen || (lvblen % 8))
438                 return -EINVAL;
439 
440         if (!try_module_get(THIS_MODULE))
441                 return -EINVAL;
442 
443         if (!dlm_user_daemon_available()) {
444                 log_print("dlm user daemon not available");
445                 error = -EUNATCH;
446                 goto out;
447         }
448 
449         if (ops && ops_result) {
450                 if (!dlm_config.ci_recover_callbacks)
451                         *ops_result = -EOPNOTSUPP;
452                 else
453                         *ops_result = 0;
454         }
455 
456         if (dlm_config.ci_recover_callbacks && cluster &&
457             strncmp(cluster, dlm_config.ci_cluster_name, DLM_LOCKSPACE_LEN)) {
458                 log_print("dlm cluster name %s mismatch %s",
459                           dlm_config.ci_cluster_name, cluster);
460                 error = -EBADR;
461                 goto out;
462         }
463 
464         error = 0;
465 
466         spin_lock(&lslist_lock);
467         list_for_each_entry(ls, &lslist, ls_list) {
468                 WARN_ON(ls->ls_create_count <= 0);
469                 if (ls->ls_namelen != namelen)
470                         continue;
471                 if (memcmp(ls->ls_name, name, namelen))
472                         continue;
473                 if (flags & DLM_LSFL_NEWEXCL) {
474                         error = -EEXIST;
475                         break;
476                 }
477                 ls->ls_create_count++;
478                 *lockspace = ls;
479                 error = 1;
480                 break;
481         }
482         spin_unlock(&lslist_lock);
483 
484         if (error)
485                 goto out;
486 
487         error = -ENOMEM;
488 
489         ls = kzalloc(sizeof(struct dlm_ls) + namelen, GFP_NOFS);
490         if (!ls)
491                 goto out;
492         memcpy(ls->ls_name, name, namelen);
493         ls->ls_namelen = namelen;
494         ls->ls_lvblen = lvblen;
495         ls->ls_count = 0;
496         ls->ls_flags = 0;
497         ls->ls_scan_time = jiffies;
498 
499         if (ops && dlm_config.ci_recover_callbacks) {
500                 ls->ls_ops = ops;
501                 ls->ls_ops_arg = ops_arg;
502         }
503 
504         if (flags & DLM_LSFL_TIMEWARN)
505                 set_bit(LSFL_TIMEWARN, &ls->ls_flags);
506 
507         /* ls_exflags are forced to match among nodes, and we don't
508            need to require all nodes to have some flags set */
509         ls->ls_exflags = (flags & ~(DLM_LSFL_TIMEWARN | DLM_LSFL_FS |
510                                     DLM_LSFL_NEWEXCL));
511 
512         size = dlm_config.ci_rsbtbl_size;
513         ls->ls_rsbtbl_size = size;
514 
515         ls->ls_rsbtbl = vmalloc(sizeof(struct dlm_rsbtable) * size);
516         if (!ls->ls_rsbtbl)
517                 goto out_lsfree;
518         for (i = 0; i < size; i++) {
519                 ls->ls_rsbtbl[i].keep.rb_node = NULL;
520                 ls->ls_rsbtbl[i].toss.rb_node = NULL;
521                 spin_lock_init(&ls->ls_rsbtbl[i].lock);
522         }
523 
524         spin_lock_init(&ls->ls_remove_spin);
525 
526         for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++) {
527                 ls->ls_remove_names[i] = kzalloc(DLM_RESNAME_MAXLEN+1,
528                                                  GFP_KERNEL);
529                 if (!ls->ls_remove_names[i])
530                         goto out_rsbtbl;
531         }
532 
533         idr_init(&ls->ls_lkbidr);
534         spin_lock_init(&ls->ls_lkbidr_spin);
535 
536         INIT_LIST_HEAD(&ls->ls_waiters);
537         mutex_init(&ls->ls_waiters_mutex);
538         INIT_LIST_HEAD(&ls->ls_orphans);
539         mutex_init(&ls->ls_orphans_mutex);
540         INIT_LIST_HEAD(&ls->ls_timeout);
541         mutex_init(&ls->ls_timeout_mutex);
542 
543         INIT_LIST_HEAD(&ls->ls_new_rsb);
544         spin_lock_init(&ls->ls_new_rsb_spin);
545 
546         INIT_LIST_HEAD(&ls->ls_nodes);
547         INIT_LIST_HEAD(&ls->ls_nodes_gone);
548         ls->ls_num_nodes = 0;
549         ls->ls_low_nodeid = 0;
550         ls->ls_total_weight = 0;
551         ls->ls_node_array = NULL;
552 
553         memset(&ls->ls_stub_rsb, 0, sizeof(struct dlm_rsb));
554         ls->ls_stub_rsb.res_ls = ls;
555 
556         ls->ls_debug_rsb_dentry = NULL;
557         ls->ls_debug_waiters_dentry = NULL;
558 
559         init_waitqueue_head(&ls->ls_uevent_wait);
560         ls->ls_uevent_result = 0;
561         init_completion(&ls->ls_members_done);
562         ls->ls_members_result = -1;
563 
564         mutex_init(&ls->ls_cb_mutex);
565         INIT_LIST_HEAD(&ls->ls_cb_delay);
566 
567         ls->ls_recoverd_task = NULL;
568         mutex_init(&ls->ls_recoverd_active);
569         spin_lock_init(&ls->ls_recover_lock);
570         spin_lock_init(&ls->ls_rcom_spin);
571         get_random_bytes(&ls->ls_rcom_seq, sizeof(uint64_t));
572         ls->ls_recover_status = 0;
573         ls->ls_recover_seq = 0;
574         ls->ls_recover_args = NULL;
575         init_rwsem(&ls->ls_in_recovery);
576         init_rwsem(&ls->ls_recv_active);
577         INIT_LIST_HEAD(&ls->ls_requestqueue);
578         mutex_init(&ls->ls_requestqueue_mutex);
579         mutex_init(&ls->ls_clear_proc_locks);
580 
581         ls->ls_recover_buf = kmalloc(dlm_config.ci_buffer_size, GFP_NOFS);
582         if (!ls->ls_recover_buf)
583                 goto out_lkbidr;
584 
585         ls->ls_slot = 0;
586         ls->ls_num_slots = 0;
587         ls->ls_slots_size = 0;
588         ls->ls_slots = NULL;
589 
590         INIT_LIST_HEAD(&ls->ls_recover_list);
591         spin_lock_init(&ls->ls_recover_list_lock);
592         idr_init(&ls->ls_recover_idr);
593         spin_lock_init(&ls->ls_recover_idr_lock);
594         ls->ls_recover_list_count = 0;
595         ls->ls_local_handle = ls;
596         init_waitqueue_head(&ls->ls_wait_general);
597         INIT_LIST_HEAD(&ls->ls_root_list);
598         init_rwsem(&ls->ls_root_sem);
599 
600         spin_lock(&lslist_lock);
601         ls->ls_create_count = 1;
602         list_add(&ls->ls_list, &lslist);
603         spin_unlock(&lslist_lock);
604 
605         if (flags & DLM_LSFL_FS) {
606                 error = dlm_callback_start(ls);
607                 if (error) {
608                         log_error(ls, "can't start dlm_callback %d", error);
609                         goto out_delist;
610                 }
611         }
612 
613         init_waitqueue_head(&ls->ls_recover_lock_wait);
614 
615         /*
616          * Once started, dlm_recoverd first looks for ls in lslist, then
617          * initializes ls_in_recovery as locked in "down" mode.  We need
618          * to wait for the wakeup from dlm_recoverd because in_recovery
619          * has to start out in down mode.
620          */
621 
622         error = dlm_recoverd_start(ls);
623         if (error) {
624                 log_error(ls, "can't start dlm_recoverd %d", error);
625                 goto out_callback;
626         }
627 
628         wait_event(ls->ls_recover_lock_wait,
629                    test_bit(LSFL_RECOVER_LOCK, &ls->ls_flags));
630 
631         ls->ls_kobj.kset = dlm_kset;
632         error = kobject_init_and_add(&ls->ls_kobj, &dlm_ktype, NULL,
633                                      "%s", ls->ls_name);
634         if (error)
635                 goto out_recoverd;
636         kobject_uevent(&ls->ls_kobj, KOBJ_ADD);
637 
638         /* let kobject handle freeing of ls if there's an error */
639         do_unreg = 1;
640 
641         /* This uevent triggers dlm_controld in userspace to add us to the
642            group of nodes that are members of this lockspace (managed by the
643            cluster infrastructure.)  Once it's done that, it tells us who the
644            current lockspace members are (via configfs) and then tells the
645            lockspace to start running (via sysfs) in dlm_ls_start(). */
646 
647         error = do_uevent(ls, 1);
648         if (error)
649                 goto out_recoverd;
650 
651         wait_for_completion(&ls->ls_members_done);
652         error = ls->ls_members_result;
653         if (error)
654                 goto out_members;
655 
656         dlm_create_debug_file(ls);
657 
658         log_rinfo(ls, "join complete");
659         *lockspace = ls;
660         return 0;
661 
662  out_members:
663         do_uevent(ls, 0);
664         dlm_clear_members(ls);
665         kfree(ls->ls_node_array);
666  out_recoverd:
667         dlm_recoverd_stop(ls);
668  out_callback:
669         dlm_callback_stop(ls);
670  out_delist:
671         spin_lock(&lslist_lock);
672         list_del(&ls->ls_list);
673         spin_unlock(&lslist_lock);
674         idr_destroy(&ls->ls_recover_idr);
675         kfree(ls->ls_recover_buf);
676  out_lkbidr:
677         idr_destroy(&ls->ls_lkbidr);
678         for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++) {
679                 if (ls->ls_remove_names[i])
680                         kfree(ls->ls_remove_names[i]);
681         }
682  out_rsbtbl:
683         vfree(ls->ls_rsbtbl);
684  out_lsfree:
685         if (do_unreg)
686                 kobject_put(&ls->ls_kobj);
687         else
688                 kfree(ls);
689  out:
690         module_put(THIS_MODULE);
691         return error;
692 }
693 
694 int dlm_new_lockspace(const char *name, const char *cluster,
695                       uint32_t flags, int lvblen,
696                       const struct dlm_lockspace_ops *ops, void *ops_arg,
697                       int *ops_result, dlm_lockspace_t **lockspace)
698 {
699         int error = 0;
700 
701         mutex_lock(&ls_lock);
702         if (!ls_count)
703                 error = threads_start();
704         if (error)
705                 goto out;
706 
707         error = new_lockspace(name, cluster, flags, lvblen, ops, ops_arg,
708                               ops_result, lockspace);
709         if (!error)
710                 ls_count++;
711         if (error > 0)
712                 error = 0;
713         if (!ls_count)
714                 threads_stop();
715  out:
716         mutex_unlock(&ls_lock);
717         return error;
718 }
719 
720 static int lkb_idr_is_local(int id, void *p, void *data)
721 {
722         struct dlm_lkb *lkb = p;
723 
724         return lkb->lkb_nodeid == 0 && lkb->lkb_grmode != DLM_LOCK_IV;
725 }
726 
727 static int lkb_idr_is_any(int id, void *p, void *data)
728 {
729         return 1;
730 }
731 
732 static int lkb_idr_free(int id, void *p, void *data)
733 {
734         struct dlm_lkb *lkb = p;
735 
736         if (lkb->lkb_lvbptr && lkb->lkb_flags & DLM_IFL_MSTCPY)
737                 dlm_free_lvb(lkb->lkb_lvbptr);
738 
739         dlm_free_lkb(lkb);
740         return 0;
741 }
742 
743 /* NOTE: We check the lkbidr here rather than the resource table.
744    This is because there may be LKBs queued as ASTs that have been unlinked
745    from their RSBs and are pending deletion once the AST has been delivered */
746 
747 static int lockspace_busy(struct dlm_ls *ls, int force)
748 {
749         int rv;
750 
751         spin_lock(&ls->ls_lkbidr_spin);
752         if (force == 0) {
753                 rv = idr_for_each(&ls->ls_lkbidr, lkb_idr_is_any, ls);
754         } else if (force == 1) {
755                 rv = idr_for_each(&ls->ls_lkbidr, lkb_idr_is_local, ls);
756         } else {
757                 rv = 0;
758         }
759         spin_unlock(&ls->ls_lkbidr_spin);
760         return rv;
761 }
762 
763 static int release_lockspace(struct dlm_ls *ls, int force)
764 {
765         struct dlm_rsb *rsb;
766         struct rb_node *n;
767         int i, busy, rv;
768 
769         busy = lockspace_busy(ls, force);
770 
771         spin_lock(&lslist_lock);
772         if (ls->ls_create_count == 1) {
773                 if (busy) {
774                         rv = -EBUSY;
775                 } else {
776                         /* remove_lockspace takes ls off lslist */
777                         ls->ls_create_count = 0;
778                         rv = 0;
779                 }
780         } else if (ls->ls_create_count > 1) {
781                 rv = --ls->ls_create_count;
782         } else {
783                 rv = -EINVAL;
784         }
785         spin_unlock(&lslist_lock);
786 
787         if (rv) {
788                 log_debug(ls, "release_lockspace no remove %d", rv);
789                 return rv;
790         }
791 
792         dlm_device_deregister(ls);
793 
794         if (force < 3 && dlm_user_daemon_available())
795                 do_uevent(ls, 0);
796 
797         dlm_recoverd_stop(ls);
798 
799         dlm_callback_stop(ls);
800 
801         remove_lockspace(ls);
802 
803         dlm_delete_debug_file(ls);
804 
805         kfree(ls->ls_recover_buf);
806 
807         /*
808          * Free all lkb's in idr
809          */
810 
811         idr_for_each(&ls->ls_lkbidr, lkb_idr_free, ls);
812         idr_destroy(&ls->ls_lkbidr);
813 
814         /*
815          * Free all rsb's on rsbtbl[] lists
816          */
817 
818         for (i = 0; i < ls->ls_rsbtbl_size; i++) {
819                 while ((n = rb_first(&ls->ls_rsbtbl[i].keep))) {
820                         rsb = rb_entry(n, struct dlm_rsb, res_hashnode);
821                         rb_erase(n, &ls->ls_rsbtbl[i].keep);
822                         dlm_free_rsb(rsb);
823                 }
824 
825                 while ((n = rb_first(&ls->ls_rsbtbl[i].toss))) {
826                         rsb = rb_entry(n, struct dlm_rsb, res_hashnode);
827                         rb_erase(n, &ls->ls_rsbtbl[i].toss);
828                         dlm_free_rsb(rsb);
829                 }
830         }
831 
832         vfree(ls->ls_rsbtbl);
833 
834         for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++)
835                 kfree(ls->ls_remove_names[i]);
836 
837         while (!list_empty(&ls->ls_new_rsb)) {
838                 rsb = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb,
839                                        res_hashchain);
840                 list_del(&rsb->res_hashchain);
841                 dlm_free_rsb(rsb);
842         }
843 
844         /*
845          * Free structures on any other lists
846          */
847 
848         dlm_purge_requestqueue(ls);
849         kfree(ls->ls_recover_args);
850         dlm_clear_members(ls);
851         dlm_clear_members_gone(ls);
852         kfree(ls->ls_node_array);
853         log_rinfo(ls, "release_lockspace final free");
854         kobject_put(&ls->ls_kobj);
855         /* The ls structure will be freed when the kobject is done with */
856 
857         module_put(THIS_MODULE);
858         return 0;
859 }
860 
861 /*
862  * Called when a system has released all its locks and is not going to use the
863  * lockspace any longer.  We free everything we're managing for this lockspace.
864  * Remaining nodes will go through the recovery process as if we'd died.  The
865  * lockspace must continue to function as usual, participating in recoveries,
866  * until this returns.
867  *
868  * Force has 4 possible values:
869  * 0 - don't destroy locksapce if it has any LKBs
870  * 1 - destroy lockspace if it has remote LKBs but not if it has local LKBs
871  * 2 - destroy lockspace regardless of LKBs
872  * 3 - destroy lockspace as part of a forced shutdown
873  */
874 
875 int dlm_release_lockspace(void *lockspace, int force)
876 {
877         struct dlm_ls *ls;
878         int error;
879 
880         ls = dlm_find_lockspace_local(lockspace);
881         if (!ls)
882                 return -EINVAL;
883         dlm_put_lockspace(ls);
884 
885         mutex_lock(&ls_lock);
886         error = release_lockspace(ls, force);
887         if (!error)
888                 ls_count--;
889         if (!ls_count)
890                 threads_stop();
891         mutex_unlock(&ls_lock);
892 
893         return error;
894 }
895 
896 void dlm_stop_lockspaces(void)
897 {
898         struct dlm_ls *ls;
899         int count;
900 
901  restart:
902         count = 0;
903         spin_lock(&lslist_lock);
904         list_for_each_entry(ls, &lslist, ls_list) {
905                 if (!test_bit(LSFL_RUNNING, &ls->ls_flags)) {
906                         count++;
907                         continue;
908                 }
909                 spin_unlock(&lslist_lock);
910                 log_error(ls, "no userland control daemon, stopping lockspace");
911                 dlm_ls_stop(ls);
912                 goto restart;
913         }
914         spin_unlock(&lslist_lock);
915 
916         if (count)
917                 log_print("dlm user daemon left %d lockspaces", count);
918 }
919 
920 

~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

kernel.org | git.kernel.org | LWN.net | Project Home | Wiki (Japanese) | Wiki (English) | SVN repository | Mail admin

Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.

osdn.jp