LU-7830 ost: do not evict during truncate

[fs/lustre-release.git] / lustre / ofd / ofd_dev.c

diff --git a/lustre/ofd/ofd_dev.c b/lustre/ofd/ofd_dev.c
index d48472b..0a2ec19 100644 (file)
--- a/lustre/ofd/ofd_dev.c
+++ b/lustre/ofd/ofd_dev.c
@@ -2021,6 +2021,8 @@ static int ofd_punch_hdl(struct tgt_session_info *tsi)
 
        ENTRY;
 
+       OBD_FAIL_TIMEOUT(OBD_FAIL_OST_PAUSE_PUNCH, cfs_fail_val);
+
        /* check that we do support OBD_CONNECT_TRUNCLOCK. */
        CLASSERT(OST_CONNECT_SUPPORTED & OBD_CONNECT_TRUNCLOCK);
 
@@ -2106,6 +2108,97 @@ out:
 }
 
 /**
+ * OFD request handler for OST_LADVISE RPC.
+ *
+ * Tune cache or perfetch policies according to advices.
+ *
+ * \param[in] tsi      target session environment for this request
+ *
+ * \retval             0 if successful
+ * \retval             negative errno on error
+ */
+static int ofd_ladvise_hdl(struct tgt_session_info *tsi)
+{
+       struct ptlrpc_request   *req = tgt_ses_req(tsi);
+       struct obd_export       *exp = tsi->tsi_exp;
+       struct ofd_device       *ofd = ofd_exp(exp);
+       struct ost_body         *body, *repbody;
+       struct ofd_thread_info  *info;
+       struct ofd_object       *fo;
+       const struct lu_env     *env = req->rq_svc_thread->t_env;
+       int                      rc = 0;
+       struct lu_ladvise       *ladvise;
+       int                      num_advise;
+       struct ladvise_hdr      *ladvise_hdr;
+       int                      i;
+       ENTRY;
+
+       body = tsi->tsi_ost_body;
+
+       if ((body->oa.o_valid & OBD_MD_FLID) != OBD_MD_FLID)
+               RETURN(err_serious(-EPROTO));
+
+       ladvise_hdr = req_capsule_client_get(tsi->tsi_pill,
+                                            &RMF_OST_LADVISE_HDR);
+       if (ladvise_hdr == NULL)
+               RETURN(err_serious(-EPROTO));
+
+       if (ladvise_hdr->lah_magic != LADVISE_MAGIC ||
+           ladvise_hdr->lah_count < 1)
+               RETURN(err_serious(-EPROTO));
+
+       if ((ladvise_hdr->lah_flags & (~LF_MASK)) != 0)
+               RETURN(err_serious(-EPROTO));
+
+       ladvise = req_capsule_client_get(tsi->tsi_pill, &RMF_OST_LADVISE);
+       if (ladvise == NULL)
+               RETURN(err_serious(-EPROTO));
+
+       num_advise = req_capsule_get_size(&req->rq_pill,
+                                         &RMF_OST_LADVISE, RCL_CLIENT) /
+                    sizeof(*ladvise);
+       if (num_advise < ladvise_hdr->lah_count)
+               RETURN(err_serious(-EPROTO));
+
+       repbody = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
+       repbody->oa = body->oa;
+
+       info = ofd_info_init(env, exp);
+
+       rc = ostid_to_fid(&info->fti_fid, &body->oa.o_oi,
+                         ofd->ofd_lut.lut_lsd.lsd_osd_index);
+       if (rc != 0)
+               RETURN(rc);
+
+       fo = ofd_object_find(env, ofd, &info->fti_fid);
+       if (IS_ERR(fo)) {
+               rc = PTR_ERR(fo);
+               RETURN(rc);
+       }
+       LASSERT(fo != NULL);
+
+       for (i = 0; i < num_advise; i++, ladvise++) {
+               if (ladvise->lla_end <= ladvise->lla_start) {
+                       rc = err_serious(-EPROTO);
+                       break;
+               }
+
+               /* Handle different advice types */
+               switch (ladvise->lla_advice) {
+               default:
+                       rc = -ENOTSUPP;
+                       break;
+               }
+               if (rc != 0)
+                       break;
+       }
+
+       ofd_object_put(env, fo);
+       req->rq_status = rc;
+       RETURN(rc);
+}
+
+/**
  * OFD request handler for OST_QUOTACTL RPC.
  *
  * This is part of request processing to validate incoming request fields,
@@ -2118,11 +2211,10 @@ out:
  */
 static int ofd_quotactl(struct tgt_session_info *tsi)
 {
-       struct obd_quotactl     *oqctl, *repoqc;
-       struct lu_nodemap       *nodemap =
-               tsi->tsi_exp->exp_target_data.ted_nodemap;
-       int                      id;
-       int                      rc;
+       struct obd_quotactl *oqctl, *repoqc;
+       struct lu_nodemap *nodemap;
+       int id;
+       int rc;
 
        ENTRY;
 
@@ -2136,6 +2228,10 @@ static int ofd_quotactl(struct tgt_session_info *tsi)
 
        *repoqc = *oqctl;
 
+       nodemap = nodemap_get_from_exp(tsi->tsi_exp);
+       if (IS_ERR(nodemap))
+               RETURN(PTR_ERR(nodemap));
+
        id = repoqc->qc_id;
        if (oqctl->qc_type == USRQUOTA)
                id = nodemap_map_id(nodemap, NODEMAP_UID,
@@ -2146,6 +2242,8 @@ static int ofd_quotactl(struct tgt_session_info *tsi)
                                    NODEMAP_CLIENT_TO_FS,
                                    repoqc->qc_id);
 
+       nodemap_putref(nodemap);
+
        if (repoqc->qc_id != id)
                swap(repoqc->qc_id, id);
 
@@ -2450,6 +2548,10 @@ static int ofd_punch_hpreq_lock_match(struct ptlrpc_request *req,
                                      struct ldlm_lock *lock)
 {
        struct tgt_session_info *tsi;
+       struct obdo             *oa;
+       struct ldlm_extent       ext;
+
+       ENTRY;
 
        /* Don't use tgt_ses_info() to get session info, because lock_match()
         * can be called while request has no processing thread yet. */
@@ -2462,9 +2564,20 @@ static int ofd_punch_hpreq_lock_match(struct ptlrpc_request *req,
        LASSERT(tsi->tsi_ost_body != NULL);
        if (tsi->tsi_ost_body->oa.o_valid & OBD_MD_FLHANDLE &&
            tsi->tsi_ost_body->oa.o_handle.cookie == lock->l_handle.h_cookie)
-               return 1;
+               RETURN(1);
 
-       return 0;
+       oa = &tsi->tsi_ost_body->oa;
+       ext.start = oa->o_size;
+       ext.end   = oa->o_blocks;
+
+       LASSERT(lock->l_resource != NULL);
+       if (!ostid_res_name_eq(&oa->o_oi, &lock->l_resource->lr_name))
+               RETURN(0);
+
+       if (!(lock->l_granted_mode & (LCK_PW | LCK_GROUP)))
+               RETURN(0);
+
+       RETURN(ldlm_extent_overlap(&lock->l_policy_data.l_extent, &ext));
 }
 
 /**
@@ -2624,6 +2737,7 @@ TGT_OST_HDL_HP(HABEO_CORPUS| HABEO_REFERO | MUTABOR,
                                                        ofd_hp_punch),
 TGT_OST_HDL(HABEO_CORPUS| HABEO_REFERO,        OST_SYNC,       ofd_sync_hdl),
 TGT_OST_HDL(0          | HABEO_REFERO, OST_QUOTACTL,   ofd_quotactl),
+TGT_OST_HDL(HABEO_CORPUS | HABEO_REFERO, OST_LADVISE,  ofd_ladvise_hdl),
 };
 
 static struct tgt_opc_slice ofd_common_slice[] = {
@@ -2751,6 +2865,7 @@ static int ofd_init0(const struct lu_env *env, struct ofd_device *m,
        ofd_slc_set(m);
        m->ofd_grant_compat_disable = 0;
        m->ofd_soft_sync_limit = OFD_SOFT_SYNC_LIMIT_DEFAULT;
+       m->ofd_brw_size = ONE_MB_BRW_SIZE;
 
        /* statfs data */
        spin_lock_init(&m->ofd_osfs_lock);