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