LU-1187 osd: add remote entry insert for ZFS DNE.

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

diff --git a/lustre/quota/qsd_handler.c b/lustre/quota/qsd_handler.c
index 68803d0..805be89 100644 (file)
--- a/lustre/quota/qsd_handler.c
+++ b/lustre/quota/qsd_handler.c
@@ -21,7 +21,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2012 Intel, Inc.
+ * Copyright (c) 2012, Intel Corporation.
  * Use is subject to license terms.
  *
  * Author: Johann Lombardi <johann.lombardi@intel.com>
@@ -94,10 +94,10 @@ static int qsd_ready(struct lquota_entry *lqe, struct lustre_handle *lockh)
        struct ldlm_lock        *lock;
        ENTRY;
 
-       cfs_read_lock(&qsd->qsd_lock);
+       read_lock(&qsd->qsd_lock);
        /* is the qsd about to shut down? */
        if (qsd->qsd_stopping) {
-               cfs_read_unlock(&qsd->qsd_lock);
+               read_unlock(&qsd->qsd_lock);
                LQUOTA_DEBUG(lqe, "dropping quota req since qsd is stopping");
                /* Target is about to shut down, client will retry */
                RETURN(-EINPROGRESS);
@@ -107,7 +107,7 @@ static int qsd_ready(struct lquota_entry *lqe, struct lustre_handle *lockh)
        if (qsd->qsd_exp_valid)
                imp = class_exp2cliimp(qsd->qsd_exp);
        if (imp == NULL || imp->imp_invalid) {
-               cfs_read_unlock(&qsd->qsd_lock);
+               read_unlock(&qsd->qsd_lock);
                LQUOTA_DEBUG(lqe, "connection to master not ready");
                RETURN(-ENOTCONN);
        }
@@ -120,7 +120,7 @@ static int qsd_ready(struct lquota_entry *lqe, struct lustre_handle *lockh)
         * If the previous reintegration failed for some reason, we'll
         * re-trigger it here as well. */
        if (!qqi->qqi_glb_uptodate || !qqi->qqi_slv_uptodate) {
-               cfs_read_unlock(&qsd->qsd_lock);
+               read_unlock(&qsd->qsd_lock);
                LQUOTA_DEBUG(lqe, "not up-to-date, dropping request and "
                             "kicking off reintegration");
                qsd_start_reint_thread(qqi);
@@ -130,7 +130,7 @@ static int qsd_ready(struct lquota_entry *lqe, struct lustre_handle *lockh)
        /* Fill the remote global lock handle, master will check this handle
         * to see if the slave is sending request with stale lock */
        lustre_handle_copy(lockh, &qqi->qqi_lockh);
-       cfs_read_unlock(&qsd->qsd_lock);
+       read_unlock(&qsd->qsd_lock);
 
        if (!lustre_handle_is_used(lockh))
                RETURN(-ENOLCK);
@@ -308,7 +308,7 @@ static void qsd_req_completion(const struct lu_env *env,
                               struct quota_body *reqbody,
                               struct quota_body *repbody,
                               struct lustre_handle *lockh,
-                              union ldlm_wire_lvb *lvb,
+                              struct lquota_lvb *lvb,
                               void *arg, int ret)
 {
        struct lquota_entry     *lqe = (struct lquota_entry *)arg;
@@ -390,9 +390,9 @@ static void qsd_req_completion(const struct lu_env *env,
 
        /* extract information from lvb */
        if (ret == 0 && lvb != 0) {
-               if (lvb->l_lquota.lvb_id_qunit != 0)
-                       qsd_set_qunit(lqe, lvb->l_lquota.lvb_id_qunit);
-               if (lvb->l_lquota.lvb_flags & LQUOTA_FL_EDQUOT)
+               if (lvb->lvb_id_qunit != 0)
+                       qsd_set_qunit(lqe, lvb->lvb_id_qunit);
+               if (lvb->lvb_flags & LQUOTA_FL_EDQUOT)
                        lqe->lqe_edquot = true;
                else
                        lqe->lqe_edquot = false;
@@ -407,6 +407,10 @@ static void qsd_req_completion(const struct lu_env *env,
                lqe->lqe_nopreacq = true;
 out:
        adjust = qsd_adjust_needed(lqe);
+       if (reqbody && req_is_acq(reqbody->qb_flags) && ret != -EDQUOT) {
+               lqe->lqe_acq_rc = ret;
+               lqe->lqe_acq_time = cfs_time_current_64();
+       }
 out_noadjust:
        qsd_request_exit(lqe);
        lqe_write_unlock(lqe);
@@ -558,6 +562,14 @@ static int qsd_acquire_remote(const struct lu_env *env,
                RETURN(0);
        }
 
+       /* check whether an acquire request completed recently */
+       if (lqe->lqe_acq_rc != 0 &&
+           cfs_time_before_64(cfs_time_shift_64(-1), lqe->lqe_acq_time)) {
+               lqe_write_unlock(lqe);
+               LQUOTA_DEBUG(lqe, "using cached return code %d", lqe->lqe_acq_rc);
+               RETURN(lqe->lqe_acq_rc);
+       }
+
        /* only 1 quota request in flight for a given ID is allowed */
        rc = qsd_request_enter(lqe);
        if (rc) {
@@ -578,7 +590,7 @@ static int qsd_acquire_remote(const struct lu_env *env,
        /* check whether we already own a valid lock for this ID */
        rc = qsd_id_lock_match(&qti->qti_lockh, &qbody->qb_lockh);
        if (rc) {
-               union ldlm_wire_lvb *lvb;
+               struct lquota_lvb *lvb;
 
                OBD_ALLOC_PTR(lvb);
                if (lvb == NULL) {
@@ -823,12 +835,12 @@ int qsd_op_begin(const struct lu_env *env, struct qsd_instance *qsd,
                RETURN(0);
 
        /* We don't enforce quota until the qsd_instance is started */
-       cfs_read_lock(&qsd->qsd_lock);
+       read_lock(&qsd->qsd_lock);
        if (!qsd->qsd_started) {
-               cfs_read_unlock(&qsd->qsd_lock);
+               read_unlock(&qsd->qsd_lock);
                RETURN(0);
        }
-       cfs_read_unlock(&qsd->qsd_lock);
+       read_unlock(&qsd->qsd_lock);
 
        /* ignore block quota on MDTs, ignore inode quota on OSTs */
        if ((!qsd->qsd_is_md && !qi->lqi_is_blk) ||
@@ -898,8 +910,13 @@ int qsd_adjust(const struct lu_env *env, struct lquota_entry *lqe)
 
        memset(qbody, 0, sizeof(*qbody));
        rc = qsd_ready(lqe, &qbody->qb_glb_lockh);
-       if (rc)
-               RETURN(rc);
+       if (rc) {
+               /* add to adjust list again to trigger adjustment later when
+                * slave is ready */
+               LQUOTA_DEBUG(lqe, "delaying adjustment since qsd isn't ready");
+               qsd_adjust_schedule(lqe, true, false);
+               RETURN(0);
+       }
 
        qqi = lqe2qqi(lqe);
        qsd = qqi->qqi_qsd;
@@ -909,7 +926,7 @@ int qsd_adjust(const struct lu_env *env, struct lquota_entry *lqe)
        /* fill qb_count & qb_flags */
        if (!qsd_calc_adjust(lqe, qbody)) {
                lqe_write_unlock(lqe);
-               LQUOTA_DEBUG(lqe, "No adjust required");
+               LQUOTA_DEBUG(lqe, "no adjustment required");
                RETURN(0);
        }
 
@@ -965,7 +982,7 @@ int qsd_adjust(const struct lu_env *env, struct lquota_entry *lqe)
                                    qsd_req_completion, qqi, &qti->qti_lockh,
                                    lqe);
        } else {
-               union ldlm_wire_lvb *lvb;
+               struct lquota_lvb *lvb;
 
                OBD_ALLOC_PTR(lvb);
                if (lvb == NULL)
@@ -1058,12 +1075,12 @@ void qsd_op_end(const struct lu_env *env, struct qsd_instance *qsd,
                RETURN_EXIT;
 
        /* We don't enforce quota until the qsd_instance is started */
-       cfs_read_lock(&qsd->qsd_lock);
+       read_lock(&qsd->qsd_lock);
        if (!qsd->qsd_started) {
-               cfs_read_unlock(&qsd->qsd_lock);
+               read_unlock(&qsd->qsd_lock);
                RETURN_EXIT;
        }
-       cfs_read_unlock(&qsd->qsd_lock);
+       read_unlock(&qsd->qsd_lock);
 
        LASSERT(trans != NULL);
 
@@ -1110,12 +1127,12 @@ void qsd_op_adjust(const struct lu_env *env, struct qsd_instance *qsd,
                RETURN_EXIT;
 
        /* We don't enforce quota until the qsd_instance is started */
-       cfs_read_lock(&qsd->qsd_lock);
+       read_lock(&qsd->qsd_lock);
        if (!qsd->qsd_started) {
-               cfs_read_unlock(&qsd->qsd_lock);
+               read_unlock(&qsd->qsd_lock);
                RETURN_EXIT;
        }
-       cfs_read_unlock(&qsd->qsd_lock);
+       read_unlock(&qsd->qsd_lock);
 
        qqi = qsd->qsd_type_array[qtype];
        LASSERT(qqi);
@@ -1124,12 +1141,12 @@ void qsd_op_adjust(const struct lu_env *env, struct qsd_instance *qsd,
            qid->qid_uid == 0)
                RETURN_EXIT;
 
-       cfs_read_lock(&qsd->qsd_lock);
+       read_lock(&qsd->qsd_lock);
        if (!qsd->qsd_started) {
-               cfs_read_unlock(&qsd->qsd_lock);
+               read_unlock(&qsd->qsd_lock);
                RETURN_EXIT;
        }
-       cfs_read_unlock(&qsd->qsd_lock);
+       read_unlock(&qsd->qsd_lock);
 
        lqe = lqe_locate(env, qqi->qqi_site, qid);
        if (IS_ERR(lqe)) {