EX-8027 pcc: wait for in-progress attaches when remove PCC

When remove a PCC backend from a client, it should wait for all
in-progress attaches finished. Othwise, it results in the failure
of the PCC backend umount operation.

The reason is that the PCC copy is referenced in the kernel, not
used by any applications in user space and the tool "lsof" can not
check whether the target PCC backend is used or not.

Change-Id: I05b268e75841f9f17e77819ed20c85c78d7c6ad6
Signed-off-by: Qian Yingjin <qian@ddn.com>
Reviewed-on: https://review.whamcloud.com/c/ex/lustre-release/+/51940
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Feng Lei <flei@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>

diff --git a/lustre/llite/pcc.c b/lustre/llite/pcc.c
index 2650153..aa5e19a 100644 (file)
--- a/lustre/llite/pcc.c
+++ b/lustre/llite/pcc.c
@@ -1032,6 +1032,8 @@ pcc_dataset_flags_check(struct pcc_super *super, struct pcc_cmd *cmd)
        return 0;
 }
 
+static void pcc_dataset_put(struct pcc_dataset *dataset);
+
 /**
  * pcc_dataset_add - Add a Cache policy to control which files need be
  * cached and where it will be cached.
@@ -1075,6 +1077,8 @@ pcc_dataset_add(struct pcc_super *super, struct pcc_cmd *cmd)
        dataset->pccd_flags = cmd->u.pccc_add.pccc_flags;
        dataset->pccd_hsmtool_type = cmd->u.pccc_add.pccc_hsmtool_type;
        atomic_set(&dataset->pccd_refcount, 1);
+       atomic_set(&dataset->pccd_attach_nr, 0);
+       init_waitqueue_head(&dataset->pccd_attach_waitq);
 
        rc = pcc_dataset_rule_init(&dataset->pccd_rule, cmd);
        if (rc) {
@@ -1105,8 +1109,8 @@ pcc_dataset_add(struct pcc_super *super, struct pcc_cmd *cmd)
        return rc;
 }
 
-struct pcc_dataset *
-pcc_dataset_get(struct pcc_super *super, enum lu_pcc_type type, __u32 id)
+static struct pcc_dataset *
+pcc_dataset_attach_get(struct pcc_super *super, enum lu_pcc_type type, __u32 id)
 {
        struct pcc_dataset *dataset;
        struct pcc_dataset *selected = NULL;
@@ -1127,6 +1131,7 @@ pcc_dataset_get(struct pcc_super *super, enum lu_pcc_type type, __u32 id)
                     !(dataset->pccd_flags & PCC_DATASET_ROPCC)))
                        continue;
                atomic_inc(&dataset->pccd_refcount);
+               atomic_inc(&dataset->pccd_attach_nr);
                selected = dataset;
                break;
        }
@@ -1137,7 +1142,7 @@ pcc_dataset_get(struct pcc_super *super, enum lu_pcc_type type, __u32 id)
        return selected;
 }
 
-void
+static void
 pcc_dataset_put(struct pcc_dataset *dataset)
 {
        if (atomic_dec_and_test(&dataset->pccd_refcount)) {
@@ -1147,6 +1152,15 @@ pcc_dataset_put(struct pcc_dataset *dataset)
        }
 }
 
+static void
+pcc_dataset_attach_put(struct pcc_dataset *dataset)
+{
+       if (atomic_dec_and_test(&dataset->pccd_attach_nr))
+               wake_up(&dataset->pccd_attach_waitq);
+
+       pcc_dataset_put(dataset);
+}
+
 static int
 pcc_dataset_del(struct pcc_super *super, char *pathname)
 {
@@ -1159,10 +1173,12 @@ pcc_dataset_del(struct pcc_super *super, char *pathname)
                dataset = list_entry(l, struct pcc_dataset, pccd_linkage);
                if (strcmp(dataset->pccd_pathname, pathname) == 0) {
                        list_del_init(&dataset->pccd_linkage);
+                       up_write(&super->pccs_rw_sem);
+                       l_wait_event_abortable(dataset->pccd_attach_waitq,
+                               atomic_read(&dataset->pccd_attach_nr) == 0);
                        pcc_dataset_put(dataset);
                        super->pccs_generation++;
-                       rc = 0;
-                       break;
+                       return 0;
                }
        }
        up_write(&super->pccs_rw_sem);
@@ -1201,13 +1217,18 @@ pcc_super_dump(struct pcc_super *super, struct seq_file *m)
 
 static void pcc_remove_datasets(struct pcc_super *super)
 {
-       struct pcc_dataset *dataset, *tmp;
+       struct pcc_dataset *dataset;
 
        down_write(&super->pccs_rw_sem);
-       list_for_each_entry_safe(dataset, tmp,
-                                &super->pccs_datasets, pccd_linkage) {
+       while ((dataset = list_first_entry_or_null(&super->pccs_datasets,
+                                                  struct pcc_dataset,
+                                                  pccd_linkage)) != NULL) {
                list_del(&dataset->pccd_linkage);
+               up_write(&super->pccs_rw_sem);
+               l_wait_event_abortable(dataset->pccd_attach_waitq,
+                               atomic_read(&dataset->pccd_attach_nr) == 0);
                pcc_dataset_put(dataset);
+               down_write(&super->pccs_rw_sem);
        }
        super->pccs_generation++;
        up_write(&super->pccs_rw_sem);
@@ -4077,8 +4098,8 @@ int pcc_readwrite_attach(struct file *file, struct inode *inode,
        if (rc)
                RETURN(rc);
 
-       dataset = pcc_dataset_get(&ll_i2sbi(inode)->ll_pcc_super,
-                                 LU_PCC_READWRITE, archive_id);
+       dataset = pcc_dataset_attach_get(&ll_i2sbi(inode)->ll_pcc_super,
+                                        LU_PCC_READWRITE, archive_id);
        if (dataset == NULL)
                RETURN(-ENOENT);
 
@@ -4108,7 +4129,7 @@ out_unlock:
                dput(dentry);
        }
 out_dataset_put:
-       pcc_dataset_put(dataset);
+       pcc_dataset_attach_put(dataset);
        RETURN(rc);
 }
 
@@ -4269,8 +4290,8 @@ static int pcc_readonly_attach(struct file *file, struct inode *inode,
        if (rc)
                RETURN(rc);
 
-       dataset = pcc_dataset_get(&ll_s2sbi(inode->i_sb)->ll_pcc_super,
-                                 LU_PCC_READONLY, roid);
+       dataset = pcc_dataset_attach_get(&ll_s2sbi(inode->i_sb)->ll_pcc_super,
+                                        LU_PCC_READONLY, roid);
        if (dataset == NULL)
                RETURN(-ENOENT);
 
@@ -4279,6 +4300,8 @@ static int pcc_readonly_attach(struct file *file, struct inode *inode,
        if (rc)
                GOTO(out_dataset_put, rc);
 
+       OBD_FAIL_TIMEOUT(OBD_FAIL_LLITE_PCC_ATTACH_PAUSE, cfs_fail_val);
+
        pcc_inode_lock(inode);
        old_cred = override_creds(super->pccs_cred);
        if (gen != ll_layout_version_get(lli)) {
@@ -4334,7 +4357,7 @@ out_put_unlock:
        revert_creds(old_cred);
        pcc_inode_unlock(inode);
 out_dataset_put:
-       pcc_dataset_put(dataset);
+       pcc_dataset_attach_put(dataset);
 
        RETURN(rc);
 }

diff --git a/lustre/llite/pcc.h b/lustre/llite/pcc.h
index 99b789b..828a288 100644 (file)
--- a/lustre/llite/pcc.h
+++ b/lustre/llite/pcc.h
@@ -151,6 +151,8 @@ struct pcc_dataset {
        struct list_head        pccd_linkage;  /* Linked to pccs_datasets */
        atomic_t                pccd_refcount; /* Reference count */
        enum hsmtool_type       pccd_hsmtool_type; /* HSM copytool type */
+       atomic_t                pccd_attach_nr; /* # of in-progress attach */
+       wait_queue_head_t       pccd_attach_waitq; /* Waitq for attach */
 };
 
 #define PCC_DEFAULT_ASYNC_THRESHOLD    (256 << 20)
@@ -351,7 +353,6 @@ void pcc_create_attach_cleanup(struct super_block *sb,
 struct pcc_dataset *pcc_dataset_match_get(struct pcc_super *super,
                                          enum lu_pcc_type type,
                                          struct pcc_matcher *matcher);
-void pcc_dataset_put(struct pcc_dataset *dataset);
 void pcc_inode_free(struct inode *inode);
 void pcc_layout_invalidate(struct inode *inode);
 

diff --git a/lustre/tests/sanity-pcc.sh b/lustre/tests/sanity-pcc.sh
old mode 100644 (file)
new mode 100755 (executable)
index 064501c..d305c86
--- a/lustre/tests/sanity-pcc.sh
+++ b/lustre/tests/sanity-pcc.sh
@@ -4889,6 +4889,43 @@ test_105() {
 }
 run_test 105 "Return valid flag for PCC file that is still valid cached"
 
+test_106() {
+       local loopfile="$TMP/$tfile"
+       local mntpt="/mnt/pcc.$tdir"
+       local hsm_root="$mntpt/$tdir"
+       local file=$DIR/$tfile
+
+       $LCTL get_param -n mdc.*.connect_flags | grep -q pcc_ro ||
+               skip "Server does not support PCC-RO"
+
+       setup_loopdev client $loopfile $mntpt 600
+       mkdir $hsm_root || error "mkdir $hsm_root failed"
+       setup_pcc_mapping client \
+               "projid={100}\ roid=$HSM_ARCHIVE_NUMBER\ pccro=1"
+
+       local pid
+
+       echo "Test for PCC dataset clear (PCC_CLEAR_ALL)"
+       echo "QQQQQ" > $file || error "failed to write $file"
+       # define OBD_FAIL_LLITE_PCC_ATTACH_PAUSE        0x1414
+       $LCTL set_param fail_loc=0x1414 fail_val=5
+       $LFS pcc attach $file &
+       pid=$!
+       sleep 1
+       $LCTL set_param llite.*.pcc="clear"
+       wait $pid || error "PCC attach for $file failed"
+       $LFS pcc state $file
+       $LFS pcc detach $file || error "failed to detch $file"
+
+       echo "Test for PCC dataset del (PCC_DEL_DATASET)"
+       setup_pcc_mapping client \
+               "projid={100}\ roid=$HSM_ARCHIVE_NUMBER\ pccro=1"
+       $LFS pcc attach $file &
+       sleep 1
+       $LCTL pcc clear -v $MOUNT
+}
+run_test 106 "PCC backend cleanup shound wait for all attaching finished"
+
 test_200() {
        local loopfile="$TMP/$tfile"
        local mntpt="/mnt/pcc.$tdir"