LU-8460 osc: max_pages_per_rpc should be chunk size aligned

[fs/lustre-release.git] / lustre / osc / osc_request.c

diff --git a/lustre/osc/osc_request.c b/lustre/osc/osc_request.c
index 9b2a805..1cc95fa 100644 (file)
--- a/lustre/osc/osc_request.c
+++ b/lustre/osc/osc_request.c
@@ -937,20 +937,25 @@ static void osc_init_grant(struct client_obd *cli, struct obd_connect_data *ocd)
 
        if (OCD_HAS_FLAG(ocd, GRANT_PARAM)) {
                u64 size;
+               int chunk_mask;
 
                /* overhead for each extent insertion */
                cli->cl_grant_extent_tax = ocd->ocd_grant_tax_kb << 10;
                /* determine the appropriate chunk size used by osc_extent. */
-               cli->cl_chunkbits = max_t(int, PAGE_CACHE_SHIFT,
+               cli->cl_chunkbits = max_t(int, PAGE_SHIFT,
                                          ocd->ocd_grant_blkbits);
+               /* max_pages_per_rpc must be chunk aligned */
+               chunk_mask = ~((1 << (cli->cl_chunkbits - PAGE_SHIFT)) - 1);
+               cli->cl_max_pages_per_rpc = (cli->cl_max_pages_per_rpc +
+                                            ~chunk_mask) & chunk_mask;
                /* determine maximum extent size, in #pages */
                size = (u64)ocd->ocd_grant_max_blks << ocd->ocd_grant_blkbits;
-               cli->cl_max_extent_pages = size >> PAGE_CACHE_SHIFT;
+               cli->cl_max_extent_pages = size >> PAGE_SHIFT;
                if (cli->cl_max_extent_pages == 0)
                        cli->cl_max_extent_pages = 1;
        } else {
                cli->cl_grant_extent_tax = 0;
-               cli->cl_chunkbits = PAGE_CACHE_SHIFT;
+               cli->cl_chunkbits = PAGE_SHIFT;
                cli->cl_max_extent_pages = DT_MAX_BRW_PAGES;
        }
        spin_unlock(&cli->cl_loi_list_lock);
@@ -1391,7 +1396,8 @@ static int osc_brw_fini_request(struct ptlrpc_request *req, int rc)
         /* set/clear over quota flag for a uid/gid */
         if (lustre_msg_get_opc(req->rq_reqmsg) == OST_WRITE &&
             body->oa.o_valid & (OBD_MD_FLUSRQUOTA | OBD_MD_FLGRPQUOTA)) {
-                unsigned int qid[MAXQUOTAS] = { body->oa.o_uid, body->oa.o_gid };
+               unsigned int qid[LL_MAXQUOTAS] =
+                                       {body->oa.o_uid, body->oa.o_gid};
 
                 CDEBUG(D_QUOTA, "setdq for [%u %u] with valid "LPX64", flags %x\n",
                        body->oa.o_uid, body->oa.o_gid, body->oa.o_valid,
@@ -1903,6 +1909,7 @@ int osc_build_rpc(const struct lu_env *env, struct client_obd *cli,
        DEBUG_REQ(D_INODE, req, "%d pages, aa %p. now %ur/%uw in flight",
                  page_count, aa, cli->cl_r_in_flight,
                  cli->cl_w_in_flight);
+       OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_DELAY_IO, 4);
 
        ptlrpcd_add_req(req);
        rc = 0;
@@ -1931,17 +1938,11 @@ out:
        RETURN(rc);
 }
 
-static int osc_set_lock_data_with_check(struct ldlm_lock *lock,
-                                        struct ldlm_enqueue_info *einfo)
+static int osc_set_lock_data(struct ldlm_lock *lock, void *data)
 {
-        void *data = einfo->ei_cbdata;
         int set = 0;
 
         LASSERT(lock != NULL);
-        LASSERT(lock->l_blocking_ast == einfo->ei_cb_bl);
-        LASSERT(lock->l_resource->lr_type == einfo->ei_type);
-        LASSERT(lock->l_completion_ast == einfo->ei_cb_cp);
-        LASSERT(lock->l_glimpse_ast == einfo->ei_cb_gl);
 
         lock_res_and_lock(lock);
 
@@ -1955,21 +1956,6 @@ static int osc_set_lock_data_with_check(struct ldlm_lock *lock,
        return set;
 }
 
-static int osc_set_data_with_check(struct lustre_handle *lockh,
-                                   struct ldlm_enqueue_info *einfo)
-{
-        struct ldlm_lock *lock = ldlm_handle2lock(lockh);
-        int set = 0;
-
-        if (lock != NULL) {
-                set = osc_set_lock_data_with_check(lock, einfo);
-                LDLM_LOCK_PUT(lock);
-        } else
-                CERROR("lockh %p, data %p - client evicted?\n",
-                       lockh, einfo->ei_cbdata);
-        return set;
-}
-
 static int osc_enqueue_fini(struct ptlrpc_request *req,
                            osc_enqueue_upcall_f upcall, void *cookie,
                            struct lustre_handle *lockh, enum ldlm_mode mode,
@@ -2135,7 +2121,7 @@ int osc_enqueue_base(struct obd_export *exp, struct ldlm_res_id *res_id,
                        ldlm_lock_decref(&lockh, mode);
                        LDLM_LOCK_PUT(matched);
                        RETURN(-ECANCELED);
-               } else if (osc_set_lock_data_with_check(matched, einfo)) {
+               } else if (osc_set_lock_data(matched, einfo->ei_cbdata)) {
                        *flags |= LDLM_FL_LVB_READY;
 
                        /* We already have a lock, and it's referenced. */
@@ -2246,21 +2232,20 @@ int osc_match_base(struct obd_export *exp, struct ldlm_res_id *res_id,
                 rc |= LCK_PW;
         rc = ldlm_lock_match(obd->obd_namespace, lflags,
                              res_id, type, policy, rc, lockh, unref);
-        if (rc) {
-                if (data != NULL) {
-                        if (!osc_set_data_with_check(lockh, data)) {
-                                if (!(lflags & LDLM_FL_TEST_LOCK))
-                                        ldlm_lock_decref(lockh, rc);
-                                RETURN(0);
-                        }
-                }
-                if (!(lflags & LDLM_FL_TEST_LOCK) && mode != rc) {
-                        ldlm_lock_addref(lockh, LCK_PR);
-                        ldlm_lock_decref(lockh, LCK_PW);
-                }
-                RETURN(rc);
-        }
-        RETURN(rc);
+       if (rc == 0 || lflags & LDLM_FL_TEST_LOCK)
+               RETURN(rc);
+
+       if (data != NULL) {
+               struct ldlm_lock *lock = ldlm_handle2lock(lockh);
+
+               LASSERT(lock != NULL);
+               if (!osc_set_lock_data(lock, data)) {
+                       ldlm_lock_decref(lockh, rc);
+                       rc = 0;
+               }
+               LDLM_LOCK_PUT(lock);
+       }
+       RETURN(rc);
 }
 
 static int osc_statfs_interpret(const struct lu_env *env,