fs/dlm/lockspace.c

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