radosgw: refactor compression code

Signed-off-by: Adam Kupczyk <akupczyk@mirantis.com>

diff --git a/src/rgw/rgw_compression.cc b/src/rgw/rgw_compression.cc
index 1a7758182e481084b49ed7010c06e1ab95067910..b9637d01d09b9d2c96ca24e0d244d7cfea500831 100644 (file)
--- a/src/rgw/rgw_compression.cc
+++ b/src/rgw/rgw_compression.cc
@@ -59,7 +59,6 @@ RGWGetObj_Decompress::RGWGetObj_Decompress(CephContext* cct_,
                                                                 partial_content(partial_content_),
                                                                 q_ofs(0),
                                                                 q_len(0),
-                                                                first_data(true),
                                                                 cur_ofs(0)
 {
   compressor = Compressor::create(cct, cs_info->compression_type);
@@ -69,7 +68,8 @@ RGWGetObj_Decompress::RGWGetObj_Decompress(CephContext* cct_,
 
 int RGWGetObj_Decompress::handle_data(bufferlist& bl, off_t bl_ofs, off_t bl_len)
 {
-  ldout(cct, 10) << "Compression for rgw is enabled, decompress part " << bl_len << dendl;
+  ldout(cct, 10) << "Compression for rgw is enabled, decompress part "
+      << "bl_ofs="<< bl_ofs << bl_len << dendl;
 
   if (!compressor.get()) {
     // if compressor isn't available - error, because cannot return decompressed data?
@@ -79,6 +79,7 @@ int RGWGetObj_Decompress::handle_data(bufferlist& bl, off_t bl_ofs, off_t bl_len
   bufferlist out_bl, in_bl, temp_in_bl;
   bl.copy(bl_ofs, bl_len, temp_in_bl); 
   bl_ofs = 0;
+  int r;
   if (waiting.length() != 0) {
     in_bl.append(waiting);
     in_bl.append(temp_in_bl);        
@@ -87,9 +88,9 @@ int RGWGetObj_Decompress::handle_data(bufferlist& bl, off_t bl_ofs, off_t bl_len
     in_bl.claim(temp_in_bl);
   }
   bl_len = in_bl.length();
-
+  
   while (first_block <= last_block) {
-    bufferlist tmp, tmp_out;
+    bufferlist tmp;
     off_t ofs_in_bl = first_block->new_ofs - cur_ofs;
     if (ofs_in_bl + (off_t)first_block->len > bl_len) {
       // not complete block, put it to waiting
@@ -99,48 +100,37 @@ int RGWGetObj_Decompress::handle_data(bufferlist& bl, off_t bl_ofs, off_t bl_len
       break;
     }
     in_bl.copy(ofs_in_bl, first_block->len, tmp);
-    int r = compressor->decompress(tmp, tmp_out);
-    if (r < 0) {
-      lderr(cct) << "Compression failed with exit code " << r << dendl;
-      return r;
+    int cr = compressor->decompress(tmp, out_bl);
+    if (cr < 0) {
+      lderr(cct) << "Compression failed with exit code " << cr << dendl;
+      return cr;
     }
-    if (first_block == last_block && partial_content)
-      tmp_out.copy(0, q_len, out_bl);
-    else {
-      out_bl.append(tmp_out);    
-      if (cct->_conf->rgw_max_chunk_size && (out_bl.length() >= cct->_conf->rgw_max_chunk_size)) {
-        if (first_data && partial_content && out_bl.length() != 0)
-          bl_ofs =  q_ofs;
-        
-        if (first_data && out_bl.length() != 0)
-          first_data = false;
-        
-        r = next->handle_data(out_bl, bl_ofs, out_bl.length() - bl_ofs);
-        if (r < 0) {
-          lderr(cct) << "handle_data failed with exit code " << r << dendl;
-          return r;       
-        }  
-        bl_ofs = 0;
-        out_bl.clear();
+    ++first_block;
+    while (out_bl.length() - q_ofs >= cct->_conf->rgw_max_chunk_size)
+    {
+      off_t ch_len = std::min<off_t>(cct->_conf->rgw_max_chunk_size, q_len);
+      q_len -= ch_len;
+      r = next->handle_data(out_bl, q_ofs, ch_len);
+      if (r < 0) {
+        lderr(cct) << "handle_data failed with exit code " << r << dendl;
+        return r;
       }
+      out_bl.splice(0, q_ofs + ch_len);
+      q_ofs = 0;
     }
-    ++first_block;
- }
-
-  if (first_data && partial_content && out_bl.length() != 0)
-    bl_ofs =  q_ofs;
-
-  if (first_data && out_bl.length() != 0)
-    first_data = false;
+  }
 
   cur_ofs += bl_len;
 
-  if (out_bl.length()) {
-    assert(out_bl.length() >= bl_ofs); 
-    return next->handle_data(out_bl, bl_ofs, out_bl.length() - bl_ofs);
-  } 
-
-  return 0;
+  off_t ch_len = std::min<off_t>(out_bl.length() - q_ofs, q_len);
+  r = next->handle_data(out_bl, q_ofs, ch_len);
+  if (r < 0) {
+    lderr(cct) << "handle_data failed with exit code " << r << dendl;
+    return r;
+  }
+  q_len -= ch_len;
+  q_ofs = 0;
+  return r;
 }
 
 int RGWGetObj_Decompress::fixup_range(off_t& ofs, off_t& end)
@@ -169,12 +159,11 @@ int RGWGetObj_Decompress::fixup_range(off_t& ofs, off_t& end)
   }
 
   q_ofs = ofs - first_block->old_ofs;
-  q_len = end - last_block->old_ofs + 1;
+  q_len = end + 1 - ofs;
 
   ofs = first_block->new_ofs;
   end = last_block->new_ofs + last_block->len - 1;
 
-  first_data = true;
   cur_ofs = ofs;
   waiting.clear();
 

diff --git a/src/rgw/rgw_compression.h b/src/rgw/rgw_compression.h
index e58d959e294a4e9f64efaa9d76b0089b908efba5..721c510f02ab538be96bdb36ff30ea19ae644ecf 100644 (file)
--- a/src/rgw/rgw_compression.h
+++ b/src/rgw/rgw_compression.h
@@ -17,7 +17,6 @@ class RGWGetObj_Decompress : public RGWGetObj_Filter
   bool partial_content;
   vector<compression_block>::iterator first_block, last_block;
   off_t q_ofs, q_len;
-  bool first_data;
   uint64_t cur_ofs;
   bufferlist waiting;
 public:

diff --git a/src/test/rgw/test_rgw_compression.cc b/src/test/rgw/test_rgw_compression.cc
index b254ebc81ad5b014fd3d39b988f869bad68e4a67..e10bbdc3049c9cb16cbec5c8f15dfd93354c0121 100644 (file)
--- a/src/test/rgw/test_rgw_compression.cc
+++ b/src/test/rgw/test_rgw_compression.cc
@@ -154,7 +154,7 @@ TEST(Compress, LimitedChunkSize)
     decompress.handle_data(c_sink.get_sink(), 0, c_sink.get_sink().length());
     decompress.handle_data(empty, 0, 0);
 
-    ASSERT_LT(d_sink.get_size(), (size_t)g_ceph_context->_conf->rgw_max_chunk_size);
+    ASSERT_LE(d_sink.get_size(), (size_t)g_ceph_context->_conf->rgw_max_chunk_size);
   }
 }