LU-6939 nrs: add lock to protect TBF rule linkage

[fs/lustre-release.git] / lustre / ptlrpc / nrs_tbf.c

diff --git a/lustre/ptlrpc/nrs_tbf.c b/lustre/ptlrpc/nrs_tbf.c
index 1162c6f..7f92e10 100644 (file)
--- a/lustre/ptlrpc/nrs_tbf.c
+++ b/lustre/ptlrpc/nrs_tbf.c
@@ -116,7 +116,9 @@ nrs_tbf_cli_rule_put(struct nrs_tbf_client *cli)
 {
        LASSERT(!list_empty(&cli->tc_linkage));
        LASSERT(cli->tc_rule);
+       spin_lock(&cli->tc_rule->tr_rule_lock);
        list_del_init(&cli->tc_linkage);
+       spin_unlock(&cli->tc_rule->tr_rule_lock);
        nrs_tbf_rule_put(cli->tc_rule);
        cli->tc_rule = NULL;
 }
@@ -146,7 +148,8 @@ nrs_tbf_cli_reset(struct nrs_tbf_head *head,
                  struct nrs_tbf_rule *rule,
                  struct nrs_tbf_client *cli)
 {
-       if (!list_empty(&cli->tc_linkage)) {
+       spin_lock(&cli->tc_rule_lock);
+       if (cli->tc_rule != NULL && !list_empty(&cli->tc_linkage)) {
                LASSERT(rule != cli->tc_rule);
                nrs_tbf_cli_rule_put(cli);
        }
@@ -154,7 +157,10 @@ nrs_tbf_cli_reset(struct nrs_tbf_head *head,
        LASSERT(list_empty(&cli->tc_linkage));
        /* Rule's ref is added before called */
        cli->tc_rule = rule;
+       spin_lock(&rule->tr_rule_lock);
        list_add_tail(&cli->tc_linkage, &rule->tr_cli_list);
+       spin_unlock(&rule->tr_rule_lock);
+       spin_unlock(&cli->tc_rule_lock);
        nrs_tbf_cli_reset_value(head, cli);
 }
 
@@ -252,6 +258,7 @@ nrs_tbf_cli_init(struct nrs_tbf_head *head,
        head->th_ops->o_cli_init(cli, req);
        INIT_LIST_HEAD(&cli->tc_list);
        INIT_LIST_HEAD(&cli->tc_linkage);
+       spin_lock_init(&cli->tc_rule_lock);
        atomic_set(&cli->tc_ref, 1);
        rule = nrs_tbf_rule_match(head, cli);
        nrs_tbf_cli_reset(head, rule, cli);
@@ -263,7 +270,9 @@ nrs_tbf_cli_fini(struct nrs_tbf_client *cli)
        LASSERT(list_empty(&cli->tc_list));
        LASSERT(!cli->tc_in_heap);
        LASSERT(atomic_read(&cli->tc_ref) == 0);
+       spin_lock(&cli->tc_rule_lock);
        nrs_tbf_cli_rule_put(cli);
+       spin_unlock(&cli->tc_rule_lock);
        OBD_FREE_PTR(cli);
 }
 
@@ -292,6 +301,9 @@ nrs_tbf_rule_start(struct ptlrpc_nrs_policy *policy,
        atomic_set(&rule->tr_ref, 1);
        INIT_LIST_HEAD(&rule->tr_cli_list);
        INIT_LIST_HEAD(&rule->tr_nids);
+       INIT_LIST_HEAD(&rule->tr_linkage);
+       spin_lock_init(&rule->tr_rule_lock);
+       rule->tr_head = head;
 
        rc = head->th_ops->o_rule_init(policy, rule, start);
        if (rc) {
@@ -309,7 +321,6 @@ nrs_tbf_rule_start(struct ptlrpc_nrs_policy *policy,
                return -EEXIST;
        }
        list_add(&rule->tr_linkage, &head->th_list);
-       rule->tr_head = head;
        spin_unlock(&head->th_rule_lock);
        atomic_inc(&head->th_rule_sequence);
        if (start->tc_rule_flags & NTRS_DEFAULT) {