fs/dlm/lockspace.c

   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