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_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 atomic_inc(&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 atomic_inc(&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 atomic_inc(&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 if (atomic_dec_and_test(&ls->ls_count)) 363 wake_up(&ls->ls_count_wait); 364 } 365 366 static void remove_lockspace(struct dlm_ls *ls) 367 { 368 retry: 369 wait_event(ls->ls_count_wait, atomic_read(&ls->ls_count) == 0); 370 371 spin_lock(&lslist_lock); 372 if (atomic_read(&ls->ls_count) != 0) { 373 spin_unlock(&lslist_lock); 374 goto retry; 375 } 376 377 WARN_ON(ls->ls_create_count != 0); 378 list_del(&ls->ls_list); 379 spin_unlock(&lslist_lock); 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_midcomms_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 int new_lockspace(const char *name, const char *cluster, 408 uint32_t flags, int lvblen, 409 const struct dlm_lockspace_ops *ops, void *ops_arg, 410 int *ops_result, dlm_lockspace_t **lockspace) 411 { 412 struct dlm_ls *ls; 413 int i, size, error; 414 int do_unreg = 0; 415 int namelen = strlen(name); 416 417 if (namelen > DLM_LOCKSPACE_LEN || namelen == 0) 418 return -EINVAL; 419 420 if (!lvblen || (lvblen % 8)) 421 return -EINVAL; 422 423 if (!try_module_get(THIS_MODULE)) 424 return -EINVAL; 425 426 if (!dlm_user_daemon_available()) { 427 log_print("dlm user daemon not available"); 428 error = -EUNATCH; 429 goto out; 430 } 431 432 if (ops && ops_result) { 433 if (!dlm_config.ci_recover_callbacks) 434 *ops_result = -EOPNOTSUPP; 435 else 436 *ops_result = 0; 437 } 438 439 if (!cluster) 440 log_print("dlm cluster name '%s' is being used without an application provided cluster name", 441 dlm_config.ci_cluster_name); 442 443 if (dlm_config.ci_recover_callbacks && cluster && 444 strncmp(cluster, dlm_config.ci_cluster_name, DLM_LOCKSPACE_LEN)) { 445 log_print("dlm cluster name '%s' does not match " 446 "the application cluster name '%s'", 447 dlm_config.ci_cluster_name, cluster); 448 error = -EBADR; 449 goto out; 450 } 451 452 error = 0; 453 454 spin_lock(&lslist_lock); 455 list_for_each_entry(ls, &lslist, ls_list) { 456 WARN_ON(ls->ls_create_count <= 0); 457 if (ls->ls_namelen != namelen) 458 continue; 459 if (memcmp(ls->ls_name, name, namelen)) 460 continue; 461 if (flags & DLM_LSFL_NEWEXCL) { 462 error = -EEXIST; 463 break; 464 } 465 ls->ls_create_count++; 466 *lockspace = ls; 467 error = 1; 468 break; 469 } 470 spin_unlock(&lslist_lock); 471 472 if (error) 473 goto out; 474 475 error = -ENOMEM; 476 477 ls = kzalloc(sizeof(struct dlm_ls) + namelen, GFP_NOFS); 478 if (!ls) 479 goto out; 480 memcpy(ls->ls_name, name, namelen); 481 ls->ls_namelen = namelen; 482 ls->ls_lvblen = lvblen; 483 atomic_set(&ls->ls_count, 0); 484 init_waitqueue_head(&ls->ls_count_wait); 485 ls->ls_flags = 0; 486 ls->ls_scan_time = jiffies; 487 488 if (ops && dlm_config.ci_recover_callbacks) { 489 ls->ls_ops = ops; 490 ls->ls_ops_arg = ops_arg; 491 } 492 493 if (flags & DLM_LSFL_TIMEWARN) 494 set_bit(LSFL_TIMEWARN, &ls->ls_flags); 495 496 /* ls_exflags are forced to match among nodes, and we don't 497 need to require all nodes to have some flags set */ 498 ls->ls_exflags = (flags & ~(DLM_LSFL_TIMEWARN | DLM_LSFL_FS | 499 DLM_LSFL_NEWEXCL)); 500 501 size = READ_ONCE(dlm_config.ci_rsbtbl_size); 502 ls->ls_rsbtbl_size = size; 503 504 ls->ls_rsbtbl = vmalloc(array_size(size, sizeof(struct dlm_rsbtable))); 505 if (!ls->ls_rsbtbl) 506 goto out_lsfree; 507 for (i = 0; i < size; i++) { 508 ls->ls_rsbtbl[i].keep.rb_node = NULL; 509 ls->ls_rsbtbl[i].toss.rb_node = NULL; 510 spin_lock_init(&ls->ls_rsbtbl[i].lock); 511 } 512 513 spin_lock_init(&ls->ls_remove_spin); 514 init_waitqueue_head(&ls->ls_remove_wait); 515 516 for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++) { 517 ls->ls_remove_names[i] = kzalloc(DLM_RESNAME_MAXLEN+1, 518 GFP_KERNEL); 519 if (!ls->ls_remove_names[i]) 520 goto out_rsbtbl; 521 } 522 523 idr_init(&ls->ls_lkbidr); 524 spin_lock_init(&ls->ls_lkbidr_spin); 525 526 INIT_LIST_HEAD(&ls->ls_waiters); 527 mutex_init(&ls->ls_waiters_mutex); 528 INIT_LIST_HEAD(&ls->ls_orphans); 529 mutex_init(&ls->ls_orphans_mutex); 530 INIT_LIST_HEAD(&ls->ls_timeout); 531 mutex_init(&ls->ls_timeout_mutex); 532 533 INIT_LIST_HEAD(&ls->ls_new_rsb); 534 spin_lock_init(&ls->ls_new_rsb_spin); 535 536 INIT_LIST_HEAD(&ls->ls_nodes); 537 INIT_LIST_HEAD(&ls->ls_nodes_gone); 538 ls->ls_num_nodes = 0; 539 ls->ls_low_nodeid = 0; 540 ls->ls_total_weight = 0; 541 ls->ls_node_array = NULL; 542 543 memset(&ls->ls_stub_rsb, 0, sizeof(struct dlm_rsb)); 544 ls->ls_stub_rsb.res_ls = ls; 545 546 ls->ls_debug_rsb_dentry = NULL; 547 ls->ls_debug_waiters_dentry = NULL; 548 549 init_waitqueue_head(&ls->ls_uevent_wait); 550 ls->ls_uevent_result = 0; 551 init_completion(&ls->ls_members_done); 552 ls->ls_members_result = -1; 553 554 mutex_init(&ls->ls_cb_mutex); 555 INIT_LIST_HEAD(&ls->ls_cb_delay); 556 557 ls->ls_recoverd_task = NULL; 558 mutex_init(&ls->ls_recoverd_active); 559 spin_lock_init(&ls->ls_recover_lock); 560 spin_lock_init(&ls->ls_rcom_spin); 561 get_random_bytes(&ls->ls_rcom_seq, sizeof(uint64_t)); 562 ls->ls_recover_status = 0; 563 ls->ls_recover_seq = 0; 564 ls->ls_recover_args = NULL; 565 init_rwsem(&ls->ls_in_recovery); 566 init_rwsem(&ls->ls_recv_active); 567 INIT_LIST_HEAD(&ls->ls_requestqueue); 568 atomic_set(&ls->ls_requestqueue_cnt, 0); 569 init_waitqueue_head(&ls->ls_requestqueue_wait); 570 mutex_init(&ls->ls_requestqueue_mutex); 571 mutex_init(&ls->ls_clear_proc_locks); 572 573 /* Due backwards compatibility with 3.1 we need to use maximum 574 * possible dlm message size to be sure the message will fit and 575 * not having out of bounds issues. However on sending side 3.2 576 * might send less. 577 */ 578 ls->ls_recover_buf = kmalloc(DLM_MAX_SOCKET_BUFSIZE, GFP_NOFS); 579 if (!ls->ls_recover_buf) 580 goto out_lkbidr; 581 582 ls->ls_slot = 0; 583 ls->ls_num_slots = 0; 584 ls->ls_slots_size = 0; 585 ls->ls_slots = NULL; 586 587 INIT_LIST_HEAD(&ls->ls_recover_list); 588 spin_lock_init(&ls->ls_recover_list_lock); 589 idr_init(&ls->ls_recover_idr); 590 spin_lock_init(&ls->ls_recover_idr_lock); 591 ls->ls_recover_list_count = 0; 592 ls->ls_local_handle = ls; 593 init_waitqueue_head(&ls->ls_wait_general); 594 INIT_LIST_HEAD(&ls->ls_root_list); 595 init_rwsem(&ls->ls_root_sem); 596 597 spin_lock(&lslist_lock); 598 ls->ls_create_count = 1; 599 list_add(&ls->ls_list, &lslist); 600 spin_unlock(&lslist_lock); 601 602 if (flags & DLM_LSFL_FS) { 603 error = dlm_callback_start(ls); 604 if (error) { 605 log_error(ls, "can't start dlm_callback %d", error); 606 goto out_delist; 607 } 608 } 609 610 init_waitqueue_head(&ls->ls_recover_lock_wait); 611 612 /* 613 * Once started, dlm_recoverd first looks for ls in lslist, then 614 * initializes ls_in_recovery as locked in "down" mode. We need 615 * to wait for the wakeup from dlm_recoverd because in_recovery 616 * has to start out in down mode. 617 */ 618 619 error = dlm_recoverd_start(ls); 620 if (error) { 621 log_error(ls, "can't start dlm_recoverd %d", error); 622 goto out_callback; 623 } 624 625 wait_event(ls->ls_recover_lock_wait, 626 test_bit(LSFL_RECOVER_LOCK, &ls->ls_flags)); 627 628 /* let kobject handle freeing of ls if there's an error */ 629 do_unreg = 1; 630 631 ls->ls_kobj.kset = dlm_kset; 632 error = kobject_init_and_add(&ls->ls_kobj, &dlm_ktype, NULL, 633 "%s", ls->ls_name); 634 if (error) 635 goto out_recoverd; 636 kobject_uevent(&ls->ls_kobj, KOBJ_ADD); 637 638 /* This uevent triggers dlm_controld in userspace to add us to the 639 group of nodes that are members of this lockspace (managed by the 640 cluster infrastructure.) Once it's done that, it tells us who the 641 current lockspace members are (via configfs) and then tells the 642 lockspace to start running (via sysfs) in dlm_ls_start(). */ 643 644 error = do_uevent(ls, 1); 645 if (error) 646 goto out_recoverd; 647 648 wait_for_completion(&ls->ls_members_done); 649 error = ls->ls_members_result; 650 if (error) 651 goto out_members; 652 653 dlm_create_debug_file(ls); 654 655 log_rinfo(ls, "join complete"); 656 *lockspace = ls; 657 return 0; 658 659 out_members: 660 do_uevent(ls, 0); 661 dlm_clear_members(ls); 662 kfree(ls->ls_node_array); 663 out_recoverd: 664 dlm_recoverd_stop(ls); 665 out_callback: 666 dlm_callback_stop(ls); 667 out_delist: 668 spin_lock(&lslist_lock); 669 list_del(&ls->ls_list); 670 spin_unlock(&lslist_lock); 671 idr_destroy(&ls->ls_recover_idr); 672 kfree(ls->ls_recover_buf); 673 out_lkbidr: 674 idr_destroy(&ls->ls_lkbidr); 675 out_rsbtbl: 676 for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++) 677 kfree(ls->ls_remove_names[i]); 678 vfree(ls->ls_rsbtbl); 679 out_lsfree: 680 if (do_unreg) 681 kobject_put(&ls->ls_kobj); 682 else 683 kfree(ls); 684 out: 685 module_put(THIS_MODULE); 686 return error; 687 } 688 689 int dlm_new_lockspace(const char *name, const char *cluster, 690 uint32_t flags, int lvblen, 691 const struct dlm_lockspace_ops *ops, void *ops_arg, 692 int *ops_result, dlm_lockspace_t **lockspace) 693 { 694 int error = 0; 695 696 mutex_lock(&ls_lock); 697 if (!ls_count) 698 error = threads_start(); 699 if (error) 700 goto out; 701 702 error = new_lockspace(name, cluster, flags, lvblen, ops, ops_arg, 703 ops_result, lockspace); 704 if (!error) 705 ls_count++; 706 if (error > 0) 707 error = 0; 708 if (!ls_count) { 709 dlm_scand_stop(); 710 dlm_midcomms_shutdown(); 711 dlm_lowcomms_stop(); 712 } 713 out: 714 mutex_unlock(&ls_lock); 715 return error; 716 } 717 718 static int lkb_idr_is_local(int id, void *p, void *data) 719 { 720 struct dlm_lkb *lkb = p; 721 722 return lkb->lkb_nodeid == 0 && lkb->lkb_grmode != DLM_LOCK_IV; 723 } 724 725 static int lkb_idr_is_any(int id, void *p, void *data) 726 { 727 return 1; 728 } 729 730 static int lkb_idr_free(int id, void *p, void *data) 731 { 732 struct dlm_lkb *lkb = p; 733 734 if (lkb->lkb_lvbptr && lkb->lkb_flags & DLM_IFL_MSTCPY) 735 dlm_free_lvb(lkb->lkb_lvbptr); 736 737 dlm_free_lkb(lkb); 738 return 0; 739 } 740 741 /* NOTE: We check the lkbidr here rather than the resource table. 742 This is because there may be LKBs queued as ASTs that have been unlinked 743 from their RSBs and are pending deletion once the AST has been delivered */ 744 745 static int lockspace_busy(struct dlm_ls *ls, int force) 746 { 747 int rv; 748 749 spin_lock(&ls->ls_lkbidr_spin); 750 if (force == 0) { 751 rv = idr_for_each(&ls->ls_lkbidr, lkb_idr_is_any, ls); 752 } else if (force == 1) { 753 rv = idr_for_each(&ls->ls_lkbidr, lkb_idr_is_local, ls); 754 } else { 755 rv = 0; 756 } 757 spin_unlock(&ls->ls_lkbidr_spin); 758 return rv; 759 } 760 761 static int release_lockspace(struct dlm_ls *ls, int force) 762 { 763 struct dlm_rsb *rsb; 764 struct rb_node *n; 765 int i, busy, rv; 766 767 busy = lockspace_busy(ls, force); 768 769 spin_lock(&lslist_lock); 770 if (ls->ls_create_count == 1) { 771 if (busy) { 772 rv = -EBUSY; 773 } else { 774 /* remove_lockspace takes ls off lslist */ 775 ls->ls_create_count = 0; 776 rv = 0; 777 } 778 } else if (ls->ls_create_count > 1) { 779 rv = --ls->ls_create_count; 780 } else { 781 rv = -EINVAL; 782 } 783 spin_unlock(&lslist_lock); 784 785 if (rv) { 786 log_debug(ls, "release_lockspace no remove %d", rv); 787 return rv; 788 } 789 790 dlm_device_deregister(ls); 791 792 if (force < 3 && dlm_user_daemon_available()) 793 do_uevent(ls, 0); 794 795 dlm_recoverd_stop(ls); 796 797 if (ls_count == 1) { 798 dlm_scand_stop(); 799 dlm_clear_members(ls); 800 dlm_midcomms_shutdown(); 801 } 802 803 dlm_callback_stop(ls); 804 805 remove_lockspace(ls); 806 807 dlm_delete_debug_file(ls); 808 809 idr_destroy(&ls->ls_recover_idr); 810 kfree(ls->ls_recover_buf); 811 812 /* 813 * Free all lkb's in idr 814 */ 815 816 idr_for_each(&ls->ls_lkbidr, lkb_idr_free, ls); 817 idr_destroy(&ls->ls_lkbidr); 818 819 /* 820 * Free all rsb's on rsbtbl[] lists 821 */ 822 823 for (i = 0; i < ls->ls_rsbtbl_size; i++) { 824 while ((n = rb_first(&ls->ls_rsbtbl[i].keep))) { 825 rsb = rb_entry(n, struct dlm_rsb, res_hashnode); 826 rb_erase(n, &ls->ls_rsbtbl[i].keep); 827 dlm_free_rsb(rsb); 828 } 829 830 while ((n = rb_first(&ls->ls_rsbtbl[i].toss))) { 831 rsb = rb_entry(n, struct dlm_rsb, res_hashnode); 832 rb_erase(n, &ls->ls_rsbtbl[i].toss); 833 dlm_free_rsb(rsb); 834 } 835 } 836 837 vfree(ls->ls_rsbtbl); 838 839 for (i = 0; i < DLM_REMOVE_NAMES_MAX; i++) 840 kfree(ls->ls_remove_names[i]); 841 842 while (!list_empty(&ls->ls_new_rsb)) { 843 rsb = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb, 844 res_hashchain); 845 list_del(&rsb->res_hashchain); 846 dlm_free_rsb(rsb); 847 } 848 849 /* 850 * Free structures on any other lists 851 */ 852 853 dlm_purge_requestqueue(ls); 854 kfree(ls->ls_recover_args); 855 dlm_clear_members(ls); 856 dlm_clear_members_gone(ls); 857 kfree(ls->ls_node_array); 858 log_rinfo(ls, "release_lockspace final free"); 859 kobject_put(&ls->ls_kobj); 860 /* The ls structure will be freed when the kobject is done with */ 861 862 module_put(THIS_MODULE); 863 return 0; 864 } 865 866 /* 867 * Called when a system has released all its locks and is not going to use the 868 * lockspace any longer. We free everything we're managing for this lockspace. 869 * Remaining nodes will go through the recovery process as if we'd died. The 870 * lockspace must continue to function as usual, participating in recoveries, 871 * until this returns. 872 * 873 * Force has 4 possible values: 874 * 0 - don't destroy lockspace if it has any LKBs 875 * 1 - destroy lockspace if it has remote LKBs but not if it has local LKBs 876 * 2 - destroy lockspace regardless of LKBs 877 * 3 - destroy lockspace as part of a forced shutdown 878 */ 879 880 int dlm_release_lockspace(void *lockspace, int force) 881 { 882 struct dlm_ls *ls; 883 int error; 884 885 ls = dlm_find_lockspace_local(lockspace); 886 if (!ls) 887 return -EINVAL; 888 dlm_put_lockspace(ls); 889 890 mutex_lock(&ls_lock); 891 error = release_lockspace(ls, force); 892 if (!error) 893 ls_count--; 894 if (!ls_count) 895 dlm_lowcomms_stop(); 896 mutex_unlock(&ls_lock); 897 898 return error; 899 } 900 901 void dlm_stop_lockspaces(void) 902 { 903 struct dlm_ls *ls; 904 int count; 905 906 restart: 907 count = 0; 908 spin_lock(&lslist_lock); 909 list_for_each_entry(ls, &lslist, ls_list) { 910 if (!test_bit(LSFL_RUNNING, &ls->ls_flags)) { 911 count++; 912 continue; 913 } 914 spin_unlock(&lslist_lock); 915 log_error(ls, "no userland control daemon, stopping lockspace"); 916 dlm_ls_stop(ls); 917 goto restart; 918 } 919 spin_unlock(&lslist_lock); 920 921 if (count) 922 log_print("dlm user daemon left %d lockspaces", count); 923 } 924 925