db_stress.cc 86.4 KB
Newer Older
1
//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
S
Siying Dong 已提交
2 3 4
//  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).
5
//
6 7 8
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.
9
//
X
Xing Jin 已提交
10 11 12 13 14 15 16 17 18 19 20 21
// The test uses an array to compare against values written to the database.
// Keys written to the array are in 1:1 correspondence to the actual values in
// the database according to the formula in the function GenerateValue.

// Space is reserved in the array from 0 to FLAGS_max_key and values are
// randomly written/deleted/read from those positions. During verification we
// compare all the positions in the array. To shorten/elongate the running
// time, you could change the settings: FLAGS_max_key, FLAGS_ops_per_thread,
// (sometimes also FLAGS_threads).
//
// NOTE that if FLAGS_test_batches_snapshots is set, the test will have
// different behavior. See comment of the flag for details.
A
amayank 已提交
22

23 24 25 26 27 28 29 30
#ifndef GFLAGS
#include <cstdio>
int main() {
  fprintf(stderr, "Please install gflags to run rocksdb tools\n");
  return 1;
}
#else

31
#define __STDC_FORMAT_MACROS
32
#include <fcntl.h>
33
#include <inttypes.h>
34 35
#include <stdio.h>
#include <stdlib.h>
36
#include <sys/types.h>
37
#include <algorithm>
38
#include <chrono>
39
#include <exception>
40 41
#include <thread>

42
#include <gflags/gflags.h>
43 44
#include "db/db_impl.h"
#include "db/version_set.h"
45
#include "hdfs/env_hdfs.h"
46
#include "monitoring/histogram.h"
47
#include "port/port.h"
48 49
#include "rocksdb/cache.h"
#include "rocksdb/env.h"
50 51
#include "rocksdb/slice.h"
#include "rocksdb/slice_transform.h"
52 53 54
#include "rocksdb/statistics.h"
#include "rocksdb/utilities/db_ttl.h"
#include "rocksdb/write_batch.h"
55
#include "util/coding.h"
56
#include "util/compression.h"
57
#include "util/crc32c.h"
58
#include "util/logging.h"
59 60
#include "util/mutexlock.h"
#include "util/random.h"
61
#include "util/string_util.h"
62 63 64 65
// SyncPoint is not supported in Released Windows Mode.
#if !(defined NDEBUG) || !defined(OS_WIN)
#include "util/sync_point.h"
#endif  // !(defined NDEBUG) || !defined(OS_WIN)
66
#include "util/testutil.h"
67

D
Deon Nicholas 已提交
68
#include "utilities/merge_operators.h"
69

70 71 72
using GFLAGS::ParseCommandLineFlags;
using GFLAGS::RegisterFlagValidator;
using GFLAGS::SetUsageMessage;
73

74
static const long KB = 1024;
75

76 77
static bool ValidateUint32Range(const char* flagname, uint64_t value) {
  if (value > std::numeric_limits<uint32_t>::max()) {
K
kailiu 已提交
78 79 80 81
    fprintf(stderr,
            "Invalid value for --%s: %lu, overflow\n",
            flagname,
            (unsigned long)value);
82 83 84 85
    return false;
  }
  return true;
}
86

87
DEFINE_uint64(seed, 2341234, "Seed for PRNG");
I
Igor Canadi 已提交
88
static const bool FLAGS_seed_dummy __attribute__((unused)) =
89
    RegisterFlagValidator(&FLAGS_seed, &ValidateUint32Range);
90

I
Igor Canadi 已提交
91
DEFINE_int64(max_key, 1 * KB* KB,
92
             "Max number of key/values to place in database");
93

I
Igor Canadi 已提交
94 95
DEFINE_int32(column_families, 10, "Number of column families");

A
Andrew Kryczka 已提交
96 97 98 99 100 101 102
DEFINE_int64(
    active_width, 0,
    "Number of keys in active span of the key-range at any given time. The "
    "span begins with its left endpoint at key 0, gradually moves rightwards, "
    "and ends with its right endpoint at max_key. If set to 0, active_width "
    "will be sanitized to be equal to max_key.");

A
Andres Noetzli 已提交
103
// TODO(noetzli) Add support for single deletes
104
DEFINE_bool(test_batches_snapshots, false,
A
Andres Noetzli 已提交
105
            "If set, the test uses MultiGet(), MultiPut() and MultiDelete()"
106 107 108 109 110 111 112 113 114
            " which read/write/delete multiple keys in a batch. In this mode,"
            " we do not verify db content by comparing the content with the "
            "pre-allocated array. Instead, we do partial verification inside"
            " MultiGet() by checking various values in a batch. Benefit of"
            " this mode:\n"
            "\t(a) No need to acquire mutexes during writes (less cache "
            "flushes in multi-core leading to speed up)\n"
            "\t(b) No long validation at the end (more speed up)\n"
            "\t(c) Test snapshot and atomicity of batch writes");
115

116
DEFINE_int32(threads, 32, "Number of concurrent threads to run.");
117

118 119 120 121
DEFINE_int32(ttl, -1,
             "Opens the db with this ttl value if this is not -1. "
             "Carefully specify a large value such that verifications on "
             "deleted values don't fail");
122

123 124
DEFINE_int32(value_size_mult, 8,
             "Size of value will be this number times rand_int(1,3) bytes");
125

126 127
DEFINE_int32(compaction_readahead_size, 0, "Compaction readahead size");

128
DEFINE_bool(verify_before_write, false, "Verify before write");
129

130
DEFINE_bool(histogram, false, "Print histogram of operation timings");
131

132 133
DEFINE_bool(destroy_db_initially, true,
            "Destroys the database dir before start if this is true");
134

135 136 137 138
DEFINE_bool(verbose, false, "Verbose");

DEFINE_bool(progress_reports, true,
            "If true, db_stress will report number of finished operations");
139

140 141 142
DEFINE_uint64(db_write_buffer_size, rocksdb::Options().db_write_buffer_size,
              "Number of bytes to buffer in all memtables before compacting");

143 144
DEFINE_int32(write_buffer_size,
             static_cast<int32_t>(rocksdb::Options().write_buffer_size),
145
             "Number of bytes to buffer in memtable before compacting");
146

147 148 149 150
DEFINE_int32(max_write_buffer_number,
             rocksdb::Options().max_write_buffer_number,
             "The number of in-memory memtables. "
             "Each memtable is of size FLAGS_write_buffer_size.");
151

152 153 154 155 156 157 158 159 160 161
DEFINE_int32(min_write_buffer_number_to_merge,
             rocksdb::Options().min_write_buffer_number_to_merge,
             "The minimum number of write buffers that will be merged together "
             "before writing to storage. This is cheap because it is an "
             "in-memory merge. If this feature is not enabled, then all these "
             "write buffers are flushed to L0 as separate files and this "
             "increases read amplification because a get request has to check "
             "in all of these files. Also, an in-memory merge may result in "
             "writing less data to storage if there are duplicate records in"
             " each of these individual write buffers.");
162

163 164 165 166 167 168 169 170 171 172 173 174 175 176
DEFINE_int32(max_write_buffer_number_to_maintain,
             rocksdb::Options().max_write_buffer_number_to_maintain,
             "The total maximum number of write buffers to maintain in memory "
             "including copies of buffers that have already been flushed. "
             "Unlike max_write_buffer_number, this parameter does not affect "
             "flushing. This controls the minimum amount of write history "
             "that will be available in memory for conflict checking when "
             "Transactions are used. If this value is too low, some "
             "transactions may fail at commit time due to not being able to "
             "determine whether there were any write conflicts. Setting this "
             "value to 0 will cause write buffers to be freed immediately "
             "after they are flushed.  If this value is set to -1, "
             "'max_write_buffer_number' will be used.");

177 178 179
DEFINE_int32(open_files, rocksdb::Options().max_open_files,
             "Maximum number of files to keep open at the same time "
             "(use default if == 0)");
180

181 182 183 184
DEFINE_int64(compressed_cache_size, -1,
             "Number of bytes to use as a cache of compressed data."
             " Negative means use default settings.");

185
DEFINE_int32(compaction_style, rocksdb::Options().compaction_style, "");
186

187 188 189
DEFINE_int32(level0_file_num_compaction_trigger,
             rocksdb::Options().level0_file_num_compaction_trigger,
             "Level0 compaction start trigger");
190

191 192 193
DEFINE_int32(level0_slowdown_writes_trigger,
             rocksdb::Options().level0_slowdown_writes_trigger,
             "Number of files in level-0 that will slow down writes");
194

195 196 197
DEFINE_int32(level0_stop_writes_trigger,
             rocksdb::Options().level0_stop_writes_trigger,
             "Number of files in level-0 that will trigger put stop.");
198

199 200
DEFINE_int32(block_size,
             static_cast<int32_t>(rocksdb::BlockBasedTableOptions().block_size),
201
             "Number of bytes in a block.");
202

203 204 205 206
DEFINE_int32(max_background_compactions,
             rocksdb::Options().max_background_compactions,
             "The maximum number of concurrent background compactions "
             "that can occur in parallel.");
207

208 209 210 211
DEFINE_int32(compaction_thread_pool_adjust_interval, 0,
             "The interval (in milliseconds) to adjust compaction thread pool "
             "size. Don't change it periodically if the value is 0.");

C
clark.kang 已提交
212
DEFINE_int32(compaction_thread_pool_variations, 2,
A
Andres Noetzli 已提交
213
             "Range of background thread pool size variations when adjusted "
214 215
             "periodically.");

I
Igor Canadi 已提交
216 217 218 219
DEFINE_int32(max_background_flushes, rocksdb::Options().max_background_flushes,
             "The maximum number of concurrent background flushes "
             "that can occur in parallel.");

220 221
DEFINE_int32(universal_size_ratio, 0, "The ratio of file sizes that trigger"
             " compaction in universal style");
222

223 224
DEFINE_int32(universal_min_merge_width, 0, "The minimum number of files to "
             "compact in universal style compaction");
225

226 227
DEFINE_int32(universal_max_merge_width, 0, "The max number of files to compact"
             " in universal style compaction");
228

229 230
DEFINE_int32(universal_max_size_amplification_percent, 0,
             "The max size amplification for universal style compaction");
231

I
Igor Canadi 已提交
232 233 234 235 236
DEFINE_int32(clear_column_family_one_in, 1000000,
             "With a chance of 1/N, delete a column family and then recreate "
             "it again. If N == 0, never drop/create column families. "
             "When test_batches_snapshots is true, this flag has no effect");

L
Lei Jin 已提交
237 238 239
DEFINE_int32(set_options_one_in, 0,
             "With a chance of 1/N, change some random options");

240 241 242
DEFINE_int32(set_in_place_one_in, 0,
             "With a chance of 1/N, toggle in place support option");

243
DEFINE_int64(cache_size, 2LL * KB * KB * KB,
244
             "Number of bytes to use as a cache of uncompressed data.");
245

Y
Yi Wu 已提交
246 247 248
DEFINE_bool(use_clock_cache, false,
            "Replace default LRU block cache with clock cache.");

249
DEFINE_uint64(subcompactions, 1,
250 251
              "Maximum number of subcompactions to divide L0-L1 compactions "
              "into.");
252

253
DEFINE_bool(allow_concurrent_memtable_write, false,
254 255 256 257 258
            "Allow multi-writers to update mem tables in parallel.");

DEFINE_bool(enable_write_thread_adaptive_yield, true,
            "Use a yielding spin loop for brief writer thread waits.");

259 260 261
static const bool FLAGS_subcompactions_dummy __attribute__((unused)) =
    RegisterFlagValidator(&FLAGS_subcompactions, &ValidateUint32Range);

262 263 264 265 266 267 268 269 270
static bool ValidateInt32Positive(const char* flagname, int32_t value) {
  if (value < 0) {
    fprintf(stderr, "Invalid value for --%s: %d, must be >=0\n",
            flagname, value);
    return false;
  }
  return true;
}
DEFINE_int32(reopen, 10, "Number of times database reopens");
I
Igor Canadi 已提交
271
static const bool FLAGS_reopen_dummy __attribute__((unused)) =
272
    RegisterFlagValidator(&FLAGS_reopen, &ValidateInt32Positive);
273

274 275
DEFINE_int32(bloom_bits, 10, "Bloom filter bits per key. "
             "Negative means use default settings.");
276

277 278 279
DEFINE_bool(use_block_based_filter, false, "use block based filter"
              "instead of full filter for block based table");

280
DEFINE_string(db, "", "Use the db with the following name.");
281

282 283
DEFINE_bool(verify_checksum, false,
            "Verify checksum for every block read from storage");
284

285
DEFINE_bool(mmap_read, rocksdb::Options().allow_mmap_reads,
286
            "Allow reads to occur via mmap-ing files");
287

288
DEFINE_bool(mmap_write, rocksdb::Options().allow_mmap_writes,
289 290
            "Allow writes to occur via mmap-ing files");

291
DEFINE_bool(use_direct_reads, rocksdb::Options().use_direct_reads,
292 293
            "Use O_DIRECT for reading data");

294 295
DEFINE_bool(use_direct_io_for_flush_and_compaction,
            rocksdb::Options().use_direct_io_for_flush_and_compaction,
296 297
            "Use O_DIRECT for writing data");

298 299 300
// Database statistics
static std::shared_ptr<rocksdb::Statistics> dbstats;
DEFINE_bool(statistics, false, "Create database statistics");
301

302
DEFINE_bool(sync, false, "Sync all writes to disk");
303

304
DEFINE_bool(use_fsync, false, "If true, issue fsync instead of fdatasync");
305

306 307 308
DEFINE_int32(kill_random_test, 0,
             "If non-zero, kill at various points in source code with "
             "probability 1/this");
I
Igor Canadi 已提交
309
static const bool FLAGS_kill_random_test_dummy __attribute__((unused)) =
310
    RegisterFlagValidator(&FLAGS_kill_random_test, &ValidateInt32Positive);
311
extern int rocksdb_kill_odds;
312

313 314 315 316 317
DEFINE_string(kill_prefix_blacklist, "",
              "If non-empty, kill points with prefix in the list given will be"
              " skipped. Items are comma-separated.");
extern std::vector<std::string> rocksdb_kill_prefix_blacklist;

318
DEFINE_bool(disable_wal, false, "If true, do not write WAL for write.");
319

320 321
DEFINE_int32(target_file_size_base, 64 * KB,
             "Target level-1 file size for compaction");
322

323
DEFINE_int32(target_file_size_multiplier, 1,
A
Andres Noetzli 已提交
324
             "A multiplier to compute target level-N file size (N >= 2)");
325

326
DEFINE_uint64(max_bytes_for_level_base, 256 * KB, "Max bytes for level-1");
327

328 329
DEFINE_double(max_bytes_for_level_multiplier, 2,
              "A multiplier to compute max bytes for level-N (N >= 2)");
330

331 332 333
DEFINE_int32(range_deletion_width, 10,
             "The width of the range deletion intervals.");

334 335 336 337 338
DEFINE_uint64(rate_limiter_bytes_per_sec, 0, "Set options.rate_limiter value.");

DEFINE_bool(rate_limit_bg_reads, false,
            "Use options.rate_limiter on compaction reads");

339 340
// Temporarily disable this to allows it to detect new bugs
DEFINE_int32(compact_files_one_in, 0,
341 342 343
             "If non-zero, then CompactFiles() will be called one for every N "
             "operations IN AVERAGE.  0 indicates CompactFiles() is disabled.");

344 345 346 347 348 349 350 351
static bool ValidateInt32Percent(const char* flagname, int32_t value) {
  if (value < 0 || value>100) {
    fprintf(stderr, "Invalid value for --%s: %d, 0<= pct <=100 \n",
            flagname, value);
    return false;
  }
  return true;
}
352

353 354
DEFINE_int32(readpercent, 10,
             "Ratio of reads to total workload (expressed as a percentage)");
I
Igor Canadi 已提交
355
static const bool FLAGS_readpercent_dummy __attribute__((unused)) =
356
    RegisterFlagValidator(&FLAGS_readpercent, &ValidateInt32Percent);
357 358 359 360

DEFINE_int32(prefixpercent, 20,
             "Ratio of prefix iterators to total workload (expressed as a"
             " percentage)");
I
Igor Canadi 已提交
361
static const bool FLAGS_prefixpercent_dummy __attribute__((unused)) =
362
    RegisterFlagValidator(&FLAGS_prefixpercent, &ValidateInt32Percent);
363 364

DEFINE_int32(writepercent, 45,
A
Andres Noetzli 已提交
365
             "Ratio of writes to total workload (expressed as a percentage)");
I
Igor Canadi 已提交
366
static const bool FLAGS_writepercent_dummy __attribute__((unused)) =
367
    RegisterFlagValidator(&FLAGS_writepercent, &ValidateInt32Percent);
368 369 370

DEFINE_int32(delpercent, 15,
             "Ratio of deletes to total workload (expressed as a percentage)");
I
Igor Canadi 已提交
371
static const bool FLAGS_delpercent_dummy __attribute__((unused)) =
372
    RegisterFlagValidator(&FLAGS_delpercent, &ValidateInt32Percent);
373

374 375 376 377 378 379
DEFINE_int32(delrangepercent, 0,
             "Ratio of range deletions to total workload (expressed as a "
             "percentage). Cannot be used with test_batches_snapshots");
static const bool FLAGS_delrangepercent_dummy __attribute__((unused)) =
    RegisterFlagValidator(&FLAGS_delrangepercent, &ValidateInt32Percent);

A
Andres Noetzli 已提交
380 381 382 383 384 385
DEFINE_int32(nooverwritepercent, 60,
             "Ratio of keys without overwrite to total workload (expressed as "
             " a percentage)");
static const bool FLAGS_nooverwritepercent_dummy __attribute__((__unused__)) =
    RegisterFlagValidator(&FLAGS_nooverwritepercent, &ValidateInt32Percent);

386 387
DEFINE_int32(iterpercent, 10, "Ratio of iterations to total workload"
             " (expressed as a percentage)");
I
Igor Canadi 已提交
388
static const bool FLAGS_iterpercent_dummy __attribute__((unused)) =
389
    RegisterFlagValidator(&FLAGS_iterpercent, &ValidateInt32Percent);
390 391

DEFINE_uint64(num_iterations, 10, "Number of iterations per MultiIterate run");
I
Igor Canadi 已提交
392
static const bool FLAGS_num_iterations_dummy __attribute__((unused)) =
393
    RegisterFlagValidator(&FLAGS_num_iterations, &ValidateUint32Range);
394

I
Igor Canadi 已提交
395
namespace {
396 397 398 399 400 401 402 403 404 405 406
enum rocksdb::CompressionType StringToCompressionType(const char* ctype) {
  assert(ctype);

  if (!strcasecmp(ctype, "none"))
    return rocksdb::kNoCompression;
  else if (!strcasecmp(ctype, "snappy"))
    return rocksdb::kSnappyCompression;
  else if (!strcasecmp(ctype, "zlib"))
    return rocksdb::kZlibCompression;
  else if (!strcasecmp(ctype, "bzip2"))
    return rocksdb::kBZip2Compression;
A
Albert Strasheim 已提交
407 408 409 410
  else if (!strcasecmp(ctype, "lz4"))
    return rocksdb::kLZ4Compression;
  else if (!strcasecmp(ctype, "lz4hc"))
    return rocksdb::kLZ4HCCompression;
411 412
  else if (!strcasecmp(ctype, "xpress"))
    return rocksdb::kXpressCompression;
413
  else if (!strcasecmp(ctype, "zstd"))
S
sdong 已提交
414
    return rocksdb::kZSTD;
415 416 417 418

  fprintf(stdout, "Cannot parse compression type '%s'\n", ctype);
  return rocksdb::kSnappyCompression; //default value
}
419 420 421 422 423 424 425 426 427 428 429 430 431 432 433

std::vector<std::string> SplitString(std::string src) {
  std::vector<std::string> ret;
  if (src.empty()) {
    return ret;
  }
  size_t pos = 0;
  size_t pos_comma;
  while ((pos_comma = src.find(',', pos)) != std::string::npos) {
    ret.push_back(src.substr(pos, pos_comma - pos));
    pos = pos_comma + 1;
  }
  ret.push_back(src.substr(pos, src.length()));
  return ret;
}
I
Igor Canadi 已提交
434 435
}  // namespace

436 437 438
DEFINE_string(compression_type, "snappy",
              "Algorithm to use to compress the database");
static enum rocksdb::CompressionType FLAGS_compression_type_e =
439
    rocksdb::kSnappyCompression;
440

441
DEFINE_string(hdfs, "", "Name of hdfs environment");
442
// posix or hdfs environment
443
static rocksdb::Env* FLAGS_env = rocksdb::Env::Default();
444

I
Igor Canadi 已提交
445 446
DEFINE_uint64(ops_per_thread, 1200000, "Number of operations per thread.");
static const bool FLAGS_ops_per_thread_dummy __attribute__((unused)) =
447
    RegisterFlagValidator(&FLAGS_ops_per_thread, &ValidateUint32Range);
448

449
DEFINE_uint64(log2_keys_per_lock, 2, "Log2 of number of keys per lock");
I
Igor Canadi 已提交
450
static const bool FLAGS_log2_keys_per_lock_dummy __attribute__((unused)) =
451
    RegisterFlagValidator(&FLAGS_log2_keys_per_lock, &ValidateUint32Range);
452

453 454
DEFINE_bool(in_place_update, false, "On true, does inplace update in memtable");

J
Jim Paton 已提交
455 456
enum RepFactory {
  kSkipList,
I
Igor Canadi 已提交
457
  kHashSkipList,
J
Jim Paton 已提交
458 459
  kVectorRep
};
I
Igor Canadi 已提交
460 461

namespace {
462 463 464 465 466 467
enum RepFactory StringToRepFactory(const char* ctype) {
  assert(ctype);

  if (!strcasecmp(ctype, "skip_list"))
    return kSkipList;
  else if (!strcasecmp(ctype, "prefix_hash"))
I
Igor Canadi 已提交
468
    return kHashSkipList;
469 470 471 472 473 474
  else if (!strcasecmp(ctype, "vector"))
    return kVectorRep;

  fprintf(stdout, "Cannot parse memreptable %s\n", ctype);
  return kSkipList;
}
I
Igor Canadi 已提交
475 476
}  // namespace

J
Jim Paton 已提交
477
static enum RepFactory FLAGS_rep_factory;
L
Lei Jin 已提交
478
DEFINE_string(memtablerep, "prefix_hash", "");
J
Jim Paton 已提交
479

480
static bool ValidatePrefixSize(const char* flagname, int32_t value) {
L
Lei Jin 已提交
481 482
  if (value < 0 || value > 8) {
    fprintf(stderr, "Invalid value for --%s: %d. 0 <= PrefixSize <= 8\n",
483 484 485 486 487
            flagname, value);
    return false;
  }
  return true;
}
I
Igor Canadi 已提交
488
DEFINE_int32(prefix_size, 7, "Control the prefix size for HashSkipListRep");
L
Lei Jin 已提交
489
static const bool FLAGS_prefix_size_dummy __attribute__((unused)) =
490
    RegisterFlagValidator(&FLAGS_prefix_size, &ValidatePrefixSize);
491 492 493

DEFINE_bool(use_merge, false, "On true, replaces all writes with a Merge "
            "that behaves like a Put");
J
Jim Paton 已提交
494

495 496 497
DEFINE_bool(use_full_merge_v1, false,
            "On true, use a merge operator that implement the deprecated "
            "version of FullMerge");
D
Deon Nicholas 已提交
498

499
namespace rocksdb {
500

501
// convert long to a big-endian slice key
D
Dmitri Smirnov 已提交
502
static std::string Key(int64_t val) {
503 504 505 506 507
  std::string little_endian_key;
  std::string big_endian_key;
  PutFixed64(&little_endian_key, val);
  assert(little_endian_key.size() == sizeof(val));
  big_endian_key.resize(sizeof(val));
D
Dmitri Smirnov 已提交
508
  for (size_t i = 0 ; i < sizeof(val); ++i) {
509 510 511 512 513
    big_endian_key[i] = little_endian_key[sizeof(val) - 1 - i];
  }
  return big_endian_key;
}

514 515
static std::string StringToHex(const std::string& str) {
  std::string result = "0x";
L
Laurent Demailly 已提交
516
  result.append(Slice(str).ToString(true));
517 518 519 520
  return result;
}


521 522 523 524 525
class StressTest;
namespace {

class Stats {
 private:
D
Dmitri Smirnov 已提交
526 527 528
  uint64_t start_;
  uint64_t finish_;
  double  seconds_;
529
  long done_;
530 531
  long gets_;
  long prefixes_;
532
  long writes_;
A
amayank 已提交
533
  long deletes_;
A
Andres Noetzli 已提交
534
  size_t single_deletes_;
535
  long iterator_size_sums_;
536
  long founds_;
537
  long iterations_;
538 539
  long range_deletions_;
  long covered_by_range_deletions_;
540
  long errors_;
541 542
  long num_compact_files_succeed_;
  long num_compact_files_failed_;
543 544
  int next_report_;
  size_t bytes_;
D
Dmitri Smirnov 已提交
545
  uint64_t last_op_finish_;
546
  HistogramImpl hist_;
547 548 549 550 551 552 553 554

 public:
  Stats() { }

  void Start() {
    next_report_ = 100;
    hist_.Clear();
    done_ = 0;
555 556
    gets_ = 0;
    prefixes_ = 0;
557
    writes_ = 0;
A
amayank 已提交
558
    deletes_ = 0;
A
Andres Noetzli 已提交
559
    single_deletes_ = 0;
560
    iterator_size_sums_ = 0;
561
    founds_ = 0;
562
    iterations_ = 0;
563 564
    range_deletions_ = 0;
    covered_by_range_deletions_ = 0;
565
    errors_ = 0;
566 567
    bytes_ = 0;
    seconds_ = 0;
568 569
    num_compact_files_succeed_ = 0;
    num_compact_files_failed_ = 0;
570 571 572 573 574 575 576 577
    start_ = FLAGS_env->NowMicros();
    last_op_finish_ = start_;
    finish_ = start_;
  }

  void Merge(const Stats& other) {
    hist_.Merge(other.hist_);
    done_ += other.done_;
578 579
    gets_ += other.gets_;
    prefixes_ += other.prefixes_;
580
    writes_ += other.writes_;
A
amayank 已提交
581
    deletes_ += other.deletes_;
A
Andres Noetzli 已提交
582
    single_deletes_ += other.single_deletes_;
583
    iterator_size_sums_ += other.iterator_size_sums_;
584
    founds_ += other.founds_;
585
    iterations_ += other.iterations_;
586 587
    range_deletions_ += other.range_deletions_;
    covered_by_range_deletions_ = other.covered_by_range_deletions_;
588
    errors_ += other.errors_;
589 590
    bytes_ += other.bytes_;
    seconds_ += other.seconds_;
591 592
    num_compact_files_succeed_ += other.num_compact_files_succeed_;
    num_compact_files_failed_ += other.num_compact_files_failed_;
593 594 595 596 597 598 599 600 601 602 603
    if (other.start_ < start_) start_ = other.start_;
    if (other.finish_ > finish_) finish_ = other.finish_;
  }

  void Stop() {
    finish_ = FLAGS_env->NowMicros();
    seconds_ = (finish_ - start_) * 1e-6;
  }

  void FinishedSingleOp() {
    if (FLAGS_histogram) {
D
Dmitri Smirnov 已提交
604 605
      auto now = FLAGS_env->NowMicros();
      auto micros = now - last_op_finish_;
606 607
      hist_.Add(micros);
      if (micros > 20000) {
D
Dmitri Smirnov 已提交
608
        fprintf(stdout, "long op: %" PRIu64 " micros%30s\r", micros, "");
609 610 611 612
      }
      last_op_finish_ = now;
    }

613
      done_++;
614
    if (FLAGS_progress_reports) {
615 616 617 618 619 620 621 622 623 624
      if (done_ >= next_report_) {
        if      (next_report_ < 1000)   next_report_ += 100;
        else if (next_report_ < 5000)   next_report_ += 500;
        else if (next_report_ < 10000)  next_report_ += 1000;
        else if (next_report_ < 50000)  next_report_ += 5000;
        else if (next_report_ < 100000) next_report_ += 10000;
        else if (next_report_ < 500000) next_report_ += 50000;
        else                            next_report_ += 100000;
        fprintf(stdout, "... finished %ld ops%30s\r", done_, "");
      }
625 626 627
    }
  }

628 629 630 631 632
  void AddBytesForWrites(int nwrites, size_t nbytes) {
    writes_ += nwrites;
    bytes_ += nbytes;
  }

633 634 635
  void AddGets(int ngets, int nfounds) {
    founds_ += nfounds;
    gets_ += ngets;
636 637
  }

638 639 640 641 642
  void AddPrefixes(int nprefixes, int count) {
    prefixes_ += nprefixes;
    iterator_size_sums_ += count;
  }

643 644 645 646
  void AddIterations(int n) {
    iterations_ += n;
  }

647 648 649 650
  void AddDeletes(int n) {
    deletes_ += n;
  }

A
Andres Noetzli 已提交
651 652
  void AddSingleDeletes(size_t n) { single_deletes_ += n; }

653 654 655 656 657 658 659 660
  void AddRangeDeletions(int n) {
    range_deletions_ += n;
  }

  void AddCoveredByRangeDeletions(int n) {
    covered_by_range_deletions_ += n;
  }

661 662
  void AddErrors(int n) {
    errors_ += n;
A
amayank 已提交
663 664
  }

665 666 667 668
  void AddNumCompactFilesSucceed(int n) { num_compact_files_succeed_ += n; }

  void AddNumCompactFilesFailed(int n) { num_compact_files_failed_ += n; }

669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685
  void Report(const char* name) {
    std::string extra;
    if (bytes_ < 1 || done_ < 1) {
      fprintf(stderr, "No writes or ops?\n");
      return;
    }

    double elapsed = (finish_ - start_) * 1e-6;
    double bytes_mb = bytes_ / 1048576.0;
    double rate = bytes_mb / elapsed;
    double throughput = (double)done_/elapsed;

    fprintf(stdout, "%-12s: ", name);
    fprintf(stdout, "%.3f micros/op %ld ops/sec\n",
            seconds_ * 1e6 / done_, (long)throughput);
    fprintf(stdout, "%-12s: Wrote %.2f MB (%.2f MB/sec) (%ld%% of %ld ops)\n",
            "", bytes_mb, rate, (100*writes_)/done_, done_);
686
    fprintf(stdout, "%-12s: Wrote %ld times\n", "", writes_);
A
amayank 已提交
687
    fprintf(stdout, "%-12s: Deleted %ld times\n", "", deletes_);
Y
yuslepukhin 已提交
688 689
    fprintf(stdout, "%-12s: Single deleted %" ROCKSDB_PRIszt " times\n", "",
           single_deletes_);
X
Xing Jin 已提交
690 691
    fprintf(stdout, "%-12s: %ld read and %ld found the key\n", "",
            gets_, founds_);
692 693 694
    fprintf(stdout, "%-12s: Prefix scanned %ld times\n", "", prefixes_);
    fprintf(stdout, "%-12s: Iterator size sum is %ld\n", "",
            iterator_size_sums_);
695
    fprintf(stdout, "%-12s: Iterated %ld times\n", "", iterations_);
696 697 698 699
    fprintf(stdout, "%-12s: Deleted %ld key-ranges\n", "", range_deletions_);
    fprintf(stdout, "%-12s: Range deletions covered %ld keys\n", "",
            covered_by_range_deletions_);

700
    fprintf(stdout, "%-12s: Got errors %ld times\n", "", errors_);
701 702
    fprintf(stdout, "%-12s: %ld CompactFiles() succeed\n", "",
            num_compact_files_succeed_);
I
Islam AbdelRahman 已提交
703
    fprintf(stdout, "%-12s: %ld CompactFiles() did not succeed\n", "",
704
            num_compact_files_failed_);
705 706 707 708 709 710 711 712 713 714 715

    if (FLAGS_histogram) {
      fprintf(stdout, "Microseconds per op:\n%s\n", hist_.ToString().c_str());
    }
    fflush(stdout);
  }
};

// State shared by all concurrent executions of the same benchmark.
class SharedState {
 public:
716
  static const uint32_t SENTINEL;
717

718 719
  explicit SharedState(StressTest* stress_test)
      : cv_(&mu_),
720
        seed_(static_cast<uint32_t>(FLAGS_seed)),
721
        max_key_(FLAGS_max_key),
722
        log2_keys_per_lock_(static_cast<uint32_t>(FLAGS_log2_keys_per_lock)),
723 724 725 726 727 728 729
        num_threads_(FLAGS_threads),
        num_initialized_(0),
        num_populated_(0),
        vote_reopen_(0),
        num_done_(0),
        start_(false),
        start_verify_(false),
730 731
        should_stop_bg_thread_(false),
        bg_thread_finished_(false),
732
        stress_test_(stress_test),
A
Andres Noetzli 已提交
733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751
        verification_failure_(false),
        no_overwrite_ids_(FLAGS_column_families) {
    // Pick random keys in each column family that will not experience
    // overwrite

    printf("Choosing random keys with no overwrite\n");
    Random rnd(seed_);
    size_t num_no_overwrite_keys = (max_key_ * FLAGS_nooverwritepercent) / 100;
    for (auto& cf_ids : no_overwrite_ids_) {
      for (size_t i = 0; i < num_no_overwrite_keys; i++) {
        size_t rand_key;
        do {
          rand_key = rnd.Next() % max_key_;
        } while (cf_ids.find(rand_key) != cf_ids.end());
        cf_ids.insert(rand_key);
      }
      assert(cf_ids.size() == num_no_overwrite_keys);
    }

752 753 754 755
    if (FLAGS_test_batches_snapshots) {
      fprintf(stdout, "No lock creation because test_batches_snapshots set\n");
      return;
    }
I
Igor Canadi 已提交
756 757 758 759
    values_.resize(FLAGS_column_families);

    for (int i = 0; i < FLAGS_column_families; ++i) {
      values_[i] = std::vector<uint32_t>(max_key_, SENTINEL);
760
    }
761

D
Dmitri Smirnov 已提交
762
    long num_locks = static_cast<long>(max_key_ >> log2_keys_per_lock_);
763
    if (max_key_ & ((1 << log2_keys_per_lock_) - 1)) {
I
Igor Canadi 已提交
764 765 766 767
      num_locks++;
    }
    fprintf(stdout, "Creating %ld locks\n", num_locks * FLAGS_column_families);
    key_locks_.resize(FLAGS_column_families);
D
Dmitri Smirnov 已提交
768

I
Igor Canadi 已提交
769
    for (int i = 0; i < FLAGS_column_families; ++i) {
D
Dmitri Smirnov 已提交
770 771 772 773
      key_locks_[i].resize(num_locks);
      for (auto& ptr : key_locks_[i]) {
        ptr.reset(new port::Mutex);
      }
774 775 776
    }
  }

I
Igor Canadi 已提交
777
  ~SharedState() {}
778 779 780 781 782 783 784 785 786 787 788 789 790

  port::Mutex* GetMutex() {
    return &mu_;
  }

  port::CondVar* GetCondVar() {
    return &cv_;
  }

  StressTest* GetStressTest() const {
    return stress_test_;
  }

D
Dmitri Smirnov 已提交
791
  int64_t GetMaxKey() const {
792 793 794 795 796 797 798 799 800 801 802
    return max_key_;
  }

  uint32_t GetNumThreads() const {
    return num_threads_;
  }

  void IncInitialized() {
    num_initialized_++;
  }

A
amayank 已提交
803
  void IncOperated() {
804 805 806 807 808 809 810
    num_populated_++;
  }

  void IncDone() {
    num_done_++;
  }

811 812 813 814
  void IncVotedReopen() {
    vote_reopen_ = (vote_reopen_ + 1) % num_threads_;
  }

815 816 817 818
  bool AllInitialized() const {
    return num_initialized_ >= num_threads_;
  }

A
amayank 已提交
819
  bool AllOperated() const {
820 821 822 823 824 825 826
    return num_populated_ >= num_threads_;
  }

  bool AllDone() const {
    return num_done_ >= num_threads_;
  }

827 828 829 830
  bool AllVotedReopen() {
    return (vote_reopen_ == 0);
  }

831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846
  void SetStart() {
    start_ = true;
  }

  void SetStartVerify() {
    start_verify_ = true;
  }

  bool Started() const {
    return start_;
  }

  bool VerifyStarted() const {
    return start_verify_;
  }

I
Igor Canadi 已提交
847 848 849 850
  void SetVerificationFailure() { verification_failure_.store(true); }

  bool HasVerificationFailedYet() { return verification_failure_.load(); }

I
Igor Canadi 已提交
851
  port::Mutex* GetMutexForKey(int cf, long key) {
D
Dmitri Smirnov 已提交
852
    return key_locks_[cf][key >> log2_keys_per_lock_].get();
853 854
  }

I
Igor Canadi 已提交
855 856
  void LockColumnFamily(int cf) {
    for (auto& mutex : key_locks_[cf]) {
D
Dmitri Smirnov 已提交
857
      mutex->Lock();
I
Igor Canadi 已提交
858
    }
859 860
  }

I
Igor Canadi 已提交
861 862
  void UnlockColumnFamily(int cf) {
    for (auto& mutex : key_locks_[cf]) {
D
Dmitri Smirnov 已提交
863
      mutex->Unlock();
I
Igor Canadi 已提交
864
    }
865 866
  }

I
Igor Canadi 已提交
867 868
  void ClearColumnFamily(int cf) {
    std::fill(values_[cf].begin(), values_[cf].end(), SENTINEL);
A
amayank 已提交
869 870
  }

D
Dmitri Smirnov 已提交
871
  void Put(int cf, int64_t key, uint32_t value_base) {
I
Igor Canadi 已提交
872
    values_[cf][key] = value_base;
873 874
  }

D
Dmitri Smirnov 已提交
875
  uint32_t Get(int cf, int64_t key) const { return values_[cf][key]; }
I
Igor Canadi 已提交
876

D
Dmitri Smirnov 已提交
877
  void Delete(int cf, int64_t key) { values_[cf][key] = SENTINEL; }
I
Igor Canadi 已提交
878

D
Dmitri Smirnov 已提交
879
  void SingleDelete(int cf, int64_t key) { values_[cf][key] = SENTINEL; }
A
Andres Noetzli 已提交
880

881 882 883 884 885 886 887 888 889 890 891
  int DeleteRange(int cf, int64_t begin_key, int64_t end_key) {
    int covered = 0;
    for (int64_t key = begin_key; key < end_key; ++key) {
      if (values_[cf][key] != SENTINEL) {
        ++covered;
      }
      values_[cf][key] = SENTINEL;
    }
    return covered;
  }

D
Dmitri Smirnov 已提交
892
  bool AllowsOverwrite(int cf, int64_t key) {
A
Andres Noetzli 已提交
893 894 895
    return no_overwrite_ids_[cf].find(key) == no_overwrite_ids_[cf].end();
  }

D
Dmitri Smirnov 已提交
896
  bool Exists(int cf, int64_t key) { return values_[cf][key] != SENTINEL; }
A
Andres Noetzli 已提交
897

I
Igor Canadi 已提交
898 899
  uint32_t GetSeed() const { return seed_; }

900 901 902 903 904 905 906 907
  void SetShouldStopBgThread() { should_stop_bg_thread_ = true; }

  bool ShoudStopBgThread() { return should_stop_bg_thread_; }

  void SetBgThreadFinish() { bg_thread_finished_ = true; }

  bool BgThreadFinished() const { return bg_thread_finished_; }

908 909 910 911
 private:
  port::Mutex mu_;
  port::CondVar cv_;
  const uint32_t seed_;
D
Dmitri Smirnov 已提交
912
  const int64_t max_key_;
913 914 915 916
  const uint32_t log2_keys_per_lock_;
  const int num_threads_;
  long num_initialized_;
  long num_populated_;
917
  long vote_reopen_;
918 919 920
  long num_done_;
  bool start_;
  bool start_verify_;
921 922
  bool should_stop_bg_thread_;
  bool bg_thread_finished_;
923
  StressTest* stress_test_;
I
Igor Canadi 已提交
924
  std::atomic<bool> verification_failure_;
925

A
Andres Noetzli 已提交
926 927 928
  // Keys that should not be overwritten
  std::vector<std::set<size_t> > no_overwrite_ids_;

I
Igor Canadi 已提交
929
  std::vector<std::vector<uint32_t>> values_;
D
Dmitri Smirnov 已提交
930 931
  // Has to make it owned by a smart ptr as port::Mutex is not copyable
  // and storing it in the container may require copying depending on the impl.
S
sdong 已提交
932
  std::vector<std::vector<std::unique_ptr<port::Mutex> > > key_locks_;
933 934
};

935 936
const uint32_t SharedState::SENTINEL = 0xffffffff;

937 938 939 940 941 942 943
// Per-thread state for concurrent executions of the same benchmark.
struct ThreadState {
  uint32_t tid; // 0..n-1
  Random rand;  // Has different seeds for different threads
  SharedState* shared;
  Stats stats;

I
Igor Canadi 已提交
944
  ThreadState(uint32_t index, SharedState* _shared)
I
Igor Canadi 已提交
945
      : tid(index), rand(1000 + index + _shared->GetSeed()), shared(_shared) {}
946 947
};

948 949
class DbStressListener : public EventListener {
 public:
A
Aaron Gao 已提交
950 951 952
  DbStressListener(const std::string& db_name,
                   const std::vector<DbPath>& db_paths)
      : db_name_(db_name), db_paths_(db_paths) {}
953 954
  virtual ~DbStressListener() {}
#ifndef ROCKSDB_LITE
A
Aaron Gao 已提交
955
  virtual void OnFlushCompleted(DB* db, const FlushJobInfo& info) override {
956 957
    assert(db);
    assert(db->GetName() == db_name_);
958 959
    assert(IsValidColumnFamilyName(info.cf_name));
    VerifyFilePath(info.file_path);
960 961
    // pretending doing some work here
    std::this_thread::sleep_for(
A
Aaron Gao 已提交
962
        std::chrono::microseconds(Random::GetTLSInstance()->Uniform(5000)));
963 964
  }

A
Aaron Gao 已提交
965 966
  virtual void OnCompactionCompleted(DB* db,
                                     const CompactionJobInfo& ci) override {
967 968 969 970 971 972 973 974 975 976 977 978
    assert(db);
    assert(db->GetName() == db_name_);
    assert(IsValidColumnFamilyName(ci.cf_name));
    assert(ci.input_files.size() + ci.output_files.size() > 0U);
    for (const auto& file_path : ci.input_files) {
      VerifyFilePath(file_path);
    }
    for (const auto& file_path : ci.output_files) {
      VerifyFilePath(file_path);
    }
    // pretending doing some work here
    std::this_thread::sleep_for(
A
Aaron Gao 已提交
979
        std::chrono::microseconds(Random::GetTLSInstance()->Uniform(5000)));
980 981
  }

A
Aaron Gao 已提交
982
  virtual void OnTableFileCreated(const TableFileCreationInfo& info) override {
983 984 985
    assert(info.db_name == db_name_);
    assert(IsValidColumnFamilyName(info.cf_name));
    VerifyFilePath(info.file_path);
986
    assert(info.job_id > 0 || FLAGS_compact_files_one_in > 0);
987 988 989 990 991 992
    if (info.status.ok()) {
      assert(info.file_size > 0);
      assert(info.table_properties.data_size > 0);
      assert(info.table_properties.raw_key_size > 0);
      assert(info.table_properties.num_entries > 0);
    }
993 994
  }

995 996 997 998 999
 protected:
  bool IsValidColumnFamilyName(const std::string& cf_name) const {
    if (cf_name == kDefaultColumnFamilyName) {
      return true;
    }
Y
Yueh-Hsuan Chiang 已提交
1000 1001 1002 1003
    // The column family names in the stress tests are numbers.
    for (size_t i = 0; i < cf_name.size(); ++i) {
      if (cf_name[i] < '0' || cf_name[i] > '9') {
        return false;
1004 1005
      }
    }
Y
Yueh-Hsuan Chiang 已提交
1006
    return true;
1007 1008 1009
  }

  void VerifyFileDir(const std::string& file_dir) {
1010
#ifndef NDEBUG
1011 1012 1013 1014 1015 1016 1017 1018 1019
    if (db_name_ == file_dir) {
      return;
    }
    for (const auto& db_path : db_paths_) {
      if (db_path.path == file_dir) {
        return;
      }
    }
    assert(false);
1020
#endif  // !NDEBUG
1021 1022 1023
  }

  void VerifyFileName(const std::string& file_name) {
1024
#ifndef NDEBUG
1025 1026 1027 1028 1029
    uint64_t file_number;
    FileType file_type;
    bool result = ParseFileName(file_name, &file_number, &file_type);
    assert(result);
    assert(file_type == kTableFile);
1030
#endif  // !NDEBUG
1031 1032 1033
  }

  void VerifyFilePath(const std::string& file_path) {
1034
#ifndef NDEBUG
1035 1036 1037 1038 1039 1040 1041 1042 1043
    size_t pos = file_path.find_last_of("/");
    if (pos == std::string::npos) {
      VerifyFileName(file_path);
    } else {
      if (pos > 0) {
        VerifyFileDir(file_path.substr(0, pos));
      }
      VerifyFileName(file_path.substr(pos));
    }
1044
#endif  // !NDEBUG
1045 1046 1047 1048 1049 1050 1051 1052
  }
#endif  // !ROCKSDB_LITE

 private:
  std::string db_name_;
  std::vector<DbPath> db_paths_;
};

1053 1054 1055 1056 1057
}  // namespace

class StressTest {
 public:
  StressTest()
Y
Yi Wu 已提交
1058 1059
      : cache_(NewCache(FLAGS_cache_size)),
        compressed_cache_(NewLRUCache(FLAGS_compressed_cache_size)),
1060
        filter_policy_(FLAGS_bloom_bits >= 0
Y
Yi Wu 已提交
1061 1062 1063 1064
                           ? FLAGS_use_block_based_filter
                                 ? NewBloomFilterPolicy(FLAGS_bloom_bits, true)
                                 : NewBloomFilterPolicy(FLAGS_bloom_bits, false)
                           : nullptr),
1065
        db_(nullptr),
1066
        new_column_family_name_(1),
1067
        num_times_reopened_(0) {
1068 1069 1070 1071 1072
    if (FLAGS_destroy_db_initially) {
      std::vector<std::string> files;
      FLAGS_env->GetChildren(FLAGS_db, &files);
      for (unsigned int i = 0; i < files.size(); i++) {
        if (Slice(files[i]).starts_with("heap-")) {
1073
          FLAGS_env->DeleteFile(FLAGS_db + "/" + files[i]);
1074
        }
1075
      }
1076
      DestroyDB(FLAGS_db, Options());
1077 1078 1079 1080
    }
  }

  ~StressTest() {
I
Igor Canadi 已提交
1081 1082 1083 1084
    for (auto cf : column_families_) {
      delete cf;
    }
    column_families_.clear();
1085 1086 1087
    delete db_;
  }

Y
Yi Wu 已提交
1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103
  std::shared_ptr<Cache> NewCache(size_t capacity) {
    if (capacity <= 0) {
      return nullptr;
    }
    if (FLAGS_use_clock_cache) {
      auto cache = NewClockCache((size_t)capacity);
      if (!cache) {
        fprintf(stderr, "Clock cache not supported.");
        exit(1);
      }
      return cache;
    } else {
      return NewLRUCache((size_t)capacity);
    }
  }

L
Lei Jin 已提交
1104 1105 1106 1107
  bool BuildOptionsTable() {
    if (FLAGS_set_options_one_in <= 0) {
      return true;
    }
D
Dmitri Smirnov 已提交
1108

S
sdong 已提交
1109 1110 1111
    std::unordered_map<std::string, std::vector<std::string> > options_tbl = {
        {"write_buffer_size",
         {ToString(FLAGS_write_buffer_size),
1112
          ToString(FLAGS_write_buffer_size * 2),
S
sdong 已提交
1113 1114 1115
          ToString(FLAGS_write_buffer_size * 4)}},
        {"max_write_buffer_number",
         {ToString(FLAGS_max_write_buffer_number),
1116
          ToString(FLAGS_max_write_buffer_number * 2),
S
sdong 已提交
1117 1118 1119 1120 1121 1122 1123 1124 1125
          ToString(FLAGS_max_write_buffer_number * 4)}},
        {"arena_block_size",
         {
             ToString(Options().arena_block_size),
             ToString(FLAGS_write_buffer_size / 4),
             ToString(FLAGS_write_buffer_size / 8),
         }},
        {"memtable_prefix_bloom_bits", {"0", "8", "10"}},
        {"memtable_prefix_bloom_probes", {"4", "5", "6"}},
1126
        {"memtable_huge_page_size", {"0", ToString(2 * 1024 * 1024)}},
S
sdong 已提交
1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150
        {"max_successive_merges", {"0", "2", "4"}},
        {"inplace_update_num_locks", {"100", "200", "300"}},
        // TODO(ljin): enable test for this option
        // {"disable_auto_compactions", {"100", "200", "300"}},
        {"soft_rate_limit", {"0", "0.5", "0.9"}},
        {"hard_rate_limit", {"0", "1.1", "2.0"}},
        {"level0_file_num_compaction_trigger",
         {
             ToString(FLAGS_level0_file_num_compaction_trigger),
             ToString(FLAGS_level0_file_num_compaction_trigger + 2),
             ToString(FLAGS_level0_file_num_compaction_trigger + 4),
         }},
        {"level0_slowdown_writes_trigger",
         {
             ToString(FLAGS_level0_slowdown_writes_trigger),
             ToString(FLAGS_level0_slowdown_writes_trigger + 2),
             ToString(FLAGS_level0_slowdown_writes_trigger + 4),
         }},
        {"level0_stop_writes_trigger",
         {
             ToString(FLAGS_level0_stop_writes_trigger),
             ToString(FLAGS_level0_stop_writes_trigger + 2),
             ToString(FLAGS_level0_stop_writes_trigger + 4),
         }},
1151
        {"max_compaction_bytes",
S
sdong 已提交
1152
         {
1153 1154 1155
             ToString(FLAGS_target_file_size_base * 5),
             ToString(FLAGS_target_file_size_base * 15),
             ToString(FLAGS_target_file_size_base * 100),
S
sdong 已提交
1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177
         }},
        {"target_file_size_base",
         {
             ToString(FLAGS_target_file_size_base),
             ToString(FLAGS_target_file_size_base * 2),
             ToString(FLAGS_target_file_size_base * 4),
         }},
        {"target_file_size_multiplier",
         {
             ToString(FLAGS_target_file_size_multiplier), "1", "2",
         }},
        {"max_bytes_for_level_base",
         {
             ToString(FLAGS_max_bytes_for_level_base / 2),
             ToString(FLAGS_max_bytes_for_level_base),
             ToString(FLAGS_max_bytes_for_level_base * 2),
         }},
        {"max_bytes_for_level_multiplier",
         {
             ToString(FLAGS_max_bytes_for_level_multiplier), "1", "2",
         }},
        {"max_sequential_skip_in_iterations", {"4", "8", "12"}},
1178 1179
        {"use_direct_reads", {"false", "true"}},
        {"use_direct_io_for_flush_and_compaction", {"false", "true"}},
L
Lei Jin 已提交
1180
    };
D
Dmitri Smirnov 已提交
1181 1182 1183

    options_table_ = std::move(options_tbl);

L
Lei Jin 已提交
1184 1185 1186 1187 1188 1189
    for (const auto& iter : options_table_) {
      options_index_.push_back(iter.first);
    }
    return true;
  }

I
Igor Canadi 已提交
1190
  bool Run() {
1191
    PrintEnv();
L
Lei Jin 已提交
1192
    BuildOptionsTable();
1193 1194 1195 1196 1197 1198 1199 1200 1201
    Open();
    SharedState shared(this);
    uint32_t n = shared.GetNumThreads();

    std::vector<ThreadState*> threads(n);
    for (uint32_t i = 0; i < n; i++) {
      threads[i] = new ThreadState(i, &shared);
      FLAGS_env->StartThread(ThreadBody, threads[i]);
    }
1202 1203 1204 1205 1206
    ThreadState bg_thread(0, &shared);
    if (FLAGS_compaction_thread_pool_adjust_interval > 0) {
      FLAGS_env->StartThread(PoolSizeChangeThread, &bg_thread);
    }

1207
    // Each thread goes through the following states:
A
amayank 已提交
1208 1209
    // initializing -> wait for others to init -> read/populate/depopulate
    // wait for others to operate -> verify -> done
1210 1211 1212 1213 1214 1215 1216

    {
      MutexLock l(shared.GetMutex());
      while (!shared.AllInitialized()) {
        shared.GetCondVar()->Wait();
      }

D
Dmitri Smirnov 已提交
1217
      auto now = FLAGS_env->NowMicros();
1218
      fprintf(stdout, "%s Starting database operations\n",
D
Dmitri Smirnov 已提交
1219
              FLAGS_env->TimeToString(now/1000000).c_str());
A
amayank 已提交
1220

1221 1222
      shared.SetStart();
      shared.GetCondVar()->SignalAll();
A
amayank 已提交
1223
      while (!shared.AllOperated()) {
1224 1225 1226
        shared.GetCondVar()->Wait();
      }

1227
      now = FLAGS_env->NowMicros();
1228 1229 1230 1231 1232 1233 1234
      if (FLAGS_test_batches_snapshots) {
        fprintf(stdout, "%s Limited verification already done during gets\n",
                FLAGS_env->TimeToString((uint64_t) now/1000000).c_str());
      } else {
        fprintf(stdout, "%s Starting verification\n",
                FLAGS_env->TimeToString((uint64_t) now/1000000).c_str());
      }
1235

1236 1237 1238 1239 1240 1241 1242
      shared.SetStartVerify();
      shared.GetCondVar()->SignalAll();
      while (!shared.AllDone()) {
        shared.GetCondVar()->Wait();
      }
    }

1243
    for (unsigned int i = 1; i < n; i++) {
1244 1245 1246 1247
      threads[0]->stats.Merge(threads[i]->stats);
    }
    threads[0]->stats.Report("Stress Test");

1248
    for (unsigned int i = 0; i < n; i++) {
1249
      delete threads[i];
1250
      threads[i] = nullptr;
1251
    }
D
Dmitri Smirnov 已提交
1252
    auto now = FLAGS_env->NowMicros();
1253 1254
    if (!FLAGS_test_batches_snapshots) {
      fprintf(stdout, "%s Verification successful\n",
D
Dmitri Smirnov 已提交
1255
              FLAGS_env->TimeToString(now/1000000).c_str());
1256
    }
1257
    PrintStatistics();
I
Igor Canadi 已提交
1258

1259 1260 1261 1262 1263 1264 1265 1266
    if (FLAGS_compaction_thread_pool_adjust_interval > 0) {
      MutexLock l(shared.GetMutex());
      shared.SetShouldStopBgThread();
      while (!shared.BgThreadFinished()) {
        shared.GetCondVar()->Wait();
      }
    }

I
Igor Canadi 已提交
1267 1268 1269 1270 1271
    if (shared.HasVerificationFailedYet()) {
      printf("Verification failed :(\n");
      return false;
    }
    return true;
1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289
  }

 private:

  static void ThreadBody(void* v) {
    ThreadState* thread = reinterpret_cast<ThreadState*>(v);
    SharedState* shared = thread->shared;

    {
      MutexLock l(shared->GetMutex());
      shared->IncInitialized();
      if (shared->AllInitialized()) {
        shared->GetCondVar()->SignalAll();
      }
      while (!shared->Started()) {
        shared->GetCondVar()->Wait();
      }
    }
A
amayank 已提交
1290
    thread->shared->GetStressTest()->OperateDb(thread);
1291 1292 1293

    {
      MutexLock l(shared->GetMutex());
A
amayank 已提交
1294 1295
      shared->IncOperated();
      if (shared->AllOperated()) {
1296 1297 1298 1299 1300 1301 1302
        shared->GetCondVar()->SignalAll();
      }
      while (!shared->VerifyStarted()) {
        shared->GetCondVar()->Wait();
      }
    }

1303
    if (!FLAGS_test_batches_snapshots) {
M
Mayank Agarwal 已提交
1304
      thread->shared->GetStressTest()->VerifyDb(thread);
1305
    }
1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316

    {
      MutexLock l(shared->GetMutex());
      shared->IncDone();
      if (shared->AllDone()) {
        shared->GetCondVar()->SignalAll();
      }
    }

  }

1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332
  static void PoolSizeChangeThread(void* v) {
    assert(FLAGS_compaction_thread_pool_adjust_interval > 0);
    ThreadState* thread = reinterpret_cast<ThreadState*>(v);
    SharedState* shared = thread->shared;

    while (true) {
      {
        MutexLock l(shared->GetMutex());
        if (shared->ShoudStopBgThread()) {
          shared->SetBgThreadFinish();
          shared->GetCondVar()->SignalAll();
          return;
        }
      }

      auto thread_pool_size_base = FLAGS_max_background_compactions;
I
Igor Canadi 已提交
1333
      auto thread_pool_size_var = FLAGS_compaction_thread_pool_variations;
1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348
      int new_thread_pool_size =
          thread_pool_size_base - thread_pool_size_var +
          thread->rand.Next() % (thread_pool_size_var * 2 + 1);
      if (new_thread_pool_size < 1) {
        new_thread_pool_size = 1;
      }
      FLAGS_env->SetBackgroundThreads(new_thread_pool_size);
      // Sleep up to 3 seconds
      FLAGS_env->SleepForMicroseconds(
          thread->rand.Next() % FLAGS_compaction_thread_pool_adjust_interval *
              1000 +
          1);
    }
  }

1349 1350 1351
  // Given a key K and value V, this puts ("0"+K, "0"+V), ("1"+K, "1"+V), ...
  // ("9"+K, "9"+V) in DB atomically i.e in a single batch.
  // Also refer MultiGet.
I
Igor Canadi 已提交
1352 1353 1354
  Status MultiPut(ThreadState* thread, const WriteOptions& writeoptions,
                  ColumnFamilyHandle* column_family, const Slice& key,
                  const Slice& value, size_t sz) {
1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365
    std::string keys[10] = {"9", "8", "7", "6", "5",
                            "4", "3", "2", "1", "0"};
    std::string values[10] = {"9", "8", "7", "6", "5",
                              "4", "3", "2", "1", "0"};
    Slice value_slices[10];
    WriteBatch batch;
    Status s;
    for (int i = 0; i < 10; i++) {
      keys[i] += key.ToString();
      values[i] += value.ToString();
      value_slices[i] = values[i];
1366
      if (FLAGS_use_merge) {
I
Igor Canadi 已提交
1367
        batch.Merge(column_family, keys[i], value_slices[i]);
D
Deon Nicholas 已提交
1368
      } else {
I
Igor Canadi 已提交
1369
        batch.Put(column_family, keys[i], value_slices[i]);
D
Deon Nicholas 已提交
1370
      }
1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386
    }

    s = db_->Write(writeoptions, &batch);
    if (!s.ok()) {
      fprintf(stderr, "multiput error: %s\n", s.ToString().c_str());
      thread->stats.AddErrors(1);
    } else {
      // we did 10 writes each of size sz + 1
      thread->stats.AddBytesForWrites(10, (sz + 1) * 10);
    }

    return s;
  }

  // Given a key K, this deletes ("0"+K), ("1"+K),... ("9"+K)
  // in DB atomically i.e in a single batch. Also refer MultiGet.
I
Igor Canadi 已提交
1387 1388
  Status MultiDelete(ThreadState* thread, const WriteOptions& writeoptions,
                     ColumnFamilyHandle* column_family, const Slice& key) {
1389 1390 1391 1392 1393 1394 1395
    std::string keys[10] = {"9", "7", "5", "3", "1",
                            "8", "6", "4", "2", "0"};

    WriteBatch batch;
    Status s;
    for (int i = 0; i < 10; i++) {
      keys[i] += key.ToString();
I
Igor Canadi 已提交
1396
      batch.Delete(column_family, keys[i]);
1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413
    }

    s = db_->Write(writeoptions, &batch);
    if (!s.ok()) {
      fprintf(stderr, "multidelete error: %s\n", s.ToString().c_str());
      thread->stats.AddErrors(1);
    } else {
      thread->stats.AddDeletes(10);
    }

    return s;
  }

  // Given a key K, this gets values for "0"+K, "1"+K,..."9"+K
  // in the same snapshot, and verifies that all the values are of the form
  // "0"+V, "1"+V,..."9"+V.
  // ASSUMES that MultiPut was used to put (K, V) into the DB.
I
Igor Canadi 已提交
1414 1415 1416
  Status MultiGet(ThreadState* thread, const ReadOptions& readoptions,
                  ColumnFamilyHandle* column_family, const Slice& key,
                  std::string* value) {
1417 1418 1419 1420 1421 1422 1423 1424 1425
    std::string keys[10] = {"0", "1", "2", "3", "4", "5", "6", "7", "8", "9"};
    Slice key_slices[10];
    std::string values[10];
    ReadOptions readoptionscopy = readoptions;
    readoptionscopy.snapshot = db_->GetSnapshot();
    Status s;
    for (int i = 0; i < 10; i++) {
      keys[i] += key.ToString();
      key_slices[i] = keys[i];
I
Igor Canadi 已提交
1426
      s = db_->Get(readoptionscopy, column_family, key_slices[i], value);
1427 1428 1429 1430 1431 1432 1433 1434
      if (!s.ok() && !s.IsNotFound()) {
        fprintf(stderr, "get error: %s\n", s.ToString().c_str());
        values[i] = "";
        thread->stats.AddErrors(1);
        // we continue after error rather than exiting so that we can
        // find more errors if any
      } else if (s.IsNotFound()) {
        values[i] = "";
1435
        thread->stats.AddGets(1, 0);
1436 1437 1438 1439 1440 1441
      } else {
        values[i] = *value;

        char expected_prefix = (keys[i])[0];
        char actual_prefix = (values[i])[0];
        if (actual_prefix != expected_prefix) {
1442
          fprintf(stderr, "error expected prefix = %c actual = %c\n",
1443 1444 1445
                  expected_prefix, actual_prefix);
        }
        (values[i])[0] = ' '; // blank out the differing character
1446
        thread->stats.AddGets(1, 1);
1447 1448 1449 1450 1451 1452 1453
      }
    }
    db_->ReleaseSnapshot(readoptionscopy.snapshot);

    // Now that we retrieved all values, check that they all match
    for (int i = 1; i < 10; i++) {
      if (values[i] != values[0]) {
1454
        fprintf(stderr, "error : inconsistent values for key %s: %s, %s\n",
1455 1456
                key.ToString(true).c_str(), StringToHex(values[0]).c_str(),
                StringToHex(values[i]).c_str());
1457 1458 1459 1460 1461 1462 1463 1464
      // we continue after error rather than exiting so that we can
      // find more errors if any
      }
    }

    return s;
  }

L
Lei Jin 已提交
1465 1466 1467 1468 1469
  // Given a key, this does prefix scans for "0"+P, "1"+P,..."9"+P
  // in the same snapshot where P is the first FLAGS_prefix_size - 1 bytes
  // of the key. Each of these 10 scans returns a series of values;
  // each series should be the same length, and it is verified for each
  // index i that all the i'th values are of the form "0"+V, "1"+V,..."9"+V.
1470
  // ASSUMES that MultiPut was used to put (K, V)
I
Igor Canadi 已提交
1471 1472
  Status MultiPrefixScan(ThreadState* thread, const ReadOptions& readoptions,
                         ColumnFamilyHandle* column_family,
L
Lei Jin 已提交
1473
                         const Slice& key) {
1474 1475 1476 1477 1478 1479 1480 1481
    std::string prefixes[10] = {"0", "1", "2", "3", "4",
                                "5", "6", "7", "8", "9"};
    Slice prefix_slices[10];
    ReadOptions readoptionscopy[10];
    const Snapshot* snapshot = db_->GetSnapshot();
    Iterator* iters[10];
    Status s = Status::OK();
    for (int i = 0; i < 10; i++) {
L
Lei Jin 已提交
1482 1483 1484
      prefixes[i] += key.ToString();
      prefixes[i].resize(FLAGS_prefix_size);
      prefix_slices[i] = Slice(prefixes[i]);
1485 1486
      readoptionscopy[i] = readoptions;
      readoptionscopy[i].snapshot = snapshot;
I
Igor Canadi 已提交
1487
      iters[i] = db_->NewIterator(readoptionscopy[i], column_family);
I
Igor Canadi 已提交
1488
      iters[i]->Seek(prefix_slices[i]);
1489 1490 1491
    }

    int count = 0;
1492
    while (iters[0]->Valid() && iters[0]->key().starts_with(prefix_slices[0])) {
1493 1494 1495 1496 1497
      count++;
      std::string values[10];
      // get list of all values for this iteration
      for (int i = 0; i < 10; i++) {
        // no iterator should finish before the first one
1498 1499
        assert(iters[i]->Valid() &&
               iters[i]->key().starts_with(prefix_slices[i]));
1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513
        values[i] = iters[i]->value().ToString();

        char expected_first = (prefixes[i])[0];
        char actual_first = (values[i])[0];

        if (actual_first != expected_first) {
          fprintf(stderr, "error expected first = %c actual = %c\n",
                  expected_first, actual_first);
        }
        (values[i])[0] = ' '; // blank out the differing character
      }
      // make sure all values are equivalent
      for (int i = 0; i < 10; i++) {
        if (values[i] != values[0]) {
1514 1515 1516
          fprintf(stderr, "error : %d, inconsistent values for prefix %s: %s, %s\n",
                  i, prefixes[i].c_str(), StringToHex(values[0]).c_str(),
                  StringToHex(values[i]).c_str());
1517 1518 1519 1520 1521 1522 1523 1524 1525 1526
          // we continue after error rather than exiting so that we can
          // find more errors if any
        }
        iters[i]->Next();
      }
    }

    // cleanup iterators and snapshot
    for (int i = 0; i < 10; i++) {
      // if the first iterator finished, they should have all finished
1527 1528
      assert(!iters[i]->Valid() ||
             !iters[i]->key().starts_with(prefix_slices[i]));
1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542
      assert(iters[i]->status().ok());
      delete iters[i];
    }
    db_->ReleaseSnapshot(snapshot);

    if (s.ok()) {
      thread->stats.AddPrefixes(1, count);
    } else {
      thread->stats.AddErrors(1);
    }

    return s;
  }

1543 1544
  // Given a key K, this creates an iterator which scans to K and then
  // does a random sequence of Next/Prev operations.
I
Igor Canadi 已提交
1545 1546
  Status MultiIterate(ThreadState* thread, const ReadOptions& readoptions,
                      ColumnFamilyHandle* column_family, const Slice& key) {
1547 1548 1549 1550
    Status s;
    const Snapshot* snapshot = db_->GetSnapshot();
    ReadOptions readoptionscopy = readoptions;
    readoptionscopy.snapshot = snapshot;
I
Igor Canadi 已提交
1551
    unique_ptr<Iterator> iter(db_->NewIterator(readoptionscopy, column_family));
1552 1553

    iter->Seek(key);
1554
    for (uint64_t i = 0; i < FLAGS_num_iterations && iter->Valid(); i++) {
1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572
      if (thread->rand.OneIn(2)) {
        iter->Next();
      } else {
        iter->Prev();
      }
    }

    if (s.ok()) {
      thread->stats.AddIterations(1);
    } else {
      thread->stats.AddErrors(1);
    }

    db_->ReleaseSnapshot(snapshot);

    return s;
  }

1573
  Status SetOptions(ThreadState* thread) {
L
Lei Jin 已提交
1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599
    assert(FLAGS_set_options_one_in > 0);
    std::unordered_map<std::string, std::string> opts;
    std::string name = options_index_[
      thread->rand.Next() % options_index_.size()];
    int value_idx = thread->rand.Next() % options_table_[name].size();
    if (name == "soft_rate_limit" || name == "hard_rate_limit") {
      opts["soft_rate_limit"] = options_table_["soft_rate_limit"][value_idx];
      opts["hard_rate_limit"] = options_table_["hard_rate_limit"][value_idx];
    } else if (name == "level0_file_num_compaction_trigger" ||
               name == "level0_slowdown_writes_trigger" ||
               name == "level0_stop_writes_trigger") {
      opts["level0_file_num_compaction_trigger"] =
        options_table_["level0_file_num_compaction_trigger"][value_idx];
      opts["level0_slowdown_writes_trigger"] =
        options_table_["level0_slowdown_writes_trigger"][value_idx];
      opts["level0_stop_writes_trigger"] =
        options_table_["level0_stop_writes_trigger"][value_idx];
    } else {
      opts[name] = options_table_[name][value_idx];
    }

    int rand_cf_idx = thread->rand.Next() % FLAGS_column_families;
    auto cfh = column_families_[rand_cf_idx];
    return db_->SetOptions(cfh, opts);
  }

A
amayank 已提交
1600
  void OperateDb(ThreadState* thread) {
1601 1602
    ReadOptions read_opts(FLAGS_verify_checksum, true);
    WriteOptions write_opts;
A
Andres Noetzli 已提交
1603
    auto shared = thread->shared;
A
amayank 已提交
1604
    char value[100];
D
Dmitri Smirnov 已提交
1605
    auto max_key = thread->shared->GetMaxKey();
1606 1607 1608 1609 1610
    std::string from_db;
    if (FLAGS_sync) {
      write_opts.sync = true;
    }
    write_opts.disableWAL = FLAGS_disable_wal;
1611 1612 1613
    const int prefixBound = (int)FLAGS_readpercent + (int)FLAGS_prefixpercent;
    const int writeBound = prefixBound + (int)FLAGS_writepercent;
    const int delBound = writeBound + (int)FLAGS_delpercent;
1614
    const int delRangeBound = delBound + (int)FLAGS_delrangepercent;
1615 1616

    thread->stats.Start();
1617
    for (uint64_t i = 0; i < FLAGS_ops_per_thread; i++) {
I
Igor Canadi 已提交
1618 1619 1620 1621
      if (thread->shared->HasVerificationFailedYet()) {
        break;
      }
      if (i != 0 && (i % (FLAGS_ops_per_thread / (FLAGS_reopen + 1))) == 0) {
1622
        {
1623
          thread->stats.FinishedSingleOp();
1624 1625 1626 1627 1628 1629 1630 1631 1632
          MutexLock l(thread->shared->GetMutex());
          thread->shared->IncVotedReopen();
          if (thread->shared->AllVotedReopen()) {
            thread->shared->GetStressTest()->Reopen();
            thread->shared->GetCondVar()->SignalAll();
          }
          else {
            thread->shared->GetCondVar()->Wait();
          }
1633 1634
          // Commenting this out as we don't want to reset stats on each open.
          // thread->stats.Start();
1635 1636
        }
      }
1637

L
Lei Jin 已提交
1638 1639 1640 1641 1642 1643
      // Change Options
      if (FLAGS_set_options_one_in > 0 &&
          thread->rand.OneIn(FLAGS_set_options_one_in)) {
        SetOptions(thread);
      }

1644 1645 1646 1647 1648
      if (FLAGS_set_in_place_one_in > 0 &&
          thread->rand.OneIn(FLAGS_set_in_place_one_in)) {
        options_.inplace_update_support ^= options_.inplace_update_support;
      }

I
Igor Canadi 已提交
1649
      if (!FLAGS_test_batches_snapshots &&
1650
          FLAGS_clear_column_family_one_in != 0 && FLAGS_column_families > 1) {
I
Igor Canadi 已提交
1651 1652 1653 1654
        if (thread->rand.OneIn(FLAGS_clear_column_family_one_in)) {
          // drop column family and then create it again (can't drop default)
          int cf = thread->rand.Next() % (FLAGS_column_families - 1) + 1;
          std::string new_name =
1655
              ToString(new_column_family_name_.fetch_add(1));
I
Igor Canadi 已提交
1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666
          {
            MutexLock l(thread->shared->GetMutex());
            fprintf(
                stdout,
                "[CF %d] Dropping and recreating column family. new name: %s\n",
                cf, new_name.c_str());
          }
          thread->shared->LockColumnFamily(cf);
          Status s __attribute__((unused));
          s = db_->DropColumnFamily(column_families_[cf]);
          delete column_families_[cf];
1667 1668 1669 1670 1671
          if (!s.ok()) {
            fprintf(stderr, "dropping column family error: %s\n",
                s.ToString().c_str());
            std::terminate();
          }
I
Igor Canadi 已提交
1672 1673 1674 1675
          s = db_->CreateColumnFamily(ColumnFamilyOptions(options_), new_name,
                                      &column_families_[cf]);
          column_family_names_[cf] = new_name;
          thread->shared->ClearColumnFamily(cf);
1676 1677 1678 1679 1680
          if (!s.ok()) {
            fprintf(stderr, "creating column family error: %s\n",
                s.ToString().c_str());
            std::terminate();
          }
I
Igor Canadi 已提交
1681 1682 1683 1684
          thread->shared->UnlockColumnFamily(cf);
        }
      }

1685
#ifndef ROCKSDB_LITE  // Lite does not support GetColumnFamilyMetaData
1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734
      if (FLAGS_compact_files_one_in > 0 &&
          thread->rand.Uniform(FLAGS_compact_files_one_in) == 0) {
        auto* random_cf =
            column_families_[thread->rand.Next() % FLAGS_column_families];
        rocksdb::ColumnFamilyMetaData cf_meta_data;
        db_->GetColumnFamilyMetaData(random_cf, &cf_meta_data);

        // Randomly compact up to three consecutive files from a level
        const int kMaxRetry = 3;
        for (int attempt = 0; attempt < kMaxRetry; ++attempt) {
          size_t random_level = thread->rand.Uniform(
              static_cast<int>(cf_meta_data.levels.size()));

          const auto& files = cf_meta_data.levels[random_level].files;
          if (files.size() > 0) {
            size_t random_file_index =
                thread->rand.Uniform(static_cast<int>(files.size()));
            if (files[random_file_index].being_compacted) {
              // Retry as the selected file is currently being compacted
              continue;
            }

            std::vector<std::string> input_files;
            input_files.push_back(files[random_file_index].name);
            if (random_file_index > 0 &&
                !files[random_file_index - 1].being_compacted) {
              input_files.push_back(files[random_file_index - 1].name);
            }
            if (random_file_index + 1 < files.size() &&
                !files[random_file_index + 1].being_compacted) {
              input_files.push_back(files[random_file_index + 1].name);
            }

            size_t output_level =
                std::min(random_level + 1, cf_meta_data.levels.size() - 1);
            auto s =
                db_->CompactFiles(CompactionOptions(), random_cf, input_files,
                                  static_cast<int>(output_level));
            if (!s.ok()) {
              printf("Unable to perform CompactFiles(): %s\n",
                     s.ToString().c_str());
              thread->stats.AddNumCompactFilesFailed(1);
            } else {
              thread->stats.AddNumCompactFilesSucceed(1);
            }
            break;
          }
        }
      }
1735
#endif                // !ROCKSDB_LITE
1736

A
Andrew Kryczka 已提交
1737 1738 1739 1740 1741
      const double completed_ratio =
          static_cast<double>(i) / FLAGS_ops_per_thread;
      const int64_t base_key = static_cast<int64_t>(
          completed_ratio * (FLAGS_max_key - FLAGS_active_width));
      long rand_key = base_key + thread->rand.Next() % FLAGS_active_width;
I
Igor Canadi 已提交
1742
      int rand_column_family = thread->rand.Next() % FLAGS_column_families;
1743 1744
      std::string keystr = Key(rand_key);
      Slice key = keystr;
I
Igor Canadi 已提交
1745 1746 1747
      std::unique_ptr<MutexLock> l;
      if (!FLAGS_test_batches_snapshots) {
        l.reset(new MutexLock(
A
Andres Noetzli 已提交
1748
            shared->GetMutexForKey(rand_column_family, rand_key)));
I
Igor Canadi 已提交
1749 1750
      }
      auto column_family = column_families_[rand_column_family];
1751

A
Andres Noetzli 已提交
1752
      int prob_op = thread->rand.Uniform(100);
1753
      if (prob_op >= 0 && prob_op < (int)FLAGS_readpercent) {
1754
        // OPERATION read
1755
        if (!FLAGS_test_batches_snapshots) {
I
Igor Canadi 已提交
1756
          Status s = db_->Get(read_opts, column_family, key, &from_db);
1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767
          if (s.ok()) {
            // found case
            thread->stats.AddGets(1, 1);
          } else if (s.IsNotFound()) {
            // not found case
            thread->stats.AddGets(1, 0);
          } else {
            // errors case
            thread->stats.AddErrors(1);
          }
        } else {
I
Igor Canadi 已提交
1768
          MultiGet(thread, read_opts, column_family, key, &from_db);
1769
        }
1770 1771
      } else if ((int)FLAGS_readpercent <= prob_op && prob_op < prefixBound) {
        // OPERATION prefix scan
L
Lei Jin 已提交
1772 1773
        // keys are 8 bytes long, prefix size is FLAGS_prefix_size. There are
        // (8 - FLAGS_prefix_size) bytes besides the prefix. So there will
I
Igor Canadi 已提交
1774 1775
        // be 2 ^ ((8 - FLAGS_prefix_size) * 8) possible keys with the same
        // prefix
1776
        if (!FLAGS_test_batches_snapshots) {
L
Lei Jin 已提交
1777
          Slice prefix = Slice(key.data(), FLAGS_prefix_size);
I
Igor Canadi 已提交
1778
          Iterator* iter = db_->NewIterator(read_opts, column_family);
L
Lei Jin 已提交
1779
          int64_t count = 0;
I
Igor Canadi 已提交
1780 1781
          for (iter->Seek(prefix);
               iter->Valid() && iter->key().starts_with(prefix); iter->Next()) {
L
Lei Jin 已提交
1782
            ++count;
1783
          }
I
Igor Canadi 已提交
1784 1785
          assert(count <=
                 (static_cast<int64_t>(1) << ((8 - FLAGS_prefix_size) * 8)));
1786
          if (iter->status().ok()) {
1787
            thread->stats.AddPrefixes(1, static_cast<int>(count));
1788 1789 1790 1791
          } else {
            thread->stats.AddErrors(1);
          }
          delete iter;
1792
        } else {
1793
          MultiPrefixScan(thread, read_opts, column_family, key);
A
amayank 已提交
1794
        }
1795 1796
      } else if (prefixBound <= prob_op && prob_op < writeBound) {
        // OPERATION write
1797 1798 1799
        uint32_t value_base = thread->rand.Next();
        size_t sz = GenerateValue(value_base, value, sizeof(value));
        Slice v(value, sz);
1800
        if (!FLAGS_test_batches_snapshots) {
A
Andres Noetzli 已提交
1801 1802 1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815
          // If the chosen key does not allow overwrite and it already
          // exists, choose another key.
          while (!shared->AllowsOverwrite(rand_column_family, rand_key) &&
                 shared->Exists(rand_column_family, rand_key)) {
            l.reset();
            rand_key = thread->rand.Next() % max_key;
            rand_column_family = thread->rand.Next() % FLAGS_column_families;
            l.reset(new MutexLock(
                shared->GetMutexForKey(rand_column_family, rand_key)));
          }

          keystr = Key(rand_key);
          key = keystr;
          column_family = column_families_[rand_column_family];

1816
          if (FLAGS_verify_before_write) {
I
Igor Canadi 已提交
1817 1818
            std::string keystr2 = Key(rand_key);
            Slice k = keystr2;
I
Igor Canadi 已提交
1819
            Status s = db_->Get(read_opts, column_family, k, &from_db);
A
Andres Noetzli 已提交
1820 1821
            if (!VerifyValue(rand_column_family, rand_key, read_opts,
                             thread->shared, from_db, s, true)) {
I
Igor Canadi 已提交
1822 1823
              break;
            }
1824
          }
A
Andres Noetzli 已提交
1825
          shared->Put(rand_column_family, rand_key, value_base);
1826
          Status s;
1827
          if (FLAGS_use_merge) {
1828
            s = db_->Merge(write_opts, column_family, key, v);
D
Deon Nicholas 已提交
1829
          } else {
1830 1831 1832 1833 1834
            s = db_->Put(write_opts, column_family, key, v);
          }
          if (!s.ok()) {
            fprintf(stderr, "put or merge error: %s\n", s.ToString().c_str());
            std::terminate();
D
Deon Nicholas 已提交
1835
          }
1836 1837
          thread->stats.AddBytesForWrites(1, sz);
        } else {
I
Igor Canadi 已提交
1838
          MultiPut(thread, write_opts, column_family, key, v, sz);
1839
        }
1840 1841
        PrintKeyValue(rand_column_family, static_cast<uint32_t>(rand_key),
                      value, sz);
1842 1843
      } else if (writeBound <= prob_op && prob_op < delBound) {
        // OPERATION delete
1844
        if (!FLAGS_test_batches_snapshots) {
A
Andres Noetzli 已提交
1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878
          // If the chosen key does not allow overwrite and it does not exist,
          // choose another key.
          while (!shared->AllowsOverwrite(rand_column_family, rand_key) &&
                 !shared->Exists(rand_column_family, rand_key)) {
            l.reset();
            rand_key = thread->rand.Next() % max_key;
            rand_column_family = thread->rand.Next() % FLAGS_column_families;
            l.reset(new MutexLock(
                shared->GetMutexForKey(rand_column_family, rand_key)));
          }

          keystr = Key(rand_key);
          key = keystr;
          column_family = column_families_[rand_column_family];

          // Use delete if the key may be overwritten and a single deletion
          // otherwise.
          if (shared->AllowsOverwrite(rand_column_family, rand_key)) {
            shared->Delete(rand_column_family, rand_key);
            Status s = db_->Delete(write_opts, column_family, key);
            thread->stats.AddDeletes(1);
            if (!s.ok()) {
              fprintf(stderr, "delete error: %s\n", s.ToString().c_str());
              std::terminate();
            }
          } else {
            shared->SingleDelete(rand_column_family, rand_key);
            Status s = db_->SingleDelete(write_opts, column_family, key);
            thread->stats.AddSingleDeletes(1);
            if (!s.ok()) {
              fprintf(stderr, "single delete error: %s\n",
                      s.ToString().c_str());
              std::terminate();
            }
1879
          }
1880
        } else {
I
Igor Canadi 已提交
1881
          MultiDelete(thread, write_opts, column_family, key);
1882
        }
1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910 1911 1912 1913 1914 1915 1916 1917 1918 1919 1920 1921 1922 1923
      } else if (delBound <= prob_op && prob_op < delRangeBound) {
        // OPERATION delete range
        if (!FLAGS_test_batches_snapshots) {
          std::vector<std::unique_ptr<MutexLock>> range_locks;
          // delete range does not respect disallowed overwrites. the keys for
          // which overwrites are disallowed are randomly distributed so it
          // could be expensive to find a range where each key allows
          // overwrites.
          if (rand_key > max_key - FLAGS_range_deletion_width) {
            l.reset();
            rand_key = thread->rand.Next() %
                       (max_key - FLAGS_range_deletion_width + 1);
            range_locks.emplace_back(new MutexLock(
                shared->GetMutexForKey(rand_column_family, rand_key)));
          } else {
            range_locks.emplace_back(std::move(l));
          }
          for (int j = 1; j < FLAGS_range_deletion_width; ++j) {
            if (((rand_key + j) & ((1 << FLAGS_log2_keys_per_lock) - 1)) == 0) {
              range_locks.emplace_back(new MutexLock(
                    shared->GetMutexForKey(rand_column_family, rand_key + j)));
            }
          }

          keystr = Key(rand_key);
          key = keystr;
          column_family = column_families_[rand_column_family];
          std::string end_keystr = Key(rand_key + FLAGS_range_deletion_width);
          Slice end_key = end_keystr;
          int covered = shared->DeleteRange(
              rand_column_family, rand_key,
              rand_key + FLAGS_range_deletion_width);
          Status s = db_->DeleteRange(write_opts, column_family, key, end_key);
          if (!s.ok()) {
            fprintf(stderr, "delete range error: %s\n",
                    s.ToString().c_str());
            std::terminate();
          }
          thread->stats.AddRangeDeletions(1);
          thread->stats.AddCoveredByRangeDeletions(covered);
        }
1924 1925
      } else {
        // OPERATION iterate
I
Igor Canadi 已提交
1926
        MultiIterate(thread, read_opts, column_family, key);
1927
      }
1928 1929
      thread->stats.FinishedSingleOp();
    }
1930

1931 1932 1933
    thread->stats.Stop();
  }

M
Mayank Agarwal 已提交
1934
  void VerifyDb(ThreadState* thread) const {
1935
    ReadOptions options(FLAGS_verify_checksum, true);
I
Igor Canadi 已提交
1936
    auto shared = thread->shared;
1937 1938 1939 1940
    const int64_t max_key = shared->GetMaxKey();
    const int64_t keys_per_thread = max_key / shared->GetNumThreads();
    int64_t start = keys_per_thread * thread->tid;
    int64_t end = start + keys_per_thread;
I
Igor Canadi 已提交
1941
    if (thread->tid == shared->GetNumThreads() - 1) {
1942 1943
      end = max_key;
    }
I
Igor Canadi 已提交
1944
    for (size_t cf = 0; cf < column_families_.size(); ++cf) {
I
Igor Canadi 已提交
1945 1946 1947
      if (thread->shared->HasVerificationFailedYet()) {
        break;
      }
I
Igor Canadi 已提交
1948 1949 1950 1951 1952
      if (!thread->rand.OneIn(2)) {
        // Use iterator to verify this range
        unique_ptr<Iterator> iter(
            db_->NewIterator(options, column_families_[cf]));
        iter->Seek(Key(start));
D
Dmitri Smirnov 已提交
1953
        for (auto i = start; i < end; i++) {
I
Igor Canadi 已提交
1954 1955 1956
          if (thread->shared->HasVerificationFailedYet()) {
            break;
          }
1957 1958 1959 1960 1961 1962
          // TODO(ljin): update "long" to uint64_t
          // Reseek when the prefix changes
          if (i % (static_cast<int64_t>(1) << 8 * (8 - FLAGS_prefix_size)) ==
              0) {
            iter->Seek(Key(i));
          }
I
Igor Canadi 已提交
1963 1964 1965 1966 1967 1968 1969 1970 1971 1972 1973
          std::string from_db;
          std::string keystr = Key(i);
          Slice k = keystr;
          Status s = iter->status();
          if (iter->Valid()) {
            if (iter->key().compare(k) > 0) {
              s = Status::NotFound(Slice());
            } else if (iter->key().compare(k) == 0) {
              from_db = iter->value().ToString();
              iter->Next();
            } else if (iter->key().compare(k) < 0) {
1974 1975
              VerificationAbort(shared, "An out of range key was found",
                                static_cast<int>(cf), i);
I
Igor Canadi 已提交
1976 1977 1978 1979
            }
          } else {
            // The iterator found no value for the key in question, so do not
            // move to the next item in the iterator
M
Mayank Agarwal 已提交
1980 1981
            s = Status::NotFound(Slice());
          }
1982 1983
          VerifyValue(static_cast<int>(cf), i, options, shared, from_db, s,
                      true);
I
Igor Canadi 已提交
1984
          if (from_db.length()) {
1985 1986
            PrintKeyValue(static_cast<int>(cf), static_cast<uint32_t>(i),
                          from_db.data(), from_db.length());
I
Igor Canadi 已提交
1987
          }
M
Mayank Agarwal 已提交
1988
        }
I
Igor Canadi 已提交
1989 1990
      } else {
        // Use Get to verify this range
D
Dmitri Smirnov 已提交
1991
        for (auto i = start; i < end; i++) {
I
Igor Canadi 已提交
1992 1993 1994
          if (thread->shared->HasVerificationFailedYet()) {
            break;
          }
I
Igor Canadi 已提交
1995 1996 1997 1998
          std::string from_db;
          std::string keystr = Key(i);
          Slice k = keystr;
          Status s = db_->Get(options, column_families_[cf], k, &from_db);
1999 2000
          VerifyValue(static_cast<int>(cf), i, options, shared, from_db, s,
                      true);
I
Igor Canadi 已提交
2001
          if (from_db.length()) {
2002 2003
            PrintKeyValue(static_cast<int>(cf), static_cast<uint32_t>(i),
                          from_db.data(), from_db.length());
I
Igor Canadi 已提交
2004
          }
M
Mayank Agarwal 已提交
2005
        }
2006 2007 2008 2009
      }
    }
  }

I
Igor Canadi 已提交
2010
  void VerificationAbort(SharedState* shared, std::string msg, int cf,
D
Dmitri Smirnov 已提交
2011 2012
                         int64_t key) const {
    printf("Verification failed for column family %d key %" PRIi64 ": %s\n", cf, key,
I
Igor Canadi 已提交
2013 2014
           msg.c_str());
    shared->SetVerificationFailure();
2015 2016
  }

D
Dmitri Smirnov 已提交
2017
  bool VerifyValue(int cf, int64_t key, const ReadOptions& opts,
I
Igor Canadi 已提交
2018
                   SharedState* shared, const std::string& value_from_db,
I
Igor Canadi 已提交
2019
                   Status s, bool strict = false) const {
I
Igor Canadi 已提交
2020 2021 2022
    if (shared->HasVerificationFailedYet()) {
      return false;
    }
2023
    // compare value_from_db with the value in the shared state
A
amayank 已提交
2024
    char value[100];
I
Igor Canadi 已提交
2025
    uint32_t value_base = shared->Get(cf, key);
2026
    if (value_base == SharedState::SENTINEL && !strict) {
I
Igor Canadi 已提交
2027
      return true;
2028 2029
    }

2030
    if (s.ok()) {
2031
      if (value_base == SharedState::SENTINEL) {
I
Igor Canadi 已提交
2032 2033
        VerificationAbort(shared, "Unexpected value found", cf, key);
        return false;
2034
      }
A
amayank 已提交
2035
      size_t sz = GenerateValue(value_base, value, sizeof(value));
2036
      if (value_from_db.length() != sz) {
I
Igor Canadi 已提交
2037 2038
        VerificationAbort(shared, "Length of value read is not equal", cf, key);
        return false;
2039
      }
2040
      if (memcmp(value_from_db.data(), value, sz) != 0) {
I
Igor Canadi 已提交
2041 2042 2043
        VerificationAbort(shared, "Contents of value read don't match", cf,
                          key);
        return false;
2044 2045 2046
      }
    } else {
      if (value_base != SharedState::SENTINEL) {
I
Igor Canadi 已提交
2047 2048
        VerificationAbort(shared, "Value not found: " + s.ToString(), cf, key);
        return false;
2049 2050
      }
    }
I
Igor Canadi 已提交
2051
    return true;
2052 2053
  }

D
Dmitri Smirnov 已提交
2054
  static void PrintKeyValue(int cf, int64_t key, const char* value,
I
Igor Canadi 已提交
2055 2056 2057 2058
                            size_t sz) {
    if (!FLAGS_verbose) {
      return;
    }
D
Dmitri Smirnov 已提交
2059
    fprintf(stdout, "[CF %d] %" PRIi64 " == > (%" ROCKSDB_PRIszt ") ", cf, key, sz);
I
Igor Canadi 已提交
2060
    for (size_t i = 0; i < sz; i++) {
2061 2062 2063 2064 2065 2066 2067 2068 2069 2070 2071 2072
      fprintf(stdout, "%X", value[i]);
    }
    fprintf(stdout, "\n");
  }

  static size_t GenerateValue(uint32_t rand, char *v, size_t max_sz) {
    size_t value_sz = ((rand % 3) + 1) * FLAGS_value_size_mult;
    assert(value_sz <= max_sz && value_sz >= sizeof(uint32_t));
    *((uint32_t*)v) = rand;
    for (size_t i=sizeof(uint32_t); i < value_sz; i++) {
      v[i] = (char)(rand ^ i);
    }
2073
    v[value_sz] = '\0';
2074 2075 2076 2077
    return value_sz; // the size of the value set.
  }

  void PrintEnv() const {
A
Andres Noetzli 已提交
2078
    fprintf(stdout, "RocksDB version           : %d.%d\n", kMajorVersion,
I
Igor Canadi 已提交
2079
            kMinorVersion);
A
Andres Noetzli 已提交
2080
    fprintf(stdout, "Column families           : %d\n", FLAGS_column_families);
I
Igor Canadi 已提交
2081
    if (!FLAGS_test_batches_snapshots) {
A
Andres Noetzli 已提交
2082
      fprintf(stdout, "Clear CFs one in          : %d\n",
I
Igor Canadi 已提交
2083 2084
              FLAGS_clear_column_family_one_in);
    }
A
Andres Noetzli 已提交
2085 2086
    fprintf(stdout, "Number of threads         : %d\n", FLAGS_threads);
    fprintf(stdout, "Ops per thread            : %lu\n",
K
kailiu 已提交
2087
            (unsigned long)FLAGS_ops_per_thread);
2088 2089 2090 2091
    std::string ttl_state("unused");
    if (FLAGS_ttl > 0) {
      ttl_state = NumberToString(FLAGS_ttl);
    }
A
Andres Noetzli 已提交
2092 2093 2094 2095 2096
    fprintf(stdout, "Time to live(sec)         : %s\n", ttl_state.c_str());
    fprintf(stdout, "Read percentage           : %d%%\n", FLAGS_readpercent);
    fprintf(stdout, "Prefix percentage         : %d%%\n", FLAGS_prefixpercent);
    fprintf(stdout, "Write percentage          : %d%%\n", FLAGS_writepercent);
    fprintf(stdout, "Delete percentage         : %d%%\n", FLAGS_delpercent);
2097
    fprintf(stdout, "Delete range percentage   : %d%%\n", FLAGS_delrangepercent);
A
Andres Noetzli 已提交
2098 2099 2100 2101 2102 2103 2104 2105
    fprintf(stdout, "No overwrite percentage   : %d%%\n",
            FLAGS_nooverwritepercent);
    fprintf(stdout, "Iterate percentage        : %d%%\n", FLAGS_iterpercent);
    fprintf(stdout, "DB-write-buffer-size      : %" PRIu64 "\n",
            FLAGS_db_write_buffer_size);
    fprintf(stdout, "Write-buffer-size         : %d\n",
            FLAGS_write_buffer_size);
    fprintf(stdout, "Iterations                : %lu\n",
K
kailiu 已提交
2106
            (unsigned long)FLAGS_num_iterations);
A
Andres Noetzli 已提交
2107
    fprintf(stdout, "Max key                   : %lu\n",
K
kailiu 已提交
2108
            (unsigned long)FLAGS_max_key);
A
Andres Noetzli 已提交
2109 2110 2111 2112
    fprintf(stdout, "Ratio #ops/#keys          : %f\n",
            (1.0 * FLAGS_ops_per_thread * FLAGS_threads) / FLAGS_max_key);
    fprintf(stdout, "Num times DB reopens      : %d\n", FLAGS_reopen);
    fprintf(stdout, "Batches/snapshots         : %d\n",
2113
            FLAGS_test_batches_snapshots);
A
Andres Noetzli 已提交
2114 2115
    fprintf(stdout, "Do update in place        : %d\n", FLAGS_in_place_update);
    fprintf(stdout, "Num keys per lock         : %d\n",
2116
            1 << FLAGS_log2_keys_per_lock);
2117
    std::string compression = CompressionTypeToString(FLAGS_compression_type_e);
A
Andres Noetzli 已提交
2118
    fprintf(stdout, "Compression               : %s\n", compression.c_str());
2119 2120
    fprintf(stdout, "Max subcompactions        : %" PRIu64 "\n",
            FLAGS_subcompactions);
J
Jim Paton 已提交
2121 2122 2123 2124 2125 2126

    const char* memtablerep = "";
    switch (FLAGS_rep_factory) {
      case kSkipList:
        memtablerep = "skip_list";
        break;
I
Igor Canadi 已提交
2127
      case kHashSkipList:
J
Jim Paton 已提交
2128 2129 2130 2131 2132 2133 2134
        memtablerep = "prefix_hash";
        break;
      case kVectorRep:
        memtablerep = "vector";
        break;
    }

A
Andres Noetzli 已提交
2135
    fprintf(stdout, "Memtablerep               : %s\n", memtablerep);
J
Jim Paton 已提交
2136

2137 2138 2139 2140 2141 2142 2143 2144
    fprintf(stdout, "Test kill odd             : %d\n", rocksdb_kill_odds);
    if (!rocksdb_kill_prefix_blacklist.empty()) {
      fprintf(stdout, "Skipping kill points prefixes:\n");
      for (auto& p : rocksdb_kill_prefix_blacklist) {
        fprintf(stdout, "  %s\n", p.c_str());
      }
    }

2145 2146 2147 2148
    fprintf(stdout, "------------------------------------------------\n");
  }

  void Open() {
2149
    assert(db_ == nullptr);
2150 2151 2152 2153
    BlockBasedTableOptions block_based_options;
    block_based_options.block_cache = cache_;
    block_based_options.block_cache_compressed = compressed_cache_;
    block_based_options.block_size = FLAGS_block_size;
2154
    block_based_options.format_version = 2;
2155 2156 2157
    block_based_options.filter_policy = filter_policy_;
    options_.table_factory.reset(
        NewBlockBasedTableFactory(block_based_options));
2158
    options_.db_write_buffer_size = FLAGS_db_write_buffer_size;
I
Igor Canadi 已提交
2159 2160 2161 2162
    options_.write_buffer_size = FLAGS_write_buffer_size;
    options_.max_write_buffer_number = FLAGS_max_write_buffer_number;
    options_.min_write_buffer_number_to_merge =
        FLAGS_min_write_buffer_number_to_merge;
2163 2164
    options_.max_write_buffer_number_to_maintain =
        FLAGS_max_write_buffer_number_to_maintain;
I
Igor Canadi 已提交
2165 2166 2167 2168 2169 2170 2171 2172 2173
    options_.max_background_compactions = FLAGS_max_background_compactions;
    options_.max_background_flushes = FLAGS_max_background_flushes;
    options_.compaction_style =
        static_cast<rocksdb::CompactionStyle>(FLAGS_compaction_style);
    options_.prefix_extractor.reset(NewFixedPrefixTransform(FLAGS_prefix_size));
    options_.max_open_files = FLAGS_open_files;
    options_.statistics = dbstats;
    options_.env = FLAGS_env;
    options_.use_fsync = FLAGS_use_fsync;
2174
    options_.compaction_readahead_size = FLAGS_compaction_readahead_size;
I
Igor Canadi 已提交
2175
    options_.allow_mmap_reads = FLAGS_mmap_read;
2176 2177
    options_.allow_mmap_writes = FLAGS_mmap_write;
    options_.use_direct_reads = FLAGS_use_direct_reads;
2178 2179
    options_.use_direct_io_for_flush_and_compaction =
        FLAGS_use_direct_io_for_flush_and_compaction;
I
Igor Canadi 已提交
2180 2181 2182 2183
    options_.target_file_size_base = FLAGS_target_file_size_base;
    options_.target_file_size_multiplier = FLAGS_target_file_size_multiplier;
    options_.max_bytes_for_level_base = FLAGS_max_bytes_for_level_base;
    options_.max_bytes_for_level_multiplier =
2184
        FLAGS_max_bytes_for_level_multiplier;
I
Igor Canadi 已提交
2185 2186 2187 2188 2189 2190 2191 2192
    options_.level0_stop_writes_trigger = FLAGS_level0_stop_writes_trigger;
    options_.level0_slowdown_writes_trigger =
        FLAGS_level0_slowdown_writes_trigger;
    options_.level0_file_num_compaction_trigger =
        FLAGS_level0_file_num_compaction_trigger;
    options_.compression = FLAGS_compression_type_e;
    options_.create_if_missing = true;
    options_.max_manifest_file_size = 10 * 1024;
2193
    options_.inplace_update_support = FLAGS_in_place_update;
2194
    options_.max_subcompactions = static_cast<uint32_t>(FLAGS_subcompactions);
2195 2196 2197 2198
    options_.allow_concurrent_memtable_write =
        FLAGS_allow_concurrent_memtable_write;
    options_.enable_write_thread_adaptive_yield =
        FLAGS_enable_write_thread_adaptive_yield;
2199 2200 2201 2202 2203 2204 2205 2206 2207 2208
    if (FLAGS_rate_limiter_bytes_per_sec > 0) {
      options_.rate_limiter.reset(NewGenericRateLimiter(
          FLAGS_rate_limiter_bytes_per_sec, 1000 /* refill_period_us */,
          10 /* fairness */,
          FLAGS_rate_limit_bg_reads ? RateLimiter::Mode::kReadsOnly
                                    : RateLimiter::Mode::kWritesOnly));
      if (FLAGS_rate_limit_bg_reads) {
        options_.new_table_reader_for_compaction_inputs = true;
      }
    }
2209 2210

    if (FLAGS_prefix_size == 0 && FLAGS_rep_factory == kHashSkipList) {
J
Jim Paton 已提交
2211
      fprintf(stderr,
2212
              "prefeix_size cannot be zero if memtablerep == prefix_hash\n");
J
Jim Paton 已提交
2213 2214
      exit(1);
    }
2215 2216 2217 2218 2219
    if (FLAGS_prefix_size != 0 && FLAGS_rep_factory != kHashSkipList) {
      fprintf(stderr,
              "WARNING: prefix_size is non-zero but "
              "memtablerep != prefix_hash\n");
    }
J
Jim Paton 已提交
2220 2221 2222 2223
    switch (FLAGS_rep_factory) {
      case kSkipList:
        // no need to do anything
        break;
S
sdong 已提交
2224 2225 2226 2227
#ifndef ROCKSDB_LITE
      case kHashSkipList:
        options_.memtable_factory.reset(NewHashSkipListRepFactory(10000));
        break;
J
Jim Paton 已提交
2228
      case kVectorRep:
I
Igor Canadi 已提交
2229
        options_.memtable_factory.reset(new VectorRepFactory());
J
Jim Paton 已提交
2230
        break;
S
sdong 已提交
2231 2232 2233 2234 2235 2236
#else
      default:
        fprintf(stderr,
                "RocksdbLite only supports skip list mem table. Skip "
                "--rep_factory\n");
#endif  // ROCKSDB_LITE
J
Jim Paton 已提交
2237
    }
S
sdong 已提交
2238

2239 2240 2241 2242
    if (FLAGS_use_full_merge_v1) {
      options_.merge_operator = MergeOperators::CreateDeprecatedPutOperator();
    } else {
      options_.merge_operator = MergeOperators::CreatePutOperator();
D
Deon Nicholas 已提交
2243 2244
    }

2245 2246
    // set universal style compaction configurations, if applicable
    if (FLAGS_universal_size_ratio != 0) {
I
Igor Canadi 已提交
2247 2248
      options_.compaction_options_universal.size_ratio =
          FLAGS_universal_size_ratio;
2249 2250
    }
    if (FLAGS_universal_min_merge_width != 0) {
I
Igor Canadi 已提交
2251 2252
      options_.compaction_options_universal.min_merge_width =
          FLAGS_universal_min_merge_width;
2253 2254
    }
    if (FLAGS_universal_max_merge_width != 0) {
I
Igor Canadi 已提交
2255 2256
      options_.compaction_options_universal.max_merge_width =
          FLAGS_universal_max_merge_width;
2257 2258
    }
    if (FLAGS_universal_max_size_amplification_percent != 0) {
I
Igor Canadi 已提交
2259 2260
      options_.compaction_options_universal.max_size_amplification_percent =
          FLAGS_universal_max_size_amplification_percent;
2261 2262
    }

2263
    fprintf(stdout, "DB path: [%s]\n", FLAGS_db.c_str());
2264

2265 2266
    Status s;
    if (FLAGS_ttl == -1) {
I
Igor Canadi 已提交
2267 2268 2269 2270 2271 2272 2273
      std::vector<std::string> existing_column_families;
      s = DB::ListColumnFamilies(DBOptions(options_), FLAGS_db,
                                 &existing_column_families);  // ignore errors
      if (!s.ok()) {
        // DB doesn't exist
        assert(existing_column_families.empty());
        assert(column_family_names_.empty());
2274
        column_family_names_.push_back(kDefaultColumnFamilyName);
I
Igor Canadi 已提交
2275 2276 2277 2278 2279 2280 2281
      } else if (column_family_names_.empty()) {
        // this is the first call to the function Open()
        column_family_names_ = existing_column_families;
      } else {
        // this is a reopen. just assert that existing column_family_names are
        // equivalent to what we remember
        auto sorted_cfn = column_family_names_;
2282 2283 2284
        std::sort(sorted_cfn.begin(), sorted_cfn.end());
        std::sort(existing_column_families.begin(),
                  existing_column_families.end());
I
Igor Canadi 已提交
2285 2286 2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302
        if (sorted_cfn != existing_column_families) {
          fprintf(stderr,
                  "Expected column families differ from the existing:\n");
          printf("Expected: {");
          for (auto cf : sorted_cfn) {
            printf("%s ", cf.c_str());
          }
          printf("}\n");
          printf("Existing: {");
          for (auto cf : existing_column_families) {
            printf("%s ", cf.c_str());
          }
          printf("}\n");
        }
        assert(sorted_cfn == existing_column_families);
      }
      std::vector<ColumnFamilyDescriptor> cf_descriptors;
      for (auto name : column_family_names_) {
2303
        if (name != kDefaultColumnFamilyName) {
I
Igor Canadi 已提交
2304 2305 2306 2307 2308
          new_column_family_name_ =
              std::max(new_column_family_name_.load(), std::stoi(name) + 1);
        }
        cf_descriptors.emplace_back(name, ColumnFamilyOptions(options_));
      }
I
Igor Canadi 已提交
2309
      while (cf_descriptors.size() < (size_t)FLAGS_column_families) {
2310
        std::string name = ToString(new_column_family_name_.load());
I
Igor Canadi 已提交
2311 2312 2313 2314
        new_column_family_name_++;
        cf_descriptors.emplace_back(name, ColumnFamilyOptions(options_));
        column_family_names_.push_back(name);
      }
2315 2316
      options_.listeners.clear();
      options_.listeners.emplace_back(
Y
Yueh-Hsuan Chiang 已提交
2317
          new DbStressListener(FLAGS_db, options_.db_paths));
I
Igor Canadi 已提交
2318
      options_.create_missing_column_families = true;
I
Igor Canadi 已提交
2319 2320 2321 2322
      s = DB::Open(DBOptions(options_), FLAGS_db, cf_descriptors,
                   &column_families_, &db_);
      assert(!s.ok() || column_families_.size() ==
                            static_cast<size_t>(FLAGS_column_families));
2323
    } else {
S
sdong 已提交
2324
#ifndef ROCKSDB_LITE
I
Igor Canadi 已提交
2325 2326 2327
      DBWithTTL* db_with_ttl;
      s = DBWithTTL::Open(options_, FLAGS_db, &db_with_ttl, FLAGS_ttl);
      db_ = db_with_ttl;
S
sdong 已提交
2328 2329 2330 2331
#else
      fprintf(stderr, "TTL is not supported in RocksDBLite\n");
      exit(1);
#endif
2332
    }
2333 2334 2335 2336 2337 2338
    if (!s.ok()) {
      fprintf(stderr, "open error: %s\n", s.ToString().c_str());
      exit(1);
    }
  }

2339
  void Reopen() {
I
Igor Canadi 已提交
2340 2341
    for (auto cf : column_families_) {
      delete cf;
2342
    }
I
Igor Canadi 已提交
2343 2344
    column_families_.clear();
    delete db_;
2345
    db_ = nullptr;
2346 2347

    num_times_reopened_++;
D
Dmitri Smirnov 已提交
2348
    auto now = FLAGS_env->NowMicros();
2349
    fprintf(stdout, "%s Reopening database for the %dth time\n",
D
Dmitri Smirnov 已提交
2350
            FLAGS_env->TimeToString(now/1000000).c_str(),
2351
            num_times_reopened_);
2352 2353 2354
    Open();
  }

2355 2356
  void PrintStatistics() {
    if (dbstats) {
2357
      fprintf(stdout, "STATISTICS:\n%s\n", dbstats->ToString().c_str());
2358 2359 2360 2361
    }
  }

 private:
2362 2363 2364
  std::shared_ptr<Cache> cache_;
  std::shared_ptr<Cache> compressed_cache_;
  std::shared_ptr<const FilterPolicy> filter_policy_;
2365
  DB* db_;
I
Igor Canadi 已提交
2366 2367 2368 2369
  Options options_;
  std::vector<ColumnFamilyHandle*> column_families_;
  std::vector<std::string> column_family_names_;
  std::atomic<int> new_column_family_name_;
2370
  int num_times_reopened_;
L
Lei Jin 已提交
2371 2372
  std::unordered_map<std::string, std::vector<std::string>> options_table_;
  std::vector<std::string> options_index_;
2373 2374
};

2375
}  // namespace rocksdb
2376 2377

int main(int argc, char** argv) {
2378 2379 2380
  SetUsageMessage(std::string("\nUSAGE:\n") + std::string(argv[0]) +
                  " [OPTIONS]...");
  ParseCommandLineFlags(&argc, &argv, true);
S
Sagar Vemuri 已提交
2381 2382
#if !defined(NDEBUG) && !defined(OS_MACOSX) && !defined(OS_WIN) && \
  !defined(OS_SOLARIS) && !defined(OS_AIX)
2383
  rocksdb::SyncPoint::GetInstance()->SetCallBack(
2384 2385 2386 2387
    "NewWritableFile:O_DIRECT", [&](void* arg) {
      int* val = static_cast<int*>(arg);
      *val &= ~O_DIRECT;
    });
2388
  rocksdb::SyncPoint::GetInstance()->SetCallBack(
2389 2390 2391 2392
    "NewRandomAccessFile:O_DIRECT", [&](void* arg) {
      int* val = static_cast<int*>(arg);
      *val &= ~O_DIRECT;
    });
2393 2394
  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
#endif
2395 2396 2397 2398 2399 2400 2401 2402

  if (FLAGS_statistics) {
    dbstats = rocksdb::CreateDBStatistics();
  }
  FLAGS_compression_type_e =
    StringToCompressionType(FLAGS_compression_type.c_str());
  if (!FLAGS_hdfs.empty()) {
    FLAGS_env  = new rocksdb::HdfsEnv(FLAGS_hdfs);
2403
  }
2404
  FLAGS_rep_factory = StringToRepFactory(FLAGS_memtablerep.c_str());
2405

2406 2407 2408 2409
  // The number of background threads should be at least as much the
  // max number of concurrent compactions.
  FLAGS_env->SetBackgroundThreads(FLAGS_max_background_compactions);

L
Lei Jin 已提交
2410 2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421
  if (FLAGS_prefixpercent > 0 && FLAGS_prefix_size <= 0) {
    fprintf(stderr,
            "Error: prefixpercent is non-zero while prefix_size is "
            "not positive!\n");
    exit(1);
  }
  if (FLAGS_test_batches_snapshots && FLAGS_prefix_size <= 0) {
    fprintf(stderr,
            "Error: please specify prefix_size for "
            "test_batches_snapshots test!\n");
    exit(1);
  }
2422
  if ((FLAGS_readpercent + FLAGS_prefixpercent +
2423 2424
       FLAGS_writepercent + FLAGS_delpercent + FLAGS_delrangepercent +
       FLAGS_iterpercent) != 100) {
2425
      fprintf(stderr,
2426 2427
              "Error: Read+Prefix+Write+Delete+DeleteRange+Iterate percents != "
              "100!\n");
A
amayank 已提交
2428 2429
      exit(1);
  }
2430 2431 2432 2433
  if (FLAGS_disable_wal == 1 && FLAGS_reopen > 0) {
      fprintf(stderr, "Error: Db cannot reopen safely with disable_wal set!\n");
      exit(1);
  }
2434
  if ((unsigned)FLAGS_reopen >= FLAGS_ops_per_thread) {
K
kailiu 已提交
2435 2436 2437 2438 2439
      fprintf(stderr,
              "Error: #DB-reopens should be < ops_per_thread\n"
              "Provided reopens = %d and ops_per_thread = %lu\n",
              FLAGS_reopen,
              (unsigned long)FLAGS_ops_per_thread);
2440 2441
      exit(1);
  }
2442 2443 2444 2445 2446
  if (FLAGS_test_batches_snapshots && FLAGS_delrangepercent > 0) {
    fprintf(stderr, "Error: nonzero delrangepercent unsupported in "
                    "test_batches_snapshots mode\n");
    exit(1);
  }
A
Andrew Kryczka 已提交
2447 2448 2449 2450 2451 2452
  if (FLAGS_active_width > FLAGS_max_key) {
    fprintf(stderr, "Error: active_width can be at most max_key\n");
    exit(1);
  } else if (FLAGS_active_width == 0) {
    FLAGS_active_width = FLAGS_max_key;
  }
2453

2454
  // Choose a location for the test database if none given with --db=<path>
2455 2456
  if (FLAGS_db.empty()) {
      std::string default_db_path;
2457
      rocksdb::Env::Default()->GetTestDirectory(&default_db_path);
2458
      default_db_path += "/dbstress";
2459
      FLAGS_db = default_db_path;
2460 2461
  }

2462 2463 2464
  rocksdb_kill_odds = FLAGS_kill_random_test;
  rocksdb_kill_prefix_blacklist = SplitString(FLAGS_kill_prefix_blacklist);

2465
  rocksdb::StressTest stress;
I
Igor Canadi 已提交
2466 2467 2468 2469 2470
  if (stress.Run()) {
    return 0;
  } else {
    return 1;
  }
2471
}
2472 2473

#endif  // GFLAGS