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).
10 fprintf(stderr
, "Please install gflags to run rocksdb tools\n");
18 #include "db/write_batch_internal.h"
19 #include "rocksdb/db.h"
20 #include "rocksdb/types.h"
21 #include "test_util/testutil.h"
22 #include "util/gflags_compat.h"
24 // Run a thread to perform Put's.
25 // Another thread uses GetUpdatesSince API to keep getting the updates.
27 // --num_inserts = the num of inserts the first thread should perform.
28 // --wal_ttl = the wal ttl for the run.
30 using namespace ROCKSDB_NAMESPACE
;
32 using GFLAGS_NAMESPACE::ParseCommandLineFlags
;
33 using GFLAGS_NAMESPACE::SetUsageMessage
;
35 struct DataPumpThread
{
37 DB
* db
; // Assumption DB is Open'ed already.
40 static void DataPumpThreadBody(void* arg
) {
41 DataPumpThread
* t
= reinterpret_cast<DataPumpThread
*>(arg
);
45 while (i
++ < t
->no_records
) {
46 if (!db
->Put(WriteOptions(), Slice(rnd
.RandomString(500)),
47 Slice(rnd
.RandomString(500)))
49 fprintf(stderr
, "Error in put\n");
55 struct ReplicationThread
{
56 std::atomic
<bool> stop
;
58 volatile size_t no_read
;
61 static void ReplicationThreadBody(void* arg
) {
62 ReplicationThread
* t
= reinterpret_cast<ReplicationThread
*>(arg
);
64 std::unique_ptr
<TransactionLogIterator
> iter
;
65 SequenceNumber currentSeqNum
= 1;
66 while (!t
->stop
.load(std::memory_order_acquire
)) {
69 while (!db
->GetUpdatesSince(currentSeqNum
, &iter
).ok()) {
70 if (t
->stop
.load(std::memory_order_acquire
)) {
74 fprintf(stderr
, "Refreshing iterator\n");
75 for (; iter
->Valid(); iter
->Next(), t
->no_read
++, currentSeqNum
++) {
76 BatchResult res
= iter
->GetBatch();
77 if (res
.sequence
!= currentSeqNum
) {
78 fprintf(stderr
, "Missed a seq no. b/w %ld and %ld\n",
79 (long)currentSeqNum
, (long)res
.sequence
);
86 DEFINE_uint64(num_inserts
, 1000,
87 "the num of inserts the first thread should"
89 DEFINE_uint64(wal_ttl_seconds
, 1000, "the wal ttl for the run(in seconds)");
90 DEFINE_uint64(wal_size_limit_MB
, 10,
91 "the wal size limit for the run"
94 int main(int argc
, const char** argv
) {
96 std::string("\nUSAGE:\n") + std::string(argv
[0]) +
97 " --num_inserts=<num_inserts> --wal_ttl_seconds=<WAL_ttl_seconds>" +
98 " --wal_size_limit_MB=<WAL_size_limit_MB>");
99 ParseCommandLineFlags(&argc
, const_cast<char***>(&argv
), true);
101 Env
* env
= Env::Default();
102 std::string default_db_path
;
103 env
->GetTestDirectory(&default_db_path
);
104 default_db_path
+= "db_repl_stress";
106 options
.create_if_missing
= true;
107 options
.WAL_ttl_seconds
= FLAGS_wal_ttl_seconds
;
108 options
.WAL_size_limit_MB
= FLAGS_wal_size_limit_MB
;
110 DestroyDB(default_db_path
, options
);
112 Status s
= DB::Open(options
, default_db_path
, &db
);
115 fprintf(stderr
, "Could not open DB due to %s\n", s
.ToString().c_str());
119 DataPumpThread dataPump
;
120 dataPump
.no_records
= FLAGS_num_inserts
;
122 env
->StartThread(DataPumpThreadBody
, &dataPump
);
124 ReplicationThread replThread
;
126 replThread
.no_read
= 0;
127 replThread
.stop
.store(false, std::memory_order_release
);
129 env
->StartThread(ReplicationThreadBody
, &replThread
);
130 while (replThread
.no_read
< FLAGS_num_inserts
)
132 replThread
.stop
.store(true, std::memory_order_release
);
133 if (replThread
.no_read
< dataPump
.no_records
) {
134 // no. read should be => than inserted.
136 "No. of Record's written and read not same\nRead : %" ROCKSDB_PRIszt
137 " Written : %" ROCKSDB_PRIszt
"\n",
138 replThread
.no_read
, dataPump
.no_records
);
141 fprintf(stderr
, "Successful!\n");
147 #else // ROCKSDB_LITE
149 int main(int /*argc*/, char** /*argv*/) {
150 fprintf(stderr
, "Not supported in lite mode.\n");
153 #endif // ROCKSDB_LITE