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