ceph/src/test/rgw/test_rgw_compression.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 #include "gtest/gtest.h"
   4
   5 #include "rgw_compression.h"
   6
   7 class ut_get_sink : public RGWGetObj_Filter {
   8   bufferlist sink;
   9 public:
  10   ut_get_sink() {}
  11   virtual ~ut_get_sink() {}
  12
  13   int handle_data(bufferlist& bl, off_t bl_ofs, off_t bl_len) override
  14   {
  15     auto& bl_buffers = bl.buffers();
  16     auto i = bl_buffers.begin();
  17     while (bl_len > 0)
  18     {
  19       ceph_assert(i != bl_buffers.end());
  20       off_t len = std::min<off_t>(bl_len, i->length());
  21       sink.append(*i, 0, len);
  22       bl_len -= len;
  23       i++;
  24     }
  25     return 0;
  26   }
  27   bufferlist& get_sink()
  28   {
  29     return sink;
  30   }
  31 };
  32
  33 class ut_get_sink_size : public RGWGetObj_Filter {
  34   size_t max_size = 0;
  35 public:
  36   ut_get_sink_size() {}
  37   virtual ~ut_get_sink_size() {}
  38
  39   int handle_data(bufferlist& bl, off_t bl_ofs, off_t bl_len) override
  40   {
  41     if (bl_len > (off_t)max_size)
  42       max_size = bl_len;
  43     return 0;
  44   }
  45   size_t get_size()
  46   {
  47     return max_size;
  48   }
  49 };
  50
  51 class ut_put_sink: public rgw::sal::DataProcessor
  52 {
  53   bufferlist sink;
  54 public:
  55   int process(bufferlist&& bl, uint64_t ofs) override
  56   {
  57     sink.claim_append(bl);
  58     return 0;
  59   }
  60   bufferlist&  get_sink()
  61   {
  62     return sink;
  63   }
  64 };
  65
  66
  67 struct MockGetDataCB : public RGWGetObj_Filter {
  68   int handle_data(bufferlist& bl, off_t bl_ofs, off_t bl_len) override {
  69     return 0;
  70   }
  71 } cb;
  72
  73 using range_t = std::pair<off_t, off_t>;
  74
  75 // call filter->fixup_range() and return the range as a pair. this makes it easy
  76 // to fit on a single line for ASSERT_EQ()
  77 range_t fixup_range(RGWGetObj_Decompress *filter, off_t ofs, off_t end)
  78 {
  79   filter->fixup_range(ofs, end);
  80   return {ofs, end};
  81 }
  82
  83
  84 TEST(Decompress, FixupRangePartial)
  85 {
  86   RGWCompressionInfo cs_info;
  87
  88   // array of blocks with original len=8, compressed to len=6
  89   auto& blocks = cs_info.blocks;
  90   blocks.emplace_back(compression_block{0, 0, 6});
  91   blocks.emplace_back(compression_block{8, 6, 6});
  92   blocks.emplace_back(compression_block{16, 12, 6});
  93   blocks.emplace_back(compression_block{24, 18, 6});
  94
  95   const bool partial = true;
  96   RGWGetObj_Decompress decompress(g_ceph_context, &cs_info, partial, &cb);
  97
  98   // test translation from logical ranges to compressed ranges
  99   ASSERT_EQ(range_t(0, 5), fixup_range(&decompress, 0, 1));
 100   ASSERT_EQ(range_t(0, 5), fixup_range(&decompress, 1, 7));
 101   ASSERT_EQ(range_t(0, 11), fixup_range(&decompress, 7, 8));
 102   ASSERT_EQ(range_t(0, 11), fixup_range(&decompress, 0, 9));
 103   ASSERT_EQ(range_t(0, 11), fixup_range(&decompress, 7, 9));
 104   ASSERT_EQ(range_t(6, 11), fixup_range(&decompress, 8, 9));
 105   ASSERT_EQ(range_t(6, 17), fixup_range(&decompress, 8, 16));
 106   ASSERT_EQ(range_t(6, 17), fixup_range(&decompress, 8, 17));
 107   ASSERT_EQ(range_t(12, 23), fixup_range(&decompress, 16, 24));
 108   ASSERT_EQ(range_t(12, 23), fixup_range(&decompress, 16, 999));
 109   ASSERT_EQ(range_t(18, 23), fixup_range(&decompress, 998, 999));
 110 }
 111
 112 TEST(Compress, LimitedChunkSize)
 113 {
 114   CompressorRef plugin;
 115   plugin = Compressor::create(g_ceph_context, Compressor::COMP_ALG_ZLIB);
 116   ASSERT_NE(plugin.get(), nullptr);
 117
 118   for (size_t s = 100 ; s < 10000000 ; s = s*5/4)
 119   {
 120     bufferptr bp(s);
 121     bufferlist bl;
 122     bl.append(bp);
 123
 124     ut_put_sink c_sink;
 125     RGWPutObj_Compress compressor(g_ceph_context, plugin, &c_sink);
 126     compressor.process(std::move(bl), 0);
 127     compressor.process({}, s); // flush
 128
 129     RGWCompressionInfo cs_info;
 130     cs_info.compression_type = plugin->get_type_name();
 131     cs_info.orig_size = s;
 132     cs_info.compressor_message = compressor.get_compressor_message();
 133     cs_info.blocks = move(compressor.get_compression_blocks());
 134
 135     ut_get_sink_size d_sink;
 136     RGWGetObj_Decompress decompress(g_ceph_context, &cs_info, false, &d_sink);
 137
 138     off_t f_begin = 0;
 139     off_t f_end = s - 1;
 140     decompress.fixup_range(f_begin, f_end);
 141
 142     decompress.handle_data(c_sink.get_sink(), 0, c_sink.get_sink().length());
 143     bufferlist empty;
 144     decompress.handle_data(empty, 0, 0);
 145
 146     ASSERT_LE(d_sink.get_size(), (size_t)g_ceph_context->_conf->rgw_max_chunk_size);
 147   }
 148 }
 149
 150
 151 TEST(Compress, BillionZeros)
 152 {
 153   CompressorRef plugin;
 154   ut_put_sink c_sink;
 155   plugin = Compressor::create(g_ceph_context, Compressor::COMP_ALG_ZLIB);
 156   ASSERT_NE(plugin.get(), nullptr);
 157   RGWPutObj_Compress compressor(g_ceph_context, plugin, &c_sink);
 158
 159   constexpr size_t size = 1000000;
 160   bufferptr bp(size);
 161   bufferlist bl;
 162   bl.append(bp);
 163
 164   for (int i=0; i<1000;i++)
 165     compressor.process(bufferlist{bl}, size*i);
 166   compressor.process({}, size*1000); // flush
 167
 168   RGWCompressionInfo cs_info;
 169   cs_info.compression_type = plugin->get_type_name();
 170   cs_info.orig_size = size*1000;
 171   cs_info.compressor_message = compressor.get_compressor_message();
 172   cs_info.blocks = move(compressor.get_compression_blocks());
 173
 174   ut_get_sink d_sink;
 175   RGWGetObj_Decompress decompress(g_ceph_context, &cs_info, false, &d_sink);
 176
 177   off_t f_begin = 0;
 178   off_t f_end = size*1000 - 1;
 179   decompress.fixup_range(f_begin, f_end);
 180
 181   decompress.handle_data(c_sink.get_sink(), 0, c_sink.get_sink().length());
 182   bufferlist empty;
 183   decompress.handle_data(empty, 0, 0);
 184
 185   ASSERT_EQ(d_sink.get_sink().length() , size*1000);
 186 }