Merge tag 'dlm-6.18' of git://git.kernel.org/pub/scm/linux/kernel/git/teigland/linux-dlm

+2 -2

drivers/md/md-cluster.c

··· 979 979 lockres_free(cinfo->resync_lockres); 980 980 lockres_free(cinfo->bitmap_lockres); 981 981 if (cinfo->lockspace) 982 - dlm_release_lockspace(cinfo->lockspace, 2); 982 + dlm_release_lockspace(cinfo->lockspace, DLM_RELEASE_NORMAL); 983 983 mddev->cluster_info = NULL; 984 984 kfree(cinfo); 985 985 return ret; ··· 1042 1042 lockres_free(cinfo->resync_lockres); 1043 1043 lockres_free(cinfo->bitmap_lockres); 1044 1044 unlock_all_bitmaps(mddev); 1045 - dlm_release_lockspace(cinfo->lockspace, 2); 1045 + dlm_release_lockspace(cinfo->lockspace, DLM_RELEASE_NORMAL); 1046 1046 kfree(cinfo); 1047 1047 return 0; 1048 1048 }

+62 -2

fs/dlm/config.c

··· 26 26 /* 27 27 * /config/dlm/<cluster>/spaces/<space>/nodes/<node>/nodeid (refers to <node>) 28 28 * /config/dlm/<cluster>/spaces/<space>/nodes/<node>/weight 29 + * /config/dlm/<cluster>/spaces/<space>/nodes/<node>/release_recover 29 30 * /config/dlm/<cluster>/comms/<comm>/nodeid (refers to <comm>) 30 31 * /config/dlm/<cluster>/comms/<comm>/local 31 32 * /config/dlm/<cluster>/comms/<comm>/addr (write only) ··· 268 267 enum { 269 268 NODE_ATTR_NODEID = 0, 270 269 NODE_ATTR_WEIGHT, 270 + NODE_ATTR_RELEASE_RECOVER, 271 271 }; 272 272 273 273 struct dlm_clusters { ··· 282 280 struct dlm_space { 283 281 struct config_group group; 284 282 struct list_head members; 283 + struct list_head members_gone; 284 + int members_gone_count; 285 285 struct mutex members_lock; 286 286 int members_count; 287 287 struct dlm_nodes *nds; ··· 314 310 int weight; 315 311 int new; 316 312 int comm_seq; /* copy of cm->seq when nd->nodeid is set */ 313 + unsigned int release_recover; 314 + }; 315 + 316 + struct dlm_member_gone { 317 + int nodeid; 318 + unsigned int release_recover; 319 + 320 + struct list_head list; /* space->members_gone */ 317 321 }; 318 322 319 323 static struct configfs_group_operations clusters_ops = { ··· 492 480 configfs_add_default_group(&nds->ns_group, &sp->group); 493 481 494 482 INIT_LIST_HEAD(&sp->members); 483 + INIT_LIST_HEAD(&sp->members_gone); 495 484 mutex_init(&sp->members_lock); 496 485 sp->members_count = 0; 497 486 sp->nds = nds; ··· 600 587 { 601 588 struct dlm_space *sp = config_item_to_space(g->cg_item.ci_parent); 602 589 struct dlm_node *nd = config_item_to_node(i); 590 + struct dlm_member_gone *mb_gone; 591 + 592 + mb_gone = kzalloc(sizeof(*mb_gone), GFP_KERNEL); 593 + if (!mb_gone) 594 + return; 603 595 604 596 mutex_lock(&sp->members_lock); 605 597 list_del(&nd->list); 606 598 sp->members_count--; 599 + 600 + mb_gone->nodeid = nd->nodeid; 601 + mb_gone->release_recover = nd->release_recover; 602 + list_add(&mb_gone->list, &sp->members_gone); 603 + sp->members_gone_count++; 607 604 mutex_unlock(&sp->members_lock); 608 605 609 606 config_item_put(i); ··· 838 815 return len; 839 816 } 840 817 818 + static ssize_t node_release_recover_show(struct config_item *item, char *buf) 819 + { 820 + struct dlm_node *n = config_item_to_node(item); 821 + 822 + return sprintf(buf, "%u\n", n->release_recover); 823 + } 824 + 825 + static ssize_t node_release_recover_store(struct config_item *item, 826 + const char *buf, size_t len) 827 + { 828 + struct dlm_node *n = config_item_to_node(item); 829 + int rc; 830 + 831 + rc = kstrtouint(buf, 0, &n->release_recover); 832 + if (rc) 833 + return rc; 834 + 835 + return len; 836 + } 837 + 841 838 CONFIGFS_ATTR(node_, nodeid); 842 839 CONFIGFS_ATTR(node_, weight); 840 + CONFIGFS_ATTR(node_, release_recover); 843 841 844 842 static struct configfs_attribute *node_attrs[] = { 845 843 [NODE_ATTR_NODEID] = &node_attr_nodeid, 846 844 [NODE_ATTR_WEIGHT] = &node_attr_weight, 845 + [NODE_ATTR_RELEASE_RECOVER] = &node_attr_release_recover, 847 846 NULL, 848 847 }; 849 848 ··· 927 882 int dlm_config_nodes(char *lsname, struct dlm_config_node **nodes_out, 928 883 int *count_out) 929 884 { 885 + struct dlm_member_gone *mb_gone, *mb_safe; 886 + struct dlm_config_node *nodes, *node; 930 887 struct dlm_space *sp; 931 888 struct dlm_node *nd; 932 - struct dlm_config_node *nodes, *node; 933 889 int rv, count; 934 890 935 891 sp = get_space(lsname); ··· 944 898 goto out; 945 899 } 946 900 947 - count = sp->members_count; 901 + count = sp->members_count + sp->members_gone_count; 948 902 949 903 nodes = kcalloc(count, sizeof(struct dlm_config_node), GFP_NOFS); 950 904 if (!nodes) { ··· 961 915 node++; 962 916 963 917 nd->new = 0; 918 + } 919 + 920 + /* we delay the remove on nodes until here as configfs does 921 + * not support addtional attributes for rmdir(). 922 + */ 923 + list_for_each_entry_safe(mb_gone, mb_safe, &sp->members_gone, list) { 924 + node->nodeid = mb_gone->nodeid; 925 + node->release_recover = mb_gone->release_recover; 926 + node->gone = true; 927 + node++; 928 + 929 + list_del(&mb_gone->list); 930 + sp->members_gone_count--; 931 + kfree(mb_gone); 964 932 } 965 933 966 934 *count_out = count;

+2

fs/dlm/config.h

··· 17 17 struct dlm_config_node { 18 18 int nodeid; 19 19 int weight; 20 + bool gone; 20 21 int new; 21 22 uint32_t comm_seq; 23 + unsigned int release_recover; 22 24 }; 23 25 24 26 extern const struct rhashtable_params dlm_rhash_rsb_params;

+1 -1

fs/dlm/lock.c

··· 5576 5576 5577 5577 if (rl->rl_status == DLM_LKSTS_CONVERT && middle_conversion(lkb)) { 5578 5578 /* We may need to adjust grmode depending on other granted locks. */ 5579 - log_limit(ls, "%s %x middle convert gr %d rq %d remote %d %x", 5579 + log_rinfo(ls, "%s %x middle convert gr %d rq %d remote %d %x", 5580 5580 __func__, lkb->lkb_id, lkb->lkb_grmode, 5581 5581 lkb->lkb_rqmode, lkb->lkb_nodeid, lkb->lkb_remid); 5582 5582 rsb_set_flag(r, RSB_RECOVER_CONVERT);

+26 -20

fs/dlm/lockspace.c

··· 186 186 187 187 static struct kset *dlm_kset; 188 188 189 - static int do_uevent(struct dlm_ls *ls, int in) 189 + static int do_uevent(struct dlm_ls *ls, int in, unsigned int release_recover) 190 190 { 191 - if (in) 191 + char message[512] = {}; 192 + char *envp[] = { message, NULL }; 193 + 194 + if (in) { 192 195 kobject_uevent(&ls->ls_kobj, KOBJ_ONLINE); 193 - else 194 - kobject_uevent(&ls->ls_kobj, KOBJ_OFFLINE); 196 + } else { 197 + snprintf(message, 511, "RELEASE_RECOVER=%u", release_recover); 198 + kobject_uevent_env(&ls->ls_kobj, KOBJ_OFFLINE, envp); 199 + } 195 200 196 201 log_rinfo(ls, "%s the lockspace group...", in ? "joining" : "leaving"); 197 202 ··· 580 575 current lockspace members are (via configfs) and then tells the 581 576 lockspace to start running (via sysfs) in dlm_ls_start(). */ 582 577 583 - error = do_uevent(ls, 1); 578 + error = do_uevent(ls, 1, 0); 584 579 if (error < 0) 585 580 goto out_recoverd; 586 581 ··· 597 592 return 0; 598 593 599 594 out_members: 600 - do_uevent(ls, 0); 595 + do_uevent(ls, 0, 0); 601 596 dlm_clear_members(ls); 602 597 kfree(ls->ls_node_array); 603 598 out_recoverd: ··· 676 671 This is because there may be LKBs queued as ASTs that have been unlinked 677 672 from their RSBs and are pending deletion once the AST has been delivered */ 678 673 679 - static int lockspace_busy(struct dlm_ls *ls, int force) 674 + static int lockspace_busy(struct dlm_ls *ls, unsigned int release_option) 680 675 { 681 676 struct dlm_lkb *lkb; 682 677 unsigned long id; 683 678 int rv = 0; 684 679 685 680 read_lock_bh(&ls->ls_lkbxa_lock); 686 - if (force == 0) { 681 + if (release_option == DLM_RELEASE_NO_LOCKS) { 687 682 xa_for_each(&ls->ls_lkbxa, id, lkb) { 688 683 rv = 1; 689 684 break; 690 685 } 691 - } else if (force == 1) { 686 + } else if (release_option == DLM_RELEASE_UNUSED) { 687 + /* TODO: handle this UNUSED option as NO_LOCKS in later patch */ 692 688 xa_for_each(&ls->ls_lkbxa, id, lkb) { 693 689 if (lkb->lkb_nodeid == 0 && 694 690 lkb->lkb_grmode != DLM_LOCK_IV) { ··· 704 698 return rv; 705 699 } 706 700 707 - static int release_lockspace(struct dlm_ls *ls, int force) 701 + static int release_lockspace(struct dlm_ls *ls, unsigned int release_option) 708 702 { 709 703 int busy, rv; 710 704 711 - busy = lockspace_busy(ls, force); 705 + busy = lockspace_busy(ls, release_option); 712 706 713 707 spin_lock_bh(&lslist_lock); 714 708 if (ls->ls_create_count == 1) { ··· 736 730 737 731 dlm_device_deregister(ls); 738 732 739 - if (force < 3 && dlm_user_daemon_available()) 740 - do_uevent(ls, 0); 733 + if (release_option != DLM_RELEASE_NO_EVENT && 734 + dlm_user_daemon_available()) 735 + do_uevent(ls, 0, (release_option == DLM_RELEASE_RECOVER)); 741 736 742 737 dlm_recoverd_stop(ls); 743 738 ··· 789 782 * lockspace must continue to function as usual, participating in recoveries, 790 783 * until this returns. 791 784 * 792 - * Force has 4 possible values: 793 - * 0 - don't destroy lockspace if it has any LKBs 794 - * 1 - destroy lockspace if it has remote LKBs but not if it has local LKBs 795 - * 2 - destroy lockspace regardless of LKBs 796 - * 3 - destroy lockspace as part of a forced shutdown 785 + * See DLM_RELEASE defines for release_option values and their meaning. 797 786 */ 798 787 799 - int dlm_release_lockspace(void *lockspace, int force) 788 + int dlm_release_lockspace(void *lockspace, unsigned int release_option) 800 789 { 801 790 struct dlm_ls *ls; 802 791 int error; 792 + 793 + if (release_option > __DLM_RELEASE_MAX) 794 + return -EINVAL; 803 795 804 796 ls = dlm_find_lockspace_local(lockspace); 805 797 if (!ls) ··· 806 800 dlm_put_lockspace(ls); 807 801 808 802 mutex_lock(&ls_lock); 809 - error = release_lockspace(ls, force); 803 + error = release_lockspace(ls, release_option); 810 804 if (!error) 811 805 ls_count--; 812 806 if (!ls_count)

+21 -6

fs/dlm/member.c

··· 478 478 ls->ls_ops->recover_prep(ls->ls_ops_arg); 479 479 } 480 480 481 - static void dlm_lsop_recover_slot(struct dlm_ls *ls, struct dlm_member *memb) 481 + static void dlm_lsop_recover_slot(struct dlm_ls *ls, struct dlm_member *memb, 482 + unsigned int release_recover) 482 483 { 483 484 struct dlm_slot slot; 484 485 uint32_t seq; ··· 496 495 497 496 error = dlm_comm_seq(memb->nodeid, &seq, false); 498 497 499 - if (!error && seq == memb->comm_seq) 498 + if (!release_recover && !error && seq == memb->comm_seq) 500 499 return; 501 500 502 501 slot.nodeid = memb->nodeid; ··· 553 552 struct dlm_member *memb, *safe; 554 553 struct dlm_config_node *node; 555 554 int i, error, neg = 0, low = -1; 555 + unsigned int release_recover; 556 556 557 557 /* previously removed members that we've not finished removing need to 558 558 * count as a negative change so the "neg" recovery steps will happen ··· 571 569 572 570 list_for_each_entry_safe(memb, safe, &ls->ls_nodes, list) { 573 571 node = find_config_node(rv, memb->nodeid); 574 - if (node && !node->new) 572 + if (!node) { 573 + log_error(ls, "remove member %d invalid", 574 + memb->nodeid); 575 + return -EFAULT; 576 + } 577 + 578 + if (!node->new && !node->gone) 575 579 continue; 576 580 577 - if (!node) { 578 - log_rinfo(ls, "remove member %d", memb->nodeid); 581 + release_recover = 0; 582 + 583 + if (node->gone) { 584 + release_recover = node->release_recover; 585 + log_rinfo(ls, "remove member %d%s", memb->nodeid, 586 + release_recover ? " (release_recover)" : ""); 579 587 } else { 580 588 /* removed and re-added */ 581 589 log_rinfo(ls, "remove member %d comm_seq %u %u", ··· 596 584 list_move(&memb->list, &ls->ls_nodes_gone); 597 585 remove_remote_member(memb->nodeid); 598 586 ls->ls_num_nodes--; 599 - dlm_lsop_recover_slot(ls, memb); 587 + dlm_lsop_recover_slot(ls, memb, release_recover); 600 588 } 601 589 602 590 /* add new members to ls_nodes */ 603 591 604 592 for (i = 0; i < rv->nodes_count; i++) { 605 593 node = &rv->nodes[i]; 594 + if (node->gone) 595 + continue; 596 + 606 597 if (dlm_is_member(ls, node->nodeid)) 607 598 continue; 608 599 error = dlm_add_member(ls, node);

+1 -1

fs/dlm/recover.c

··· 842 842 */ 843 843 if (((lkb->lkb_grmode == DLM_LOCK_PR) && (other_grmode == DLM_LOCK_CW)) || 844 844 ((lkb->lkb_grmode == DLM_LOCK_CW) && (other_grmode == DLM_LOCK_PR))) { 845 - log_limit(ls, "%s %x gr %d rq %d, remote %d %x, other_lkid %u, other gr %d, set gr=NL", 845 + log_rinfo(ls, "%s %x gr %d rq %d, remote %d %x, other_lkid %u, other gr %d, set gr=NL", 846 846 __func__, lkb->lkb_id, lkb->lkb_grmode, 847 847 lkb->lkb_rqmode, lkb->lkb_nodeid, 848 848 lkb->lkb_remid, other_lkid, other_grmode);

+3 -3

fs/dlm/user.c

··· 425 425 dlm_put_lockspace(ls); 426 426 427 427 if (error) 428 - dlm_release_lockspace(lockspace, 0); 428 + dlm_release_lockspace(lockspace, DLM_RELEASE_NO_LOCKS); 429 429 else 430 430 error = ls->ls_device.minor; 431 431 ··· 436 436 { 437 437 dlm_lockspace_t *lockspace; 438 438 struct dlm_ls *ls; 439 - int error, force = 0; 439 + int error, force = DLM_RELEASE_NO_LOCKS; 440 440 441 441 if (!capable(CAP_SYS_ADMIN)) 442 442 return -EPERM; ··· 446 446 return -ENOENT; 447 447 448 448 if (params->flags & DLM_USER_LSFLG_FORCEFREE) 449 - force = 2; 449 + force = DLM_RELEASE_NORMAL; 450 450 451 451 lockspace = ls; 452 452 dlm_put_lockspace(ls);

+2 -2

fs/gfs2/lock_dlm.c

··· 1418 1418 return 0; 1419 1419 1420 1420 fail_release: 1421 - dlm_release_lockspace(ls->ls_dlm, 2); 1421 + dlm_release_lockspace(ls->ls_dlm, DLM_RELEASE_NORMAL); 1422 1422 fail_free: 1423 1423 free_recover_size(ls); 1424 1424 fail: ··· 1456 1456 release: 1457 1457 down_write(&ls->ls_sem); 1458 1458 if (ls->ls_dlm) { 1459 - dlm_release_lockspace(ls->ls_dlm, 2); 1459 + dlm_release_lockspace(ls->ls_dlm, DLM_RELEASE_NORMAL); 1460 1460 ls->ls_dlm = NULL; 1461 1461 } 1462 1462 up_write(&ls->ls_sem);

+1 -1

fs/ocfs2/stack_user.c

··· 952 952 static int user_cluster_disconnect(struct ocfs2_cluster_connection *conn) 953 953 { 954 954 version_unlock(conn); 955 - dlm_release_lockspace(conn->cc_lockspace, 2); 955 + dlm_release_lockspace(conn->cc_lockspace, DLM_RELEASE_NORMAL); 956 956 conn->cc_lockspace = NULL; 957 957 ocfs2_live_connection_drop(conn->cc_private); 958 958 conn->cc_private = NULL;

+32 -1

include/linux/dlm.h

··· 88 88 int *ops_result, dlm_lockspace_t **lockspace); 89 89 90 90 /* 91 + * dlm_release_lockspace() release_option values: 92 + * 93 + * DLM_RELEASE_NO_LOCKS returns -EBUSY if any locks (lkb's) 94 + * exist in the local lockspace. 95 + * 96 + * DLM_RELEASE_UNUSED previous value that is no longer used. 97 + * 98 + * DLM_RELEASE_NORMAL releases the lockspace regardless of any 99 + * locks managed in the local lockspace. 100 + * 101 + * DLM_RELEASE_NO_EVENT release the lockspace regardless of any 102 + * locks managed in the local lockspace, and does not submit 103 + * a leave event to the cluster manager, so other nodes will 104 + * not be notified that the node should be removed from the 105 + * list of lockspace members. 106 + * 107 + * DLM_RELEASE_RECOVER like DLM_RELEASE_NORMAL, but the remaining 108 + * nodes will handle the removal of the node as if the node 109 + * had failed, e.g. the recover_slot() callback would be used. 110 + */ 111 + #define DLM_RELEASE_NO_LOCKS 0 112 + #define DLM_RELEASE_UNUSED 1 113 + #define DLM_RELEASE_NORMAL 2 114 + #define DLM_RELEASE_NO_EVENT 3 115 + #define DLM_RELEASE_RECOVER 4 116 + #define __DLM_RELEASE_MAX DLM_RELEASE_RECOVER 117 + 118 + /* 91 119 * dlm_release_lockspace 92 120 * 93 121 * Stop a lockspace. 122 + * 123 + * release_option: see DLM_RELEASE values above. 94 124 */ 95 125 96 - int dlm_release_lockspace(dlm_lockspace_t *lockspace, int force); 126 + int dlm_release_lockspace(dlm_lockspace_t *lockspace, 127 + unsigned int release_option); 97 128 98 129 /* 99 130 * dlm_lock

Configure Feed

Configure Feed