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

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

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

~ [ 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