YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

//  Copyright (c) 2011-present, 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.

//

// The following only applies to changes made to this file as part of YugaByte development.

//

// Portions Copyright (c) YugaByte, Inc.

//

// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except

// in compliance with the License.  You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software distributed under the License

// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express

// or implied.  See the License for the specific language governing permissions and limitations

// under the License.

//

// 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.

//

// 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.

#ifndef GFLAGS

#include <cstdio>

int main() {

  fprintf(stderr, "Please install gflags to run rocksdb tools\n");

  return 1;

}

#else

#ifndef __STDC_FORMAT_MACROS

#define __STDC_FORMAT_MACROS

#endif

#include <inttypes.h>

#include <stdio.h>

#include <stdlib.h>

#include <sys/types.h>

#include <chrono>

#include <exception>

#include <thread>

#include <gflags/gflags.h>

#include "yb/rocksdb/db/db_impl.h"

#include "yb/rocksdb/db/filename.h"

#include "yb/rocksdb/db/version_set.h"

#include "yb/rocksdb/hdfs/env_hdfs.h"

#include "yb/rocksdb/port/port.h"

#include "yb/rocksdb/cache.h"

#include "yb/rocksdb/env.h"

#include "yb/rocksdb/filter_policy.h"

#include "yb/rocksdb/slice_transform.h"

#include "yb/rocksdb/statistics.h"

#include "yb/rocksdb/utilities/db_ttl.h"

#include "yb/rocksdb/write_batch.h"

#include "yb/rocksdb/util/coding.h"

#include "yb/rocksdb/util/compression.h"

#include "yb/rocksdb/util/crc32c.h"

#include "yb/rocksdb/util/histogram.h"

#include "yb/rocksdb/util/logging.h"

#include "yb/rocksdb/util/mutexlock.h"

#include "yb/rocksdb/util/random.h"

#include "yb/rocksdb/util/testutil.h"

#include "yb/rocksdb/utilities/merge_operators.h"

#include "yb/util/slice.h"

#include "yb/util/string_util.h"

using GFLAGS::ParseCommandLineFlags;

using GFLAGS::RegisterFlagValidator;

using GFLAGS::SetUsageMessage;

static const int64_t KB = 1024;

static bool ValidateUint32Range(const char* flagname, uint64_t value) {

  if (value > std::numeric_limits<uint32_t>::max()) {

    fprintf(stderr,

            "Invalid value for --%s: %" PRIu64 ", overflow\n",

            flagname,

            value);

    return false;

  }

  return true;

}

DEFINE_uint64(seed, 2341234, "Seed for PRNG");

static const bool FLAGS_seed_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_seed, &ValidateUint32Range);

DEFINE_int64(max_key, 1 * KB* KB,

             "Max number of key/values to place in database");

DEFINE_int32(column_families, 10, "Number of column families");

// TODO(noetzli) Add support for single deletes

DEFINE_bool(test_batches_snapshots, false,

            "If set, the test uses MultiGet(), MultiPut() and MultiDelete()"

            " 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");

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

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");

DEFINE_int32(value_size_mult, 8,

             "Size of value will be this number times rand_int(1,3) bytes");

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

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

DEFINE_bool(destroy_db_initially, true,

            "Destroys the database dir before start if this is true");

DEFINE_bool(verbose, false, "Verbose");

DEFINE_bool(progress_reports, true,

            "If true, db_stress will report number of finished operations");

DEFINE_uint64(db_write_buffer_size, rocksdb::Options().db_write_buffer_size,

              "Number of bytes to buffer in all memtables before compacting");

DEFINE_int32(write_buffer_size,

             static_cast<int32_t>(rocksdb::Options().write_buffer_size),

             "Number of bytes to buffer in memtable before compacting");

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.");

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.");

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.");

DEFINE_int32(open_files, rocksdb::Options().max_open_files,

             "Maximum number of files to keep open at the same time "

             "(use default if == 0)");

DEFINE_int64(compressed_cache_size, -1,

             "Number of bytes to use as a cache of compressed data."

             " Negative means use default settings.");

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

DEFINE_int32(level0_file_num_compaction_trigger,

             rocksdb::Options().level0_file_num_compaction_trigger,

             "Level0 compaction start trigger");

DEFINE_int32(level0_slowdown_writes_trigger,

             rocksdb::Options().level0_slowdown_writes_trigger,

             "Number of files in level-0 that will slow down writes");

DEFINE_int32(level0_stop_writes_trigger,

             rocksdb::Options().level0_stop_writes_trigger,

             "Number of files in level-0 that will trigger put stop.");

DEFINE_int32(block_size,

             static_cast<int32_t>(rocksdb::BlockBasedTableOptions().block_size),

             "Number of bytes in a block.");

DEFINE_int32(max_background_compactions,

             rocksdb::Options().max_background_compactions,

             "The maximum number of concurrent background compactions "

             "that can occur in parallel.");

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.");

DEFINE_int32(compaction_thread_pool_variations, 2,

             "Range of background thread pool size variations when adjusted "

             "periodically.");

DEFINE_int32(max_background_flushes, rocksdb::Options().max_background_flushes,

             "The maximum number of concurrent background flushes "

             "that can occur in parallel.");

DEFINE_int32(universal_size_ratio, 0, "The ratio of file sizes that trigger"

             " compaction in universal style");

DEFINE_int32(universal_min_merge_width, 0, "The minimum number of files to "

             "compact in universal style compaction");

DEFINE_int32(universal_max_merge_width, 0, "The max number of files to compact"

             " in universal style compaction");

DEFINE_int32(universal_max_size_amplification_percent, 0,

             "The max size amplification for universal style compaction");

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");

DEFINE_int32(set_options_one_in, 0,

             "With a chance of 1/N, change some random options");

DEFINE_int32(set_in_place_one_in, 0,

             "With a chance of 1/N, toggle in place support option");

DEFINE_int64(cache_size, 2LL * KB * KB * KB,

             "Number of bytes to use as a cache of uncompressed data.");

DEFINE_uint64(subcompactions, 1,

             "Maximum number of subcompactions to divide L0-L1 compactions "

             "into.");

DEFINE_bool(allow_concurrent_memtable_write, true,

            "Allow multi-writers to update mem tables in parallel.");

DEFINE_bool(enable_write_thread_adaptive_yield, true,

            "Use a yielding spin loop for brief writer thread waits.");

static const bool FLAGS_subcompactions_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_subcompactions, &ValidateUint32Range);

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");

static const bool FLAGS_reopen_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_reopen, &ValidateInt32Positive);

DEFINE_int32(bloom_bits, 10, "Bloom filter bits per key. "

             "Negative means use default settings.");

DEFINE_bool(use_block_based_filter, false, "use block based filter"

              "instead of full filter for block based table");

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

DEFINE_bool(verify_checksum, false,

            "Verify checksum for every block read from storage");

DEFINE_bool(mmap_read, rocksdb::EnvOptions().use_mmap_reads,

            "Allow reads to occur via mmap-ing files");

// Database statistics

static std::shared_ptr<rocksdb::Statistics> dbstats;

DEFINE_bool(statistics, false, "Create database statistics");

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

DEFINE_bool(disable_data_sync, false,

            "If true, do not wait until data is synced to disk.");

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

DEFINE_int32(kill_random_test, 0,

             "If non-zero, kill at various points in source code with "

             "probability 1/this");

static const bool FLAGS_kill_random_test_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_kill_random_test, &ValidateInt32Positive);

extern int rocksdb_kill_odds;

DEFINE_string(kill_prefix_blacklist, "",

              "If non-empty, kill points with prefix in the list given will be"

              " skipped. Items are comma-separated.");

extern std::vector<std::string> rocksdb_kill_prefix_blacklist;

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

DEFINE_int32(target_file_size_base, 64 * KB,

             "Target level-1 file size for compaction");

DEFINE_int32(target_file_size_multiplier, 1,

             "A multiplier to compute target level-N file size (N >= 2)");

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

DEFINE_int32(max_bytes_for_level_multiplier, 2,

             "A multiplier to compute max bytes for level-N (N >= 2)");

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)");

static const bool FLAGS_readpercent_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_readpercent, &ValidateInt32Percent);

DEFINE_int32(prefixpercent, 20,

             "Ratio of prefix iterators to total workload (expressed as a"

             " percentage)");

static const bool FLAGS_prefixpercent_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_prefixpercent, &ValidateInt32Percent);

DEFINE_int32(writepercent, 45,

             "Ratio of writes to total workload (expressed as a percentage)");

static const bool FLAGS_writepercent_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_writepercent, &ValidateInt32Percent);

DEFINE_int32(delpercent, 15,

             "Ratio of deletes to total workload (expressed as a percentage)");

static const bool FLAGS_delpercent_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_delpercent, &ValidateInt32Percent);

DEFINE_int32(nooverwritepercent, 60,

             "Ratio of keys without overwrite to total workload (expressed as "

             " a percentage)");

static const bool FLAGS_nooverwritepercent_dummy __attribute__((__unused__)) =

    RegisterFlagValidator(&FLAGS_nooverwritepercent, &ValidateInt32Percent);

DEFINE_int32(iterpercent, 10, "Ratio of iterations to total workload"

             " (expressed as a percentage)");

static const bool FLAGS_iterpercent_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_iterpercent, &ValidateInt32Percent);

DEFINE_uint64(num_iterations, 10, "Number of iterations per MultiIterate run");

static const bool FLAGS_num_iterations_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_num_iterations, &ValidateUint32Range);

namespace {

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;

  else if (!strcasecmp(ctype, "lz4"))

    return rocksdb::kLZ4Compression;

  else if (!strcasecmp(ctype, "lz4hc"))

    return rocksdb::kLZ4HCCompression;

  else if (!strcasecmp(ctype, "zstd"))

    return rocksdb::kZSTDNotFinalCompression;

  fprintf(stdout, "Cannot parse compression type '%s'\n", ctype);

  return rocksdb::kSnappyCompression; // default value

}

std::vector<std::string> SplitString(std::string src) {

  std::vector<std::string> ret;

  if (src.empty()) {

    return ret;

  }

  size_t pos = 0;

  size_t pos_comma;

  while ((pos_comma = src.find(',', pos)) != std::string::npos) {

    ret.push_back(src.substr(pos, pos_comma - pos));

    pos = pos_comma + 1;

  }

  ret.push_back(src.substr(pos, src.length()));

  return ret;

}

}  // namespace

DEFINE_string(compression_type, "snappy",

              "Algorithm to use to compress the database");

static enum rocksdb::CompressionType FLAGS_compression_type_e =

    rocksdb::kSnappyCompression;

DEFINE_string(hdfs, "", "Name of hdfs environment");

// posix or hdfs environment

static rocksdb::Env* FLAGS_env = rocksdb::Env::Default();

DEFINE_uint64(ops_per_thread, 1200000, "Number of operations per thread.");

static const bool FLAGS_ops_per_thread_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_ops_per_thread, &ValidateUint32Range);

DEFINE_uint64(log2_keys_per_lock, 2, "Log2 of number of keys per lock");

static const bool FLAGS_log2_keys_per_lock_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_log2_keys_per_lock, &ValidateUint32Range);

DEFINE_bool(filter_deletes, false, "On true, deletes use KeyMayExist to drop"

            " the delete if key not present");

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

enum RepFactory {

  kSkipList,

  kHashSkipList,

  kVectorRep

};

namespace {

enum RepFactory StringToRepFactory(const char* ctype) {

  assert(ctype);

  if (!strcasecmp(ctype, "skip_list"))

    return kSkipList;

  else if (!strcasecmp(ctype, "prefix_hash"))

    return kHashSkipList;

  else if (!strcasecmp(ctype, "vector"))

    return kVectorRep;

  fprintf(stdout, "Cannot parse memreptable %s\n", ctype);

  return kSkipList;

}

}  // namespace

static enum RepFactory FLAGS_rep_factory;

DEFINE_string(memtablerep, "prefix_hash", "");

static bool ValidatePrefixSize(const char* flagname, int32_t value) {

  if (value < 0 || value > 8) {

    fprintf(stderr, "Invalid value for --%s: %d. 0 <= PrefixSize <= 8\n",

            flagname, value);

    return false;

  }

  return true;

}

DEFINE_int32(prefix_size, 7, "Control the prefix size for HashSkipListRep");

static const bool FLAGS_prefix_size_dummy __attribute__((unused)) =

    RegisterFlagValidator(&FLAGS_prefix_size, &ValidatePrefixSize);

DEFINE_bool(use_merge, false, "On true, replaces all writes with a Merge "

            "that behaves like a Put");

namespace rocksdb {

// convert long to a big-endian slice key

static std::string Key(int64_t 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 (size_t i = 0 ; i < sizeof(val); ++i) {

    big_endian_key[i] = little_endian_key[sizeof(val) - 1 - i];

  }

  return big_endian_key;

}

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, sizeof(buf), "%02X", (unsigned char)str[i]);

    result += buf;

  }

  return result;

}

class StressTest;

namespace {

class Stats {

 private:

  uint64_t start_;

  uint64_t finish_;

  double  seconds_;

  int64_t done_;

  int64_t gets_;

  int64_t prefixes_;

  int64_t writes_;

  int64_t deletes_;

  size_t single_deletes_;

  int64_t iterator_size_sums_;

  int64_t founds_;

  int64_t iterations_;

  int64_t errors_;

  int next_report_;

  size_t bytes_;

  uint64_t last_op_finish_;

  HistogramImpl hist_;

 public:

  Stats() { }

  void Start() {

    next_report_ = 100;

    hist_.Clear();

    done_ = 0;

    gets_ = 0;

    prefixes_ = 0;

    writes_ = 0;

    deletes_ = 0;

    single_deletes_ = 0;

    iterator_size_sums_ = 0;

    founds_ = 0;

    iterations_ = 0;

    errors_ = 0;

    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_;

    gets_ += other.gets_;

    prefixes_ += other.prefixes_;

    writes_ += other.writes_;

    deletes_ += other.deletes_;

    single_deletes_ += other.single_deletes_;

    iterator_size_sums_ += other.iterator_size_sums_;

    founds_ += other.founds_;

    iterations_ += other.iterations_;

    errors_ += other.errors_;

    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) {

      auto now = FLAGS_env->NowMicros();

      auto micros = now - last_op_finish_;

      hist_.Add(micros);

      if (micros > 20000) {

        fprintf(stdout, "long op: %" PRIu64 " micros%30s\r", micros, "");

      }

      last_op_finish_ = now;

    }

      done_++;

    if (FLAGS_progress_reports) {

      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 %" PRId64 " ops%30s\r", done_, "");

      }

    }

  }

  void AddBytesForWrites(int nwrites, size_t nbytes) {

    writes_ += nwrites;

    bytes_ += nbytes;

  }

  void AddGets(int ngets, int nfounds) {

    founds_ += nfounds;

    gets_ += ngets;

  }

  void AddPrefixes(int nprefixes, int count) {

    prefixes_ += nprefixes;

    iterator_size_sums_ += count;

  }

  void AddIterations(int n) {

    iterations_ += n;

  }

  void AddDeletes(int n) {

    deletes_ += n;

  }

  void AddSingleDeletes(size_t n) { single_deletes_ += n; }

  void AddErrors(int n) {

    errors_ += n;

  }

  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;

    int64_t throughput = static_cast<int64_t>(done_ / elapsed);

    fprintf(stdout, "%-12s: ", name);

    fprintf(stdout, "%.3f micros/op %" PRId64 " ops/sec\n",

            seconds_ * 1e6 / done_, throughput);

    fprintf(stdout, "%-12s: Wrote %.2f MB (%.2f MB/sec) (%" PRId64 "%% of %" PRId64 " ops)\n",

            "", bytes_mb, rate, (100 * writes_) / done_, done_);

    fprintf(stdout, "%-12s: Wrote %" PRId64 " times\n", "", writes_);

    fprintf(stdout, "%-12s: Deleted %" PRId64 " times\n", "", deletes_);

    fprintf(stdout, "%-12s: Single deleted %" ROCKSDB_PRIszt " times\n", "",

           single_deletes_);

    fprintf(stdout, "%-12s: %" PRId64 " read and %" PRId64 " found the key\n", "",

            gets_, founds_);

    fprintf(stdout, "%-12s: Prefix scanned %" PRId64 " times\n", "", prefixes_);

    fprintf(stdout, "%-12s: Iterator size sum is %" PRId64 "\n", "", iterator_size_sums_);

    fprintf(stdout, "%-12s: Iterated %" PRId64 " times\n", "", iterations_);

    fprintf(stdout, "%-12s: Got errors %" PRId64 " times\n", "", errors_);

    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:

  static const uint32_t SENTINEL;

  explicit SharedState(StressTest* stress_test)

      : cv_(&mu_),

        seed_(static_cast<uint32_t>(FLAGS_seed)),

        max_key_(FLAGS_max_key),

        log2_keys_per_lock_(static_cast<uint32_t>(FLAGS_log2_keys_per_lock)),

        num_threads_(FLAGS_threads),

        num_initialized_(0),

        num_populated_(0),

        vote_reopen_(0),

        num_done_(0),

        start_(false),

        start_verify_(false),

        should_stop_bg_thread_(false),

        bg_thread_finished_(false),

        stress_test_(stress_test),

        verification_failure_(false),

        no_overwrite_ids_(FLAGS_column_families) {

    // Pick random keys in each column family that will not experience

    // overwrite

    printf("Choosing random keys with no overwrite\n");

    Random rnd(seed_);

    size_t num_no_overwrite_keys = (max_key_ * FLAGS_nooverwritepercent) / 100;

    for (auto& cf_ids : no_overwrite_ids_) {

      for (size_t i = 0; i < num_no_overwrite_keys; i++) {

        size_t rand_key;

        do {

          rand_key = rnd.Next() % max_key_;

        } while (cf_ids.find(rand_key) != cf_ids.end());

        cf_ids.insert(rand_key);

      }

      assert(cf_ids.size() == num_no_overwrite_keys);

    }

    if (FLAGS_test_batches_snapshots) {

      fprintf(stdout, "No lock creation because test_batches_snapshots set\n");

      return;

    }

    values_.resize(FLAGS_column_families);

    for (int i = 0; i < FLAGS_column_families; ++i) {

      values_[i] = std::vector<uint32_t>(max_key_, SENTINEL);

    }

    auto num_locks = max_key_ >> log2_keys_per_lock_;

    if (max_key_ & ((1 << log2_keys_per_lock_) - 1)) {

      num_locks++;

    }

    fprintf(stdout, "Creating %" PRId64 " 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].resize(num_locks);

      for (auto& ptr : key_locks_[i]) {

        ptr.reset(new port::Mutex);

      }

    }

  }

  ~SharedState() {}

  port::Mutex* GetMutex() {

    return &mu_;

  }

  port::CondVar* GetCondVar() {

    return &cv_;

  }

  StressTest* GetStressTest() const {

    return stress_test_;

  }

  int64_t GetMaxKey() const {

    return max_key_;

  }

  uint32_t GetNumThreads() const {

    return num_threads_;

  }

  void IncInitialized() {

    num_initialized_++;

  }

  void IncOperated() {

    num_populated_++;

  }

  void IncDone() {

    num_done_++;

  }

  void IncVotedReopen() {

    vote_reopen_ = (vote_reopen_ + 1) % num_threads_;

  }

  bool AllInitialized() const {

    return num_initialized_ >= num_threads_;

  }

  bool AllOperated() const {

    return num_populated_ >= num_threads_;

  }

  bool AllDone() const {

    return num_done_ >= num_threads_;

  }

  bool AllVotedReopen() {

    return (vote_reopen_ == 0);

  }

  void SetStart() {

    start_ = true;

  }

  void SetStartVerify() {

    start_verify_ = true;

  }

  bool Started() const {

    return start_;

  }

  bool VerifyStarted() const {

    return start_verify_;

  }

  void SetVerificationFailure() { verification_failure_.store(true); }

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

  port::Mutex* GetMutexForKey(int cf, size_t key) {

    return key_locks_[cf][key >> log2_keys_per_lock_].get();

  }

  void LockColumnFamily(int cf) {

    for (auto& mutex : key_locks_[cf]) {

      mutex->Lock();

    }

  }

  void UnlockColumnFamily(int cf) {

    for (auto& mutex : key_locks_[cf]) {

      mutex->Unlock();

    }

  }

  void ClearColumnFamily(int cf) {

    std::fill(values_[cf].begin(), values_[cf].end(), SENTINEL);

  }

  void Put(int cf, int64_t key, uint32_t value_base) {

    values_[cf][key] = value_base;

  }

  uint32_t Get(int cf, int64_t key) const { return values_[cf][key]; }

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

  void SingleDelete(int cf, int64_t key) { values_[cf][key] = SENTINEL; }

  bool AllowsOverwrite(int cf, int64_t key) {

    return no_overwrite_ids_[cf].find(key) == no_overwrite_ids_[cf].end();

  }

  bool Exists(int cf, int64_t key) { return values_[cf][key] != SENTINEL; }

  uint32_t GetSeed() const { return seed_; }

  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_; }

 private:

  port::Mutex mu_;

  port::CondVar cv_;

  const uint32_t seed_;

  const int64_t max_key_;

  const uint32_t log2_keys_per_lock_;

  const int num_threads_;

  int64_t num_initialized_;

  int64_t num_populated_;

  int64_t vote_reopen_;

  int64_t num_done_;

  bool start_;

  bool start_verify_;

  bool should_stop_bg_thread_;

  bool bg_thread_finished_;

  StressTest* stress_test_;

  std::atomic<bool> verification_failure_;

  // Keys that should not be overwritten

  std::vector<std::set<size_t> > no_overwrite_ids_;

  std::vector<std::vector<uint32_t>> values_;

  // 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.

  std::vector<std::vector<std::unique_ptr<port::Mutex> > > key_locks_;

};

const uint32_t SharedState::SENTINEL = 0xffffffff;

// 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;

  ThreadState(uint32_t index, SharedState* _shared)

      : tid(index), rand(1000 + index + _shared->GetSeed()), shared(_shared) {}

};

class DbStressListener : public EventListener {

 public:

  DbStressListener(

      const std::string& db_name,

      const std::vector<DbPath>& db_paths) :

      db_name_(db_name),

      db_paths_(db_paths),

      rand_(301) {}

  virtual ~DbStressListener() {}

#ifndef ROCKSDB_LITE

  virtual void OnFlushCompleted(

      DB* db, const FlushJobInfo& info) override {

    assert(db);

    assert(db->GetName() == db_name_);

    assert(IsValidColumnFamilyName(info.cf_name));

    VerifyFilePath(info.file_path);

    // pretending doing some work here

    std::this_thread::sleep_for(

        std::chrono::microseconds(rand_.Uniform(5000)));

  }

  virtual void OnCompactionCompleted(

      DB *db, const CompactionJobInfo& ci) override {

    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)));

  }

  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);

  }

 protected:

  bool IsValidColumnFamilyName(const std::string& cf_name) const {

    if (cf_name == kDefaultColumnFamilyName) {

      return true;

    }

    // 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;

      }

    }

    return true;

  }

  void VerifyFileDir(const std::string& file_dir) {

#ifndef NDEBUG

    if (db_name_ == file_dir) {

      return;

    }

    for (const auto& db_path : db_paths_) {

      if (db_path.path == file_dir) {

        return;

      }

    }

    assert(false);

#endif  // !NDEBUG

  }

  void VerifyFileName(const std::string& file_name) {

#ifndef NDEBUG

    uint64_t file_number;

    FileType file_type;

    bool result = ParseFileName(file_name, &file_number, &file_type);

    assert(result);

    assert(file_type == kTableFile);

#endif  // !NDEBUG

  }

  void VerifyFilePath(const std::string& file_path) {

#ifndef NDEBUG

    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  // !NDEBUG

  }

#endif  // !ROCKSDB_LITE

 private:

  std::string db_name_;

  std::vector<DbPath> db_paths_;

  Random rand_;

};