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