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