ceph/src/rocksdb/utilities/persistent_cache/persistent_cache_test.cc

   1 //  Copyright (c) 2013, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under the BSD-style license found in the
   3 //  LICENSE file in the root directory of this source tree. An additional grant
   4 //  of patent rights can be found in the PATENTS file in the same directory.
   5 //
   6 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   7 // Use of this source code is governed by a BSD-style license that can be
   8 // found in the LICENSE file. See the AUTHORS file for names of contributors.
   9 #ifndef ROCKSDB_LITE
  10
  11 #include "utilities/persistent_cache/persistent_cache_test.h"
  12
  13 #include <functional>
  14 #include <memory>
  15 #include <thread>
  16
  17 #include "utilities/persistent_cache/block_cache_tier.h"
  18
  19 namespace rocksdb {
  20
  21 static const double kStressFactor = .125;
  22
  23 #ifdef OS_LINUX
  24 static void OnOpenForRead(void* arg) {
  25   int* val = static_cast<int*>(arg);
  26   *val &= ~O_DIRECT;
  27   rocksdb::SyncPoint::GetInstance()->SetCallBack(
  28       "NewRandomAccessFile:O_DIRECT",
  29       std::bind(OnOpenForRead, std::placeholders::_1));
  30 }
  31
  32 static void OnOpenForWrite(void* arg) {
  33   int* val = static_cast<int*>(arg);
  34   *val &= ~O_DIRECT;
  35   rocksdb::SyncPoint::GetInstance()->SetCallBack(
  36       "NewWritableFile:O_DIRECT",
  37       std::bind(OnOpenForWrite, std::placeholders::_1));
  38 }
  39 #endif
  40
  41 static void RemoveDirectory(const std::string& folder) {
  42   std::vector<std::string> files;
  43   Status status = Env::Default()->GetChildren(folder, &files);
  44   if (!status.ok()) {
  45     // we assume the directory does not exist
  46     return;
  47   }
  48
  49   // cleanup files with the patter :digi:.rc
  50   for (auto file : files) {
  51     if (file == "." || file == "..") {
  52       continue;
  53     }
  54     status = Env::Default()->DeleteFile(folder + "/" + file);
  55     assert(status.ok());
  56   }
  57
  58   status = Env::Default()->DeleteDir(folder);
  59   assert(status.ok());
  60 }
  61
  62 static void OnDeleteDir(void* arg) {
  63   char* dir = static_cast<char*>(arg);
  64   RemoveDirectory(std::string(dir));
  65 }
  66
  67 //
  68 // Simple logger that prints message on stdout
  69 //
  70 class ConsoleLogger : public Logger {
  71  public:
  72   using Logger::Logv;
  73   ConsoleLogger() : Logger(InfoLogLevel::ERROR_LEVEL) {}
  74
  75   void Logv(const char* format, va_list ap) override {
  76     MutexLock _(&lock_);
  77     vprintf(format, ap);
  78     printf("\n");
  79   }
  80
  81   port::Mutex lock_;
  82 };
  83
  84 // construct a tiered RAM+Block cache
  85 std::unique_ptr<PersistentTieredCache> NewTieredCache(
  86     const size_t mem_size, const PersistentCacheConfig& opt) {
  87   std::unique_ptr<PersistentTieredCache> tcache(new PersistentTieredCache());
  88   // create primary tier
  89   assert(mem_size);
  90   auto pcache = std::shared_ptr<PersistentCacheTier>(new VolatileCacheTier(
  91       /*is_compressed*/ true, mem_size));
  92   tcache->AddTier(pcache);
  93   // create secondary tier
  94   auto scache = std::shared_ptr<PersistentCacheTier>(new BlockCacheTier(opt));
  95   tcache->AddTier(scache);
  96
  97   Status s = tcache->Open();
  98   assert(s.ok());
  99   return tcache;
 100 }
 101
 102 // create block cache
 103 std::unique_ptr<PersistentCacheTier> NewBlockCache(
 104     Env* env, const std::string& path,
 105     const uint64_t max_size = std::numeric_limits<uint64_t>::max(),
 106     const bool enable_direct_writes = false) {
 107   const uint32_t max_file_size = static_cast<uint32_t>(12 * 1024 * 1024 * kStressFactor);
 108   auto log = std::make_shared<ConsoleLogger>();
 109   PersistentCacheConfig opt(env, path, max_size, log);
 110   opt.cache_file_size = max_file_size;
 111   opt.max_write_pipeline_backlog_size = std::numeric_limits<uint64_t>::max();
 112   opt.enable_direct_writes = enable_direct_writes;
 113   std::unique_ptr<PersistentCacheTier> scache(new BlockCacheTier(opt));
 114   Status s = scache->Open();
 115   assert(s.ok());
 116   return scache;
 117 }
 118
 119 // create a new cache tier
 120 std::unique_ptr<PersistentTieredCache> NewTieredCache(
 121     Env* env, const std::string& path, const uint64_t max_volatile_cache_size,
 122     const uint64_t max_block_cache_size =
 123         std::numeric_limits<uint64_t>::max()) {
 124   const uint32_t max_file_size = static_cast<uint32_t>(12 * 1024 * 1024 * kStressFactor);
 125   auto log = std::make_shared<ConsoleLogger>();
 126   auto opt = PersistentCacheConfig(env, path, max_block_cache_size, log);
 127   opt.cache_file_size = max_file_size;
 128   opt.max_write_pipeline_backlog_size = std::numeric_limits<uint64_t>::max();
 129   // create tier out of the two caches
 130   auto cache = NewTieredCache(max_volatile_cache_size, opt);
 131   return cache;
 132 }
 133
 134 PersistentCacheTierTest::PersistentCacheTierTest()
 135     : path_(test::TmpDir(Env::Default()) + "/cache_test") {
 136 #ifdef OS_LINUX
 137   rocksdb::SyncPoint::GetInstance()->EnableProcessing();
 138   rocksdb::SyncPoint::GetInstance()->SetCallBack("NewRandomAccessFile:O_DIRECT",
 139                                                  OnOpenForRead);
 140   rocksdb::SyncPoint::GetInstance()->SetCallBack("NewWritableFile:O_DIRECT",
 141                                                  OnOpenForWrite);
 142 #endif
 143 }
 144
 145 // Block cache tests
 146 TEST_F(PersistentCacheTierTest, DISABLED_BlockCacheInsertWithFileCreateError) {
 147   cache_ = NewBlockCache(Env::Default(), path_,
 148                          /*size=*/std::numeric_limits<uint64_t>::max(),
 149                          /*direct_writes=*/ false);
 150   rocksdb::SyncPoint::GetInstance()->SetCallBack(
 151     "BlockCacheTier::NewCacheFile:DeleteDir", OnDeleteDir);
 152
 153   RunNegativeInsertTest(/*nthreads=*/ 1,
 154                         /*max_keys*/
 155                           static_cast<size_t>(10 * 1024 * kStressFactor));
 156
 157   rocksdb::SyncPoint::GetInstance()->ClearAllCallBacks();
 158 }
 159
 160 #ifdef TRAVIS
 161 // Travis is unable to handle the normal version of the tests running out of
 162 // fds, out of space and timeouts. This is an easier version of the test
 163 // specifically written for Travis
 164 TEST_F(PersistentCacheTierTest, BasicTest) {
 165   cache_ = std::make_shared<VolatileCacheTier>();
 166   RunInsertTest(/*nthreads=*/1, /*max_keys=*/1024);
 167
 168   cache_ = NewBlockCache(Env::Default(), path_,
 169                          /*size=*/std::numeric_limits<uint64_t>::max(),
 170                          /*direct_writes=*/true);
 171   RunInsertTest(/*nthreads=*/1, /*max_keys=*/1024);
 172
 173   cache_ = NewTieredCache(Env::Default(), path_,
 174                           /*memory_size=*/static_cast<size_t>(1 * 1024 * 1024));
 175   RunInsertTest(/*nthreads=*/1, /*max_keys=*/1024);
 176 }
 177 #else
 178 // Volatile cache tests
 179 TEST_F(PersistentCacheTierTest, VolatileCacheInsert) {
 180   for (auto nthreads : {1, 5}) {
 181     for (auto max_keys :
 182          {10 * 1024 * kStressFactor, 1 * 1024 * 1024 * kStressFactor}) {
 183       cache_ = std::make_shared<VolatileCacheTier>();
 184       RunInsertTest(nthreads, static_cast<size_t>(max_keys));
 185     }
 186   }
 187 }
 188
 189 TEST_F(PersistentCacheTierTest, VolatileCacheInsertWithEviction) {
 190   for (auto nthreads : {1, 5}) {
 191     for (auto max_keys : {1 * 1024 * 1024 * kStressFactor}) {
 192       cache_ = std::make_shared<VolatileCacheTier>(
 193           /*compressed=*/true, /*size=*/static_cast<size_t>(1 * 1024 * 1024 * kStressFactor));
 194       RunInsertTestWithEviction(nthreads, static_cast<size_t>(max_keys));
 195     }
 196   }
 197 }
 198
 199 // Block cache tests
 200 TEST_F(PersistentCacheTierTest, BlockCacheInsert) {
 201   for (auto direct_writes : {true, false}) {
 202     for (auto nthreads : {1, 5}) {
 203       for (auto max_keys :
 204            {10 * 1024 * kStressFactor, 1 * 1024 * 1024 * kStressFactor}) {
 205         cache_ = NewBlockCache(Env::Default(), path_,
 206                                /*size=*/std::numeric_limits<uint64_t>::max(),
 207                                direct_writes);
 208         RunInsertTest(nthreads, static_cast<size_t>(max_keys));
 209       }
 210     }
 211   }
 212 }
 213
 214 TEST_F(PersistentCacheTierTest, BlockCacheInsertWithEviction) {
 215   for (auto nthreads : {1, 5}) {
 216     for (auto max_keys : {1 * 1024 * 1024 * kStressFactor}) {
 217       cache_ = NewBlockCache(Env::Default(), path_,
 218                              /*max_size=*/static_cast<size_t>(200 * 1024 * 1024 * kStressFactor));
 219       RunInsertTestWithEviction(nthreads, static_cast<size_t>(max_keys));
 220     }
 221   }
 222 }
 223
 224 // Tiered cache tests
 225 TEST_F(PersistentCacheTierTest, TieredCacheInsert) {
 226   for (auto nthreads : {1, 5}) {
 227     for (auto max_keys :
 228          {10 * 1024 * kStressFactor, 1 * 1024 * 1024 * kStressFactor}) {
 229       cache_ = NewTieredCache(Env::Default(), path_,
 230                               /*memory_size=*/static_cast<size_t>(1 * 1024 * 1024 * kStressFactor));
 231       RunInsertTest(nthreads, static_cast<size_t>(max_keys));
 232     }
 233   }
 234 }
 235
 236 // the tests causes a lot of file deletions which Travis limited testing
 237 // environment cannot handle
 238 TEST_F(PersistentCacheTierTest, TieredCacheInsertWithEviction) {
 239   for (auto nthreads : {1, 5}) {
 240     for (auto max_keys : {1 * 1024 * 1024 * kStressFactor}) {
 241       cache_ = NewTieredCache(
 242           Env::Default(), path_,
 243           /*memory_size=*/static_cast<size_t>(1 * 1024 * 1024 * kStressFactor),
 244           /*block_cache_size*/ static_cast<size_t>(200 * 1024 * 1024 * kStressFactor));
 245       RunInsertTestWithEviction(nthreads, static_cast<size_t>(max_keys));
 246     }
 247   }
 248 }
 249 #endif
 250
 251 std::shared_ptr<PersistentCacheTier> MakeVolatileCache(
 252     const std::string& /*dbname*/) {
 253   return std::make_shared<VolatileCacheTier>();
 254 }
 255
 256 std::shared_ptr<PersistentCacheTier> MakeBlockCache(const std::string& dbname) {
 257   return NewBlockCache(Env::Default(), dbname);
 258 }
 259
 260 std::shared_ptr<PersistentCacheTier> MakeTieredCache(
 261     const std::string& dbname) {
 262   const auto memory_size = 1 * 1024 * 1024 * kStressFactor;
 263   return NewTieredCache(Env::Default(), dbname, static_cast<size_t>(memory_size));
 264 }
 265
 266 #ifdef OS_LINUX
 267 static void UniqueIdCallback(void* arg) {
 268   int* result = reinterpret_cast<int*>(arg);
 269   if (*result == -1) {
 270     *result = 0;
 271   }
 272
 273   rocksdb::SyncPoint::GetInstance()->ClearTrace();
 274   rocksdb::SyncPoint::GetInstance()->SetCallBack(
 275       "GetUniqueIdFromFile:FS_IOC_GETVERSION", UniqueIdCallback);
 276 }
 277 #endif
 278
 279 TEST_F(PersistentCacheTierTest, FactoryTest) {
 280   for (auto nvm_opt : {true, false}) {
 281     ASSERT_FALSE(cache_);
 282     auto log = std::make_shared<ConsoleLogger>();
 283     std::shared_ptr<PersistentCache> cache;
 284     ASSERT_OK(NewPersistentCache(Env::Default(), path_,
 285                                  /*size=*/1 * 1024 * 1024 * 1024, log, nvm_opt,
 286                                  &cache));
 287     ASSERT_TRUE(cache);
 288     ASSERT_EQ(cache->Stats().size(), 1);
 289     ASSERT_TRUE(cache->Stats()[0].size());
 290     cache.reset();
 291   }
 292 }
 293
 294 PersistentCacheDBTest::PersistentCacheDBTest() : DBTestBase("/cache_test") {
 295 #ifdef OS_LINUX
 296   rocksdb::SyncPoint::GetInstance()->EnableProcessing();
 297   rocksdb::SyncPoint::GetInstance()->SetCallBack(
 298       "GetUniqueIdFromFile:FS_IOC_GETVERSION", UniqueIdCallback);
 299   rocksdb::SyncPoint::GetInstance()->SetCallBack("NewRandomAccessFile:O_DIRECT",
 300                                                  OnOpenForRead);
 301 #endif
 302 }
 303
 304 // test template
 305 void PersistentCacheDBTest::RunTest(
 306     const std::function<std::shared_ptr<PersistentCacheTier>(bool)>& new_pcache,
 307     const size_t max_keys = 100 * 1024, const size_t max_usecase = 5) {
 308   if (!Snappy_Supported()) {
 309     return;
 310   }
 311
 312   // number of insertion interations
 313   int num_iter = static_cast<int>(max_keys * kStressFactor);
 314
 315   for (size_t iter = 0; iter < max_usecase; iter++) {
 316     Options options;
 317     options.write_buffer_size =
 318       static_cast<size_t>(64 * 1024 * kStressFactor);  // small write buffer
 319     options.statistics = rocksdb::CreateDBStatistics();
 320     options = CurrentOptions(options);
 321
 322     // setup page cache
 323     std::shared_ptr<PersistentCacheTier> pcache;
 324     BlockBasedTableOptions table_options;
 325     table_options.cache_index_and_filter_blocks = true;
 326
 327     const size_t size_max = std::numeric_limits<size_t>::max();
 328
 329     switch (iter) {
 330       case 0:
 331         // page cache, block cache, no-compressed cache
 332         pcache = new_pcache(/*is_compressed=*/true);
 333         table_options.persistent_cache = pcache;
 334         table_options.block_cache = NewLRUCache(size_max);
 335         table_options.block_cache_compressed = nullptr;
 336         options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 337         break;
 338       case 1:
 339         // page cache, block cache, compressed cache
 340         pcache = new_pcache(/*is_compressed=*/true);
 341         table_options.persistent_cache = pcache;
 342         table_options.block_cache = NewLRUCache(size_max);
 343         table_options.block_cache_compressed = NewLRUCache(size_max);
 344         options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 345         break;
 346       case 2:
 347         // page cache, block cache, compressed cache + KNoCompression
 348         // both block cache and compressed cache, but DB is not compressed
 349         // also, make block cache sizes bigger, to trigger block cache hits
 350         pcache = new_pcache(/*is_compressed=*/true);
 351         table_options.persistent_cache = pcache;
 352         table_options.block_cache = NewLRUCache(size_max);
 353         table_options.block_cache_compressed = NewLRUCache(size_max);
 354         options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 355         options.compression = kNoCompression;
 356         break;
 357       case 3:
 358         // page cache, no block cache, no compressed cache
 359         pcache = new_pcache(/*is_compressed=*/false);
 360         table_options.persistent_cache = pcache;
 361         table_options.block_cache = nullptr;
 362         table_options.block_cache_compressed = nullptr;
 363         options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 364         break;
 365       case 4:
 366         // page cache, no block cache, no compressed cache
 367         // Page cache caches compressed blocks
 368         pcache = new_pcache(/*is_compressed=*/true);
 369         table_options.persistent_cache = pcache;
 370         table_options.block_cache = nullptr;
 371         table_options.block_cache_compressed = nullptr;
 372         options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 373         break;
 374       default:
 375         ASSERT_TRUE(false);
 376     }
 377
 378     std::vector<std::string> values;
 379     // insert data
 380     Insert(options, table_options, num_iter, &values);
 381     // flush all data in cache to device
 382     pcache->TEST_Flush();
 383     // verify data
 384     Verify(num_iter, values);
 385
 386     auto block_miss = TestGetTickerCount(options, BLOCK_CACHE_MISS);
 387     auto compressed_block_hit =
 388         TestGetTickerCount(options, BLOCK_CACHE_COMPRESSED_HIT);
 389     auto compressed_block_miss =
 390         TestGetTickerCount(options, BLOCK_CACHE_COMPRESSED_MISS);
 391     auto page_hit = TestGetTickerCount(options, PERSISTENT_CACHE_HIT);
 392     auto page_miss = TestGetTickerCount(options, PERSISTENT_CACHE_MISS);
 393
 394     // check that we triggered the appropriate code paths in the cache
 395     switch (iter) {
 396       case 0:
 397         // page cache, block cache, no-compressed cache
 398         ASSERT_GT(page_miss, 0);
 399         ASSERT_GT(page_hit, 0);
 400         ASSERT_GT(block_miss, 0);
 401         ASSERT_EQ(compressed_block_miss, 0);
 402         ASSERT_EQ(compressed_block_hit, 0);
 403         break;
 404       case 1:
 405         // page cache, block cache, compressed cache
 406         ASSERT_GT(page_miss, 0);
 407         ASSERT_GT(block_miss, 0);
 408         ASSERT_GT(compressed_block_miss, 0);
 409         break;
 410       case 2:
 411         // page cache, block cache, compressed cache + KNoCompression
 412         ASSERT_GT(page_miss, 0);
 413         ASSERT_GT(page_hit, 0);
 414         ASSERT_GT(block_miss, 0);
 415         ASSERT_GT(compressed_block_miss, 0);
 416         // remember kNoCompression
 417         ASSERT_EQ(compressed_block_hit, 0);
 418         break;
 419       case 3:
 420       case 4:
 421         // page cache, no block cache, no compressed cache
 422         ASSERT_GT(page_miss, 0);
 423         ASSERT_GT(page_hit, 0);
 424         ASSERT_EQ(compressed_block_hit, 0);
 425         ASSERT_EQ(compressed_block_miss, 0);
 426         break;
 427       default:
 428         ASSERT_TRUE(false);
 429     }
 430
 431     options.create_if_missing = true;
 432     DestroyAndReopen(options);
 433
 434     pcache->Close();
 435   }
 436 }
 437
 438 #ifdef TRAVIS
 439 // Travis is unable to handle the normal version of the tests running out of
 440 // fds, out of space and timeouts. This is an easier version of the test
 441 // specifically written for Travis
 442 TEST_F(PersistentCacheDBTest, BasicTest) {
 443   RunTest(std::bind(&MakeBlockCache, dbname_), /*max_keys=*/1024,
 444           /*max_usecase=*/1);
 445 }
 446 #else
 447 // test table with block page cache
 448 TEST_F(PersistentCacheDBTest, BlockCacheTest) {
 449   RunTest(std::bind(&MakeBlockCache, dbname_));
 450 }
 451
 452 // test table with volatile page cache
 453 TEST_F(PersistentCacheDBTest, VolatileCacheTest) {
 454   RunTest(std::bind(&MakeVolatileCache, dbname_));
 455 }
 456
 457 // test table with tiered page cache
 458 TEST_F(PersistentCacheDBTest, TieredCacheTest) {
 459   RunTest(std::bind(&MakeTieredCache, dbname_));
 460 }
 461 #endif
 462
 463 }  // namespace rocksdb
 464
 465 int main(int argc, char** argv) {
 466   ::testing::InitGoogleTest(&argc, argv);
 467   return RUN_ALL_TESTS();
 468 }
 469 #else
 470 int main() { return 0; }
 471 #endif