LU-16271 ptlrpc: fix eviction right after recovery

[fs/lustre-release.git] / lustre / quota / qsd_handler.c

diff --git a/lustre/quota/qsd_handler.c b/lustre/quota/qsd_handler.c
index ea75142..01433b6 100644 (file)
--- a/lustre/quota/qsd_handler.c
+++ b/lustre/quota/qsd_handler.c
@@ -857,6 +857,10 @@ int qsd_op_begin(const struct lu_env *env, struct qsd_instance *qsd,
        bool    found = false;
        ENTRY;
 
+       /* fast path, ignore quota enforcement request for root owned files */
+       if (qi->lqi_id.qid_uid == 0)
+               return 0;
+
        if (unlikely(qsd == NULL))
                RETURN(0);
 
@@ -880,7 +884,7 @@ int qsd_op_begin(const struct lu_env *env, struct qsd_instance *qsd,
         *    - quota isn't enforced for this quota type
         * or - the user/group is root
         * or - quota accounting isn't enabled */
-       if (!qsd_type_enabled(qsd, qi->lqi_type) || qi->lqi_id.qid_uid == 0 ||
+       if (!qsd_type_enabled(qsd, qi->lqi_type) ||
            (qsd->qsd_type_array[qi->lqi_type])->qqi_acct_failed)
                RETURN(0);
 
@@ -1064,8 +1068,16 @@ static void qsd_op_end0(const struct lu_env *env, struct qsd_qtype_info *qqi,
                qsd_refresh_usage(env, lqe);
 
        lqe_write_lock(lqe);
-       if (qid->lqi_space > 0)
-               lqe->lqe_pending_write -= qid->lqi_space;
+       if (qid->lqi_space > 0) {
+               if (lqe->lqe_pending_write < qid->lqi_space) {
+                       LQUOTA_ERROR(lqe,
+                                    "More pending write quota to reduce (pending %llu, space %lld)\n",
+                                    lqe->lqe_pending_write, qid->lqi_space);
+                       lqe->lqe_pending_write = 0;
+               } else {
+                       lqe->lqe_pending_write -= qid->lqi_space;
+               }
+       }
        if (env != NULL)
                adjust = qsd_adjust_needed(lqe);
        else
@@ -1136,6 +1148,72 @@ void qsd_op_end(const struct lu_env *env, struct qsd_instance *qsd,
 }
 EXPORT_SYMBOL(qsd_op_end);
 
+/* Simple wrapper on top of qsd API which implement quota transfer for osd
+ * setattr needs. As a reminder, only the root user can change ownership of
+ * a file, that's why EDQUOT & EINPROGRESS errors are discarded
+ */
+int qsd_transfer(const struct lu_env *env, struct qsd_instance *qsd,
+                struct lquota_trans *trans, unsigned int qtype,
+                u64 orig_id, u64 new_id, u64 bspace,
+                struct lquota_id_info *qi)
+{
+       int rc;
+
+       if (unlikely(!qsd))
+               return 0;
+
+       LASSERT(qtype < LL_MAXQUOTAS);
+       if (qtype == PRJQUOTA)
+               if (!projid_valid(make_kprojid(&init_user_ns, new_id)))
+                       return -EINVAL;
+
+       qi->lqi_type = qtype;
+
+       /* inode accounting */
+       qi->lqi_is_blk = false;
+
+       /* one more inode for the new owner ... */
+       qi->lqi_id.qid_uid = new_id;
+       qi->lqi_space = 1;
+       rc = qsd_op_begin(env, qsd, trans, qi, NULL);
+       if (rc == -EDQUOT || rc == -EINPROGRESS)
+               rc = 0;
+       if (rc)
+               return rc;
+
+       /* and one less inode for the current id */
+       qi->lqi_id.qid_uid = orig_id;
+       qi->lqi_space = -1;
+       /* can't get EDQUOT when reducing usage */
+       rc = qsd_op_begin(env, qsd, trans, qi, NULL);
+       if (rc == -EINPROGRESS)
+               rc = 0;
+       if (rc)
+               return rc;
+
+       /* block accounting */
+       qi->lqi_is_blk = true;
+
+       /* more blocks for the new owner ... */
+       qi->lqi_id.qid_uid = new_id;
+       qi->lqi_space = bspace;
+       rc = qsd_op_begin(env, qsd, trans, qi, NULL);
+       if (rc == -EDQUOT || rc == -EINPROGRESS)
+               rc = 0;
+       if (rc)
+               return rc;
+
+       /* and finally less blocks for the current owner */
+       qi->lqi_id.qid_uid = orig_id;
+       qi->lqi_space = -bspace;
+       rc = qsd_op_begin(env, qsd, trans, qi, NULL);
+       /* can't get EDQUOT when reducing usage */
+       if (rc == -EINPROGRESS)
+               rc = 0;
+       return rc;
+}
+EXPORT_SYMBOL(qsd_transfer);
+
 /**
  * Trigger pre-acquire/release if necessary.
  * It's only used by ldiskfs osd so far. When unlink a file in ldiskfs, the
@@ -1224,10 +1302,9 @@ int qsd_reserve_or_free_quota(const struct lu_env *env,
                              struct qsd_instance *qsd,
                              struct lquota_id_info *qi)
 {
-       struct lquota_entry *lqe;
        struct qsd_qtype_info  *qqi;
-       int rc = 0;
        bool is_free = qi->lqi_space < 0;
+       int rc = 0;
 
        ENTRY;
 
@@ -1261,31 +1338,21 @@ int qsd_reserve_or_free_quota(const struct lu_env *env,
         * or - the user/group is root
         * or - quota accounting isn't enabled
         */
-       if (!qsd_type_enabled(qsd, qi->lqi_type) || qi->lqi_id.qid_uid == 0 ||
-           (qsd->qsd_type_array[qi->lqi_type])->qqi_acct_failed)
+       if (!is_free &&
+           (!qsd_type_enabled(qsd, qi->lqi_type) || qi->lqi_id.qid_uid == 0 ||
+             (qsd->qsd_type_array[qi->lqi_type])->qqi_acct_failed))
                RETURN(0);
 
        if (is_free) {
-               /* look up lquota entry associated with qid */
-               lqe = lqe_locate(env, qqi->qqi_site, &qi->lqi_id);
-               if (IS_ERR(lqe))
-                       RETURN(PTR_ERR(lqe));
-               if (!lqe->lqe_enforced) {
-                       lqe_putref(lqe);
-                       RETURN(0);
-               }
-
-               qi->lqi_qentry = lqe;
-
-               /* lqe will be put in qsd_op_end0 */
                qsd_op_end0(env, qsd->qsd_type_array[qi->lqi_type], qi);
-               qi->lqi_qentry = NULL;
        } else {
-               /* manage quota enforcement for this ID */
-               rc = qsd_op_begin0(env, qsd->qsd_type_array[qi->lqi_type], qi,
-                                  qi->lqi_space, NULL);
+               long long qspace = qi->lqi_space;
 
-               if (qi->lqi_qentry != NULL) {
+               /* the acquired quota will add to lqi_space in qsd_op_begin0 */
+               qi->lqi_space = 0;
+               rc = qsd_op_begin0(env, qsd->qsd_type_array[qi->lqi_type], qi,
+                                  qspace, NULL);
+               if (rc && qi->lqi_qentry) {
                        lqe_putref(qi->lqi_qentry);
                        qi->lqi_qentry = NULL;
                }