ceph/src/test/compressor/test_compression.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph distributed storage system
   5  *
   6  * Copyright (C) 2015 Mirantis, Inc.
   7  *
   8  * Author: Alyona Kiseleva <akiselyova@mirantis.com>
   9  *
  10  *  This library is free software; you can redistribute it and/or
  11  *  modify it under the terms of the GNU Lesser General Public
  12  *  License as published by the Free Software Foundation; either
  13  *  version 2.1 of the License, or (at your option) any later version.
  14  *
  15  */
  16
  17 #include <errno.h>
  18 #include <signal.h>
  19 #include <stdlib.h>
  20 #include "gtest/gtest.h"
  21 #include "common/ceph_context.h"
  22 #include "common/config.h"
  23 #include "compressor/Compressor.h"
  24 #include "compressor/CompressionPlugin.h"
  25 #include "global/global_context.h"
  26 #include "osd/OSDMap.h"
  27
  28 using namespace std;
  29
  30 class CompressorTest : public ::testing::Test,
  31                         public ::testing::WithParamInterface<const char*> {
  32 public:
  33   std::string plugin;
  34   CompressorRef compressor;
  35   bool old_zlib_isal;
  36
  37   CompressorTest() {
  38     // note for later
  39     old_zlib_isal = g_conf()->compressor_zlib_isal;
  40
  41     plugin = GetParam();
  42     size_t pos = plugin.find('/');
  43     if (pos != std::string::npos) {
  44       string isal = plugin.substr(pos + 1);
  45       plugin = plugin.substr(0, pos);
  46       if (isal == "isal") {
  47         g_conf().set_val("compressor_zlib_isal", "true");
  48         g_ceph_context->_conf.apply_changes(nullptr);
  49       } else if (isal == "noisal") {
  50         g_conf().set_val("compressor_zlib_isal", "false");
  51         g_ceph_context->_conf.apply_changes(nullptr);
  52       } else {
  53         ceph_abort_msg("bad option");
  54       }
  55     }
  56     cout << "[plugin " << plugin << " (" << GetParam() << ")]" << std::endl;
  57   }
  58   ~CompressorTest() override {
  59     g_conf().set_val("compressor_zlib_isal", old_zlib_isal ? "true" : "false");
  60     g_ceph_context->_conf.apply_changes(nullptr);
  61   }
  62
  63   void SetUp() override {
  64     compressor = Compressor::create(g_ceph_context, plugin);
  65     ASSERT_TRUE(compressor);
  66   }
  67   void TearDown() override {
  68     compressor.reset();
  69   }
  70 };
  71
  72 TEST_P(CompressorTest, load_plugin)
  73 {
  74 }
  75
  76 TEST_P(CompressorTest, small_round_trip)
  77 {
  78   bufferlist orig;
  79   orig.append("This is a short string.  There are many strings like it but this one is mine.");
  80   bufferlist compressed;
  81   boost::optional<int32_t> compressor_message;
  82   int r = compressor->compress(orig, compressed, compressor_message);
  83   ASSERT_EQ(0, r);
  84   bufferlist decompressed;
  85   r = compressor->decompress(compressed, decompressed, compressor_message);
  86   ASSERT_EQ(0, r);
  87   ASSERT_EQ(decompressed.length(), orig.length());
  88   ASSERT_TRUE(decompressed.contents_equal(orig));
  89   cout << "orig " << orig.length() << " compressed " << compressed.length()
  90        << " with " << GetParam() << std::endl;
  91 }
  92
  93 TEST_P(CompressorTest, big_round_trip_repeated)
  94 {
  95   unsigned len = 1048576 * 4;
  96   bufferlist orig;
  97   while (orig.length() < len) {
  98     orig.append("This is a short string.  There are many strings like it but this one is mine.");
  99   }
 100   bufferlist compressed;
 101   boost::optional<int32_t> compressor_message;
 102   int r = compressor->compress(orig, compressed, compressor_message);
 103   ASSERT_EQ(0, r);
 104   bufferlist decompressed;
 105   r = compressor->decompress(compressed, decompressed, compressor_message);
 106   ASSERT_EQ(0, r);
 107   ASSERT_EQ(decompressed.length(), orig.length());
 108   ASSERT_TRUE(decompressed.contents_equal(orig));
 109   cout << "orig " << orig.length() << " compressed " << compressed.length()
 110        << " with " << GetParam() << std::endl;
 111 }
 112
 113 TEST_P(CompressorTest, big_round_trip_randomish)
 114 {
 115   unsigned len = 1048576 * 10;//269;
 116   bufferlist orig;
 117   const char *alphabet = "abcdefghijklmnopqrstuvwxyz";
 118   if (false) {
 119     while (orig.length() < len) {
 120       orig.append(alphabet[rand() % 10]);
 121     }
 122   } else {
 123     bufferptr bp(len);
 124     char *p = bp.c_str();
 125     for (unsigned i=0; i<len; ++i) {
 126       p[i] = alphabet[rand() % 10];
 127     }
 128     orig.append(bp);
 129   }
 130   bufferlist compressed;
 131   boost::optional<int32_t> compressor_message;
 132   int r = compressor->compress(orig, compressed, compressor_message);
 133   ASSERT_EQ(0, r);
 134   bufferlist decompressed;
 135   r = compressor->decompress(compressed, decompressed, compressor_message);
 136   ASSERT_EQ(0, r);
 137   ASSERT_EQ(decompressed.length(), orig.length());
 138   ASSERT_TRUE(decompressed.contents_equal(orig));
 139   cout << "orig " << orig.length() << " compressed " << compressed.length()
 140        << " with " << GetParam() << std::endl;
 141 }
 142
 143 #if 0
 144 TEST_P(CompressorTest, big_round_trip_file)
 145 {
 146   bufferlist orig;
 147   int fd = ::open("bin/ceph-osd", O_RDONLY);
 148   struct stat st;
 149   ::fstat(fd, &st);
 150   orig.read_fd(fd, st.st_size);
 151
 152   bufferlist compressed;
 153   int r = compressor->compress(orig, compressed);
 154   ASSERT_EQ(0, r);
 155   bufferlist decompressed;
 156   r = compressor->decompress(compressed, decompressed);
 157   ASSERT_EQ(0, r);
 158   ASSERT_EQ(decompressed.length(), orig.length());
 159   ASSERT_TRUE(decompressed.contents_equal(orig));
 160   cout << "orig " << orig.length() << " compressed " << compressed.length()
 161        << " with " << GetParam() << std::endl;
 162 }
 163 #endif
 164
 165
 166 TEST_P(CompressorTest, round_trip_osdmap)
 167 {
 168 #include "osdmaps/osdmap.2982809.h"
 169
 170   auto compressor = Compressor::create(g_ceph_context, plugin);
 171   bufferlist orig;
 172   orig.append((char*)osdmap_a, sizeof(osdmap_a));
 173   cout << "orig length " << orig.length() << std::endl;
 174   uint32_t size = 128*1024;
 175   OSDMap *o = new OSDMap;
 176   o->decode(orig);
 177   bufferlist fbl;
 178   o->encode(fbl, o->get_encoding_features() | CEPH_FEATURE_RESERVED);
 179   ASSERT_TRUE(fbl.contents_equal(orig));
 180   for (int j = 0; j < 3; j++) {
 181     bufferlist chunk;
 182     uint32_t l = std::min(size, fbl.length() - j*size);
 183     chunk.substr_of(fbl, j*size, l);
 184     //fbl.rebuild();
 185     bufferlist compressed;
 186     boost::optional<int32_t> compressor_message;
 187     int r = compressor->compress(chunk, compressed, compressor_message);
 188     ASSERT_EQ(0, r);
 189     bufferlist decompressed;
 190     r = compressor->decompress(compressed, decompressed, compressor_message);
 191     ASSERT_EQ(0, r);
 192     ASSERT_EQ(decompressed.length(), chunk.length());
 193     if (!decompressed.contents_equal(chunk)) {
 194       cout << "FAILED, orig bl was\n" << fbl << std::endl;
 195       ASSERT_TRUE(decompressed.contents_equal(chunk));
 196     }
 197     cout << "chunk " << chunk.length()
 198          << " compressed " << compressed.length()
 199          << " decompressed " << decompressed.length()
 200          << " with " << plugin << std::endl;
 201   }
 202   delete o;
 203 }
 204
 205 TEST_P(CompressorTest, compress_decompress)
 206 {
 207   const char* test = "This is test text";
 208   int res;
 209   int len = strlen(test);
 210   bufferlist in, out;
 211   bufferlist after;
 212   bufferlist exp;
 213   in.append(test, len);
 214   boost::optional<int32_t> compressor_message;
 215   res = compressor->compress(in, out, compressor_message);
 216   EXPECT_EQ(res, 0);
 217   res = compressor->decompress(out, after, compressor_message);
 218   EXPECT_EQ(res, 0);
 219   exp.append(test);
 220   EXPECT_TRUE(exp.contents_equal(after));
 221   after.clear();
 222   size_t compressed_len = out.length();
 223   out.append_zero(12);
 224   auto it = out.cbegin();
 225   res = compressor->decompress(it, compressed_len, after, compressor_message);
 226   EXPECT_EQ(res, 0);
 227   EXPECT_TRUE(exp.contents_equal(after));
 228
 229   //large block and non-begin iterator for continuous block
 230   std::string data;
 231   data.resize(0x10000 * 1);
 232   for(size_t i = 0; i < data.size(); i++)
 233     data[i] = i / 256;
 234   in.clear();
 235   out.clear();
 236   in.append(data);
 237   exp = in;
 238   res = compressor->compress(in, out, compressor_message);
 239   EXPECT_EQ(res, 0);
 240   compressed_len = out.length();
 241   out.append_zero(0x10000 - out.length());
 242   after.clear();
 243   out.c_str();
 244   bufferlist prefix;
 245   prefix.append(string("some prefix"));
 246   size_t prefix_len = prefix.length();
 247   prefix.claim_append(out);
 248   out.swap(prefix);
 249   it = out.cbegin();
 250   it += prefix_len;
 251   res = compressor->decompress(it, compressed_len, after, compressor_message);
 252   EXPECT_EQ(res, 0);
 253   EXPECT_TRUE(exp.contents_equal(after));
 254 }
 255
 256 TEST_P(CompressorTest, sharded_input_decompress)
 257 {
 258   const size_t small_prefix_size=3;
 259
 260   string test(128*1024,0);
 261   int len = test.size();
 262   bufferlist in, out;
 263   in.append(test.c_str(), len);
 264   boost::optional<int32_t> compressor_message;
 265   int res = compressor->compress(in, out, compressor_message);
 266   EXPECT_EQ(res, 0);
 267   EXPECT_GT(out.length(), small_prefix_size);
 268
 269   bufferlist out2, tmp;
 270   tmp.substr_of(out, 0, small_prefix_size );
 271   out2.append( tmp );
 272   size_t left = out.length()-small_prefix_size;
 273   size_t offs = small_prefix_size;
 274   while( left > 0 ){
 275     size_t shard_size = std::min<size_t>(2048, left);
 276     tmp.substr_of(out, offs, shard_size );
 277     out2.append( tmp );
 278     left -= shard_size;
 279     offs += shard_size;
 280   }
 281
 282   bufferlist after;
 283   res = compressor->decompress(out2, after, compressor_message);
 284   EXPECT_EQ(res, 0);
 285 }
 286
 287 void test_compress(CompressorRef compressor, size_t size)
 288 {
 289   char* data = (char*) malloc(size);
 290   for (size_t t = 0; t < size; t++) {
 291     data[t] = (t & 0xff) | (t >> 8);
 292   }
 293   bufferlist in;
 294   in.append(data, size);
 295   for (size_t t = 0; t < 10000; t++) {
 296     bufferlist out;
 297     boost::optional<int32_t> compressor_message;
 298     int res = compressor->compress(in, out, compressor_message);
 299     EXPECT_EQ(res, 0);
 300   }
 301   free(data);
 302 }
 303
 304 void test_decompress(CompressorRef compressor, size_t size)
 305 {
 306   char* data = (char*) malloc(size);
 307   for (size_t t = 0; t < size; t++) {
 308     data[t] = (t & 0xff) | (t >> 8);
 309   }
 310   bufferlist in, out;
 311   in.append(data, size);
 312   boost::optional<int32_t> compressor_message;
 313   int res = compressor->compress(in, out, compressor_message);
 314   EXPECT_EQ(res, 0);
 315   for (size_t t = 0; t < 10000; t++) {
 316     bufferlist out_dec;
 317     int res = compressor->decompress(out, out_dec, compressor_message);
 318     EXPECT_EQ(res, 0);
 319   }
 320   free(data);
 321 }
 322
 323 TEST_P(CompressorTest, compress_1024)
 324 {
 325   test_compress(compressor, 1024);
 326 }
 327
 328 TEST_P(CompressorTest, compress_2048)
 329 {
 330   test_compress(compressor, 2048);
 331 }
 332
 333 TEST_P(CompressorTest, compress_4096)
 334 {
 335   test_compress(compressor, 4096);
 336 }
 337
 338 TEST_P(CompressorTest, compress_8192)
 339 {
 340   test_compress(compressor, 8192);
 341 }
 342
 343 TEST_P(CompressorTest, compress_16384)
 344 {
 345   test_compress(compressor, 16384);
 346 }
 347
 348 TEST_P(CompressorTest, decompress_1024)
 349 {
 350   test_decompress(compressor, 1024);
 351 }
 352
 353 TEST_P(CompressorTest, decompress_2048)
 354 {
 355   test_decompress(compressor, 2048);
 356 }
 357
 358 TEST_P(CompressorTest, decompress_4096)
 359 {
 360   test_decompress(compressor, 4096);
 361 }
 362
 363 TEST_P(CompressorTest, decompress_8192)
 364 {
 365   test_decompress(compressor, 8192);
 366 }
 367
 368 TEST_P(CompressorTest, decompress_16384)
 369 {
 370   test_decompress(compressor, 16384);
 371 }
 372
 373
 374 INSTANTIATE_TEST_SUITE_P(
 375   Compressor,
 376   CompressorTest,
 377   ::testing::Values(
 378 #ifdef HAVE_LZ4
 379     "lz4",
 380 #endif
 381 #ifdef __x86_64__
 382     "zlib/isal",
 383 #endif
 384     "zlib/noisal",
 385     "snappy",
 386 #ifdef HAVE_BROTLI
 387     "brotli",
 388 #endif
 389     "zstd"));
 390
 391 #ifdef __x86_64__
 392
 393 TEST(ZlibCompressor, zlib_isal_compatibility)
 394 {
 395   g_conf().set_val("compressor_zlib_isal", "true");
 396   g_ceph_context->_conf.apply_changes(nullptr);
 397   CompressorRef isal = Compressor::create(g_ceph_context, "zlib");
 398   if (!isal) {
 399     // skip the test if the plugin is not ready
 400     return;
 401   }
 402   g_conf().set_val("compressor_zlib_isal", "false");
 403   g_ceph_context->_conf.apply_changes(nullptr);
 404   CompressorRef zlib = Compressor::create(g_ceph_context, "zlib");
 405   char test[101];
 406   srand(time(0));
 407   for (int i=0; i<100; ++i)
 408     test[i] = 'a' + rand()%26;
 409   test[100] = '\0';
 410   int len = strlen(test);
 411   bufferlist in, out;
 412   in.append(test, len);
 413   // isal -> zlib
 414   boost::optional<int32_t> compressor_message;
 415   int res = isal->compress(in, out, compressor_message);
 416   EXPECT_EQ(res, 0);
 417   bufferlist after;
 418   res = zlib->decompress(out, after, compressor_message);
 419   EXPECT_EQ(res, 0);
 420   bufferlist exp;
 421   exp.append(static_cast<char*>(test));
 422   EXPECT_TRUE(exp.contents_equal(after));
 423   after.clear();
 424   out.clear();
 425   exp.clear();
 426   // zlib -> isal
 427   res = zlib->compress(in, out, compressor_message);
 428   EXPECT_EQ(res, 0);
 429   res = isal->decompress(out, after, compressor_message);
 430   EXPECT_EQ(res, 0);
 431   exp.append(static_cast<char*>(test));
 432   EXPECT_TRUE(exp.contents_equal(after));
 433 }
 434 #endif
 435
 436 TEST(CompressionPlugin, all)
 437 {
 438   CompressorRef compressor;
 439   PluginRegistry *reg = g_ceph_context->get_plugin_registry();
 440   EXPECT_TRUE(reg);
 441   CompressionPlugin *factory = dynamic_cast<CompressionPlugin*>(reg->get_with_load("compressor", "invalid"));
 442   EXPECT_FALSE(factory);
 443   factory = dynamic_cast<CompressionPlugin*>(reg->get_with_load("compressor", "example"));
 444   ASSERT_TRUE(factory);
 445   stringstream ss;
 446   EXPECT_EQ(0, factory->factory(&compressor, &ss));
 447   EXPECT_TRUE(compressor.get());
 448   {
 449     std::lock_guard l(reg->lock);
 450     EXPECT_EQ(-ENOENT, reg->remove("compressor", "does not exist"));
 451     EXPECT_EQ(0, reg->remove("compressor", "example"));
 452     EXPECT_EQ(0, reg->load("compressor", "example"));
 453   }
 454 }
 455
 456 #ifdef __x86_64__
 457
 458 TEST(ZlibCompressor, isal_compress_zlib_decompress_random)
 459 {
 460   g_conf().set_val("compressor_zlib_isal", "true");
 461   g_ceph_context->_conf.apply_changes(nullptr);
 462   CompressorRef isal = Compressor::create(g_ceph_context, "zlib");
 463   if (!isal) {
 464     // skip the test if the plugin is not ready
 465     return;
 466   }
 467   g_conf().set_val("compressor_zlib_isal", "false");
 468   g_ceph_context->_conf.apply_changes(nullptr);
 469   CompressorRef zlib = Compressor::create(g_ceph_context, "zlib");
 470
 471   for (int cnt=0; cnt<100; cnt++)
 472   {
 473     srand(cnt + 1000);
 474     int log2 = (rand()%18) + 1;
 475     int size = (rand() % (1 << log2)) + 1;
 476
 477     char test[size];
 478     for (int i=0; i<size; ++i)
 479       test[i] = rand()%256;
 480     bufferlist in, out;
 481     in.append(test, size);
 482
 483     boost::optional<int32_t> compressor_message;
 484     int res = isal->compress(in, out, compressor_message);
 485     EXPECT_EQ(res, 0);
 486     bufferlist after;
 487     res = zlib->decompress(out, after, compressor_message);
 488     EXPECT_EQ(res, 0);
 489     bufferlist exp;
 490     exp.append(test, size);
 491     EXPECT_TRUE(exp.contents_equal(after));
 492   }
 493 }
 494
 495 TEST(ZlibCompressor, isal_compress_zlib_decompress_walk)
 496 {
 497   g_conf().set_val("compressor_zlib_isal", "true");
 498   g_ceph_context->_conf.apply_changes(nullptr);
 499   CompressorRef isal = Compressor::create(g_ceph_context, "zlib");
 500   if (!isal) {
 501     // skip the test if the plugin is not ready
 502     return;
 503   }
 504   g_conf().set_val("compressor_zlib_isal", "false");
 505   g_ceph_context->_conf.apply_changes(nullptr);
 506   CompressorRef zlib = Compressor::create(g_ceph_context, "zlib");
 507
 508   for (int cnt=0; cnt<100; cnt++)
 509   {
 510     srand(cnt + 1000);
 511     int log2 = (rand()%18) + 1;
 512     int size = (rand() % (1 << log2)) + 1;
 513
 514     int range = 1;
 515
 516     char test[size];
 517     test[0] = rand()%256;
 518     for (int i=1; i<size; ++i)
 519       test[i] = test[i-1] + rand()%(range*2+1) - range;
 520     bufferlist in, out;
 521     in.append(test, size);
 522
 523     boost::optional<int32_t> compressor_message;
 524     int res = isal->compress(in, out, compressor_message);
 525     EXPECT_EQ(res, 0);
 526     bufferlist after;
 527     res = zlib->decompress(out, after, compressor_message);
 528     EXPECT_EQ(res, 0);
 529     bufferlist exp;
 530     exp.append(test, size);
 531     EXPECT_TRUE(exp.contents_equal(after));
 532   }
 533 }
 534
 535 #endif  // __x86_64__
 536
 537 #ifdef HAVE_QATZIP
 538 TEST(QAT, enc_qat_dec_noqat) {
 539 #ifdef HAVE_LZ4
 540   const char* alg_collection[] = {"zlib", "lz4", "snappy"};
 541 #else
 542   const char* alg_collection[] = {"zlib", "snappy"};
 543 #endif
 544   for (auto alg : alg_collection) {
 545     g_conf().set_val("qat_compressor_enabled", "true");
 546     CompressorRef q = Compressor::create(g_ceph_context, alg);
 547     g_conf().set_val("qat_compressor_enabled", "false");
 548     CompressorRef noq = Compressor::create(g_ceph_context, alg);
 549
 550     // generate random buffer
 551     for (int cnt=0; cnt<100; cnt++) {
 552       srand(cnt + 1000);
 553       int log2 = (rand()%18) + 1;
 554       int size = (rand() % (1 << log2)) + 1;
 555
 556       char test[size];
 557       for (int i=0; i<size; ++i)
 558         test[i] = rand()%256;
 559       bufferlist in, out;
 560       in.append(test, size);
 561
 562       boost::optional<int32_t> compressor_message;
 563       int res = q->compress(in, out, compressor_message);
 564       EXPECT_EQ(res, 0);
 565       bufferlist after;
 566       res = noq->decompress(out, after, compressor_message);
 567       EXPECT_EQ(res, 0);
 568       bufferlist exp;
 569       exp.append(test, size);
 570       EXPECT_TRUE(exp.contents_equal(after));
 571     }
 572   }
 573 }
 574
 575 TEST(QAT, enc_noqat_dec_qat) {
 576 #ifdef HAVE_LZ4
 577   const char* alg_collection[] = {"zlib", "lz4", "snappy"};
 578 #else
 579   const char* alg_collection[] = {"zlib", "snappy"};
 580 #endif
 581   for (auto alg : alg_collection) {
 582     g_conf().set_val("qat_compressor_enabled", "true");
 583     CompressorRef q = Compressor::create(g_ceph_context, alg);
 584     g_conf().set_val("qat_compressor_enabled", "false");
 585     CompressorRef noq = Compressor::create(g_ceph_context, alg);
 586
 587     // generate random buffer
 588     for (int cnt=0; cnt<100; cnt++) {
 589       srand(cnt + 1000);
 590       int log2 = (rand()%18) + 1;
 591       int size = (rand() % (1 << log2)) + 1;
 592
 593       char test[size];
 594       for (int i=0; i<size; ++i)
 595         test[i] = rand()%256;
 596       bufferlist in, out;
 597       in.append(test, size);
 598
 599       boost::optional<int32_t> compressor_message;
 600       int res = noq->compress(in, out, compressor_message);
 601       EXPECT_EQ(res, 0);
 602       bufferlist after;
 603       res = q->decompress(out, after, compressor_message);
 604       EXPECT_EQ(res, 0);
 605       bufferlist exp;
 606       exp.append(test, size);
 607       EXPECT_TRUE(exp.contents_equal(after));
 608     }
 609   }
 610 }
 611
 612 #endif  // HAVE_QATZIP