1 // Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
2 // This source code is licensed under both the GPLv2 (found in the
3 // COPYING file in the root directory) and Apache 2.0 License
4 // (found in the LICENSE.Apache file in the root 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
11 #ifdef GFLAGS
12 #include "db_stress_tool/db_stress_common.h"
13 #include "db_stress_tool/db_stress_driver.h"
14 #include "rocksdb/convenience.h"
15
16 namespace ROCKSDB_NAMESPACE {
StressTest()17 StressTest::StressTest()
18 : cache_(NewCache(FLAGS_cache_size)),
19 compressed_cache_(NewLRUCache(FLAGS_compressed_cache_size)),
20 filter_policy_(FLAGS_bloom_bits >= 0
21 ? FLAGS_use_block_based_filter
22 ? NewBloomFilterPolicy(FLAGS_bloom_bits, true)
23 : NewBloomFilterPolicy(FLAGS_bloom_bits, false)
24 : nullptr),
25 db_(nullptr),
26 #ifndef ROCKSDB_LITE
27 txn_db_(nullptr),
28 #endif
29 new_column_family_name_(1),
30 num_times_reopened_(0),
31 db_preload_finished_(false),
32 cmp_db_(nullptr) {
33 if (FLAGS_destroy_db_initially) {
34 std::vector<std::string> files;
35 db_stress_env->GetChildren(FLAGS_db, &files);
36 for (unsigned int i = 0; i < files.size(); i++) {
37 if (Slice(files[i]).starts_with("heap-")) {
38 db_stress_env->DeleteFile(FLAGS_db + "/" + files[i]);
39 }
40 }
41
42 Options options;
43 // Remove files without preserving manfiest files
44 #ifndef ROCKSDB_LITE
45 const Status s = !FLAGS_use_blob_db
46 ? DestroyDB(FLAGS_db, options)
47 : blob_db::DestroyBlobDB(FLAGS_db, options,
48 blob_db::BlobDBOptions());
49 #else
50 const Status s = DestroyDB(FLAGS_db, options);
51 #endif // !ROCKSDB_LITE
52
53 if (!s.ok()) {
54 fprintf(stderr, "Cannot destroy original db: %s\n", s.ToString().c_str());
55 exit(1);
56 }
57 }
58 }
59
~StressTest()60 StressTest::~StressTest() {
61 for (auto cf : column_families_) {
62 delete cf;
63 }
64 column_families_.clear();
65 delete db_;
66
67 assert(secondaries_.size() == secondary_cfh_lists_.size());
68 size_t n = secondaries_.size();
69 for (size_t i = 0; i != n; ++i) {
70 for (auto* cf : secondary_cfh_lists_[i]) {
71 delete cf;
72 }
73 secondary_cfh_lists_[i].clear();
74 delete secondaries_[i];
75 }
76 secondaries_.clear();
77
78 for (auto* cf : cmp_cfhs_) {
79 delete cf;
80 }
81 cmp_cfhs_.clear();
82 delete cmp_db_;
83 }
84
NewCache(size_t capacity)85 std::shared_ptr<Cache> StressTest::NewCache(size_t capacity) {
86 if (capacity <= 0) {
87 return nullptr;
88 }
89 if (FLAGS_use_clock_cache) {
90 auto cache = NewClockCache((size_t)capacity);
91 if (!cache) {
92 fprintf(stderr, "Clock cache not supported.");
93 exit(1);
94 }
95 return cache;
96 } else {
97 return NewLRUCache((size_t)capacity);
98 }
99 }
100
BuildOptionsTable()101 bool StressTest::BuildOptionsTable() {
102 if (FLAGS_set_options_one_in <= 0) {
103 return true;
104 }
105
106 std::unordered_map<std::string, std::vector<std::string>> options_tbl = {
107 {"write_buffer_size",
108 {ToString(options_.write_buffer_size),
109 ToString(options_.write_buffer_size * 2),
110 ToString(options_.write_buffer_size * 4)}},
111 {"max_write_buffer_number",
112 {ToString(options_.max_write_buffer_number),
113 ToString(options_.max_write_buffer_number * 2),
114 ToString(options_.max_write_buffer_number * 4)}},
115 {"arena_block_size",
116 {
117 ToString(options_.arena_block_size),
118 ToString(options_.write_buffer_size / 4),
119 ToString(options_.write_buffer_size / 8),
120 }},
121 {"memtable_huge_page_size", {"0", ToString(2 * 1024 * 1024)}},
122 {"max_successive_merges", {"0", "2", "4"}},
123 {"inplace_update_num_locks", {"100", "200", "300"}},
124 // TODO(ljin): enable test for this option
125 // {"disable_auto_compactions", {"100", "200", "300"}},
126 {"soft_rate_limit", {"0", "0.5", "0.9"}},
127 {"hard_rate_limit", {"0", "1.1", "2.0"}},
128 {"level0_file_num_compaction_trigger",
129 {
130 ToString(options_.level0_file_num_compaction_trigger),
131 ToString(options_.level0_file_num_compaction_trigger + 2),
132 ToString(options_.level0_file_num_compaction_trigger + 4),
133 }},
134 {"level0_slowdown_writes_trigger",
135 {
136 ToString(options_.level0_slowdown_writes_trigger),
137 ToString(options_.level0_slowdown_writes_trigger + 2),
138 ToString(options_.level0_slowdown_writes_trigger + 4),
139 }},
140 {"level0_stop_writes_trigger",
141 {
142 ToString(options_.level0_stop_writes_trigger),
143 ToString(options_.level0_stop_writes_trigger + 2),
144 ToString(options_.level0_stop_writes_trigger + 4),
145 }},
146 {"max_compaction_bytes",
147 {
148 ToString(options_.target_file_size_base * 5),
149 ToString(options_.target_file_size_base * 15),
150 ToString(options_.target_file_size_base * 100),
151 }},
152 {"target_file_size_base",
153 {
154 ToString(options_.target_file_size_base),
155 ToString(options_.target_file_size_base * 2),
156 ToString(options_.target_file_size_base * 4),
157 }},
158 {"target_file_size_multiplier",
159 {
160 ToString(options_.target_file_size_multiplier),
161 "1",
162 "2",
163 }},
164 {"max_bytes_for_level_base",
165 {
166 ToString(options_.max_bytes_for_level_base / 2),
167 ToString(options_.max_bytes_for_level_base),
168 ToString(options_.max_bytes_for_level_base * 2),
169 }},
170 {"max_bytes_for_level_multiplier",
171 {
172 ToString(options_.max_bytes_for_level_multiplier),
173 "1",
174 "2",
175 }},
176 {"max_sequential_skip_in_iterations", {"4", "8", "12"}},
177 };
178
179 options_table_ = std::move(options_tbl);
180
181 for (const auto& iter : options_table_) {
182 options_index_.push_back(iter.first);
183 }
184 return true;
185 }
186
InitDb()187 void StressTest::InitDb() {
188 uint64_t now = db_stress_env->NowMicros();
189 fprintf(stdout, "%s Initializing db_stress\n",
190 db_stress_env->TimeToString(now / 1000000).c_str());
191 PrintEnv();
192 Open();
193 BuildOptionsTable();
194 }
195
InitReadonlyDb(SharedState * shared)196 void StressTest::InitReadonlyDb(SharedState* shared) {
197 uint64_t now = db_stress_env->NowMicros();
198 fprintf(stdout, "%s Preloading db with %" PRIu64 " KVs\n",
199 db_stress_env->TimeToString(now / 1000000).c_str(), FLAGS_max_key);
200 PreloadDbAndReopenAsReadOnly(FLAGS_max_key, shared);
201 }
202
VerifySecondaries()203 bool StressTest::VerifySecondaries() {
204 #ifndef ROCKSDB_LITE
205 if (FLAGS_test_secondary) {
206 uint64_t now = db_stress_env->NowMicros();
207 fprintf(
208 stdout, "%s Start to verify secondaries against primary\n",
209 db_stress_env->TimeToString(static_cast<uint64_t>(now) / 1000000).c_str());
210 }
211 for (size_t k = 0; k != secondaries_.size(); ++k) {
212 Status s = secondaries_[k]->TryCatchUpWithPrimary();
213 if (!s.ok()) {
214 fprintf(stderr, "Secondary failed to catch up with primary\n");
215 return false;
216 }
217 ReadOptions ropts;
218 ropts.total_order_seek = true;
219 // Verify only the default column family since the primary may have
220 // dropped other column families after most recent reopen.
221 std::unique_ptr<Iterator> iter1(db_->NewIterator(ropts));
222 std::unique_ptr<Iterator> iter2(secondaries_[k]->NewIterator(ropts));
223 for (iter1->SeekToFirst(), iter2->SeekToFirst();
224 iter1->Valid() && iter2->Valid(); iter1->Next(), iter2->Next()) {
225 if (iter1->key().compare(iter2->key()) != 0 ||
226 iter1->value().compare(iter2->value())) {
227 fprintf(stderr,
228 "Secondary %d contains different data from "
229 "primary.\nPrimary: %s : %s\nSecondary: %s : %s\n",
230 static_cast<int>(k),
231 iter1->key().ToString(/*hex=*/true).c_str(),
232 iter1->value().ToString(/*hex=*/true).c_str(),
233 iter2->key().ToString(/*hex=*/true).c_str(),
234 iter2->value().ToString(/*hex=*/true).c_str());
235 return false;
236 }
237 }
238 if (iter1->Valid() && !iter2->Valid()) {
239 fprintf(stderr,
240 "Secondary %d record count is smaller than that of primary\n",
241 static_cast<int>(k));
242 return false;
243 } else if (!iter1->Valid() && iter2->Valid()) {
244 fprintf(stderr,
245 "Secondary %d record count is larger than that of primary\n",
246 static_cast<int>(k));
247 return false;
248 }
249 }
250 if (FLAGS_test_secondary) {
251 uint64_t now = db_stress_env->NowMicros();
252 fprintf(
253 stdout, "%s Verification of secondaries succeeded\n",
254 db_stress_env->TimeToString(static_cast<uint64_t>(now) / 1000000).c_str());
255 }
256 #endif // ROCKSDB_LITE
257 return true;
258 }
259
AssertSame(DB * db,ColumnFamilyHandle * cf,ThreadState::SnapshotState & snap_state)260 Status StressTest::AssertSame(DB* db, ColumnFamilyHandle* cf,
261 ThreadState::SnapshotState& snap_state) {
262 Status s;
263 if (cf->GetName() != snap_state.cf_at_name) {
264 return s;
265 }
266 ReadOptions ropt;
267 ropt.snapshot = snap_state.snapshot;
268 PinnableSlice exp_v(&snap_state.value);
269 exp_v.PinSelf();
270 PinnableSlice v;
271 s = db->Get(ropt, cf, snap_state.key, &v);
272 if (!s.ok() && !s.IsNotFound()) {
273 return s;
274 }
275 if (snap_state.status != s) {
276 return Status::Corruption(
277 "The snapshot gave inconsistent results for key " +
278 ToString(Hash(snap_state.key.c_str(), snap_state.key.size(), 0)) +
279 " in cf " + cf->GetName() + ": (" + snap_state.status.ToString() +
280 ") vs. (" + s.ToString() + ")");
281 }
282 if (s.ok()) {
283 if (exp_v != v) {
284 return Status::Corruption("The snapshot gave inconsistent values: (" +
285 exp_v.ToString() + ") vs. (" + v.ToString() +
286 ")");
287 }
288 }
289 if (snap_state.key_vec != nullptr) {
290 // When `prefix_extractor` is set, seeking to beginning and scanning
291 // across prefixes are only supported with `total_order_seek` set.
292 ropt.total_order_seek = true;
293 std::unique_ptr<Iterator> iterator(db->NewIterator(ropt));
294 std::unique_ptr<std::vector<bool>> tmp_bitvec(
295 new std::vector<bool>(FLAGS_max_key));
296 for (iterator->SeekToFirst(); iterator->Valid(); iterator->Next()) {
297 uint64_t key_val;
298 if (GetIntVal(iterator->key().ToString(), &key_val)) {
299 (*tmp_bitvec.get())[key_val] = true;
300 }
301 }
302 if (!std::equal(snap_state.key_vec->begin(), snap_state.key_vec->end(),
303 tmp_bitvec.get()->begin())) {
304 return Status::Corruption("Found inconsistent keys at this snapshot");
305 }
306 }
307 return Status::OK();
308 }
309
VerificationAbort(SharedState * shared,std::string msg,Status s) const310 void StressTest::VerificationAbort(SharedState* shared, std::string msg,
311 Status s) const {
312 fprintf(stderr, "Verification failed: %s. Status is %s\n", msg.c_str(),
313 s.ToString().c_str());
314 shared->SetVerificationFailure();
315 }
316
VerificationAbort(SharedState * shared,std::string msg,int cf,int64_t key) const317 void StressTest::VerificationAbort(SharedState* shared, std::string msg, int cf,
318 int64_t key) const {
319 fprintf(stderr,
320 "Verification failed for column family %d key %" PRIi64 ": %s\n", cf,
321 key, msg.c_str());
322 shared->SetVerificationFailure();
323 }
324
PrintStatistics()325 void StressTest::PrintStatistics() {
326 if (dbstats) {
327 fprintf(stdout, "STATISTICS:\n%s\n", dbstats->ToString().c_str());
328 }
329 if (dbstats_secondaries) {
330 fprintf(stdout, "Secondary instances STATISTICS:\n%s\n",
331 dbstats_secondaries->ToString().c_str());
332 }
333 }
334
335 // Currently PreloadDb has to be single-threaded.
PreloadDbAndReopenAsReadOnly(int64_t number_of_keys,SharedState * shared)336 void StressTest::PreloadDbAndReopenAsReadOnly(int64_t number_of_keys,
337 SharedState* shared) {
338 WriteOptions write_opts;
339 write_opts.disableWAL = FLAGS_disable_wal;
340 if (FLAGS_sync) {
341 write_opts.sync = true;
342 }
343 char value[100];
344 int cf_idx = 0;
345 Status s;
346 for (auto cfh : column_families_) {
347 for (int64_t k = 0; k != number_of_keys; ++k) {
348 std::string key_str = Key(k);
349 Slice key = key_str;
350 size_t sz = GenerateValue(0 /*value_base*/, value, sizeof(value));
351 Slice v(value, sz);
352 shared->Put(cf_idx, k, 0, true /* pending */);
353
354 if (FLAGS_use_merge) {
355 if (!FLAGS_use_txn) {
356 s = db_->Merge(write_opts, cfh, key, v);
357 } else {
358 #ifndef ROCKSDB_LITE
359 Transaction* txn;
360 s = NewTxn(write_opts, &txn);
361 if (s.ok()) {
362 s = txn->Merge(cfh, key, v);
363 if (s.ok()) {
364 s = CommitTxn(txn);
365 }
366 }
367 #endif
368 }
369 } else {
370 if (!FLAGS_use_txn) {
371 s = db_->Put(write_opts, cfh, key, v);
372 } else {
373 #ifndef ROCKSDB_LITE
374 Transaction* txn;
375 s = NewTxn(write_opts, &txn);
376 if (s.ok()) {
377 s = txn->Put(cfh, key, v);
378 if (s.ok()) {
379 s = CommitTxn(txn);
380 }
381 }
382 #endif
383 }
384 }
385
386 shared->Put(cf_idx, k, 0, false /* pending */);
387 if (!s.ok()) {
388 break;
389 }
390 }
391 if (!s.ok()) {
392 break;
393 }
394 ++cf_idx;
395 }
396 if (s.ok()) {
397 s = db_->Flush(FlushOptions(), column_families_);
398 }
399 if (s.ok()) {
400 for (auto cf : column_families_) {
401 delete cf;
402 }
403 column_families_.clear();
404 delete db_;
405 db_ = nullptr;
406 #ifndef ROCKSDB_LITE
407 txn_db_ = nullptr;
408 #endif
409
410 db_preload_finished_.store(true);
411 auto now = db_stress_env->NowMicros();
412 fprintf(stdout, "%s Reopening database in read-only\n",
413 db_stress_env->TimeToString(now / 1000000).c_str());
414 // Reopen as read-only, can ignore all options related to updates
415 Open();
416 } else {
417 fprintf(stderr, "Failed to preload db");
418 exit(1);
419 }
420 }
421
SetOptions(ThreadState * thread)422 Status StressTest::SetOptions(ThreadState* thread) {
423 assert(FLAGS_set_options_one_in > 0);
424 std::unordered_map<std::string, std::string> opts;
425 std::string name =
426 options_index_[thread->rand.Next() % options_index_.size()];
427 int value_idx = thread->rand.Next() % options_table_[name].size();
428 if (name == "soft_rate_limit" || name == "hard_rate_limit") {
429 opts["soft_rate_limit"] = options_table_["soft_rate_limit"][value_idx];
430 opts["hard_rate_limit"] = options_table_["hard_rate_limit"][value_idx];
431 } else if (name == "level0_file_num_compaction_trigger" ||
432 name == "level0_slowdown_writes_trigger" ||
433 name == "level0_stop_writes_trigger") {
434 opts["level0_file_num_compaction_trigger"] =
435 options_table_["level0_file_num_compaction_trigger"][value_idx];
436 opts["level0_slowdown_writes_trigger"] =
437 options_table_["level0_slowdown_writes_trigger"][value_idx];
438 opts["level0_stop_writes_trigger"] =
439 options_table_["level0_stop_writes_trigger"][value_idx];
440 } else {
441 opts[name] = options_table_[name][value_idx];
442 }
443
444 int rand_cf_idx = thread->rand.Next() % FLAGS_column_families;
445 auto cfh = column_families_[rand_cf_idx];
446 return db_->SetOptions(cfh, opts);
447 }
448
449 #ifndef ROCKSDB_LITE
NewTxn(WriteOptions & write_opts,Transaction ** txn)450 Status StressTest::NewTxn(WriteOptions& write_opts, Transaction** txn) {
451 if (!FLAGS_use_txn) {
452 return Status::InvalidArgument("NewTxn when FLAGS_use_txn is not set");
453 }
454 static std::atomic<uint64_t> txn_id = {0};
455 TransactionOptions txn_options;
456 *txn = txn_db_->BeginTransaction(write_opts, txn_options);
457 auto istr = std::to_string(txn_id.fetch_add(1));
458 Status s = (*txn)->SetName("xid" + istr);
459 return s;
460 }
461
CommitTxn(Transaction * txn)462 Status StressTest::CommitTxn(Transaction* txn) {
463 if (!FLAGS_use_txn) {
464 return Status::InvalidArgument("CommitTxn when FLAGS_use_txn is not set");
465 }
466 Status s = txn->Prepare();
467 if (s.ok()) {
468 s = txn->Commit();
469 }
470 delete txn;
471 return s;
472 }
473
RollbackTxn(Transaction * txn)474 Status StressTest::RollbackTxn(Transaction* txn) {
475 if (!FLAGS_use_txn) {
476 return Status::InvalidArgument(
477 "RollbackTxn when FLAGS_use_txn is not"
478 " set");
479 }
480 Status s = txn->Rollback();
481 delete txn;
482 return s;
483 }
484 #endif
485
OperateDb(ThreadState * thread)486 void StressTest::OperateDb(ThreadState* thread) {
487 ReadOptions read_opts(FLAGS_verify_checksum, true);
488 WriteOptions write_opts;
489 auto shared = thread->shared;
490 char value[100];
491 std::string from_db;
492 if (FLAGS_sync) {
493 write_opts.sync = true;
494 }
495 write_opts.disableWAL = FLAGS_disable_wal;
496 const int prefixBound = static_cast<int>(FLAGS_readpercent) +
497 static_cast<int>(FLAGS_prefixpercent);
498 const int writeBound = prefixBound + static_cast<int>(FLAGS_writepercent);
499 const int delBound = writeBound + static_cast<int>(FLAGS_delpercent);
500 const int delRangeBound = delBound + static_cast<int>(FLAGS_delrangepercent);
501 const uint64_t ops_per_open = FLAGS_ops_per_thread / (FLAGS_reopen + 1);
502
503 thread->stats.Start();
504 for (int open_cnt = 0; open_cnt <= FLAGS_reopen; ++open_cnt) {
505 if (thread->shared->HasVerificationFailedYet() ||
506 thread->shared->ShouldStopTest()) {
507 break;
508 }
509 if (open_cnt != 0) {
510 thread->stats.FinishedSingleOp();
511 MutexLock l(thread->shared->GetMutex());
512 while (!thread->snapshot_queue.empty()) {
513 db_->ReleaseSnapshot(thread->snapshot_queue.front().second.snapshot);
514 delete thread->snapshot_queue.front().second.key_vec;
515 thread->snapshot_queue.pop();
516 }
517 thread->shared->IncVotedReopen();
518 if (thread->shared->AllVotedReopen()) {
519 thread->shared->GetStressTest()->Reopen(thread);
520 thread->shared->GetCondVar()->SignalAll();
521 } else {
522 thread->shared->GetCondVar()->Wait();
523 }
524 // Commenting this out as we don't want to reset stats on each open.
525 // thread->stats.Start();
526 }
527
528 for (uint64_t i = 0; i < ops_per_open; i++) {
529 if (thread->shared->HasVerificationFailedYet()) {
530 break;
531 }
532
533 // Change Options
534 if (thread->rand.OneInOpt(FLAGS_set_options_one_in)) {
535 SetOptions(thread);
536 }
537
538 if (thread->rand.OneInOpt(FLAGS_set_in_place_one_in)) {
539 options_.inplace_update_support ^= options_.inplace_update_support;
540 }
541
542 if (thread->tid == 0 && FLAGS_verify_db_one_in > 0 &&
543 thread->rand.OneIn(FLAGS_verify_db_one_in)) {
544 ContinuouslyVerifyDb(thread);
545 if (thread->shared->ShouldStopTest()) {
546 break;
547 }
548 }
549
550 MaybeClearOneColumnFamily(thread);
551
552 if (thread->rand.OneInOpt(FLAGS_sync_wal_one_in)) {
553 Status s = db_->SyncWAL();
554 if (!s.ok() && !s.IsNotSupported()) {
555 fprintf(stderr, "SyncWAL() failed: %s\n", s.ToString().c_str());
556 }
557 }
558
559 int rand_column_family = thread->rand.Next() % FLAGS_column_families;
560 ColumnFamilyHandle* column_family = column_families_[rand_column_family];
561
562 if (thread->rand.OneInOpt(FLAGS_compact_files_one_in)) {
563 TestCompactFiles(thread, column_family);
564 }
565
566 int64_t rand_key = GenerateOneKey(thread, i);
567 std::string keystr = Key(rand_key);
568 Slice key = keystr;
569 std::unique_ptr<MutexLock> lock;
570 if (ShouldAcquireMutexOnKey()) {
571 lock.reset(new MutexLock(
572 shared->GetMutexForKey(rand_column_family, rand_key)));
573 }
574
575 if (thread->rand.OneInOpt(FLAGS_compact_range_one_in)) {
576 TestCompactRange(thread, rand_key, key, column_family);
577 if (thread->shared->HasVerificationFailedYet()) {
578 break;
579 }
580 }
581
582 std::vector<int> rand_column_families =
583 GenerateColumnFamilies(FLAGS_column_families, rand_column_family);
584
585 if (thread->rand.OneInOpt(FLAGS_flush_one_in)) {
586 Status status = TestFlush(rand_column_families);
587 if (!status.ok()) {
588 fprintf(stdout, "Unable to perform Flush(): %s\n",
589 status.ToString().c_str());
590 }
591 }
592
593 #ifndef ROCKSDB_LITE
594 // Every 1 in N verify the one of the following: 1) GetLiveFiles
595 // 2) GetSortedWalFiles 3) GetCurrentWalFile. Each time, randomly select
596 // one of them to run the test.
597 if (thread->rand.OneInOpt(FLAGS_get_live_files_and_wal_files_one_in)) {
598 Status status = VerifyGetLiveAndWalFiles(thread);
599 if (!status.ok()) {
600 VerificationAbort(shared, "VerifyGetLiveAndWalFiles status not OK",
601 status);
602 }
603 }
604 #endif // !ROCKSDB_LITE
605
606 if (thread->rand.OneInOpt(FLAGS_pause_background_one_in)) {
607 Status status = TestPauseBackground(thread);
608 if (!status.ok()) {
609 VerificationAbort(
610 shared, "Pause/ContinueBackgroundWork status not OK", status);
611 }
612 }
613
614 #ifndef ROCKSDB_LITE
615 if (thread->rand.OneInOpt(FLAGS_verify_checksum_one_in)) {
616 Status status = db_->VerifyChecksum();
617 if (!status.ok()) {
618 VerificationAbort(shared, "VerifyChecksum status not OK", status);
619 }
620 }
621 #endif
622
623 std::vector<int64_t> rand_keys = GenerateKeys(rand_key);
624
625 if (thread->rand.OneInOpt(FLAGS_ingest_external_file_one_in)) {
626 TestIngestExternalFile(thread, rand_column_families, rand_keys, lock);
627 }
628
629 if (thread->rand.OneInOpt(FLAGS_backup_one_in)) {
630 Status s = TestBackupRestore(thread, rand_column_families, rand_keys);
631 if (!s.ok()) {
632 VerificationAbort(shared, "Backup/restore gave inconsistent state",
633 s);
634 }
635 }
636
637 if (thread->rand.OneInOpt(FLAGS_checkpoint_one_in)) {
638 Status s = TestCheckpoint(thread, rand_column_families, rand_keys);
639 if (!s.ok()) {
640 VerificationAbort(shared, "Checkpoint gave inconsistent state", s);
641 }
642 }
643
644 #ifndef ROCKSDB_LITE
645 if (thread->rand.OneInOpt(FLAGS_approximate_size_one_in)) {
646 Status s =
647 TestApproximateSize(thread, i, rand_column_families, rand_keys);
648 if (!s.ok()) {
649 VerificationAbort(shared, "ApproximateSize Failed", s);
650 }
651 }
652 #endif // !ROCKSDB_LITE
653 if (thread->rand.OneInOpt(FLAGS_acquire_snapshot_one_in)) {
654 TestAcquireSnapshot(thread, rand_column_family, keystr, i);
655 }
656
657 /*always*/ {
658 Status s = MaybeReleaseSnapshots(thread, i);
659 if (!s.ok()) {
660 VerificationAbort(shared, "Snapshot gave inconsistent state", s);
661 }
662 }
663
664 int prob_op = thread->rand.Uniform(100);
665 // Reset this in case we pick something other than a read op. We don't
666 // want to use a stale value when deciding at the beginning of the loop
667 // whether to vote to reopen
668 if (prob_op >= 0 && prob_op < static_cast<int>(FLAGS_readpercent)) {
669 assert(0 <= prob_op);
670 // OPERATION read
671 if (FLAGS_use_multiget) {
672 // Leave room for one more iteration of the loop with a single key
673 // batch. This is to ensure that each thread does exactly the same
674 // number of ops
675 int multiget_batch_size = static_cast<int>(
676 std::min(static_cast<uint64_t>(thread->rand.Uniform(64)),
677 FLAGS_ops_per_thread - i - 1));
678 // If its the last iteration, ensure that multiget_batch_size is 1
679 multiget_batch_size = std::max(multiget_batch_size, 1);
680 rand_keys = GenerateNKeys(thread, multiget_batch_size, i);
681 TestMultiGet(thread, read_opts, rand_column_families, rand_keys);
682 i += multiget_batch_size - 1;
683 } else {
684 TestGet(thread, read_opts, rand_column_families, rand_keys);
685 }
686 } else if (prob_op < prefixBound) {
687 assert(static_cast<int>(FLAGS_readpercent) <= prob_op);
688 // OPERATION prefix scan
689 // keys are 8 bytes long, prefix size is FLAGS_prefix_size. There are
690 // (8 - FLAGS_prefix_size) bytes besides the prefix. So there will
691 // be 2 ^ ((8 - FLAGS_prefix_size) * 8) possible keys with the same
692 // prefix
693 TestPrefixScan(thread, read_opts, rand_column_families, rand_keys);
694 } else if (prob_op < writeBound) {
695 assert(prefixBound <= prob_op);
696 // OPERATION write
697 TestPut(thread, write_opts, read_opts, rand_column_families, rand_keys,
698 value, lock);
699 } else if (prob_op < delBound) {
700 assert(writeBound <= prob_op);
701 // OPERATION delete
702 TestDelete(thread, write_opts, rand_column_families, rand_keys, lock);
703 } else if (prob_op < delRangeBound) {
704 assert(delBound <= prob_op);
705 // OPERATION delete range
706 TestDeleteRange(thread, write_opts, rand_column_families, rand_keys,
707 lock);
708 } else {
709 assert(delRangeBound <= prob_op);
710 // OPERATION iterate
711 int num_seeks = static_cast<int>(
712 std::min(static_cast<uint64_t>(thread->rand.Uniform(4)),
713 FLAGS_ops_per_thread - i - 1));
714 rand_keys = GenerateNKeys(thread, num_seeks, i);
715 i += num_seeks - 1;
716 TestIterate(thread, read_opts, rand_column_families, rand_keys);
717 }
718 thread->stats.FinishedSingleOp();
719 #ifndef ROCKSDB_LITE
720 uint32_t tid = thread->tid;
721 assert(secondaries_.empty() ||
722 static_cast<size_t>(tid) < secondaries_.size());
723 if (thread->rand.OneInOpt(FLAGS_secondary_catch_up_one_in)) {
724 Status s = secondaries_[tid]->TryCatchUpWithPrimary();
725 if (!s.ok()) {
726 VerificationAbort(shared, "Secondary instance failed to catch up", s);
727 break;
728 }
729 }
730 #endif
731 }
732 }
733 while (!thread->snapshot_queue.empty()) {
734 db_->ReleaseSnapshot(thread->snapshot_queue.front().second.snapshot);
735 delete thread->snapshot_queue.front().second.key_vec;
736 thread->snapshot_queue.pop();
737 }
738
739 thread->stats.Stop();
740 }
741
742 #ifndef ROCKSDB_LITE
743 // Generated a list of keys that close to boundaries of SST keys.
744 // If there isn't any SST file in the DB, return empty list.
GetWhiteBoxKeys(ThreadState * thread,DB * db,ColumnFamilyHandle * cfh,size_t num_keys)745 std::vector<std::string> StressTest::GetWhiteBoxKeys(ThreadState* thread,
746 DB* db,
747 ColumnFamilyHandle* cfh,
748 size_t num_keys) {
749 ColumnFamilyMetaData cfmd;
750 db->GetColumnFamilyMetaData(cfh, &cfmd);
751 std::vector<std::string> boundaries;
752 for (const LevelMetaData& lmd : cfmd.levels) {
753 for (const SstFileMetaData& sfmd : lmd.files) {
754 boundaries.push_back(sfmd.smallestkey);
755 boundaries.push_back(sfmd.largestkey);
756 }
757 }
758 if (boundaries.empty()) {
759 return {};
760 }
761
762 std::vector<std::string> ret;
763 for (size_t j = 0; j < num_keys; j++) {
764 std::string k =
765 boundaries[thread->rand.Uniform(static_cast<int>(boundaries.size()))];
766 if (thread->rand.OneIn(3)) {
767 // Reduce one byte from the string
768 for (int i = static_cast<int>(k.length()) - 1; i >= 0; i--) {
769 uint8_t cur = k[i];
770 if (cur > 0) {
771 k[i] = static_cast<char>(cur - 1);
772 break;
773 } else if (i > 0) {
774 k[i] = 0xFFu;
775 }
776 }
777 } else if (thread->rand.OneIn(2)) {
778 // Add one byte to the string
779 for (int i = static_cast<int>(k.length()) - 1; i >= 0; i--) {
780 uint8_t cur = k[i];
781 if (cur < 255) {
782 k[i] = static_cast<char>(cur + 1);
783 break;
784 } else if (i > 0) {
785 k[i] = 0x00;
786 }
787 }
788 }
789 ret.push_back(k);
790 }
791 return ret;
792 }
793 #endif // !ROCKSDB_LITE
794
795 // Given a key K, this creates an iterator which scans to K and then
796 // does a random sequence of Next/Prev operations.
TestIterate(ThreadState * thread,const ReadOptions & read_opts,const std::vector<int> & rand_column_families,const std::vector<int64_t> & rand_keys)797 Status StressTest::TestIterate(ThreadState* thread,
798 const ReadOptions& read_opts,
799 const std::vector<int>& rand_column_families,
800 const std::vector<int64_t>& rand_keys) {
801 Status s;
802 const Snapshot* snapshot = db_->GetSnapshot();
803 ReadOptions readoptionscopy = read_opts;
804 readoptionscopy.snapshot = snapshot;
805
806 bool expect_total_order = false;
807 if (thread->rand.OneIn(16)) {
808 // When prefix extractor is used, it's useful to cover total order seek.
809 readoptionscopy.total_order_seek = true;
810 expect_total_order = true;
811 } else if (thread->rand.OneIn(4)) {
812 readoptionscopy.total_order_seek = false;
813 readoptionscopy.auto_prefix_mode = true;
814 expect_total_order = true;
815 } else if (options_.prefix_extractor.get() == nullptr) {
816 expect_total_order = true;
817 }
818
819 std::string upper_bound_str;
820 Slice upper_bound;
821 if (thread->rand.OneIn(16)) {
822 // in 1/16 chance, set a iterator upper bound
823 int64_t rand_upper_key = GenerateOneKey(thread, FLAGS_ops_per_thread);
824 upper_bound_str = Key(rand_upper_key);
825 upper_bound = Slice(upper_bound_str);
826 // uppder_bound can be smaller than seek key, but the query itself
827 // should not crash either.
828 readoptionscopy.iterate_upper_bound = &upper_bound;
829 }
830 std::string lower_bound_str;
831 Slice lower_bound;
832 if (thread->rand.OneIn(16)) {
833 // in 1/16 chance, enable iterator lower bound
834 int64_t rand_lower_key = GenerateOneKey(thread, FLAGS_ops_per_thread);
835 lower_bound_str = Key(rand_lower_key);
836 lower_bound = Slice(lower_bound_str);
837 // uppder_bound can be smaller than seek key, but the query itself
838 // should not crash either.
839 readoptionscopy.iterate_lower_bound = &lower_bound;
840 }
841
842 auto cfh = column_families_[rand_column_families[0]];
843 std::unique_ptr<Iterator> iter(db_->NewIterator(readoptionscopy, cfh));
844
845 std::vector<std::string> key_str;
846 if (thread->rand.OneIn(16)) {
847 // Generate keys close to lower or upper bound of SST files.
848 key_str = GetWhiteBoxKeys(thread, db_, cfh, rand_keys.size());
849 }
850 if (key_str.empty()) {
851 // If key string is not geneerated using white block keys,
852 // Use randomized key passe in.
853 for (int64_t rkey : rand_keys) {
854 key_str.push_back(Key(rkey));
855 }
856 }
857
858 std::string op_logs;
859 const size_t kOpLogsLimit = 10000;
860
861 for (const std::string& skey : key_str) {
862 if (op_logs.size() > kOpLogsLimit) {
863 // Shouldn't take too much memory for the history log. Clear it.
864 op_logs = "(cleared...)\n";
865 }
866
867 Slice key = skey;
868
869 if (readoptionscopy.iterate_upper_bound != nullptr &&
870 thread->rand.OneIn(2)) {
871 // 1/2 chance, change the upper bound.
872 // It is possible that it is changed without first use, but there is no
873 // problem with that.
874 int64_t rand_upper_key = GenerateOneKey(thread, FLAGS_ops_per_thread);
875 upper_bound_str = Key(rand_upper_key);
876 upper_bound = Slice(upper_bound_str);
877 } else if (readoptionscopy.iterate_lower_bound != nullptr &&
878 thread->rand.OneIn(4)) {
879 // 1/4 chance, change the lower bound.
880 // It is possible that it is changed without first use, but there is no
881 // problem with that.
882 int64_t rand_lower_key = GenerateOneKey(thread, FLAGS_ops_per_thread);
883 lower_bound_str = Key(rand_lower_key);
884 lower_bound = Slice(lower_bound_str);
885 }
886
887 // Record some options to op_logs;
888 op_logs += "total_order_seek: ";
889 op_logs += (readoptionscopy.total_order_seek ? "1 " : "0 ");
890 op_logs += "auto_prefix_mode: ";
891 op_logs += (readoptionscopy.auto_prefix_mode ? "1 " : "0 ");
892 if (readoptionscopy.iterate_upper_bound != nullptr) {
893 op_logs += "ub: " + upper_bound.ToString(true) + " ";
894 }
895 if (readoptionscopy.iterate_lower_bound != nullptr) {
896 op_logs += "lb: " + lower_bound.ToString(true) + " ";
897 }
898
899 // Set up an iterator and does the same without bounds and with total
900 // order seek and compare the results. This is to identify bugs related
901 // to bounds, prefix extractor or reseeking. Sometimes we are comparing
902 // iterators with the same set-up, and it doesn't hurt to check them
903 // to be equal.
904 ReadOptions cmp_ro;
905 cmp_ro.snapshot = snapshot;
906 cmp_ro.total_order_seek = true;
907 ColumnFamilyHandle* cmp_cfh =
908 GetControlCfh(thread, rand_column_families[0]);
909 std::unique_ptr<Iterator> cmp_iter(db_->NewIterator(cmp_ro, cmp_cfh));
910 bool diverged = false;
911
912 bool support_seek_first_or_last = expect_total_order;
913
914 LastIterateOp last_op;
915 if (support_seek_first_or_last && thread->rand.OneIn(100)) {
916 iter->SeekToFirst();
917 cmp_iter->SeekToFirst();
918 last_op = kLastOpSeekToFirst;
919 op_logs += "STF ";
920 } else if (support_seek_first_or_last && thread->rand.OneIn(100)) {
921 iter->SeekToLast();
922 cmp_iter->SeekToLast();
923 last_op = kLastOpSeekToLast;
924 op_logs += "STL ";
925 } else if (thread->rand.OneIn(8)) {
926 iter->SeekForPrev(key);
927 cmp_iter->SeekForPrev(key);
928 last_op = kLastOpSeekForPrev;
929 op_logs += "SFP " + key.ToString(true) + " ";
930 } else {
931 iter->Seek(key);
932 cmp_iter->Seek(key);
933 last_op = kLastOpSeek;
934 op_logs += "S " + key.ToString(true) + " ";
935 }
936 VerifyIterator(thread, cmp_cfh, readoptionscopy, iter.get(), cmp_iter.get(),
937 last_op, key, op_logs, &diverged);
938
939 bool no_reverse =
940 (FLAGS_memtablerep == "prefix_hash" && !expect_total_order);
941 for (uint64_t i = 0; i < FLAGS_num_iterations && iter->Valid(); i++) {
942 if (no_reverse || thread->rand.OneIn(2)) {
943 iter->Next();
944 if (!diverged) {
945 assert(cmp_iter->Valid());
946 cmp_iter->Next();
947 }
948 op_logs += "N";
949 } else {
950 iter->Prev();
951 if (!diverged) {
952 assert(cmp_iter->Valid());
953 cmp_iter->Prev();
954 }
955 op_logs += "P";
956 }
957 last_op = kLastOpNextOrPrev;
958 VerifyIterator(thread, cmp_cfh, readoptionscopy, iter.get(),
959 cmp_iter.get(), last_op, key, op_logs, &diverged);
960 }
961
962 if (s.ok()) {
963 thread->stats.AddIterations(1);
964 } else {
965 fprintf(stderr, "TestIterate error: %s\n", s.ToString().c_str());
966 thread->stats.AddErrors(1);
967 break;
968 }
969
970 op_logs += "; ";
971 }
972
973 db_->ReleaseSnapshot(snapshot);
974
975 return s;
976 }
977
978 #ifndef ROCKSDB_LITE
979 // Test the return status of GetLiveFiles, GetSortedWalFiles, and
980 // GetCurrentWalFile. Each time, randomly select one of them to run
981 // and return the status.
VerifyGetLiveAndWalFiles(ThreadState * thread)982 Status StressTest::VerifyGetLiveAndWalFiles(ThreadState* thread) {
983 int case_num = thread->rand.Uniform(3);
984 if (case_num == 0) {
985 std::vector<std::string> live_file;
986 uint64_t manifest_size;
987 return db_->GetLiveFiles(live_file, &manifest_size);
988 }
989
990 if (case_num == 1) {
991 VectorLogPtr log_ptr;
992 return db_->GetSortedWalFiles(log_ptr);
993 }
994
995 if (case_num == 2) {
996 std::unique_ptr<LogFile> cur_wal_file;
997 return db_->GetCurrentWalFile(&cur_wal_file);
998 }
999 assert(false);
1000 return Status::Corruption("Undefined case happens!");
1001 }
1002 #endif // !ROCKSDB_LITE
1003
1004 // Compare the two iterator, iter and cmp_iter are in the same position,
1005 // unless iter might be made invalidate or undefined because of
1006 // upper or lower bounds, or prefix extractor.
1007 // Will flag failure if the verification fails.
1008 // diverged = true if the two iterator is already diverged.
1009 // True if verification passed, false if not.
VerifyIterator(ThreadState * thread,ColumnFamilyHandle * cmp_cfh,const ReadOptions & ro,Iterator * iter,Iterator * cmp_iter,LastIterateOp op,const Slice & seek_key,const std::string & op_logs,bool * diverged)1010 void StressTest::VerifyIterator(ThreadState* thread,
1011 ColumnFamilyHandle* cmp_cfh,
1012 const ReadOptions& ro, Iterator* iter,
1013 Iterator* cmp_iter, LastIterateOp op,
1014 const Slice& seek_key,
1015 const std::string& op_logs, bool* diverged) {
1016 if (*diverged) {
1017 return;
1018 }
1019
1020 if (op == kLastOpSeekToFirst && ro.iterate_lower_bound != nullptr) {
1021 // SeekToFirst() with lower bound is not well defined.
1022 *diverged = true;
1023 return;
1024 } else if (op == kLastOpSeekToLast && ro.iterate_upper_bound != nullptr) {
1025 // SeekToLast() with higher bound is not well defined.
1026 *diverged = true;
1027 return;
1028 } else if (op == kLastOpSeek && ro.iterate_lower_bound != nullptr &&
1029 (options_.comparator->Compare(*ro.iterate_lower_bound, seek_key) >=
1030 0 ||
1031 (ro.iterate_upper_bound != nullptr &&
1032 options_.comparator->Compare(*ro.iterate_lower_bound,
1033 *ro.iterate_upper_bound) >= 0))) {
1034 // Lower bound behavior is not well defined if it is larger than
1035 // seek key or upper bound. Disable the check for now.
1036 *diverged = true;
1037 return;
1038 } else if (op == kLastOpSeekForPrev && ro.iterate_upper_bound != nullptr &&
1039 (options_.comparator->Compare(*ro.iterate_upper_bound, seek_key) <=
1040 0 ||
1041 (ro.iterate_lower_bound != nullptr &&
1042 options_.comparator->Compare(*ro.iterate_lower_bound,
1043 *ro.iterate_upper_bound) >= 0))) {
1044 // Uppder bound behavior is not well defined if it is smaller than
1045 // seek key or lower bound. Disable the check for now.
1046 *diverged = true;
1047 return;
1048 }
1049
1050 const SliceTransform* pe = (ro.total_order_seek || ro.auto_prefix_mode)
1051 ? nullptr
1052 : options_.prefix_extractor.get();
1053 const Comparator* cmp = options_.comparator;
1054
1055 if (iter->Valid() && !cmp_iter->Valid()) {
1056 if (pe != nullptr) {
1057 if (!pe->InDomain(seek_key)) {
1058 // Prefix seek a non-in-domain key is undefined. Skip checking for
1059 // this scenario.
1060 *diverged = true;
1061 return;
1062 } else if (!pe->InDomain(iter->key())) {
1063 // out of range is iterator key is not in domain anymore.
1064 *diverged = true;
1065 return;
1066 } else if (pe->Transform(iter->key()) != pe->Transform(seek_key)) {
1067 *diverged = true;
1068 return;
1069 }
1070 }
1071 fprintf(stderr,
1072 "Control interator is invalid but iterator has key %s "
1073 "%s\n",
1074 iter->key().ToString(true).c_str(), op_logs.c_str());
1075
1076 *diverged = true;
1077 } else if (cmp_iter->Valid()) {
1078 // Iterator is not valid. It can be legimate if it has already been
1079 // out of upper or lower bound, or filtered out by prefix iterator.
1080 const Slice& total_order_key = cmp_iter->key();
1081
1082 if (pe != nullptr) {
1083 if (!pe->InDomain(seek_key)) {
1084 // Prefix seek a non-in-domain key is undefined. Skip checking for
1085 // this scenario.
1086 *diverged = true;
1087 return;
1088 }
1089
1090 if (!pe->InDomain(total_order_key) ||
1091 pe->Transform(total_order_key) != pe->Transform(seek_key)) {
1092 // If the prefix is exhausted, the only thing needs to check
1093 // is the iterator isn't return a position in prefix.
1094 // Either way, checking can stop from here.
1095 *diverged = true;
1096 if (!iter->Valid() || !pe->InDomain(iter->key()) ||
1097 pe->Transform(iter->key()) != pe->Transform(seek_key)) {
1098 return;
1099 }
1100 fprintf(stderr,
1101 "Iterator stays in prefix but contol doesn't"
1102 " iterator key %s control iterator key %s %s\n",
1103 iter->key().ToString(true).c_str(),
1104 cmp_iter->key().ToString(true).c_str(), op_logs.c_str());
1105 }
1106 }
1107 // Check upper or lower bounds.
1108 if (!*diverged) {
1109 if ((iter->Valid() && iter->key() != cmp_iter->key()) ||
1110 (!iter->Valid() &&
1111 (ro.iterate_upper_bound == nullptr ||
1112 cmp->Compare(total_order_key, *ro.iterate_upper_bound) < 0) &&
1113 (ro.iterate_lower_bound == nullptr ||
1114 cmp->Compare(total_order_key, *ro.iterate_lower_bound) > 0))) {
1115 fprintf(stderr,
1116 "Iterator diverged from control iterator which"
1117 " has value %s %s\n",
1118 total_order_key.ToString(true).c_str(), op_logs.c_str());
1119 if (iter->Valid()) {
1120 fprintf(stderr, "iterator has value %s\n",
1121 iter->key().ToString(true).c_str());
1122 } else {
1123 fprintf(stderr, "iterator is not valid\n");
1124 }
1125 *diverged = true;
1126 }
1127 }
1128 }
1129 if (*diverged) {
1130 fprintf(stderr, "Control CF %s\n", cmp_cfh->GetName().c_str());
1131 thread->stats.AddErrors(1);
1132 // Fail fast to preserve the DB state.
1133 thread->shared->SetVerificationFailure();
1134 }
1135 }
1136
1137 #ifdef ROCKSDB_LITE
TestBackupRestore(ThreadState *,const std::vector<int> &,const std::vector<int64_t> &)1138 Status StressTest::TestBackupRestore(
1139 ThreadState* /* thread */,
1140 const std::vector<int>& /* rand_column_families */,
1141 const std::vector<int64_t>& /* rand_keys */) {
1142 assert(false);
1143 fprintf(stderr,
1144 "RocksDB lite does not support "
1145 "TestBackupRestore\n");
1146 std::terminate();
1147 }
1148
TestCheckpoint(ThreadState *,const std::vector<int> &,const std::vector<int64_t> &)1149 Status StressTest::TestCheckpoint(
1150 ThreadState* /* thread */,
1151 const std::vector<int>& /* rand_column_families */,
1152 const std::vector<int64_t>& /* rand_keys */) {
1153 assert(false);
1154 fprintf(stderr,
1155 "RocksDB lite does not support "
1156 "TestCheckpoint\n");
1157 std::terminate();
1158 }
1159
TestCompactFiles(ThreadState *,ColumnFamilyHandle *)1160 void StressTest::TestCompactFiles(ThreadState* /* thread */,
1161 ColumnFamilyHandle* /* column_family */) {
1162 assert(false);
1163 fprintf(stderr,
1164 "RocksDB lite does not support "
1165 "CompactFiles\n");
1166 std::terminate();
1167 }
1168 #else // ROCKSDB_LITE
TestBackupRestore(ThreadState * thread,const std::vector<int> & rand_column_families,const std::vector<int64_t> & rand_keys)1169 Status StressTest::TestBackupRestore(
1170 ThreadState* thread, const std::vector<int>& rand_column_families,
1171 const std::vector<int64_t>& rand_keys) {
1172 // Note the column families chosen by `rand_column_families` cannot be
1173 // dropped while the locks for `rand_keys` are held. So we should not have
1174 // to worry about accessing those column families throughout this function.
1175 assert(rand_column_families.size() == rand_keys.size());
1176 std::string backup_dir = FLAGS_db + "/.backup" + ToString(thread->tid);
1177 std::string restore_dir = FLAGS_db + "/.restore" + ToString(thread->tid);
1178 BackupableDBOptions backup_opts(backup_dir);
1179 BackupEngine* backup_engine = nullptr;
1180 Status s = BackupEngine::Open(db_stress_env, backup_opts, &backup_engine);
1181 if (s.ok()) {
1182 s = backup_engine->CreateNewBackup(db_);
1183 }
1184 if (s.ok()) {
1185 delete backup_engine;
1186 backup_engine = nullptr;
1187 s = BackupEngine::Open(db_stress_env, backup_opts, &backup_engine);
1188 }
1189 if (s.ok()) {
1190 s = backup_engine->RestoreDBFromLatestBackup(restore_dir /* db_dir */,
1191 restore_dir /* wal_dir */);
1192 }
1193 if (s.ok()) {
1194 s = backup_engine->PurgeOldBackups(0 /* num_backups_to_keep */);
1195 }
1196 DB* restored_db = nullptr;
1197 std::vector<ColumnFamilyHandle*> restored_cf_handles;
1198 if (s.ok()) {
1199 Options restore_options(options_);
1200 restore_options.listeners.clear();
1201 std::vector<ColumnFamilyDescriptor> cf_descriptors;
1202 // TODO(ajkr): `column_family_names_` is not safe to access here when
1203 // `clear_column_family_one_in != 0`. But we can't easily switch to
1204 // `ListColumnFamilies` to get names because it won't necessarily give
1205 // the same order as `column_family_names_`.
1206 assert(FLAGS_clear_column_family_one_in == 0);
1207 for (auto name : column_family_names_) {
1208 cf_descriptors.emplace_back(name, ColumnFamilyOptions(restore_options));
1209 }
1210 s = DB::Open(DBOptions(restore_options), restore_dir, cf_descriptors,
1211 &restored_cf_handles, &restored_db);
1212 }
1213 // for simplicity, currently only verifies existence/non-existence of a few
1214 // keys
1215 for (size_t i = 0; s.ok() && i < rand_column_families.size(); ++i) {
1216 std::string key_str = Key(rand_keys[i]);
1217 Slice key = key_str;
1218 std::string restored_value;
1219 Status get_status = restored_db->Get(
1220 ReadOptions(), restored_cf_handles[rand_column_families[i]], key,
1221 &restored_value);
1222 bool exists = thread->shared->Exists(rand_column_families[i], rand_keys[i]);
1223 if (get_status.ok()) {
1224 if (!exists) {
1225 s = Status::Corruption("key exists in restore but not in original db");
1226 }
1227 } else if (get_status.IsNotFound()) {
1228 if (exists) {
1229 s = Status::Corruption("key exists in original db but not in restore");
1230 }
1231 } else {
1232 s = get_status;
1233 }
1234 }
1235 if (backup_engine != nullptr) {
1236 delete backup_engine;
1237 backup_engine = nullptr;
1238 }
1239 if (restored_db != nullptr) {
1240 for (auto* cf_handle : restored_cf_handles) {
1241 restored_db->DestroyColumnFamilyHandle(cf_handle);
1242 }
1243 delete restored_db;
1244 restored_db = nullptr;
1245 }
1246 if (!s.ok()) {
1247 fprintf(stderr, "A backup/restore operation failed with: %s\n",
1248 s.ToString().c_str());
1249 }
1250 return s;
1251 }
1252
1253 #ifndef ROCKSDB_LITE
TestApproximateSize(ThreadState * thread,uint64_t iteration,const std::vector<int> & rand_column_families,const std::vector<int64_t> & rand_keys)1254 Status StressTest::TestApproximateSize(
1255 ThreadState* thread, uint64_t iteration,
1256 const std::vector<int>& rand_column_families,
1257 const std::vector<int64_t>& rand_keys) {
1258 // rand_keys likely only has one key. Just use the first one.
1259 assert(!rand_keys.empty());
1260 assert(!rand_column_families.empty());
1261 int64_t key1 = rand_keys[0];
1262 int64_t key2;
1263 if (thread->rand.OneIn(2)) {
1264 // Two totally random keys. This tends to cover large ranges.
1265 key2 = GenerateOneKey(thread, iteration);
1266 if (key2 < key1) {
1267 std::swap(key1, key2);
1268 }
1269 } else {
1270 // Unless users pass a very large FLAGS_max_key, it we should not worry
1271 // about overflow. It is for testing, so we skip the overflow checking
1272 // for simplicity.
1273 key2 = key1 + static_cast<int64_t>(thread->rand.Uniform(1000));
1274 }
1275 std::string key1_str = Key(key1);
1276 std::string key2_str = Key(key2);
1277 Range range{Slice(key1_str), Slice(key2_str)};
1278 SizeApproximationOptions sao;
1279 sao.include_memtabtles = thread->rand.OneIn(2);
1280 if (sao.include_memtabtles) {
1281 sao.include_files = thread->rand.OneIn(2);
1282 }
1283 if (thread->rand.OneIn(2)) {
1284 if (thread->rand.OneIn(2)) {
1285 sao.files_size_error_margin = 0.0;
1286 } else {
1287 sao.files_size_error_margin =
1288 static_cast<double>(thread->rand.Uniform(3));
1289 }
1290 }
1291 uint64_t result;
1292 return db_->GetApproximateSizes(
1293 sao, column_families_[rand_column_families[0]], &range, 1, &result);
1294 }
1295 #endif // ROCKSDB_LITE
1296
TestCheckpoint(ThreadState * thread,const std::vector<int> & rand_column_families,const std::vector<int64_t> & rand_keys)1297 Status StressTest::TestCheckpoint(ThreadState* thread,
1298 const std::vector<int>& rand_column_families,
1299 const std::vector<int64_t>& rand_keys) {
1300 // Note the column families chosen by `rand_column_families` cannot be
1301 // dropped while the locks for `rand_keys` are held. So we should not have
1302 // to worry about accessing those column families throughout this function.
1303 assert(rand_column_families.size() == rand_keys.size());
1304 std::string checkpoint_dir =
1305 FLAGS_db + "/.checkpoint" + ToString(thread->tid);
1306 Options tmp_opts(options_);
1307 tmp_opts.listeners.clear();
1308 tmp_opts.env = db_stress_env->target();
1309
1310 DestroyDB(checkpoint_dir, tmp_opts);
1311
1312 Checkpoint* checkpoint = nullptr;
1313 Status s = Checkpoint::Create(db_, &checkpoint);
1314 if (s.ok()) {
1315 s = checkpoint->CreateCheckpoint(checkpoint_dir);
1316 }
1317 std::vector<ColumnFamilyHandle*> cf_handles;
1318 DB* checkpoint_db = nullptr;
1319 if (s.ok()) {
1320 delete checkpoint;
1321 checkpoint = nullptr;
1322 Options options(options_);
1323 options.listeners.clear();
1324 std::vector<ColumnFamilyDescriptor> cf_descs;
1325 // TODO(ajkr): `column_family_names_` is not safe to access here when
1326 // `clear_column_family_one_in != 0`. But we can't easily switch to
1327 // `ListColumnFamilies` to get names because it won't necessarily give
1328 // the same order as `column_family_names_`.
1329 if (FLAGS_clear_column_family_one_in == 0) {
1330 for (const auto& name : column_family_names_) {
1331 cf_descs.emplace_back(name, ColumnFamilyOptions(options));
1332 }
1333 s = DB::OpenForReadOnly(DBOptions(options), checkpoint_dir, cf_descs,
1334 &cf_handles, &checkpoint_db);
1335 }
1336 }
1337 if (checkpoint_db != nullptr) {
1338 for (size_t i = 0; s.ok() && i < rand_column_families.size(); ++i) {
1339 std::string key_str = Key(rand_keys[i]);
1340 Slice key = key_str;
1341 std::string value;
1342 Status get_status = checkpoint_db->Get(
1343 ReadOptions(), cf_handles[rand_column_families[i]], key, &value);
1344 bool exists =
1345 thread->shared->Exists(rand_column_families[i], rand_keys[i]);
1346 if (get_status.ok()) {
1347 if (!exists) {
1348 s = Status::Corruption(
1349 "key exists in checkpoint but not in original db");
1350 }
1351 } else if (get_status.IsNotFound()) {
1352 if (exists) {
1353 s = Status::Corruption(
1354 "key exists in original db but not in checkpoint");
1355 }
1356 } else {
1357 s = get_status;
1358 }
1359 }
1360 for (auto cfh : cf_handles) {
1361 delete cfh;
1362 }
1363 cf_handles.clear();
1364 delete checkpoint_db;
1365 checkpoint_db = nullptr;
1366 }
1367
1368 DestroyDB(checkpoint_dir, tmp_opts);
1369
1370 if (!s.ok()) {
1371 fprintf(stderr, "A checkpoint operation failed with: %s\n",
1372 s.ToString().c_str());
1373 }
1374 return s;
1375 }
1376
TestCompactFiles(ThreadState * thread,ColumnFamilyHandle * column_family)1377 void StressTest::TestCompactFiles(ThreadState* thread,
1378 ColumnFamilyHandle* column_family) {
1379 ROCKSDB_NAMESPACE::ColumnFamilyMetaData cf_meta_data;
1380 db_->GetColumnFamilyMetaData(column_family, &cf_meta_data);
1381
1382 // Randomly compact up to three consecutive files from a level
1383 const int kMaxRetry = 3;
1384 for (int attempt = 0; attempt < kMaxRetry; ++attempt) {
1385 size_t random_level =
1386 thread->rand.Uniform(static_cast<int>(cf_meta_data.levels.size()));
1387
1388 const auto& files = cf_meta_data.levels[random_level].files;
1389 if (files.size() > 0) {
1390 size_t random_file_index =
1391 thread->rand.Uniform(static_cast<int>(files.size()));
1392 if (files[random_file_index].being_compacted) {
1393 // Retry as the selected file is currently being compacted
1394 continue;
1395 }
1396
1397 std::vector<std::string> input_files;
1398 input_files.push_back(files[random_file_index].name);
1399 if (random_file_index > 0 &&
1400 !files[random_file_index - 1].being_compacted) {
1401 input_files.push_back(files[random_file_index - 1].name);
1402 }
1403 if (random_file_index + 1 < files.size() &&
1404 !files[random_file_index + 1].being_compacted) {
1405 input_files.push_back(files[random_file_index + 1].name);
1406 }
1407
1408 size_t output_level =
1409 std::min(random_level + 1, cf_meta_data.levels.size() - 1);
1410 auto s = db_->CompactFiles(CompactionOptions(), column_family,
1411 input_files, static_cast<int>(output_level));
1412 if (!s.ok()) {
1413 fprintf(stdout, "Unable to perform CompactFiles(): %s\n",
1414 s.ToString().c_str());
1415 thread->stats.AddNumCompactFilesFailed(1);
1416 } else {
1417 thread->stats.AddNumCompactFilesSucceed(1);
1418 }
1419 break;
1420 }
1421 }
1422 }
1423 #endif // ROCKSDB_LITE
1424
TestFlush(const std::vector<int> & rand_column_families)1425 Status StressTest::TestFlush(const std::vector<int>& rand_column_families) {
1426 FlushOptions flush_opts;
1427 std::vector<ColumnFamilyHandle*> cfhs;
1428 std::for_each(rand_column_families.begin(), rand_column_families.end(),
1429 [this, &cfhs](int k) { cfhs.push_back(column_families_[k]); });
1430 return db_->Flush(flush_opts, cfhs);
1431 }
1432
TestPauseBackground(ThreadState * thread)1433 Status StressTest::TestPauseBackground(ThreadState* thread) {
1434 Status status = db_->PauseBackgroundWork();
1435 if (!status.ok()) {
1436 return status;
1437 }
1438 // To avoid stalling/deadlocking ourself in this thread, just
1439 // sleep here during pause and let other threads do db operations.
1440 // Sleep up to ~16 seconds (2**24 microseconds), but very skewed
1441 // toward short pause. (1 chance in 25 of pausing >= 1s;
1442 // 1 chance in 625 of pausing full 16s.)
1443 int pwr2_micros =
1444 std::min(thread->rand.Uniform(25), thread->rand.Uniform(25));
1445 db_stress_env->SleepForMicroseconds(1 << pwr2_micros);
1446 return db_->ContinueBackgroundWork();
1447 }
1448
TestAcquireSnapshot(ThreadState * thread,int rand_column_family,const std::string & keystr,uint64_t i)1449 void StressTest::TestAcquireSnapshot(ThreadState* thread,
1450 int rand_column_family,
1451 const std::string& keystr, uint64_t i) {
1452 Slice key = keystr;
1453 ColumnFamilyHandle* column_family = column_families_[rand_column_family];
1454 #ifndef ROCKSDB_LITE
1455 auto db_impl = reinterpret_cast<DBImpl*>(db_->GetRootDB());
1456 const bool ww_snapshot = thread->rand.OneIn(10);
1457 const Snapshot* snapshot =
1458 ww_snapshot ? db_impl->GetSnapshotForWriteConflictBoundary()
1459 : db_->GetSnapshot();
1460 #else
1461 const Snapshot* snapshot = db_->GetSnapshot();
1462 #endif // !ROCKSDB_LITE
1463 ReadOptions ropt;
1464 ropt.snapshot = snapshot;
1465 std::string value_at;
1466 // When taking a snapshot, we also read a key from that snapshot. We
1467 // will later read the same key before releasing the snapshot and
1468 // verify that the results are the same.
1469 auto status_at = db_->Get(ropt, column_family, key, &value_at);
1470 std::vector<bool>* key_vec = nullptr;
1471
1472 if (FLAGS_compare_full_db_state_snapshot && (thread->tid == 0)) {
1473 key_vec = new std::vector<bool>(FLAGS_max_key);
1474 // When `prefix_extractor` is set, seeking to beginning and scanning
1475 // across prefixes are only supported with `total_order_seek` set.
1476 ropt.total_order_seek = true;
1477 std::unique_ptr<Iterator> iterator(db_->NewIterator(ropt));
1478 for (iterator->SeekToFirst(); iterator->Valid(); iterator->Next()) {
1479 uint64_t key_val;
1480 if (GetIntVal(iterator->key().ToString(), &key_val)) {
1481 (*key_vec)[key_val] = true;
1482 }
1483 }
1484 }
1485
1486 ThreadState::SnapshotState snap_state = {
1487 snapshot, rand_column_family, column_family->GetName(),
1488 keystr, status_at, value_at,
1489 key_vec};
1490 uint64_t hold_for = FLAGS_snapshot_hold_ops;
1491 if (FLAGS_long_running_snapshots) {
1492 // Hold 10% of snapshots for 10x more
1493 if (thread->rand.OneIn(10)) {
1494 assert(hold_for < port::kMaxInt64 / 10);
1495 hold_for *= 10;
1496 // Hold 1% of snapshots for 100x more
1497 if (thread->rand.OneIn(10)) {
1498 assert(hold_for < port::kMaxInt64 / 10);
1499 hold_for *= 10;
1500 }
1501 }
1502 }
1503 uint64_t release_at = std::min(FLAGS_ops_per_thread - 1, i + hold_for);
1504 thread->snapshot_queue.emplace(release_at, snap_state);
1505 }
1506
MaybeReleaseSnapshots(ThreadState * thread,uint64_t i)1507 Status StressTest::MaybeReleaseSnapshots(ThreadState* thread, uint64_t i) {
1508 while (!thread->snapshot_queue.empty() &&
1509 i >= thread->snapshot_queue.front().first) {
1510 auto snap_state = thread->snapshot_queue.front().second;
1511 assert(snap_state.snapshot);
1512 // Note: this is unsafe as the cf might be dropped concurrently. But
1513 // it is ok since unclean cf drop is cunnrently not supported by write
1514 // prepared transactions.
1515 Status s = AssertSame(db_, column_families_[snap_state.cf_at], snap_state);
1516 db_->ReleaseSnapshot(snap_state.snapshot);
1517 delete snap_state.key_vec;
1518 thread->snapshot_queue.pop();
1519 if (!s.ok()) {
1520 return s;
1521 }
1522 }
1523 return Status::OK();
1524 }
1525
TestCompactRange(ThreadState * thread,int64_t rand_key,const Slice & start_key,ColumnFamilyHandle * column_family)1526 void StressTest::TestCompactRange(ThreadState* thread, int64_t rand_key,
1527 const Slice& start_key,
1528 ColumnFamilyHandle* column_family) {
1529 int64_t end_key_num;
1530 if (port::kMaxInt64 - rand_key < FLAGS_compact_range_width) {
1531 end_key_num = port::kMaxInt64;
1532 } else {
1533 end_key_num = FLAGS_compact_range_width + rand_key;
1534 }
1535 std::string end_key_buf = Key(end_key_num);
1536 Slice end_key(end_key_buf);
1537
1538 CompactRangeOptions cro;
1539 cro.exclusive_manual_compaction = static_cast<bool>(thread->rand.Next() % 2);
1540 cro.change_level = static_cast<bool>(thread->rand.Next() % 2);
1541 std::vector<BottommostLevelCompaction> bottom_level_styles = {
1542 BottommostLevelCompaction::kSkip,
1543 BottommostLevelCompaction::kIfHaveCompactionFilter,
1544 BottommostLevelCompaction::kForce,
1545 BottommostLevelCompaction::kForceOptimized};
1546 cro.bottommost_level_compaction =
1547 bottom_level_styles[thread->rand.Next() %
1548 static_cast<uint32_t>(bottom_level_styles.size())];
1549 cro.allow_write_stall = static_cast<bool>(thread->rand.Next() % 2);
1550 cro.max_subcompactions = static_cast<uint32_t>(thread->rand.Next() % 4);
1551
1552 const Snapshot* pre_snapshot = nullptr;
1553 uint32_t pre_hash = 0;
1554 if (thread->rand.OneIn(2)) {
1555 // Do some validation by declaring a snapshot and compare the data before
1556 // and after the compaction
1557 pre_snapshot = db_->GetSnapshot();
1558 pre_hash =
1559 GetRangeHash(thread, pre_snapshot, column_family, start_key, end_key);
1560 }
1561
1562 Status status = db_->CompactRange(cro, column_family, &start_key, &end_key);
1563
1564 if (!status.ok()) {
1565 fprintf(stdout, "Unable to perform CompactRange(): %s\n",
1566 status.ToString().c_str());
1567 }
1568
1569 if (pre_snapshot != nullptr) {
1570 uint32_t post_hash =
1571 GetRangeHash(thread, pre_snapshot, column_family, start_key, end_key);
1572 if (pre_hash != post_hash) {
1573 fprintf(stderr,
1574 "Data hash different before and after compact range "
1575 "start_key %s end_key %s\n",
1576 start_key.ToString(true).c_str(), end_key.ToString(true).c_str());
1577 thread->stats.AddErrors(1);
1578 // Fail fast to preserve the DB state.
1579 thread->shared->SetVerificationFailure();
1580 }
1581 db_->ReleaseSnapshot(pre_snapshot);
1582 }
1583 }
1584
GetRangeHash(ThreadState * thread,const Snapshot * snapshot,ColumnFamilyHandle * column_family,const Slice & start_key,const Slice & end_key)1585 uint32_t StressTest::GetRangeHash(ThreadState* thread, const Snapshot* snapshot,
1586 ColumnFamilyHandle* column_family,
1587 const Slice& start_key,
1588 const Slice& end_key) {
1589 const std::string kCrcCalculatorSepearator = ";";
1590 uint32_t crc = 0;
1591 ReadOptions ro;
1592 ro.snapshot = snapshot;
1593 ro.total_order_seek = true;
1594 std::unique_ptr<Iterator> it(db_->NewIterator(ro, column_family));
1595 for (it->Seek(start_key);
1596 it->Valid() && options_.comparator->Compare(it->key(), end_key) <= 0;
1597 it->Next()) {
1598 crc = crc32c::Extend(crc, it->key().data(), it->key().size());
1599 crc = crc32c::Extend(crc, kCrcCalculatorSepearator.data(), 1);
1600 crc = crc32c::Extend(crc, it->value().data(), it->value().size());
1601 crc = crc32c::Extend(crc, kCrcCalculatorSepearator.data(), 1);
1602 }
1603 if (!it->status().ok()) {
1604 fprintf(stderr, "Iterator non-OK when calculating range CRC: %s\n",
1605 it->status().ToString().c_str());
1606 thread->stats.AddErrors(1);
1607 // Fail fast to preserve the DB state.
1608 thread->shared->SetVerificationFailure();
1609 }
1610 return crc;
1611 }
1612
PrintEnv() const1613 void StressTest::PrintEnv() const {
1614 fprintf(stdout, "RocksDB version : %d.%d\n", kMajorVersion,
1615 kMinorVersion);
1616 fprintf(stdout, "Format version : %d\n", FLAGS_format_version);
1617 fprintf(stdout, "TransactionDB : %s\n",
1618 FLAGS_use_txn ? "true" : "false");
1619 #ifndef ROCKSDB_LITE
1620 fprintf(stdout, "BlobDB : %s\n",
1621 FLAGS_use_blob_db ? "true" : "false");
1622 #endif // !ROCKSDB_LITE
1623 fprintf(stdout, "Read only mode : %s\n",
1624 FLAGS_read_only ? "true" : "false");
1625 fprintf(stdout, "Atomic flush : %s\n",
1626 FLAGS_atomic_flush ? "true" : "false");
1627 fprintf(stdout, "Column families : %d\n", FLAGS_column_families);
1628 if (!FLAGS_test_batches_snapshots) {
1629 fprintf(stdout, "Clear CFs one in : %d\n",
1630 FLAGS_clear_column_family_one_in);
1631 }
1632 fprintf(stdout, "Number of threads : %d\n", FLAGS_threads);
1633 fprintf(stdout, "Ops per thread : %lu\n",
1634 (unsigned long)FLAGS_ops_per_thread);
1635 std::string ttl_state("unused");
1636 if (FLAGS_ttl > 0) {
1637 ttl_state = NumberToString(FLAGS_ttl);
1638 }
1639 fprintf(stdout, "Time to live(sec) : %s\n", ttl_state.c_str());
1640 fprintf(stdout, "Read percentage : %d%%\n", FLAGS_readpercent);
1641 fprintf(stdout, "Prefix percentage : %d%%\n", FLAGS_prefixpercent);
1642 fprintf(stdout, "Write percentage : %d%%\n", FLAGS_writepercent);
1643 fprintf(stdout, "Delete percentage : %d%%\n", FLAGS_delpercent);
1644 fprintf(stdout, "Delete range percentage : %d%%\n", FLAGS_delrangepercent);
1645 fprintf(stdout, "No overwrite percentage : %d%%\n",
1646 FLAGS_nooverwritepercent);
1647 fprintf(stdout, "Iterate percentage : %d%%\n", FLAGS_iterpercent);
1648 fprintf(stdout, "DB-write-buffer-size : %" PRIu64 "\n",
1649 FLAGS_db_write_buffer_size);
1650 fprintf(stdout, "Write-buffer-size : %d\n", FLAGS_write_buffer_size);
1651 fprintf(stdout, "Iterations : %lu\n",
1652 (unsigned long)FLAGS_num_iterations);
1653 fprintf(stdout, "Max key : %lu\n",
1654 (unsigned long)FLAGS_max_key);
1655 fprintf(stdout, "Ratio #ops/#keys : %f\n",
1656 (1.0 * FLAGS_ops_per_thread * FLAGS_threads) / FLAGS_max_key);
1657 fprintf(stdout, "Num times DB reopens : %d\n", FLAGS_reopen);
1658 fprintf(stdout, "Batches/snapshots : %d\n",
1659 FLAGS_test_batches_snapshots);
1660 fprintf(stdout, "Do update in place : %d\n", FLAGS_in_place_update);
1661 fprintf(stdout, "Num keys per lock : %d\n",
1662 1 << FLAGS_log2_keys_per_lock);
1663 std::string compression = CompressionTypeToString(compression_type_e);
1664 fprintf(stdout, "Compression : %s\n", compression.c_str());
1665 std::string bottommost_compression =
1666 CompressionTypeToString(bottommost_compression_type_e);
1667 fprintf(stdout, "Bottommost Compression : %s\n",
1668 bottommost_compression.c_str());
1669 std::string checksum = ChecksumTypeToString(checksum_type_e);
1670 fprintf(stdout, "Checksum type : %s\n", checksum.c_str());
1671 fprintf(stdout, "Bloom bits / key : %s\n",
1672 FormatDoubleParam(FLAGS_bloom_bits).c_str());
1673 fprintf(stdout, "Max subcompactions : %" PRIu64 "\n",
1674 FLAGS_subcompactions);
1675 fprintf(stdout, "Use MultiGet : %s\n",
1676 FLAGS_use_multiget ? "true" : "false");
1677
1678 const char* memtablerep = "";
1679 switch (FLAGS_rep_factory) {
1680 case kSkipList:
1681 memtablerep = "skip_list";
1682 break;
1683 case kHashSkipList:
1684 memtablerep = "prefix_hash";
1685 break;
1686 case kVectorRep:
1687 memtablerep = "vector";
1688 break;
1689 }
1690
1691 fprintf(stdout, "Memtablerep : %s\n", memtablerep);
1692
1693 fprintf(stdout, "Test kill odd : %d\n", rocksdb_kill_odds);
1694 if (!rocksdb_kill_prefix_blacklist.empty()) {
1695 fprintf(stdout, "Skipping kill points prefixes:\n");
1696 for (auto& p : rocksdb_kill_prefix_blacklist) {
1697 fprintf(stdout, " %s\n", p.c_str());
1698 }
1699 }
1700 fprintf(stdout, "Periodic Compaction Secs : %" PRIu64 "\n",
1701 FLAGS_periodic_compaction_seconds);
1702 fprintf(stdout, "Compaction TTL : %" PRIu64 "\n",
1703 FLAGS_compaction_ttl);
1704 fprintf(stdout, "Background Purge : %d\n",
1705 static_cast<int>(FLAGS_avoid_unnecessary_blocking_io));
1706 fprintf(stdout, "Write DB ID to manifest : %d\n",
1707 static_cast<int>(FLAGS_write_dbid_to_manifest));
1708 fprintf(stdout, "Max Write Batch Group Size: %" PRIu64 "\n",
1709 FLAGS_max_write_batch_group_size_bytes);
1710 fprintf(stdout, "Use dynamic level : %d\n",
1711 static_cast<int>(FLAGS_level_compaction_dynamic_level_bytes));
1712
1713 fprintf(stdout, "------------------------------------------------\n");
1714 }
1715
Open()1716 void StressTest::Open() {
1717 assert(db_ == nullptr);
1718 #ifndef ROCKSDB_LITE
1719 assert(txn_db_ == nullptr);
1720 #endif
1721 if (FLAGS_options_file.empty()) {
1722 BlockBasedTableOptions block_based_options;
1723 block_based_options.block_cache = cache_;
1724 block_based_options.cache_index_and_filter_blocks =
1725 FLAGS_cache_index_and_filter_blocks;
1726 block_based_options.block_cache_compressed = compressed_cache_;
1727 block_based_options.checksum = checksum_type_e;
1728 block_based_options.block_size = FLAGS_block_size;
1729 block_based_options.format_version =
1730 static_cast<uint32_t>(FLAGS_format_version);
1731 block_based_options.index_block_restart_interval =
1732 static_cast<int32_t>(FLAGS_index_block_restart_interval);
1733 block_based_options.filter_policy = filter_policy_;
1734 block_based_options.partition_filters = FLAGS_partition_filters;
1735 block_based_options.index_type =
1736 static_cast<BlockBasedTableOptions::IndexType>(FLAGS_index_type);
1737 options_.table_factory.reset(
1738 NewBlockBasedTableFactory(block_based_options));
1739 options_.db_write_buffer_size = FLAGS_db_write_buffer_size;
1740 options_.write_buffer_size = FLAGS_write_buffer_size;
1741 options_.max_write_buffer_number = FLAGS_max_write_buffer_number;
1742 options_.min_write_buffer_number_to_merge =
1743 FLAGS_min_write_buffer_number_to_merge;
1744 options_.max_write_buffer_number_to_maintain =
1745 FLAGS_max_write_buffer_number_to_maintain;
1746 options_.max_write_buffer_size_to_maintain =
1747 FLAGS_max_write_buffer_size_to_maintain;
1748 options_.memtable_prefix_bloom_size_ratio =
1749 FLAGS_memtable_prefix_bloom_size_ratio;
1750 options_.memtable_whole_key_filtering = FLAGS_memtable_whole_key_filtering;
1751 options_.max_background_compactions = FLAGS_max_background_compactions;
1752 options_.max_background_flushes = FLAGS_max_background_flushes;
1753 options_.compaction_style =
1754 static_cast<ROCKSDB_NAMESPACE::CompactionStyle>(FLAGS_compaction_style);
1755 if (FLAGS_prefix_size >= 0) {
1756 options_.prefix_extractor.reset(
1757 NewFixedPrefixTransform(FLAGS_prefix_size));
1758 }
1759 options_.max_open_files = FLAGS_open_files;
1760 options_.statistics = dbstats;
1761 options_.env = db_stress_env;
1762 options_.use_fsync = FLAGS_use_fsync;
1763 options_.compaction_readahead_size = FLAGS_compaction_readahead_size;
1764 options_.allow_mmap_reads = FLAGS_mmap_read;
1765 options_.allow_mmap_writes = FLAGS_mmap_write;
1766 options_.use_direct_reads = FLAGS_use_direct_reads;
1767 options_.use_direct_io_for_flush_and_compaction =
1768 FLAGS_use_direct_io_for_flush_and_compaction;
1769 options_.recycle_log_file_num =
1770 static_cast<size_t>(FLAGS_recycle_log_file_num);
1771 options_.target_file_size_base = FLAGS_target_file_size_base;
1772 options_.target_file_size_multiplier = FLAGS_target_file_size_multiplier;
1773 options_.max_bytes_for_level_base = FLAGS_max_bytes_for_level_base;
1774 options_.max_bytes_for_level_multiplier =
1775 FLAGS_max_bytes_for_level_multiplier;
1776 options_.level0_stop_writes_trigger = FLAGS_level0_stop_writes_trigger;
1777 options_.level0_slowdown_writes_trigger =
1778 FLAGS_level0_slowdown_writes_trigger;
1779 options_.level0_file_num_compaction_trigger =
1780 FLAGS_level0_file_num_compaction_trigger;
1781 options_.compression = compression_type_e;
1782 options_.bottommost_compression = bottommost_compression_type_e;
1783 options_.compression_opts.max_dict_bytes = FLAGS_compression_max_dict_bytes;
1784 options_.compression_opts.zstd_max_train_bytes =
1785 FLAGS_compression_zstd_max_train_bytes;
1786 options_.create_if_missing = true;
1787 options_.max_manifest_file_size = FLAGS_max_manifest_file_size;
1788 options_.inplace_update_support = FLAGS_in_place_update;
1789 options_.max_subcompactions = static_cast<uint32_t>(FLAGS_subcompactions);
1790 options_.allow_concurrent_memtable_write =
1791 FLAGS_allow_concurrent_memtable_write;
1792 options_.periodic_compaction_seconds = FLAGS_periodic_compaction_seconds;
1793 options_.ttl = FLAGS_compaction_ttl;
1794 options_.enable_pipelined_write = FLAGS_enable_pipelined_write;
1795 options_.enable_write_thread_adaptive_yield =
1796 FLAGS_enable_write_thread_adaptive_yield;
1797 options_.compaction_options_universal.size_ratio =
1798 FLAGS_universal_size_ratio;
1799 options_.compaction_options_universal.min_merge_width =
1800 FLAGS_universal_min_merge_width;
1801 options_.compaction_options_universal.max_merge_width =
1802 FLAGS_universal_max_merge_width;
1803 options_.compaction_options_universal.max_size_amplification_percent =
1804 FLAGS_universal_max_size_amplification_percent;
1805 options_.atomic_flush = FLAGS_atomic_flush;
1806 options_.avoid_unnecessary_blocking_io =
1807 FLAGS_avoid_unnecessary_blocking_io;
1808 options_.write_dbid_to_manifest = FLAGS_write_dbid_to_manifest;
1809 options_.max_write_batch_group_size_bytes =
1810 FLAGS_max_write_batch_group_size_bytes;
1811 options_.level_compaction_dynamic_level_bytes =
1812 FLAGS_level_compaction_dynamic_level_bytes;
1813 } else {
1814 #ifdef ROCKSDB_LITE
1815 fprintf(stderr, "--options_file not supported in lite mode\n");
1816 exit(1);
1817 #else
1818 DBOptions db_options;
1819 std::vector<ColumnFamilyDescriptor> cf_descriptors;
1820 Status s = LoadOptionsFromFile(FLAGS_options_file, db_stress_env,
1821 &db_options, &cf_descriptors);
1822 db_options.env = new DbStressEnvWrapper(db_stress_env);
1823 if (!s.ok()) {
1824 fprintf(stderr, "Unable to load options file %s --- %s\n",
1825 FLAGS_options_file.c_str(), s.ToString().c_str());
1826 exit(1);
1827 }
1828 options_ = Options(db_options, cf_descriptors[0].options);
1829 #endif // ROCKSDB_LITE
1830 }
1831
1832 if (FLAGS_rate_limiter_bytes_per_sec > 0) {
1833 options_.rate_limiter.reset(NewGenericRateLimiter(
1834 FLAGS_rate_limiter_bytes_per_sec, 1000 /* refill_period_us */,
1835 10 /* fairness */,
1836 FLAGS_rate_limit_bg_reads ? RateLimiter::Mode::kReadsOnly
1837 : RateLimiter::Mode::kWritesOnly));
1838 if (FLAGS_rate_limit_bg_reads) {
1839 options_.new_table_reader_for_compaction_inputs = true;
1840 }
1841 }
1842
1843 if (FLAGS_prefix_size == 0 && FLAGS_rep_factory == kHashSkipList) {
1844 fprintf(stderr,
1845 "prefeix_size cannot be zero if memtablerep == prefix_hash\n");
1846 exit(1);
1847 }
1848 if (FLAGS_prefix_size != 0 && FLAGS_rep_factory != kHashSkipList) {
1849 fprintf(stderr,
1850 "WARNING: prefix_size is non-zero but "
1851 "memtablerep != prefix_hash\n");
1852 }
1853 switch (FLAGS_rep_factory) {
1854 case kSkipList:
1855 // no need to do anything
1856 break;
1857 #ifndef ROCKSDB_LITE
1858 case kHashSkipList:
1859 options_.memtable_factory.reset(NewHashSkipListRepFactory(10000));
1860 break;
1861 case kVectorRep:
1862 options_.memtable_factory.reset(new VectorRepFactory());
1863 break;
1864 #else
1865 default:
1866 fprintf(stderr,
1867 "RocksdbLite only supports skip list mem table. Skip "
1868 "--rep_factory\n");
1869 #endif // ROCKSDB_LITE
1870 }
1871
1872 if (FLAGS_use_full_merge_v1) {
1873 options_.merge_operator = MergeOperators::CreateDeprecatedPutOperator();
1874 } else {
1875 options_.merge_operator = MergeOperators::CreatePutOperator();
1876 }
1877
1878 fprintf(stdout, "DB path: [%s]\n", FLAGS_db.c_str());
1879
1880 Status s;
1881 if (FLAGS_ttl == -1) {
1882 std::vector<std::string> existing_column_families;
1883 s = DB::ListColumnFamilies(DBOptions(options_), FLAGS_db,
1884 &existing_column_families); // ignore errors
1885 if (!s.ok()) {
1886 // DB doesn't exist
1887 assert(existing_column_families.empty());
1888 assert(column_family_names_.empty());
1889 column_family_names_.push_back(kDefaultColumnFamilyName);
1890 } else if (column_family_names_.empty()) {
1891 // this is the first call to the function Open()
1892 column_family_names_ = existing_column_families;
1893 } else {
1894 // this is a reopen. just assert that existing column_family_names are
1895 // equivalent to what we remember
1896 auto sorted_cfn = column_family_names_;
1897 std::sort(sorted_cfn.begin(), sorted_cfn.end());
1898 std::sort(existing_column_families.begin(),
1899 existing_column_families.end());
1900 if (sorted_cfn != existing_column_families) {
1901 fprintf(stderr, "Expected column families differ from the existing:\n");
1902 fprintf(stderr, "Expected: {");
1903 for (auto cf : sorted_cfn) {
1904 fprintf(stderr, "%s ", cf.c_str());
1905 }
1906 fprintf(stderr, "}\n");
1907 fprintf(stderr, "Existing: {");
1908 for (auto cf : existing_column_families) {
1909 fprintf(stderr, "%s ", cf.c_str());
1910 }
1911 fprintf(stderr, "}\n");
1912 }
1913 assert(sorted_cfn == existing_column_families);
1914 }
1915 std::vector<ColumnFamilyDescriptor> cf_descriptors;
1916 for (auto name : column_family_names_) {
1917 if (name != kDefaultColumnFamilyName) {
1918 new_column_family_name_ =
1919 std::max(new_column_family_name_.load(), std::stoi(name) + 1);
1920 }
1921 cf_descriptors.emplace_back(name, ColumnFamilyOptions(options_));
1922 }
1923 while (cf_descriptors.size() < (size_t)FLAGS_column_families) {
1924 std::string name = ToString(new_column_family_name_.load());
1925 new_column_family_name_++;
1926 cf_descriptors.emplace_back(name, ColumnFamilyOptions(options_));
1927 column_family_names_.push_back(name);
1928 }
1929 options_.listeners.clear();
1930 options_.listeners.emplace_back(
1931 new DbStressListener(FLAGS_db, options_.db_paths, cf_descriptors));
1932 options_.create_missing_column_families = true;
1933 if (!FLAGS_use_txn) {
1934 #ifndef ROCKSDB_LITE
1935 if (FLAGS_use_blob_db) {
1936 blob_db::BlobDBOptions blob_db_options;
1937 blob_db_options.min_blob_size = FLAGS_blob_db_min_blob_size;
1938 blob_db_options.bytes_per_sync = FLAGS_blob_db_bytes_per_sync;
1939 blob_db_options.blob_file_size = FLAGS_blob_db_file_size;
1940 blob_db_options.enable_garbage_collection = FLAGS_blob_db_enable_gc;
1941 blob_db_options.garbage_collection_cutoff = FLAGS_blob_db_gc_cutoff;
1942
1943 blob_db::BlobDB* blob_db = nullptr;
1944 s = blob_db::BlobDB::Open(options_, blob_db_options, FLAGS_db,
1945 cf_descriptors, &column_families_, &blob_db);
1946 if (s.ok()) {
1947 db_ = blob_db;
1948 }
1949 } else
1950 #endif // !ROCKSDB_LITE
1951 {
1952 if (db_preload_finished_.load() && FLAGS_read_only) {
1953 s = DB::OpenForReadOnly(DBOptions(options_), FLAGS_db, cf_descriptors,
1954 &column_families_, &db_);
1955 } else {
1956 s = DB::Open(DBOptions(options_), FLAGS_db, cf_descriptors,
1957 &column_families_, &db_);
1958 }
1959 }
1960 } else {
1961 #ifndef ROCKSDB_LITE
1962 TransactionDBOptions txn_db_options;
1963 assert(FLAGS_txn_write_policy <= TxnDBWritePolicy::WRITE_UNPREPARED);
1964 txn_db_options.write_policy =
1965 static_cast<TxnDBWritePolicy>(FLAGS_txn_write_policy);
1966 if (FLAGS_unordered_write) {
1967 assert(txn_db_options.write_policy == TxnDBWritePolicy::WRITE_PREPARED);
1968 options_.unordered_write = true;
1969 options_.two_write_queues = true;
1970 txn_db_options.skip_concurrency_control = true;
1971 }
1972 s = TransactionDB::Open(options_, txn_db_options, FLAGS_db,
1973 cf_descriptors, &column_families_, &txn_db_);
1974 if (!s.ok()) {
1975 fprintf(stderr, "Error in opening the TransactionDB [%s]\n",
1976 s.ToString().c_str());
1977 fflush(stderr);
1978 }
1979 assert(s.ok());
1980 db_ = txn_db_;
1981 // after a crash, rollback to commit recovered transactions
1982 std::vector<Transaction*> trans;
1983 txn_db_->GetAllPreparedTransactions(&trans);
1984 Random rand(static_cast<uint32_t>(FLAGS_seed));
1985 for (auto txn : trans) {
1986 if (rand.OneIn(2)) {
1987 s = txn->Commit();
1988 assert(s.ok());
1989 } else {
1990 s = txn->Rollback();
1991 assert(s.ok());
1992 }
1993 delete txn;
1994 }
1995 trans.clear();
1996 txn_db_->GetAllPreparedTransactions(&trans);
1997 assert(trans.size() == 0);
1998 #endif
1999 }
2000 assert(!s.ok() || column_families_.size() ==
2001 static_cast<size_t>(FLAGS_column_families));
2002
2003 if (FLAGS_test_secondary) {
2004 #ifndef ROCKSDB_LITE
2005 secondaries_.resize(FLAGS_threads);
2006 std::fill(secondaries_.begin(), secondaries_.end(), nullptr);
2007 secondary_cfh_lists_.clear();
2008 secondary_cfh_lists_.resize(FLAGS_threads);
2009 Options tmp_opts;
2010 // TODO(yanqin) support max_open_files != -1 for secondary instance.
2011 tmp_opts.max_open_files = -1;
2012 tmp_opts.statistics = dbstats_secondaries;
2013 tmp_opts.env = db_stress_env;
2014 for (size_t i = 0; i != static_cast<size_t>(FLAGS_threads); ++i) {
2015 const std::string secondary_path =
2016 FLAGS_secondaries_base + "/" + std::to_string(i);
2017 s = DB::OpenAsSecondary(tmp_opts, FLAGS_db, secondary_path,
2018 cf_descriptors, &secondary_cfh_lists_[i],
2019 &secondaries_[i]);
2020 if (!s.ok()) {
2021 break;
2022 }
2023 }
2024 assert(s.ok());
2025 #else
2026 fprintf(stderr, "Secondary is not supported in RocksDBLite\n");
2027 exit(1);
2028 #endif
2029 }
2030 if (FLAGS_continuous_verification_interval > 0 && !cmp_db_) {
2031 Options tmp_opts;
2032 // TODO(yanqin) support max_open_files != -1 for secondary instance.
2033 tmp_opts.max_open_files = -1;
2034 tmp_opts.env = db_stress_env;
2035 std::string secondary_path = FLAGS_secondaries_base + "/cmp_database";
2036 s = DB::OpenAsSecondary(tmp_opts, FLAGS_db, secondary_path,
2037 cf_descriptors, &cmp_cfhs_, &cmp_db_);
2038 assert(!s.ok() ||
2039 cmp_cfhs_.size() == static_cast<size_t>(FLAGS_column_families));
2040 }
2041 } else {
2042 #ifndef ROCKSDB_LITE
2043 DBWithTTL* db_with_ttl;
2044 s = DBWithTTL::Open(options_, FLAGS_db, &db_with_ttl, FLAGS_ttl);
2045 db_ = db_with_ttl;
2046 if (FLAGS_test_secondary) {
2047 secondaries_.resize(FLAGS_threads);
2048 std::fill(secondaries_.begin(), secondaries_.end(), nullptr);
2049 Options tmp_opts;
2050 tmp_opts.env = options_.env;
2051 // TODO(yanqin) support max_open_files != -1 for secondary instance.
2052 tmp_opts.max_open_files = -1;
2053 for (size_t i = 0; i != static_cast<size_t>(FLAGS_threads); ++i) {
2054 const std::string secondary_path =
2055 FLAGS_secondaries_base + "/" + std::to_string(i);
2056 s = DB::OpenAsSecondary(tmp_opts, FLAGS_db, secondary_path,
2057 &secondaries_[i]);
2058 if (!s.ok()) {
2059 break;
2060 }
2061 }
2062 }
2063 #else
2064 fprintf(stderr, "TTL is not supported in RocksDBLite\n");
2065 exit(1);
2066 #endif
2067 }
2068 if (!s.ok()) {
2069 fprintf(stderr, "open error: %s\n", s.ToString().c_str());
2070 exit(1);
2071 }
2072 }
2073
Reopen(ThreadState * thread)2074 void StressTest::Reopen(ThreadState* thread) {
2075 #ifndef ROCKSDB_LITE
2076 // BG jobs in WritePrepared must be canceled first because i) they can access
2077 // the db via a callbac ii) they hold on to a snapshot and the upcoming
2078 // ::Close would complain about it.
2079 const bool write_prepared = FLAGS_use_txn && FLAGS_txn_write_policy != 0;
2080 bool bg_canceled = false;
2081 if (write_prepared || thread->rand.OneIn(2)) {
2082 const bool wait =
2083 write_prepared || static_cast<bool>(thread->rand.OneIn(2));
2084 CancelAllBackgroundWork(db_, wait);
2085 bg_canceled = wait;
2086 }
2087 assert(!write_prepared || bg_canceled);
2088 (void) bg_canceled;
2089 #else
2090 (void) thread;
2091 #endif
2092
2093 for (auto cf : column_families_) {
2094 delete cf;
2095 }
2096 column_families_.clear();
2097
2098 #ifndef ROCKSDB_LITE
2099 if (thread->rand.OneIn(2)) {
2100 Status s = db_->Close();
2101 if (!s.ok()) {
2102 fprintf(stderr, "Non-ok close status: %s\n", s.ToString().c_str());
2103 fflush(stderr);
2104 }
2105 assert(s.ok());
2106 }
2107 #endif
2108 delete db_;
2109 db_ = nullptr;
2110 #ifndef ROCKSDB_LITE
2111 txn_db_ = nullptr;
2112 #endif
2113
2114 assert(secondaries_.size() == secondary_cfh_lists_.size());
2115 size_t n = secondaries_.size();
2116 for (size_t i = 0; i != n; ++i) {
2117 for (auto* cf : secondary_cfh_lists_[i]) {
2118 delete cf;
2119 }
2120 secondary_cfh_lists_[i].clear();
2121 delete secondaries_[i];
2122 }
2123 secondaries_.clear();
2124
2125 num_times_reopened_++;
2126 auto now = db_stress_env->NowMicros();
2127 fprintf(stdout, "%s Reopening database for the %dth time\n",
2128 db_stress_env->TimeToString(now / 1000000).c_str(),
2129 num_times_reopened_);
2130 Open();
2131 }
2132 } // namespace ROCKSDB_NAMESPACE
2133 #endif // GFLAGS
2134