1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
|
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
#ifndef GFLAGS
#include <cstdio>
int main() {
fprintf(stderr, "Please install gflags to run rocksdb tools\n");
return 1;
}
#else
#include <atomic>
#include <cstdio>
#include "db/write_batch_internal.h"
#include "rocksdb/db.h"
#include "rocksdb/types.h"
#include "test_util/testutil.h"
#include "util/gflags_compat.h"
// Run a thread to perform Put's.
// Another thread uses GetUpdatesSince API to keep getting the updates.
// options :
// --num_inserts = the num of inserts the first thread should perform.
// --wal_ttl = the wal ttl for the run.
DEFINE_uint64(num_inserts, 1000,
"the num of inserts the first thread should"
" perform.");
DEFINE_uint64(wal_ttl_seconds, 1000, "the wal ttl for the run(in seconds)");
DEFINE_uint64(wal_size_limit_MB, 10,
"the wal size limit for the run"
"(in MB)");
using ROCKSDB_NAMESPACE::BatchResult;
using ROCKSDB_NAMESPACE::DB;
using ROCKSDB_NAMESPACE::DestroyDB;
using ROCKSDB_NAMESPACE::Env;
using ROCKSDB_NAMESPACE::Options;
using ROCKSDB_NAMESPACE::Random;
using ROCKSDB_NAMESPACE::SequenceNumber;
using ROCKSDB_NAMESPACE::Slice;
using ROCKSDB_NAMESPACE::Status;
using ROCKSDB_NAMESPACE::TransactionLogIterator;
using ROCKSDB_NAMESPACE::WriteOptions;
using GFLAGS_NAMESPACE::ParseCommandLineFlags;
using GFLAGS_NAMESPACE::SetUsageMessage;
struct DataPumpThread {
DB* db; // Assumption DB is Open'ed already.
};
static void DataPumpThreadBody(void* arg) {
DataPumpThread* t = static_cast<DataPumpThread*>(arg);
DB* db = t->db;
Random rnd(301);
uint64_t i = 0;
while (i++ < FLAGS_num_inserts) {
if (!db->Put(WriteOptions(), Slice(rnd.RandomString(500)),
Slice(rnd.RandomString(500)))
.ok()) {
fprintf(stderr, "Error in put\n");
exit(1);
}
}
}
int main(int argc, const char** argv) {
SetUsageMessage(
std::string("\nUSAGE:\n") + std::string(argv[0]) +
" --num_inserts=<num_inserts> --wal_ttl_seconds=<WAL_ttl_seconds>" +
" --wal_size_limit_MB=<WAL_size_limit_MB>");
ParseCommandLineFlags(&argc, const_cast<char***>(&argv), true);
Env* env = Env::Default();
std::string default_db_path;
env->GetTestDirectory(&default_db_path);
default_db_path += "db_repl_stress";
Options options;
options.create_if_missing = true;
options.WAL_ttl_seconds = FLAGS_wal_ttl_seconds;
options.WAL_size_limit_MB = FLAGS_wal_size_limit_MB;
DB* db;
DestroyDB(default_db_path, options);
Status s = DB::Open(options, default_db_path, &db);
if (!s.ok()) {
fprintf(stderr, "Could not open DB due to %s\n", s.ToString().c_str());
exit(1);
}
DataPumpThread dataPump;
dataPump.db = db;
env->StartThread(DataPumpThreadBody, &dataPump);
std::unique_ptr<TransactionLogIterator> iter;
SequenceNumber currentSeqNum = 1;
uint64_t num_read = 0;
for (;;) {
iter.reset();
// Continue to probe a bit more after all received
size_t probes = 0;
while (!db->GetUpdatesSince(currentSeqNum, &iter).ok()) {
probes++;
if (probes > 100 && num_read >= FLAGS_num_inserts) {
if (num_read > FLAGS_num_inserts) {
fprintf(stderr, "Too many updates read: %ld expected: %ld\n",
(long)num_read, (long)FLAGS_num_inserts);
exit(1);
}
fprintf(stderr, "Successful!\n");
return 0;
}
}
fprintf(stderr, "Refreshing iterator\n");
for (; iter->Valid(); iter->Next(), num_read++, currentSeqNum++) {
BatchResult res = iter->GetBatch();
if (res.sequence != currentSeqNum) {
fprintf(stderr, "Missed a seq no. b/w %ld and %ld\n",
(long)currentSeqNum, (long)res.sequence);
exit(1);
}
}
}
}
#endif // GFLAGS
|
