Fixes #14400 + some refactoring to avoid input buffer modification in future.

Signed-off-by: Igor Fedotov <ifedotov@mirantis.com>

diff --git a/src/compressor/Compressor.h b/src/compressor/Compressor.h
index c891bbef2ede4cf3afce175f823aa6ff2fabe5eb..6e39ceccc3769b6e80af84def16c5e6a2fd5a8d6 100644 (file)
--- a/src/compressor/Compressor.h
+++ b/src/compressor/Compressor.h
@@ -25,8 +25,8 @@ typedef shared_ptr<Compressor> CompressorRef;
 class Compressor {
  public:
   virtual ~Compressor() {}
-  virtual int compress(bufferlist &in, bufferlist &out) = 0;
-  virtual int decompress(bufferlist &in, bufferlist &out) = 0;
+  virtual int compress(const bufferlist &in, bufferlist &out) = 0;
+  virtual int decompress(const bufferlist &in, bufferlist &out) = 0;
 
   static CompressorRef create(CephContext *cct, const string &type);
 };

diff --git a/src/compressor/snappy/SnappyCompressor.h b/src/compressor/snappy/SnappyCompressor.h
index 40ec4509a3ac2ba42252243f275a1f25b07eae65..8a6166bd975bf14be56eff1dde5abb675987417e 100644 (file)
--- a/src/compressor/snappy/SnappyCompressor.h
+++ b/src/compressor/snappy/SnappyCompressor.h
@@ -26,7 +26,7 @@ class BufferlistSource : public snappy::Source {
   size_t left;
 
  public:
-  BufferlistSource(bufferlist &data): pb(data.buffers().begin()), pb_off(0), left(data.length()) {}
+  BufferlistSource(const bufferlist &data): pb(data.buffers().begin()), pb_off(0), left(data.length()) {}
   virtual ~BufferlistSource() {}
   virtual size_t Available() const { return left; }
   virtual const char* Peek(size_t* len) {
@@ -53,7 +53,7 @@ class SnappyCompressor : public Compressor {
  public:
   virtual ~SnappyCompressor() {}
   virtual const char* get_method_name() { return "snappy"; }
-  virtual int compress(bufferlist &src, bufferlist &dst) {
+  virtual int compress(const bufferlist &src, bufferlist &dst) {
     BufferlistSource source(src);
     bufferptr ptr(snappy::MaxCompressedLength(src.length()));
     snappy::UncheckedByteArraySink sink(ptr.c_str());
@@ -61,12 +61,14 @@ class SnappyCompressor : public Compressor {
     dst.append(ptr, 0, sink.CurrentDestination()-ptr.c_str());
     return 0;
   }
-  virtual int decompress(bufferlist &src, bufferlist &dst) {
-    BufferlistSource source(src);
+  virtual int decompress(const bufferlist &src, bufferlist &dst) {
     size_t res_len = 0;
     // Trick, decompress only need first 32bits buffer
-    if (!snappy::GetUncompressedLength(src.get_contiguous(0, 8), 8, &res_len))
+    bufferlist tmp;
+    tmp.substr_of( src, 0, 4 );
+    if (!snappy::GetUncompressedLength(tmp.c_str(), tmp.length(), &res_len))
       return -1;
+    BufferlistSource source(src);
     bufferptr ptr(res_len);
     if (snappy::RawUncompress(&source, ptr.c_str())) {
       dst.append(ptr);

diff --git a/src/test/compressor/compressor_example.h b/src/test/compressor/compressor_example.h
index 3a96f8807d333a437c0e032425cbd3c4c24274a0..148c63aa9893ad4da2cc5030a7f96cd9a224b5a4 100644 (file)
--- a/src/test/compressor/compressor_example.h
+++ b/src/test/compressor/compressor_example.h
@@ -31,13 +31,13 @@ class CompressorExample : public Compressor {
 public:
   virtual ~CompressorExample() {}
 
-  virtual int compress(bufferlist &in, bufferlist &out)
+  virtual int compress(const bufferlist &in, bufferlist &out)
   {
     out = in;
     return 0;
   }
 
-  virtual int decompress(bufferlist &in, bufferlist &out)
+  virtual int decompress(const bufferlist &in, bufferlist &out)
   {
     out = in;
     return 0;

diff --git a/src/test/compressor/test_compression_snappy.cc b/src/test/compressor/test_compression_snappy.cc
index 69b0ebfaeda4fb43fcaa5b6dda841f508341832c..556390b59db301fea29c9fb06acfe124ba4ad5d8 100644 (file)
--- a/src/test/compressor/test_compression_snappy.cc
+++ b/src/test/compressor/test_compression_snappy.cc
@@ -27,7 +27,7 @@ TEST(SnappyCompressor, compress_decompress)
 {
   SnappyCompressor sp;
   EXPECT_EQ(sp.get_method_name(), "snappy");
-  char* test = "This is test text";
+  const char* test = "This is test text";
   int len = strlen(test);
   bufferlist in, out;
   in.append(test, len);
@@ -38,6 +38,38 @@ TEST(SnappyCompressor, compress_decompress)
   EXPECT_EQ(res, 0);
 }
 
+TEST(SnappyCompressor, sharded_input_decompress)
+{
+  const size_t small_prefix_size=3;
+
+  SnappyCompressor sp;
+  EXPECT_EQ(sp.get_method_name(), "snappy");
+  string test(128*1024,0);
+  int len = test.size();
+  bufferlist in, out;
+  in.append(test.c_str(), len);
+  int res = sp.compress(in, out);
+  EXPECT_EQ(res, 0);
+  EXPECT_GT(out.length(), small_prefix_size);
+  
+  bufferlist out2, tmp;
+  tmp.substr_of(out, 0, small_prefix_size );
+  out2.append( tmp );
+  size_t left = out.length()-small_prefix_size;
+  size_t offs = small_prefix_size;
+  while( left > 0 ){
+    size_t shard_size = MIN( 2048, left ); 
+    tmp.substr_of(out, offs, shard_size );
+    out2.append( tmp );
+    left -= shard_size;
+    offs += shard_size;
+  }
+
+  bufferlist after;
+  res = sp.decompress(out2, after);
+  EXPECT_EQ(res, 0);
+}
+
 int main(int argc, char **argv) {
   vector<const char*> args;
   argv_to_vec(argc, (const char **)argv, args);