LU-5278 echo: request pages in batches

rather than fetch them one by one.

Change-Id: I55dc5066402d1a6393f8edfe8cf71892a397339c
Signed-off-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Reviewed-on: http://review.whamcloud.com/13612
Reviewed-by: Nathaniel Clark <nathaniel.l.clark@intel.com>
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>

diff --git a/lustre/obdecho/echo_client.c b/lustre/obdecho/echo_client.c
index 5388768..3aa7759 100644 (file)
--- a/lustre/obdecho/echo_client.c
+++ b/lustre/obdecho/echo_client.c
@@ -2415,7 +2415,7 @@ static int echo_client_prep_commit(const struct lu_env *env,
 {
        struct obd_ioobj         ioo;
        struct niobuf_local     *lnb;
-       struct niobuf_remote    *rnb;
+       struct niobuf_remote     rnb;
        u64                      off;
        u64                      npages, tot_pages;
        int i, ret = 0, brw_flags = 0;
@@ -2428,40 +2428,35 @@ static int echo_client_prep_commit(const struct lu_env *env,
        npages = batch >> PAGE_CACHE_SHIFT;
        tot_pages = count >> PAGE_CACHE_SHIFT;
 
-        OBD_ALLOC(lnb, npages * sizeof(struct niobuf_local));
-        OBD_ALLOC(rnb, npages * sizeof(struct niobuf_remote));
-
-        if (lnb == NULL || rnb == NULL)
-                GOTO(out, ret = -ENOMEM);
+       OBD_ALLOC(lnb, npages * sizeof(struct niobuf_local));
+       if (lnb == NULL)
+               GOTO(out, ret = -ENOMEM);
 
        if (rw == OBD_BRW_WRITE && async)
                brw_flags |= OBD_BRW_ASYNC;
 
-        obdo_to_ioobj(oa, &ioo);
-
-        off = offset;
+       obdo_to_ioobj(oa, &ioo);
 
-        for(; tot_pages; tot_pages -= npages) {
-                int lpages;
+       off = offset;
 
-                if (tot_pages < npages)
-                        npages = tot_pages;
+       for (; tot_pages > 0; tot_pages -= npages) {
+               int lpages;
 
-               for (i = 0; i < npages; i++, off += PAGE_CACHE_SIZE) {
-                       rnb[i].rnb_offset = off;
-                       rnb[i].rnb_len = PAGE_CACHE_SIZE;
-                       rnb[i].rnb_flags = brw_flags;
-                }
+               if (tot_pages < npages)
+                       npages = tot_pages;
 
-                ioo.ioo_bufcnt = npages;
+               rnb.rnb_offset = off;
+               rnb.rnb_len = npages * PAGE_CACHE_SIZE;
+               rnb.rnb_flags = brw_flags;
+               ioo.ioo_bufcnt = 1;
+               off += npages * PAGE_CACHE_SIZE;
 
                 lpages = npages;
-               ret = obd_preprw(env, rw, exp, oa, 1, &ioo, rnb, &lpages, lnb);
+               ret = obd_preprw(env, rw, exp, oa, 1, &ioo, &rnb, &lpages, lnb);
                 if (ret != 0)
                         GOTO(out, ret);
-                LASSERT(lpages == npages);
 
-                for (i = 0; i < lpages; i++) {
+               for (i = 0; i < lpages; i++) {
                        struct page *page = lnb[i].lnb_page;
 
                        /* read past eof? */
@@ -2478,17 +2473,17 @@ static int echo_client_prep_commit(const struct lu_env *env,
 
                        if (rw == OBD_BRW_WRITE)
                                echo_client_page_debug_setup(page, rw,
-                                                           ostid_id(&oa->o_oi),
-                                                            rnb[i].rnb_offset,
-                                                            rnb[i].rnb_len);
+                                                       ostid_id(&oa->o_oi),
+                                                       lnb[i].lnb_file_offset,
+                                                       lnb[i].lnb_len);
                        else
                                echo_client_page_debug_check(page,
-                                                           ostid_id(&oa->o_oi),
-                                                            rnb[i].rnb_offset,
-                                                            rnb[i].rnb_len);
+                                                       ostid_id(&oa->o_oi),
+                                                       lnb[i].lnb_file_offset,
+                                                       lnb[i].lnb_len);
                }
 
-               ret = obd_commitrw(env, rw, exp, oa, 1, &ioo, rnb, npages, lnb,
+               ret = obd_commitrw(env, rw, exp, oa, 1, &ioo, &rnb, npages, lnb,
                                   ret);
                 if (ret != 0)
                         GOTO(out, ret);
@@ -2496,14 +2491,12 @@ static int echo_client_prep_commit(const struct lu_env *env,
                /* Reuse env context. */
                lu_context_exit((struct lu_context *)&env->le_ctx);
                lu_context_enter((struct lu_context *)&env->le_ctx);
-        }
+       }
 
 out:
-        if (lnb)
-                OBD_FREE(lnb, npages * sizeof(struct niobuf_local));
-        if (rnb)
-                OBD_FREE(rnb, npages * sizeof(struct niobuf_remote));
-        RETURN(ret);
+       if (lnb)
+               OBD_FREE(lnb, npages * sizeof(struct niobuf_local));
+       RETURN(ret);
 }
 
 static int echo_client_brw_ioctl(const struct lu_env *env, int rw,