EX-7818 osc: don't check for start inside the chunk

Chunk size is the same for the whole request and every
chunk offset is multiple to a chunk size.

No need to search for compression header in every page.
It is enough to check every with offset multiple to a chunk
size.

Test-Parameters: testlist=sanity-compr env=ONLY="1-6"
Signed-off-by: Artem Blagodarenko <ablagodarenko@ddn.com>
Signed-off-by: Patrick Farrell <pfarrell@whamcloud.com>
Change-Id: Ie2ef645130656279e152ea1f7e6db01cb33836ca
Reviewed-on: https://review.whamcloud.com/c/ex/lustre-release/+/51650
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>

diff --git a/contrib/scripts/spelling.txt b/contrib/scripts/spelling.txt
index 22e1097..3a80486 100644 (file)
--- a/contrib/scripts/spelling.txt
+++ b/contrib/scripts/spelling.txt
@@ -98,6 +98,7 @@ cfs_time_current||jiffies
 cfs_time_current_64||ktime_get
 cfs_time_current_sec||ktime_get_real_seconds
 CLASSERT||BUILD_BUG_ON()
+cp_chunk_log_bits||consider cp_compr_chunk_bits helper?
 msecs_to_jiffies||cfs_time_seconds
 DEFINE_TIMER||CFS_DEFINE_TIMER
 define OBD_CONNECT||see "XXX README XXX" below and contact adilger@whamcloud.com

diff --git a/lustre/include/cl_object.h b/lustre/include/cl_object.h
index 965160c..1475adf 100644 (file)
--- a/lustre/include/cl_object.h
+++ b/lustre/include/cl_object.h
@@ -805,6 +805,8 @@ struct cl_page {
        /** Assigned if doing a sync_io */
        struct cl_sync_io       *cp_sync_io;
 };
+#define cl_page_compr_bits(clpage) \
+       (clpage->cp_chunk_log_bits + COMPR_CHUNK_MIN_BITS)
 
 /**
  * Per-layer part of cl_page.

diff --git a/lustre/osc/osc_compress.c b/lustre/osc/osc_compress.c
index 3c62553..cfccf8e 100644 (file)
--- a/lustre/osc/osc_compress.c
+++ b/lustre/osc/osc_compress.c
@@ -317,27 +317,31 @@ int compress_request(const char *obd_name, struct obdo *oa,
                     struct brw_page **pga, struct brw_page ***cpga,
                     u32 page_count, int *pcount)
 {
-       int chunk_size;
-       int pages_in_chunk;
-       int pga_i;
-       int cpga_i = 0;
-       int chunk_start = 0;
-       void *src = NULL;
-       void *dst = NULL;
+       struct cl_page *clpage;
+       unsigned int applied_type = LL_COMPR_TYPE_UNCHANGED;
+       enum cpga_fill_bits fill_bits = 0;
        unsigned int src_size;
        unsigned int dst_size;
+       int chunk_start = 0;
        void *wrkmem = NULL;
+       int pages_in_chunk;
+       int dest_buf_bits;
+       int src_buf_bits;
+       void *src = NULL;
+       void *dst = NULL;
+       int chunk_size;
+       int chunk_bits;
+       int cpga_i = 0;
+       int count = 0;
        int done = 0;
        int rc = 0;
-       int count = 0;
-       enum cpga_fill_bits fill_bits = 0;
-       struct cl_page *clpage;
-       unsigned int applied_type = LL_COMPR_TYPE_UNCHANGED;
+       int pga_i;
 
        ENTRY;
 
        clpage = oap2cl_page(brw_page2oap(pga[chunk_start]));
-       chunk_size = (1 <<  (clpage->cp_chunk_log_bits + COMPR_CHUNK_MIN_BITS));
+       chunk_bits = cl_page_compr_bits(clpage);
+       chunk_size = (1 << chunk_bits);
        pages_in_chunk = chunk_size / PAGE_SIZE;
 
        if (clpage->cp_type == CPT_TRANSIENT)
@@ -346,10 +350,10 @@ int compress_request(const char *obd_name, struct obdo *oa,
                fill_bits |= CPGA_FILL_ENCRYPTED;
 
        OBD_ALLOC(*cpga, page_count * sizeof(**cpga));
-       sptlrpc_enc_pool_get_buf(&src,
-               clpage->cp_chunk_log_bits + COMPR_CHUNK_MIN_BITS);
-       sptlrpc_enc_pool_get_buf(&wrkmem,
-               clpage->cp_chunk_log_bits + COMPR_CHUNK_MIN_BITS + 1);
+       src_buf_bits = chunk_bits;
+       dest_buf_bits = chunk_bits + 1;
+       sptlrpc_enc_pool_get_buf(&src, src_buf_bits);
+       sptlrpc_enc_pool_get_buf(&wrkmem, dest_buf_bits);
 
        if (*cpga == NULL || wrkmem == NULL || src == NULL)
                GOTO(out, rc = -ENOMEM);
@@ -359,7 +363,10 @@ int compress_request(const char *obd_name, struct obdo *oa,
                    (pga_i == page_count - 1) ||
                    !can_merge_pages(pga[pga_i], pga[pga_i + 1])) {
                        clpage = oap2cl_page(brw_page2oap(pga[chunk_start]));
-                       /* TDB: change chunk size, reallocate src */
+                       /* TDB: change chunk size, reallocate src.
+                        * NB: This shouldn't happen as long as RPCs are for
+                        * a single component/object
+                        */
                        CDEBUG(D_SEC, "Chunk [%i,%i], type %i, level %i\n",
                               chunk_start, pga_i, clpage->cp_comp_type,
                               clpage->cp_comp_level);
@@ -367,9 +374,7 @@ int compress_request(const char *obd_name, struct obdo *oa,
                        merge_chunk(pga, chunk_start, pga_i + 1 - chunk_start,
                                    src, &src_size);
                        dst_size = 2 * chunk_size;
-                       sptlrpc_enc_pool_get_buf(&dst,
-                               clpage->cp_chunk_log_bits +
-                               COMPR_CHUNK_MIN_BITS + 1);
+                       sptlrpc_enc_pool_get_buf(&dst, dest_buf_bits);
                        if (dst == NULL)
                                GOTO(out, rc = -ENOMEM);
 
@@ -398,14 +403,11 @@ int compress_request(const char *obd_name, struct obdo *oa,
                                       fill_bits);
 
                        if (!done) {
-                               sptlrpc_enc_pool_put_buf(&dst,
-                                       clpage->cp_chunk_log_bits +
-                                       COMPR_CHUNK_MIN_BITS + 1);
+                               sptlrpc_enc_pool_put_buf(&dst, dest_buf_bits);
                        } else {
                                (*cpga)[cpga_i]->bp_cmp_chunk = dst;
                                (*cpga)[cpga_i]->bp_cmp_chunk_size =
-                                       clpage->cp_chunk_log_bits +
-                                       COMPR_CHUNK_MIN_BITS + 1;
+                                       dest_buf_bits;
                        }
 
                        if (rc)
@@ -422,12 +424,10 @@ int compress_request(const char *obd_name, struct obdo *oa,
        *pcount = cpga_i;
 out:
        if (wrkmem != NULL)
-               sptlrpc_enc_pool_put_buf(&wrkmem,
-                       clpage->cp_chunk_log_bits + COMPR_CHUNK_MIN_BITS + 1);
+               sptlrpc_enc_pool_put_buf(&wrkmem, dest_buf_bits);
 
        if (src != NULL)
-               sptlrpc_enc_pool_put_buf(&src,
-                       clpage->cp_chunk_log_bits + COMPR_CHUNK_MIN_BITS);
+               sptlrpc_enc_pool_put_buf(&src, src_buf_bits);
 
        if (rc != 0 && *cpga != NULL) {
                free_cpga(*cpga, page_count);
@@ -489,113 +489,92 @@ fail:
 
 int decompress_request(struct osc_brw_async_args *aa, int page_count)
 {
-       enum {CS_BEGIN, CS_OUTSIDE, CS_INSIDE} state = CS_BEGIN;
-       bool is_start = 0;
-       struct ll_compr_hdr *tllch = NULL, *llch = NULL;
-       int chunk_start = 0;
        struct brw_page **pga = aa->aa_ppga;
+       struct ll_compr_hdr *llch = NULL;
+       struct cl_page *clpage;
+       unsigned int src_size;
+       unsigned int dst_size;
+       int pages_in_chunk;
        char *src = NULL;
        char *dst = NULL;
-       unsigned int dst_size;
-       int pages_in_chunk = 0;
-       int chunk_size = 0;
-       int chunk_bits = 0;
-       int rc = 0;
-       int done = 0;
+       int chunk_bits;
+       int chunk_size;
        int count = 0;
+       int done = 0;
+       int buf_bits;
+       int rc = 0;
        int i = 0;
-       unsigned int src_size;
        ENTRY;
 
-       for (i = 0; i <= page_count; i++) {
+       clpage = oap2cl_page(brw_page2oap(pga[0]));
+       /* no compression */
+       if (clpage->cp_comp_type == LL_COMPR_TYPE_NONE)
+               RETURN(0);
+       chunk_bits = cl_page_compr_bits(clpage);
+       chunk_size = 1 << chunk_bits;
+       buf_bits = chunk_bits + 1;
+       pages_in_chunk = chunk_size / PAGE_SIZE;
 
-               if (i < page_count)
-                       is_start = is_chunk_start(pga[i]->pg, &tllch);
+       for (i = 0; i < page_count; i+=pages_in_chunk) {
 
-               switch (state) {
-               case CS_BEGIN:
-                       chunk_start = 0;
-                       count = 0;
-                       fallthrough;
-               case CS_OUTSIDE:
-                       if ((i == page_count) || !is_start) {
-                               state = CS_OUTSIDE;
-                               break;
-                       }
-                       chunk_start = i;
-                       state = CS_INSIDE;
-                       llch = tllch;
-                       chunk_bits = llch->llch_chunk_log_bits +
-                               COMPR_CHUNK_MIN_BITS;
-                       chunk_size = 1 << chunk_bits;
-                       pages_in_chunk = chunk_size / PAGE_SIZE;
-                       if (!src) { /* get chunk size once */
-                               chunk_bits = llch->llch_chunk_log_bits +
-                                       COMPR_CHUNK_MIN_BITS;
-                               chunk_size = 1 << chunk_bits;
-                               pages_in_chunk = chunk_size / PAGE_SIZE;
-                       }
-                       CDEBUG(D_SEC, "chunk_size: %i, pages_in_chunk: %i\n",
-                              chunk_size, pages_in_chunk);
-                       break;
-               case CS_INSIDE:
-                       LASSERT(pages_in_chunk != 0);
-                       if ((i < page_count &&
-                           (i - chunk_start) < pages_in_chunk &&
-                           !is_start) || llch == NULL)
-                               break;
-                       if (!src) {
-                               LASSERT(chunk_size != 0);
-                               sptlrpc_enc_pool_get_buf((void **)&src,
-                                       chunk_bits + 1);
-                               sptlrpc_enc_pool_get_buf((void **)&dst,
-                                       chunk_bits + 1);
-                               if (src == NULL || dst == NULL)
-                                       GOTO(out, rc = -ENOMEM);
-                       }
+               if (!is_chunk_start(pga[i]->pg, &llch))
+                       continue;
 
-                       CDEBUG(D_SEC, "Merge chunk start %i, i: %i, src: %px\n",
-                              chunk_start, i, src);
-                       merge_chunk(pga, chunk_start, i - chunk_start, src,
-                                   &src_size);
-                       LASSERT(src_size <= chunk_size);
-                       dst_size = 2 * chunk_size;
-                       CDEBUG(D_SEC, "Compressed size %lu, type %i\n",
-                              llch->llch_compr_size + sizeof(*llch),
-                              llch->llch_compr_type);
-
-                       rc = decompress_chunk(aa,
-                                               src + llch->llch_header_size,
-                                               llch->llch_compr_size,
-                                               dst, &dst_size,
-                                               llch->llch_compr_type,
-                                               llch->llch_compr_level);
-                       if (rc)
-                               GOTO(out, rc);
+               if (!src) { /* get chunk size once */
+                       int rpc_chunk_bits;
 
-                       CDEBUG(D_SEC, "Decompressed size %u, status %i\n",
-                              dst_size, done);
+                       rpc_chunk_bits = llch->llch_chunk_log_bits +
+                               COMPR_CHUNK_MIN_BITS;
+                       /* the chunk bits in every chunk should be the same */
+                       LASSERT(rpc_chunk_bits == chunk_bits);
 
-                       LASSERT(dst_size <= chunk_size);
-                       unmerge_chunk(pga, chunk_start, i - chunk_start,
-                                     dst, dst_size);
+                       CDEBUG(D_SEC, "chunk_size: %i, pages_in_chunk: %i\n",
+                               chunk_size, pages_in_chunk);
 
-                       count++;
-                       i--;
-                       state = CS_OUTSIDE;
-                       break;
-               default:
-                       GOTO(out, rc = -EIO);
+                       sptlrpc_enc_pool_get_buf((void **)&src,
+                               buf_bits);
+                       sptlrpc_enc_pool_get_buf((void **)&dst,
+                               buf_bits);
+                       if (src == NULL || dst == NULL)
+                               GOTO(out, rc = -ENOMEM);
                }
 
+               LASSERT((page_count - i) >= pages_in_chunk);
+               CDEBUG(D_SEC, "Merge chunk start %i, src: %px\n", i, src);
+
+               merge_chunk(pga, i, pages_in_chunk, src,
+                           &src_size);
+               LASSERT(src_size <= chunk_size);
+               dst_size = 2 * chunk_size;
+               CDEBUG(D_SEC, "Compressed size %lu, type %i\n",
+                      llch->llch_compr_size + sizeof(*llch),
+                      llch->llch_compr_type);
+
+               rc = decompress_chunk(aa,
+                                       src + llch->llch_header_size,
+                                       llch->llch_compr_size,
+                                       dst, &dst_size,
+                                       llch->llch_compr_type,
+                                       llch->llch_compr_level);
+               if (rc)
+                       GOTO(out, rc);
+
+               CDEBUG(D_SEC, "Decompressed size %u, status %i\n",
+                      dst_size, done);
+
+               LASSERT(dst_size <= chunk_size);
+               unmerge_chunk(pga, i, pages_in_chunk,
+                             dst, dst_size);
+
+               count++;
        }
        CDEBUG(D_SEC, "Decompressed %i pages (%i chunks)\n", page_count, count);
 out:
        if (src != NULL)
-               sptlrpc_enc_pool_put_buf(&src, chunk_bits + 1);
+               sptlrpc_enc_pool_put_buf(&src, buf_bits);
 
        if (dst != NULL)
-               sptlrpc_enc_pool_put_buf(&dst, chunk_bits + 1);
+               sptlrpc_enc_pool_put_buf(&dst, buf_bits);
 
        RETURN(rc);
 }