fs/dlm/lockspace.c

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