ceph/src/rocksdb/db/db_dynamic_level_test.cc

   1 //  Copyright (c) 2011-present, 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
  10 // Introduction of SyncPoint effectively disabled building and running this test
  11 // in Release build.
  12 // which is a pity, it is a good test
  13 #if !defined(ROCKSDB_LITE)
  14
  15 #include "db/db_test_util.h"
  16 #include "port/port.h"
  17 #include "port/stack_trace.h"
  18
  19 namespace rocksdb {
  20 class DBTestDynamicLevel : public DBTestBase {
  21  public:
  22   DBTestDynamicLevel() : DBTestBase("/db_dynamic_level_test") {}
  23 };
  24
  25 TEST_F(DBTestDynamicLevel, DynamicLevelMaxBytesBase) {
  26   if (!Snappy_Supported() || !LZ4_Supported()) {
  27     return;
  28   }
  29   // Use InMemoryEnv, or it would be too slow.
  30   unique_ptr<Env> env(new MockEnv(env_));
  31
  32   const int kNKeys = 1000;
  33   int keys[kNKeys];
  34
  35   auto verify_func = [&]() {
  36     for (int i = 0; i < kNKeys; i++) {
  37       ASSERT_NE("NOT_FOUND", Get(Key(i)));
  38       ASSERT_NE("NOT_FOUND", Get(Key(kNKeys * 2 + i)));
  39       if (i < kNKeys / 10) {
  40         ASSERT_EQ("NOT_FOUND", Get(Key(kNKeys + keys[i])));
  41       } else {
  42         ASSERT_NE("NOT_FOUND", Get(Key(kNKeys + keys[i])));
  43       }
  44     }
  45   };
  46
  47   Random rnd(301);
  48   for (int ordered_insert = 0; ordered_insert <= 1; ordered_insert++) {
  49     for (int i = 0; i < kNKeys; i++) {
  50       keys[i] = i;
  51     }
  52     if (ordered_insert == 0) {
  53       std::random_shuffle(std::begin(keys), std::end(keys));
  54     }
  55     for (int max_background_compactions = 1; max_background_compactions < 4;
  56          max_background_compactions += 2) {
  57       Options options;
  58       options.env = env.get();
  59       options.create_if_missing = true;
  60       options.db_write_buffer_size = 2048;
  61       options.write_buffer_size = 2048;
  62       options.max_write_buffer_number = 2;
  63       options.level0_file_num_compaction_trigger = 2;
  64       options.level0_slowdown_writes_trigger = 2;
  65       options.level0_stop_writes_trigger = 2;
  66       options.target_file_size_base = 2048;
  67       options.level_compaction_dynamic_level_bytes = true;
  68       options.max_bytes_for_level_base = 10240;
  69       options.max_bytes_for_level_multiplier = 4;
  70       options.soft_rate_limit = 1.1;
  71       options.max_background_compactions = max_background_compactions;
  72       options.num_levels = 5;
  73
  74       options.compression_per_level.resize(3);
  75       options.compression_per_level[0] = kNoCompression;
  76       options.compression_per_level[1] = kLZ4Compression;
  77       options.compression_per_level[2] = kSnappyCompression;
  78       options.env = env_;
  79
  80       DestroyAndReopen(options);
  81
  82       for (int i = 0; i < kNKeys; i++) {
  83         int key = keys[i];
  84         ASSERT_OK(Put(Key(kNKeys + key), RandomString(&rnd, 102)));
  85         ASSERT_OK(Put(Key(key), RandomString(&rnd, 102)));
  86         ASSERT_OK(Put(Key(kNKeys * 2 + key), RandomString(&rnd, 102)));
  87         ASSERT_OK(Delete(Key(kNKeys + keys[i / 10])));
  88         env_->SleepForMicroseconds(5000);
  89       }
  90
  91       uint64_t int_prop;
  92       ASSERT_TRUE(db_->GetIntProperty("rocksdb.background-errors", &int_prop));
  93       ASSERT_EQ(0U, int_prop);
  94
  95       // Verify DB
  96       for (int j = 0; j < 2; j++) {
  97         verify_func();
  98         if (j == 0) {
  99           Reopen(options);
 100         }
 101       }
 102
 103       // Test compact range works
 104       dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
 105       // All data should be in the last level.
 106       ColumnFamilyMetaData cf_meta;
 107       db_->GetColumnFamilyMetaData(&cf_meta);
 108       ASSERT_EQ(5U, cf_meta.levels.size());
 109       for (int i = 0; i < 4; i++) {
 110         ASSERT_EQ(0U, cf_meta.levels[i].files.size());
 111       }
 112       ASSERT_GT(cf_meta.levels[4U].files.size(), 0U);
 113       verify_func();
 114
 115       Close();
 116     }
 117   }
 118
 119   env_->SetBackgroundThreads(1, Env::LOW);
 120   env_->SetBackgroundThreads(1, Env::HIGH);
 121 }
 122
 123 // Test specific cases in dynamic max bytes
 124 TEST_F(DBTestDynamicLevel, DynamicLevelMaxBytesBase2) {
 125   Random rnd(301);
 126   int kMaxKey = 1000000;
 127
 128   Options options = CurrentOptions();
 129   options.create_if_missing = true;
 130   options.db_write_buffer_size = 204800;
 131   options.write_buffer_size = 20480;
 132   options.max_write_buffer_number = 2;
 133   options.level0_file_num_compaction_trigger = 2;
 134   options.level0_slowdown_writes_trigger = 9999;
 135   options.level0_stop_writes_trigger = 9999;
 136   options.target_file_size_base = 9102;
 137   options.level_compaction_dynamic_level_bytes = true;
 138   options.max_bytes_for_level_base = 40960;
 139   options.max_bytes_for_level_multiplier = 4;
 140   options.max_background_compactions = 2;
 141   options.num_levels = 5;
 142   options.max_compaction_bytes = 0;  // Force not expanding in compactions
 143   BlockBasedTableOptions table_options;
 144   table_options.block_size = 1024;
 145   options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 146
 147   DestroyAndReopen(options);
 148   ASSERT_OK(dbfull()->SetOptions({
 149       {"disable_auto_compactions", "true"},
 150   }));
 151
 152   uint64_t int_prop;
 153   std::string str_prop;
 154
 155   // Initial base level is the last level
 156   ASSERT_TRUE(db_->GetIntProperty("rocksdb.base-level", &int_prop));
 157   ASSERT_EQ(4U, int_prop);
 158
 159   // Put about 28K to L0
 160   for (int i = 0; i < 70; i++) {
 161     ASSERT_OK(Put(Key(static_cast<int>(rnd.Uniform(kMaxKey))),
 162                   RandomString(&rnd, 380)));
 163   }
 164   ASSERT_OK(dbfull()->SetOptions({
 165       {"disable_auto_compactions", "false"},
 166   }));
 167   Flush();
 168   dbfull()->TEST_WaitForCompact();
 169   ASSERT_TRUE(db_->GetIntProperty("rocksdb.base-level", &int_prop));
 170   ASSERT_EQ(4U, int_prop);
 171
 172   // Insert extra about 28K to L0. After they are compacted to L4, base level
 173   // should be changed to L3.
 174   ASSERT_OK(dbfull()->SetOptions({
 175       {"disable_auto_compactions", "true"},
 176   }));
 177   for (int i = 0; i < 70; i++) {
 178     ASSERT_OK(Put(Key(static_cast<int>(rnd.Uniform(kMaxKey))),
 179                   RandomString(&rnd, 380)));
 180   }
 181
 182   ASSERT_OK(dbfull()->SetOptions({
 183       {"disable_auto_compactions", "false"},
 184   }));
 185   Flush();
 186   dbfull()->TEST_WaitForCompact();
 187   ASSERT_TRUE(db_->GetIntProperty("rocksdb.base-level", &int_prop));
 188   ASSERT_EQ(3U, int_prop);
 189   ASSERT_TRUE(db_->GetProperty("rocksdb.num-files-at-level1", &str_prop));
 190   ASSERT_EQ("0", str_prop);
 191   ASSERT_TRUE(db_->GetProperty("rocksdb.num-files-at-level2", &str_prop));
 192   ASSERT_EQ("0", str_prop);
 193
 194   // Trigger parallel compaction, and the first one would change the base
 195   // level.
 196   // Hold compaction jobs to make sure
 197   rocksdb::SyncPoint::GetInstance()->SetCallBack(
 198       "CompactionJob::Run():Start",
 199       [&](void* arg) { env_->SleepForMicroseconds(100000); });
 200   rocksdb::SyncPoint::GetInstance()->EnableProcessing();
 201   ASSERT_OK(dbfull()->SetOptions({
 202       {"disable_auto_compactions", "true"},
 203   }));
 204   // Write about 40K more
 205   for (int i = 0; i < 100; i++) {
 206     ASSERT_OK(Put(Key(static_cast<int>(rnd.Uniform(kMaxKey))),
 207                   RandomString(&rnd, 380)));
 208   }
 209   ASSERT_OK(dbfull()->SetOptions({
 210       {"disable_auto_compactions", "false"},
 211   }));
 212   Flush();
 213   // Wait for 200 milliseconds before proceeding compactions to make sure two
 214   // parallel ones are executed.
 215   env_->SleepForMicroseconds(200000);
 216   dbfull()->TEST_WaitForCompact();
 217   ASSERT_TRUE(db_->GetIntProperty("rocksdb.base-level", &int_prop));
 218   ASSERT_EQ(3U, int_prop);
 219   rocksdb::SyncPoint::GetInstance()->DisableProcessing();
 220
 221   // Trigger a condition that the compaction changes base level and L0->Lbase
 222   // happens at the same time.
 223   // We try to make last levels' targets to be 40K, 160K, 640K, add triggers
 224   // another compaction from 40K->160K.
 225   ASSERT_OK(dbfull()->SetOptions({
 226       {"disable_auto_compactions", "true"},
 227   }));
 228   // Write about 650K more.
 229   // Each file is about 11KB, with 9KB of data.
 230   for (int i = 0; i < 1300; i++) {
 231     ASSERT_OK(Put(Key(static_cast<int>(rnd.Uniform(kMaxKey))),
 232                   RandomString(&rnd, 380)));
 233   }
 234   ASSERT_OK(dbfull()->SetOptions({
 235       {"disable_auto_compactions", "false"},
 236   }));
 237   Flush();
 238   dbfull()->TEST_WaitForCompact();
 239   ASSERT_TRUE(db_->GetIntProperty("rocksdb.base-level", &int_prop));
 240   ASSERT_EQ(2U, int_prop);
 241
 242   // A manual compaction will trigger the base level to become L2
 243   // Keep Writing data until base level changed 2->1. There will be L0->L2
 244   // compaction going on at the same time.
 245   rocksdb::SyncPoint::GetInstance()->DisableProcessing();
 246   rocksdb::SyncPoint::GetInstance()->ClearAllCallBacks();
 247
 248   rocksdb::SyncPoint::GetInstance()->LoadDependency({
 249       {"CompactionJob::Run():Start", "DynamicLevelMaxBytesBase2:0"},
 250       {"DynamicLevelMaxBytesBase2:1", "CompactionJob::Run():End"},
 251       {"DynamicLevelMaxBytesBase2:compact_range_finish",
 252        "FlushJob::WriteLevel0Table"},
 253   });
 254   rocksdb::SyncPoint::GetInstance()->EnableProcessing();
 255
 256   rocksdb::port::Thread thread([this] {
 257     TEST_SYNC_POINT("DynamicLevelMaxBytesBase2:compact_range_start");
 258     ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
 259     TEST_SYNC_POINT("DynamicLevelMaxBytesBase2:compact_range_finish");
 260   });
 261
 262   TEST_SYNC_POINT("DynamicLevelMaxBytesBase2:0");
 263   for (int i = 0; i < 2; i++) {
 264     ASSERT_OK(Put(Key(static_cast<int>(rnd.Uniform(kMaxKey))),
 265                   RandomString(&rnd, 380)));
 266   }
 267   TEST_SYNC_POINT("DynamicLevelMaxBytesBase2:1");
 268
 269   Flush();
 270
 271   thread.join();
 272
 273   rocksdb::SyncPoint::GetInstance()->DisableProcessing();
 274   rocksdb::SyncPoint::GetInstance()->ClearAllCallBacks();
 275
 276   ASSERT_TRUE(db_->GetIntProperty("rocksdb.base-level", &int_prop));
 277   ASSERT_EQ(1U, int_prop);
 278 }
 279
 280 // Test specific cases in dynamic max bytes
 281 TEST_F(DBTestDynamicLevel, DynamicLevelMaxBytesCompactRange) {
 282   Random rnd(301);
 283   int kMaxKey = 1000000;
 284
 285   Options options = CurrentOptions();
 286   options.create_if_missing = true;
 287   options.db_write_buffer_size = 2048;
 288   options.write_buffer_size = 2048;
 289   options.max_write_buffer_number = 2;
 290   options.level0_file_num_compaction_trigger = 2;
 291   options.level0_slowdown_writes_trigger = 9999;
 292   options.level0_stop_writes_trigger = 9999;
 293   options.target_file_size_base = 2;
 294   options.level_compaction_dynamic_level_bytes = true;
 295   options.max_bytes_for_level_base = 10240;
 296   options.max_bytes_for_level_multiplier = 4;
 297   options.max_background_compactions = 1;
 298   const int kNumLevels = 5;
 299   options.num_levels = kNumLevels;
 300   options.max_compaction_bytes = 1;  // Force not expanding in compactions
 301   BlockBasedTableOptions table_options;
 302   table_options.block_size = 1024;
 303   options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 304
 305   DestroyAndReopen(options);
 306
 307   // Compact against empty DB
 308   dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
 309
 310   uint64_t int_prop;
 311   std::string str_prop;
 312
 313   // Initial base level is the last level
 314   ASSERT_TRUE(db_->GetIntProperty("rocksdb.base-level", &int_prop));
 315   ASSERT_EQ(4U, int_prop);
 316
 317   // Put about 7K to L0
 318   for (int i = 0; i < 140; i++) {
 319     ASSERT_OK(Put(Key(static_cast<int>(rnd.Uniform(kMaxKey))),
 320                   RandomString(&rnd, 80)));
 321   }
 322   Flush();
 323   dbfull()->TEST_WaitForCompact();
 324   if (NumTableFilesAtLevel(0) == 0) {
 325     // Make sure level 0 is not empty
 326     ASSERT_OK(Put(Key(static_cast<int>(rnd.Uniform(kMaxKey))),
 327                   RandomString(&rnd, 80)));
 328     Flush();
 329   }
 330
 331   ASSERT_TRUE(db_->GetIntProperty("rocksdb.base-level", &int_prop));
 332   ASSERT_EQ(3U, int_prop);
 333   ASSERT_TRUE(db_->GetProperty("rocksdb.num-files-at-level1", &str_prop));
 334   ASSERT_EQ("0", str_prop);
 335   ASSERT_TRUE(db_->GetProperty("rocksdb.num-files-at-level2", &str_prop));
 336   ASSERT_EQ("0", str_prop);
 337
 338   rocksdb::SyncPoint::GetInstance()->DisableProcessing();
 339   rocksdb::SyncPoint::GetInstance()->ClearAllCallBacks();
 340
 341   std::set<int> output_levels;
 342   rocksdb::SyncPoint::GetInstance()->SetCallBack(
 343       "CompactionPicker::CompactRange:Return", [&](void* arg) {
 344         Compaction* compaction = reinterpret_cast<Compaction*>(arg);
 345         output_levels.insert(compaction->output_level());
 346       });
 347   rocksdb::SyncPoint::GetInstance()->EnableProcessing();
 348
 349   dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr);
 350   ASSERT_EQ(output_levels.size(), 2);
 351   ASSERT_TRUE(output_levels.find(3) != output_levels.end());
 352   ASSERT_TRUE(output_levels.find(4) != output_levels.end());
 353   ASSERT_TRUE(db_->GetProperty("rocksdb.num-files-at-level0", &str_prop));
 354   ASSERT_EQ("0", str_prop);
 355   ASSERT_TRUE(db_->GetProperty("rocksdb.num-files-at-level3", &str_prop));
 356   ASSERT_EQ("0", str_prop);
 357   // Base level is still level 3.
 358   ASSERT_TRUE(db_->GetIntProperty("rocksdb.base-level", &int_prop));
 359   ASSERT_EQ(3U, int_prop);
 360 }
 361
 362 TEST_F(DBTestDynamicLevel, DynamicLevelMaxBytesBaseInc) {
 363   Options options = CurrentOptions();
 364   options.create_if_missing = true;
 365   options.db_write_buffer_size = 2048;
 366   options.write_buffer_size = 2048;
 367   options.max_write_buffer_number = 2;
 368   options.level0_file_num_compaction_trigger = 2;
 369   options.level0_slowdown_writes_trigger = 2;
 370   options.level0_stop_writes_trigger = 2;
 371   options.target_file_size_base = 2048;
 372   options.level_compaction_dynamic_level_bytes = true;
 373   options.max_bytes_for_level_base = 10240;
 374   options.max_bytes_for_level_multiplier = 4;
 375   options.soft_rate_limit = 1.1;
 376   options.max_background_compactions = 2;
 377   options.num_levels = 5;
 378
 379   DestroyAndReopen(options);
 380
 381   int non_trivial = 0;
 382   rocksdb::SyncPoint::GetInstance()->SetCallBack(
 383       "DBImpl::BackgroundCompaction:NonTrivial",
 384       [&](void* arg) { non_trivial++; });
 385   rocksdb::SyncPoint::GetInstance()->EnableProcessing();
 386
 387   Random rnd(301);
 388   const int total_keys = 3000;
 389   const int random_part_size = 100;
 390   for (int i = 0; i < total_keys; i++) {
 391     std::string value = RandomString(&rnd, random_part_size);
 392     PutFixed32(&value, static_cast<uint32_t>(i));
 393     ASSERT_OK(Put(Key(i), value));
 394   }
 395   Flush();
 396   dbfull()->TEST_WaitForCompact();
 397   rocksdb::SyncPoint::GetInstance()->DisableProcessing();
 398
 399   ASSERT_EQ(non_trivial, 0);
 400
 401   for (int i = 0; i < total_keys; i++) {
 402     std::string value = Get(Key(i));
 403     ASSERT_EQ(DecodeFixed32(value.c_str() + random_part_size),
 404               static_cast<uint32_t>(i));
 405   }
 406
 407   env_->SetBackgroundThreads(1, Env::LOW);
 408   env_->SetBackgroundThreads(1, Env::HIGH);
 409 }
 410
 411 TEST_F(DBTestDynamicLevel, DISABLED_MigrateToDynamicLevelMaxBytesBase) {
 412   Random rnd(301);
 413   const int kMaxKey = 2000;
 414
 415   Options options;
 416   options.create_if_missing = true;
 417   options.db_write_buffer_size = 2048;
 418   options.write_buffer_size = 2048;
 419   options.max_write_buffer_number = 8;
 420   options.level0_file_num_compaction_trigger = 4;
 421   options.level0_slowdown_writes_trigger = 4;
 422   options.level0_stop_writes_trigger = 8;
 423   options.target_file_size_base = 2048;
 424   options.level_compaction_dynamic_level_bytes = false;
 425   options.max_bytes_for_level_base = 10240;
 426   options.max_bytes_for_level_multiplier = 4;
 427   options.soft_rate_limit = 1.1;
 428   options.num_levels = 8;
 429
 430   DestroyAndReopen(options);
 431
 432   auto verify_func = [&](int num_keys, bool if_sleep) {
 433     for (int i = 0; i < num_keys; i++) {
 434       ASSERT_NE("NOT_FOUND", Get(Key(kMaxKey + i)));
 435       if (i < num_keys / 10) {
 436         ASSERT_EQ("NOT_FOUND", Get(Key(i)));
 437       } else {
 438         ASSERT_NE("NOT_FOUND", Get(Key(i)));
 439       }
 440       if (if_sleep && i % 1000 == 0) {
 441         // Without it, valgrind may choose not to give another
 442         // thread a chance to run before finishing the function,
 443         // causing the test to be extremely slow.
 444         env_->SleepForMicroseconds(1);
 445       }
 446     }
 447   };
 448
 449   int total_keys = 1000;
 450   for (int i = 0; i < total_keys; i++) {
 451     ASSERT_OK(Put(Key(i), RandomString(&rnd, 102)));
 452     ASSERT_OK(Put(Key(kMaxKey + i), RandomString(&rnd, 102)));
 453     ASSERT_OK(Delete(Key(i / 10)));
 454   }
 455   verify_func(total_keys, false);
 456   dbfull()->TEST_WaitForCompact();
 457
 458   options.level_compaction_dynamic_level_bytes = true;
 459   options.disable_auto_compactions = true;
 460   Reopen(options);
 461   verify_func(total_keys, false);
 462
 463   std::atomic_bool compaction_finished;
 464   compaction_finished = false;
 465   // Issue manual compaction in one thread and still verify DB state
 466   // in main thread.
 467   rocksdb::port::Thread t([&]() {
 468     CompactRangeOptions compact_options;
 469     compact_options.change_level = true;
 470     compact_options.target_level = options.num_levels - 1;
 471     dbfull()->CompactRange(compact_options, nullptr, nullptr);
 472     compaction_finished.store(true);
 473   });
 474   do {
 475     verify_func(total_keys, true);
 476   } while (!compaction_finished.load());
 477   t.join();
 478
 479   ASSERT_OK(dbfull()->SetOptions({
 480       {"disable_auto_compactions", "false"},
 481   }));
 482
 483   int total_keys2 = 2000;
 484   for (int i = total_keys; i < total_keys2; i++) {
 485     ASSERT_OK(Put(Key(i), RandomString(&rnd, 102)));
 486     ASSERT_OK(Put(Key(kMaxKey + i), RandomString(&rnd, 102)));
 487     ASSERT_OK(Delete(Key(i / 10)));
 488   }
 489
 490   verify_func(total_keys2, false);
 491   dbfull()->TEST_WaitForCompact();
 492   verify_func(total_keys2, false);
 493
 494   // Base level is not level 1
 495   ASSERT_EQ(NumTableFilesAtLevel(1), 0);
 496   ASSERT_EQ(NumTableFilesAtLevel(2), 0);
 497 }
 498 }  // namespace rocksdb
 499
 500 #endif  // !defined(ROCKSDB_LITE)
 501
 502 int main(int argc, char** argv) {
 503 #if !defined(ROCKSDB_LITE)
 504   rocksdb::port::InstallStackTraceHandler();
 505   ::testing::InitGoogleTest(&argc, argv);
 506   return RUN_ALL_TESTS();
 507 #else
 508   return 0;
 509 #endif
 510 }