db_stress.cc 71.8 KB
Newer Older
1 2 3 4 5
//  Copyright (c) 2013, Facebook, Inc.  All rights reserved.
//  This source code is licensed under the BSD-style license found in the
//  LICENSE file in the root directory of this source tree. An additional grant
//  of patent rights can be found in the PATENTS file in the same directory.
//
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 32
#define __STDC_FORMAT_MACROS
#include <inttypes.h>
33 34
#include <stdio.h>
#include <stdlib.h>
35 36
#include <sys/types.h>
#include <chrono>
37
#include <exception>
38 39
#include <thread>

40
#include <gflags/gflags.h>
41 42
#include "db/db_impl.h"
#include "db/version_set.h"
I
Igor Canadi 已提交
43
#include "rocksdb/statistics.h"
44
#include "rocksdb/cache.h"
45
#include "rocksdb/utilities/db_ttl.h"
46 47
#include "rocksdb/env.h"
#include "rocksdb/write_batch.h"
48 49
#include "rocksdb/slice.h"
#include "rocksdb/slice_transform.h"
50
#include "port/port.h"
51
#include "util/coding.h"
52 53 54 55 56
#include "util/crc32c.h"
#include "util/histogram.h"
#include "util/mutexlock.h"
#include "util/random.h"
#include "util/testutil.h"
57
#include "util/logging.h"
58
#include "hdfs/env_hdfs.h"
D
Deon Nicholas 已提交
59
#include "utilities/merge_operators.h"
60
#include "util/string_util.h"
61

62 63 64
using GFLAGS::ParseCommandLineFlags;
using GFLAGS::RegisterFlagValidator;
using GFLAGS::SetUsageMessage;
65

66
static const long KB = 1024;
67

68 69
static bool ValidateUint32Range(const char* flagname, uint64_t value) {
  if (value > std::numeric_limits<uint32_t>::max()) {
K
kailiu 已提交
70 71 72 73
    fprintf(stderr,
            "Invalid value for --%s: %lu, overflow\n",
            flagname,
            (unsigned long)value);
74 75 76 77
    return false;
  }
  return true;
}
78

79
DEFINE_uint64(seed, 2341234, "Seed for PRNG");
I
Igor Canadi 已提交
80
static const bool FLAGS_seed_dummy __attribute__((unused)) =
81
    RegisterFlagValidator(&FLAGS_seed, &ValidateUint32Range);
82

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

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

88
DEFINE_bool(test_batches_snapshots, false,
A
Andres Noetzli 已提交
89
            "If set, the test uses MultiGet(), MultiPut() and MultiDelete()"
90 91 92 93 94 95 96 97 98
            " 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");
99

100
DEFINE_int32(threads, 32, "Number of concurrent threads to run.");
101

102 103 104 105
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");
106

107 108
DEFINE_int32(value_size_mult, 8,
             "Size of value will be this number times rand_int(1,3) bytes");
109

110
DEFINE_bool(verify_before_write, false, "Verify before write");
111

112
DEFINE_bool(histogram, false, "Print histogram of operation timings");
113

114 115
DEFINE_bool(destroy_db_initially, true,
            "Destroys the database dir before start if this is true");
116

117 118 119 120
DEFINE_bool(verbose, false, "Verbose");

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

122 123 124
DEFINE_uint64(db_write_buffer_size, rocksdb::Options().db_write_buffer_size,
              "Number of bytes to buffer in all memtables before compacting");

125 126
DEFINE_int32(write_buffer_size,
             static_cast<int32_t>(rocksdb::Options().write_buffer_size),
127
             "Number of bytes to buffer in memtable before compacting");
128

129 130 131 132
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.");
133

134 135 136 137 138 139 140 141 142 143
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.");
144

145 146 147 148 149 150 151 152 153 154 155 156 157 158
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.");

159 160 161
DEFINE_int32(open_files, rocksdb::Options().max_open_files,
             "Maximum number of files to keep open at the same time "
             "(use default if == 0)");
162

163 164 165 166
DEFINE_int64(compressed_cache_size, -1,
             "Number of bytes to use as a cache of compressed data."
             " Negative means use default settings.");

167
DEFINE_int32(compaction_style, rocksdb::Options().compaction_style, "");
168

169 170 171
DEFINE_int32(level0_file_num_compaction_trigger,
             rocksdb::Options().level0_file_num_compaction_trigger,
             "Level0 compaction start trigger");
172

173 174 175
DEFINE_int32(level0_slowdown_writes_trigger,
             rocksdb::Options().level0_slowdown_writes_trigger,
             "Number of files in level-0 that will slow down writes");
176

177 178 179
DEFINE_int32(level0_stop_writes_trigger,
             rocksdb::Options().level0_stop_writes_trigger,
             "Number of files in level-0 that will trigger put stop.");
180

181 182
DEFINE_int32(block_size,
             static_cast<int32_t>(rocksdb::BlockBasedTableOptions().block_size),
183
             "Number of bytes in a block.");
184

185 186 187 188
DEFINE_int32(max_background_compactions,
             rocksdb::Options().max_background_compactions,
             "The maximum number of concurrent background compactions "
             "that can occur in parallel.");
189

190 191 192 193
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 已提交
194
DEFINE_int32(compaction_thread_pool_variations, 2,
A
Andres Noetzli 已提交
195
             "Range of background thread pool size variations when adjusted "
196 197
             "periodically.");

I
Igor Canadi 已提交
198 199 200 201
DEFINE_int32(max_background_flushes, rocksdb::Options().max_background_flushes,
             "The maximum number of concurrent background flushes "
             "that can occur in parallel.");

202 203
DEFINE_int32(universal_size_ratio, 0, "The ratio of file sizes that trigger"
             " compaction in universal style");
204

205 206
DEFINE_int32(universal_min_merge_width, 0, "The minimum number of files to "
             "compact in universal style compaction");
207

208 209
DEFINE_int32(universal_max_merge_width, 0, "The max number of files to compact"
             " in universal style compaction");
210

211 212
DEFINE_int32(universal_max_size_amplification_percent, 0,
             "The max size amplification for universal style compaction");
213

I
Igor Canadi 已提交
214 215 216 217 218
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 已提交
219 220 221
DEFINE_int32(set_options_one_in, 0,
             "With a chance of 1/N, change some random options");

222 223 224
DEFINE_int32(set_in_place_one_in, 0,
             "With a chance of 1/N, toggle in place support option");

225 226
DEFINE_int64(cache_size, 2 * KB * KB * KB,
             "Number of bytes to use as a cache of uncompressed data.");
227

228 229 230 231 232 233 234 235 236
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 已提交
237
static const bool FLAGS_reopen_dummy __attribute__((unused)) =
238
    RegisterFlagValidator(&FLAGS_reopen, &ValidateInt32Positive);
239

240 241
DEFINE_int32(bloom_bits, 10, "Bloom filter bits per key. "
             "Negative means use default settings.");
242

243 244 245
DEFINE_bool(use_block_based_filter, false, "use block based filter"
              "instead of full filter for block based table");

246
DEFINE_string(db, "", "Use the db with the following name.");
247

248 249
DEFINE_bool(verify_checksum, false,
            "Verify checksum for every block read from storage");
250

251 252
DEFINE_bool(mmap_read, rocksdb::EnvOptions().use_mmap_reads,
            "Allow reads to occur via mmap-ing files");
253

254 255 256
// Database statistics
static std::shared_ptr<rocksdb::Statistics> dbstats;
DEFINE_bool(statistics, false, "Create database statistics");
257

258
DEFINE_bool(sync, false, "Sync all writes to disk");
259

260 261
DEFINE_bool(disable_data_sync, false,
            "If true, do not wait until data is synced to disk.");
A
amayank 已提交
262

263
DEFINE_bool(use_fsync, false, "If true, issue fsync instead of fdatasync");
264

265 266 267
DEFINE_int32(kill_random_test, 0,
             "If non-zero, kill at various points in source code with "
             "probability 1/this");
I
Igor Canadi 已提交
268
static const bool FLAGS_kill_random_test_dummy __attribute__((unused)) =
269
    RegisterFlagValidator(&FLAGS_kill_random_test, &ValidateInt32Positive);
270
extern int rocksdb_kill_odds;
271

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

274 275
DEFINE_int32(target_file_size_base, 64 * KB,
             "Target level-1 file size for compaction");
276

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

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

282 283
DEFINE_int32(max_bytes_for_level_multiplier, 2,
             "A multiplier to compute max bytes for level-N (N >= 2)");
284

285 286 287 288 289 290 291 292 293 294
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;
}
DEFINE_int32(readpercent, 10,
             "Ratio of reads to total workload (expressed as a percentage)");
I
Igor Canadi 已提交
295
static const bool FLAGS_readpercent_dummy __attribute__((unused)) =
296
    RegisterFlagValidator(&FLAGS_readpercent, &ValidateInt32Percent);
297 298 299 300

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

DEFINE_int32(writepercent, 45,
A
Andres Noetzli 已提交
305
             "Ratio of writes to total workload (expressed as a percentage)");
I
Igor Canadi 已提交
306
static const bool FLAGS_writepercent_dummy __attribute__((unused)) =
307
    RegisterFlagValidator(&FLAGS_writepercent, &ValidateInt32Percent);
308 309 310

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

DEFINE_int32(iterpercent, 10, "Ratio of iterations to total workload"
             " (expressed as a percentage)");
I
Igor Canadi 已提交
316
static const bool FLAGS_iterpercent_dummy __attribute__((unused)) =
317
    RegisterFlagValidator(&FLAGS_iterpercent, &ValidateInt32Percent);
318 319

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

I
Igor Canadi 已提交
323
namespace {
324 325 326 327 328 329 330 331 332 333 334
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 已提交
335 336 337 338
  else if (!strcasecmp(ctype, "lz4"))
    return rocksdb::kLZ4Compression;
  else if (!strcasecmp(ctype, "lz4hc"))
    return rocksdb::kLZ4HCCompression;
339 340 341 342

  fprintf(stdout, "Cannot parse compression type '%s'\n", ctype);
  return rocksdb::kSnappyCompression; //default value
}
I
Igor Canadi 已提交
343 344
}  // namespace

345 346 347
DEFINE_string(compression_type, "snappy",
              "Algorithm to use to compress the database");
static enum rocksdb::CompressionType FLAGS_compression_type_e =
348
    rocksdb::kSnappyCompression;
349

350
DEFINE_string(hdfs, "", "Name of hdfs environment");
351
// posix or hdfs environment
352
static rocksdb::Env* FLAGS_env = rocksdb::Env::Default();
353

I
Igor Canadi 已提交
354 355
DEFINE_uint64(ops_per_thread, 1200000, "Number of operations per thread.");
static const bool FLAGS_ops_per_thread_dummy __attribute__((unused)) =
356
    RegisterFlagValidator(&FLAGS_ops_per_thread, &ValidateUint32Range);
357

358
DEFINE_uint64(log2_keys_per_lock, 2, "Log2 of number of keys per lock");
I
Igor Canadi 已提交
359
static const bool FLAGS_log2_keys_per_lock_dummy __attribute__((unused)) =
360
    RegisterFlagValidator(&FLAGS_log2_keys_per_lock, &ValidateUint32Range);
361

362 363
DEFINE_bool(filter_deletes, false, "On true, deletes use KeyMayExist to drop"
            " the delete if key not present");
364

365 366
DEFINE_bool(in_place_update, false, "On true, does inplace update in memtable");

J
Jim Paton 已提交
367 368
enum RepFactory {
  kSkipList,
I
Igor Canadi 已提交
369
  kHashSkipList,
J
Jim Paton 已提交
370 371
  kVectorRep
};
I
Igor Canadi 已提交
372 373

namespace {
374 375 376 377 378 379
enum RepFactory StringToRepFactory(const char* ctype) {
  assert(ctype);

  if (!strcasecmp(ctype, "skip_list"))
    return kSkipList;
  else if (!strcasecmp(ctype, "prefix_hash"))
I
Igor Canadi 已提交
380
    return kHashSkipList;
381 382 383 384 385 386
  else if (!strcasecmp(ctype, "vector"))
    return kVectorRep;

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

J
Jim Paton 已提交
389
static enum RepFactory FLAGS_rep_factory;
L
Lei Jin 已提交
390
DEFINE_string(memtablerep, "prefix_hash", "");
J
Jim Paton 已提交
391

392
static bool ValidatePrefixSize(const char* flagname, int32_t value) {
L
Lei Jin 已提交
393 394
  if (value < 0 || value > 8) {
    fprintf(stderr, "Invalid value for --%s: %d. 0 <= PrefixSize <= 8\n",
395 396 397 398 399
            flagname, value);
    return false;
  }
  return true;
}
I
Igor Canadi 已提交
400
DEFINE_int32(prefix_size, 7, "Control the prefix size for HashSkipListRep");
L
Lei Jin 已提交
401
static const bool FLAGS_prefix_size_dummy __attribute__((unused)) =
402
    RegisterFlagValidator(&FLAGS_prefix_size, &ValidatePrefixSize);
403 404 405

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

D
Deon Nicholas 已提交
407

408
namespace rocksdb {
409

410 411 412 413 414 415 416 417 418 419 420 421 422
// convert long to a big-endian slice key
static std::string Key(long val) {
  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));
  for (int i=0; i<(int)sizeof(val); i++) {
    big_endian_key[i] = little_endian_key[sizeof(val) - 1 - i];
  }
  return big_endian_key;
}

423 424 425 426 427 428 429 430 431 432 433
static std::string StringToHex(const std::string& str) {
  std::string result = "0x";
  char buf[10];
  for (size_t i = 0; i < str.length(); i++) {
    snprintf(buf, 10, "%02X", (unsigned char)str[i]);
    result += buf;
  }
  return result;
}


434 435 436 437 438 439 440 441 442
class StressTest;
namespace {

class Stats {
 private:
  double start_;
  double finish_;
  double seconds_;
  long done_;
443 444
  long gets_;
  long prefixes_;
445
  long writes_;
A
amayank 已提交
446
  long deletes_;
447
  long iterator_size_sums_;
448
  long founds_;
449
  long iterations_;
450
  long errors_;
451 452 453
  int next_report_;
  size_t bytes_;
  double last_op_finish_;
454
  HistogramImpl hist_;
455 456 457 458 459 460 461 462

 public:
  Stats() { }

  void Start() {
    next_report_ = 100;
    hist_.Clear();
    done_ = 0;
463 464
    gets_ = 0;
    prefixes_ = 0;
465
    writes_ = 0;
A
amayank 已提交
466
    deletes_ = 0;
467
    iterator_size_sums_ = 0;
468
    founds_ = 0;
469
    iterations_ = 0;
470
    errors_ = 0;
471 472 473 474 475 476 477 478 479 480
    bytes_ = 0;
    seconds_ = 0;
    start_ = FLAGS_env->NowMicros();
    last_op_finish_ = start_;
    finish_ = start_;
  }

  void Merge(const Stats& other) {
    hist_.Merge(other.hist_);
    done_ += other.done_;
481 482
    gets_ += other.gets_;
    prefixes_ += other.prefixes_;
483
    writes_ += other.writes_;
A
amayank 已提交
484
    deletes_ += other.deletes_;
485
    iterator_size_sums_ += other.iterator_size_sums_;
486
    founds_ += other.founds_;
487
    iterations_ += other.iterations_;
488
    errors_ += other.errors_;
489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505
    bytes_ += other.bytes_;
    seconds_ += other.seconds_;
    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) {
      double now = FLAGS_env->NowMicros();
      double micros = now - last_op_finish_;
      hist_.Add(micros);
      if (micros > 20000) {
506
        fprintf(stdout, "long op: %.1f micros%30s\r", micros, "");
507 508 509 510
      }
      last_op_finish_ = now;
    }

511
      done_++;
512
    if (FLAGS_progress_reports) {
513 514 515 516 517 518 519 520 521 522
      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_, "");
      }
523 524 525
    }
  }

526 527 528 529 530
  void AddBytesForWrites(int nwrites, size_t nbytes) {
    writes_ += nwrites;
    bytes_ += nbytes;
  }

531 532 533
  void AddGets(int ngets, int nfounds) {
    founds_ += nfounds;
    gets_ += ngets;
534 535
  }

536 537 538 539 540
  void AddPrefixes(int nprefixes, int count) {
    prefixes_ += nprefixes;
    iterator_size_sums_ += count;
  }

541 542 543 544
  void AddIterations(int n) {
    iterations_ += n;
  }

545 546 547 548
  void AddDeletes(int n) {
    deletes_ += n;
  }

549 550
  void AddErrors(int n) {
    errors_ += n;
A
amayank 已提交
551 552
  }

553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569
  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_);
570
    fprintf(stdout, "%-12s: Wrote %ld times\n", "", writes_);
A
amayank 已提交
571
    fprintf(stdout, "%-12s: Deleted %ld times\n", "", deletes_);
X
Xing Jin 已提交
572 573
    fprintf(stdout, "%-12s: %ld read and %ld found the key\n", "",
            gets_, founds_);
574 575 576
    fprintf(stdout, "%-12s: Prefix scanned %ld times\n", "", prefixes_);
    fprintf(stdout, "%-12s: Iterator size sum is %ld\n", "",
            iterator_size_sums_);
577
    fprintf(stdout, "%-12s: Iterated %ld times\n", "", iterations_);
578
    fprintf(stdout, "%-12s: Got errors %ld times\n", "", errors_);
579 580 581 582 583 584 585 586 587 588 589

    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:
590
  static const uint32_t SENTINEL;
591

592 593
  explicit SharedState(StressTest* stress_test)
      : cv_(&mu_),
594
        seed_(static_cast<uint32_t>(FLAGS_seed)),
595
        max_key_(FLAGS_max_key),
596
        log2_keys_per_lock_(static_cast<uint32_t>(FLAGS_log2_keys_per_lock)),
597 598 599 600 601 602 603
        num_threads_(FLAGS_threads),
        num_initialized_(0),
        num_populated_(0),
        vote_reopen_(0),
        num_done_(0),
        start_(false),
        start_verify_(false),
604 605
        should_stop_bg_thread_(false),
        bg_thread_finished_(false),
606 607
        stress_test_(stress_test),
        verification_failure_(false) {
608 609 610 611
    if (FLAGS_test_batches_snapshots) {
      fprintf(stdout, "No lock creation because test_batches_snapshots set\n");
      return;
    }
I
Igor Canadi 已提交
612 613 614 615
    values_.resize(FLAGS_column_families);

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

618 619
    long num_locks = (max_key_ >> log2_keys_per_lock_);
    if (max_key_ & ((1 << log2_keys_per_lock_) - 1)) {
I
Igor Canadi 已提交
620 621 622 623
      num_locks++;
    }
    fprintf(stdout, "Creating %ld locks\n", num_locks * FLAGS_column_families);
    key_locks_.resize(FLAGS_column_families);
D
Dmitri Smirnov 已提交
624

I
Igor Canadi 已提交
625
    for (int i = 0; i < FLAGS_column_families; ++i) {
D
Dmitri Smirnov 已提交
626 627 628 629
      key_locks_[i].resize(num_locks);
      for (auto& ptr : key_locks_[i]) {
        ptr.reset(new port::Mutex);
      }
630 631 632
    }
  }

I
Igor Canadi 已提交
633
  ~SharedState() {}
634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658

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

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

  StressTest* GetStressTest() const {
    return stress_test_;
  }

  long GetMaxKey() const {
    return max_key_;
  }

  uint32_t GetNumThreads() const {
    return num_threads_;
  }

  void IncInitialized() {
    num_initialized_++;
  }

A
amayank 已提交
659
  void IncOperated() {
660 661 662 663 664 665 666
    num_populated_++;
  }

  void IncDone() {
    num_done_++;
  }

667 668 669 670
  void IncVotedReopen() {
    vote_reopen_ = (vote_reopen_ + 1) % num_threads_;
  }

671 672 673 674
  bool AllInitialized() const {
    return num_initialized_ >= num_threads_;
  }

A
amayank 已提交
675
  bool AllOperated() const {
676 677 678 679 680 681 682
    return num_populated_ >= num_threads_;
  }

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

683 684 685 686
  bool AllVotedReopen() {
    return (vote_reopen_ == 0);
  }

687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702
  void SetStart() {
    start_ = true;
  }

  void SetStartVerify() {
    start_verify_ = true;
  }

  bool Started() const {
    return start_;
  }

  bool VerifyStarted() const {
    return start_verify_;
  }

I
Igor Canadi 已提交
703 704 705 706
  void SetVerificationFailure() { verification_failure_.store(true); }

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

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

I
Igor Canadi 已提交
711 712
  void LockColumnFamily(int cf) {
    for (auto& mutex : key_locks_[cf]) {
D
Dmitri Smirnov 已提交
713
      mutex->Lock();
I
Igor Canadi 已提交
714
    }
715 716
  }

I
Igor Canadi 已提交
717 718
  void UnlockColumnFamily(int cf) {
    for (auto& mutex : key_locks_[cf]) {
D
Dmitri Smirnov 已提交
719
      mutex->Unlock();
I
Igor Canadi 已提交
720
    }
721 722
  }

I
Igor Canadi 已提交
723 724
  void ClearColumnFamily(int cf) {
    std::fill(values_[cf].begin(), values_[cf].end(), SENTINEL);
A
amayank 已提交
725 726
  }

I
Igor Canadi 已提交
727 728
  void Put(int cf, long key, uint32_t value_base) {
    values_[cf][key] = value_base;
729 730
  }

I
Igor Canadi 已提交
731 732 733 734 735 736
  uint32_t Get(int cf, long key) const { return values_[cf][key]; }

  void Delete(int cf, long key) { values_[cf][key] = SENTINEL; }

  uint32_t GetSeed() const { return seed_; }

737 738 739 740 741 742 743 744
  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_; }

745 746 747 748 749 750 751 752 753
 private:
  port::Mutex mu_;
  port::CondVar cv_;
  const uint32_t seed_;
  const long max_key_;
  const uint32_t log2_keys_per_lock_;
  const int num_threads_;
  long num_initialized_;
  long num_populated_;
754
  long vote_reopen_;
755 756 757
  long num_done_;
  bool start_;
  bool start_verify_;
758 759
  bool should_stop_bg_thread_;
  bool bg_thread_finished_;
760
  StressTest* stress_test_;
I
Igor Canadi 已提交
761
  std::atomic<bool> verification_failure_;
762

I
Igor Canadi 已提交
763
  std::vector<std::vector<uint32_t>> values_;
D
Dmitri Smirnov 已提交
764 765
  // 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 已提交
766
  std::vector<std::vector<std::unique_ptr<port::Mutex> > > key_locks_;
767 768
};

769 770
const uint32_t SharedState::SENTINEL = 0xffffffff;

771 772 773 774 775 776 777
// 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 已提交
778
  ThreadState(uint32_t index, SharedState* _shared)
I
Igor Canadi 已提交
779
      : tid(index), rand(1000 + index + _shared->GetSeed()), shared(_shared) {}
780 781
};

782 783 784 785
class DbStressListener : public EventListener {
 public:
  DbStressListener(
      const std::string& db_name,
Y
Yueh-Hsuan Chiang 已提交
786
      const std::vector<DbPath>& db_paths) :
787 788 789 790 791 792
      db_name_(db_name),
      db_paths_(db_paths),
      rand_(301) {}
  virtual ~DbStressListener() {}
#ifndef ROCKSDB_LITE
  virtual void OnFlushCompleted(
793
      DB* db, const FlushJobInfo& info) override {
794 795
    assert(db);
    assert(db->GetName() == db_name_);
796 797
    assert(IsValidColumnFamilyName(info.cf_name));
    VerifyFilePath(info.file_path);
798 799 800 801 802 803
    // pretending doing some work here
    std::this_thread::sleep_for(
        std::chrono::microseconds(rand_.Uniform(5000)));
  }

  virtual void OnCompactionCompleted(
804
      DB *db, const CompactionJobInfo& ci) override {
805 806 807 808 809 810 811 812 813 814 815 816 817 818 819
    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(
        std::chrono::microseconds(rand_.Uniform(5000)));
  }

820 821 822 823 824 825 826 827 828 829 830 831
  virtual void OnTableFileCreated(
      const TableFileCreationInfo& info) override {
    assert(info.db_name == db_name_);
    assert(IsValidColumnFamilyName(info.cf_name));
    VerifyFilePath(info.file_path);
    assert(info.file_size > 0);
    assert(info.job_id > 0);
    assert(info.table_properties.data_size > 0);
    assert(info.table_properties.raw_key_size > 0);
    assert(info.table_properties.num_entries > 0);
  }

832 833 834 835 836
 protected:
  bool IsValidColumnFamilyName(const std::string& cf_name) const {
    if (cf_name == kDefaultColumnFamilyName) {
      return true;
    }
Y
Yueh-Hsuan Chiang 已提交
837 838 839 840
    // 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;
841 842
      }
    }
Y
Yueh-Hsuan Chiang 已提交
843
    return true;
844 845 846
  }

  void VerifyFileDir(const std::string& file_dir) {
847
#ifndef NDEBUG
848 849 850 851 852 853 854 855 856
    if (db_name_ == file_dir) {
      return;
    }
    for (const auto& db_path : db_paths_) {
      if (db_path.path == file_dir) {
        return;
      }
    }
    assert(false);
857
#endif  // !NDEBUG
858 859 860
  }

  void VerifyFileName(const std::string& file_name) {
861
#ifndef NDEBUG
862 863 864 865 866
    uint64_t file_number;
    FileType file_type;
    bool result = ParseFileName(file_name, &file_number, &file_type);
    assert(result);
    assert(file_type == kTableFile);
867
#endif  // !NDEBUG
868 869 870
  }

  void VerifyFilePath(const std::string& file_path) {
871
#ifndef NDEBUG
872 873 874 875 876 877 878 879 880
    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));
    }
881
#endif  // !NDEBUG
882 883 884 885 886 887 888 889 890
  }
#endif  // !ROCKSDB_LITE

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

891 892 893 894 895 896
}  // namespace

class StressTest {
 public:
  StressTest()
      : cache_(NewLRUCache(FLAGS_cache_size)),
I
Igor Canadi 已提交
897 898 899
        compressed_cache_(FLAGS_compressed_cache_size >= 0
                              ? NewLRUCache(FLAGS_compressed_cache_size)
                              : nullptr),
900
        filter_policy_(FLAGS_bloom_bits >= 0
901 902 903 904
                   ? FLAGS_use_block_based_filter
                     ? NewBloomFilterPolicy(FLAGS_bloom_bits, true)
                     : NewBloomFilterPolicy(FLAGS_bloom_bits, false)
                   : nullptr),
905
        db_(nullptr),
906
        new_column_family_name_(1),
907
        num_times_reopened_(0) {
908 909 910 911 912
    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-")) {
913
          FLAGS_env->DeleteFile(FLAGS_db + "/" + files[i]);
914
        }
915
      }
916
      DestroyDB(FLAGS_db, Options());
917 918 919 920
    }
  }

  ~StressTest() {
I
Igor Canadi 已提交
921 922 923 924
    for (auto cf : column_families_) {
      delete cf;
    }
    column_families_.clear();
925 926 927
    delete db_;
  }

L
Lei Jin 已提交
928 929 930 931
  bool BuildOptionsTable() {
    if (FLAGS_set_options_one_in <= 0) {
      return true;
    }
D
Dmitri Smirnov 已提交
932

S
sdong 已提交
933 934 935
    std::unordered_map<std::string, std::vector<std::string> > options_tbl = {
        {"write_buffer_size",
         {ToString(FLAGS_write_buffer_size),
936
          ToString(FLAGS_write_buffer_size * 2),
S
sdong 已提交
937 938 939
          ToString(FLAGS_write_buffer_size * 4)}},
        {"max_write_buffer_number",
         {ToString(FLAGS_max_write_buffer_number),
940
          ToString(FLAGS_max_write_buffer_number * 2),
S
sdong 已提交
941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015
          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"}},
        {"memtable_prefix_bloom_huge_page_tlb_size",
         {"0", ToString(2 * 1024 * 1024)}},
        {"max_successive_merges", {"0", "2", "4"}},
        {"filter_deletes", {"0", "1"}},
        {"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),
         }},
        {"max_grandparent_overlap_factor",
         {
             ToString(Options().max_grandparent_overlap_factor - 5),
             ToString(Options().max_grandparent_overlap_factor),
             ToString(Options().max_grandparent_overlap_factor + 5),
         }},
        {"expanded_compaction_factor",
         {
             ToString(Options().expanded_compaction_factor - 5),
             ToString(Options().expanded_compaction_factor),
             ToString(Options().expanded_compaction_factor + 5),
         }},
        {"source_compaction_factor",
         {
             ToString(Options().source_compaction_factor),
             ToString(Options().source_compaction_factor * 2),
             ToString(Options().source_compaction_factor * 4),
         }},
        {"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"}},
L
Lei Jin 已提交
1016
    };
D
Dmitri Smirnov 已提交
1017 1018 1019

    options_table_ = std::move(options_tbl);

L
Lei Jin 已提交
1020 1021 1022 1023 1024 1025
    for (const auto& iter : options_table_) {
      options_index_.push_back(iter.first);
    }
    return true;
  }

I
Igor Canadi 已提交
1026
  bool Run() {
1027
    PrintEnv();
L
Lei Jin 已提交
1028
    BuildOptionsTable();
1029 1030 1031 1032 1033 1034 1035 1036 1037
    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]);
    }
1038 1039 1040 1041 1042
    ThreadState bg_thread(0, &shared);
    if (FLAGS_compaction_thread_pool_adjust_interval > 0) {
      FLAGS_env->StartThread(PoolSizeChangeThread, &bg_thread);
    }

1043
    // Each thread goes through the following states:
A
amayank 已提交
1044 1045
    // initializing -> wait for others to init -> read/populate/depopulate
    // wait for others to operate -> verify -> done
1046 1047 1048 1049 1050 1051 1052

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

1053 1054 1055
      double now = FLAGS_env->NowMicros();
      fprintf(stdout, "%s Starting database operations\n",
              FLAGS_env->TimeToString((uint64_t) now/1000000).c_str());
A
amayank 已提交
1056

1057 1058
      shared.SetStart();
      shared.GetCondVar()->SignalAll();
A
amayank 已提交
1059
      while (!shared.AllOperated()) {
1060 1061 1062
        shared.GetCondVar()->Wait();
      }

1063
      now = FLAGS_env->NowMicros();
1064 1065 1066 1067 1068 1069 1070
      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());
      }
1071

1072 1073 1074 1075 1076 1077 1078
      shared.SetStartVerify();
      shared.GetCondVar()->SignalAll();
      while (!shared.AllDone()) {
        shared.GetCondVar()->Wait();
      }
    }

1079
    for (unsigned int i = 1; i < n; i++) {
1080 1081 1082 1083
      threads[0]->stats.Merge(threads[i]->stats);
    }
    threads[0]->stats.Report("Stress Test");

1084
    for (unsigned int i = 0; i < n; i++) {
1085
      delete threads[i];
1086
      threads[i] = nullptr;
1087
    }
1088
    double now = FLAGS_env->NowMicros();
1089 1090 1091 1092
    if (!FLAGS_test_batches_snapshots) {
      fprintf(stdout, "%s Verification successful\n",
              FLAGS_env->TimeToString((uint64_t) now/1000000).c_str());
    }
1093
    PrintStatistics();
I
Igor Canadi 已提交
1094

1095 1096 1097 1098 1099 1100 1101 1102
    if (FLAGS_compaction_thread_pool_adjust_interval > 0) {
      MutexLock l(shared.GetMutex());
      shared.SetShouldStopBgThread();
      while (!shared.BgThreadFinished()) {
        shared.GetCondVar()->Wait();
      }
    }

I
Igor Canadi 已提交
1103 1104 1105 1106 1107
    if (shared.HasVerificationFailedYet()) {
      printf("Verification failed :(\n");
      return false;
    }
    return true;
1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125
  }

 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 已提交
1126
    thread->shared->GetStressTest()->OperateDb(thread);
1127 1128 1129

    {
      MutexLock l(shared->GetMutex());
A
amayank 已提交
1130 1131
      shared->IncOperated();
      if (shared->AllOperated()) {
1132 1133 1134 1135 1136 1137 1138
        shared->GetCondVar()->SignalAll();
      }
      while (!shared->VerifyStarted()) {
        shared->GetCondVar()->Wait();
      }
    }

1139
    if (!FLAGS_test_batches_snapshots) {
M
Mayank Agarwal 已提交
1140
      thread->shared->GetStressTest()->VerifyDb(thread);
1141
    }
1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152

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

  }

1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168
  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 已提交
1169
      auto thread_pool_size_var = FLAGS_compaction_thread_pool_variations;
1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184
      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);
    }
  }

1185 1186 1187
  // 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 已提交
1188 1189 1190
  Status MultiPut(ThreadState* thread, const WriteOptions& writeoptions,
                  ColumnFamilyHandle* column_family, const Slice& key,
                  const Slice& value, size_t sz) {
1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201
    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];
1202
      if (FLAGS_use_merge) {
I
Igor Canadi 已提交
1203
        batch.Merge(column_family, keys[i], value_slices[i]);
D
Deon Nicholas 已提交
1204
      } else {
I
Igor Canadi 已提交
1205
        batch.Put(column_family, keys[i], value_slices[i]);
D
Deon Nicholas 已提交
1206
      }
1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222
    }

    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 已提交
1223 1224
  Status MultiDelete(ThreadState* thread, const WriteOptions& writeoptions,
                     ColumnFamilyHandle* column_family, const Slice& key) {
1225 1226 1227 1228 1229 1230 1231
    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 已提交
1232
      batch.Delete(column_family, keys[i]);
1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249
    }

    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 已提交
1250 1251 1252
  Status MultiGet(ThreadState* thread, const ReadOptions& readoptions,
                  ColumnFamilyHandle* column_family, const Slice& key,
                  std::string* value) {
1253 1254 1255 1256 1257 1258 1259 1260 1261
    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 已提交
1262
      s = db_->Get(readoptionscopy, column_family, key_slices[i], value);
1263 1264 1265 1266 1267 1268 1269 1270
      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] = "";
1271
        thread->stats.AddGets(1, 0);
1272 1273 1274 1275 1276 1277
      } else {
        values[i] = *value;

        char expected_prefix = (keys[i])[0];
        char actual_prefix = (values[i])[0];
        if (actual_prefix != expected_prefix) {
1278
          fprintf(stderr, "error expected prefix = %c actual = %c\n",
1279 1280 1281
                  expected_prefix, actual_prefix);
        }
        (values[i])[0] = ' '; // blank out the differing character
1282
        thread->stats.AddGets(1, 1);
1283 1284 1285 1286 1287 1288 1289
      }
    }
    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]) {
1290
        fprintf(stderr, "error : inconsistent values for key %s: %s, %s\n",
1291 1292
                key.ToString(true).c_str(), StringToHex(values[0]).c_str(),
                StringToHex(values[i]).c_str());
1293 1294 1295 1296 1297 1298 1299 1300
      // we continue after error rather than exiting so that we can
      // find more errors if any
      }
    }

    return s;
  }

L
Lei Jin 已提交
1301 1302 1303 1304 1305
  // 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.
1306
  // ASSUMES that MultiPut was used to put (K, V)
I
Igor Canadi 已提交
1307 1308
  Status MultiPrefixScan(ThreadState* thread, const ReadOptions& readoptions,
                         ColumnFamilyHandle* column_family,
L
Lei Jin 已提交
1309
                         const Slice& key) {
1310 1311 1312 1313 1314 1315 1316 1317
    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 已提交
1318 1319 1320
      prefixes[i] += key.ToString();
      prefixes[i].resize(FLAGS_prefix_size);
      prefix_slices[i] = Slice(prefixes[i]);
1321 1322
      readoptionscopy[i] = readoptions;
      readoptionscopy[i].snapshot = snapshot;
I
Igor Canadi 已提交
1323
      iters[i] = db_->NewIterator(readoptionscopy[i], column_family);
I
Igor Canadi 已提交
1324
      iters[i]->Seek(prefix_slices[i]);
1325 1326 1327
    }

    int count = 0;
1328
    while (iters[0]->Valid() && iters[0]->key().starts_with(prefix_slices[0])) {
1329 1330 1331 1332 1333
      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
1334 1335
        assert(iters[i]->Valid() &&
               iters[i]->key().starts_with(prefix_slices[i]));
1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349
        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]) {
1350 1351 1352
          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());
1353 1354 1355 1356 1357 1358 1359 1360 1361 1362
          // 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
1363 1364
      assert(!iters[i]->Valid() ||
             !iters[i]->key().starts_with(prefix_slices[i]));
1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378
      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;
  }

1379 1380
  // 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 已提交
1381 1382
  Status MultiIterate(ThreadState* thread, const ReadOptions& readoptions,
                      ColumnFamilyHandle* column_family, const Slice& key) {
1383 1384 1385 1386
    Status s;
    const Snapshot* snapshot = db_->GetSnapshot();
    ReadOptions readoptionscopy = readoptions;
    readoptionscopy.snapshot = snapshot;
I
Igor Canadi 已提交
1387
    unique_ptr<Iterator> iter(db_->NewIterator(readoptionscopy, column_family));
1388 1389

    iter->Seek(key);
1390
    for (uint64_t i = 0; i < FLAGS_num_iterations && iter->Valid(); i++) {
1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408
      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;
  }

1409
  Status SetOptions(ThreadState* thread) {
L
Lei Jin 已提交
1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435
    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 已提交
1436
  void OperateDb(ThreadState* thread) {
1437 1438
    ReadOptions read_opts(FLAGS_verify_checksum, true);
    WriteOptions write_opts;
A
amayank 已提交
1439
    char value[100];
1440 1441 1442 1443 1444 1445
    long max_key = thread->shared->GetMaxKey();
    std::string from_db;
    if (FLAGS_sync) {
      write_opts.sync = true;
    }
    write_opts.disableWAL = FLAGS_disable_wal;
1446 1447 1448
    const int prefixBound = (int)FLAGS_readpercent + (int)FLAGS_prefixpercent;
    const int writeBound = prefixBound + (int)FLAGS_writepercent;
    const int delBound = writeBound + (int)FLAGS_delpercent;
1449 1450

    thread->stats.Start();
1451
    for (uint64_t i = 0; i < FLAGS_ops_per_thread; i++) {
I
Igor Canadi 已提交
1452 1453 1454 1455
      if (thread->shared->HasVerificationFailedYet()) {
        break;
      }
      if (i != 0 && (i % (FLAGS_ops_per_thread / (FLAGS_reopen + 1))) == 0) {
1456
        {
1457
          thread->stats.FinishedSingleOp();
1458 1459 1460 1461 1462 1463 1464 1465 1466
          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();
          }
1467 1468
          // Commenting this out as we don't want to reset stats on each open.
          // thread->stats.Start();
1469 1470
        }
      }
1471

L
Lei Jin 已提交
1472 1473 1474 1475 1476 1477
      // Change Options
      if (FLAGS_set_options_one_in > 0 &&
          thread->rand.OneIn(FLAGS_set_options_one_in)) {
        SetOptions(thread);
      }

1478 1479 1480 1481 1482
      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 已提交
1483
      if (!FLAGS_test_batches_snapshots &&
1484
          FLAGS_clear_column_family_one_in != 0 && FLAGS_column_families > 1) {
I
Igor Canadi 已提交
1485 1486 1487 1488
        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 =
1489
              ToString(new_column_family_name_.fetch_add(1));
I
Igor Canadi 已提交
1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500
          {
            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];
1501 1502 1503 1504 1505
          if (!s.ok()) {
            fprintf(stderr, "dropping column family error: %s\n",
                s.ToString().c_str());
            std::terminate();
          }
I
Igor Canadi 已提交
1506 1507 1508 1509
          s = db_->CreateColumnFamily(ColumnFamilyOptions(options_), new_name,
                                      &column_families_[cf]);
          column_family_names_[cf] = new_name;
          thread->shared->ClearColumnFamily(cf);
1510 1511 1512 1513 1514
          if (!s.ok()) {
            fprintf(stderr, "creating column family error: %s\n",
                s.ToString().c_str());
            std::terminate();
          }
I
Igor Canadi 已提交
1515 1516 1517 1518
          thread->shared->UnlockColumnFamily(cf);
        }
      }

1519
      long rand_key = thread->rand.Next() % max_key;
I
Igor Canadi 已提交
1520
      int rand_column_family = thread->rand.Next() % FLAGS_column_families;
1521 1522 1523
      std::string keystr = Key(rand_key);
      Slice key = keystr;
      int prob_op = thread->rand.Uniform(100);
I
Igor Canadi 已提交
1524 1525 1526 1527 1528 1529
      std::unique_ptr<MutexLock> l;
      if (!FLAGS_test_batches_snapshots) {
        l.reset(new MutexLock(
            thread->shared->GetMutexForKey(rand_column_family, rand_key)));
      }
      auto column_family = column_families_[rand_column_family];
1530 1531

      if (prob_op >= 0 && prob_op < (int)FLAGS_readpercent) {
1532
        // OPERATION read
1533
        if (!FLAGS_test_batches_snapshots) {
I
Igor Canadi 已提交
1534
          Status s = db_->Get(read_opts, column_family, key, &from_db);
1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545
          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 已提交
1546
          MultiGet(thread, read_opts, column_family, key, &from_db);
1547
        }
1548 1549
      } else if ((int)FLAGS_readpercent <= prob_op && prob_op < prefixBound) {
        // OPERATION prefix scan
L
Lei Jin 已提交
1550 1551
        // 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 已提交
1552 1553
        // be 2 ^ ((8 - FLAGS_prefix_size) * 8) possible keys with the same
        // prefix
1554
        if (!FLAGS_test_batches_snapshots) {
L
Lei Jin 已提交
1555
          Slice prefix = Slice(key.data(), FLAGS_prefix_size);
I
Igor Canadi 已提交
1556
          Iterator* iter = db_->NewIterator(read_opts, column_family);
L
Lei Jin 已提交
1557
          int64_t count = 0;
I
Igor Canadi 已提交
1558 1559
          for (iter->Seek(prefix);
               iter->Valid() && iter->key().starts_with(prefix); iter->Next()) {
L
Lei Jin 已提交
1560
            ++count;
1561
          }
I
Igor Canadi 已提交
1562 1563
          assert(count <=
                 (static_cast<int64_t>(1) << ((8 - FLAGS_prefix_size) * 8)));
1564
          if (iter->status().ok()) {
1565
            thread->stats.AddPrefixes(1, static_cast<int>(count));
1566 1567 1568 1569
          } else {
            thread->stats.AddErrors(1);
          }
          delete iter;
1570
        } else {
1571
          MultiPrefixScan(thread, read_opts, column_family, key);
A
amayank 已提交
1572
        }
1573 1574
      } else if (prefixBound <= prob_op && prob_op < writeBound) {
        // OPERATION write
1575 1576 1577
        uint32_t value_base = thread->rand.Next();
        size_t sz = GenerateValue(value_base, value, sizeof(value));
        Slice v(value, sz);
1578
        if (!FLAGS_test_batches_snapshots) {
1579
          if (FLAGS_verify_before_write) {
I
Igor Canadi 已提交
1580 1581
            std::string keystr2 = Key(rand_key);
            Slice k = keystr2;
I
Igor Canadi 已提交
1582
            Status s = db_->Get(read_opts, column_family, k, &from_db);
I
Igor Canadi 已提交
1583 1584 1585 1586
            if (VerifyValue(rand_column_family, rand_key, read_opts,
                            thread->shared, from_db, s, true) == false) {
              break;
            }
1587
          }
I
Igor Canadi 已提交
1588
          thread->shared->Put(rand_column_family, rand_key, value_base);
1589
          Status s;
1590
          if (FLAGS_use_merge) {
1591
            s = db_->Merge(write_opts, column_family, key, v);
D
Deon Nicholas 已提交
1592
          } else {
1593 1594 1595 1596 1597
            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 已提交
1598
          }
1599 1600
          thread->stats.AddBytesForWrites(1, sz);
        } else {
I
Igor Canadi 已提交
1601
          MultiPut(thread, write_opts, column_family, key, v, sz);
1602
        }
1603 1604
        PrintKeyValue(rand_column_family, static_cast<uint32_t>(rand_key),
                      value, sz);
1605 1606
      } else if (writeBound <= prob_op && prob_op < delBound) {
        // OPERATION delete
1607
        if (!FLAGS_test_batches_snapshots) {
I
Igor Canadi 已提交
1608
          thread->shared->Delete(rand_column_family, rand_key);
1609
          Status s = db_->Delete(write_opts, column_family, key);
1610
          thread->stats.AddDeletes(1);
1611 1612 1613 1614
          if (!s.ok()) {
            fprintf(stderr, "delete error: %s\n", s.ToString().c_str());
            std::terminate();
          }
1615
        } else {
I
Igor Canadi 已提交
1616
          MultiDelete(thread, write_opts, column_family, key);
1617
        }
1618 1619
      } else {
        // OPERATION iterate
I
Igor Canadi 已提交
1620
        MultiIterate(thread, read_opts, column_family, key);
1621
      }
1622 1623
      thread->stats.FinishedSingleOp();
    }
1624

1625 1626 1627
    thread->stats.Stop();
  }

M
Mayank Agarwal 已提交
1628
  void VerifyDb(ThreadState* thread) const {
1629
    ReadOptions options(FLAGS_verify_checksum, true);
I
Igor Canadi 已提交
1630 1631 1632
    auto shared = thread->shared;
    static const long max_key = shared->GetMaxKey();
    static const long keys_per_thread = max_key / shared->GetNumThreads();
M
Mayank Agarwal 已提交
1633
    long start = keys_per_thread * thread->tid;
1634
    long end = start + keys_per_thread;
I
Igor Canadi 已提交
1635
    if (thread->tid == shared->GetNumThreads() - 1) {
1636 1637
      end = max_key;
    }
I
Igor Canadi 已提交
1638
    for (size_t cf = 0; cf < column_families_.size(); ++cf) {
I
Igor Canadi 已提交
1639 1640 1641
      if (thread->shared->HasVerificationFailedYet()) {
        break;
      }
I
Igor Canadi 已提交
1642 1643 1644 1645 1646 1647
      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));
        for (long i = start; i < end; i++) {
I
Igor Canadi 已提交
1648 1649 1650
          if (thread->shared->HasVerificationFailedYet()) {
            break;
          }
1651 1652 1653 1654 1655 1656
          // 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 已提交
1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667
          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) {
1668 1669
              VerificationAbort(shared, "An out of range key was found",
                                static_cast<int>(cf), i);
I
Igor Canadi 已提交
1670 1671 1672 1673
            }
          } 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 已提交
1674 1675
            s = Status::NotFound(Slice());
          }
1676 1677
          VerifyValue(static_cast<int>(cf), i, options, shared, from_db, s,
                      true);
I
Igor Canadi 已提交
1678
          if (from_db.length()) {
1679 1680
            PrintKeyValue(static_cast<int>(cf), static_cast<uint32_t>(i),
                          from_db.data(), from_db.length());
I
Igor Canadi 已提交
1681
          }
M
Mayank Agarwal 已提交
1682
        }
I
Igor Canadi 已提交
1683 1684 1685
      } else {
        // Use Get to verify this range
        for (long i = start; i < end; i++) {
I
Igor Canadi 已提交
1686 1687 1688
          if (thread->shared->HasVerificationFailedYet()) {
            break;
          }
I
Igor Canadi 已提交
1689 1690 1691 1692
          std::string from_db;
          std::string keystr = Key(i);
          Slice k = keystr;
          Status s = db_->Get(options, column_families_[cf], k, &from_db);
1693 1694
          VerifyValue(static_cast<int>(cf), i, options, shared, from_db, s,
                      true);
I
Igor Canadi 已提交
1695
          if (from_db.length()) {
1696 1697
            PrintKeyValue(static_cast<int>(cf), static_cast<uint32_t>(i),
                          from_db.data(), from_db.length());
I
Igor Canadi 已提交
1698
          }
M
Mayank Agarwal 已提交
1699
        }
1700 1701 1702 1703
      }
    }
  }

I
Igor Canadi 已提交
1704 1705 1706 1707 1708
  void VerificationAbort(SharedState* shared, std::string msg, int cf,
                         long key) const {
    printf("Verification failed for column family %d key %ld: %s\n", cf, key,
           msg.c_str());
    shared->SetVerificationFailure();
1709 1710
  }

I
Igor Canadi 已提交
1711 1712
  bool VerifyValue(int cf, long key, const ReadOptions& opts,
                   SharedState* shared, const std::string& value_from_db,
I
Igor Canadi 已提交
1713
                   Status s, bool strict = false) const {
I
Igor Canadi 已提交
1714 1715 1716
    if (shared->HasVerificationFailedYet()) {
      return false;
    }
1717
    // compare value_from_db with the value in the shared state
A
amayank 已提交
1718
    char value[100];
I
Igor Canadi 已提交
1719
    uint32_t value_base = shared->Get(cf, key);
1720
    if (value_base == SharedState::SENTINEL && !strict) {
I
Igor Canadi 已提交
1721
      return true;
1722 1723
    }

1724
    if (s.ok()) {
1725
      if (value_base == SharedState::SENTINEL) {
I
Igor Canadi 已提交
1726 1727
        VerificationAbort(shared, "Unexpected value found", cf, key);
        return false;
1728
      }
A
amayank 已提交
1729
      size_t sz = GenerateValue(value_base, value, sizeof(value));
1730
      if (value_from_db.length() != sz) {
I
Igor Canadi 已提交
1731 1732
        VerificationAbort(shared, "Length of value read is not equal", cf, key);
        return false;
1733
      }
1734
      if (memcmp(value_from_db.data(), value, sz) != 0) {
I
Igor Canadi 已提交
1735 1736 1737
        VerificationAbort(shared, "Contents of value read don't match", cf,
                          key);
        return false;
1738 1739 1740
      }
    } else {
      if (value_base != SharedState::SENTINEL) {
I
Igor Canadi 已提交
1741 1742
        VerificationAbort(shared, "Value not found: " + s.ToString(), cf, key);
        return false;
1743 1744
      }
    }
I
Igor Canadi 已提交
1745
    return true;
1746 1747
  }

I
Igor Canadi 已提交
1748 1749 1750 1751 1752 1753 1754
  static void PrintKeyValue(int cf, uint32_t key, const char* value,
                            size_t sz) {
    if (!FLAGS_verbose) {
      return;
    }
    fprintf(stdout, "[CF %d] %u ==> (%u) ", cf, key, (unsigned int)sz);
    for (size_t i = 0; i < sz; i++) {
1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765 1766
      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);
    }
1767
    v[value_sz] = '\0';
1768 1769 1770 1771
    return value_sz; // the size of the value set.
  }

  void PrintEnv() const {
I
Igor Canadi 已提交
1772 1773 1774 1775 1776 1777 1778
    fprintf(stdout, "RocksDB version     : %d.%d\n", kMajorVersion,
            kMinorVersion);
    fprintf(stdout, "Column families     : %d\n", FLAGS_column_families);
    if (!FLAGS_test_batches_snapshots) {
      fprintf(stdout, "Clear CFs one in    : %d\n",
              FLAGS_clear_column_family_one_in);
    }
1779
    fprintf(stdout, "Number of threads   : %d\n", FLAGS_threads);
K
kailiu 已提交
1780 1781 1782
    fprintf(stdout,
            "Ops per thread      : %lu\n",
            (unsigned long)FLAGS_ops_per_thread);
1783 1784 1785 1786 1787
    std::string ttl_state("unused");
    if (FLAGS_ttl > 0) {
      ttl_state = NumberToString(FLAGS_ttl);
    }
    fprintf(stdout, "Time to live(sec)   : %s\n", ttl_state.c_str());
K
Kai Liu 已提交
1788 1789 1790 1791 1792
    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);
    fprintf(stdout, "Iterate percentage  : %d%%\n", FLAGS_iterpercent);
1793 1794
    fprintf(stdout, "DB-write-buffer-size: %" PRIu64 "\n",
        FLAGS_db_write_buffer_size);
1795
    fprintf(stdout, "Write-buffer-size   : %d\n", FLAGS_write_buffer_size);
K
kailiu 已提交
1796 1797 1798 1799 1800 1801
    fprintf(stdout,
            "Iterations          : %lu\n",
            (unsigned long)FLAGS_num_iterations);
    fprintf(stdout,
            "Max key             : %lu\n",
            (unsigned long)FLAGS_max_key);
X
Xing Jin 已提交
1802 1803
    fprintf(stdout, "Ratio #ops/#keys    : %f\n",
            (1.0 * FLAGS_ops_per_thread * FLAGS_threads)/FLAGS_max_key);
1804
    fprintf(stdout, "Num times DB reopens: %d\n", FLAGS_reopen);
1805 1806
    fprintf(stdout, "Batches/snapshots   : %d\n",
            FLAGS_test_batches_snapshots);
1807
    fprintf(stdout, "Deletes use filter  : %d\n",
1808
            FLAGS_filter_deletes);
1809 1810
    fprintf(stdout, "Do update in place  : %d\n",
            FLAGS_in_place_update);
1811
    fprintf(stdout, "Num keys per lock   : %d\n",
1812 1813
            1 << FLAGS_log2_keys_per_lock);

1814
    const char* compression = "";
1815
    switch (FLAGS_compression_type_e) {
1816
      case rocksdb::kNoCompression:
1817
        compression = "none";
1818
        break;
1819
      case rocksdb::kSnappyCompression:
1820
        compression = "snappy";
1821
        break;
1822
      case rocksdb::kZlibCompression:
1823
        compression = "zlib";
1824
        break;
1825
      case rocksdb::kBZip2Compression:
1826
        compression = "bzip2";
1827
        break;
A
Albert Strasheim 已提交
1828 1829
      case rocksdb::kLZ4Compression:
        compression = "lz4";
1830
        break;
A
Albert Strasheim 已提交
1831 1832 1833 1834
      case rocksdb::kLZ4HCCompression:
        compression = "lz4hc";
        break;
      }
1835 1836

    fprintf(stdout, "Compression         : %s\n", compression);
J
Jim Paton 已提交
1837 1838 1839 1840 1841 1842

    const char* memtablerep = "";
    switch (FLAGS_rep_factory) {
      case kSkipList:
        memtablerep = "skip_list";
        break;
I
Igor Canadi 已提交
1843
      case kHashSkipList:
J
Jim Paton 已提交
1844 1845 1846 1847 1848 1849 1850 1851 1852
        memtablerep = "prefix_hash";
        break;
      case kVectorRep:
        memtablerep = "vector";
        break;
    }

    fprintf(stdout, "Memtablerep         : %s\n", memtablerep);

1853 1854 1855 1856
    fprintf(stdout, "------------------------------------------------\n");
  }

  void Open() {
1857
    assert(db_ == nullptr);
1858 1859 1860 1861
    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;
1862
    block_based_options.format_version = 2;
1863 1864 1865
    block_based_options.filter_policy = filter_policy_;
    options_.table_factory.reset(
        NewBlockBasedTableFactory(block_based_options));
1866
    options_.db_write_buffer_size = FLAGS_db_write_buffer_size;
I
Igor Canadi 已提交
1867 1868 1869 1870
    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;
1871 1872
    options_.max_write_buffer_number_to_maintain =
        FLAGS_max_write_buffer_number_to_maintain;
I
Igor Canadi 已提交
1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883
    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_.disableDataSync = FLAGS_disable_data_sync;
    options_.use_fsync = FLAGS_use_fsync;
    options_.allow_mmap_reads = FLAGS_mmap_read;
1884
    rocksdb_kill_odds = FLAGS_kill_random_test;
I
Igor Canadi 已提交
1885 1886 1887 1888
    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 =
1889
        FLAGS_max_bytes_for_level_multiplier;
I
Igor Canadi 已提交
1890 1891 1892 1893 1894 1895 1896 1897 1898
    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;
    options_.filter_deletes = FLAGS_filter_deletes;
1899
    options_.inplace_update_support = FLAGS_in_place_update;
I
Igor Canadi 已提交
1900
    if ((FLAGS_prefix_size == 0) == (FLAGS_rep_factory == kHashSkipList)) {
J
Jim Paton 已提交
1901 1902 1903 1904 1905 1906 1907 1908
      fprintf(stderr,
            "prefix_size should be non-zero iff memtablerep == prefix_hash\n");
      exit(1);
    }
    switch (FLAGS_rep_factory) {
      case kSkipList:
        // no need to do anything
        break;
S
sdong 已提交
1909 1910 1911 1912
#ifndef ROCKSDB_LITE
      case kHashSkipList:
        options_.memtable_factory.reset(NewHashSkipListRepFactory(10000));
        break;
J
Jim Paton 已提交
1913
      case kVectorRep:
I
Igor Canadi 已提交
1914
        options_.memtable_factory.reset(new VectorRepFactory());
J
Jim Paton 已提交
1915
        break;
S
sdong 已提交
1916 1917 1918 1919 1920 1921
#else
      default:
        fprintf(stderr,
                "RocksdbLite only supports skip list mem table. Skip "
                "--rep_factory\n");
#endif  // ROCKSDB_LITE
J
Jim Paton 已提交
1922
    }
S
sdong 已提交
1923

1924
    if (FLAGS_use_merge) {
I
Igor Canadi 已提交
1925
      options_.merge_operator = MergeOperators::CreatePutOperator();
D
Deon Nicholas 已提交
1926 1927
    }

1928 1929
    // set universal style compaction configurations, if applicable
    if (FLAGS_universal_size_ratio != 0) {
I
Igor Canadi 已提交
1930 1931
      options_.compaction_options_universal.size_ratio =
          FLAGS_universal_size_ratio;
1932 1933
    }
    if (FLAGS_universal_min_merge_width != 0) {
I
Igor Canadi 已提交
1934 1935
      options_.compaction_options_universal.min_merge_width =
          FLAGS_universal_min_merge_width;
1936 1937
    }
    if (FLAGS_universal_max_merge_width != 0) {
I
Igor Canadi 已提交
1938 1939
      options_.compaction_options_universal.max_merge_width =
          FLAGS_universal_max_merge_width;
1940 1941
    }
    if (FLAGS_universal_max_size_amplification_percent != 0) {
I
Igor Canadi 已提交
1942 1943
      options_.compaction_options_universal.max_size_amplification_percent =
          FLAGS_universal_max_size_amplification_percent;
1944 1945
    }

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

1948 1949
    Status s;
    if (FLAGS_ttl == -1) {
I
Igor Canadi 已提交
1950 1951 1952 1953 1954 1955 1956
      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());
1957
        column_family_names_.push_back(kDefaultColumnFamilyName);
I
Igor Canadi 已提交
1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979 1980 1981 1982 1983 1984
      } 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_;
        sort(sorted_cfn.begin(), sorted_cfn.end());
        sort(existing_column_families.begin(), existing_column_families.end());
        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_) {
1985
        if (name != kDefaultColumnFamilyName) {
I
Igor Canadi 已提交
1986 1987 1988 1989 1990
          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 已提交
1991
      while (cf_descriptors.size() < (size_t)FLAGS_column_families) {
1992
        std::string name = ToString(new_column_family_name_.load());
I
Igor Canadi 已提交
1993 1994 1995 1996
        new_column_family_name_++;
        cf_descriptors.emplace_back(name, ColumnFamilyOptions(options_));
        column_family_names_.push_back(name);
      }
1997 1998
      options_.listeners.clear();
      options_.listeners.emplace_back(
Y
Yueh-Hsuan Chiang 已提交
1999
          new DbStressListener(FLAGS_db, options_.db_paths));
I
Igor Canadi 已提交
2000
      options_.create_missing_column_families = true;
I
Igor Canadi 已提交
2001 2002 2003 2004
      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));
2005
    } else {
S
sdong 已提交
2006
#ifndef ROCKSDB_LITE
I
Igor Canadi 已提交
2007 2008 2009
      DBWithTTL* db_with_ttl;
      s = DBWithTTL::Open(options_, FLAGS_db, &db_with_ttl, FLAGS_ttl);
      db_ = db_with_ttl;
S
sdong 已提交
2010 2011 2012 2013
#else
      fprintf(stderr, "TTL is not supported in RocksDBLite\n");
      exit(1);
#endif
2014
    }
2015 2016 2017 2018 2019 2020
    if (!s.ok()) {
      fprintf(stderr, "open error: %s\n", s.ToString().c_str());
      exit(1);
    }
  }

2021
  void Reopen() {
I
Igor Canadi 已提交
2022 2023
    for (auto cf : column_families_) {
      delete cf;
2024
    }
I
Igor Canadi 已提交
2025 2026
    column_families_.clear();
    delete db_;
2027
    db_ = nullptr;
2028 2029 2030 2031 2032 2033

    num_times_reopened_++;
    double now = FLAGS_env->NowMicros();
    fprintf(stdout, "%s Reopening database for the %dth time\n",
            FLAGS_env->TimeToString((uint64_t) now/1000000).c_str(),
            num_times_reopened_);
2034 2035 2036
    Open();
  }

2037 2038
  void PrintStatistics() {
    if (dbstats) {
2039
      fprintf(stdout, "STATISTICS:\n%s\n", dbstats->ToString().c_str());
2040 2041 2042 2043
    }
  }

 private:
2044 2045 2046
  std::shared_ptr<Cache> cache_;
  std::shared_ptr<Cache> compressed_cache_;
  std::shared_ptr<const FilterPolicy> filter_policy_;
2047
  DB* db_;
I
Igor Canadi 已提交
2048 2049 2050 2051
  Options options_;
  std::vector<ColumnFamilyHandle*> column_families_;
  std::vector<std::string> column_family_names_;
  std::atomic<int> new_column_family_name_;
2052
  int num_times_reopened_;
L
Lei Jin 已提交
2053 2054
  std::unordered_map<std::string, std::vector<std::string>> options_table_;
  std::vector<std::string> options_index_;
2055 2056
};

2057
}  // namespace rocksdb
2058 2059

int main(int argc, char** argv) {
2060 2061 2062
  SetUsageMessage(std::string("\nUSAGE:\n") + std::string(argv[0]) +
                  " [OPTIONS]...");
  ParseCommandLineFlags(&argc, &argv, true);
2063 2064 2065 2066 2067 2068 2069 2070

  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);
2071
  }
2072
  FLAGS_rep_factory = StringToRepFactory(FLAGS_memtablerep.c_str());
2073

2074 2075 2076 2077
  // 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 已提交
2078 2079 2080 2081 2082 2083 2084 2085 2086 2087 2088 2089
  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);
  }
2090
  if ((FLAGS_readpercent + FLAGS_prefixpercent +
2091 2092 2093
       FLAGS_writepercent + FLAGS_delpercent + FLAGS_iterpercent) != 100) {
      fprintf(stderr,
              "Error: Read+Prefix+Write+Delete+Iterate percents != 100!\n");
A
amayank 已提交
2094 2095
      exit(1);
  }
2096 2097 2098 2099
  if (FLAGS_disable_wal == 1 && FLAGS_reopen > 0) {
      fprintf(stderr, "Error: Db cannot reopen safely with disable_wal set!\n");
      exit(1);
  }
2100
  if ((unsigned)FLAGS_reopen >= FLAGS_ops_per_thread) {
K
kailiu 已提交
2101 2102 2103 2104 2105
      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);
2106 2107 2108
      exit(1);
  }

2109
  // Choose a location for the test database if none given with --db=<path>
2110 2111
  if (FLAGS_db.empty()) {
      std::string default_db_path;
2112
      rocksdb::Env::Default()->GetTestDirectory(&default_db_path);
2113
      default_db_path += "/dbstress";
2114
      FLAGS_db = default_db_path;
2115 2116
  }

2117
  rocksdb::StressTest stress;
I
Igor Canadi 已提交
2118 2119 2120 2121 2122
  if (stress.Run()) {
    return 0;
  } else {
    return 1;
  }
2123
}
2124 2125

#endif  // GFLAGS