db_stress.cc 72.1 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,
I
Igor Canadi 已提交
89
            "If set, the test uses MultiGet(), Multiut() 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,
195 196 197
             "Range of bakground thread pool size variations when adjusted "
             "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 278
DEFINE_int32(target_file_size_multiplier, 1,
             "A multiplier to compute targe 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 305

DEFINE_int32(writepercent, 45,
             " Ratio of deletes 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 364
DEFINE_int32(purge_redundant_percent, 50,
             "Percentage of times we want to purge redundant keys in memory "
             "before flushing");
I
Igor Canadi 已提交
365
static const bool FLAGS_purge_redundant_percent_dummy __attribute__((unused)) =
366 367
    RegisterFlagValidator(&FLAGS_purge_redundant_percent,
                          &ValidateInt32Percent);
368

369 370
DEFINE_bool(filter_deletes, false, "On true, deletes use KeyMayExist to drop"
            " the delete if key not present");
371

372 373
DEFINE_bool(in_place_update, false, "On true, does inplace update in memtable");

J
Jim Paton 已提交
374 375
enum RepFactory {
  kSkipList,
I
Igor Canadi 已提交
376
  kHashSkipList,
J
Jim Paton 已提交
377 378
  kVectorRep
};
I
Igor Canadi 已提交
379 380

namespace {
381 382 383 384 385 386
enum RepFactory StringToRepFactory(const char* ctype) {
  assert(ctype);

  if (!strcasecmp(ctype, "skip_list"))
    return kSkipList;
  else if (!strcasecmp(ctype, "prefix_hash"))
I
Igor Canadi 已提交
387
    return kHashSkipList;
388 389 390 391 392 393
  else if (!strcasecmp(ctype, "vector"))
    return kVectorRep;

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

J
Jim Paton 已提交
396
static enum RepFactory FLAGS_rep_factory;
L
Lei Jin 已提交
397
DEFINE_string(memtablerep, "prefix_hash", "");
J
Jim Paton 已提交
398

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

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

D
Deon Nicholas 已提交
414

415
namespace rocksdb {
416

417 418 419 420 421 422 423 424 425 426 427 428 429
// 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;
}

430 431 432 433 434 435 436 437 438 439 440
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;
}


441 442 443 444 445 446 447 448 449
class StressTest;
namespace {

class Stats {
 private:
  double start_;
  double finish_;
  double seconds_;
  long done_;
450 451
  long gets_;
  long prefixes_;
452
  long writes_;
A
amayank 已提交
453
  long deletes_;
454
  long iterator_size_sums_;
455
  long founds_;
456
  long iterations_;
457
  long errors_;
458 459 460
  int next_report_;
  size_t bytes_;
  double last_op_finish_;
461
  HistogramImpl hist_;
462 463 464 465 466 467 468 469

 public:
  Stats() { }

  void Start() {
    next_report_ = 100;
    hist_.Clear();
    done_ = 0;
470 471
    gets_ = 0;
    prefixes_ = 0;
472
    writes_ = 0;
A
amayank 已提交
473
    deletes_ = 0;
474
    iterator_size_sums_ = 0;
475
    founds_ = 0;
476
    iterations_ = 0;
477
    errors_ = 0;
478 479 480 481 482 483 484 485 486 487
    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_;
488 489
    gets_ += other.gets_;
    prefixes_ += other.prefixes_;
490
    writes_ += other.writes_;
A
amayank 已提交
491
    deletes_ += other.deletes_;
492
    iterator_size_sums_ += other.iterator_size_sums_;
493
    founds_ += other.founds_;
494
    iterations_ += other.iterations_;
495
    errors_ += other.errors_;
496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512
    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) {
513
        fprintf(stdout, "long op: %.1f micros%30s\r", micros, "");
514 515 516 517
      }
      last_op_finish_ = now;
    }

518
      done_++;
519
    if (FLAGS_progress_reports) {
520 521 522 523 524 525 526 527 528 529
      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_, "");
      }
530 531 532
    }
  }

533 534 535 536 537
  void AddBytesForWrites(int nwrites, size_t nbytes) {
    writes_ += nwrites;
    bytes_ += nbytes;
  }

538 539 540
  void AddGets(int ngets, int nfounds) {
    founds_ += nfounds;
    gets_ += ngets;
541 542
  }

543 544 545 546 547
  void AddPrefixes(int nprefixes, int count) {
    prefixes_ += nprefixes;
    iterator_size_sums_ += count;
  }

548 549 550 551
  void AddIterations(int n) {
    iterations_ += n;
  }

552 553 554 555
  void AddDeletes(int n) {
    deletes_ += n;
  }

556 557
  void AddErrors(int n) {
    errors_ += n;
A
amayank 已提交
558 559
  }

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

    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:
597
  static const uint32_t SENTINEL;
598

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

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

625 626
    long num_locks = (max_key_ >> log2_keys_per_lock_);
    if (max_key_ & ((1 << log2_keys_per_lock_) - 1)) {
I
Igor Canadi 已提交
627 628 629 630 631 632
      num_locks++;
    }
    fprintf(stdout, "Creating %ld locks\n", num_locks * FLAGS_column_families);
    key_locks_.resize(FLAGS_column_families);
    for (int i = 0; i < FLAGS_column_families; ++i) {
      key_locks_[i] = std::vector<port::Mutex>(num_locks);
633 634 635
    }
  }

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

  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 已提交
662
  void IncOperated() {
663 664 665 666 667 668 669
    num_populated_++;
  }

  void IncDone() {
    num_done_++;
  }

670 671 672 673
  void IncVotedReopen() {
    vote_reopen_ = (vote_reopen_ + 1) % num_threads_;
  }

674 675 676 677
  bool AllInitialized() const {
    return num_initialized_ >= num_threads_;
  }

A
amayank 已提交
678
  bool AllOperated() const {
679 680 681 682 683 684 685
    return num_populated_ >= num_threads_;
  }

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

686 687 688 689
  bool AllVotedReopen() {
    return (vote_reopen_ == 0);
  }

690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705
  void SetStart() {
    start_ = true;
  }

  void SetStartVerify() {
    start_verify_ = true;
  }

  bool Started() const {
    return start_;
  }

  bool VerifyStarted() const {
    return start_verify_;
  }

I
Igor Canadi 已提交
706 707 708 709
  void SetVerificationFailure() { verification_failure_.store(true); }

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

I
Igor Canadi 已提交
710 711
  port::Mutex* GetMutexForKey(int cf, long key) {
    return &key_locks_[cf][key >> log2_keys_per_lock_];
712 713
  }

I
Igor Canadi 已提交
714 715 716 717
  void LockColumnFamily(int cf) {
    for (auto& mutex : key_locks_[cf]) {
      mutex.Lock();
    }
718 719
  }

I
Igor Canadi 已提交
720 721 722 723
  void UnlockColumnFamily(int cf) {
    for (auto& mutex : key_locks_[cf]) {
      mutex.Unlock();
    }
724 725
  }

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

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

I
Igor Canadi 已提交
734 735 736 737 738 739
  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_; }

740 741 742 743 744 745 746 747
  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_; }

748 749 750 751 752 753 754 755 756
 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_;
757
  long vote_reopen_;
758 759 760
  long num_done_;
  bool start_;
  bool start_verify_;
761 762
  bool should_stop_bg_thread_;
  bool bg_thread_finished_;
763
  StressTest* stress_test_;
I
Igor Canadi 已提交
764
  std::atomic<bool> verification_failure_;
765

I
Igor Canadi 已提交
766 767
  std::vector<std::vector<uint32_t>> values_;
  std::vector<std::vector<port::Mutex>> key_locks_;
768 769
};

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

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

783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809
class DbStressListener : public EventListener {
 public:
  DbStressListener(
      const std::string& db_name,
      const std::vector<DbPath>& db_paths,
      const std::vector<ColumnFamilyDescriptor>& cf_descs) :
      db_name_(db_name),
      db_paths_(db_paths),
      cf_descs_(cf_descs),
      rand_(301) {}
  virtual ~DbStressListener() {}
#ifndef ROCKSDB_LITE
  virtual void OnFlushCompleted(
      DB* db, const std::string& column_family_name,
      const std::string& file_path,
      bool triggered_writes_slowdown,
      bool triggered_writes_stop) override {
    assert(db);
    assert(db->GetName() == db_name_);
    assert(IsValidColumnFamilyName(column_family_name));
    VerifyFilePath(file_path);
    // pretending doing some work here
    std::this_thread::sleep_for(
        std::chrono::microseconds(rand_.Uniform(5000)));
  }

  virtual void OnCompactionCompleted(
810
      DB *db, const CompactionJobInfo& ci) override {
811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886
    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)));
  }

 protected:
  bool IsValidColumnFamilyName(const std::string& cf_name) const {
    if (cf_name == kDefaultColumnFamilyName) {
      return true;
    }
    for (const auto& cf_desc : cf_descs_) {
      if (cf_desc.name == cf_name) {
        return true;
      }
    }
    fprintf(stderr,
        "Unable to find the matched column family name "
        "for CF: %s.  Existing CF names are:\n",
        cf_name.c_str());
    for (const auto& cf_desc : cf_descs_) {
      fprintf(stderr, "  %s\n", cf_desc.name.c_str());
    }
    fflush(stderr);
    return false;
  }

  void VerifyFileDir(const std::string& file_dir) {
    if (db_name_ == file_dir) {
      return;
    }
    for (const auto& db_path : db_paths_) {
      if (db_path.path == file_dir) {
        return;
      }
    }
    assert(false);
  }

  void VerifyFileName(const std::string& file_name) {
    uint64_t file_number;
    FileType file_type;
    bool result = ParseFileName(file_name, &file_number, &file_type);
    assert(result);
    assert(file_type == kTableFile);
  }

  void VerifyFilePath(const std::string& file_path) {
    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));
    }
  }
#endif  // !ROCKSDB_LITE

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

887 888 889 890 891 892
}  // namespace

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

  ~StressTest() {
I
Igor Canadi 已提交
917 918 919 920
    for (auto cf : column_families_) {
      delete cf;
    }
    column_families_.clear();
921 922 923
    delete db_;
  }

L
Lei Jin 已提交
924 925 926 927 928 929 930
  bool BuildOptionsTable() {
    if (FLAGS_set_options_one_in <= 0) {
      return true;
    }
    options_table_ = {
      {"write_buffer_size",
        {
931 932 933
          ToString(FLAGS_write_buffer_size),
          ToString(FLAGS_write_buffer_size * 2),
          ToString(FLAGS_write_buffer_size * 4)
L
Lei Jin 已提交
934 935 936 937
        }
      },
      {"max_write_buffer_number",
        {
938 939 940
          ToString(FLAGS_max_write_buffer_number),
          ToString(FLAGS_max_write_buffer_number * 2),
          ToString(FLAGS_max_write_buffer_number * 4)
L
Lei Jin 已提交
941 942 943 944
        }
      },
      {"arena_block_size",
        {
945 946 947
          ToString(Options().arena_block_size),
          ToString(FLAGS_write_buffer_size / 4),
          ToString(FLAGS_write_buffer_size / 8),
L
Lei Jin 已提交
948 949 950 951 952 953 954
        }
      },
      {"memtable_prefix_bloom_bits", {"0", "8", "10"}},
      {"memtable_prefix_bloom_probes", {"4", "5", "6"}},
      {"memtable_prefix_bloom_huge_page_tlb_size",
        {
          "0",
955
          ToString(2 * 1024 * 1024)
L
Lei Jin 已提交
956 957 958 959 960 961 962 963 964 965 966
        }
      },
      {"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",
        {
967 968 969
          ToString(FLAGS_level0_file_num_compaction_trigger),
          ToString(FLAGS_level0_file_num_compaction_trigger + 2),
          ToString(FLAGS_level0_file_num_compaction_trigger + 4),
L
Lei Jin 已提交
970 971 972 973
        }
      },
      {"level0_slowdown_writes_trigger",
        {
974 975 976
          ToString(FLAGS_level0_slowdown_writes_trigger),
          ToString(FLAGS_level0_slowdown_writes_trigger + 2),
          ToString(FLAGS_level0_slowdown_writes_trigger + 4),
L
Lei Jin 已提交
977 978 979 980
        }
      },
      {"level0_stop_writes_trigger",
        {
981 982 983
          ToString(FLAGS_level0_stop_writes_trigger),
          ToString(FLAGS_level0_stop_writes_trigger + 2),
          ToString(FLAGS_level0_stop_writes_trigger + 4),
L
Lei Jin 已提交
984 985 986 987
        }
      },
      {"max_grandparent_overlap_factor",
        {
988 989 990
          ToString(Options().max_grandparent_overlap_factor - 5),
          ToString(Options().max_grandparent_overlap_factor),
          ToString(Options().max_grandparent_overlap_factor + 5),
L
Lei Jin 已提交
991 992 993 994
        }
      },
      {"expanded_compaction_factor",
        {
995 996 997
          ToString(Options().expanded_compaction_factor - 5),
          ToString(Options().expanded_compaction_factor),
          ToString(Options().expanded_compaction_factor + 5),
L
Lei Jin 已提交
998 999 1000 1001
        }
      },
      {"source_compaction_factor",
        {
1002 1003 1004
          ToString(Options().source_compaction_factor),
          ToString(Options().source_compaction_factor * 2),
          ToString(Options().source_compaction_factor * 4),
L
Lei Jin 已提交
1005 1006 1007 1008
        }
      },
      {"target_file_size_base",
        {
1009 1010 1011
          ToString(FLAGS_target_file_size_base),
          ToString(FLAGS_target_file_size_base * 2),
          ToString(FLAGS_target_file_size_base * 4),
L
Lei Jin 已提交
1012 1013 1014 1015
        }
      },
      {"target_file_size_multiplier",
        {
1016
          ToString(FLAGS_target_file_size_multiplier),
L
Lei Jin 已提交
1017 1018 1019 1020 1021 1022
          "1",
          "2",
        }
      },
      {"max_bytes_for_level_base",
        {
1023 1024 1025
          ToString(FLAGS_max_bytes_for_level_base / 2),
          ToString(FLAGS_max_bytes_for_level_base),
          ToString(FLAGS_max_bytes_for_level_base * 2),
L
Lei Jin 已提交
1026 1027 1028 1029
        }
      },
      {"max_bytes_for_level_multiplier",
        {
1030
          ToString(FLAGS_max_bytes_for_level_multiplier),
L
Lei Jin 已提交
1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043
          "1",
          "2",
        }
      },
      {"max_mem_compaction_level", {"0", "1", "2"}},
      {"max_sequential_skip_in_iterations", {"4", "8", "12"}},
    };
    for (const auto& iter : options_table_) {
      options_index_.push_back(iter.first);
    }
    return true;
  }

I
Igor Canadi 已提交
1044
  bool Run() {
1045
    PrintEnv();
L
Lei Jin 已提交
1046
    BuildOptionsTable();
1047 1048 1049 1050 1051 1052 1053 1054 1055
    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]);
    }
1056 1057 1058 1059 1060
    ThreadState bg_thread(0, &shared);
    if (FLAGS_compaction_thread_pool_adjust_interval > 0) {
      FLAGS_env->StartThread(PoolSizeChangeThread, &bg_thread);
    }

1061
    // Each thread goes through the following states:
A
amayank 已提交
1062 1063
    // initializing -> wait for others to init -> read/populate/depopulate
    // wait for others to operate -> verify -> done
1064 1065 1066 1067 1068 1069 1070

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

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

1075 1076
      shared.SetStart();
      shared.GetCondVar()->SignalAll();
A
amayank 已提交
1077
      while (!shared.AllOperated()) {
1078 1079 1080
        shared.GetCondVar()->Wait();
      }

1081
      now = FLAGS_env->NowMicros();
1082 1083 1084 1085 1086 1087 1088
      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());
      }
1089

1090 1091 1092 1093 1094 1095 1096
      shared.SetStartVerify();
      shared.GetCondVar()->SignalAll();
      while (!shared.AllDone()) {
        shared.GetCondVar()->Wait();
      }
    }

1097
    for (unsigned int i = 1; i < n; i++) {
1098 1099 1100 1101
      threads[0]->stats.Merge(threads[i]->stats);
    }
    threads[0]->stats.Report("Stress Test");

1102
    for (unsigned int i = 0; i < n; i++) {
1103
      delete threads[i];
1104
      threads[i] = nullptr;
1105
    }
1106
    double now = FLAGS_env->NowMicros();
1107 1108 1109 1110
    if (!FLAGS_test_batches_snapshots) {
      fprintf(stdout, "%s Verification successful\n",
              FLAGS_env->TimeToString((uint64_t) now/1000000).c_str());
    }
1111
    PrintStatistics();
I
Igor Canadi 已提交
1112

1113 1114 1115 1116 1117 1118 1119 1120
    if (FLAGS_compaction_thread_pool_adjust_interval > 0) {
      MutexLock l(shared.GetMutex());
      shared.SetShouldStopBgThread();
      while (!shared.BgThreadFinished()) {
        shared.GetCondVar()->Wait();
      }
    }

I
Igor Canadi 已提交
1121 1122 1123 1124 1125
    if (shared.HasVerificationFailedYet()) {
      printf("Verification failed :(\n");
      return false;
    }
    return true;
1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143
  }

 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 已提交
1144
    thread->shared->GetStressTest()->OperateDb(thread);
1145 1146 1147

    {
      MutexLock l(shared->GetMutex());
A
amayank 已提交
1148 1149
      shared->IncOperated();
      if (shared->AllOperated()) {
1150 1151 1152 1153 1154 1155 1156
        shared->GetCondVar()->SignalAll();
      }
      while (!shared->VerifyStarted()) {
        shared->GetCondVar()->Wait();
      }
    }

1157
    if (!FLAGS_test_batches_snapshots) {
M
Mayank Agarwal 已提交
1158
      thread->shared->GetStressTest()->VerifyDb(thread);
1159
    }
1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170

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

  }

1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186
  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 已提交
1187
      auto thread_pool_size_var = FLAGS_compaction_thread_pool_variations;
1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202
      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);
    }
  }

1203 1204 1205
  // 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 已提交
1206 1207 1208
  Status MultiPut(ThreadState* thread, const WriteOptions& writeoptions,
                  ColumnFamilyHandle* column_family, const Slice& key,
                  const Slice& value, size_t sz) {
1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219
    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];
1220
      if (FLAGS_use_merge) {
I
Igor Canadi 已提交
1221
        batch.Merge(column_family, keys[i], value_slices[i]);
D
Deon Nicholas 已提交
1222
      } else {
I
Igor Canadi 已提交
1223
        batch.Put(column_family, keys[i], value_slices[i]);
D
Deon Nicholas 已提交
1224
      }
1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240
    }

    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 已提交
1241 1242
  Status MultiDelete(ThreadState* thread, const WriteOptions& writeoptions,
                     ColumnFamilyHandle* column_family, const Slice& key) {
1243 1244 1245 1246 1247 1248 1249
    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 已提交
1250
      batch.Delete(column_family, keys[i]);
1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267
    }

    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 已提交
1268 1269 1270
  Status MultiGet(ThreadState* thread, const ReadOptions& readoptions,
                  ColumnFamilyHandle* column_family, const Slice& key,
                  std::string* value) {
1271 1272 1273 1274 1275 1276 1277 1278 1279
    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 已提交
1280
      s = db_->Get(readoptionscopy, column_family, key_slices[i], value);
1281 1282 1283 1284 1285 1286 1287 1288
      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] = "";
1289
        thread->stats.AddGets(1, 0);
1290 1291 1292 1293 1294 1295
      } else {
        values[i] = *value;

        char expected_prefix = (keys[i])[0];
        char actual_prefix = (values[i])[0];
        if (actual_prefix != expected_prefix) {
1296
          fprintf(stderr, "error expected prefix = %c actual = %c\n",
1297 1298 1299
                  expected_prefix, actual_prefix);
        }
        (values[i])[0] = ' '; // blank out the differing character
1300
        thread->stats.AddGets(1, 1);
1301 1302 1303 1304 1305 1306 1307
      }
    }
    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]) {
1308
        fprintf(stderr, "error : inconsistent values for key %s: %s, %s\n",
1309 1310
                key.ToString(true).c_str(), StringToHex(values[0]).c_str(),
                StringToHex(values[i]).c_str());
1311 1312 1313 1314 1315 1316 1317 1318
      // we continue after error rather than exiting so that we can
      // find more errors if any
      }
    }

    return s;
  }

L
Lei Jin 已提交
1319 1320 1321 1322 1323
  // 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.
1324
  // ASSUMES that MultiPut was used to put (K, V)
I
Igor Canadi 已提交
1325 1326
  Status MultiPrefixScan(ThreadState* thread, const ReadOptions& readoptions,
                         ColumnFamilyHandle* column_family,
L
Lei Jin 已提交
1327
                         const Slice& key) {
1328 1329 1330 1331 1332 1333 1334 1335
    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 已提交
1336 1337 1338
      prefixes[i] += key.ToString();
      prefixes[i].resize(FLAGS_prefix_size);
      prefix_slices[i] = Slice(prefixes[i]);
1339 1340
      readoptionscopy[i] = readoptions;
      readoptionscopy[i].snapshot = snapshot;
I
Igor Canadi 已提交
1341
      iters[i] = db_->NewIterator(readoptionscopy[i], column_family);
I
Igor Canadi 已提交
1342
      iters[i]->Seek(prefix_slices[i]);
1343 1344 1345
    }

    int count = 0;
1346
    while (iters[0]->Valid() && iters[0]->key().starts_with(prefix_slices[0])) {
1347 1348 1349 1350 1351
      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
1352 1353
        assert(iters[i]->Valid() &&
               iters[i]->key().starts_with(prefix_slices[i]));
1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367
        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]) {
1368 1369 1370
          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());
1371 1372 1373 1374 1375 1376 1377 1378 1379 1380
          // 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
1381 1382
      assert(!iters[i]->Valid() ||
             !iters[i]->key().starts_with(prefix_slices[i]));
1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396
      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;
  }

1397 1398
  // 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 已提交
1399 1400
  Status MultiIterate(ThreadState* thread, const ReadOptions& readoptions,
                      ColumnFamilyHandle* column_family, const Slice& key) {
1401 1402 1403 1404
    Status s;
    const Snapshot* snapshot = db_->GetSnapshot();
    ReadOptions readoptionscopy = readoptions;
    readoptionscopy.snapshot = snapshot;
I
Igor Canadi 已提交
1405
    unique_ptr<Iterator> iter(db_->NewIterator(readoptionscopy, column_family));
1406 1407

    iter->Seek(key);
1408
    for (uint64_t i = 0; i < FLAGS_num_iterations && iter->Valid(); i++) {
1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426
      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;
  }

1427
  Status SetOptions(ThreadState* thread) {
L
Lei Jin 已提交
1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453
    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 已提交
1454
  void OperateDb(ThreadState* thread) {
1455 1456
    ReadOptions read_opts(FLAGS_verify_checksum, true);
    WriteOptions write_opts;
A
amayank 已提交
1457
    char value[100];
1458 1459 1460 1461 1462 1463
    long max_key = thread->shared->GetMaxKey();
    std::string from_db;
    if (FLAGS_sync) {
      write_opts.sync = true;
    }
    write_opts.disableWAL = FLAGS_disable_wal;
1464 1465 1466
    const int prefixBound = (int)FLAGS_readpercent + (int)FLAGS_prefixpercent;
    const int writeBound = prefixBound + (int)FLAGS_writepercent;
    const int delBound = writeBound + (int)FLAGS_delpercent;
1467 1468

    thread->stats.Start();
1469
    for (uint64_t i = 0; i < FLAGS_ops_per_thread; i++) {
I
Igor Canadi 已提交
1470 1471 1472 1473
      if (thread->shared->HasVerificationFailedYet()) {
        break;
      }
      if (i != 0 && (i % (FLAGS_ops_per_thread / (FLAGS_reopen + 1))) == 0) {
1474
        {
1475
          thread->stats.FinishedSingleOp();
1476 1477 1478 1479 1480 1481 1482 1483 1484
          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();
          }
1485 1486
          // Commenting this out as we don't want to reset stats on each open.
          // thread->stats.Start();
1487 1488
        }
      }
1489

L
Lei Jin 已提交
1490 1491 1492 1493 1494 1495
      // Change Options
      if (FLAGS_set_options_one_in > 0 &&
          thread->rand.OneIn(FLAGS_set_options_one_in)) {
        SetOptions(thread);
      }

1496 1497 1498 1499 1500
      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 已提交
1501
      if (!FLAGS_test_batches_snapshots &&
1502
          FLAGS_clear_column_family_one_in != 0 && FLAGS_column_families > 1) {
I
Igor Canadi 已提交
1503 1504 1505 1506
        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 =
1507
              ToString(new_column_family_name_.fetch_add(1));
I
Igor Canadi 已提交
1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518
          {
            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];
1519 1520 1521 1522 1523
          if (!s.ok()) {
            fprintf(stderr, "dropping column family error: %s\n",
                s.ToString().c_str());
            std::terminate();
          }
I
Igor Canadi 已提交
1524 1525 1526 1527
          s = db_->CreateColumnFamily(ColumnFamilyOptions(options_), new_name,
                                      &column_families_[cf]);
          column_family_names_[cf] = new_name;
          thread->shared->ClearColumnFamily(cf);
1528 1529 1530 1531 1532
          if (!s.ok()) {
            fprintf(stderr, "creating column family error: %s\n",
                s.ToString().c_str());
            std::terminate();
          }
I
Igor Canadi 已提交
1533 1534 1535 1536
          thread->shared->UnlockColumnFamily(cf);
        }
      }

1537
      long rand_key = thread->rand.Next() % max_key;
I
Igor Canadi 已提交
1538
      int rand_column_family = thread->rand.Next() % FLAGS_column_families;
1539 1540 1541
      std::string keystr = Key(rand_key);
      Slice key = keystr;
      int prob_op = thread->rand.Uniform(100);
I
Igor Canadi 已提交
1542 1543 1544 1545 1546 1547
      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];
1548 1549

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

1643 1644 1645
    thread->stats.Stop();
  }

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

I
Igor Canadi 已提交
1722 1723 1724 1725 1726
  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();
1727 1728
  }

I
Igor Canadi 已提交
1729 1730
  bool VerifyValue(int cf, long key, const ReadOptions& opts,
                   SharedState* shared, const std::string& value_from_db,
I
Igor Canadi 已提交
1731
                   Status s, bool strict = false) const {
I
Igor Canadi 已提交
1732 1733 1734
    if (shared->HasVerificationFailedYet()) {
      return false;
    }
1735
    // compare value_from_db with the value in the shared state
A
amayank 已提交
1736
    char value[100];
I
Igor Canadi 已提交
1737
    uint32_t value_base = shared->Get(cf, key);
1738
    if (value_base == SharedState::SENTINEL && !strict) {
I
Igor Canadi 已提交
1739
      return true;
1740 1741
    }

1742
    if (s.ok()) {
1743
      if (value_base == SharedState::SENTINEL) {
I
Igor Canadi 已提交
1744 1745
        VerificationAbort(shared, "Unexpected value found", cf, key);
        return false;
1746
      }
A
amayank 已提交
1747
      size_t sz = GenerateValue(value_base, value, sizeof(value));
1748
      if (value_from_db.length() != sz) {
I
Igor Canadi 已提交
1749 1750
        VerificationAbort(shared, "Length of value read is not equal", cf, key);
        return false;
1751
      }
1752
      if (memcmp(value_from_db.data(), value, sz) != 0) {
I
Igor Canadi 已提交
1753 1754 1755
        VerificationAbort(shared, "Contents of value read don't match", cf,
                          key);
        return false;
1756 1757 1758
      }
    } else {
      if (value_base != SharedState::SENTINEL) {
I
Igor Canadi 已提交
1759 1760
        VerificationAbort(shared, "Value not found: " + s.ToString(), cf, key);
        return false;
1761 1762
      }
    }
I
Igor Canadi 已提交
1763
    return true;
1764 1765
  }

I
Igor Canadi 已提交
1766 1767 1768 1769 1770 1771 1772
  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++) {
1773 1774 1775 1776 1777 1778 1779 1780 1781 1782 1783 1784
      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);
    }
1785
    v[value_sz] = '\0';
1786 1787 1788 1789
    return value_sz; // the size of the value set.
  }

  void PrintEnv() const {
I
Igor Canadi 已提交
1790 1791 1792 1793 1794 1795 1796
    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);
    }
1797
    fprintf(stdout, "Number of threads   : %d\n", FLAGS_threads);
K
kailiu 已提交
1798 1799 1800
    fprintf(stdout,
            "Ops per thread      : %lu\n",
            (unsigned long)FLAGS_ops_per_thread);
1801 1802 1803 1804 1805
    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 已提交
1806 1807 1808 1809 1810
    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);
1811 1812
    fprintf(stdout, "DB-write-buffer-size: %" PRIu64 "\n",
        FLAGS_db_write_buffer_size);
1813
    fprintf(stdout, "Write-buffer-size   : %d\n", FLAGS_write_buffer_size);
K
kailiu 已提交
1814 1815 1816 1817 1818 1819
    fprintf(stdout,
            "Iterations          : %lu\n",
            (unsigned long)FLAGS_num_iterations);
    fprintf(stdout,
            "Max key             : %lu\n",
            (unsigned long)FLAGS_max_key);
X
Xing Jin 已提交
1820 1821
    fprintf(stdout, "Ratio #ops/#keys    : %f\n",
            (1.0 * FLAGS_ops_per_thread * FLAGS_threads)/FLAGS_max_key);
1822
    fprintf(stdout, "Num times DB reopens: %d\n", FLAGS_reopen);
1823 1824
    fprintf(stdout, "Batches/snapshots   : %d\n",
            FLAGS_test_batches_snapshots);
X
Xing Jin 已提交
1825
    fprintf(stdout, "Purge redundant %%   : %d\n",
1826
            FLAGS_purge_redundant_percent);
1827
    fprintf(stdout, "Deletes use filter  : %d\n",
1828
            FLAGS_filter_deletes);
1829 1830
    fprintf(stdout, "Do update in place  : %d\n",
            FLAGS_in_place_update);
1831
    fprintf(stdout, "Num keys per lock   : %d\n",
1832 1833
            1 << FLAGS_log2_keys_per_lock);

1834
    const char* compression = "";
1835
    switch (FLAGS_compression_type_e) {
1836
      case rocksdb::kNoCompression:
1837
        compression = "none";
1838
        break;
1839
      case rocksdb::kSnappyCompression:
1840
        compression = "snappy";
1841
        break;
1842
      case rocksdb::kZlibCompression:
1843
        compression = "zlib";
1844
        break;
1845
      case rocksdb::kBZip2Compression:
1846
        compression = "bzip2";
1847
        break;
A
Albert Strasheim 已提交
1848 1849
      case rocksdb::kLZ4Compression:
        compression = "lz4";
1850
        break;
A
Albert Strasheim 已提交
1851 1852 1853 1854
      case rocksdb::kLZ4HCCompression:
        compression = "lz4hc";
        break;
      }
1855 1856

    fprintf(stdout, "Compression         : %s\n", compression);
J
Jim Paton 已提交
1857 1858 1859 1860 1861 1862

    const char* memtablerep = "";
    switch (FLAGS_rep_factory) {
      case kSkipList:
        memtablerep = "skip_list";
        break;
I
Igor Canadi 已提交
1863
      case kHashSkipList:
J
Jim Paton 已提交
1864 1865 1866 1867 1868 1869 1870 1871 1872
        memtablerep = "prefix_hash";
        break;
      case kVectorRep:
        memtablerep = "vector";
        break;
    }

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

1873 1874 1875 1876
    fprintf(stdout, "------------------------------------------------\n");
  }

  void Open() {
1877
    assert(db_ == nullptr);
1878 1879 1880 1881
    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;
1882
    block_based_options.format_version = 2;
1883 1884 1885
    block_based_options.filter_policy = filter_policy_;
    options_.table_factory.reset(
        NewBlockBasedTableFactory(block_based_options));
1886
    options_.db_write_buffer_size = FLAGS_db_write_buffer_size;
I
Igor Canadi 已提交
1887 1888 1889 1890
    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;
1891 1892
    options_.max_write_buffer_number_to_maintain =
        FLAGS_max_write_buffer_number_to_maintain;
I
Igor Canadi 已提交
1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903
    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;
1904
    rocksdb_kill_odds = FLAGS_kill_random_test;
I
Igor Canadi 已提交
1905 1906 1907 1908
    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 =
1909
        FLAGS_max_bytes_for_level_multiplier;
I
Igor Canadi 已提交
1910 1911 1912 1913 1914 1915 1916 1917 1918
    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;
1919
    options_.inplace_update_support = FLAGS_in_place_update;
I
Igor Canadi 已提交
1920
    if ((FLAGS_prefix_size == 0) == (FLAGS_rep_factory == kHashSkipList)) {
J
Jim Paton 已提交
1921 1922 1923 1924 1925 1926 1927 1928
      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 已提交
1929 1930 1931 1932
#ifndef ROCKSDB_LITE
      case kHashSkipList:
        options_.memtable_factory.reset(NewHashSkipListRepFactory(10000));
        break;
J
Jim Paton 已提交
1933
      case kVectorRep:
I
Igor Canadi 已提交
1934
        options_.memtable_factory.reset(new VectorRepFactory());
J
Jim Paton 已提交
1935
        break;
S
sdong 已提交
1936 1937 1938 1939 1940 1941
#else
      default:
        fprintf(stderr,
                "RocksdbLite only supports skip list mem table. Skip "
                "--rep_factory\n");
#endif  // ROCKSDB_LITE
J
Jim Paton 已提交
1942
    }
S
sdong 已提交
1943

1944
    static Random purge_percent(1000); // no benefit from non-determinism here
1945 1946
    if (static_cast<int32_t>(purge_percent.Uniform(100)) <
        FLAGS_purge_redundant_percent - 1) {
I
Igor Canadi 已提交
1947
      options_.purge_redundant_kvs_while_flush = false;
1948
    }
1949

1950
    if (FLAGS_use_merge) {
I
Igor Canadi 已提交
1951
      options_.merge_operator = MergeOperators::CreatePutOperator();
D
Deon Nicholas 已提交
1952 1953
    }

1954 1955
    // set universal style compaction configurations, if applicable
    if (FLAGS_universal_size_ratio != 0) {
I
Igor Canadi 已提交
1956 1957
      options_.compaction_options_universal.size_ratio =
          FLAGS_universal_size_ratio;
1958 1959
    }
    if (FLAGS_universal_min_merge_width != 0) {
I
Igor Canadi 已提交
1960 1961
      options_.compaction_options_universal.min_merge_width =
          FLAGS_universal_min_merge_width;
1962 1963
    }
    if (FLAGS_universal_max_merge_width != 0) {
I
Igor Canadi 已提交
1964 1965
      options_.compaction_options_universal.max_merge_width =
          FLAGS_universal_max_merge_width;
1966 1967
    }
    if (FLAGS_universal_max_size_amplification_percent != 0) {
I
Igor Canadi 已提交
1968 1969
      options_.compaction_options_universal.max_size_amplification_percent =
          FLAGS_universal_max_size_amplification_percent;
1970 1971
    }

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

1974 1975
    Status s;
    if (FLAGS_ttl == -1) {
I
Igor Canadi 已提交
1976 1977 1978 1979 1980 1981 1982
      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());
1983
        column_family_names_.push_back(kDefaultColumnFamilyName);
I
Igor Canadi 已提交
1984 1985 1986 1987 1988 1989 1990 1991 1992 1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010
      } 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_) {
2011
        if (name != kDefaultColumnFamilyName) {
I
Igor Canadi 已提交
2012 2013 2014 2015 2016
          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 已提交
2017
      while (cf_descriptors.size() < (size_t)FLAGS_column_families) {
2018
        std::string name = ToString(new_column_family_name_.load());
I
Igor Canadi 已提交
2019 2020 2021 2022
        new_column_family_name_++;
        cf_descriptors.emplace_back(name, ColumnFamilyOptions(options_));
        column_family_names_.push_back(name);
      }
2023 2024 2025
      options_.listeners.clear();
      options_.listeners.emplace_back(
          new DbStressListener(FLAGS_db, options_.db_paths, cf_descriptors));
I
Igor Canadi 已提交
2026
      options_.create_missing_column_families = true;
I
Igor Canadi 已提交
2027 2028 2029 2030
      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));
2031
    } else {
S
sdong 已提交
2032
#ifndef ROCKSDB_LITE
I
Igor Canadi 已提交
2033 2034 2035
      DBWithTTL* db_with_ttl;
      s = DBWithTTL::Open(options_, FLAGS_db, &db_with_ttl, FLAGS_ttl);
      db_ = db_with_ttl;
S
sdong 已提交
2036 2037 2038 2039
#else
      fprintf(stderr, "TTL is not supported in RocksDBLite\n");
      exit(1);
#endif
2040
    }
2041 2042 2043 2044 2045 2046
    if (!s.ok()) {
      fprintf(stderr, "open error: %s\n", s.ToString().c_str());
      exit(1);
    }
  }

2047
  void Reopen() {
I
Igor Canadi 已提交
2048 2049
    for (auto cf : column_families_) {
      delete cf;
2050
    }
I
Igor Canadi 已提交
2051 2052
    column_families_.clear();
    delete db_;
2053
    db_ = nullptr;
2054 2055 2056 2057 2058 2059

    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_);
2060 2061 2062
    Open();
  }

2063 2064
  void PrintStatistics() {
    if (dbstats) {
2065
      fprintf(stdout, "STATISTICS:\n%s\n", dbstats->ToString().c_str());
2066 2067 2068 2069
    }
  }

 private:
2070 2071 2072
  std::shared_ptr<Cache> cache_;
  std::shared_ptr<Cache> compressed_cache_;
  std::shared_ptr<const FilterPolicy> filter_policy_;
2073
  DB* db_;
I
Igor Canadi 已提交
2074 2075 2076 2077
  Options options_;
  std::vector<ColumnFamilyHandle*> column_families_;
  std::vector<std::string> column_family_names_;
  std::atomic<int> new_column_family_name_;
2078
  int num_times_reopened_;
L
Lei Jin 已提交
2079 2080
  std::unordered_map<std::string, std::vector<std::string>> options_table_;
  std::vector<std::string> options_index_;
2081 2082
};

2083
}  // namespace rocksdb
2084 2085

int main(int argc, char** argv) {
2086 2087 2088
  SetUsageMessage(std::string("\nUSAGE:\n") + std::string(argv[0]) +
                  " [OPTIONS]...");
  ParseCommandLineFlags(&argc, &argv, true);
2089 2090 2091 2092 2093 2094 2095 2096

  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);
2097
  }
2098
  FLAGS_rep_factory = StringToRepFactory(FLAGS_memtablerep.c_str());
2099

2100 2101 2102 2103
  // 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 已提交
2104 2105 2106 2107 2108 2109 2110 2111 2112 2113 2114 2115
  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);
  }
2116
  if ((FLAGS_readpercent + FLAGS_prefixpercent +
2117 2118 2119
       FLAGS_writepercent + FLAGS_delpercent + FLAGS_iterpercent) != 100) {
      fprintf(stderr,
              "Error: Read+Prefix+Write+Delete+Iterate percents != 100!\n");
A
amayank 已提交
2120 2121
      exit(1);
  }
2122 2123 2124 2125
  if (FLAGS_disable_wal == 1 && FLAGS_reopen > 0) {
      fprintf(stderr, "Error: Db cannot reopen safely with disable_wal set!\n");
      exit(1);
  }
2126
  if ((unsigned)FLAGS_reopen >= FLAGS_ops_per_thread) {
K
kailiu 已提交
2127 2128 2129 2130 2131
      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);
2132 2133 2134
      exit(1);
  }

2135
  // Choose a location for the test database if none given with --db=<path>
2136 2137
  if (FLAGS_db.empty()) {
      std::string default_db_path;
2138
      rocksdb::Env::Default()->GetTestDirectory(&default_db_path);
2139
      default_db_path += "/dbstress";
2140
      FLAGS_db = default_db_path;
2141 2142
  }

2143
  rocksdb::StressTest stress;
I
Igor Canadi 已提交
2144 2145 2146 2147 2148
  if (stress.Run()) {
    return 0;
  } else {
    return 1;
  }
2149
}
2150 2151

#endif  // GFLAGS