LU-8801 quota: fix ldiskfs accounting iterator

walk_tree_dqentry() should always record the information of the
index block it walked through no matter if a valid entry is found
in this call, so that it can continue the next search from correct
position.

Added a test case in s-q to verify accounting iterator.

Reading proc file acct_user/group will become extremely slow when
there are large amount of id entries, because each seq->start():
lprocfs_quota_seq_start() calls it->next() one by one to move the
cursor to last processed position. This patch changed it to use the
it->load() on last processed key, and reading performance improved
significantly.

Signed-off-by: Niu Yawei <yawei.niu@intel.com>
Change-Id: I2ddcd35e505233fa4de821fefa5fbe87be4dbe19
Reviewed-on: https://review.whamcloud.com/23583
Tested-by: Jenkins
Reviewed-by: Fan Yong <fan.yong@intel.com>
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Bobi Jam <bobijam@hotmail.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/osd-ldiskfs/osd_quota_fmt.c b/lustre/osd-ldiskfs/osd_quota_fmt.c
index a18f388..7f2a06e 100644 (file)
--- a/lustre/osd-ldiskfs/osd_quota_fmt.c
+++ b/lustre/osd-ldiskfs/osd_quota_fmt.c
@@ -286,12 +286,10 @@ int walk_tree_dqentry(const struct lu_env *env, struct osd_object *obj,
                                                depth + 1, 0, it);
                else
                        ret = walk_block_dqentry(env, obj, type, blk, 0, it);
-       }
 
-       if (ret == 0) { /* Entry found */
-               it->oiq_blk[depth + 1] = blk;
-               it->oiq_index[depth] = index;
        }
+       it->oiq_blk[depth + 1] = blk;
+       it->oiq_index[depth] = index;
 
 out_buf:
        freedqbuf(buf);

diff --git a/lustre/quota/lproc_quota.c b/lustre/quota/lproc_quota.c
index 8c8ad65..1f2584f 100644 (file)
--- a/lustre/quota/lproc_quota.c
+++ b/lustre/quota/lproc_quota.c
@@ -41,8 +41,11 @@
  * It is passed to seq_start/stop/next/show which can thus use the same lu_env
  * to be used with the iterator API */
 struct lquota_procfs {
-       struct dt_object        *lqp_obj;
-       struct lu_env            lqp_env;
+       struct dt_object *lqp_obj;
+       struct lu_env lqp_env;
+       struct dt_it *lqp_it;
+       __u64 lqp_cookie;
+       __u64 lqp_first_cookie;
 };
 
 /* global shared environment */
@@ -58,27 +61,20 @@ static void *lprocfs_quota_seq_start(struct seq_file *p, loff_t *pos)
 
        if (offset == 0)
                return SEQ_START_TOKEN;
-       offset--;
 
        if (lqp->lqp_obj == NULL)
                /* accounting not enabled. */
                return NULL;
 
-       /* initialize iterator */
-       iops = &lqp->lqp_obj->do_index_ops->dio_it;
-       it = iops->init(&lqp->lqp_env, lqp->lqp_obj, 0);
-       if (IS_ERR(it)) {
-               CERROR("%s: failed to initialize iterator: rc = %ld\n",
-                      lqp->lqp_obj->do_lu.lo_dev->ld_obd->obd_name,
-                      PTR_ERR(it));
+       if (lqp->lqp_it == NULL) /* reach the end */
                return NULL;
-       }
 
-       /* move on to the first valid record */
-       rc = iops->load(&lqp->lqp_env, it, 0);
-       if (rc < 0) { /* Error */
-               goto not_found;
-       } else if (rc == 0) {
+       offset--;
+       /* move on to the the last processed entry */
+       iops = &lqp->lqp_obj->do_index_ops->dio_it;
+       it = lqp->lqp_it;
+       rc = iops->load(&lqp->lqp_env, it, lqp->lqp_cookie);
+       if (rc == 0 && offset == 0) {
                /*
                 * Iterator didn't find record with exactly the key requested.
                 *
@@ -92,17 +88,25 @@ static void *lprocfs_quota_seq_start(struct seq_file *p, loff_t *pos)
                rc = iops->next(&lqp->lqp_env, it);
                if (rc != 0)
                        goto not_found;
+               lqp->lqp_cookie = iops->store(&lqp->lqp_env, it);
+       } else if (rc <= 0) {
+               goto not_found;
        }
-       while (offset--) {
-               rc = iops->next(&lqp->lqp_env, it);
-               if (rc != 0) /* Error or reach the end */
-                       goto not_found;
-       }
+
+       /* The id entry could be deleted while iteration, and above ->load()
+        * operation will reset cursor to the first cookie (ldiskfs), we
+        * need to break in such situation. */
+       if (offset == 0)
+               lqp->lqp_first_cookie = lqp->lqp_cookie;
+       else if (lqp->lqp_cookie == lqp->lqp_first_cookie)
+               goto not_found;
+
        return it;
 
 not_found:
        iops->put(&lqp->lqp_env, it);
        iops->fini(&lqp->lqp_env, it);
+       lqp->lqp_it = NULL;
        return NULL;
 }
 
@@ -120,7 +124,6 @@ static void lprocfs_quota_seq_stop(struct seq_file *p, void *v)
        /* if something wrong happened during ->seq_show, we need to release
         * the iterator here */
        iops->put(&lqp->lqp_env, it);
-       iops->fini(&lqp->lqp_env, it);
 }
 
 static void *lprocfs_quota_seq_next(struct seq_file *p, void *v, loff_t *pos)
@@ -139,12 +142,17 @@ static void *lprocfs_quota_seq_next(struct seq_file *p, void *v, loff_t *pos)
        if (v == SEQ_START_TOKEN)
                return lprocfs_quota_seq_start(p, pos);
 
+       if (lqp->lqp_it == NULL) /* reach the end */
+               return NULL;
+
        iops = &lqp->lqp_obj->do_index_ops->dio_it;
        it = (struct dt_it *)v;
 
        rc = iops->next(&lqp->lqp_env, it);
-       if (rc == 0)
+       if (rc == 0) {
+               lqp->lqp_cookie = iops->store(&lqp->lqp_env, it);
                return it;
+       }
 
        if (rc < 0)
                CERROR("%s: seq_next failed: rc = %d\n",
@@ -153,6 +161,7 @@ static void *lprocfs_quota_seq_next(struct seq_file *p, void *v, loff_t *pos)
        /* Reach the end or error */
        iops->put(&lqp->lqp_env, it);
        iops->fini(&lqp->lqp_env, it);
+       lqp->lqp_it = NULL;
        return NULL;
 }
 
@@ -254,6 +263,8 @@ static int lprocfs_quota_seq_open(struct inode *inode, struct file *file)
        struct seq_file         *seq;
        int                      rc;
        struct lquota_procfs    *lqp;
+       const struct dt_it_ops  *iops;
+       struct dt_it *it;
 
        /* Allocate quota procfs data. This structure will be passed to
         * seq_start/stop/next/show via seq->private */
@@ -285,6 +296,20 @@ static int lprocfs_quota_seq_open(struct inode *inode, struct file *file)
        if (rc)
                goto out_env;
 
+       /* initialize iterator */
+       iops = &lqp->lqp_obj->do_index_ops->dio_it;
+       it = iops->init(&lqp->lqp_env, lqp->lqp_obj, 0);
+       if (IS_ERR(it)) {
+               rc = PTR_ERR(it);
+               CERROR("%s: failed to initialize iterator: rc = %ld\n",
+                      lqp->lqp_obj->do_lu.lo_dev->ld_obd->obd_name,
+                      PTR_ERR(it));
+               seq_release(inode, file);
+               goto out_env;
+       }
+       lqp->lqp_it = it;
+       lqp->lqp_cookie = 0;
+
        seq = file->private_data;
        seq->private = lqp;
        return 0;
@@ -300,8 +325,12 @@ static int lprocfs_quota_seq_release(struct inode *inode, struct file *file)
 {
        struct seq_file         *seq = file->private_data;
        struct lquota_procfs    *lqp = seq->private;
+       const struct dt_it_ops  *iops;
 
        LASSERT(lqp);
+       iops = &lqp->lqp_obj->do_index_ops->dio_it;
+       if (lqp->lqp_it != NULL)
+               iops->fini(&lqp->lqp_env, lqp->lqp_it);
        lu_env_fini(&lqp->lqp_env);
        OBD_FREE_PTR(lqp);
 

diff --git a/lustre/tests/sanity-quota.sh b/lustre/tests/sanity-quota.sh
index 3cd052c..f443f68 100755 (executable)
--- a/lustre/tests/sanity-quota.sh
+++ b/lustre/tests/sanity-quota.sh
@@ -2332,6 +2332,49 @@ test_37() {
 }
 run_test 37 "Quota accounted properly for file created by 'lfs setstripe'"
 
+# LU-8801
+test_38() {
+       [ $(lustre_version_code $SINGLEMDS) -lt $(version_code 2.8.60) ] &&
+               skip "Old server doesn't have LU-8801 fix." && return
+
+       [ "$UID" != 0 ] && skip_env "must run as root" && return
+
+       setup_quota_test || error "setup quota failed with $?"
+       trap cleanup_quota_test EXIT
+
+       # make sure the system is clean
+       local USED=$(getquota -u $TSTID global curspace)
+       [ $USED -ne 0 ] &&
+               error "Used space ($USED) for user $TSTID isn't 0."
+       USED=$(getquota -u $TSTID2 global curspace)
+       [ $USED -ne 0 ] &&
+               error "Used space ($USED) for user $TSTID2 isn't 0."
+
+       local TESTFILE="$DIR/$tdir/$tfile"
+       local file_cnt=10000
+
+       # Generate id entries in accounting file
+       echo "Create $file_cnt files..."
+       for i in `seq $file_cnt`; do
+               touch $TESTFILE-$i
+               chown $((file_cnt - i)):$((file_cnt - i)) $TESTFILE-$i
+       done
+       cancel_lru_locks osc
+       sync; sync_all_data || true
+
+       local procf="osd-$(facet_fstype $SINGLEMDS).$FSNAME-MDT0000"
+       procf=${procf}.quota_slave.acct_user
+       local accnt_cnt
+
+       acct_cnt=$(do_facet mds1 $LCTL get_param $procf | grep "id:" | wc -l)
+       echo "Found $acct_cnt id entries"
+
+       [ $file_cnt -eq $acct_cnt ] || error "skipped id entries"
+
+       cleanup_quota_test
+}
+run_test 38 "Quota accounting iterator doesn't skip id entries"
+
 quota_fini()
 {
        do_nodes $(comma_list $(nodes_list)) "lctl set_param debug=-quota"