EX-7601 tgt: add remote_pages for writes

When we round a write to get all of the compressed chunks,
the number of local and the number of remote pages will
differ.  We need to make sure we do the checksum and data
transfer using the number of remote pages, not the number of
local pages.

This patch calculates the number of remote pages and uses it
accordingly.

Note that just like on the read side, this patch doesn't do
anything until we're actually rounding the chunks for IO in
a later patch.

Signed-off-by: Patrick Farrell <pfarrell@whamcloud.com>
Change-Id: I38256070d68246613ce67b0bfe328f6443a95533
Reviewed-on: https://review.whamcloud.com/c/ex/lustre-release/+/52964
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Artem Blagodarenko <ablagodarenko@ddn.com>

diff --git a/lustre/target/tgt_handler.c b/lustre/target/tgt_handler.c
index 1963c1d..2085383 100644 (file)
--- a/lustre/target/tgt_handler.c
+++ b/lustre/target/tgt_handler.c
@@ -2816,17 +2816,18 @@ int tgt_brw_write(struct tgt_session_info *tsi)
        const char *obd_name = exp->exp_obd->obd_name;
        enum cksum_types cksum_type = OBD_CKSUM_CRC32;
        bool compr_rounded_write_lock = false;
+       enum ll_compr_type compr_type;
        /* '1' for consistency with code that checks !mpflag to restore */
        unsigned int mpflags = 1;
-       enum ll_compr_type type;
        bool wait_sync = false;
        bool no_reply = false;
+       int npages_remote = 0;
        int chunk_size = 0;
+       int npages_local;
        ktime_t kstart;
        int objcount;
        int niocount;
        int nob = 0;
-       int npages;
        int rc = 0;
        __u32 *rcs;
        bool mmap;
@@ -2925,8 +2926,8 @@ int tgt_brw_write(struct tgt_session_info *tsi)
        local_nb = tbc->tbc_lnb;
 
        olc = &body->oa.o_layout_compr;
-       type = olc->ol_compr_type;
-       if (type != LL_COMPR_TYPE_NONE) {
+       compr_type = olc->ol_compr_type;
+       if (compr_type != LL_COMPR_TYPE_NONE) {
                int nrbufs = ioo->ioo_bufcnt;
                unsigned int chunk_log_bits;
                __u64 chunk_start;
@@ -3002,13 +3003,24 @@ int tgt_brw_write(struct tgt_session_info *tsi)
                GOTO(out_lock, rc = -ENOMEM);
        repbody->oa = body->oa;
 
-       npages = PTLRPC_MAX_BRW_PAGES;
+       npages_local = PTLRPC_MAX_BRW_PAGES;
+       for (i = 0; i < niocount; i++)
+               npages_remote += range_to_page_count(remote_nb[i].rnb_offset,
+                                                    remote_nb[i].rnb_len);
        kstart = ktime_get();
        rc = obd_preprw(tsi->tsi_env, OBD_BRW_WRITE, exp, &repbody->oa,
-                       objcount, ioo, remote_nb, &npages, local_nb,
+                       objcount, ioo, remote_nb, &npages_local, local_nb,
                        chunk_size);
        if (rc < 0)
                GOTO(out_lock, rc);
+
+       if (compr_type == LL_COMPR_TYPE_NONE) {
+               /* if there's no compression, the local page count should be
+                * identical to that requested by the client
+                */
+               LASSERT(npages_local == npages_remote);
+       }
+
        if (body->oa.o_valid & OBD_MD_FLFLAGS &&
            body->oa.o_flags & OBD_FL_SHORT_IO) {
                unsigned int short_io_size;
@@ -3023,11 +3035,12 @@ int tgt_brw_write(struct tgt_session_info *tsi)
                               " size = %d\n", short_io_size);
 
                /* Copy short io buf to pages */
-               rc = tgt_shortio2pages(local_nb, npages, short_io_buf,
+               rc = tgt_shortio2pages(local_nb, npages_remote, short_io_buf,
                                       short_io_size);
                desc = NULL;
        } else {
-               desc = ptlrpc_prep_bulk_exp(req, npages, ioobj_max_brw_get(ioo),
+               desc = ptlrpc_prep_bulk_exp(req, npages_remote,
+                                           ioobj_max_brw_get(ioo),
                                            PTLRPC_BULK_GET_SINK,
                                            OST_BULK_PORTAL,
                                            &ptlrpc_bulk_kiov_nopin_ops);
@@ -3035,7 +3048,7 @@ int tgt_brw_write(struct tgt_session_info *tsi)
                        GOTO(skip_transfer, rc = -ENOMEM);
 
                /* NB Having prepped, we must commit... */
-               for (i = 0; i < npages; i++) {
+               for (i = 0; i < npages_remote; i++) {
                        CDEBUG(D_SEC, "adding frag, page %d, offset %lu, len %d\n",
                               i, local_nb[i].lnb_page_offset & ~PAGE_MASK,
                               local_nb[i].lnb_len);
@@ -3067,7 +3080,7 @@ skip_transfer:
                                                           cksum_type);
 
                rc = tgt_checksum_niobuf_rw(tsi->tsi_tgt, cksum_type,
-                                           local_nb, npages, OST_WRITE,
+                                           local_nb, npages_remote, OST_WRITE,
                                            &repbody->oa.o_cksum, false);
                if (rc < 0)
                        GOTO(out_commitrw, rc);
@@ -3078,7 +3091,7 @@ skip_transfer:
                        mmap = (body->oa.o_valid & OBD_MD_FLFLAGS &&
                                body->oa.o_flags & OBD_FL_MMAP);
 
-                       tgt_warn_on_cksum(req, desc, local_nb, npages,
+                       tgt_warn_on_cksum(req, desc, local_nb, npages_remote,
                                          body->oa.o_cksum,
                                          repbody->oa.o_cksum, mmap);
                        cksum_counter = 0;
@@ -3108,8 +3121,8 @@ out_commitrw:
 
        /* Must commit after prep above in all cases */
        rc = obd_commitrw(tsi->tsi_env, OBD_BRW_WRITE, exp, &repbody->oa,
-                         objcount, ioo, remote_nb, npages, local_nb, rc, nob,
-                         kstart);
+                         objcount, ioo, remote_nb, npages_local, local_nb,
+                         rc, nob, kstart);
        if (rc == -ENOTCONN)
                /* quota acquire process has been given up because
                 * either the client has been evicted or the client
@@ -3137,7 +3150,7 @@ out_commitrw:
 
                        rcs[i] = 0;
                        do {
-                               LASSERT(j < npages);
+                               LASSERT(j < npages_remote);
                                if (local_nb[j].lnb_rc < 0)
                                        rcs[i] = local_nb[j].lnb_rc;
                                len -= local_nb[j].lnb_len;
@@ -3145,7 +3158,7 @@ out_commitrw:
                        } while (len > 0);
                        LASSERT(len == 0);
                }
-               LASSERT(j == npages);
+               LASSERT(j == npages_remote);
                ptlrpc_lprocfs_brw(req, nob);
        }
 out_lock: