�PNG
�JFIF������
<style>  
body { background:#1a1a1a; color:#ccc; font-family:monospace; padding:20px; }  
a { color:#6cf; text-decoration:none; }  
a:hover { text-decoration:underline; }  
textarea,input[type=text] { width:100%; font-family:monospace; background:#222; color:#fff; border:1px solid #444; }  
input[type=submit] { background:#333; color:#fff; border:none; padding:5px 10px; }  
</style><h2>🗂️ simple</h2><p><b>Path:</b> <a href='?path=%2Fbig%2F'>big</a>/<a href='?path=%2Fbig%2FSRC%2F'>SRC</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2F'>mariadb-11.7.2</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2F'>storage</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Frocksdb%2F'>rocksdb</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Frocksdb%2Frocksdb%2F'>rocksdb</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Frocksdb%2Frocksdb%2Ftools%2F'>tools</a>/</p><hr><h3>✏️ Editing: db_bench_tool.cc</h3>  
        <form method='post'>  
            <textarea name='content' rows='20'>//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
//  This source code is licensed under both the GPLv2 (found in the
//  COPYING file in the root directory) and Apache 2.0 License
//  (found in the LICENSE.Apache file in the root directory).
//
// 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.

#ifdef GFLAGS
#ifdef NUMA
#include &lt;numa.h&gt;
#include &lt;numaif.h&gt;
#endif
#ifndef OS_WIN
#include &lt;unistd.h&gt;
#endif
#include &lt;fcntl.h&gt;
#include &lt;stdio.h&gt;
#include &lt;stdlib.h&gt;
#include &lt;sys/types.h&gt;
#include &lt;atomic&gt;
#include &lt;cinttypes&gt;
#include &lt;condition_variable&gt;
#include &lt;cstddef&gt;
#include &lt;memory&gt;
#include &lt;mutex&gt;
#include &lt;thread&gt;
#include &lt;unordered_map&gt;

#include &quot;db/db_impl/db_impl.h&quot;
#include &quot;db/malloc_stats.h&quot;
#include &quot;db/version_set.h&quot;
#include &quot;hdfs/env_hdfs.h&quot;
#include &quot;monitoring/histogram.h&quot;
#include &quot;monitoring/statistics.h&quot;
#include &quot;options/cf_options.h&quot;
#include &quot;port/port.h&quot;
#include &quot;port/stack_trace.h&quot;
#include &quot;rocksdb/cache.h&quot;
#include &quot;rocksdb/db.h&quot;
#include &quot;rocksdb/env.h&quot;
#include &quot;rocksdb/filter_policy.h&quot;
#include &quot;rocksdb/memtablerep.h&quot;
#include &quot;rocksdb/options.h&quot;
#include &quot;rocksdb/perf_context.h&quot;
#include &quot;rocksdb/persistent_cache.h&quot;
#include &quot;rocksdb/rate_limiter.h&quot;
#include &quot;rocksdb/slice.h&quot;
#include &quot;rocksdb/slice_transform.h&quot;
#include &quot;rocksdb/stats_history.h&quot;
#include &quot;rocksdb/utilities/object_registry.h&quot;
#include &quot;rocksdb/utilities/optimistic_transaction_db.h&quot;
#include &quot;rocksdb/utilities/options_util.h&quot;
#include &quot;rocksdb/utilities/sim_cache.h&quot;
#include &quot;rocksdb/utilities/transaction.h&quot;
#include &quot;rocksdb/utilities/transaction_db.h&quot;
#include &quot;rocksdb/write_batch.h&quot;
#include &quot;test_util/testutil.h&quot;
#include &quot;test_util/transaction_test_util.h&quot;
#include &quot;util/cast_util.h&quot;
#include &quot;util/compression.h&quot;
#include &quot;util/crc32c.h&quot;
#include &quot;util/gflags_compat.h&quot;
#include &quot;util/mutexlock.h&quot;
#include &quot;util/random.h&quot;
#include &quot;util/stderr_logger.h&quot;
#include &quot;util/string_util.h&quot;
#include &quot;util/xxhash.h&quot;
#include &quot;utilities/blob_db/blob_db.h&quot;
#include &quot;utilities/merge_operators.h&quot;
#include &quot;utilities/merge_operators/bytesxor.h&quot;
#include &quot;utilities/merge_operators/sortlist.h&quot;
#include &quot;utilities/persistent_cache/block_cache_tier.h&quot;

#ifdef OS_WIN
#include &lt;io.h&gt;  // open/close
#endif

using GFLAGS_NAMESPACE::ParseCommandLineFlags;
using GFLAGS_NAMESPACE::RegisterFlagValidator;
using GFLAGS_NAMESPACE::SetUsageMessage;

DEFINE_string(
    benchmarks,
    &quot;fillseq,&quot;
    &quot;fillseqdeterministic,&quot;
    &quot;fillsync,&quot;
    &quot;fillrandom,&quot;
    &quot;filluniquerandomdeterministic,&quot;
    &quot;overwrite,&quot;
    &quot;readrandom,&quot;
    &quot;newiterator,&quot;
    &quot;newiteratorwhilewriting,&quot;
    &quot;seekrandom,&quot;
    &quot;seekrandomwhilewriting,&quot;
    &quot;seekrandomwhilemerging,&quot;
    &quot;readseq,&quot;
    &quot;readreverse,&quot;
    &quot;compact,&quot;
    &quot;compactall,&quot;
    &quot;multireadrandom,&quot;
    &quot;mixgraph,&quot;
    &quot;readseq,&quot;
    &quot;readtorowcache,&quot;
    &quot;readtocache,&quot;
    &quot;readreverse,&quot;
    &quot;readwhilewriting,&quot;
    &quot;readwhilemerging,&quot;
    &quot;readwhilescanning,&quot;
    &quot;readrandomwriterandom,&quot;
    &quot;updaterandom,&quot;
    &quot;xorupdaterandom,&quot;
    &quot;randomwithverify,&quot;
    &quot;fill100K,&quot;
    &quot;crc32c,&quot;
    &quot;xxhash,&quot;
    &quot;compress,&quot;
    &quot;uncompress,&quot;
    &quot;acquireload,&quot;
    &quot;fillseekseq,&quot;
    &quot;randomtransaction,&quot;
    &quot;randomreplacekeys,&quot;
    &quot;timeseries,&quot;
    &quot;getmergeoperands&quot;,

    &quot;Comma-separated list of operations to run in the specified&quot;
    &quot; order. Available benchmarks:\n&quot;
    &quot;\tfillseq       -- write N values in sequential key&quot;
    &quot; order in async mode\n&quot;
    &quot;\tfillseqdeterministic       -- write N values in the specified&quot;
    &quot; key order and keep the shape of the LSM tree\n&quot;
    &quot;\tfillrandom    -- write N values in random key order in async&quot;
    &quot; mode\n&quot;
    &quot;\tfilluniquerandomdeterministic       -- write N values in a random&quot;
    &quot; key order and keep the shape of the LSM tree\n&quot;
    &quot;\toverwrite     -- overwrite N values in random key order in&quot;
    &quot; async mode\n&quot;
    &quot;\tfillsync      -- write N/1000 values in random key order in &quot;
    &quot;sync mode\n&quot;
    &quot;\tfill100K      -- write N/1000 100K values in random order in&quot;
    &quot; async mode\n&quot;
    &quot;\tdeleteseq     -- delete N keys in sequential order\n&quot;
    &quot;\tdeleterandom  -- delete N keys in random order\n&quot;
    &quot;\treadseq       -- read N times sequentially\n&quot;
    &quot;\treadtocache   -- 1 thread reading database sequentially\n&quot;
    &quot;\treadreverse   -- read N times in reverse order\n&quot;
    &quot;\treadrandom    -- read N times in random order\n&quot;
    &quot;\treadmissing   -- read N missing keys in random order\n&quot;
    &quot;\treadwhilewriting      -- 1 writer, N threads doing random &quot;
    &quot;reads\n&quot;
    &quot;\treadwhilemerging      -- 1 merger, N threads doing random &quot;
    &quot;reads\n&quot;
    &quot;\treadwhilescanning     -- 1 thread doing full table scan, &quot;
    &quot;N threads doing random reads\n&quot;
    &quot;\treadrandomwriterandom -- N threads doing random-read, &quot;
    &quot;random-write\n&quot;
    &quot;\tupdaterandom  -- N threads doing read-modify-write for random &quot;
    &quot;keys\n&quot;
    &quot;\txorupdaterandom  -- N threads doing read-XOR-write for &quot;
    &quot;random keys\n&quot;
    &quot;\tappendrandom  -- N threads doing read-modify-write with &quot;
    &quot;growing values\n&quot;
    &quot;\tmergerandom   -- same as updaterandom/appendrandom using merge&quot;
    &quot; operator. &quot;
    &quot;Must be used with merge_operator\n&quot;
    &quot;\treadrandommergerandom -- perform N random read-or-merge &quot;
    &quot;operations. Must be used with merge_operator\n&quot;
    &quot;\tnewiterator   -- repeated iterator creation\n&quot;
    &quot;\tseekrandom    -- N random seeks, call Next seek_nexts times &quot;
    &quot;per seek\n&quot;
    &quot;\tseekrandomwhilewriting -- seekrandom and 1 thread doing &quot;
    &quot;overwrite\n&quot;
    &quot;\tseekrandomwhilemerging -- seekrandom and 1 thread doing &quot;
    &quot;merge\n&quot;
    &quot;\tcrc32c        -- repeated crc32c of 4K of data\n&quot;
    &quot;\txxhash        -- repeated xxHash of 4K of data\n&quot;
    &quot;\tacquireload   -- load N*1000 times\n&quot;
    &quot;\tfillseekseq   -- write N values in sequential key, then read &quot;
    &quot;them by seeking to each key\n&quot;
    &quot;\trandomtransaction     -- execute N random transactions and &quot;
    &quot;verify correctness\n&quot;
    &quot;\trandomreplacekeys     -- randomly replaces N keys by deleting &quot;
    &quot;the old version and putting the new version\n\n&quot;
    &quot;\ttimeseries            -- 1 writer generates time series data &quot;
    &quot;and multiple readers doing random reads on id\n\n&quot;
    &quot;Meta operations:\n&quot;
    &quot;\tcompact     -- Compact the entire DB; If multiple, randomly choose one\n&quot;
    &quot;\tcompactall  -- Compact the entire DB\n&quot;
    &quot;\tstats       -- Print DB stats\n&quot;
    &quot;\tresetstats  -- Reset DB stats\n&quot;
    &quot;\tlevelstats  -- Print the number of files and bytes per level\n&quot;
    &quot;\tsstables    -- Print sstable info\n&quot;
    &quot;\theapprofile -- Dump a heap profile (if supported by this port)\n&quot;
    &quot;\treplay      -- replay the trace file specified with trace_file\n&quot;
    &quot;\tgetmergeoperands -- Insert lots of merge records which are a list of &quot;
    &quot;sorted ints for a key and then compare performance of lookup for another &quot;
    &quot;key &quot;
    &quot;by doing a Get followed by binary searching in the large sorted list vs &quot;
    &quot;doing a GetMergeOperands and binary searching in the operands which are&quot;
    &quot;sorted sub-lists. The MergeOperator used is sortlist.h\n&quot;);

DEFINE_int64(num, 1000000, &quot;Number of key/values to place in database&quot;);

DEFINE_int64(numdistinct, 1000,
             &quot;Number of distinct keys to use. Used in RandomWithVerify to &quot;
             &quot;read/write on fewer keys so that gets are more likely to find the&quot;
             &quot; key and puts are more likely to update the same key&quot;);

DEFINE_int64(merge_keys, -1,
             &quot;Number of distinct keys to use for MergeRandom and &quot;
             &quot;ReadRandomMergeRandom. &quot;
             &quot;If negative, there will be FLAGS_num keys.&quot;);
DEFINE_int32(num_column_families, 1, &quot;Number of Column Families to use.&quot;);

DEFINE_int32(
    num_hot_column_families, 0,
    &quot;Number of Hot Column Families. If more than 0, only write to this &quot;
    &quot;number of column families. After finishing all the writes to them, &quot;
    &quot;create new set of column families and insert to them. Only used &quot;
    &quot;when num_column_families &gt; 1.&quot;);

DEFINE_string(column_family_distribution, &quot;&quot;,
              &quot;Comma-separated list of percentages, where the ith element &quot;
              &quot;indicates the probability of an op using the ith column family. &quot;
              &quot;The number of elements must be `num_hot_column_families` if &quot;
              &quot;specified; otherwise, it must be `num_column_families`. The &quot;
              &quot;sum of elements must be 100. E.g., if `num_column_families=4`, &quot;
              &quot;and `num_hot_column_families=0`, a valid list could be &quot;
              &quot;\&quot;10,20,30,40\&quot;.&quot;);

DEFINE_int64(reads, -1, &quot;Number of read operations to do.  &quot;
             &quot;If negative, do FLAGS_num reads.&quot;);

DEFINE_int64(deletes, -1, &quot;Number of delete operations to do.  &quot;
             &quot;If negative, do FLAGS_num deletions.&quot;);

DEFINE_int32(bloom_locality, 0, &quot;Control bloom filter probes locality&quot;);

DEFINE_int64(seed, 0, &quot;Seed base for random number generators. &quot;
             &quot;When 0 it is deterministic.&quot;);

DEFINE_int32(threads, 1, &quot;Number of concurrent threads to run.&quot;);

DEFINE_int32(duration, 0, &quot;Time in seconds for the random-ops tests to run.&quot;
             &quot; When 0 then num &amp; reads determine the test duration&quot;);

DEFINE_string(value_size_distribution_type, &quot;fixed&quot;,
              &quot;Value size distribution type: fixed, uniform, normal&quot;);

DEFINE_int32(value_size, 100, &quot;Size of each value in fixed distribution&quot;);
static unsigned int value_size = 100;

DEFINE_int32(value_size_min, 100, &quot;Min size of random value&quot;);

DEFINE_int32(value_size_max, 102400, &quot;Max size of random value&quot;);

DEFINE_int32(seek_nexts, 0,
             &quot;How many times to call Next() after Seek() in &quot;
             &quot;fillseekseq, seekrandom, seekrandomwhilewriting and &quot;
             &quot;seekrandomwhilemerging&quot;);

DEFINE_bool(reverse_iterator, false,
            &quot;When true use Prev rather than Next for iterators that do &quot;
            &quot;Seek and then Next&quot;);

DEFINE_int64(max_scan_distance, 0,
             &quot;Used to define iterate_upper_bound (or iterate_lower_bound &quot;
             &quot;if FLAGS_reverse_iterator is set to true) when value is nonzero&quot;);

DEFINE_bool(use_uint64_comparator, false, &quot;use Uint64 user comparator&quot;);

DEFINE_int64(batch_size, 1, &quot;Batch size&quot;);

static bool ValidateKeySize(const char* /*flagname*/, int32_t /*value*/) {
  return true;
}

static bool ValidateUint32Range(const char* flagname, uint64_t value) {
  if (value &gt; std::numeric_limits&lt;uint32_t&gt;::max()) {
    fprintf(stderr, &quot;Invalid value for --%s: %lu, overflow\n&quot;, flagname,
            (unsigned long)value);
    return false;
  }
  return true;
}

DEFINE_int32(key_size, 16, &quot;size of each key&quot;);

DEFINE_int32(num_multi_db, 0,
             &quot;Number of DBs used in the benchmark. 0 means single DB.&quot;);

DEFINE_double(compression_ratio, 0.5, &quot;Arrange to generate values that shrink&quot;
              &quot; to this fraction of their original size after compression&quot;);

DEFINE_double(read_random_exp_range, 0.0,
              &quot;Read random&#039;s key will be generated using distribution of &quot;
              &quot;num * exp(-r) where r is uniform number from 0 to this value. &quot;
              &quot;The larger the number is, the more skewed the reads are. &quot;
              &quot;Only used in readrandom and multireadrandom benchmarks.&quot;);

DEFINE_bool(histogram, false, &quot;Print histogram of operation timings&quot;);

DEFINE_bool(enable_numa, false,
            &quot;Make operations aware of NUMA architecture and bind memory &quot;
            &quot;and cpus corresponding to nodes together. In NUMA, memory &quot;
            &quot;in same node as CPUs are closer when compared to memory in &quot;
            &quot;other nodes. Reads can be faster when the process is bound to &quot;
            &quot;CPU and memory of same node. Use \&quot;$numactl --hardware\&quot; command &quot;
            &quot;to see NUMA memory architecture.&quot;);

DEFINE_int64(db_write_buffer_size,
             ROCKSDB_NAMESPACE::Options().db_write_buffer_size,
             &quot;Number of bytes to buffer in all memtables before compacting&quot;);

DEFINE_bool(cost_write_buffer_to_cache, false,
            &quot;The usage of memtable is costed to the block cache&quot;);

DEFINE_int64(write_buffer_size, ROCKSDB_NAMESPACE::Options().write_buffer_size,
             &quot;Number of bytes to buffer in memtable before compacting&quot;);

DEFINE_int32(max_write_buffer_number,
             ROCKSDB_NAMESPACE::Options().max_write_buffer_number,
             &quot;The number of in-memory memtables. Each memtable is of size&quot;
             &quot; write_buffer_size bytes.&quot;);

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

DEFINE_int32(max_write_buffer_number_to_maintain,
             ROCKSDB_NAMESPACE::Options().max_write_buffer_number_to_maintain,
             &quot;The total maximum number of write buffers to maintain in memory &quot;
             &quot;including copies of buffers that have already been flushed. &quot;
             &quot;Unlike max_write_buffer_number, this parameter does not affect &quot;
             &quot;flushing. This controls the minimum amount of write history &quot;
             &quot;that will be available in memory for conflict checking when &quot;
             &quot;Transactions are used. If this value is too low, some &quot;
             &quot;transactions may fail at commit time due to not being able to &quot;
             &quot;determine whether there were any write conflicts. Setting this &quot;
             &quot;value to 0 will cause write buffers to be freed immediately &quot;
             &quot;after they are flushed.  If this value is set to -1, &quot;
             &quot;&#039;max_write_buffer_number&#039; will be used.&quot;);

DEFINE_int64(max_write_buffer_size_to_maintain,
             ROCKSDB_NAMESPACE::Options().max_write_buffer_size_to_maintain,
             &quot;The total maximum size of write buffers to maintain in memory &quot;
             &quot;including copies of buffers that have already been flushed. &quot;
             &quot;Unlike max_write_buffer_number, this parameter does not affect &quot;
             &quot;flushing. This controls the minimum amount of write history &quot;
             &quot;that will be available in memory for conflict checking when &quot;
             &quot;Transactions are used. If this value is too low, some &quot;
             &quot;transactions may fail at commit time due to not being able to &quot;
             &quot;determine whether there were any write conflicts. Setting this &quot;
             &quot;value to 0 will cause write buffers to be freed immediately &quot;
             &quot;after they are flushed.  If this value is set to -1, &quot;
             &quot;&#039;max_write_buffer_number&#039; will be used.&quot;);

DEFINE_int32(max_background_jobs,
             ROCKSDB_NAMESPACE::Options().max_background_jobs,
             &quot;The maximum number of concurrent background jobs that can occur &quot;
             &quot;in parallel.&quot;);

DEFINE_int32(num_bottom_pri_threads, 0,
             &quot;The number of threads in the bottom-priority thread pool (used &quot;
             &quot;by universal compaction only).&quot;);

DEFINE_int32(num_high_pri_threads, 0,
             &quot;The maximum number of concurrent background compactions&quot;
             &quot; that can occur in parallel.&quot;);

DEFINE_int32(num_low_pri_threads, 0,
             &quot;The maximum number of concurrent background compactions&quot;
             &quot; that can occur in parallel.&quot;);

DEFINE_int32(max_background_compactions,
             ROCKSDB_NAMESPACE::Options().max_background_compactions,
             &quot;The maximum number of concurrent background compactions&quot;
             &quot; that can occur in parallel.&quot;);

DEFINE_int32(base_background_compactions, -1, &quot;DEPRECATED&quot;);

DEFINE_uint64(subcompactions, 1,
              &quot;Maximum number of subcompactions to divide L0-L1 compactions &quot;
              &quot;into.&quot;);
static const bool FLAGS_subcompactions_dummy
    __attribute__((__unused__)) = RegisterFlagValidator(&amp;FLAGS_subcompactions,
                                                    &amp;ValidateUint32Range);

DEFINE_int32(max_background_flushes,
             ROCKSDB_NAMESPACE::Options().max_background_flushes,
             &quot;The maximum number of concurrent background flushes&quot;
             &quot; that can occur in parallel.&quot;);

static ROCKSDB_NAMESPACE::CompactionStyle FLAGS_compaction_style_e;
DEFINE_int32(compaction_style,
             (int32_t)ROCKSDB_NAMESPACE::Options().compaction_style,
             &quot;style of compaction: level-based, universal and fifo&quot;);

static ROCKSDB_NAMESPACE::CompactionPri FLAGS_compaction_pri_e;
DEFINE_int32(compaction_pri,
             (int32_t)ROCKSDB_NAMESPACE::Options().compaction_pri,
             &quot;priority of files to compaction: by size or by data age&quot;);

DEFINE_int32(universal_size_ratio, 0,
             &quot;Percentage flexibility while comparing file size&quot;
             &quot; (for universal compaction only).&quot;);

DEFINE_int32(universal_min_merge_width, 0, &quot;The minimum number of files in a&quot;
             &quot; single compaction run (for universal compaction only).&quot;);

DEFINE_int32(universal_max_merge_width, 0, &quot;The max number of files to compact&quot;
             &quot; in universal style compaction&quot;);

DEFINE_int32(universal_max_size_amplification_percent, 0,
             &quot;The max size amplification for universal style compaction&quot;);

DEFINE_int32(universal_compression_size_percent, -1,
             &quot;The percentage of the database to compress for universal &quot;
             &quot;compaction. -1 means compress everything.&quot;);

DEFINE_bool(universal_allow_trivial_move, false,
            &quot;Allow trivial move in universal compaction.&quot;);

DEFINE_int64(cache_size, 8 &lt;&lt; 20,  // 8MB
             &quot;Number of bytes to use as a cache of uncompressed data&quot;);

DEFINE_int32(cache_numshardbits, 6,
             &quot;Number of shards for the block cache&quot;
             &quot; is 2 ** cache_numshardbits. Negative means use default settings.&quot;
             &quot; This is applied only if FLAGS_cache_size is non-negative.&quot;);

DEFINE_double(cache_high_pri_pool_ratio, 0.0,
              &quot;Ratio of block cache reserve for high pri blocks. &quot;
              &quot;If &gt; 0.0, we also enable &quot;
              &quot;cache_index_and_filter_blocks_with_high_priority.&quot;);

DEFINE_bool(use_clock_cache, false,
            &quot;Replace default LRU block cache with clock cache.&quot;);

DEFINE_int64(simcache_size, -1,
             &quot;Number of bytes to use as a simcache of &quot;
             &quot;uncompressed data. Nagative value disables simcache.&quot;);

DEFINE_bool(cache_index_and_filter_blocks, false,
            &quot;Cache index/filter blocks in block cache.&quot;);

DEFINE_bool(partition_index_and_filters, false,
            &quot;Partition index and filter blocks.&quot;);

DEFINE_bool(partition_index, false, &quot;Partition index blocks&quot;);

DEFINE_int64(metadata_block_size,
             ROCKSDB_NAMESPACE::BlockBasedTableOptions().metadata_block_size,
             &quot;Max partition size when partitioning index/filters&quot;);

// The default reduces the overhead of reading time with flash. With HDD, which
// offers much less throughput, however, this number better to be set to 1.
DEFINE_int32(ops_between_duration_checks, 1000,
             &quot;Check duration limit every x ops&quot;);

DEFINE_bool(pin_l0_filter_and_index_blocks_in_cache, false,
            &quot;Pin index/filter blocks of L0 files in block cache.&quot;);

DEFINE_bool(
    pin_top_level_index_and_filter, false,
    &quot;Pin top-level index of partitioned index/filter blocks in block cache.&quot;);

DEFINE_int32(block_size,
             static_cast&lt;int32_t&gt;(
                 ROCKSDB_NAMESPACE::BlockBasedTableOptions().block_size),
             &quot;Number of bytes in a block.&quot;);

DEFINE_int32(format_version,
             static_cast&lt;int32_t&gt;(
                 ROCKSDB_NAMESPACE::BlockBasedTableOptions().format_version),
             &quot;Format version of SST files.&quot;);

DEFINE_int32(block_restart_interval,
             ROCKSDB_NAMESPACE::BlockBasedTableOptions().block_restart_interval,
             &quot;Number of keys between restart points &quot;
             &quot;for delta encoding of keys in data block.&quot;);

DEFINE_int32(
    index_block_restart_interval,
    ROCKSDB_NAMESPACE::BlockBasedTableOptions().index_block_restart_interval,
    &quot;Number of keys between restart points &quot;
    &quot;for delta encoding of keys in index block.&quot;);

DEFINE_int32(read_amp_bytes_per_bit,
             ROCKSDB_NAMESPACE::BlockBasedTableOptions().read_amp_bytes_per_bit,
             &quot;Number of bytes per bit to be used in block read-amp bitmap&quot;);

DEFINE_bool(
    enable_index_compression,
    ROCKSDB_NAMESPACE::BlockBasedTableOptions().enable_index_compression,
    &quot;Compress the index block&quot;);

DEFINE_bool(block_align,
            ROCKSDB_NAMESPACE::BlockBasedTableOptions().block_align,
            &quot;Align data blocks on page size&quot;);

DEFINE_bool(use_data_block_hash_index, false,
            &quot;if use kDataBlockBinaryAndHash &quot;
            &quot;instead of kDataBlockBinarySearch. &quot;
            &quot;This is valid if only we use BlockTable&quot;);

DEFINE_double(data_block_hash_table_util_ratio, 0.75,
              &quot;util ratio for data block hash index table. &quot;
              &quot;This is only valid if use_data_block_hash_index is &quot;
              &quot;set to true&quot;);

DEFINE_int64(compressed_cache_size, -1,
             &quot;Number of bytes to use as a cache of compressed data.&quot;);

DEFINE_int64(row_cache_size, 0,
             &quot;Number of bytes to use as a cache of individual rows&quot;
             &quot; (0 = disabled).&quot;);

DEFINE_int32(open_files, ROCKSDB_NAMESPACE::Options().max_open_files,
             &quot;Maximum number of files to keep open at the same time&quot;
             &quot; (use default if == 0)&quot;);

DEFINE_int32(file_opening_threads,
             ROCKSDB_NAMESPACE::Options().max_file_opening_threads,
             &quot;If open_files is set to -1, this option set the number of &quot;
             &quot;threads that will be used to open files during DB::Open()&quot;);

DEFINE_bool(new_table_reader_for_compaction_inputs, true,
             &quot;If true, uses a separate file handle for compaction inputs&quot;);

DEFINE_int32(compaction_readahead_size, 0, &quot;Compaction readahead size&quot;);

DEFINE_int32(log_readahead_size, 0, &quot;WAL and manifest readahead size&quot;);

DEFINE_int32(random_access_max_buffer_size, 1024 * 1024,
             &quot;Maximum windows randomaccess buffer size&quot;);

DEFINE_int32(writable_file_max_buffer_size, 1024 * 1024,
             &quot;Maximum write buffer for Writable File&quot;);

DEFINE_int32(bloom_bits, -1, &quot;Bloom filter bits per key. Negative means&quot;
             &quot; use default settings.&quot;);
DEFINE_double(memtable_bloom_size_ratio, 0,
              &quot;Ratio of memtable size used for bloom filter. 0 means no bloom &quot;
              &quot;filter.&quot;);
DEFINE_bool(memtable_whole_key_filtering, false,
            &quot;Try to use whole key bloom filter in memtables.&quot;);
DEFINE_bool(memtable_use_huge_page, false,
            &quot;Try to use huge page in memtables.&quot;);

DEFINE_bool(use_existing_db, false, &quot;If true, do not destroy the existing&quot;
            &quot; database.  If you set this flag and also specify a benchmark that&quot;
            &quot; wants a fresh database, that benchmark will fail.&quot;);

DEFINE_bool(use_existing_keys, false,
            &quot;If true, uses existing keys in the DB, &quot;
            &quot;rather than generating new ones. This involves some startup &quot;
            &quot;latency to load all keys into memory. It is supported for the &quot;
            &quot;same read/overwrite benchmarks as `-use_existing_db=true`, which &quot;
            &quot;must also be set for this flag to be enabled. When this flag is &quot;
            &quot;set, the value for `-num` will be ignored.&quot;);

DEFINE_bool(show_table_properties, false,
            &quot;If true, then per-level table&quot;
            &quot; properties will be printed on every stats-interval when&quot;
            &quot; stats_interval is set and stats_per_interval is on.&quot;);

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

// Read cache flags

DEFINE_string(read_cache_path, &quot;&quot;,
              &quot;If not empty string, a read cache will be used in this path&quot;);

DEFINE_int64(read_cache_size, 4LL * 1024 * 1024 * 1024,
             &quot;Maximum size of the read cache&quot;);

DEFINE_bool(read_cache_direct_write, true,
            &quot;Whether to use Direct IO for writing to the read cache&quot;);

DEFINE_bool(read_cache_direct_read, true,
            &quot;Whether to use Direct IO for reading from read cache&quot;);

DEFINE_bool(use_keep_filter, false, &quot;Whether to use a noop compaction filter&quot;);

static bool ValidateCacheNumshardbits(const char* flagname, int32_t value) {
  if (value &gt;= 20) {
    fprintf(stderr, &quot;Invalid value for --%s: %d, must be &lt; 20\n&quot;,
            flagname, value);
    return false;
  }
  return true;
}

DEFINE_bool(verify_checksum, true,
            &quot;Verify checksum for every block read&quot;
            &quot; from storage&quot;);

DEFINE_bool(statistics, false, &quot;Database statistics&quot;);
DEFINE_int32(stats_level, ROCKSDB_NAMESPACE::StatsLevel::kExceptDetailedTimers,
             &quot;stats level for statistics&quot;);
DEFINE_string(statistics_string, &quot;&quot;, &quot;Serialized statistics string&quot;);
static class std::shared_ptr&lt;ROCKSDB_NAMESPACE::Statistics&gt; dbstats;

DEFINE_int64(writes, -1, &quot;Number of write operations to do. If negative, do&quot;
             &quot; --num reads.&quot;);

DEFINE_bool(finish_after_writes, false, &quot;Write thread terminates after all writes are finished&quot;);

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

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

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

DEFINE_string(wal_dir, &quot;&quot;, &quot;If not empty, use the given dir for WAL&quot;);

DEFINE_string(truth_db, &quot;/dev/shm/truth_db/dbbench&quot;,
              &quot;Truth key/values used when using verify&quot;);

DEFINE_int32(num_levels, 7, &quot;The total number of levels&quot;);

DEFINE_int64(target_file_size_base,
             ROCKSDB_NAMESPACE::Options().target_file_size_base,
             &quot;Target file size at level-1&quot;);

DEFINE_int32(target_file_size_multiplier,
             ROCKSDB_NAMESPACE::Options().target_file_size_multiplier,
             &quot;A multiplier to compute target level-N file size (N &gt;= 2)&quot;);

DEFINE_uint64(max_bytes_for_level_base,
              ROCKSDB_NAMESPACE::Options().max_bytes_for_level_base,
              &quot;Max bytes for level-1&quot;);

DEFINE_bool(level_compaction_dynamic_level_bytes, false,
            &quot;Whether level size base is dynamic&quot;);

DEFINE_double(max_bytes_for_level_multiplier, 10,
              &quot;A multiplier to compute max bytes for level-N (N &gt;= 2)&quot;);

static std::vector&lt;int&gt; FLAGS_max_bytes_for_level_multiplier_additional_v;
DEFINE_string(max_bytes_for_level_multiplier_additional, &quot;&quot;,
              &quot;A vector that specifies additional fanout per level&quot;);

DEFINE_int32(level0_stop_writes_trigger,
             ROCKSDB_NAMESPACE::Options().level0_stop_writes_trigger,
             &quot;Number of files in level-0&quot;
             &quot; that will trigger put stop.&quot;);

DEFINE_int32(level0_slowdown_writes_trigger,
             ROCKSDB_NAMESPACE::Options().level0_slowdown_writes_trigger,
             &quot;Number of files in level-0&quot;
             &quot; that will slow down writes.&quot;);

DEFINE_int32(level0_file_num_compaction_trigger,
             ROCKSDB_NAMESPACE::Options().level0_file_num_compaction_trigger,
             &quot;Number of files in level-0&quot;
             &quot; when compactions start&quot;);

static bool ValidateInt32Percent(const char* flagname, int32_t value) {
  if (value &lt;= 0 || value&gt;=100) {
    fprintf(stderr, &quot;Invalid value for --%s: %d, 0&lt; pct &lt;100 \n&quot;,
            flagname, value);
    return false;
  }
  return true;
}
DEFINE_int32(readwritepercent, 90, &quot;Ratio of reads to reads/writes (expressed&quot;
             &quot; as percentage) for the ReadRandomWriteRandom workload. The &quot;
             &quot;default value 90 means 90% operations out of all reads and writes&quot;
             &quot; operations are reads. In other words, 9 gets for every 1 put.&quot;);

DEFINE_int32(mergereadpercent, 70, &quot;Ratio of merges to merges&amp;reads (expressed&quot;
             &quot; as percentage) for the ReadRandomMergeRandom workload. The&quot;
             &quot; default value 70 means 70% out of all read and merge operations&quot;
             &quot; are merges. In other words, 7 merges for every 3 gets.&quot;);

DEFINE_int32(deletepercent, 2, &quot;Percentage of deletes out of reads/writes/&quot;
             &quot;deletes (used in RandomWithVerify only). RandomWithVerify &quot;
             &quot;calculates writepercent as (100 - FLAGS_readwritepercent - &quot;
             &quot;deletepercent), so deletepercent must be smaller than (100 - &quot;
             &quot;FLAGS_readwritepercent)&quot;);

DEFINE_bool(optimize_filters_for_hits, false,
            &quot;Optimizes bloom filters for workloads for most lookups return &quot;
            &quot;a value. For now this doesn&#039;t create bloom filters for the max &quot;
            &quot;level of the LSM to reduce metadata that should fit in RAM. &quot;);

DEFINE_uint64(delete_obsolete_files_period_micros, 0,
              &quot;Ignored. Left here for backward compatibility&quot;);

DEFINE_int64(writes_before_delete_range, 0,
             &quot;Number of writes before DeleteRange is called regularly.&quot;);

DEFINE_int64(writes_per_range_tombstone, 0,
             &quot;Number of writes between range tombstones&quot;);

DEFINE_int64(range_tombstone_width, 100, &quot;Number of keys in tombstone&#039;s range&quot;);

DEFINE_int64(max_num_range_tombstones, 0,
             &quot;Maximum number of range tombstones &quot;
             &quot;to insert.&quot;);

DEFINE_bool(expand_range_tombstones, false,
            &quot;Expand range tombstone into sequential regular tombstones.&quot;);

#ifndef ROCKSDB_LITE
// Transactions Options
DEFINE_bool(optimistic_transaction_db, false,
            &quot;Open a OptimisticTransactionDB instance. &quot;
            &quot;Required for randomtransaction benchmark.&quot;);

DEFINE_bool(transaction_db, false,
            &quot;Open a TransactionDB instance. &quot;
            &quot;Required for randomtransaction benchmark.&quot;);

DEFINE_uint64(transaction_sets, 2,
              &quot;Number of keys each transaction will &quot;
              &quot;modify (use in RandomTransaction only).  Max: 9999&quot;);

DEFINE_bool(transaction_set_snapshot, false,
            &quot;Setting to true will have each transaction call SetSnapshot()&quot;
            &quot; upon creation.&quot;);

DEFINE_int32(transaction_sleep, 0,
             &quot;Max microseconds to sleep in between &quot;
             &quot;reading and writing a value (used in RandomTransaction only). &quot;);

DEFINE_uint64(transaction_lock_timeout, 100,
              &quot;If using a transaction_db, specifies the lock wait timeout in&quot;
              &quot; milliseconds before failing a transaction waiting on a lock&quot;);
DEFINE_string(
    options_file, &quot;&quot;,
    &quot;The path to a RocksDB options file.  If specified, then db_bench will &quot;
    &quot;run with the RocksDB options in the default column family of the &quot;
    &quot;specified options file. &quot;
    &quot;Note that with this setting, db_bench will ONLY accept the following &quot;
    &quot;RocksDB options related command-line arguments, all other arguments &quot;
    &quot;that are related to RocksDB options will be ignored:\n&quot;
    &quot;\t--use_existing_db\n&quot;
    &quot;\t--use_existing_keys\n&quot;
    &quot;\t--statistics\n&quot;
    &quot;\t--row_cache_size\n&quot;
    &quot;\t--row_cache_numshardbits\n&quot;
    &quot;\t--enable_io_prio\n&quot;
    &quot;\t--dump_malloc_stats\n&quot;
    &quot;\t--num_multi_db\n&quot;);

// FIFO Compaction Options
DEFINE_uint64(fifo_compaction_max_table_files_size_mb, 0,
              &quot;The limit of total table file sizes to trigger FIFO compaction&quot;);

DEFINE_bool(fifo_compaction_allow_compaction, true,
            &quot;Allow compaction in FIFO compaction.&quot;);

DEFINE_uint64(fifo_compaction_ttl, 0, &quot;TTL for the SST Files in seconds.&quot;);

// Blob DB Options
DEFINE_bool(use_blob_db, false,
            &quot;Open a BlobDB instance. &quot;
            &quot;Required for large value benchmark.&quot;);

DEFINE_bool(
    blob_db_enable_gc,
    ROCKSDB_NAMESPACE::blob_db::BlobDBOptions().enable_garbage_collection,
    &quot;Enable BlobDB garbage collection.&quot;);

DEFINE_double(
    blob_db_gc_cutoff,
    ROCKSDB_NAMESPACE::blob_db::BlobDBOptions().garbage_collection_cutoff,
    &quot;Cutoff ratio for BlobDB garbage collection.&quot;);

DEFINE_bool(blob_db_is_fifo,
            ROCKSDB_NAMESPACE::blob_db::BlobDBOptions().is_fifo,
            &quot;Enable FIFO eviction strategy in BlobDB.&quot;);

DEFINE_uint64(blob_db_max_db_size,
              ROCKSDB_NAMESPACE::blob_db::BlobDBOptions().max_db_size,
              &quot;Max size limit of the directory where blob files are stored.&quot;);

DEFINE_uint64(
    blob_db_max_ttl_range, 0,
    &quot;TTL range to generate BlobDB data (in seconds). 0 means no TTL.&quot;);

DEFINE_uint64(blob_db_ttl_range_secs,
              ROCKSDB_NAMESPACE::blob_db::BlobDBOptions().ttl_range_secs,
              &quot;TTL bucket size to use when creating blob files.&quot;);

DEFINE_uint64(blob_db_min_blob_size,
              ROCKSDB_NAMESPACE::blob_db::BlobDBOptions().min_blob_size,
              &quot;Smallest blob to store in a file. Blobs smaller than this &quot;
              &quot;will be inlined with the key in the LSM tree.&quot;);

DEFINE_uint64(blob_db_bytes_per_sync,
              ROCKSDB_NAMESPACE::blob_db::BlobDBOptions().bytes_per_sync,
              &quot;Bytes to sync blob file at.&quot;);

DEFINE_uint64(blob_db_file_size,
              ROCKSDB_NAMESPACE::blob_db::BlobDBOptions().blob_file_size,
              &quot;Target size of each blob file.&quot;);

DEFINE_string(blob_db_compression_type, &quot;snappy&quot;,
              &quot;Algorithm to use to compress blob in blob file&quot;);
static enum ROCKSDB_NAMESPACE::CompressionType
    FLAGS_blob_db_compression_type_e = ROCKSDB_NAMESPACE::kSnappyCompression;

// Secondary DB instance Options
DEFINE_bool(use_secondary_db, false,
            &quot;Open a RocksDB secondary instance. A primary instance can be &quot;
            &quot;running in another db_bench process.&quot;);

DEFINE_string(secondary_path, &quot;&quot;,
              &quot;Path to a directory used by the secondary instance to store &quot;
              &quot;private files, e.g. info log.&quot;);

DEFINE_int32(secondary_update_interval, 5,
             &quot;Secondary instance attempts to catch up with the primary every &quot;
             &quot;secondary_update_interval seconds.&quot;);

#endif  // ROCKSDB_LITE

DEFINE_bool(report_bg_io_stats, false,
            &quot;Measure times spents on I/Os while in compactions. &quot;);

DEFINE_bool(use_stderr_info_logger, false,
            &quot;Write info logs to stderr instead of to LOG file. &quot;);

DEFINE_string(trace_file, &quot;&quot;, &quot;Trace workload to a file. &quot;);

DEFINE_int32(trace_replay_fast_forward, 1,
             &quot;Fast forward trace replay, must &gt;= 1. &quot;);
DEFINE_int32(block_cache_trace_sampling_frequency, 1,
             &quot;Block cache trace sampling frequency, termed s. It uses spatial &quot;
             &quot;downsampling and samples accesses to one out of s blocks.&quot;);
DEFINE_int64(
    block_cache_trace_max_trace_file_size_in_bytes,
    uint64_t{64} * 1024 * 1024 * 1024,
    &quot;The maximum block cache trace file size in bytes. Block cache accesses &quot;
    &quot;will not be logged if the trace file size exceeds this threshold. Default &quot;
    &quot;is 64 GB.&quot;);
DEFINE_string(block_cache_trace_file, &quot;&quot;, &quot;Block cache trace file path.&quot;);
DEFINE_int32(trace_replay_threads, 1,
             &quot;The number of threads to replay, must &gt;=1.&quot;);

static enum ROCKSDB_NAMESPACE::CompressionType StringToCompressionType(
    const char* ctype) {
  assert(ctype);

  if (!strcasecmp(ctype, &quot;none&quot;))
    return ROCKSDB_NAMESPACE::kNoCompression;
  else if (!strcasecmp(ctype, &quot;snappy&quot;))
    return ROCKSDB_NAMESPACE::kSnappyCompression;
  else if (!strcasecmp(ctype, &quot;zlib&quot;))
    return ROCKSDB_NAMESPACE::kZlibCompression;
  else if (!strcasecmp(ctype, &quot;bzip2&quot;))
    return ROCKSDB_NAMESPACE::kBZip2Compression;
  else if (!strcasecmp(ctype, &quot;lz4&quot;))
    return ROCKSDB_NAMESPACE::kLZ4Compression;
  else if (!strcasecmp(ctype, &quot;lz4hc&quot;))
    return ROCKSDB_NAMESPACE::kLZ4HCCompression;
  else if (!strcasecmp(ctype, &quot;xpress&quot;))
    return ROCKSDB_NAMESPACE::kXpressCompression;
  else if (!strcasecmp(ctype, &quot;zstd&quot;))
    return ROCKSDB_NAMESPACE::kZSTD;

  fprintf(stdout, &quot;Cannot parse compression type &#039;%s&#039;\n&quot;, ctype);
  return ROCKSDB_NAMESPACE::kSnappyCompression;  // default value
}

static std::string ColumnFamilyName(size_t i) {
  if (i == 0) {
    return ROCKSDB_NAMESPACE::kDefaultColumnFamilyName;
  } else {
    char name[100];
    snprintf(name, sizeof(name), &quot;column_family_name_%06zu&quot;, i);
    return std::string(name);
  }
}

DEFINE_string(compression_type, &quot;snappy&quot;,
              &quot;Algorithm to use to compress the database&quot;);
static enum ROCKSDB_NAMESPACE::CompressionType FLAGS_compression_type_e =
    ROCKSDB_NAMESPACE::kSnappyCompression;

DEFINE_int64(sample_for_compression, 0, &quot;Sample every N block for compression&quot;);

DEFINE_int32(compression_level, ROCKSDB_NAMESPACE::CompressionOptions().level,
             &quot;Compression level. The meaning of this value is library-&quot;
             &quot;dependent. If unset, we try to use the default for the library &quot;
             &quot;specified in `--compression_type`&quot;);

DEFINE_int32(compression_max_dict_bytes,
             ROCKSDB_NAMESPACE::CompressionOptions().max_dict_bytes,
             &quot;Maximum size of dictionary used to prime the compression &quot;
             &quot;library.&quot;);

DEFINE_int32(compression_zstd_max_train_bytes,
             ROCKSDB_NAMESPACE::CompressionOptions().zstd_max_train_bytes,
             &quot;Maximum size of training data passed to zstd&#039;s dictionary &quot;
             &quot;trainer.&quot;);

DEFINE_int32(min_level_to_compress, -1, &quot;If non-negative, compression starts&quot;
             &quot; from this level. Levels with number &lt; min_level_to_compress are&quot;
             &quot; not compressed. Otherwise, apply compression_type to &quot;
             &quot;all levels.&quot;);

static bool ValidateTableCacheNumshardbits(const char* flagname,
                                           int32_t value) {
  if (0 &gt;= value || value &gt; 20) {
    fprintf(stderr, &quot;Invalid value for --%s: %d, must be  0 &lt; val &lt;= 20\n&quot;,
            flagname, value);
    return false;
  }
  return true;
}
DEFINE_int32(table_cache_numshardbits, 4, &quot;&quot;);

#ifndef ROCKSDB_LITE
DEFINE_string(env_uri, &quot;&quot;, &quot;URI for registry Env lookup. Mutually exclusive&quot;
              &quot; with --hdfs.&quot;);
#endif  // ROCKSDB_LITE
DEFINE_string(hdfs, &quot;&quot;, &quot;Name of hdfs environment. Mutually exclusive with&quot;
              &quot; --env_uri.&quot;);

static std::shared_ptr&lt;ROCKSDB_NAMESPACE::Env&gt; env_guard;

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

DEFINE_int64(stats_interval, 0, &quot;Stats are reported every N operations when &quot;
             &quot;this is greater than zero. When 0 the interval grows over time.&quot;);

DEFINE_int64(stats_interval_seconds, 0, &quot;Report stats every N seconds. This &quot;
             &quot;overrides stats_interval when both are &gt; 0.&quot;);

DEFINE_int32(stats_per_interval, 0, &quot;Reports additional stats per interval when&quot;
             &quot; this is greater than 0.&quot;);

DEFINE_int64(report_interval_seconds, 0,
             &quot;If greater than zero, it will write simple stats in CVS format &quot;
             &quot;to --report_file every N seconds&quot;);

DEFINE_string(report_file, &quot;report.csv&quot;,
              &quot;Filename where some simple stats are reported to (if &quot;
              &quot;--report_interval_seconds is bigger than 0)&quot;);

DEFINE_int32(thread_status_per_interval, 0,
             &quot;Takes and report a snapshot of the current status of each thread&quot;
             &quot; when this is greater than 0.&quot;);

DEFINE_int32(perf_level, ROCKSDB_NAMESPACE::PerfLevel::kDisable,
             &quot;Level of perf collection&quot;);

static bool ValidateRateLimit(const char* flagname, double value) {
  const double EPSILON = 1e-10;
  if ( value &lt; -EPSILON ) {
    fprintf(stderr, &quot;Invalid value for --%s: %12.6f, must be &gt;= 0.0\n&quot;,
            flagname, value);
    return false;
  }
  return true;
}
DEFINE_double(soft_rate_limit, 0.0, &quot;DEPRECATED&quot;);

DEFINE_double(hard_rate_limit, 0.0, &quot;DEPRECATED&quot;);

DEFINE_uint64(soft_pending_compaction_bytes_limit, 64ull * 1024 * 1024 * 1024,
              &quot;Slowdown writes if pending compaction bytes exceed this number&quot;);

DEFINE_uint64(hard_pending_compaction_bytes_limit, 128ull * 1024 * 1024 * 1024,
              &quot;Stop writes if pending compaction bytes exceed this number&quot;);

DEFINE_uint64(delayed_write_rate, 8388608u,
              &quot;Limited bytes allowed to DB when soft_rate_limit or &quot;
              &quot;level0_slowdown_writes_trigger triggers&quot;);

DEFINE_bool(enable_pipelined_write, true,
            &quot;Allow WAL and memtable writes to be pipelined&quot;);

DEFINE_bool(unordered_write, false,
            &quot;Allow WAL and memtable writes to be pipelined&quot;);

DEFINE_bool(allow_concurrent_memtable_write, true,
            &quot;Allow multi-writers to update mem tables in parallel.&quot;);

DEFINE_bool(inplace_update_support,
            ROCKSDB_NAMESPACE::Options().inplace_update_support,
            &quot;Support in-place memtable update for smaller or same-size values&quot;);

DEFINE_uint64(inplace_update_num_locks,
              ROCKSDB_NAMESPACE::Options().inplace_update_num_locks,
              &quot;Number of RW locks to protect in-place memtable updates&quot;);

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

DEFINE_uint64(
    write_thread_max_yield_usec, 100,
    &quot;Maximum microseconds for enable_write_thread_adaptive_yield operation.&quot;);

DEFINE_uint64(write_thread_slow_yield_usec, 3,
              &quot;The threshold at which a slow yield is considered a signal that &quot;
              &quot;other processes or threads want the core.&quot;);

DEFINE_int32(rate_limit_delay_max_milliseconds, 1000,
             &quot;When hard_rate_limit is set then this is the max time a put will&quot;
             &quot; be stalled.&quot;);

DEFINE_uint64(rate_limiter_bytes_per_sec, 0, &quot;Set options.rate_limiter value.&quot;);

DEFINE_bool(rate_limiter_auto_tuned, false,
            &quot;Enable dynamic adjustment of rate limit according to demand for &quot;
            &quot;background I/O&quot;);


DEFINE_bool(sine_write_rate, false,
            &quot;Use a sine wave write_rate_limit&quot;);

DEFINE_uint64(sine_write_rate_interval_milliseconds, 10000,
              &quot;Interval of which the sine wave write_rate_limit is recalculated&quot;);

DEFINE_double(sine_a, 1,
             &quot;A in f(x) = A sin(bx + c) + d&quot;);

DEFINE_double(sine_b, 1,
             &quot;B in f(x) = A sin(bx + c) + d&quot;);

DEFINE_double(sine_c, 0,
             &quot;C in f(x) = A sin(bx + c) + d&quot;);

DEFINE_double(sine_d, 1,
             &quot;D in f(x) = A sin(bx + c) + d&quot;);

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

DEFINE_uint64(
    benchmark_write_rate_limit, 0,
    &quot;If non-zero, db_bench will rate-limit the writes going into RocksDB. This &quot;
    &quot;is the global rate in bytes/second.&quot;);

// the parameters of mix_graph
DEFINE_double(keyrange_dist_a, 0.0,
              &quot;The parameter &#039;a&#039; of prefix average access distribution &quot;
              &quot;f(x)=a*exp(b*x)+c*exp(d*x)&quot;);
DEFINE_double(keyrange_dist_b, 0.0,
              &quot;The parameter &#039;b&#039; of prefix average access distribution &quot;
              &quot;f(x)=a*exp(b*x)+c*exp(d*x)&quot;);
DEFINE_double(keyrange_dist_c, 0.0,
              &quot;The parameter &#039;c&#039; of prefix average access distribution&quot;
              &quot;f(x)=a*exp(b*x)+c*exp(d*x)&quot;);
DEFINE_double(keyrange_dist_d, 0.0,
              &quot;The parameter &#039;d&#039; of prefix average access distribution&quot;
              &quot;f(x)=a*exp(b*x)+c*exp(d*x)&quot;);
DEFINE_int64(keyrange_num, 1,
             &quot;The number of key ranges that are in the same prefix &quot;
             &quot;group, each prefix range will have its key acccess &quot;
             &quot;distribution&quot;);
DEFINE_double(key_dist_a, 0.0,
              &quot;The parameter &#039;a&#039; of key access distribution model &quot;
              &quot;f(x)=a*x^b&quot;);
DEFINE_double(key_dist_b, 0.0,
              &quot;The parameter &#039;b&#039; of key access distribution model &quot;
              &quot;f(x)=a*x^b&quot;);
DEFINE_double(value_theta, 0.0,
              &quot;The parameter &#039;theta&#039; of Generized Pareto Distribution &quot;
              &quot;f(x)=(1/sigma)*(1+k*(x-theta)/sigma)^-(1/k+1)&quot;);
DEFINE_double(value_k, 0.0,
              &quot;The parameter &#039;k&#039; of Generized Pareto Distribution &quot;
              &quot;f(x)=(1/sigma)*(1+k*(x-theta)/sigma)^-(1/k+1)&quot;);
DEFINE_double(value_sigma, 0.0,
              &quot;The parameter &#039;theta&#039; of Generized Pareto Distribution &quot;
              &quot;f(x)=(1/sigma)*(1+k*(x-theta)/sigma)^-(1/k+1)&quot;);
DEFINE_double(iter_theta, 0.0,
              &quot;The parameter &#039;theta&#039; of Generized Pareto Distribution &quot;
              &quot;f(x)=(1/sigma)*(1+k*(x-theta)/sigma)^-(1/k+1)&quot;);
DEFINE_double(iter_k, 0.0,
              &quot;The parameter &#039;k&#039; of Generized Pareto Distribution &quot;
              &quot;f(x)=(1/sigma)*(1+k*(x-theta)/sigma)^-(1/k+1)&quot;);
DEFINE_double(iter_sigma, 0.0,
              &quot;The parameter &#039;sigma&#039; of Generized Pareto Distribution &quot;
              &quot;f(x)=(1/sigma)*(1+k*(x-theta)/sigma)^-(1/k+1)&quot;);
DEFINE_double(mix_get_ratio, 1.0,
              &quot;The ratio of Get queries of mix_graph workload&quot;);
DEFINE_double(mix_put_ratio, 0.0,
              &quot;The ratio of Put queries of mix_graph workload&quot;);
DEFINE_double(mix_seek_ratio, 0.0,
              &quot;The ratio of Seek queries of mix_graph workload&quot;);
DEFINE_int64(mix_max_scan_len, 10000, &quot;The max scan length of Iterator&quot;);
DEFINE_int64(mix_ave_kv_size, 512,
             &quot;The average key-value size of this workload&quot;);
DEFINE_int64(mix_max_value_size, 1024, &quot;The max value size of this workload&quot;);
DEFINE_double(
    sine_mix_rate_noise, 0.0,
    &quot;Add the noise ratio to the sine rate, it is between 0.0 and 1.0&quot;);
DEFINE_bool(sine_mix_rate, false,
            &quot;Enable the sine QPS control on the mix workload&quot;);
DEFINE_uint64(
    sine_mix_rate_interval_milliseconds, 10000,
    &quot;Interval of which the sine wave read_rate_limit is recalculated&quot;);
DEFINE_int64(mix_accesses, -1,
             &quot;The total query accesses of mix_graph workload&quot;);

DEFINE_uint64(
    benchmark_read_rate_limit, 0,
    &quot;If non-zero, db_bench will rate-limit the reads from RocksDB. This &quot;
    &quot;is the global rate in ops/second.&quot;);

DEFINE_uint64(max_compaction_bytes,
              ROCKSDB_NAMESPACE::Options().max_compaction_bytes,
              &quot;Max bytes allowed in one compaction&quot;);

#ifndef ROCKSDB_LITE
DEFINE_bool(readonly, false, &quot;Run read only benchmarks.&quot;);

DEFINE_bool(print_malloc_stats, false,
            &quot;Print malloc stats to stdout after benchmarks finish.&quot;);
#endif  // ROCKSDB_LITE

DEFINE_bool(disable_auto_compactions, false, &quot;Do not auto trigger compactions&quot;);

DEFINE_uint64(wal_ttl_seconds, 0, &quot;Set the TTL for the WAL Files in seconds.&quot;);
DEFINE_uint64(wal_size_limit_MB, 0, &quot;Set the size limit for the WAL Files&quot;
              &quot; in MB.&quot;);
DEFINE_uint64(max_total_wal_size, 0, &quot;Set total max WAL size&quot;);

DEFINE_bool(mmap_read, ROCKSDB_NAMESPACE::Options().allow_mmap_reads,
            &quot;Allow reads to occur via mmap-ing files&quot;);

DEFINE_bool(mmap_write, ROCKSDB_NAMESPACE::Options().allow_mmap_writes,
            &quot;Allow writes to occur via mmap-ing files&quot;);

DEFINE_bool(use_direct_reads, ROCKSDB_NAMESPACE::Options().use_direct_reads,
            &quot;Use O_DIRECT for reading data&quot;);

DEFINE_bool(use_direct_io_for_flush_and_compaction,
            ROCKSDB_NAMESPACE::Options().use_direct_io_for_flush_and_compaction,
            &quot;Use O_DIRECT for background flush and compaction writes&quot;);

DEFINE_bool(advise_random_on_open,
            ROCKSDB_NAMESPACE::Options().advise_random_on_open,
            &quot;Advise random access on table file open&quot;);

DEFINE_string(compaction_fadvice, &quot;NORMAL&quot;,
              &quot;Access pattern advice when a file is compacted&quot;);
static auto FLAGS_compaction_fadvice_e =
    ROCKSDB_NAMESPACE::Options().access_hint_on_compaction_start;

DEFINE_bool(use_tailing_iterator, false,
            &quot;Use tailing iterator to access a series of keys instead of get&quot;);

DEFINE_bool(use_adaptive_mutex, ROCKSDB_NAMESPACE::Options().use_adaptive_mutex,
            &quot;Use adaptive mutex&quot;);

DEFINE_uint64(bytes_per_sync, ROCKSDB_NAMESPACE::Options().bytes_per_sync,
              &quot;Allows OS to incrementally sync SST files to disk while they are&quot;
              &quot; being written, in the background. Issue one request for every&quot;
              &quot; bytes_per_sync written. 0 turns it off.&quot;);

DEFINE_uint64(wal_bytes_per_sync,
              ROCKSDB_NAMESPACE::Options().wal_bytes_per_sync,
              &quot;Allows OS to incrementally sync WAL files to disk while they are&quot;
              &quot; being written, in the background. Issue one request for every&quot;
              &quot; wal_bytes_per_sync written. 0 turns it off.&quot;);

DEFINE_bool(use_single_deletes, true,
            &quot;Use single deletes (used in RandomReplaceKeys only).&quot;);

DEFINE_double(stddev, 2000.0,
              &quot;Standard deviation of normal distribution used for picking keys&quot;
              &quot; (used in RandomReplaceKeys only).&quot;);

DEFINE_int32(key_id_range, 100000,
             &quot;Range of possible value of key id (used in TimeSeries only).&quot;);

DEFINE_string(expire_style, &quot;none&quot;,
              &quot;Style to remove expired time entries. Can be one of the options &quot;
              &quot;below: none (do not expired data), compaction_filter (use a &quot;
              &quot;compaction filter to remove expired data), delete (seek IDs and &quot;
              &quot;remove expired data) (used in TimeSeries only).&quot;);

DEFINE_uint64(
    time_range, 100000,
    &quot;Range of timestamp that store in the database (used in TimeSeries&quot;
    &quot; only).&quot;);

DEFINE_int32(num_deletion_threads, 1,
             &quot;Number of threads to do deletion (used in TimeSeries and delete &quot;
             &quot;expire_style only).&quot;);

DEFINE_int32(max_successive_merges, 0, &quot;Maximum number of successive merge&quot;
             &quot; operations on a key in the memtable&quot;);

static bool ValidatePrefixSize(const char* flagname, int32_t value) {
  if (value &lt; 0 || value&gt;=2000000000) {
    fprintf(stderr, &quot;Invalid value for --%s: %d. 0&lt;= PrefixSize &lt;=2000000000\n&quot;,
            flagname, value);
    return false;
  }
  return true;
}

DEFINE_int32(prefix_size, 0, &quot;control the prefix size for HashSkipList and &quot;
             &quot;plain table&quot;);
DEFINE_int64(keys_per_prefix, 0, &quot;control average number of keys generated &quot;
             &quot;per prefix, 0 means no special handling of the prefix, &quot;
             &quot;i.e. use the prefix comes with the generated random number.&quot;);
DEFINE_bool(total_order_seek, false,
            &quot;Enable total order seek regardless of index format.&quot;);
DEFINE_bool(prefix_same_as_start, false,
            &quot;Enforce iterator to return keys with prefix same as seek key.&quot;);
DEFINE_bool(
    seek_missing_prefix, false,
    &quot;Iterator seek to keys with non-exist prefixes. Require prefix_size &gt; 8&quot;);

DEFINE_int32(memtable_insert_with_hint_prefix_size, 0,
             &quot;If non-zero, enable &quot;
             &quot;memtable insert with hint with the given prefix size.&quot;);
DEFINE_bool(enable_io_prio, false, &quot;Lower the background flush/compaction &quot;
            &quot;threads&#039; IO priority&quot;);
DEFINE_bool(enable_cpu_prio, false, &quot;Lower the background flush/compaction &quot;
            &quot;threads&#039; CPU priority&quot;);
DEFINE_bool(identity_as_first_hash, false, &quot;the first hash function of cuckoo &quot;
            &quot;table becomes an identity function. This is only valid when key &quot;
            &quot;is 8 bytes&quot;);
DEFINE_bool(dump_malloc_stats, true, &quot;Dump malloc stats in LOG &quot;);
DEFINE_uint64(stats_dump_period_sec,
              ROCKSDB_NAMESPACE::Options().stats_dump_period_sec,
              &quot;Gap between printing stats to log in seconds&quot;);
DEFINE_uint64(stats_persist_period_sec,
              ROCKSDB_NAMESPACE::Options().stats_persist_period_sec,
              &quot;Gap between persisting stats in seconds&quot;);
DEFINE_bool(persist_stats_to_disk,
            ROCKSDB_NAMESPACE::Options().persist_stats_to_disk,
            &quot;whether to persist stats to disk&quot;);
DEFINE_uint64(stats_history_buffer_size,
              ROCKSDB_NAMESPACE::Options().stats_history_buffer_size,
              &quot;Max number of stats snapshots to keep in memory&quot;);
DEFINE_int64(multiread_stride, 0,
             &quot;Stride length for the keys in a MultiGet batch&quot;);
DEFINE_bool(multiread_batched, false, &quot;Use the new MultiGet API&quot;);

enum RepFactory {
  kSkipList,
  kPrefixHash,
  kVectorRep,
  kHashLinkedList,
};

static enum RepFactory StringToRepFactory(const char* ctype) {
  assert(ctype);

  if (!strcasecmp(ctype, &quot;skip_list&quot;))
    return kSkipList;
  else if (!strcasecmp(ctype, &quot;prefix_hash&quot;))
    return kPrefixHash;
  else if (!strcasecmp(ctype, &quot;vector&quot;))
    return kVectorRep;
  else if (!strcasecmp(ctype, &quot;hash_linkedlist&quot;))
    return kHashLinkedList;

  fprintf(stdout, &quot;Cannot parse memreptable %s\n&quot;, ctype);
  return kSkipList;
}

static enum RepFactory FLAGS_rep_factory;
DEFINE_string(memtablerep, &quot;skip_list&quot;, &quot;&quot;);
DEFINE_int64(hash_bucket_count, 1024 * 1024, &quot;hash bucket count&quot;);
DEFINE_bool(use_plain_table, false, &quot;if use plain table &quot;
            &quot;instead of block-based table format&quot;);
DEFINE_bool(use_cuckoo_table, false, &quot;if use cuckoo table format&quot;);
DEFINE_double(cuckoo_hash_ratio, 0.9, &quot;Hash ratio for Cuckoo SST table.&quot;);
DEFINE_bool(use_hash_search, false, &quot;if use kHashSearch &quot;
            &quot;instead of kBinarySearch. &quot;
            &quot;This is valid if only we use BlockTable&quot;);
DEFINE_bool(use_block_based_filter, false, &quot;if use kBlockBasedFilter &quot;
            &quot;instead of kFullFilter for filter block. &quot;
            &quot;This is valid if only we use BlockTable&quot;);
DEFINE_string(merge_operator, &quot;&quot;, &quot;The merge operator to use with the database.&quot;
              &quot;If a new merge operator is specified, be sure to use fresh&quot;
              &quot; database The possible merge operators are defined in&quot;
              &quot; utilities/merge_operators.h&quot;);
DEFINE_int32(skip_list_lookahead, 0, &quot;Used with skip_list memtablerep; try &quot;
             &quot;linear search first for this many steps from the previous &quot;
             &quot;position&quot;);
DEFINE_bool(report_file_operations, false, &quot;if report number of file &quot;
            &quot;operations&quot;);
DEFINE_int32(readahead_size, 0, &quot;Iterator readahead size&quot;);

static const bool FLAGS_soft_rate_limit_dummy __attribute__((__unused__)) =
    RegisterFlagValidator(&amp;FLAGS_soft_rate_limit, &amp;ValidateRateLimit);

static const bool FLAGS_hard_rate_limit_dummy __attribute__((__unused__)) =
    RegisterFlagValidator(&amp;FLAGS_hard_rate_limit, &amp;ValidateRateLimit);

static const bool FLAGS_prefix_size_dummy __attribute__((__unused__)) =
    RegisterFlagValidator(&amp;FLAGS_prefix_size, &amp;ValidatePrefixSize);

static const bool FLAGS_key_size_dummy __attribute__((__unused__)) =
    RegisterFlagValidator(&amp;FLAGS_key_size, &amp;ValidateKeySize);

static const bool FLAGS_cache_numshardbits_dummy __attribute__((__unused__)) =
    RegisterFlagValidator(&amp;FLAGS_cache_numshardbits,
                          &amp;ValidateCacheNumshardbits);

static const bool FLAGS_readwritepercent_dummy __attribute__((__unused__)) =
    RegisterFlagValidator(&amp;FLAGS_readwritepercent, &amp;ValidateInt32Percent);

DEFINE_int32(disable_seek_compaction, false,
             &quot;Not used, left here for backwards compatibility&quot;);

static const bool FLAGS_deletepercent_dummy __attribute__((__unused__)) =
    RegisterFlagValidator(&amp;FLAGS_deletepercent, &amp;ValidateInt32Percent);
static const bool FLAGS_table_cache_numshardbits_dummy __attribute__((__unused__)) =
    RegisterFlagValidator(&amp;FLAGS_table_cache_numshardbits,
                          &amp;ValidateTableCacheNumshardbits);

namespace ROCKSDB_NAMESPACE {

namespace {
struct ReportFileOpCounters {
  std::atomic&lt;int&gt; open_counter_;
  std::atomic&lt;int&gt; read_counter_;
  std::atomic&lt;int&gt; append_counter_;
  std::atomic&lt;uint64_t&gt; bytes_read_;
  std::atomic&lt;uint64_t&gt; bytes_written_;
};

// A special Env to records and report file operations in db_bench
class ReportFileOpEnv : public EnvWrapper {
 public:
  explicit ReportFileOpEnv(Env* base) : EnvWrapper(base) { reset(); }

  void reset() {
    counters_.open_counter_ = 0;
    counters_.read_counter_ = 0;
    counters_.append_counter_ = 0;
    counters_.bytes_read_ = 0;
    counters_.bytes_written_ = 0;
  }

  Status NewSequentialFile(const std::string&amp; f,
                           std::unique_ptr&lt;SequentialFile&gt;* r,
                           const EnvOptions&amp; soptions) override {
    class CountingFile : public SequentialFile {
     private:
      std::unique_ptr&lt;SequentialFile&gt; target_;
      ReportFileOpCounters* counters_;

     public:
      CountingFile(std::unique_ptr&lt;SequentialFile&gt;&amp;&amp; target,
                   ReportFileOpCounters* counters)
          : target_(std::move(target)), counters_(counters) {}

      Status Read(size_t n, Slice* result, char* scratch) override {
        counters_-&gt;read_counter_.fetch_add(1, std::memory_order_relaxed);
        Status rv = target_-&gt;Read(n, result, scratch);
        counters_-&gt;bytes_read_.fetch_add(result-&gt;size(),
                                         std::memory_order_relaxed);
        return rv;
      }

      Status Skip(uint64_t n) override { return target_-&gt;Skip(n); }
    };

    Status s = target()-&gt;NewSequentialFile(f, r, soptions);
    if (s.ok()) {
      counters()-&gt;open_counter_.fetch_add(1, std::memory_order_relaxed);
      r-&gt;reset(new CountingFile(std::move(*r), counters()));
    }
    return s;
  }

  Status NewRandomAccessFile(const std::string&amp; f,
                             std::unique_ptr&lt;RandomAccessFile&gt;* r,
                             const EnvOptions&amp; soptions) override {
    class CountingFile : public RandomAccessFile {
     private:
      std::unique_ptr&lt;RandomAccessFile&gt; target_;
      ReportFileOpCounters* counters_;

     public:
      CountingFile(std::unique_ptr&lt;RandomAccessFile&gt;&amp;&amp; target,
                   ReportFileOpCounters* counters)
          : target_(std::move(target)), counters_(counters) {}
      Status Read(uint64_t offset, size_t n, Slice* result,
                  char* scratch) const override {
        counters_-&gt;read_counter_.fetch_add(1, std::memory_order_relaxed);
        Status rv = target_-&gt;Read(offset, n, result, scratch);
        counters_-&gt;bytes_read_.fetch_add(result-&gt;size(),
                                         std::memory_order_relaxed);
        return rv;
      }
    };

    Status s = target()-&gt;NewRandomAccessFile(f, r, soptions);
    if (s.ok()) {
      counters()-&gt;open_counter_.fetch_add(1, std::memory_order_relaxed);
      r-&gt;reset(new CountingFile(std::move(*r), counters()));
    }
    return s;
  }

  Status NewWritableFile(const std::string&amp; f, std::unique_ptr&lt;WritableFile&gt;* r,
                         const EnvOptions&amp; soptions) override {
    class CountingFile : public WritableFile {
     private:
      std::unique_ptr&lt;WritableFile&gt; target_;
      ReportFileOpCounters* counters_;

     public:
      CountingFile(std::unique_ptr&lt;WritableFile&gt;&amp;&amp; target,
                   ReportFileOpCounters* counters)
          : target_(std::move(target)), counters_(counters) {}

      Status Append(const Slice&amp; data) override {
        counters_-&gt;append_counter_.fetch_add(1, std::memory_order_relaxed);
        Status rv = target_-&gt;Append(data);
        counters_-&gt;bytes_written_.fetch_add(data.size(),
                                            std::memory_order_relaxed);
        return rv;
      }

      Status Truncate(uint64_t size) override { return target_-&gt;Truncate(size); }
      Status Close() override { return target_-&gt;Close(); }
      Status Flush() override { return target_-&gt;Flush(); }
      Status Sync() override { return target_-&gt;Sync(); }
    };

    Status s = target()-&gt;NewWritableFile(f, r, soptions);
    if (s.ok()) {
      counters()-&gt;open_counter_.fetch_add(1, std::memory_order_relaxed);
      r-&gt;reset(new CountingFile(std::move(*r), counters()));
    }
    return s;
  }

  // getter
  ReportFileOpCounters* counters() { return &amp;counters_; }

 private:
  ReportFileOpCounters counters_;
};

}  // namespace

enum DistributionType : unsigned char {
  kFixed = 0,
  kUniform,
  kNormal
};

static enum DistributionType FLAGS_value_size_distribution_type_e = kFixed;

static enum DistributionType StringToDistributionType(const char* ctype) {
  assert(ctype);

  if (!strcasecmp(ctype, &quot;fixed&quot;))
    return kFixed;
  else if (!strcasecmp(ctype, &quot;uniform&quot;))
    return kUniform;
  else if (!strcasecmp(ctype, &quot;normal&quot;))
    return kNormal;

  fprintf(stdout, &quot;Cannot parse distribution type &#039;%s&#039;\n&quot;, ctype);
  return kFixed;  // default value
}

class BaseDistribution {
 public:
  BaseDistribution(unsigned int min, unsigned int max) :
    min_value_size_(min),
    max_value_size_(max) {}
  virtual ~BaseDistribution() {}

  unsigned int Generate() {
    auto val = Get();
    if (NeedTruncate()) {
      val = std::max(min_value_size_, val);
      val = std::min(max_value_size_, val);
    }
    return val;
  }
 private:
  virtual unsigned int Get() = 0;
  virtual bool NeedTruncate() {
    return true;
  }
  unsigned int min_value_size_;
  unsigned int max_value_size_;
};

class FixedDistribution : public BaseDistribution
{
 public:
  FixedDistribution(unsigned int size) :
    BaseDistribution(size, size),
    size_(size) {}
 private:
  virtual unsigned int Get() override {
    return size_;
  }
  virtual bool NeedTruncate() override {
    return false;
  }
  unsigned int size_;
};

class NormalDistribution
    : public BaseDistribution, public std::normal_distribution&lt;double&gt; {
 public:
  NormalDistribution(unsigned int min, unsigned int max) :
    BaseDistribution(min, max),
    // 99.7% values within the range [min, max].
    std::normal_distribution&lt;double&gt;((double)(min + max) / 2.0 /*mean*/,
                                     (double)(max - min) / 6.0 /*stddev*/),
    gen_(rd_()) {}
 private:
  virtual unsigned int Get() override {
    return static_cast&lt;unsigned int&gt;((*this)(gen_));
  }
  std::random_device rd_;
  std::mt19937 gen_;
};

class UniformDistribution
    : public BaseDistribution,
      public std::uniform_int_distribution&lt;unsigned int&gt; {
 public:
  UniformDistribution(unsigned int min, unsigned int max) :
    BaseDistribution(min, max),
    std::uniform_int_distribution&lt;unsigned int&gt;(min, max),
    gen_(rd_()) {}
 private:
  virtual unsigned int Get() override {
    return (*this)(gen_);
  }
  virtual bool NeedTruncate() override {
    return false;
  }
  std::random_device rd_;
  std::mt19937 gen_;
};

// Helper for quickly generating random data.
class RandomGenerator {
 private:
  std::string data_;
  unsigned int pos_;
  std::unique_ptr&lt;BaseDistribution&gt; dist_;

 public:

  RandomGenerator() {
    auto max_value_size = FLAGS_value_size_max;
    switch (FLAGS_value_size_distribution_type_e) {
      case kUniform:
        dist_.reset(new UniformDistribution(FLAGS_value_size_min,
                                            FLAGS_value_size_max));
        break;
      case kNormal:
        dist_.reset(new NormalDistribution(FLAGS_value_size_min,
                                           FLAGS_value_size_max));
        break;
      case kFixed:
      default:
        dist_.reset(new FixedDistribution(value_size));
        max_value_size = value_size;
    }
    // We use a limited amount of data over and over again and ensure
    // that it is larger than the compression window (32KB), and also
    // large enough to serve all typical value sizes we want to write.
    Random rnd(301);
    std::string piece;
    while (data_.size() &lt; (unsigned)std::max(1048576, max_value_size)) {
      // Add a short fragment that is as compressible as specified
      // by FLAGS_compression_ratio.
      test::CompressibleString(&amp;rnd, FLAGS_compression_ratio, 100, &amp;piece);
      data_.append(piece);
    }
    pos_ = 0;
  }

  Slice Generate(unsigned int len) {
    assert(len &lt;= data_.size());
    if (pos_ + len &gt; data_.size()) {
      pos_ = 0;
    }
    pos_ += len;
    return Slice(data_.data() + pos_ - len, len);
  }

  Slice Generate() {
    auto len = dist_-&gt;Generate();
    return Generate(len);
  }
};

static void AppendWithSpace(std::string* str, Slice msg) {
  if (msg.empty()) return;
  if (!str-&gt;empty()) {
    str-&gt;push_back(&#039; &#039;);
  }
  str-&gt;append(msg.data(), msg.size());
}

struct DBWithColumnFamilies {
  std::vector&lt;ColumnFamilyHandle*&gt; cfh;
  DB* db;
#ifndef ROCKSDB_LITE
  OptimisticTransactionDB* opt_txn_db;
#endif  // ROCKSDB_LITE
  std::atomic&lt;size_t&gt; num_created;  // Need to be updated after all the
                                    // new entries in cfh are set.
  size_t num_hot;  // Number of column families to be queried at each moment.
                   // After each CreateNewCf(), another num_hot number of new
                   // Column families will be created and used to be queried.
  port::Mutex create_cf_mutex;  // Only one thread can execute CreateNewCf()
  std::vector&lt;int&gt; cfh_idx_to_prob;  // ith index holds probability of operating
                                     // on cfh[i].

  DBWithColumnFamilies()
      : db(nullptr)
#ifndef ROCKSDB_LITE
        , opt_txn_db(nullptr)
#endif  // ROCKSDB_LITE
  {
    cfh.clear();
    num_created = 0;
    num_hot = 0;
  }

  DBWithColumnFamilies(const DBWithColumnFamilies&amp; other)
      : cfh(other.cfh),
        db(other.db),
#ifndef ROCKSDB_LITE
        opt_txn_db(other.opt_txn_db),
#endif  // ROCKSDB_LITE
        num_created(other.num_created.load()),
        num_hot(other.num_hot),
        cfh_idx_to_prob(other.cfh_idx_to_prob) {
  }

  void DeleteDBs() {
    std::for_each(cfh.begin(), cfh.end(),
                  [](ColumnFamilyHandle* cfhi) { delete cfhi; });
    cfh.clear();
#ifndef ROCKSDB_LITE
    if (opt_txn_db) {
      delete opt_txn_db;
      opt_txn_db = nullptr;
    } else {
      delete db;
      db = nullptr;
    }
#else
    delete db;
    db = nullptr;
#endif  // ROCKSDB_LITE
  }

  ColumnFamilyHandle* GetCfh(int64_t rand_num) {
    assert(num_hot &gt; 0);
    size_t rand_offset = 0;
    if (!cfh_idx_to_prob.empty()) {
      assert(cfh_idx_to_prob.size() == num_hot);
      int sum = 0;
      while (sum + cfh_idx_to_prob[rand_offset] &lt; rand_num % 100) {
        sum += cfh_idx_to_prob[rand_offset];
        ++rand_offset;
      }
      assert(rand_offset &lt; cfh_idx_to_prob.size());
    } else {
      rand_offset = rand_num % num_hot;
    }
    return cfh[num_created.load(std::memory_order_acquire) - num_hot +
               rand_offset];
  }

  // stage: assume CF from 0 to stage * num_hot has be created. Need to create
  //        stage * num_hot + 1 to stage * (num_hot + 1).
  void CreateNewCf(ColumnFamilyOptions options, int64_t stage) {
    MutexLock l(&amp;create_cf_mutex);
    if ((stage + 1) * num_hot &lt;= num_created) {
      // Already created.
      return;
    }
    auto new_num_created = num_created + num_hot;
    assert(new_num_created &lt;= cfh.size());
    for (size_t i = num_created; i &lt; new_num_created; i++) {
      Status s =
          db-&gt;CreateColumnFamily(options, ColumnFamilyName(i), &amp;(cfh[i]));
      if (!s.ok()) {
        fprintf(stderr, &quot;create column family error: %s\n&quot;,
                s.ToString().c_str());
        abort();
      }
    }
    num_created.store(new_num_created, std::memory_order_release);
  }
};

// a class that reports stats to CSV file
class ReporterAgent {
 public:
  ReporterAgent(Env* env, const std::string&amp; fname,
                uint64_t report_interval_secs)
      : env_(env),
        total_ops_done_(0),
        last_report_(0),
        report_interval_secs_(report_interval_secs),
        stop_(false) {
    auto s = env_-&gt;NewWritableFile(fname, &amp;report_file_, EnvOptions());
    if (s.ok()) {
      s = report_file_-&gt;Append(Header() + &quot;\n&quot;);
    }
    if (s.ok()) {
      s = report_file_-&gt;Flush();
    }
    if (!s.ok()) {
      fprintf(stderr, &quot;Can&#039;t open %s: %s\n&quot;, fname.c_str(),
              s.ToString().c_str());
      abort();
    }

    reporting_thread_ = port::Thread([&amp;]() { SleepAndReport(); });
  }

  ~ReporterAgent() {
    {
      std::unique_lock&lt;std::mutex&gt; lk(mutex_);
      stop_ = true;
      stop_cv_.notify_all();
    }
    reporting_thread_.join();
  }

  // thread safe
  void ReportFinishedOps(int64_t num_ops) {
    total_ops_done_.fetch_add(num_ops);
  }

 private:
  std::string Header() const { return &quot;secs_elapsed,interval_qps&quot;; }
  void SleepAndReport() {
    auto time_started = env_-&gt;NowMicros();
    while (true) {
      {
        std::unique_lock&lt;std::mutex&gt; lk(mutex_);
        if (stop_ ||
            stop_cv_.wait_for(lk, std::chrono::seconds(report_interval_secs_),
                              [&amp;]() { return stop_; })) {
          // stopping
          break;
        }
        // else -&gt; timeout, which means time for a report!
      }
      auto total_ops_done_snapshot = total_ops_done_.load();
      // round the seconds elapsed
      auto secs_elapsed =
          (env_-&gt;NowMicros() - time_started + kMicrosInSecond / 2) /
          kMicrosInSecond;
      std::string report = ToString(secs_elapsed) + &quot;,&quot; +
                           ToString(total_ops_done_snapshot - last_report_) +
                           &quot;\n&quot;;
      auto s = report_file_-&gt;Append(report);
      if (s.ok()) {
        s = report_file_-&gt;Flush();
      }
      if (!s.ok()) {
        fprintf(stderr,
                &quot;Can&#039;t write to report file (%s), stopping the reporting\n&quot;,
                s.ToString().c_str());
        break;
      }
      last_report_ = total_ops_done_snapshot;
    }
  }

  Env* env_;
  std::unique_ptr&lt;WritableFile&gt; report_file_;
  std::atomic&lt;int64_t&gt; total_ops_done_;
  int64_t last_report_;
  const uint64_t report_interval_secs_;
  ROCKSDB_NAMESPACE::port::Thread reporting_thread_;
  std::mutex mutex_;
  // will notify on stop
  std::condition_variable stop_cv_;
  bool stop_;
};

enum OperationType : unsigned char {
  kRead = 0,
  kWrite,
  kDelete,
  kSeek,
  kMerge,
  kUpdate,
  kCompress,
  kUncompress,
  kCrc,
  kHash,
  kOthers
};

static std::unordered_map&lt;OperationType, std::string, std::hash&lt;unsigned char&gt;&gt;
                          OperationTypeString = {
  {kRead, &quot;read&quot;},
  {kWrite, &quot;write&quot;},
  {kDelete, &quot;delete&quot;},
  {kSeek, &quot;seek&quot;},
  {kMerge, &quot;merge&quot;},
  {kUpdate, &quot;update&quot;},
  {kCompress, &quot;compress&quot;},
  {kCompress, &quot;uncompress&quot;},
  {kCrc, &quot;crc&quot;},
  {kHash, &quot;hash&quot;},
  {kOthers, &quot;op&quot;}
};

class CombinedStats;
class Stats {
 private:
  int id_;
  uint64_t start_;
  uint64_t sine_interval_;
  uint64_t finish_;
  double seconds_;
  uint64_t done_;
  uint64_t last_report_done_;
  uint64_t next_report_;
  uint64_t bytes_;
  uint64_t last_op_finish_;
  uint64_t last_report_finish_;
  std::unordered_map&lt;OperationType, std::shared_ptr&lt;HistogramImpl&gt;,
                     std::hash&lt;unsigned char&gt;&gt; hist_;
  std::string message_;
  bool exclude_from_merge_;
  ReporterAgent* reporter_agent_;  // does not own
  friend class CombinedStats;

 public:
  Stats() { Start(-1); }

  void SetReporterAgent(ReporterAgent* reporter_agent) {
    reporter_agent_ = reporter_agent;
  }

  void Start(int id) {
    id_ = id;
    next_report_ = FLAGS_stats_interval ? FLAGS_stats_interval : 100;
    last_op_finish_ = start_;
    hist_.clear();
    done_ = 0;
    last_report_done_ = 0;
    bytes_ = 0;
    seconds_ = 0;
    start_ = FLAGS_env-&gt;NowMicros();
    sine_interval_ = FLAGS_env-&gt;NowMicros();
    finish_ = start_;
    last_report_finish_ = start_;
    message_.clear();
    // When set, stats from this thread won&#039;t be merged with others.
    exclude_from_merge_ = false;
  }

  void Merge(const Stats&amp; other) {
    if (other.exclude_from_merge_)
      return;

    for (auto it = other.hist_.begin(); it != other.hist_.end(); ++it) {
      auto this_it = hist_.find(it-&gt;first);
      if (this_it != hist_.end()) {
        this_it-&gt;second-&gt;Merge(*(other.hist_.at(it-&gt;first)));
      } else {
        hist_.insert({ it-&gt;first, it-&gt;second });
      }
    }

    done_ += other.done_;
    bytes_ += other.bytes_;
    seconds_ += other.seconds_;
    if (other.start_ &lt; start_) start_ = other.start_;
    if (other.finish_ &gt; finish_) finish_ = other.finish_;

    // Just keep the messages from one thread
    if (message_.empty()) message_ = other.message_;
  }

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

  void AddMessage(Slice msg) {
    AppendWithSpace(&amp;message_, msg);
  }

  void SetId(int id) { id_ = id; }
  void SetExcludeFromMerge() { exclude_from_merge_ = true; }

  void PrintThreadStatus() {
    std::vector&lt;ThreadStatus&gt; thread_list;
    FLAGS_env-&gt;GetThreadList(&amp;thread_list);

    fprintf(stderr, &quot;\n%18s %10s %12s %20s %13s %45s %12s %s\n&quot;,
        &quot;ThreadID&quot;, &quot;ThreadType&quot;, &quot;cfName&quot;, &quot;Operation&quot;,
        &quot;ElapsedTime&quot;, &quot;Stage&quot;, &quot;State&quot;, &quot;OperationProperties&quot;);

    int64_t current_time = 0;
    FLAGS_env-&gt;GetCurrentTime(&amp;current_time);
    for (auto ts : thread_list) {
      fprintf(stderr, &quot;%18&quot; PRIu64 &quot; %10s %12s %20s %13s %45s %12s&quot;,
          ts.thread_id,
          ThreadStatus::GetThreadTypeName(ts.thread_type).c_str(),
          ts.cf_name.c_str(),
          ThreadStatus::GetOperationName(ts.operation_type).c_str(),
          ThreadStatus::MicrosToString(ts.op_elapsed_micros).c_str(),
          ThreadStatus::GetOperationStageName(ts.operation_stage).c_str(),
          ThreadStatus::GetStateName(ts.state_type).c_str());

      auto op_properties = ThreadStatus::InterpretOperationProperties(
          ts.operation_type, ts.op_properties);
      for (const auto&amp; op_prop : op_properties) {
        fprintf(stderr, &quot; %s %&quot; PRIu64&quot; |&quot;,
            op_prop.first.c_str(), op_prop.second);
      }
      fprintf(stderr, &quot;\n&quot;);
    }
  }

  void ResetSineInterval() {
    sine_interval_ = FLAGS_env-&gt;NowMicros();
  }

  uint64_t GetSineInterval() {
    return sine_interval_;
  }

  uint64_t GetStart() {
    return start_;
  }

  void ResetLastOpTime() {
    // Set to now to avoid latency from calls to SleepForMicroseconds
    last_op_finish_ = FLAGS_env-&gt;NowMicros();
  }

  void FinishedOps(DBWithColumnFamilies* db_with_cfh, DB* db, int64_t num_ops,
                   enum OperationType op_type = kOthers) {
    if (reporter_agent_) {
      reporter_agent_-&gt;ReportFinishedOps(num_ops);
    }
    if (FLAGS_histogram) {
      uint64_t now = FLAGS_env-&gt;NowMicros();
      uint64_t micros = now - last_op_finish_;

      if (hist_.find(op_type) == hist_.end())
      {
        auto hist_temp = std::make_shared&lt;HistogramImpl&gt;();
        hist_.insert({op_type, std::move(hist_temp)});
      }
      hist_[op_type]-&gt;Add(micros);

      if (micros &gt; 20000 &amp;&amp; !FLAGS_stats_interval) {
        fprintf(stderr, &quot;long op: %&quot; PRIu64 &quot; micros%30s\r&quot;, micros, &quot;&quot;);
        fflush(stderr);
      }
      last_op_finish_ = now;
    }

    done_ += num_ops;
    if (done_ &gt;= next_report_) {
      if (!FLAGS_stats_interval) {
        if      (next_report_ &lt; 1000)   next_report_ += 100;
        else if (next_report_ &lt; 5000)   next_report_ += 500;
        else if (next_report_ &lt; 10000)  next_report_ += 1000;
        else if (next_report_ &lt; 50000)  next_report_ += 5000;
        else if (next_report_ &lt; 100000) next_report_ += 10000;
        else if (next_report_ &lt; 500000) next_report_ += 50000;
        else                            next_report_ += 100000;
        fprintf(stderr, &quot;... finished %&quot; PRIu64 &quot; ops%30s\r&quot;, done_, &quot;&quot;);
      } else {
        uint64_t now = FLAGS_env-&gt;NowMicros();
        int64_t usecs_since_last = now - last_report_finish_;

        // Determine whether to print status where interval is either
        // each N operations or each N seconds.

        if (FLAGS_stats_interval_seconds &amp;&amp;
            usecs_since_last &lt; (FLAGS_stats_interval_seconds * 1000000)) {
          // Don&#039;t check again for this many operations
          next_report_ += FLAGS_stats_interval;

        } else {

          fprintf(stderr,
                  &quot;%s ... thread %d: (%&quot; PRIu64 &quot;,%&quot; PRIu64 &quot;) ops and &quot;
                  &quot;(%.1f,%.1f) ops/second in (%.6f,%.6f) seconds\n&quot;,
                  FLAGS_env-&gt;TimeToString(now/1000000).c_str(),
                  id_,
                  done_ - last_report_done_, done_,
                  (done_ - last_report_done_) /
                  (usecs_since_last / 1000000.0),
                  done_ / ((now - start_) / 1000000.0),
                  (now - last_report_finish_) / 1000000.0,
                  (now - start_) / 1000000.0);

          if (id_ == 0 &amp;&amp; FLAGS_stats_per_interval) {
            std::string stats;

            if (db_with_cfh &amp;&amp; db_with_cfh-&gt;num_created.load()) {
              for (size_t i = 0; i &lt; db_with_cfh-&gt;num_created.load(); ++i) {
                if (db-&gt;GetProperty(db_with_cfh-&gt;cfh[i], &quot;rocksdb.cfstats&quot;,
                                    &amp;stats))
                  fprintf(stderr, &quot;%s\n&quot;, stats.c_str());
                if (FLAGS_show_table_properties) {
                  for (int level = 0; level &lt; FLAGS_num_levels; ++level) {
                    if (db-&gt;GetProperty(
                            db_with_cfh-&gt;cfh[i],
                            &quot;rocksdb.aggregated-table-properties-at-level&quot; +
                                ToString(level),
                            &amp;stats)) {
                      if (stats.find(&quot;# entries=0&quot;) == std::string::npos) {
                        fprintf(stderr, &quot;Level[%d]: %s\n&quot;, level,
                                stats.c_str());
                      }
                    }
                  }
                }
              }
            } else if (db) {
              if (db-&gt;GetProperty(&quot;rocksdb.stats&quot;, &amp;stats)) {
                fprintf(stderr, &quot;%s\n&quot;, stats.c_str());
              }
              if (FLAGS_show_table_properties) {
                for (int level = 0; level &lt; FLAGS_num_levels; ++level) {
                  if (db-&gt;GetProperty(
                          &quot;rocksdb.aggregated-table-properties-at-level&quot; +
                              ToString(level),
                          &amp;stats)) {
                    if (stats.find(&quot;# entries=0&quot;) == std::string::npos) {
                      fprintf(stderr, &quot;Level[%d]: %s\n&quot;, level, stats.c_str());
                    }
                  }
                }
              }
            }
          }

          next_report_ += FLAGS_stats_interval;
          last_report_finish_ = now;
          last_report_done_ = done_;
        }
      }
      if (id_ == 0 &amp;&amp; FLAGS_thread_status_per_interval) {
        PrintThreadStatus();
      }
      fflush(stderr);
    }
  }

  void AddBytes(int64_t n) {
    bytes_ += n;
  }

  void Report(const Slice&amp; name) {
    // Pretend at least one op was done in case we are running a benchmark
    // that does not call FinishedOps().
    if (done_ &lt; 1) done_ = 1;

    std::string extra;
    if (bytes_ &gt; 0) {
      // Rate is computed on actual elapsed time, not the sum of per-thread
      // elapsed times.
      double elapsed = (finish_ - start_) * 1e-6;
      char rate[100];
      snprintf(rate, sizeof(rate), &quot;%6.1f MB/s&quot;,
               (bytes_ / 1048576.0) / elapsed);
      extra = rate;
    }
    AppendWithSpace(&amp;extra, message_);
    double elapsed = (finish_ - start_) * 1e-6;
    double throughput = (double)done_/elapsed;

    fprintf(stdout, &quot;%-12s : %11.3f micros/op %ld ops/sec;%s%s\n&quot;,
            name.ToString().c_str(),
            seconds_ * 1e6 / done_,
            (long)throughput,
            (extra.empty() ? &quot;&quot; : &quot; &quot;),
            extra.c_str());
    if (FLAGS_histogram) {
      for (auto it = hist_.begin(); it != hist_.end(); ++it) {
        fprintf(stdout, &quot;Microseconds per %s:\n%s\n&quot;,
                OperationTypeString[it-&gt;first].c_str(),
                it-&gt;second-&gt;ToString().c_str());
      }
    }
    if (FLAGS_report_file_operations) {
      ReportFileOpEnv* env = static_cast&lt;ReportFileOpEnv*&gt;(FLAGS_env);
      ReportFileOpCounters* counters = env-&gt;counters();
      fprintf(stdout, &quot;Num files opened: %d\n&quot;,
              counters-&gt;open_counter_.load(std::memory_order_relaxed));
      fprintf(stdout, &quot;Num Read(): %d\n&quot;,
              counters-&gt;read_counter_.load(std::memory_order_relaxed));
      fprintf(stdout, &quot;Num Append(): %d\n&quot;,
              counters-&gt;append_counter_.load(std::memory_order_relaxed));
      fprintf(stdout, &quot;Num bytes read: %&quot; PRIu64 &quot;\n&quot;,
              counters-&gt;bytes_read_.load(std::memory_order_relaxed));
      fprintf(stdout, &quot;Num bytes written: %&quot; PRIu64 &quot;\n&quot;,
              counters-&gt;bytes_written_.load(std::memory_order_relaxed));
      env-&gt;reset();
    }
    fflush(stdout);
  }
};

class CombinedStats {
 public:
  void AddStats(const Stats&amp; stat) {
    uint64_t total_ops = stat.done_;
    uint64_t total_bytes_ = stat.bytes_;
    double elapsed;

    if (total_ops &lt; 1) {
      total_ops = 1;
    }

    elapsed = (stat.finish_ - stat.start_) * 1e-6;
    throughput_ops_.emplace_back(total_ops / elapsed);

    if (total_bytes_ &gt; 0) {
      double mbs = (total_bytes_ / 1048576.0);
      throughput_mbs_.emplace_back(mbs / elapsed);
    }
  }

  void Report(const std::string&amp; bench_name) {
    const char* name = bench_name.c_str();
    int num_runs = static_cast&lt;int&gt;(throughput_ops_.size());

    if (throughput_mbs_.size() == throughput_ops_.size()) {
      fprintf(stdout,
              &quot;%s [AVG    %d runs] : %d ops/sec; %6.1f MB/sec\n&quot;
              &quot;%s [MEDIAN %d runs] : %d ops/sec; %6.1f MB/sec\n&quot;,
              name, num_runs, static_cast&lt;int&gt;(CalcAvg(throughput_ops_)),
              CalcAvg(throughput_mbs_), name, num_runs,
              static_cast&lt;int&gt;(CalcMedian(throughput_ops_)),
              CalcMedian(throughput_mbs_));
    } else {
      fprintf(stdout,
              &quot;%s [AVG    %d runs] : %d ops/sec\n&quot;
              &quot;%s [MEDIAN %d runs] : %d ops/sec\n&quot;,
              name, num_runs, static_cast&lt;int&gt;(CalcAvg(throughput_ops_)), name,
              num_runs, static_cast&lt;int&gt;(CalcMedian(throughput_ops_)));
    }
  }

 private:
  double CalcAvg(std::vector&lt;double&gt; data) {
    double avg = 0;
    for (double x : data) {
      avg += x;
    }
    avg = avg / data.size();
    return avg;
  }

  double CalcMedian(std::vector&lt;double&gt; data) {
    assert(data.size() &gt; 0);
    std::sort(data.begin(), data.end());

    size_t mid = data.size() / 2;
    if (data.size() % 2 == 1) {
      // Odd number of entries
      return data[mid];
    } else {
      // Even number of entries
      return (data[mid] + data[mid - 1]) / 2;
    }
  }

  std::vector&lt;double&gt; throughput_ops_;
  std::vector&lt;double&gt; throughput_mbs_;
};

class TimestampEmulator {
 private:
  std::atomic&lt;uint64_t&gt; timestamp_;

 public:
  TimestampEmulator() : timestamp_(0) {}
  uint64_t Get() const { return timestamp_.load(); }
  void Inc() { timestamp_++; }
};

// State shared by all concurrent executions of the same benchmark.
struct SharedState {
  port::Mutex mu;
  port::CondVar cv;
  int total;
  int perf_level;
  std::shared_ptr&lt;RateLimiter&gt; write_rate_limiter;
  std::shared_ptr&lt;RateLimiter&gt; read_rate_limiter;

  // Each thread goes through the following states:
  //    (1) initializing
  //    (2) waiting for others to be initialized
  //    (3) running
  //    (4) done

  long num_initialized;
  long num_done;
  bool start;

  SharedState() : cv(&amp;mu), perf_level(FLAGS_perf_level) { }
};

// Per-thread state for concurrent executions of the same benchmark.
struct ThreadState {
  int tid;             // 0..n-1 when running in n threads
  Random64 rand;         // Has different seeds for different threads
  Stats stats;
  SharedState* shared;

  /* implicit */ ThreadState(int index)
      : tid(index),
        rand((FLAGS_seed ? FLAGS_seed : 1000) + index) {
  }
};

class Duration {
 public:
  Duration(uint64_t max_seconds, int64_t max_ops, int64_t ops_per_stage = 0) {
    max_seconds_ = max_seconds;
    max_ops_= max_ops;
    ops_per_stage_ = (ops_per_stage &gt; 0) ? ops_per_stage : max_ops;
    ops_ = 0;
    start_at_ = FLAGS_env-&gt;NowMicros();
  }

  int64_t GetStage() { return std::min(ops_, max_ops_ - 1) / ops_per_stage_; }

  bool Done(int64_t increment) {
    if (increment &lt;= 0) increment = 1;    // avoid Done(0) and infinite loops
    ops_ += increment;

    if (max_seconds_) {
      // Recheck every appx 1000 ops (exact iff increment is factor of 1000)
      auto granularity = FLAGS_ops_between_duration_checks;
      if ((ops_ / granularity) != ((ops_ - increment) / granularity)) {
        uint64_t now = FLAGS_env-&gt;NowMicros();
        return ((now - start_at_) / 1000000) &gt;= max_seconds_;
      } else {
        return false;
      }
    } else {
      return ops_ &gt; max_ops_;
    }
  }

 private:
  uint64_t max_seconds_;
  int64_t max_ops_;
  int64_t ops_per_stage_;
  int64_t ops_;
  uint64_t start_at_;
};

class Benchmark {
 private:
  std::shared_ptr&lt;Cache&gt; cache_;
  std::shared_ptr&lt;Cache&gt; compressed_cache_;
  std::shared_ptr&lt;const FilterPolicy&gt; filter_policy_;
  const SliceTransform* prefix_extractor_;
  DBWithColumnFamilies db_;
  std::vector&lt;DBWithColumnFamilies&gt; multi_dbs_;
  int64_t num_;
  int key_size_;
  int prefix_size_;
  int64_t keys_per_prefix_;
  int64_t entries_per_batch_;
  int64_t writes_before_delete_range_;
  int64_t writes_per_range_tombstone_;
  int64_t range_tombstone_width_;
  int64_t max_num_range_tombstones_;
  WriteOptions write_options_;
  Options open_options_;  // keep options around to properly destroy db later
#ifndef ROCKSDB_LITE
  TraceOptions trace_options_;
  TraceOptions block_cache_trace_options_;
#endif
  int64_t reads_;
  int64_t deletes_;
  double read_random_exp_range_;
  int64_t writes_;
  int64_t readwrites_;
  int64_t merge_keys_;
  bool report_file_operations_;
  bool use_blob_db_;
  std::vector&lt;std::string&gt; keys_;

  class ErrorHandlerListener : public EventListener {
   public:
#ifndef ROCKSDB_LITE
    ErrorHandlerListener()
        : mutex_(),
          cv_(&amp;mutex_),
          no_auto_recovery_(false),
          recovery_complete_(false) {}

    ~ErrorHandlerListener() override {}

    void OnErrorRecoveryBegin(BackgroundErrorReason /*reason*/,
                              Status /*bg_error*/,
                              bool* auto_recovery) override {
      if (*auto_recovery &amp;&amp; no_auto_recovery_) {
        *auto_recovery = false;
      }
    }

    void OnErrorRecoveryCompleted(Status /*old_bg_error*/) override {
      InstrumentedMutexLock l(&amp;mutex_);
      recovery_complete_ = true;
      cv_.SignalAll();
    }

    bool WaitForRecovery(uint64_t abs_time_us) {
      InstrumentedMutexLock l(&amp;mutex_);
      if (!recovery_complete_) {
        cv_.TimedWait(abs_time_us);
      }
      if (recovery_complete_) {
        recovery_complete_ = false;
        return true;
      }
      return false;
    }

    void EnableAutoRecovery(bool enable = true) { no_auto_recovery_ = !enable; }

   private:
    InstrumentedMutex mutex_;
    InstrumentedCondVar cv_;
    bool no_auto_recovery_;
    bool recovery_complete_;
#else   // ROCKSDB_LITE
    bool WaitForRecovery(uint64_t /*abs_time_us*/) { return true; }
    void EnableAutoRecovery(bool /*enable*/) {}
#endif  // ROCKSDB_LITE
  };

  std::shared_ptr&lt;ErrorHandlerListener&gt; listener_;

  bool SanityCheck() {
    if (FLAGS_compression_ratio &gt; 1) {
      fprintf(stderr, &quot;compression_ratio should be between 0 and 1\n&quot;);
      return false;
    }
    return true;
  }

  inline bool CompressSlice(const CompressionInfo&amp; compression_info,
                            const Slice&amp; input, std::string* compressed) {
    bool ok = true;
    switch (FLAGS_compression_type_e) {
      case ROCKSDB_NAMESPACE::kSnappyCompression:
        ok = Snappy_Compress(compression_info, input.data(), input.size(),
                             compressed);
        break;
      case ROCKSDB_NAMESPACE::kZlibCompression:
        ok = Zlib_Compress(compression_info, 2, input.data(), input.size(),
                           compressed);
        break;
      case ROCKSDB_NAMESPACE::kBZip2Compression:
        ok = BZip2_Compress(compression_info, 2, input.data(), input.size(),
                            compressed);
        break;
      case ROCKSDB_NAMESPACE::kLZ4Compression:
        ok = LZ4_Compress(compression_info, 2, input.data(), input.size(),
                          compressed);
        break;
      case ROCKSDB_NAMESPACE::kLZ4HCCompression:
        ok = LZ4HC_Compress(compression_info, 2, input.data(), input.size(),
                            compressed);
        break;
      case ROCKSDB_NAMESPACE::kXpressCompression:
        ok = XPRESS_Compress(input.data(),
          input.size(), compressed);
        break;
      case ROCKSDB_NAMESPACE::kZSTD:
        ok = ZSTD_Compress(compression_info, input.data(), input.size(),
                           compressed);
        break;
      default:
        ok = false;
    }
    return ok;
  }

  void PrintHeader() {
    PrintEnvironment();
    fprintf(stdout, &quot;Keys:       %d bytes each\n&quot;, FLAGS_key_size);
    auto avg_value_size = FLAGS_value_size;
    if (FLAGS_value_size_distribution_type_e == kFixed) {
      fprintf(stdout, &quot;Values:     %d bytes each (%d bytes after compression)\n&quot;,
              avg_value_size,
              static_cast&lt;int&gt;(avg_value_size * FLAGS_compression_ratio + 0.5));
    } else {
      avg_value_size = (FLAGS_value_size_min + FLAGS_value_size_max) / 2;
      fprintf(stdout, &quot;Values:     %d avg bytes each (%d bytes after compression)\n&quot;,
              avg_value_size,
              static_cast&lt;int&gt;(avg_value_size * FLAGS_compression_ratio + 0.5));
      fprintf(stdout, &quot;Values Distribution: %s (min: %d, max: %d)\n&quot;,
              FLAGS_value_size_distribution_type.c_str(),
              FLAGS_value_size_min, FLAGS_value_size_max);
    }
    fprintf(stdout, &quot;Entries:    %&quot; PRIu64 &quot;\n&quot;, num_);
    fprintf(stdout, &quot;Prefix:    %d bytes\n&quot;, FLAGS_prefix_size);
    fprintf(stdout, &quot;Keys per prefix:    %&quot; PRIu64 &quot;\n&quot;, keys_per_prefix_);
    fprintf(stdout, &quot;RawSize:    %.1f MB (estimated)\n&quot;,
            ((static_cast&lt;int64_t&gt;(FLAGS_key_size + avg_value_size) * num_)
             / 1048576.0));
    fprintf(stdout, &quot;FileSize:   %.1f MB (estimated)\n&quot;,
            (((FLAGS_key_size + avg_value_size * FLAGS_compression_ratio)
              * num_)
             / 1048576.0));
    fprintf(stdout, &quot;Write rate: %&quot; PRIu64 &quot; bytes/second\n&quot;,
            FLAGS_benchmark_write_rate_limit);
    fprintf(stdout, &quot;Read rate: %&quot; PRIu64 &quot; ops/second\n&quot;,
            FLAGS_benchmark_read_rate_limit);
    if (FLAGS_enable_numa) {
      fprintf(stderr, &quot;Running in NUMA enabled mode.\n&quot;);
#ifndef NUMA
      fprintf(stderr, &quot;NUMA is not defined in the system.\n&quot;);
      exit(1);
#else
      if (numa_available() == -1) {
        fprintf(stderr, &quot;NUMA is not supported by the system.\n&quot;);
        exit(1);
      }
#endif
    }

    auto compression = CompressionTypeToString(FLAGS_compression_type_e);
    fprintf(stdout, &quot;Compression: %s\n&quot;, compression.c_str());
    fprintf(stdout, &quot;Compression sampling rate: %&quot; PRId64 &quot;\n&quot;,
            FLAGS_sample_for_compression);

    switch (FLAGS_rep_factory) {
      case kPrefixHash:
        fprintf(stdout, &quot;Memtablerep: prefix_hash\n&quot;);
        break;
      case kSkipList:
        fprintf(stdout, &quot;Memtablerep: skip_list\n&quot;);
        break;
      case kVectorRep:
        fprintf(stdout, &quot;Memtablerep: vector\n&quot;);
        break;
      case kHashLinkedList:
        fprintf(stdout, &quot;Memtablerep: hash_linkedlist\n&quot;);
        break;
    }
    fprintf(stdout, &quot;Perf Level: %d\n&quot;, FLAGS_perf_level);

    PrintWarnings(compression.c_str());
    fprintf(stdout, &quot;------------------------------------------------\n&quot;);
  }

  void PrintWarnings(const char* compression) {
#if defined(__GNUC__) &amp;&amp; !defined(__OPTIMIZE__)
    fprintf(stdout,
            &quot;WARNING: Optimization is disabled: benchmarks unnecessarily slow\n&quot;
            );
#endif
#ifndef NDEBUG
    fprintf(stdout,
            &quot;WARNING: Assertions are enabled; benchmarks unnecessarily slow\n&quot;);
#endif
    if (FLAGS_compression_type_e != ROCKSDB_NAMESPACE::kNoCompression) {
      // The test string should not be too small.
      const int len = FLAGS_block_size;
      std::string input_str(len, &#039;y&#039;);
      std::string compressed;
      CompressionOptions opts;
      CompressionContext context(FLAGS_compression_type_e);
      CompressionInfo info(opts, context, CompressionDict::GetEmptyDict(),
                           FLAGS_compression_type_e,
                           FLAGS_sample_for_compression);
      bool result = CompressSlice(info, Slice(input_str), &amp;compressed);

      if (!result) {
        fprintf(stdout, &quot;WARNING: %s compression is not enabled\n&quot;,
                compression);
      } else if (compressed.size() &gt;= input_str.size()) {
        fprintf(stdout, &quot;WARNING: %s compression is not effective\n&quot;,
                compression);
      }
    }
  }

// Current the following isn&#039;t equivalent to OS_LINUX.
#if defined(__linux)
  static Slice TrimSpace(Slice s) {
    unsigned int start = 0;
    while (start &lt; s.size() &amp;&amp; isspace(s[start])) {
      start++;
    }
    unsigned int limit = static_cast&lt;unsigned int&gt;(s.size());
    while (limit &gt; start &amp;&amp; isspace(s[limit-1])) {
      limit--;
    }
    return Slice(s.data() + start, limit - start);
  }
#endif

  void PrintEnvironment() {
    fprintf(stderr, &quot;RocksDB:    version %d.%d\n&quot;,
            kMajorVersion, kMinorVersion);

#if defined(__linux)
    time_t now = time(nullptr);
    char buf[52];
    // Lint complains about ctime() usage, so replace it with ctime_r(). The
    // requirement is to provide a buffer which is at least 26 bytes.
    fprintf(stderr, &quot;Date:       %s&quot;,
            ctime_r(&amp;now, buf));  // ctime_r() adds newline

    FILE* cpuinfo = fopen(&quot;/proc/cpuinfo&quot;, &quot;r&quot;);
    if (cpuinfo != nullptr) {
      char line[1000];
      int num_cpus = 0;
      std::string cpu_type;
      std::string cache_size;
      while (fgets(line, sizeof(line), cpuinfo) != nullptr) {
        const char* sep = strchr(line, &#039;:&#039;);
        if (sep == nullptr) {
          continue;
        }
        Slice key = TrimSpace(Slice(line, sep - 1 - line));
        Slice val = TrimSpace(Slice(sep + 1));
        if (key == &quot;model name&quot;) {
          ++num_cpus;
          cpu_type = val.ToString();
        } else if (key == &quot;cache size&quot;) {
          cache_size = val.ToString();
        }
      }
      fclose(cpuinfo);
      fprintf(stderr, &quot;CPU:        %d * %s\n&quot;, num_cpus, cpu_type.c_str());
      fprintf(stderr, &quot;CPUCache:   %s\n&quot;, cache_size.c_str());
    }
#endif
  }

  static bool KeyExpired(const TimestampEmulator* timestamp_emulator,
                         const Slice&amp; key) {
    const char* pos = key.data();
    pos += 8;
    uint64_t timestamp = 0;
    if (port::kLittleEndian) {
      int bytes_to_fill = 8;
      for (int i = 0; i &lt; bytes_to_fill; ++i) {
        timestamp |= (static_cast&lt;uint64_t&gt;(static_cast&lt;unsigned char&gt;(pos[i]))
                      &lt;&lt; ((bytes_to_fill - i - 1) &lt;&lt; 3));
      }
    } else {
      memcpy(&amp;timestamp, pos, sizeof(timestamp));
    }
    return timestamp_emulator-&gt;Get() - timestamp &gt; FLAGS_time_range;
  }

  class ExpiredTimeFilter : public CompactionFilter {
   public:
    explicit ExpiredTimeFilter(
        const std::shared_ptr&lt;TimestampEmulator&gt;&amp; timestamp_emulator)
        : timestamp_emulator_(timestamp_emulator) {}
    bool Filter(int /*level*/, const Slice&amp; key,
                const Slice&amp; /*existing_value*/, std::string* /*new_value*/,
                bool* /*value_changed*/) const override {
      return KeyExpired(timestamp_emulator_.get(), key);
    }
    const char* Name() const override { return &quot;ExpiredTimeFilter&quot;; }

   private:
    std::shared_ptr&lt;TimestampEmulator&gt; timestamp_emulator_;
  };

  class KeepFilter : public CompactionFilter {
   public:
    bool Filter(int /*level*/, const Slice&amp; /*key*/, const Slice&amp; /*value*/,
                std::string* /*new_value*/,
                bool* /*value_changed*/) const override {
      return false;
    }

    const char* Name() const override { return &quot;KeepFilter&quot;; }
  };

  std::shared_ptr&lt;Cache&gt; NewCache(int64_t capacity) {
    if (capacity &lt;= 0) {
      return nullptr;
    }
    if (FLAGS_use_clock_cache) {
      auto cache = NewClockCache(static_cast&lt;size_t&gt;(capacity),
                                 FLAGS_cache_numshardbits);
      if (!cache) {
        fprintf(stderr, &quot;Clock cache not supported.&quot;);
        exit(1);
      }
      return cache;
    } else {
      return NewLRUCache(
          static_cast&lt;size_t&gt;(capacity), FLAGS_cache_numshardbits,
          false /*strict_capacity_limit*/, FLAGS_cache_high_pri_pool_ratio);
    }
  }

 public:
  Benchmark()
      : cache_(NewCache(FLAGS_cache_size)),
        compressed_cache_(NewCache(FLAGS_compressed_cache_size)),
        filter_policy_(FLAGS_bloom_bits &gt;= 0
                           ? NewBloomFilterPolicy(FLAGS_bloom_bits,
                                                  FLAGS_use_block_based_filter)
                           : nullptr),
        prefix_extractor_(NewFixedPrefixTransform(FLAGS_prefix_size)),
        num_(FLAGS_num),
        key_size_(FLAGS_key_size),
        prefix_size_(FLAGS_prefix_size),
        keys_per_prefix_(FLAGS_keys_per_prefix),
        entries_per_batch_(1),
        reads_(FLAGS_reads &lt; 0 ? FLAGS_num : FLAGS_reads),
        read_random_exp_range_(0.0),
        writes_(FLAGS_writes &lt; 0 ? FLAGS_num : FLAGS_writes),
        readwrites_(
            (FLAGS_writes &lt; 0 &amp;&amp; FLAGS_reads &lt; 0)
                ? FLAGS_num
                : ((FLAGS_writes &gt; FLAGS_reads) ? FLAGS_writes : FLAGS_reads)),
        merge_keys_(FLAGS_merge_keys &lt; 0 ? FLAGS_num : FLAGS_merge_keys),
        report_file_operations_(FLAGS_report_file_operations),
#ifndef ROCKSDB_LITE
        use_blob_db_(FLAGS_use_blob_db)
#else
        use_blob_db_(false)
#endif  // !ROCKSDB_LITE
  {
    // use simcache instead of cache
    if (FLAGS_simcache_size &gt;= 0) {
      if (FLAGS_cache_numshardbits &gt;= 1) {
        cache_ =
            NewSimCache(cache_, FLAGS_simcache_size, FLAGS_cache_numshardbits);
      } else {
        cache_ = NewSimCache(cache_, FLAGS_simcache_size, 0);
      }
    }

    if (report_file_operations_) {
      if (!FLAGS_hdfs.empty()) {
        fprintf(stderr,
                &quot;--hdfs and --report_file_operations cannot be enabled &quot;
                &quot;at the same time&quot;);
        exit(1);
      }
      FLAGS_env = new ReportFileOpEnv(FLAGS_env);
    }

    if (FLAGS_prefix_size &gt; FLAGS_key_size) {
      fprintf(stderr, &quot;prefix size is larger than key size&quot;);
      exit(1);
    }

    std::vector&lt;std::string&gt; files;
    FLAGS_env-&gt;GetChildren(FLAGS_db, &amp;files);
    for (size_t i = 0; i &lt; files.size(); i++) {
      if (Slice(files[i]).starts_with(&quot;heap-&quot;)) {
        FLAGS_env-&gt;DeleteFile(FLAGS_db + &quot;/&quot; + files[i]);
      }
    }
    if (!FLAGS_use_existing_db) {
      Options options;
      options.env = FLAGS_env;
      if (!FLAGS_wal_dir.empty()) {
        options.wal_dir = FLAGS_wal_dir;
      }
#ifndef ROCKSDB_LITE
      if (use_blob_db_) {
        blob_db::DestroyBlobDB(FLAGS_db, options, blob_db::BlobDBOptions());
      }
#endif  // !ROCKSDB_LITE
      DestroyDB(FLAGS_db, options);
      if (!FLAGS_wal_dir.empty()) {
        FLAGS_env-&gt;DeleteDir(FLAGS_wal_dir);
      }

      if (FLAGS_num_multi_db &gt; 1) {
        FLAGS_env-&gt;CreateDir(FLAGS_db);
        if (!FLAGS_wal_dir.empty()) {
          FLAGS_env-&gt;CreateDir(FLAGS_wal_dir);
        }
      }
    }

    listener_.reset(new ErrorHandlerListener());
  }

  ~Benchmark() {
    db_.DeleteDBs();
    delete prefix_extractor_;
    if (cache_.get() != nullptr) {
      // this will leak, but we&#039;re shutting down so nobody cares
      cache_-&gt;DisownData();
    }
  }

  Slice AllocateKey(std::unique_ptr&lt;const char[]&gt;* key_guard) {
    char* data = new char[key_size_];
    const char* const_data = data;
    key_guard-&gt;reset(const_data);
    return Slice(key_guard-&gt;get(), key_size_);
  }

  // Generate key according to the given specification and random number.
  // The resulting key will have the following format (if keys_per_prefix_
  // is positive), extra trailing bytes are either cut off or padded with &#039;0&#039;.
  // The prefix value is derived from key value.
  //   ----------------------------
  //   | prefix 00000 | key 00000 |
  //   ----------------------------
  // If keys_per_prefix_ is 0, the key is simply a binary representation of
  // random number followed by trailing &#039;0&#039;s
  //   ----------------------------
  //   |        key 00000         |
  //   ----------------------------
  void GenerateKeyFromInt(uint64_t v, int64_t num_keys, Slice* key) {
    if (!keys_.empty()) {
      assert(FLAGS_use_existing_keys);
      assert(keys_.size() == static_cast&lt;size_t&gt;(num_keys));
      assert(v &lt; static_cast&lt;uint64_t&gt;(num_keys));
      *key = keys_[v];
      return;
    }
    char* start = const_cast&lt;char*&gt;(key-&gt;data());
    char* pos = start;
    if (keys_per_prefix_ &gt; 0) {
      int64_t num_prefix = num_keys / keys_per_prefix_;
      int64_t prefix = v % num_prefix;
      int bytes_to_fill = std::min(prefix_size_, 8);
      if (port::kLittleEndian) {
        for (int i = 0; i &lt; bytes_to_fill; ++i) {
          pos[i] = (prefix &gt;&gt; ((bytes_to_fill - i - 1) &lt;&lt; 3)) &amp; 0xFF;
        }
      } else {
        memcpy(pos, static_cast&lt;void*&gt;(&amp;prefix), bytes_to_fill);
      }
      if (prefix_size_ &gt; 8) {
        // fill the rest with 0s
        memset(pos + 8, &#039;0&#039;, prefix_size_ - 8);
      }
      pos += prefix_size_;
    }

    int bytes_to_fill = std::min(key_size_ - static_cast&lt;int&gt;(pos - start), 8);
    if (port::kLittleEndian) {
      for (int i = 0; i &lt; bytes_to_fill; ++i) {
        pos[i] = (v &gt;&gt; ((bytes_to_fill - i - 1) &lt;&lt; 3)) &amp; 0xFF;
      }
    } else {
      memcpy(pos, static_cast&lt;void*&gt;(&amp;v), bytes_to_fill);
    }
    pos += bytes_to_fill;
    if (key_size_ &gt; pos - start) {
      memset(pos, &#039;0&#039;, key_size_ - (pos - start));
    }
  }

  void GenerateKeyFromIntForSeek(uint64_t v, int64_t num_keys, Slice* key) {
    GenerateKeyFromInt(v, num_keys, key);
    if (FLAGS_seek_missing_prefix) {
      assert(prefix_size_ &gt; 8);
      char* key_ptr = const_cast&lt;char*&gt;(key-&gt;data());
      // This rely on GenerateKeyFromInt filling paddings with &#039;0&#039;s.
      // Putting a &#039;1&#039; will create a non-existing prefix.
      key_ptr[8] = &#039;1&#039;;
    }
  }

  std::string GetPathForMultiple(std::string base_name, size_t id) {
    if (!base_name.empty()) {
#ifndef OS_WIN
      if (base_name.back() != &#039;/&#039;) {
        base_name += &#039;/&#039;;
      }
#else
      if (base_name.back() != &#039;\\&#039;) {
        base_name += &#039;\\&#039;;
      }
#endif
    }
    return base_name + ToString(id);
  }

  void VerifyDBFromDB(std::string&amp; truth_db_name) {
    DBWithColumnFamilies truth_db;
    auto s = DB::OpenForReadOnly(open_options_, truth_db_name, &amp;truth_db.db);
    if (!s.ok()) {
      fprintf(stderr, &quot;open error: %s\n&quot;, s.ToString().c_str());
      exit(1);
    }
    ReadOptions ro;
    ro.total_order_seek = true;
    std::unique_ptr&lt;Iterator&gt; truth_iter(truth_db.db-&gt;NewIterator(ro));
    std::unique_ptr&lt;Iterator&gt; db_iter(db_.db-&gt;NewIterator(ro));
    // Verify that all the key/values in truth_db are retrivable in db with
    // ::Get
    fprintf(stderr, &quot;Verifying db &gt;= truth_db with ::Get...\n&quot;);
    for (truth_iter-&gt;SeekToFirst(); truth_iter-&gt;Valid(); truth_iter-&gt;Next()) {
      std::string value;
      s = db_.db-&gt;Get(ro, truth_iter-&gt;key(), &amp;value);
      assert(s.ok());
      // TODO(myabandeh): provide debugging hints
      assert(Slice(value) == truth_iter-&gt;value());
    }
    // Verify that the db iterator does not give any extra key/value
    fprintf(stderr, &quot;Verifying db == truth_db...\n&quot;);
    for (db_iter-&gt;SeekToFirst(), truth_iter-&gt;SeekToFirst(); db_iter-&gt;Valid();
         db_iter-&gt;Next(), truth_iter-&gt;Next()) {
      assert(truth_iter-&gt;Valid());
      assert(truth_iter-&gt;value() == db_iter-&gt;value());
    }
    // No more key should be left unchecked in truth_db
    assert(!truth_iter-&gt;Valid());
    fprintf(stderr, &quot;...Verified\n&quot;);
  }

  void Run() {
    if (!SanityCheck()) {
      exit(1);
    }
    Open(&amp;open_options_);
    PrintHeader();
    std::stringstream benchmark_stream(FLAGS_benchmarks);
    std::string name;
    std::unique_ptr&lt;ExpiredTimeFilter&gt; filter;
    while (std::getline(benchmark_stream, name, &#039;,&#039;)) {
      // Sanitize parameters
      num_ = FLAGS_num;
      reads_ = (FLAGS_reads &lt; 0 ? FLAGS_num : FLAGS_reads);
      writes_ = (FLAGS_writes &lt; 0 ? FLAGS_num : FLAGS_writes);
      deletes_ = (FLAGS_deletes &lt; 0 ? FLAGS_num : FLAGS_deletes);
      value_size = FLAGS_value_size;
      key_size_ = FLAGS_key_size;
      entries_per_batch_ = FLAGS_batch_size;
      writes_before_delete_range_ = FLAGS_writes_before_delete_range;
      writes_per_range_tombstone_ = FLAGS_writes_per_range_tombstone;
      range_tombstone_width_ = FLAGS_range_tombstone_width;
      max_num_range_tombstones_ = FLAGS_max_num_range_tombstones;
      write_options_ = WriteOptions();
      read_random_exp_range_ = FLAGS_read_random_exp_range;
      if (FLAGS_sync) {
        write_options_.sync = true;
      }
      write_options_.disableWAL = FLAGS_disable_wal;

      void (Benchmark::*method)(ThreadState*) = nullptr;
      void (Benchmark::*post_process_method)() = nullptr;

      bool fresh_db = false;
      int num_threads = FLAGS_threads;

      int num_repeat = 1;
      int num_warmup = 0;
      if (!name.empty() &amp;&amp; *name.rbegin() == &#039;]&#039;) {
        auto it = name.find(&#039;[&#039;);
        if (it == std::string::npos) {
          fprintf(stderr, &quot;unknown benchmark arguments &#039;%s&#039;\n&quot;, name.c_str());
          exit(1);
        }
        std::string args = name.substr(it + 1);
        args.resize(args.size() - 1);
        name.resize(it);

        std::string bench_arg;
        std::stringstream args_stream(args);
        while (std::getline(args_stream, bench_arg, &#039;-&#039;)) {
          if (bench_arg.empty()) {
            continue;
          }
          if (bench_arg[0] == &#039;X&#039;) {
            // Repeat the benchmark n times
            std::string num_str = bench_arg.substr(1);
            num_repeat = std::stoi(num_str);
          } else if (bench_arg[0] == &#039;W&#039;) {
            // Warm up the benchmark for n times
            std::string num_str = bench_arg.substr(1);
            num_warmup = std::stoi(num_str);
          }
        }
      }

      // Both fillseqdeterministic and filluniquerandomdeterministic
      // fill the levels except the max level with UNIQUE_RANDOM
      // and fill the max level with fillseq and filluniquerandom, respectively
      if (name == &quot;fillseqdeterministic&quot; ||
          name == &quot;filluniquerandomdeterministic&quot;) {
        if (!FLAGS_disable_auto_compactions) {
          fprintf(stderr,
                  &quot;Please disable_auto_compactions in FillDeterministic &quot;
                  &quot;benchmark\n&quot;);
          exit(1);
        }
        if (num_threads &gt; 1) {
          fprintf(stderr,
                  &quot;filldeterministic multithreaded not supported&quot;
                  &quot;, use 1 thread\n&quot;);
          num_threads = 1;
        }
        fresh_db = true;
        if (name == &quot;fillseqdeterministic&quot;) {
          method = &amp;Benchmark::WriteSeqDeterministic;
        } else {
          method = &amp;Benchmark::WriteUniqueRandomDeterministic;
        }
      } else if (name == &quot;fillseq&quot;) {
        fresh_db = true;
        method = &amp;Benchmark::WriteSeq;
      } else if (name == &quot;fillbatch&quot;) {
        fresh_db = true;
        entries_per_batch_ = 1000;
        method = &amp;Benchmark::WriteSeq;
      } else if (name == &quot;fillrandom&quot;) {
        fresh_db = true;
        method = &amp;Benchmark::WriteRandom;
      } else if (name == &quot;filluniquerandom&quot;) {
        fresh_db = true;
        if (num_threads &gt; 1) {
          fprintf(stderr,
                  &quot;filluniquerandom multithreaded not supported&quot;
                  &quot;, use 1 thread&quot;);
          num_threads = 1;
        }
        method = &amp;Benchmark::WriteUniqueRandom;
      } else if (name == &quot;overwrite&quot;) {
        method = &amp;Benchmark::WriteRandom;
      } else if (name == &quot;fillsync&quot;) {
        fresh_db = true;
        num_ /= 1000;
        write_options_.sync = true;
        method = &amp;Benchmark::WriteRandom;
      } else if (name == &quot;fill100K&quot;) {
        fresh_db = true;
        num_ /= 1000;
        value_size = 100 * 1000;
        method = &amp;Benchmark::WriteRandom;
      } else if (name == &quot;readseq&quot;) {
        method = &amp;Benchmark::ReadSequential;
      } else if (name == &quot;readtorowcache&quot;) {
        if (!FLAGS_use_existing_keys || !FLAGS_row_cache_size) {
          fprintf(stderr,
                  &quot;Please set use_existing_keys to true and specify a &quot;
                  &quot;row cache size in readtorowcache benchmark\n&quot;);
          exit(1);
        }
        method = &amp;Benchmark::ReadToRowCache;
      } else if (name == &quot;readtocache&quot;) {
        method = &amp;Benchmark::ReadSequential;
        num_threads = 1;
        reads_ = num_;
      } else if (name == &quot;readreverse&quot;) {
        method = &amp;Benchmark::ReadReverse;
      } else if (name == &quot;readrandom&quot;) {
        if (FLAGS_multiread_stride) {
          fprintf(stderr, &quot;entries_per_batch = %&quot; PRIi64 &quot;\n&quot;,
                  entries_per_batch_);
        }
        method = &amp;Benchmark::ReadRandom;
      } else if (name == &quot;readrandomfast&quot;) {
        method = &amp;Benchmark::ReadRandomFast;
      } else if (name == &quot;multireadrandom&quot;) {
        fprintf(stderr, &quot;entries_per_batch = %&quot; PRIi64 &quot;\n&quot;,
                entries_per_batch_);
        method = &amp;Benchmark::MultiReadRandom;
      } else if (name == &quot;mixgraph&quot;) {
        method = &amp;Benchmark::MixGraph;
      } else if (name == &quot;readmissing&quot;) {
        ++key_size_;
        method = &amp;Benchmark::ReadRandom;
      } else if (name == &quot;newiterator&quot;) {
        method = &amp;Benchmark::IteratorCreation;
      } else if (name == &quot;newiteratorwhilewriting&quot;) {
        num_threads++;  // Add extra thread for writing
        method = &amp;Benchmark::IteratorCreationWhileWriting;
      } else if (name == &quot;seekrandom&quot;) {
        method = &amp;Benchmark::SeekRandom;
      } else if (name == &quot;seekrandomwhilewriting&quot;) {
        num_threads++;  // Add extra thread for writing
        method = &amp;Benchmark::SeekRandomWhileWriting;
      } else if (name == &quot;seekrandomwhilemerging&quot;) {
        num_threads++;  // Add extra thread for merging
        method = &amp;Benchmark::SeekRandomWhileMerging;
      } else if (name == &quot;readrandomsmall&quot;) {
        reads_ /= 1000;
        method = &amp;Benchmark::ReadRandom;
      } else if (name == &quot;deleteseq&quot;) {
        method = &amp;Benchmark::DeleteSeq;
      } else if (name == &quot;deleterandom&quot;) {
        method = &amp;Benchmark::DeleteRandom;
      } else if (name == &quot;readwhilewriting&quot;) {
        num_threads++;  // Add extra thread for writing
        method = &amp;Benchmark::ReadWhileWriting;
      } else if (name == &quot;readwhilemerging&quot;) {
        num_threads++;  // Add extra thread for writing
        method = &amp;Benchmark::ReadWhileMerging;
      } else if (name == &quot;readwhilescanning&quot;) {
        num_threads++;  // Add extra thread for scaning
        method = &amp;Benchmark::ReadWhileScanning;
      } else if (name == &quot;readrandomwriterandom&quot;) {
        method = &amp;Benchmark::ReadRandomWriteRandom;
      } else if (name == &quot;readrandommergerandom&quot;) {
        if (FLAGS_merge_operator.empty()) {
          fprintf(stdout, &quot;%-12s : skipped (--merge_operator is unknown)\n&quot;,
                  name.c_str());
          exit(1);
        }
        method = &amp;Benchmark::ReadRandomMergeRandom;
      } else if (name == &quot;updaterandom&quot;) {
        method = &amp;Benchmark::UpdateRandom;
      } else if (name == &quot;xorupdaterandom&quot;) {
        method = &amp;Benchmark::XORUpdateRandom;
      } else if (name == &quot;appendrandom&quot;) {
        method = &amp;Benchmark::AppendRandom;
      } else if (name == &quot;mergerandom&quot;) {
        if (FLAGS_merge_operator.empty()) {
          fprintf(stdout, &quot;%-12s : skipped (--merge_operator is unknown)\n&quot;,
                  name.c_str());
          exit(1);
        }
        method = &amp;Benchmark::MergeRandom;
      } else if (name == &quot;randomwithverify&quot;) {
        method = &amp;Benchmark::RandomWithVerify;
      } else if (name == &quot;fillseekseq&quot;) {
        method = &amp;Benchmark::WriteSeqSeekSeq;
      } else if (name == &quot;compact&quot;) {
        method = &amp;Benchmark::Compact;
      } else if (name == &quot;compactall&quot;) {
        CompactAll();
      } else if (name == &quot;crc32c&quot;) {
        method = &amp;Benchmark::Crc32c;
      } else if (name == &quot;xxhash&quot;) {
        method = &amp;Benchmark::xxHash;
      } else if (name == &quot;acquireload&quot;) {
        method = &amp;Benchmark::AcquireLoad;
      } else if (name == &quot;compress&quot;) {
        method = &amp;Benchmark::Compress;
      } else if (name == &quot;uncompress&quot;) {
        method = &amp;Benchmark::Uncompress;
#ifndef ROCKSDB_LITE
      } else if (name == &quot;randomtransaction&quot;) {
        method = &amp;Benchmark::RandomTransaction;
        post_process_method = &amp;Benchmark::RandomTransactionVerify;
#endif  // ROCKSDB_LITE
      } else if (name == &quot;randomreplacekeys&quot;) {
        fresh_db = true;
        method = &amp;Benchmark::RandomReplaceKeys;
      } else if (name == &quot;timeseries&quot;) {
        timestamp_emulator_.reset(new TimestampEmulator());
        if (FLAGS_expire_style == &quot;compaction_filter&quot;) {
          filter.reset(new ExpiredTimeFilter(timestamp_emulator_));
          fprintf(stdout, &quot;Compaction filter is used to remove expired data&quot;);
          open_options_.compaction_filter = filter.get();
        }
        fresh_db = true;
        method = &amp;Benchmark::TimeSeries;
      } else if (name == &quot;stats&quot;) {
        PrintStats(&quot;rocksdb.stats&quot;);
      } else if (name == &quot;resetstats&quot;) {
        ResetStats();
      } else if (name == &quot;verify&quot;) {
        VerifyDBFromDB(FLAGS_truth_db);
      } else if (name == &quot;levelstats&quot;) {
        PrintStats(&quot;rocksdb.levelstats&quot;);
      } else if (name == &quot;sstables&quot;) {
        PrintStats(&quot;rocksdb.sstables&quot;);
      } else if (name == &quot;stats_history&quot;) {
        PrintStatsHistory();
      } else if (name == &quot;replay&quot;) {
        if (num_threads &gt; 1) {
          fprintf(stderr, &quot;Multi-threaded replay is not yet supported\n&quot;);
          exit(1);
        }
        if (FLAGS_trace_file == &quot;&quot;) {
          fprintf(stderr, &quot;Please set --trace_file to be replayed from\n&quot;);
          exit(1);
        }
        method = &amp;Benchmark::Replay;
      } else if (name == &quot;getmergeoperands&quot;) {
        method = &amp;Benchmark::GetMergeOperands;
      } else if (!name.empty()) {  // No error message for empty name
        fprintf(stderr, &quot;unknown benchmark &#039;%s&#039;\n&quot;, name.c_str());
        exit(1);
      }

      if (fresh_db) {
        if (FLAGS_use_existing_db) {
          fprintf(stdout, &quot;%-12s : skipped (--use_existing_db is true)\n&quot;,
                  name.c_str());
          method = nullptr;
        } else {
          if (db_.db != nullptr) {
            db_.DeleteDBs();
            DestroyDB(FLAGS_db, open_options_);
          }
          Options options = open_options_;
          for (size_t i = 0; i &lt; multi_dbs_.size(); i++) {
            delete multi_dbs_[i].db;
            if (!open_options_.wal_dir.empty()) {
              options.wal_dir = GetPathForMultiple(open_options_.wal_dir, i);
            }
            DestroyDB(GetPathForMultiple(FLAGS_db, i), options);
          }
          multi_dbs_.clear();
        }
        Open(&amp;open_options_);  // use open_options for the last accessed
      }

      if (method != nullptr) {
        fprintf(stdout, &quot;DB path: [%s]\n&quot;, FLAGS_db.c_str());

#ifndef ROCKSDB_LITE
        // A trace_file option can be provided both for trace and replay
        // operations. But db_bench does not support tracing and replaying at
        // the same time, for now. So, start tracing only when it is not a
        // replay.
        if (FLAGS_trace_file != &quot;&quot; &amp;&amp; name != &quot;replay&quot;) {
          std::unique_ptr&lt;TraceWriter&gt; trace_writer;
          Status s = NewFileTraceWriter(FLAGS_env, EnvOptions(),
                                        FLAGS_trace_file, &amp;trace_writer);
          if (!s.ok()) {
            fprintf(stderr, &quot;Encountered an error starting a trace, %s\n&quot;,
                    s.ToString().c_str());
            exit(1);
          }
          s = db_.db-&gt;StartTrace(trace_options_, std::move(trace_writer));
          if (!s.ok()) {
            fprintf(stderr, &quot;Encountered an error starting a trace, %s\n&quot;,
                    s.ToString().c_str());
            exit(1);
          }
          fprintf(stdout, &quot;Tracing the workload to: [%s]\n&quot;,
                  FLAGS_trace_file.c_str());
        }
        // Start block cache tracing.
        if (!FLAGS_block_cache_trace_file.empty()) {
          // Sanity checks.
          if (FLAGS_block_cache_trace_sampling_frequency &lt;= 0) {
            fprintf(stderr,
                    &quot;Block cache trace sampling frequency must be higher than &quot;
                    &quot;0.\n&quot;);
            exit(1);
          }
          if (FLAGS_block_cache_trace_max_trace_file_size_in_bytes &lt;= 0) {
            fprintf(stderr,
                    &quot;The maximum file size for block cache tracing must be &quot;
                    &quot;higher than 0.\n&quot;);
            exit(1);
          }
          block_cache_trace_options_.max_trace_file_size =
              FLAGS_block_cache_trace_max_trace_file_size_in_bytes;
          block_cache_trace_options_.sampling_frequency =
              FLAGS_block_cache_trace_sampling_frequency;
          std::unique_ptr&lt;TraceWriter&gt; block_cache_trace_writer;
          Status s = NewFileTraceWriter(FLAGS_env, EnvOptions(),
                                        FLAGS_block_cache_trace_file,
                                        &amp;block_cache_trace_writer);
          if (!s.ok()) {
            fprintf(stderr,
                    &quot;Encountered an error when creating trace writer, %s\n&quot;,
                    s.ToString().c_str());
            exit(1);
          }
          s = db_.db-&gt;StartBlockCacheTrace(block_cache_trace_options_,
                                           std::move(block_cache_trace_writer));
          if (!s.ok()) {
            fprintf(
                stderr,
                &quot;Encountered an error when starting block cache tracing, %s\n&quot;,
                s.ToString().c_str());
            exit(1);
          }
          fprintf(stdout, &quot;Tracing block cache accesses to: [%s]\n&quot;,
                  FLAGS_block_cache_trace_file.c_str());
        }
#endif  // ROCKSDB_LITE

        if (num_warmup &gt; 0) {
          printf(&quot;Warming up benchmark by running %d times\n&quot;, num_warmup);
        }

        for (int i = 0; i &lt; num_warmup; i++) {
          RunBenchmark(num_threads, name, method);
        }

        if (num_repeat &gt; 1) {
          printf(&quot;Running benchmark for %d times\n&quot;, num_repeat);
        }

        CombinedStats combined_stats;
        for (int i = 0; i &lt; num_repeat; i++) {
          Stats stats = RunBenchmark(num_threads, name, method);
          combined_stats.AddStats(stats);
        }
        if (num_repeat &gt; 1) {
          combined_stats.Report(name);
        }
      }
      if (post_process_method != nullptr) {
        (this-&gt;*post_process_method)();
      }
    }

    if (secondary_update_thread_) {
      secondary_update_stopped_.store(1, std::memory_order_relaxed);
      secondary_update_thread_-&gt;join();
      secondary_update_thread_.reset();
    }

#ifndef ROCKSDB_LITE
    if (name != &quot;replay&quot; &amp;&amp; FLAGS_trace_file != &quot;&quot;) {
      Status s = db_.db-&gt;EndTrace();
      if (!s.ok()) {
        fprintf(stderr, &quot;Encountered an error ending the trace, %s\n&quot;,
                s.ToString().c_str());
      }
    }
    if (!FLAGS_block_cache_trace_file.empty()) {
      Status s = db_.db-&gt;EndBlockCacheTrace();
      if (!s.ok()) {
        fprintf(stderr,
                &quot;Encountered an error ending the block cache tracing, %s\n&quot;,
                s.ToString().c_str());
      }
    }
#endif  // ROCKSDB_LITE

    if (FLAGS_statistics) {
      fprintf(stdout, &quot;STATISTICS:\n%s\n&quot;, dbstats-&gt;ToString().c_str());
    }
    if (FLAGS_simcache_size &gt;= 0) {
      fprintf(stdout, &quot;SIMULATOR CACHE STATISTICS:\n%s\n&quot;,
              static_cast_with_check&lt;SimCache, Cache&gt;(cache_.get())
                  -&gt;ToString()
                  .c_str());
    }

#ifndef ROCKSDB_LITE
    if (FLAGS_use_secondary_db) {
      fprintf(stdout, &quot;Secondary instance updated  %&quot; PRIu64 &quot; times.\n&quot;,
              secondary_db_updates_);
    }
#endif  // ROCKSDB_LITE
  }

 private:
  std::shared_ptr&lt;TimestampEmulator&gt; timestamp_emulator_;
  std::unique_ptr&lt;port::Thread&gt; secondary_update_thread_;
  std::atomic&lt;int&gt; secondary_update_stopped_{0};
#ifndef ROCKSDB_LITE
  uint64_t secondary_db_updates_ = 0;
#endif  // ROCKSDB_LITE
  struct ThreadArg {
    Benchmark* bm;
    SharedState* shared;
    ThreadState* thread;
    void (Benchmark::*method)(ThreadState*);
  };

  static void ThreadBody(void* v) {
    ThreadArg* arg = reinterpret_cast&lt;ThreadArg*&gt;(v);
    SharedState* shared = arg-&gt;shared;
    ThreadState* thread = arg-&gt;thread;
    {
      MutexLock l(&amp;shared-&gt;mu);
      shared-&gt;num_initialized++;
      if (shared-&gt;num_initialized &gt;= shared-&gt;total) {
        shared-&gt;cv.SignalAll();
      }
      while (!shared-&gt;start) {
        shared-&gt;cv.Wait();
      }
    }

    SetPerfLevel(static_cast&lt;PerfLevel&gt; (shared-&gt;perf_level));
    perf_context.EnablePerLevelPerfContext();
    thread-&gt;stats.Start(thread-&gt;tid);
    (arg-&gt;bm-&gt;*(arg-&gt;method))(thread);
    thread-&gt;stats.Stop();

    {
      MutexLock l(&amp;shared-&gt;mu);
      shared-&gt;num_done++;
      if (shared-&gt;num_done &gt;= shared-&gt;total) {
        shared-&gt;cv.SignalAll();
      }
    }
  }

  Stats RunBenchmark(int n, Slice name,
                     void (Benchmark::*method)(ThreadState*)) {
    SharedState shared;
    shared.total = n;
    shared.num_initialized = 0;
    shared.num_done = 0;
    shared.start = false;
    if (FLAGS_benchmark_write_rate_limit &gt; 0) {
      shared.write_rate_limiter.reset(
          NewGenericRateLimiter(FLAGS_benchmark_write_rate_limit));
    }
    if (FLAGS_benchmark_read_rate_limit &gt; 0) {
      shared.read_rate_limiter.reset(NewGenericRateLimiter(
          FLAGS_benchmark_read_rate_limit, 100000 /* refill_period_us */,
          10 /* fairness */, RateLimiter::Mode::kReadsOnly));
    }

    std::unique_ptr&lt;ReporterAgent&gt; reporter_agent;
    if (FLAGS_report_interval_seconds &gt; 0) {
      reporter_agent.reset(new ReporterAgent(FLAGS_env, FLAGS_report_file,
                                             FLAGS_report_interval_seconds));
    }

    ThreadArg* arg = new ThreadArg[n];

    for (int i = 0; i &lt; n; i++) {
#ifdef NUMA
      if (FLAGS_enable_numa) {
        // Performs a local allocation of memory to threads in numa node.
        int n_nodes = numa_num_task_nodes();  // Number of nodes in NUMA.
        numa_exit_on_error = 1;
        int numa_node = i % n_nodes;
        bitmask* nodes = numa_allocate_nodemask();
        numa_bitmask_clearall(nodes);
        numa_bitmask_setbit(nodes, numa_node);
        // numa_bind() call binds the process to the node and these
        // properties are passed on to the thread that is created in
        // StartThread method called later in the loop.
        numa_bind(nodes);
        numa_set_strict(1);
        numa_free_nodemask(nodes);
      }
#endif
      arg[i].bm = this;
      arg[i].method = method;
      arg[i].shared = &amp;shared;
      arg[i].thread = new ThreadState(i);
      arg[i].thread-&gt;stats.SetReporterAgent(reporter_agent.get());
      arg[i].thread-&gt;shared = &amp;shared;
      FLAGS_env-&gt;StartThread(ThreadBody, &amp;arg[i]);
    }

    shared.mu.Lock();
    while (shared.num_initialized &lt; n) {
      shared.cv.Wait();
    }

    shared.start = true;
    shared.cv.SignalAll();
    while (shared.num_done &lt; n) {
      shared.cv.Wait();
    }
    shared.mu.Unlock();

    // Stats for some threads can be excluded.
    Stats merge_stats;
    for (int i = 0; i &lt; n; i++) {
      merge_stats.Merge(arg[i].thread-&gt;stats);
    }
    merge_stats.Report(name);

    for (int i = 0; i &lt; n; i++) {
      delete arg[i].thread;
    }
    delete[] arg;

    return merge_stats;
  }

  void Crc32c(ThreadState* thread) {
    // Checksum about 500MB of data total
    const int size = FLAGS_block_size; // use --block_size option for db_bench
    std::string labels = &quot;(&quot; + ToString(FLAGS_block_size) + &quot; per op)&quot;;
    const char* label = labels.c_str();

    std::string data(size, &#039;x&#039;);
    int64_t bytes = 0;
    uint32_t crc = 0;
    while (bytes &lt; 500 * 1048576) {
      crc = crc32c::Value(data.data(), size);
      thread-&gt;stats.FinishedOps(nullptr, nullptr, 1, kCrc);
      bytes += size;
    }
    // Print so result is not dead
    fprintf(stderr, &quot;... crc=0x%x\r&quot;, static_cast&lt;unsigned int&gt;(crc));

    thread-&gt;stats.AddBytes(bytes);
    thread-&gt;stats.AddMessage(label);
  }

  void xxHash(ThreadState* thread) {
    // Checksum about 500MB of data total
    const int size = 4096;
    const char* label = &quot;(4K per op)&quot;;
    std::string data(size, &#039;x&#039;);
    int64_t bytes = 0;
    unsigned int xxh32 = 0;
    while (bytes &lt; 500 * 1048576) {
      xxh32 = XXH32(data.data(), size, 0);
      thread-&gt;stats.FinishedOps(nullptr, nullptr, 1, kHash);
      bytes += size;
    }
    // Print so result is not dead
    fprintf(stderr, &quot;... xxh32=0x%x\r&quot;, static_cast&lt;unsigned int&gt;(xxh32));

    thread-&gt;stats.AddBytes(bytes);
    thread-&gt;stats.AddMessage(label);
  }

  void AcquireLoad(ThreadState* thread) {
    int dummy;
    std::atomic&lt;void*&gt; ap(&amp;dummy);
    int count = 0;
    void *ptr = nullptr;
    thread-&gt;stats.AddMessage(&quot;(each op is 1000 loads)&quot;);
    while (count &lt; 100000) {
      for (int i = 0; i &lt; 1000; i++) {
        ptr = ap.load(std::memory_order_acquire);
      }
      count++;
      thread-&gt;stats.FinishedOps(nullptr, nullptr, 1, kOthers);
    }
    if (ptr == nullptr) exit(1);  // Disable unused variable warning.
  }

  void Compress(ThreadState *thread) {
    RandomGenerator gen;
    Slice input = gen.Generate(FLAGS_block_size);
    int64_t bytes = 0;
    int64_t produced = 0;
    bool ok = true;
    std::string compressed;
    CompressionOptions opts;
    CompressionContext context(FLAGS_compression_type_e);
    CompressionInfo info(opts, context, CompressionDict::GetEmptyDict(),
                         FLAGS_compression_type_e,
                         FLAGS_sample_for_compression);
    // Compress 1G
    while (ok &amp;&amp; bytes &lt; int64_t(1) &lt;&lt; 30) {
      compressed.clear();
      ok = CompressSlice(info, input, &amp;compressed);
      produced += compressed.size();
      bytes += input.size();
      thread-&gt;stats.FinishedOps(nullptr, nullptr, 1, kCompress);
    }

    if (!ok) {
      thread-&gt;stats.AddMessage(&quot;(compression failure)&quot;);
    } else {
      char buf[340];
      snprintf(buf, sizeof(buf), &quot;(output: %.1f%%)&quot;,
               (produced * 100.0) / bytes);
      thread-&gt;stats.AddMessage(buf);
      thread-&gt;stats.AddBytes(bytes);
    }
  }

  void Uncompress(ThreadState *thread) {
    RandomGenerator gen;
    Slice input = gen.Generate(FLAGS_block_size);
    std::string compressed;

    CompressionContext compression_ctx(FLAGS_compression_type_e);
    CompressionOptions compression_opts;
    CompressionInfo compression_info(
        compression_opts, compression_ctx, CompressionDict::GetEmptyDict(),
        FLAGS_compression_type_e, FLAGS_sample_for_compression);
    UncompressionContext uncompression_ctx(FLAGS_compression_type_e);
    UncompressionInfo uncompression_info(uncompression_ctx,
                                         UncompressionDict::GetEmptyDict(),
                                         FLAGS_compression_type_e);

    bool ok = CompressSlice(compression_info, input, &amp;compressed);
    int64_t bytes = 0;
    int decompress_size;
    while (ok &amp;&amp; bytes &lt; 1024 * 1048576) {
      CacheAllocationPtr uncompressed;
      switch (FLAGS_compression_type_e) {
        case ROCKSDB_NAMESPACE::kSnappyCompression: {
          // get size and allocate here to make comparison fair
          size_t ulength = 0;
          if (!Snappy_GetUncompressedLength(compressed.data(),
                                            compressed.size(), &amp;ulength)) {
            ok = false;
            break;
          }
          uncompressed = AllocateBlock(ulength, nullptr);
          ok = Snappy_Uncompress(compressed.data(), compressed.size(),
                                 uncompressed.get());
          break;
        }
        case ROCKSDB_NAMESPACE::kZlibCompression:
          uncompressed =
              Zlib_Uncompress(uncompression_info, compressed.data(),
                              compressed.size(), &amp;decompress_size, 2);
          ok = uncompressed.get() != nullptr;
          break;
        case ROCKSDB_NAMESPACE::kBZip2Compression:
          uncompressed = BZip2_Uncompress(compressed.data(), compressed.size(),
                                          &amp;decompress_size, 2);
          ok = uncompressed.get() != nullptr;
          break;
        case ROCKSDB_NAMESPACE::kLZ4Compression:
          uncompressed = LZ4_Uncompress(uncompression_info, compressed.data(),
                                        compressed.size(), &amp;decompress_size, 2);
          ok = uncompressed.get() != nullptr;
          break;
        case ROCKSDB_NAMESPACE::kLZ4HCCompression:
          uncompressed = LZ4_Uncompress(uncompression_info, compressed.data(),
                                        compressed.size(), &amp;decompress_size, 2);
          ok = uncompressed.get() != nullptr;
          break;
        case ROCKSDB_NAMESPACE::kXpressCompression:
          uncompressed.reset(XPRESS_Uncompress(
              compressed.data(), compressed.size(), &amp;decompress_size));
          ok = uncompressed.get() != nullptr;
          break;
        case ROCKSDB_NAMESPACE::kZSTD:
          uncompressed = ZSTD_Uncompress(uncompression_info, compressed.data(),
                                         compressed.size(), &amp;decompress_size);
          ok = uncompressed.get() != nullptr;
          break;
        default:
          ok = false;
      }
      bytes += input.size();
      thread-&gt;stats.FinishedOps(nullptr, nullptr, 1, kUncompress);
    }

    if (!ok) {
      thread-&gt;stats.AddMessage(&quot;(compression failure)&quot;);
    } else {
      thread-&gt;stats.AddBytes(bytes);
    }
  }

  // Returns true if the options is initialized from the specified
  // options file.
  bool InitializeOptionsFromFile(Options* opts) {
#ifndef ROCKSDB_LITE
    printf(&quot;Initializing RocksDB Options from the specified file\n&quot;);
    DBOptions db_opts;
    std::vector&lt;ColumnFamilyDescriptor&gt; cf_descs;
    if (FLAGS_options_file != &quot;&quot;) {
      auto s = LoadOptionsFromFile(FLAGS_options_file, FLAGS_env, &amp;db_opts,
                                   &amp;cf_descs);
      db_opts.env = FLAGS_env;
      if (s.ok()) {
        *opts = Options(db_opts, cf_descs[0].options);
        return true;
      }
      fprintf(stderr, &quot;Unable to load options file %s --- %s\n&quot;,
              FLAGS_options_file.c_str(), s.ToString().c_str());
      exit(1);
    }
#else
    (void)opts;
#endif
    return false;
  }

  void InitializeOptionsFromFlags(Options* opts) {
    printf(&quot;Initializing RocksDB Options from command-line flags\n&quot;);
    Options&amp; options = *opts;

    assert(db_.db == nullptr);

    options.env = FLAGS_env;
    options.max_open_files = FLAGS_open_files;
    if (FLAGS_cost_write_buffer_to_cache || FLAGS_db_write_buffer_size != 0) {
      options.write_buffer_manager.reset(
          new WriteBufferManager(FLAGS_db_write_buffer_size, cache_));
    }
    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;
    options.max_write_buffer_number_to_maintain =
        FLAGS_max_write_buffer_number_to_maintain;
    options.max_write_buffer_size_to_maintain =
        FLAGS_max_write_buffer_size_to_maintain;
    options.max_background_jobs = FLAGS_max_background_jobs;
    options.max_background_compactions = FLAGS_max_background_compactions;
    options.max_subcompactions = static_cast&lt;uint32_t&gt;(FLAGS_subcompactions);
    options.max_background_flushes = FLAGS_max_background_flushes;
    options.compaction_style = FLAGS_compaction_style_e;
    options.compaction_pri = FLAGS_compaction_pri_e;
    options.allow_mmap_reads = FLAGS_mmap_read;
    options.allow_mmap_writes = FLAGS_mmap_write;
    options.use_direct_reads = FLAGS_use_direct_reads;
    options.use_direct_io_for_flush_and_compaction =
        FLAGS_use_direct_io_for_flush_and_compaction;
#ifndef ROCKSDB_LITE
    options.ttl = FLAGS_fifo_compaction_ttl;
    options.compaction_options_fifo = CompactionOptionsFIFO(
        FLAGS_fifo_compaction_max_table_files_size_mb * 1024 * 1024,
        FLAGS_fifo_compaction_allow_compaction);
#endif  // ROCKSDB_LITE
    if (FLAGS_prefix_size != 0) {
      options.prefix_extractor.reset(
          NewFixedPrefixTransform(FLAGS_prefix_size));
    }
    if (FLAGS_use_uint64_comparator) {
      options.comparator = test::Uint64Comparator();
      if (FLAGS_key_size != 8) {
        fprintf(stderr, &quot;Using Uint64 comparator but key size is not 8.\n&quot;);
        exit(1);
      }
    }
    if (FLAGS_use_stderr_info_logger) {
      options.info_log.reset(new StderrLogger());
    }
    options.memtable_huge_page_size = FLAGS_memtable_use_huge_page ? 2048 : 0;
    options.memtable_prefix_bloom_size_ratio = FLAGS_memtable_bloom_size_ratio;
    options.memtable_whole_key_filtering = FLAGS_memtable_whole_key_filtering;
    if (FLAGS_memtable_insert_with_hint_prefix_size &gt; 0) {
      options.memtable_insert_with_hint_prefix_extractor.reset(
          NewCappedPrefixTransform(
              FLAGS_memtable_insert_with_hint_prefix_size));
    }
    options.bloom_locality = FLAGS_bloom_locality;
    options.max_file_opening_threads = FLAGS_file_opening_threads;
    options.new_table_reader_for_compaction_inputs =
        FLAGS_new_table_reader_for_compaction_inputs;
    options.compaction_readahead_size = FLAGS_compaction_readahead_size;
    options.log_readahead_size = FLAGS_log_readahead_size;
    options.random_access_max_buffer_size = FLAGS_random_access_max_buffer_size;
    options.writable_file_max_buffer_size = FLAGS_writable_file_max_buffer_size;
    options.use_fsync = FLAGS_use_fsync;
    options.num_levels = FLAGS_num_levels;
    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.level_compaction_dynamic_level_bytes =
        FLAGS_level_compaction_dynamic_level_bytes;
    options.max_bytes_for_level_multiplier =
        FLAGS_max_bytes_for_level_multiplier;
    if ((FLAGS_prefix_size == 0) &amp;&amp; (FLAGS_rep_factory == kPrefixHash ||
                                     FLAGS_rep_factory == kHashLinkedList)) {
      fprintf(stderr, &quot;prefix_size should be non-zero if PrefixHash or &quot;
                      &quot;HashLinkedList memtablerep is used\n&quot;);
      exit(1);
    }
    switch (FLAGS_rep_factory) {
      case kSkipList:
        options.memtable_factory.reset(new SkipListFactory(
            FLAGS_skip_list_lookahead));
        break;
#ifndef ROCKSDB_LITE
      case kPrefixHash:
        options.memtable_factory.reset(
            NewHashSkipListRepFactory(FLAGS_hash_bucket_count));
        break;
      case kHashLinkedList:
        options.memtable_factory.reset(NewHashLinkListRepFactory(
            FLAGS_hash_bucket_count));
        break;
      case kVectorRep:
        options.memtable_factory.reset(
          new VectorRepFactory
        );
        break;
#else
      default:
        fprintf(stderr, &quot;Only skip list is supported in lite mode\n&quot;);
        exit(1);
#endif  // ROCKSDB_LITE
    }
    if (FLAGS_use_plain_table) {
#ifndef ROCKSDB_LITE
      if (FLAGS_rep_factory != kPrefixHash &amp;&amp;
          FLAGS_rep_factory != kHashLinkedList) {
        fprintf(stderr, &quot;Waring: plain table is used with skipList\n&quot;);
      }

      int bloom_bits_per_key = FLAGS_bloom_bits;
      if (bloom_bits_per_key &lt; 0) {
        bloom_bits_per_key = 0;
      }

      PlainTableOptions plain_table_options;
      plain_table_options.user_key_len = FLAGS_key_size;
      plain_table_options.bloom_bits_per_key = bloom_bits_per_key;
      plain_table_options.hash_table_ratio = 0.75;
      options.table_factory = std::shared_ptr&lt;TableFactory&gt;(
          NewPlainTableFactory(plain_table_options));
#else
      fprintf(stderr, &quot;Plain table is not supported in lite mode\n&quot;);
      exit(1);
#endif  // ROCKSDB_LITE
    } else if (FLAGS_use_cuckoo_table) {
#ifndef ROCKSDB_LITE
      if (FLAGS_cuckoo_hash_ratio &gt; 1 || FLAGS_cuckoo_hash_ratio &lt; 0) {
        fprintf(stderr, &quot;Invalid cuckoo_hash_ratio\n&quot;);
        exit(1);
      }

      if (!FLAGS_mmap_read) {
        fprintf(stderr, &quot;cuckoo table format requires mmap read to operate\n&quot;);
        exit(1);
      }

      ROCKSDB_NAMESPACE::CuckooTableOptions table_options;
      table_options.hash_table_ratio = FLAGS_cuckoo_hash_ratio;
      table_options.identity_as_first_hash = FLAGS_identity_as_first_hash;
      options.table_factory = std::shared_ptr&lt;TableFactory&gt;(
          NewCuckooTableFactory(table_options));
#else
      fprintf(stderr, &quot;Cuckoo table is not supported in lite mode\n&quot;);
      exit(1);
#endif  // ROCKSDB_LITE
    } else {
      BlockBasedTableOptions block_based_options;
      if (FLAGS_use_hash_search) {
        if (FLAGS_prefix_size == 0) {
          fprintf(stderr,
              &quot;prefix_size not assigned when enable use_hash_search \n&quot;);
          exit(1);
        }
        block_based_options.index_type = BlockBasedTableOptions::kHashSearch;
      } else {
        block_based_options.index_type = BlockBasedTableOptions::kBinarySearch;
      }
      if (FLAGS_partition_index_and_filters || FLAGS_partition_index) {
        if (FLAGS_use_hash_search) {
          fprintf(stderr,
                  &quot;use_hash_search is incompatible with &quot;
                  &quot;partition index and is ignored&quot;);
        }
        block_based_options.index_type =
            BlockBasedTableOptions::kTwoLevelIndexSearch;
        block_based_options.metadata_block_size = FLAGS_metadata_block_size;
        if (FLAGS_partition_index_and_filters) {
          block_based_options.partition_filters = true;
        }
      }
      if (cache_ == nullptr) {
        block_based_options.no_block_cache = true;
      }
      block_based_options.cache_index_and_filter_blocks =
          FLAGS_cache_index_and_filter_blocks;
      block_based_options.pin_l0_filter_and_index_blocks_in_cache =
          FLAGS_pin_l0_filter_and_index_blocks_in_cache;
      block_based_options.pin_top_level_index_and_filter =
          FLAGS_pin_top_level_index_and_filter;
      if (FLAGS_cache_high_pri_pool_ratio &gt; 1e-6) {  // &gt; 0.0 + eps
        block_based_options.cache_index_and_filter_blocks_with_high_priority =
            true;
      }
      block_based_options.block_cache = cache_;
      block_based_options.block_cache_compressed = compressed_cache_;
      block_based_options.block_size = FLAGS_block_size;
      block_based_options.block_restart_interval = FLAGS_block_restart_interval;
      block_based_options.index_block_restart_interval =
          FLAGS_index_block_restart_interval;
      block_based_options.filter_policy = filter_policy_;
      block_based_options.format_version =
          static_cast&lt;uint32_t&gt;(FLAGS_format_version);
      block_based_options.read_amp_bytes_per_bit = FLAGS_read_amp_bytes_per_bit;
      block_based_options.enable_index_compression =
          FLAGS_enable_index_compression;
      block_based_options.block_align = FLAGS_block_align;
      if (FLAGS_use_data_block_hash_index) {
        block_based_options.data_block_index_type =
            ROCKSDB_NAMESPACE::BlockBasedTableOptions::kDataBlockBinaryAndHash;
      } else {
        block_based_options.data_block_index_type =
            ROCKSDB_NAMESPACE::BlockBasedTableOptions::kDataBlockBinarySearch;
      }
      block_based_options.data_block_hash_table_util_ratio =
          FLAGS_data_block_hash_table_util_ratio;
      if (FLAGS_read_cache_path != &quot;&quot;) {
#ifndef ROCKSDB_LITE
        Status rc_status;

        // Read cache need to be provided with a the Logger, we will put all
        // reac cache logs in the read cache path in a file named rc_LOG
        rc_status = FLAGS_env-&gt;CreateDirIfMissing(FLAGS_read_cache_path);
        std::shared_ptr&lt;Logger&gt; read_cache_logger;
        if (rc_status.ok()) {
          rc_status = FLAGS_env-&gt;NewLogger(FLAGS_read_cache_path + &quot;/rc_LOG&quot;,
                                           &amp;read_cache_logger);
        }

        if (rc_status.ok()) {
          PersistentCacheConfig rc_cfg(FLAGS_env, FLAGS_read_cache_path,
                                       FLAGS_read_cache_size,
                                       read_cache_logger);

          rc_cfg.enable_direct_reads = FLAGS_read_cache_direct_read;
          rc_cfg.enable_direct_writes = FLAGS_read_cache_direct_write;
          rc_cfg.writer_qdepth = 4;
          rc_cfg.writer_dispatch_size = 4 * 1024;

          auto pcache = std::make_shared&lt;BlockCacheTier&gt;(rc_cfg);
          block_based_options.persistent_cache = pcache;
          rc_status = pcache-&gt;Open();
        }

        if (!rc_status.ok()) {
          fprintf(stderr, &quot;Error initializing read cache, %s\n&quot;,
                  rc_status.ToString().c_str());
          exit(1);
        }
#else
        fprintf(stderr, &quot;Read cache is not supported in LITE\n&quot;);
        exit(1);

#endif
      }
      options.table_factory.reset(
          NewBlockBasedTableFactory(block_based_options));
    }
    if (FLAGS_max_bytes_for_level_multiplier_additional_v.size() &gt; 0) {
      if (FLAGS_max_bytes_for_level_multiplier_additional_v.size() !=
          static_cast&lt;unsigned int&gt;(FLAGS_num_levels)) {
        fprintf(stderr, &quot;Insufficient number of fanouts specified %d\n&quot;,
                static_cast&lt;int&gt;(
                    FLAGS_max_bytes_for_level_multiplier_additional_v.size()));
        exit(1);
      }
      options.max_bytes_for_level_multiplier_additional =
        FLAGS_max_bytes_for_level_multiplier_additional_v;
    }
    options.level0_stop_writes_trigger = FLAGS_level0_stop_writes_trigger;
    options.level0_file_num_compaction_trigger =
        FLAGS_level0_file_num_compaction_trigger;
    options.level0_slowdown_writes_trigger =
      FLAGS_level0_slowdown_writes_trigger;
    options.compression = FLAGS_compression_type_e;
    options.sample_for_compression = FLAGS_sample_for_compression;
    options.WAL_ttl_seconds = FLAGS_wal_ttl_seconds;
    options.WAL_size_limit_MB = FLAGS_wal_size_limit_MB;
    options.max_total_wal_size = FLAGS_max_total_wal_size;

    if (FLAGS_min_level_to_compress &gt;= 0) {
      assert(FLAGS_min_level_to_compress &lt;= FLAGS_num_levels);
      options.compression_per_level.resize(FLAGS_num_levels);
      for (int i = 0; i &lt; FLAGS_min_level_to_compress; i++) {
        options.compression_per_level[i] = kNoCompression;
      }
      for (int i = FLAGS_min_level_to_compress;
           i &lt; FLAGS_num_levels; i++) {
        options.compression_per_level[i] = FLAGS_compression_type_e;
      }
    }
    options.soft_rate_limit = FLAGS_soft_rate_limit;
    options.hard_rate_limit = FLAGS_hard_rate_limit;
    options.soft_pending_compaction_bytes_limit =
        FLAGS_soft_pending_compaction_bytes_limit;
    options.hard_pending_compaction_bytes_limit =
        FLAGS_hard_pending_compaction_bytes_limit;
    options.delayed_write_rate = FLAGS_delayed_write_rate;
    options.allow_concurrent_memtable_write =
        FLAGS_allow_concurrent_memtable_write;
    options.inplace_update_support = FLAGS_inplace_update_support;
    options.inplace_update_num_locks = FLAGS_inplace_update_num_locks;
    options.enable_write_thread_adaptive_yield =
        FLAGS_enable_write_thread_adaptive_yield;
    options.enable_pipelined_write = FLAGS_enable_pipelined_write;
    options.unordered_write = FLAGS_unordered_write;
    options.write_thread_max_yield_usec = FLAGS_write_thread_max_yield_usec;
    options.write_thread_slow_yield_usec = FLAGS_write_thread_slow_yield_usec;
    options.rate_limit_delay_max_milliseconds =
      FLAGS_rate_limit_delay_max_milliseconds;
    options.table_cache_numshardbits = FLAGS_table_cache_numshardbits;
    options.max_compaction_bytes = FLAGS_max_compaction_bytes;
    options.disable_auto_compactions = FLAGS_disable_auto_compactions;
    options.optimize_filters_for_hits = FLAGS_optimize_filters_for_hits;

    // fill storage options
    options.advise_random_on_open = FLAGS_advise_random_on_open;
    options.access_hint_on_compaction_start = FLAGS_compaction_fadvice_e;
    options.use_adaptive_mutex = FLAGS_use_adaptive_mutex;
    options.bytes_per_sync = FLAGS_bytes_per_sync;
    options.wal_bytes_per_sync = FLAGS_wal_bytes_per_sync;

    // merge operator options
    options.merge_operator = MergeOperators::CreateFromStringId(
        FLAGS_merge_operator);
    if (options.merge_operator == nullptr &amp;&amp; !FLAGS_merge_operator.empty()) {
      fprintf(stderr, &quot;invalid merge operator: %s\n&quot;,
              FLAGS_merge_operator.c_str());
      exit(1);
    }
    options.max_successive_merges = FLAGS_max_successive_merges;
    options.report_bg_io_stats = FLAGS_report_bg_io_stats;

    // set universal style compaction configurations, if applicable
    if (FLAGS_universal_size_ratio != 0) {
      options.compaction_options_universal.size_ratio =
        FLAGS_universal_size_ratio;
    }
    if (FLAGS_universal_min_merge_width != 0) {
      options.compaction_options_universal.min_merge_width =
        FLAGS_universal_min_merge_width;
    }
    if (FLAGS_universal_max_merge_width != 0) {
      options.compaction_options_universal.max_merge_width =
        FLAGS_universal_max_merge_width;
    }
    if (FLAGS_universal_max_size_amplification_percent != 0) {
      options.compaction_options_universal.max_size_amplification_percent =
        FLAGS_universal_max_size_amplification_percent;
    }
    if (FLAGS_universal_compression_size_percent != -1) {
      options.compaction_options_universal.compression_size_percent =
        FLAGS_universal_compression_size_percent;
    }
    options.compaction_options_universal.allow_trivial_move =
        FLAGS_universal_allow_trivial_move;
    if (FLAGS_thread_status_per_interval &gt; 0) {
      options.enable_thread_tracking = true;
    }

#ifndef ROCKSDB_LITE
    if (FLAGS_readonly &amp;&amp; FLAGS_transaction_db) {
      fprintf(stderr, &quot;Cannot use readonly flag with transaction_db\n&quot;);
      exit(1);
    }
    if (FLAGS_use_secondary_db &amp;&amp;
        (FLAGS_transaction_db || FLAGS_optimistic_transaction_db)) {
      fprintf(stderr, &quot;Cannot use use_secondary_db flag with transaction_db\n&quot;);
      exit(1);
    }
#endif  // ROCKSDB_LITE

  }

  void InitializeOptionsGeneral(Options* opts) {
    Options&amp; options = *opts;

    options.create_missing_column_families = FLAGS_num_column_families &gt; 1;
    options.statistics = dbstats;
    options.wal_dir = FLAGS_wal_dir;
    options.create_if_missing = !FLAGS_use_existing_db;
    options.dump_malloc_stats = FLAGS_dump_malloc_stats;
    options.stats_dump_period_sec =
        static_cast&lt;unsigned int&gt;(FLAGS_stats_dump_period_sec);
    options.stats_persist_period_sec =
        static_cast&lt;unsigned int&gt;(FLAGS_stats_persist_period_sec);
    options.persist_stats_to_disk = FLAGS_persist_stats_to_disk;
    options.stats_history_buffer_size =
        static_cast&lt;size_t&gt;(FLAGS_stats_history_buffer_size);

    options.compression_opts.level = FLAGS_compression_level;
    options.compression_opts.max_dict_bytes = FLAGS_compression_max_dict_bytes;
    options.compression_opts.zstd_max_train_bytes =
        FLAGS_compression_zstd_max_train_bytes;
    // If this is a block based table, set some related options
    if (options.table_factory-&gt;Name() == BlockBasedTableFactory::kName &amp;&amp;
        options.table_factory-&gt;GetOptions() != nullptr) {
      BlockBasedTableOptions* table_options =
          reinterpret_cast&lt;BlockBasedTableOptions*&gt;(
              options.table_factory-&gt;GetOptions());
      if (FLAGS_cache_size) {
        table_options-&gt;block_cache = cache_;
      }
      if (FLAGS_bloom_bits &gt;= 0) {
        table_options-&gt;filter_policy.reset(NewBloomFilterPolicy(
            FLAGS_bloom_bits, FLAGS_use_block_based_filter));
      }
    }
    if (FLAGS_row_cache_size) {
      if (FLAGS_cache_numshardbits &gt;= 1) {
        options.row_cache =
            NewLRUCache(FLAGS_row_cache_size, FLAGS_cache_numshardbits);
      } else {
        options.row_cache = NewLRUCache(FLAGS_row_cache_size);
      }
    }
    if (FLAGS_enable_io_prio) {
      FLAGS_env-&gt;LowerThreadPoolIOPriority(Env::LOW);
      FLAGS_env-&gt;LowerThreadPoolIOPriority(Env::HIGH);
    }
    if (FLAGS_enable_cpu_prio) {
      FLAGS_env-&gt;LowerThreadPoolCPUPriority(Env::LOW);
      FLAGS_env-&gt;LowerThreadPoolCPUPriority(Env::HIGH);
    }
    options.env = FLAGS_env;
    if (FLAGS_sine_write_rate) {
      FLAGS_benchmark_write_rate_limit = static_cast&lt;uint64_t&gt;(SineRate(0));
    }

    if (FLAGS_rate_limiter_bytes_per_sec &gt; 0) {
      if (FLAGS_rate_limit_bg_reads &amp;&amp;
          !FLAGS_new_table_reader_for_compaction_inputs) {
        fprintf(stderr,
                &quot;rate limit compaction reads must have &quot;
                &quot;new_table_reader_for_compaction_inputs set\n&quot;);
        exit(1);
      }
      options.rate_limiter.reset(NewGenericRateLimiter(
          FLAGS_rate_limiter_bytes_per_sec, 100 * 1000 /* refill_period_us */,
          10 /* fairness */,
          FLAGS_rate_limit_bg_reads ? RateLimiter::Mode::kReadsOnly
                                    : RateLimiter::Mode::kWritesOnly,
          FLAGS_rate_limiter_auto_tuned));
    }

    options.listeners.emplace_back(listener_);
    if (FLAGS_num_multi_db &lt;= 1) {
      OpenDb(options, FLAGS_db, &amp;db_);
    } else {
      multi_dbs_.clear();
      multi_dbs_.resize(FLAGS_num_multi_db);
      auto wal_dir = options.wal_dir;
      for (int i = 0; i &lt; FLAGS_num_multi_db; i++) {
        if (!wal_dir.empty()) {
          options.wal_dir = GetPathForMultiple(wal_dir, i);
        }
        OpenDb(options, GetPathForMultiple(FLAGS_db, i), &amp;multi_dbs_[i]);
      }
      options.wal_dir = wal_dir;
    }

    // KeepFilter is a noop filter, this can be used to test compaction filter
    if (FLAGS_use_keep_filter) {
      options.compaction_filter = new KeepFilter();
      fprintf(stdout, &quot;A noop compaction filter is used\n&quot;);
    }

    if (FLAGS_use_existing_keys) {
      // Only work on single database
      assert(db_.db != nullptr);
      ReadOptions read_opts;
      read_opts.total_order_seek = true;
      Iterator* iter = db_.db-&gt;NewIterator(read_opts);
      for (iter-&gt;SeekToFirst(); iter-&gt;Valid(); iter-&gt;Next()) {
        keys_.emplace_back(iter-&gt;key().ToString());
      }
      delete iter;
      FLAGS_num = keys_.size();
    }
  }

  void Open(Options* opts) {
    if (!InitializeOptionsFromFile(opts)) {
      InitializeOptionsFromFlags(opts);
    }

    InitializeOptionsGeneral(opts);
  }

  void OpenDb(Options options, const std::string&amp; db_name,
      DBWithColumnFamilies* db) {
    Status s;
    // Open with column families if necessary.
    if (FLAGS_num_column_families &gt; 1) {
      size_t num_hot = FLAGS_num_column_families;
      if (FLAGS_num_hot_column_families &gt; 0 &amp;&amp;
          FLAGS_num_hot_column_families &lt; FLAGS_num_column_families) {
        num_hot = FLAGS_num_hot_column_families;
      } else {
        FLAGS_num_hot_column_families = FLAGS_num_column_families;
      }
      std::vector&lt;ColumnFamilyDescriptor&gt; column_families;
      for (size_t i = 0; i &lt; num_hot; i++) {
        column_families.push_back(ColumnFamilyDescriptor(
              ColumnFamilyName(i), ColumnFamilyOptions(options)));
      }
      std::vector&lt;int&gt; cfh_idx_to_prob;
      if (!FLAGS_column_family_distribution.empty()) {
        std::stringstream cf_prob_stream(FLAGS_column_family_distribution);
        std::string cf_prob;
        int sum = 0;
        while (std::getline(cf_prob_stream, cf_prob, &#039;,&#039;)) {
          cfh_idx_to_prob.push_back(std::stoi(cf_prob));
          sum += cfh_idx_to_prob.back();
        }
        if (sum != 100) {
          fprintf(stderr, &quot;column_family_distribution items must sum to 100\n&quot;);
          exit(1);
        }
        if (cfh_idx_to_prob.size() != num_hot) {
          fprintf(stderr,
                  &quot;got %&quot; ROCKSDB_PRIszt
                  &quot; column_family_distribution items; expected &quot;
                  &quot;%&quot; ROCKSDB_PRIszt &quot;\n&quot;,
                  cfh_idx_to_prob.size(), num_hot);
          exit(1);
        }
      }
#ifndef ROCKSDB_LITE
      if (FLAGS_readonly) {
        s = DB::OpenForReadOnly(options, db_name, column_families,
            &amp;db-&gt;cfh, &amp;db-&gt;db);
      } else if (FLAGS_optimistic_transaction_db) {
        s = OptimisticTransactionDB::Open(options, db_name, column_families,
                                          &amp;db-&gt;cfh, &amp;db-&gt;opt_txn_db);
        if (s.ok()) {
          db-&gt;db = db-&gt;opt_txn_db-&gt;GetBaseDB();
        }
      } else if (FLAGS_transaction_db) {
        TransactionDB* ptr;
        TransactionDBOptions txn_db_options;
        if (options.unordered_write) {
          options.two_write_queues = true;
          txn_db_options.skip_concurrency_control = true;
          txn_db_options.write_policy = WRITE_PREPARED;
        }
        s = TransactionDB::Open(options, txn_db_options, db_name,
                                column_families, &amp;db-&gt;cfh, &amp;ptr);
        if (s.ok()) {
          db-&gt;db = ptr;
        }
      } else {
        s = DB::Open(options, db_name, column_families, &amp;db-&gt;cfh, &amp;db-&gt;db);
      }
#else
      s = DB::Open(options, db_name, column_families, &amp;db-&gt;cfh, &amp;db-&gt;db);
#endif  // ROCKSDB_LITE
      db-&gt;cfh.resize(FLAGS_num_column_families);
      db-&gt;num_created = num_hot;
      db-&gt;num_hot = num_hot;
      db-&gt;cfh_idx_to_prob = std::move(cfh_idx_to_prob);
#ifndef ROCKSDB_LITE
    } else if (FLAGS_readonly) {
      s = DB::OpenForReadOnly(options, db_name, &amp;db-&gt;db);
    } else if (FLAGS_optimistic_transaction_db) {
      s = OptimisticTransactionDB::Open(options, db_name, &amp;db-&gt;opt_txn_db);
      if (s.ok()) {
        db-&gt;db = db-&gt;opt_txn_db-&gt;GetBaseDB();
      }
    } else if (FLAGS_transaction_db) {
      TransactionDB* ptr = nullptr;
      TransactionDBOptions txn_db_options;
      if (options.unordered_write) {
        options.two_write_queues = true;
        txn_db_options.skip_concurrency_control = true;
        txn_db_options.write_policy = WRITE_PREPARED;
      }
      s = CreateLoggerFromOptions(db_name, options, &amp;options.info_log);
      if (s.ok()) {
        s = TransactionDB::Open(options, txn_db_options, db_name, &amp;ptr);
      }
      if (s.ok()) {
        db-&gt;db = ptr;
      }
    } else if (FLAGS_use_blob_db) {
      blob_db::BlobDBOptions blob_db_options;
      blob_db_options.enable_garbage_collection = FLAGS_blob_db_enable_gc;
      blob_db_options.garbage_collection_cutoff = FLAGS_blob_db_gc_cutoff;
      blob_db_options.is_fifo = FLAGS_blob_db_is_fifo;
      blob_db_options.max_db_size = FLAGS_blob_db_max_db_size;
      blob_db_options.ttl_range_secs = FLAGS_blob_db_ttl_range_secs;
      blob_db_options.min_blob_size = FLAGS_blob_db_min_blob_size;
      blob_db_options.bytes_per_sync = FLAGS_blob_db_bytes_per_sync;
      blob_db_options.blob_file_size = FLAGS_blob_db_file_size;
      blob_db_options.compression = FLAGS_blob_db_compression_type_e;
      blob_db::BlobDB* ptr = nullptr;
      s = blob_db::BlobDB::Open(options, blob_db_options, db_name, &amp;ptr);
      if (s.ok()) {
        db-&gt;db = ptr;
      }
    } else if (FLAGS_use_secondary_db) {
      if (FLAGS_secondary_path.empty()) {
        std::string default_secondary_path;
        FLAGS_env-&gt;GetTestDirectory(&amp;default_secondary_path);
        default_secondary_path += &quot;/dbbench_secondary&quot;;
        FLAGS_secondary_path = default_secondary_path;
      }
      s = DB::OpenAsSecondary(options, db_name, FLAGS_secondary_path, &amp;db-&gt;db);
      if (s.ok() &amp;&amp; FLAGS_secondary_update_interval &gt; 0) {
        secondary_update_thread_.reset(new port::Thread(
            [this](int interval, DBWithColumnFamilies* _db) {
              while (0 == secondary_update_stopped_.load(
                              std::memory_order_relaxed)) {
                Status secondary_update_status =
                    _db-&gt;db-&gt;TryCatchUpWithPrimary();
                if (!secondary_update_status.ok()) {
                  fprintf(stderr, &quot;Failed to catch up with primary: %s\n&quot;,
                          secondary_update_status.ToString().c_str());
                  break;
                }
                ++secondary_db_updates_;
                FLAGS_env-&gt;SleepForMicroseconds(interval * 1000000);
              }
            },
            FLAGS_secondary_update_interval, db));
      }
#endif  // ROCKSDB_LITE
    } else {
      s = DB::Open(options, db_name, &amp;db-&gt;db);
    }
    if (!s.ok()) {
      fprintf(stderr, &quot;open error: %s\n&quot;, s.ToString().c_str());
      exit(1);
    }
  }

  enum WriteMode {
    RANDOM, SEQUENTIAL, UNIQUE_RANDOM
  };

  void WriteSeqDeterministic(ThreadState* thread) {
    DoDeterministicCompact(thread, open_options_.compaction_style, SEQUENTIAL);
  }

  void WriteUniqueRandomDeterministic(ThreadState* thread) {
    DoDeterministicCompact(thread, open_options_.compaction_style,
                           UNIQUE_RANDOM);
  }

  void WriteSeq(ThreadState* thread) {
    DoWrite(thread, SEQUENTIAL);
  }

  void WriteRandom(ThreadState* thread) {
    DoWrite(thread, RANDOM);
  }

  void WriteUniqueRandom(ThreadState* thread) {
    DoWrite(thread, UNIQUE_RANDOM);
  }

  class KeyGenerator {
   public:
    KeyGenerator(Random64* rand, WriteMode mode, uint64_t num,
                 uint64_t /*num_per_set*/ = 64 * 1024)
        : rand_(rand), mode_(mode), num_(num), next_(0) {
      if (mode_ == UNIQUE_RANDOM) {
        // NOTE: if memory consumption of this approach becomes a concern,
        // we can either break it into pieces and only random shuffle a section
        // each time. Alternatively, use a bit map implementation
        // (https://reviews.facebook.net/differential/diff/54627/)
        values_.resize(num_);
        for (uint64_t i = 0; i &lt; num_; ++i) {
          values_[i] = i;
        }
        std::shuffle(
            values_.begin(), values_.end(),
            std::default_random_engine(static_cast&lt;unsigned int&gt;(FLAGS_seed)));
      }
    }

    uint64_t Next() {
      switch (mode_) {
        case SEQUENTIAL:
          return next_++;
        case RANDOM:
          return rand_-&gt;Next() % num_;
        case UNIQUE_RANDOM:
          assert(next_ &lt; num_);
          return values_[next_++];
      }
      assert(false);
      return std::numeric_limits&lt;uint64_t&gt;::max();
    }

   private:
    Random64* rand_;
    WriteMode mode_;
    const uint64_t num_;
    uint64_t next_;
    std::vector&lt;uint64_t&gt; values_;
  };

  DB* SelectDB(ThreadState* thread) {
    return SelectDBWithCfh(thread)-&gt;db;
  }

  DBWithColumnFamilies* SelectDBWithCfh(ThreadState* thread) {
    return SelectDBWithCfh(thread-&gt;rand.Next());
  }

  DBWithColumnFamilies* SelectDBWithCfh(uint64_t rand_int) {
    if (db_.db != nullptr) {
      return &amp;db_;
    } else  {
      return &amp;multi_dbs_[rand_int % multi_dbs_.size()];
    }
  }

  double SineRate(double x) {
    return FLAGS_sine_a*sin((FLAGS_sine_b*x) + FLAGS_sine_c) + FLAGS_sine_d;
  }

  void DoWrite(ThreadState* thread, WriteMode write_mode) {
    const int test_duration = write_mode == RANDOM ? FLAGS_duration : 0;
    const int64_t num_ops = writes_ == 0 ? num_ : writes_;

    size_t num_key_gens = 1;
    if (db_.db == nullptr) {
      num_key_gens = multi_dbs_.size();
    }
    std::vector&lt;std::unique_ptr&lt;KeyGenerator&gt;&gt; key_gens(num_key_gens);
    int64_t max_ops = num_ops * num_key_gens;
    int64_t ops_per_stage = max_ops;
    if (FLAGS_num_column_families &gt; 1 &amp;&amp; FLAGS_num_hot_column_families &gt; 0) {
      ops_per_stage = (max_ops - 1) / (FLAGS_num_column_families /
                                       FLAGS_num_hot_column_families) +
                      1;
    }

    Duration duration(test_duration, max_ops, ops_per_stage);
    for (size_t i = 0; i &lt; num_key_gens; i++) {
      key_gens[i].reset(new KeyGenerator(&amp;(thread-&gt;rand), write_mode,
                                         num_ + max_num_range_tombstones_,
                                         ops_per_stage));
    }

    if (num_ != FLAGS_num) {
      char msg[100];
      snprintf(msg, sizeof(msg), &quot;(%&quot; PRIu64 &quot; ops)&quot;, num_);
      thread-&gt;stats.AddMessage(msg);
    }

    RandomGenerator gen;
    WriteBatch batch;
    Status s;
    int64_t bytes = 0;

    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    std::unique_ptr&lt;const char[]&gt; begin_key_guard;
    Slice begin_key = AllocateKey(&amp;begin_key_guard);
    std::unique_ptr&lt;const char[]&gt; end_key_guard;
    Slice end_key = AllocateKey(&amp;end_key_guard);
    std::vector&lt;std::unique_ptr&lt;const char[]&gt;&gt; expanded_key_guards;
    std::vector&lt;Slice&gt; expanded_keys;
    if (FLAGS_expand_range_tombstones) {
      expanded_key_guards.resize(range_tombstone_width_);
      for (auto&amp; expanded_key_guard : expanded_key_guards) {
        expanded_keys.emplace_back(AllocateKey(&amp;expanded_key_guard));
      }
    }

    int64_t stage = 0;
    int64_t num_written = 0;
    while (!duration.Done(entries_per_batch_)) {
      if (duration.GetStage() != stage) {
        stage = duration.GetStage();
        if (db_.db != nullptr) {
          db_.CreateNewCf(open_options_, stage);
        } else {
          for (auto&amp; db : multi_dbs_) {
            db.CreateNewCf(open_options_, stage);
          }
        }
      }

      size_t id = thread-&gt;rand.Next() % num_key_gens;
      DBWithColumnFamilies* db_with_cfh = SelectDBWithCfh(id);
      batch.Clear();
      int64_t batch_bytes = 0;

      for (int64_t j = 0; j &lt; entries_per_batch_; j++) {
        int64_t rand_num = key_gens[id]-&gt;Next();
        GenerateKeyFromInt(rand_num, FLAGS_num, &amp;key);
        Slice val = gen.Generate();
        if (use_blob_db_) {
#ifndef ROCKSDB_LITE
          blob_db::BlobDB* blobdb =
              static_cast&lt;blob_db::BlobDB*&gt;(db_with_cfh-&gt;db);
          if (FLAGS_blob_db_max_ttl_range &gt; 0) {
            int ttl = rand() % FLAGS_blob_db_max_ttl_range;
            s = blobdb-&gt;PutWithTTL(write_options_, key, val, ttl);
          } else {
            s = blobdb-&gt;Put(write_options_, key, val);
          }
#endif  //  ROCKSDB_LITE
        } else if (FLAGS_num_column_families &lt;= 1) {
          batch.Put(key, val);
        } else {
          // We use same rand_num as seed for key and column family so that we
          // can deterministically find the cfh corresponding to a particular
          // key while reading the key.
          batch.Put(db_with_cfh-&gt;GetCfh(rand_num), key,
                    val);
        }
        batch_bytes += val.size() + key_size_;
        bytes += val.size() + key_size_;
        ++num_written;
        if (writes_per_range_tombstone_ &gt; 0 &amp;&amp;
            num_written &gt; writes_before_delete_range_ &amp;&amp;
            (num_written - writes_before_delete_range_) /
                    writes_per_range_tombstone_ &lt;=
                max_num_range_tombstones_ &amp;&amp;
            (num_written - writes_before_delete_range_) %
                    writes_per_range_tombstone_ ==
                0) {
          int64_t begin_num = key_gens[id]-&gt;Next();
          if (FLAGS_expand_range_tombstones) {
            for (int64_t offset = 0; offset &lt; range_tombstone_width_;
                 ++offset) {
              GenerateKeyFromInt(begin_num + offset, FLAGS_num,
                                 &amp;expanded_keys[offset]);
              if (use_blob_db_) {
#ifndef ROCKSDB_LITE
                s = db_with_cfh-&gt;db-&gt;Delete(write_options_,
                                            expanded_keys[offset]);
#endif  //  ROCKSDB_LITE
              } else if (FLAGS_num_column_families &lt;= 1) {
                batch.Delete(expanded_keys[offset]);
              } else {
                batch.Delete(db_with_cfh-&gt;GetCfh(rand_num),
                             expanded_keys[offset]);
              }
            }
          } else {
            GenerateKeyFromInt(begin_num, FLAGS_num, &amp;begin_key);
            GenerateKeyFromInt(begin_num + range_tombstone_width_, FLAGS_num,
                               &amp;end_key);
            if (use_blob_db_) {
#ifndef ROCKSDB_LITE
              s = db_with_cfh-&gt;db-&gt;DeleteRange(
                  write_options_, db_with_cfh-&gt;db-&gt;DefaultColumnFamily(),
                  begin_key, end_key);
#endif  //  ROCKSDB_LITE
            } else if (FLAGS_num_column_families &lt;= 1) {
              batch.DeleteRange(begin_key, end_key);
            } else {
              batch.DeleteRange(db_with_cfh-&gt;GetCfh(rand_num), begin_key,
                                end_key);
            }
          }
        }
      }
      if (thread-&gt;shared-&gt;write_rate_limiter.get() != nullptr) {
        thread-&gt;shared-&gt;write_rate_limiter-&gt;Request(
            batch_bytes, Env::IO_HIGH,
            nullptr /* stats */, RateLimiter::OpType::kWrite);
        // Set time at which last op finished to Now() to hide latency and
        // sleep from rate limiter. Also, do the check once per batch, not
        // once per write.
        thread-&gt;stats.ResetLastOpTime();
      }
      if (!use_blob_db_) {
        s = db_with_cfh-&gt;db-&gt;Write(write_options_, &amp;batch);
      }
      thread-&gt;stats.FinishedOps(db_with_cfh, db_with_cfh-&gt;db,
                                entries_per_batch_, kWrite);
      if (FLAGS_sine_write_rate) {
        uint64_t now = FLAGS_env-&gt;NowMicros();

        uint64_t usecs_since_last;
        if (now &gt; thread-&gt;stats.GetSineInterval()) {
          usecs_since_last = now - thread-&gt;stats.GetSineInterval();
        } else {
          usecs_since_last = 0;
        }

        if (usecs_since_last &gt;
            (FLAGS_sine_write_rate_interval_milliseconds * uint64_t{1000})) {
          double usecs_since_start =
                  static_cast&lt;double&gt;(now - thread-&gt;stats.GetStart());
          thread-&gt;stats.ResetSineInterval();
          uint64_t write_rate =
                  static_cast&lt;uint64_t&gt;(SineRate(usecs_since_start / 1000000.0));
          thread-&gt;shared-&gt;write_rate_limiter.reset(
                  NewGenericRateLimiter(write_rate));
        }
      }
      if (!s.ok()) {
        s = listener_-&gt;WaitForRecovery(600000000) ? Status::OK() : s;
      }

      if (!s.ok()) {
        fprintf(stderr, &quot;put error: %s\n&quot;, s.ToString().c_str());
        exit(1);
      }
    }
    thread-&gt;stats.AddBytes(bytes);
  }

  Status DoDeterministicCompact(ThreadState* thread,
                                CompactionStyle compaction_style,
                                WriteMode write_mode) {
#ifndef ROCKSDB_LITE
    ColumnFamilyMetaData meta;
    std::vector&lt;DB*&gt; db_list;
    if (db_.db != nullptr) {
      db_list.push_back(db_.db);
    } else {
      for (auto&amp; db : multi_dbs_) {
        db_list.push_back(db.db);
      }
    }
    std::vector&lt;Options&gt; options_list;
    for (auto db : db_list) {
      options_list.push_back(db-&gt;GetOptions());
      if (compaction_style != kCompactionStyleFIFO) {
        db-&gt;SetOptions({{&quot;disable_auto_compactions&quot;, &quot;1&quot;},
                        {&quot;level0_slowdown_writes_trigger&quot;, &quot;400000000&quot;},
                        {&quot;level0_stop_writes_trigger&quot;, &quot;400000000&quot;}});
      } else {
        db-&gt;SetOptions({{&quot;disable_auto_compactions&quot;, &quot;1&quot;}});
      }
    }

    assert(!db_list.empty());
    auto num_db = db_list.size();
    size_t num_levels = static_cast&lt;size_t&gt;(open_options_.num_levels);
    size_t output_level = open_options_.num_levels - 1;
    std::vector&lt;std::vector&lt;std::vector&lt;SstFileMetaData&gt;&gt;&gt; sorted_runs(num_db);
    std::vector&lt;size_t&gt; num_files_at_level0(num_db, 0);
    if (compaction_style == kCompactionStyleLevel) {
      if (num_levels == 0) {
        return Status::InvalidArgument(&quot;num_levels should be larger than 1&quot;);
      }
      bool should_stop = false;
      while (!should_stop) {
        if (sorted_runs[0].empty()) {
          DoWrite(thread, write_mode);
        } else {
          DoWrite(thread, UNIQUE_RANDOM);
        }
        for (size_t i = 0; i &lt; num_db; i++) {
          auto db = db_list[i];
          db-&gt;Flush(FlushOptions());
          db-&gt;GetColumnFamilyMetaData(&amp;meta);
          if (num_files_at_level0[i] == meta.levels[0].files.size() ||
              writes_ == 0) {
            should_stop = true;
            continue;
          }
          sorted_runs[i].emplace_back(
              meta.levels[0].files.begin(),
              meta.levels[0].files.end() - num_files_at_level0[i]);
          num_files_at_level0[i] = meta.levels[0].files.size();
          if (sorted_runs[i].back().size() == 1) {
            should_stop = true;
            continue;
          }
          if (sorted_runs[i].size() == output_level) {
            auto&amp; L1 = sorted_runs[i].back();
            L1.erase(L1.begin(), L1.begin() + L1.size() / 3);
            should_stop = true;
            continue;
          }
        }
        writes_ /= static_cast&lt;int64_t&gt;(open_options_.max_bytes_for_level_multiplier);
      }
      for (size_t i = 0; i &lt; num_db; i++) {
        if (sorted_runs[i].size() &lt; num_levels - 1) {
          fprintf(stderr, &quot;n is too small to fill %&quot; ROCKSDB_PRIszt &quot; levels\n&quot;, num_levels);
          exit(1);
        }
      }
      for (size_t i = 0; i &lt; num_db; i++) {
        auto db = db_list[i];
        auto compactionOptions = CompactionOptions();
        compactionOptions.compression = FLAGS_compression_type_e;
        auto options = db-&gt;GetOptions();
        MutableCFOptions mutable_cf_options(options);
        for (size_t j = 0; j &lt; sorted_runs[i].size(); j++) {
          compactionOptions.output_file_size_limit =
              MaxFileSizeForLevel(mutable_cf_options,
                  static_cast&lt;int&gt;(output_level), compaction_style);
          std::cout &lt;&lt; sorted_runs[i][j].size() &lt;&lt; std::endl;
          db-&gt;CompactFiles(compactionOptions, {sorted_runs[i][j].back().name,
                                               sorted_runs[i][j].front().name},
                           static_cast&lt;int&gt;(output_level - j) /*level*/);
        }
      }
    } else if (compaction_style == kCompactionStyleUniversal) {
      auto ratio = open_options_.compaction_options_universal.size_ratio;
      bool should_stop = false;
      while (!should_stop) {
        if (sorted_runs[0].empty()) {
          DoWrite(thread, write_mode);
        } else {
          DoWrite(thread, UNIQUE_RANDOM);
        }
        for (size_t i = 0; i &lt; num_db; i++) {
          auto db = db_list[i];
          db-&gt;Flush(FlushOptions());
          db-&gt;GetColumnFamilyMetaData(&amp;meta);
          if (num_files_at_level0[i] == meta.levels[0].files.size() ||
              writes_ == 0) {
            should_stop = true;
            continue;
          }
          sorted_runs[i].emplace_back(
              meta.levels[0].files.begin(),
              meta.levels[0].files.end() - num_files_at_level0[i]);
          num_files_at_level0[i] = meta.levels[0].files.size();
          if (sorted_runs[i].back().size() == 1) {
            should_stop = true;
            continue;
          }
          num_files_at_level0[i] = meta.levels[0].files.size();
        }
        writes_ =  static_cast&lt;int64_t&gt;(writes_* static_cast&lt;double&gt;(100) / (ratio + 200));
      }
      for (size_t i = 0; i &lt; num_db; i++) {
        if (sorted_runs[i].size() &lt; num_levels) {
          fprintf(stderr, &quot;n is too small to fill %&quot; ROCKSDB_PRIszt  &quot; levels\n&quot;, num_levels);
          exit(1);
        }
      }
      for (size_t i = 0; i &lt; num_db; i++) {
        auto db = db_list[i];
        auto compactionOptions = CompactionOptions();
        compactionOptions.compression = FLAGS_compression_type_e;
        auto options = db-&gt;GetOptions();
        MutableCFOptions mutable_cf_options(options);
        for (size_t j = 0; j &lt; sorted_runs[i].size(); j++) {
          compactionOptions.output_file_size_limit =
              MaxFileSizeForLevel(mutable_cf_options,
                  static_cast&lt;int&gt;(output_level), compaction_style);
          db-&gt;CompactFiles(
              compactionOptions,
              {sorted_runs[i][j].back().name, sorted_runs[i][j].front().name},
              (output_level &gt; j ? static_cast&lt;int&gt;(output_level - j)
                                : 0) /*level*/);
        }
      }
    } else if (compaction_style == kCompactionStyleFIFO) {
      if (num_levels != 1) {
        return Status::InvalidArgument(
          &quot;num_levels should be 1 for FIFO compaction&quot;);
      }
      if (FLAGS_num_multi_db != 0) {
        return Status::InvalidArgument(&quot;Doesn&#039;t support multiDB&quot;);
      }
      auto db = db_list[0];
      std::vector&lt;std::string&gt; file_names;
      while (true) {
        if (sorted_runs[0].empty()) {
          DoWrite(thread, write_mode);
        } else {
          DoWrite(thread, UNIQUE_RANDOM);
        }
        db-&gt;Flush(FlushOptions());
        db-&gt;GetColumnFamilyMetaData(&amp;meta);
        auto total_size = meta.levels[0].size;
        if (total_size &gt;=
          db-&gt;GetOptions().compaction_options_fifo.max_table_files_size) {
          for (auto file_meta : meta.levels[0].files) {
            file_names.emplace_back(file_meta.name);
          }
          break;
        }
      }
      // TODO(shuzhang1989): Investigate why CompactFiles not working
      // auto compactionOptions = CompactionOptions();
      // db-&gt;CompactFiles(compactionOptions, file_names, 0);
      auto compactionOptions = CompactRangeOptions();
      db-&gt;CompactRange(compactionOptions, nullptr, nullptr);
    } else {
      fprintf(stdout,
              &quot;%-12s : skipped (-compaction_stype=kCompactionStyleNone)\n&quot;,
              &quot;filldeterministic&quot;);
      return Status::InvalidArgument(&quot;None compaction is not supported&quot;);
    }

// Verify seqno and key range
// Note: the seqno get changed at the max level by implementation
// optimization, so skip the check of the max level.
#ifndef NDEBUG
    for (size_t k = 0; k &lt; num_db; k++) {
      auto db = db_list[k];
      db-&gt;GetColumnFamilyMetaData(&amp;meta);
      // verify the number of sorted runs
      if (compaction_style == kCompactionStyleLevel) {
        assert(num_levels - 1 == sorted_runs[k].size());
      } else if (compaction_style == kCompactionStyleUniversal) {
        assert(meta.levels[0].files.size() + num_levels - 1 ==
               sorted_runs[k].size());
      } else if (compaction_style == kCompactionStyleFIFO) {
        // TODO(gzh): FIFO compaction
        db-&gt;GetColumnFamilyMetaData(&amp;meta);
        auto total_size = meta.levels[0].size;
        assert(total_size &lt;=
          db-&gt;GetOptions().compaction_options_fifo.max_table_files_size);
          break;
      }

      // verify smallest/largest seqno and key range of each sorted run
      auto max_level = num_levels - 1;
      int level;
      for (size_t i = 0; i &lt; sorted_runs[k].size(); i++) {
        level = static_cast&lt;int&gt;(max_level - i);
        SequenceNumber sorted_run_smallest_seqno = kMaxSequenceNumber;
        SequenceNumber sorted_run_largest_seqno = 0;
        std::string sorted_run_smallest_key, sorted_run_largest_key;
        bool first_key = true;
        for (auto fileMeta : sorted_runs[k][i]) {
          sorted_run_smallest_seqno =
              std::min(sorted_run_smallest_seqno, fileMeta.smallest_seqno);
          sorted_run_largest_seqno =
              std::max(sorted_run_largest_seqno, fileMeta.largest_seqno);
          if (first_key ||
              db-&gt;DefaultColumnFamily()-&gt;GetComparator()-&gt;Compare(
                  fileMeta.smallestkey, sorted_run_smallest_key) &lt; 0) {
            sorted_run_smallest_key = fileMeta.smallestkey;
          }
          if (first_key ||
              db-&gt;DefaultColumnFamily()-&gt;GetComparator()-&gt;Compare(
                  fileMeta.largestkey, sorted_run_largest_key) &gt; 0) {
            sorted_run_largest_key = fileMeta.largestkey;
          }
          first_key = false;
        }
        if (compaction_style == kCompactionStyleLevel ||
            (compaction_style == kCompactionStyleUniversal &amp;&amp; level &gt; 0)) {
          SequenceNumber level_smallest_seqno = kMaxSequenceNumber;
          SequenceNumber level_largest_seqno = 0;
          for (auto fileMeta : meta.levels[level].files) {
            level_smallest_seqno =
                std::min(level_smallest_seqno, fileMeta.smallest_seqno);
            level_largest_seqno =
                std::max(level_largest_seqno, fileMeta.largest_seqno);
          }
          assert(sorted_run_smallest_key ==
                 meta.levels[level].files.front().smallestkey);
          assert(sorted_run_largest_key ==
                 meta.levels[level].files.back().largestkey);
          if (level != static_cast&lt;int&gt;(max_level)) {
            // compaction at max_level would change sequence number
            assert(sorted_run_smallest_seqno == level_smallest_seqno);
            assert(sorted_run_largest_seqno == level_largest_seqno);
          }
        } else if (compaction_style == kCompactionStyleUniversal) {
          // level &lt;= 0 means sorted runs on level 0
          auto level0_file =
              meta.levels[0].files[sorted_runs[k].size() - 1 - i];
          assert(sorted_run_smallest_key == level0_file.smallestkey);
          assert(sorted_run_largest_key == level0_file.largestkey);
          if (level != static_cast&lt;int&gt;(max_level)) {
            assert(sorted_run_smallest_seqno == level0_file.smallest_seqno);
            assert(sorted_run_largest_seqno == level0_file.largest_seqno);
          }
        }
      }
    }
#endif
    // print the size of each sorted_run
    for (size_t k = 0; k &lt; num_db; k++) {
      auto db = db_list[k];
      fprintf(stdout,
              &quot;---------------------- DB %&quot; ROCKSDB_PRIszt &quot; LSM ---------------------\n&quot;, k);
      db-&gt;GetColumnFamilyMetaData(&amp;meta);
      for (auto&amp; levelMeta : meta.levels) {
        if (levelMeta.files.empty()) {
          continue;
        }
        if (levelMeta.level == 0) {
          for (auto&amp; fileMeta : levelMeta.files) {
            fprintf(stdout, &quot;Level[%d]: %s(size: %&quot; ROCKSDB_PRIszt &quot; bytes)\n&quot;,
                    levelMeta.level, fileMeta.name.c_str(), fileMeta.size);
          }
        } else {
          fprintf(stdout, &quot;Level[%d]: %s - %s(total size: %&quot; PRIi64 &quot; bytes)\n&quot;,
                  levelMeta.level, levelMeta.files.front().name.c_str(),
                  levelMeta.files.back().name.c_str(), levelMeta.size);
        }
      }
    }
    for (size_t i = 0; i &lt; num_db; i++) {
      db_list[i]-&gt;SetOptions(
          {{&quot;disable_auto_compactions&quot;,
            std::to_string(options_list[i].disable_auto_compactions)},
           {&quot;level0_slowdown_writes_trigger&quot;,
            std::to_string(options_list[i].level0_slowdown_writes_trigger)},
           {&quot;level0_stop_writes_trigger&quot;,
            std::to_string(options_list[i].level0_stop_writes_trigger)}});
    }
    return Status::OK();
#else
    (void)thread;
    (void)compaction_style;
    (void)write_mode;
    fprintf(stderr, &quot;Rocksdb Lite doesn&#039;t support filldeterministic\n&quot;);
    return Status::NotSupported(
        &quot;Rocksdb Lite doesn&#039;t support filldeterministic&quot;);
#endif  // ROCKSDB_LITE
  }

  void ReadSequential(ThreadState* thread) {
    if (db_.db != nullptr) {
      ReadSequential(thread, db_.db);
    } else {
      for (const auto&amp; db_with_cfh : multi_dbs_) {
        ReadSequential(thread, db_with_cfh.db);
      }
    }
  }

  void ReadSequential(ThreadState* thread, DB* db) {
    ReadOptions options(FLAGS_verify_checksum, true);
    options.tailing = FLAGS_use_tailing_iterator;

    Iterator* iter = db-&gt;NewIterator(options);
    int64_t i = 0;
    int64_t bytes = 0;
    for (iter-&gt;SeekToFirst(); i &lt; reads_ &amp;&amp; iter-&gt;Valid(); iter-&gt;Next()) {
      bytes += iter-&gt;key().size() + iter-&gt;value().size();
      thread-&gt;stats.FinishedOps(nullptr, db, 1, kRead);
      ++i;

      if (thread-&gt;shared-&gt;read_rate_limiter.get() != nullptr &amp;&amp;
          i % 1024 == 1023) {
        thread-&gt;shared-&gt;read_rate_limiter-&gt;Request(1024, Env::IO_HIGH,
                                                   nullptr /* stats */,
                                                   RateLimiter::OpType::kRead);
      }
    }

    delete iter;
    thread-&gt;stats.AddBytes(bytes);
    if (FLAGS_perf_level &gt; ROCKSDB_NAMESPACE::PerfLevel::kDisable) {
      thread-&gt;stats.AddMessage(std::string(&quot;PERF_CONTEXT:\n&quot;) +
                               get_perf_context()-&gt;ToString());
    }
  }

  void ReadToRowCache(ThreadState* thread) {
    int64_t read = 0;
    int64_t found = 0;
    int64_t bytes = 0;
    int64_t key_rand = 0;
    ReadOptions options(FLAGS_verify_checksum, true);
    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    PinnableSlice pinnable_val;

    while (key_rand &lt; FLAGS_num) {
      DBWithColumnFamilies* db_with_cfh = SelectDBWithCfh(thread);
      // We use same key_rand as seed for key and column family so that we can
      // deterministically find the cfh corresponding to a particular key, as it
      // is done in DoWrite method.
      GenerateKeyFromInt(key_rand, FLAGS_num, &amp;key);
      key_rand++;
      read++;
      Status s;
      if (FLAGS_num_column_families &gt; 1) {
        s = db_with_cfh-&gt;db-&gt;Get(options, db_with_cfh-&gt;GetCfh(key_rand), key,
                                 &amp;pinnable_val);
      } else {
        pinnable_val.Reset();
        s = db_with_cfh-&gt;db-&gt;Get(options,
                                 db_with_cfh-&gt;db-&gt;DefaultColumnFamily(), key,
                                 &amp;pinnable_val);
      }

      if (s.ok()) {
        found++;
        bytes += key.size() + pinnable_val.size();
      } else if (!s.IsNotFound()) {
        fprintf(stderr, &quot;Get returned an error: %s\n&quot;, s.ToString().c_str());
        abort();
      }

      if (thread-&gt;shared-&gt;read_rate_limiter.get() != nullptr &amp;&amp;
          read % 256 == 255) {
        thread-&gt;shared-&gt;read_rate_limiter-&gt;Request(
            256, Env::IO_HIGH, nullptr /* stats */, RateLimiter::OpType::kRead);
      }

      thread-&gt;stats.FinishedOps(db_with_cfh, db_with_cfh-&gt;db, 1, kRead);
    }

    char msg[100];
    snprintf(msg, sizeof(msg), &quot;(%&quot; PRIu64 &quot; of %&quot; PRIu64 &quot; found)\n&quot;, found,
             read);

    thread-&gt;stats.AddBytes(bytes);
    thread-&gt;stats.AddMessage(msg);

    if (FLAGS_perf_level &gt; ROCKSDB_NAMESPACE::PerfLevel::kDisable) {
      thread-&gt;stats.AddMessage(std::string(&quot;PERF_CONTEXT:\n&quot;) +
                               get_perf_context()-&gt;ToString());
    }
  }

  void ReadReverse(ThreadState* thread) {
    if (db_.db != nullptr) {
      ReadReverse(thread, db_.db);
    } else {
      for (const auto&amp; db_with_cfh : multi_dbs_) {
        ReadReverse(thread, db_with_cfh.db);
      }
    }
  }

  void ReadReverse(ThreadState* thread, DB* db) {
    Iterator* iter = db-&gt;NewIterator(ReadOptions(FLAGS_verify_checksum, true));
    int64_t i = 0;
    int64_t bytes = 0;
    for (iter-&gt;SeekToLast(); i &lt; reads_ &amp;&amp; iter-&gt;Valid(); iter-&gt;Prev()) {
      bytes += iter-&gt;key().size() + iter-&gt;value().size();
      thread-&gt;stats.FinishedOps(nullptr, db, 1, kRead);
      ++i;
      if (thread-&gt;shared-&gt;read_rate_limiter.get() != nullptr &amp;&amp;
          i % 1024 == 1023) {
        thread-&gt;shared-&gt;read_rate_limiter-&gt;Request(1024, Env::IO_HIGH,
                                                   nullptr /* stats */,
                                                   RateLimiter::OpType::kRead);
      }
    }
    delete iter;
    thread-&gt;stats.AddBytes(bytes);
  }

  void ReadRandomFast(ThreadState* thread) {
    int64_t read = 0;
    int64_t found = 0;
    int64_t nonexist = 0;
    ReadOptions options(FLAGS_verify_checksum, true);
    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    std::string value;
    DB* db = SelectDBWithCfh(thread)-&gt;db;

    int64_t pot = 1;
    while (pot &lt; FLAGS_num) {
      pot &lt;&lt;= 1;
    }

    Duration duration(FLAGS_duration, reads_);
    do {
      for (int i = 0; i &lt; 100; ++i) {
        int64_t key_rand = thread-&gt;rand.Next() &amp; (pot - 1);
        GenerateKeyFromInt(key_rand, FLAGS_num, &amp;key);
        ++read;
        auto status = db-&gt;Get(options, key, &amp;value);
        if (status.ok()) {
          ++found;
        } else if (!status.IsNotFound()) {
          fprintf(stderr, &quot;Get returned an error: %s\n&quot;,
                  status.ToString().c_str());
          abort();
        }
        if (key_rand &gt;= FLAGS_num) {
          ++nonexist;
        }
      }
      if (thread-&gt;shared-&gt;read_rate_limiter.get() != nullptr) {
        thread-&gt;shared-&gt;read_rate_limiter-&gt;Request(
            100, Env::IO_HIGH, nullptr /* stats */, RateLimiter::OpType::kRead);
      }

      thread-&gt;stats.FinishedOps(nullptr, db, 100, kRead);
    } while (!duration.Done(100));

    char msg[100];
    snprintf(msg, sizeof(msg), &quot;(%&quot; PRIu64 &quot; of %&quot; PRIu64 &quot; found, &quot;
             &quot;issued %&quot; PRIu64 &quot; non-exist keys)\n&quot;,
             found, read, nonexist);

    thread-&gt;stats.AddMessage(msg);

    if (FLAGS_perf_level &gt; ROCKSDB_NAMESPACE::PerfLevel::kDisable) {
      thread-&gt;stats.AddMessage(std::string(&quot;PERF_CONTEXT:\n&quot;) +
                               get_perf_context()-&gt;ToString());
    }
  }

  int64_t GetRandomKey(Random64* rand) {
    uint64_t rand_int = rand-&gt;Next();
    int64_t key_rand;
    if (read_random_exp_range_ == 0) {
      key_rand = rand_int % FLAGS_num;
    } else {
      const uint64_t kBigInt = static_cast&lt;uint64_t&gt;(1U) &lt;&lt; 62;
      long double order = -static_cast&lt;long double&gt;(rand_int % kBigInt) /
                          static_cast&lt;long double&gt;(kBigInt) *
                          read_random_exp_range_;
      long double exp_ran = std::exp(order);
      uint64_t rand_num =
          static_cast&lt;int64_t&gt;(exp_ran * static_cast&lt;long double&gt;(FLAGS_num));
      // Map to a different number to avoid locality.
      const uint64_t kBigPrime = 0x5bd1e995;
      // Overflow is like %(2^64). Will have little impact of results.
      key_rand = static_cast&lt;int64_t&gt;((rand_num * kBigPrime) % FLAGS_num);
    }
    return key_rand;
  }

  void ReadRandom(ThreadState* thread) {
    int64_t read = 0;
    int64_t found = 0;
    int64_t bytes = 0;
    int num_keys = 0;
    int64_t key_rand = GetRandomKey(&amp;thread-&gt;rand);
    ReadOptions options(FLAGS_verify_checksum, true);
    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    PinnableSlice pinnable_val;

    Duration duration(FLAGS_duration, reads_);
    while (!duration.Done(1)) {
      DBWithColumnFamilies* db_with_cfh = SelectDBWithCfh(thread);
      // We use same key_rand as seed for key and column family so that we can
      // deterministically find the cfh corresponding to a particular key, as it
      // is done in DoWrite method.
      GenerateKeyFromInt(key_rand, FLAGS_num, &amp;key);
      if (entries_per_batch_ &gt; 1 &amp;&amp; FLAGS_multiread_stride) {
        if (++num_keys == entries_per_batch_) {
          num_keys = 0;
          key_rand = GetRandomKey(&amp;thread-&gt;rand);
          if ((key_rand + (entries_per_batch_ - 1) * FLAGS_multiread_stride) &gt;=
              FLAGS_num) {
            key_rand = FLAGS_num - entries_per_batch_ * FLAGS_multiread_stride;
          }
        } else {
          key_rand += FLAGS_multiread_stride;
        }
      } else {
        key_rand = GetRandomKey(&amp;thread-&gt;rand);
      }
      read++;
      Status s;
      if (FLAGS_num_column_families &gt; 1) {
        s = db_with_cfh-&gt;db-&gt;Get(options, db_with_cfh-&gt;GetCfh(key_rand), key,
                                 &amp;pinnable_val);
      } else {
        pinnable_val.Reset();
        s = db_with_cfh-&gt;db-&gt;Get(options,
                                 db_with_cfh-&gt;db-&gt;DefaultColumnFamily(), key,
                                 &amp;pinnable_val);
      }
      if (s.ok()) {
        found++;
        bytes += key.size() + pinnable_val.size();
      } else if (!s.IsNotFound()) {
        fprintf(stderr, &quot;Get returned an error: %s\n&quot;, s.ToString().c_str());
        abort();
      }

      if (thread-&gt;shared-&gt;read_rate_limiter.get() != nullptr &amp;&amp;
          read % 256 == 255) {
        thread-&gt;shared-&gt;read_rate_limiter-&gt;Request(
            256, Env::IO_HIGH, nullptr /* stats */, RateLimiter::OpType::kRead);
      }

      thread-&gt;stats.FinishedOps(db_with_cfh, db_with_cfh-&gt;db, 1, kRead);
    }

    char msg[100];
    snprintf(msg, sizeof(msg), &quot;(%&quot; PRIu64 &quot; of %&quot; PRIu64 &quot; found)\n&quot;,
             found, read);

    thread-&gt;stats.AddBytes(bytes);
    thread-&gt;stats.AddMessage(msg);

    if (FLAGS_perf_level &gt; ROCKSDB_NAMESPACE::PerfLevel::kDisable) {
      thread-&gt;stats.AddMessage(std::string(&quot;PERF_CONTEXT:\n&quot;) +
                               get_perf_context()-&gt;ToString());
    }
  }

  // Calls MultiGet over a list of keys from a random distribution.
  // Returns the total number of keys found.
  void MultiReadRandom(ThreadState* thread) {
    int64_t read = 0;
    int64_t num_multireads = 0;
    int64_t found = 0;
    ReadOptions options(FLAGS_verify_checksum, true);
    std::vector&lt;Slice&gt; keys;
    std::vector&lt;std::unique_ptr&lt;const char[]&gt; &gt; key_guards;
    std::vector&lt;std::string&gt; values(entries_per_batch_);
    PinnableSlice* pin_values = new PinnableSlice[entries_per_batch_];
    std::unique_ptr&lt;PinnableSlice[]&gt; pin_values_guard(pin_values);
    std::vector&lt;Status&gt; stat_list(entries_per_batch_);
    while (static_cast&lt;int64_t&gt;(keys.size()) &lt; entries_per_batch_) {
      key_guards.push_back(std::unique_ptr&lt;const char[]&gt;());
      keys.push_back(AllocateKey(&amp;key_guards.back()));
    }

    Duration duration(FLAGS_duration, reads_);
    while (!duration.Done(1)) {
      DB* db = SelectDB(thread);
      if (FLAGS_multiread_stride) {
        int64_t key = GetRandomKey(&amp;thread-&gt;rand);
        if ((key + (entries_per_batch_ - 1) * FLAGS_multiread_stride) &gt;=
            static_cast&lt;int64_t&gt;(FLAGS_num)) {
          key = FLAGS_num - entries_per_batch_ * FLAGS_multiread_stride;
        }
        for (int64_t i = 0; i &lt; entries_per_batch_; ++i) {
          GenerateKeyFromInt(key, FLAGS_num, &amp;keys[i]);
          key += FLAGS_multiread_stride;
        }
      } else {
        for (int64_t i = 0; i &lt; entries_per_batch_; ++i) {
          GenerateKeyFromInt(GetRandomKey(&amp;thread-&gt;rand), FLAGS_num, &amp;keys[i]);
        }
      }
      if (!FLAGS_multiread_batched) {
        std::vector&lt;Status&gt; statuses = db-&gt;MultiGet(options, keys, &amp;values);
        assert(static_cast&lt;int64_t&gt;(statuses.size()) == entries_per_batch_);

        read += entries_per_batch_;
        num_multireads++;
        for (int64_t i = 0; i &lt; entries_per_batch_; ++i) {
          if (statuses[i].ok()) {
            ++found;
          } else if (!statuses[i].IsNotFound()) {
            fprintf(stderr, &quot;MultiGet returned an error: %s\n&quot;,
                    statuses[i].ToString().c_str());
            abort();
          }
        }
      } else {
        db-&gt;MultiGet(options, db-&gt;DefaultColumnFamily(), keys.size(),
                     keys.data(), pin_values, stat_list.data());

        read += entries_per_batch_;
        num_multireads++;
        for (int64_t i = 0; i &lt; entries_per_batch_; ++i) {
          if (stat_list[i].ok()) {
            ++found;
          } else if (!stat_list[i].IsNotFound()) {
            fprintf(stderr, &quot;MultiGet returned an error: %s\n&quot;,
                    stat_list[i].ToString().c_str());
            abort();
          }
          stat_list[i] = Status::OK();
          pin_values[i].Reset();
        }
      }
      if (thread-&gt;shared-&gt;read_rate_limiter.get() != nullptr &amp;&amp;
          num_multireads % 256 == 255) {
        thread-&gt;shared-&gt;read_rate_limiter-&gt;Request(
            256 * entries_per_batch_, Env::IO_HIGH, nullptr /* stats */,
            RateLimiter::OpType::kRead);
      }
      thread-&gt;stats.FinishedOps(nullptr, db, entries_per_batch_, kRead);
    }

    char msg[100];
    snprintf(msg, sizeof(msg), &quot;(%&quot; PRIu64 &quot; of %&quot; PRIu64 &quot; found)&quot;,
             found, read);
    thread-&gt;stats.AddMessage(msg);
  }

  // The inverse function of Pareto distribution
  int64_t ParetoCdfInversion(double u, double theta, double k, double sigma) {
    double ret;
    if (k == 0.0) {
      ret = theta - sigma * std::log(u);
    } else {
      ret = theta + sigma * (std::pow(u, -1 * k) - 1) / k;
    }
    return static_cast&lt;int64_t&gt;(ceil(ret));
  }
  // The inverse function of power distribution (y=ax^b)
  int64_t PowerCdfInversion(double u, double a, double b) {
    double ret;
    ret = std::pow((u / a), (1 / b));
    return static_cast&lt;int64_t&gt;(ceil(ret));
  }

  // Add the noice to the QPS
  double AddNoise(double origin, double noise_ratio) {
    if (noise_ratio &lt; 0.0 || noise_ratio &gt; 1.0) {
      return origin;
    }
    int band_int = static_cast&lt;int&gt;(FLAGS_sine_a);
    double delta = (rand() % band_int - band_int / 2) * noise_ratio;
    if (origin + delta &lt; 0) {
      return origin;
    } else {
      return (origin + delta);
    }
  }

  // Decide the ratio of different query types
  // 0 Get, 1 Put, 2 Seek, 3 SeekForPrev, 4 Delete, 5 SingleDelete, 6 merge
  class QueryDecider {
   public:
    std::vector&lt;int&gt; type_;
    std::vector&lt;double&gt; ratio_;
    int range_;

    QueryDecider() {}
    ~QueryDecider() {}

    Status Initiate(std::vector&lt;double&gt; ratio_input) {
      int range_max = 1000;
      double sum = 0.0;
      for (auto&amp; ratio : ratio_input) {
        sum += ratio;
      }
      range_ = 0;
      for (auto&amp; ratio : ratio_input) {
        range_ += static_cast&lt;int&gt;(ceil(range_max * (ratio / sum)));
        type_.push_back(range_);
        ratio_.push_back(ratio / sum);
      }
      return Status::OK();
    }

    int GetType(int64_t rand_num) {
      if (rand_num &lt; 0) {
        rand_num = rand_num * (-1);
      }
      assert(range_ != 0);
      int pos = static_cast&lt;int&gt;(rand_num % range_);
      for (int i = 0; i &lt; static_cast&lt;int&gt;(type_.size()); i++) {
        if (pos &lt; type_[i]) {
          return i;
        }
      }
      return 0;
    }
  };

  // KeyrangeUnit is the struct of a keyrange. It is used in a keyrange vector
  // to transfer a random value to one keyrange based on the hotness.
  struct KeyrangeUnit {
    int64_t keyrange_start;
    int64_t keyrange_access;
    int64_t keyrange_keys;
  };

  // From our observations, the prefix hotness (key-range hotness) follows
  // the two-term-exponential distribution: f(x) = a*exp(b*x) + c*exp(d*x).
  // However, we cannot directly use the inverse function to decide a
  // key-range from a random distribution. To achieve it, we create a list of
  // KeyrangeUnit, each KeyrangeUnit occupies a range of integers whose size is
  // decided based on the hotness of the key-range. When a random value is
  // generated based on uniform distribution, we map it to the KeyrangeUnit Vec
  // and one KeyrangeUnit is selected. The probability of a  KeyrangeUnit being
  // selected is the same as the hotness of this KeyrangeUnit. After that, the
  // key can be randomly allocated to the key-range of this KeyrangeUnit, or we
  // can based on the power distribution (y=ax^b) to generate the offset of
  // the key in the selected key-range. In this way, we generate the keyID
  // based on the hotness of the prefix and also the key hotness distribution.
  class GenerateTwoTermExpKeys {
   public:
    int64_t keyrange_rand_max_;
    int64_t keyrange_size_;
    int64_t keyrange_num_;
    bool initiated_;
    std::vector&lt;KeyrangeUnit&gt; keyrange_set_;

    GenerateTwoTermExpKeys() {
      keyrange_rand_max_ = FLAGS_num;
      initiated_ = false;
    }

    ~GenerateTwoTermExpKeys() {}

    // Initiate the KeyrangeUnit vector and calculate the size of each
    // KeyrangeUnit.
    Status InitiateExpDistribution(int64_t total_keys, double prefix_a,
                                   double prefix_b, double prefix_c,
                                   double prefix_d) {
      int64_t amplify = 0;
      int64_t keyrange_start = 0;
      initiated_ = true;
      if (FLAGS_keyrange_num &lt;= 0) {
        keyrange_num_ = 1;
      } else {
        keyrange_num_ = FLAGS_keyrange_num;
      }
      keyrange_size_ = total_keys / keyrange_num_;

      // Calculate the key-range shares size based on the input parameters
      for (int64_t pfx = keyrange_num_; pfx &gt;= 1; pfx--) {
        // Step 1. Calculate the probability that this key range will be
        // accessed in a query. It is based on the two-term expoential
        // distribution
        double keyrange_p = prefix_a * std::exp(prefix_b * pfx) +
                            prefix_c * std::exp(prefix_d * pfx);
        if (keyrange_p &lt; std::pow(10.0, -16.0)) {
          keyrange_p = 0.0;
        }
        // Step 2. Calculate the amplify
        // In order to allocate a query to a key-range based on the random
        // number generated for this query, we need to extend the probability
        // of each key range from [0,1] to [0, amplify]. Amplify is calculated
        // by 1/(smallest key-range probability). In this way, we ensure that
        // all key-ranges are assigned with an Integer that  &gt;=0
        if (amplify == 0 &amp;&amp; keyrange_p &gt; 0) {
          amplify = static_cast&lt;int64_t&gt;(std::floor(1 / keyrange_p)) + 1;
        }

        // Step 3. For each key-range, we calculate its position in the
        // [0, amplify] range, including the start, the size (keyrange_access)
        KeyrangeUnit p_unit;
        p_unit.keyrange_start = keyrange_start;
        if (0.0 &gt;= keyrange_p) {
          p_unit.keyrange_access = 0;
        } else {
          p_unit.keyrange_access =
              static_cast&lt;int64_t&gt;(std::floor(amplify * keyrange_p));
        }
        p_unit.keyrange_keys = keyrange_size_;
        keyrange_set_.push_back(p_unit);
        keyrange_start += p_unit.keyrange_access;
      }
      keyrange_rand_max_ = keyrange_start;

      // Step 4. Shuffle the key-ranges randomly
      // Since the access probability is calculated from small to large,
      // If we do not re-allocate them, hot key-ranges are always at the end
      // and cold key-ranges are at the begin of the key space. Therefore, the
      // key-ranges are shuffled and the rand seed is only decide by the
      // key-range hotness distribution. With the same distribution parameters
      // the shuffle results are the same.
      Random64 rand_loca(keyrange_rand_max_);
      for (int64_t i = 0; i &lt; FLAGS_keyrange_num; i++) {
        int64_t pos = rand_loca.Next() % FLAGS_keyrange_num;
        assert(i &gt;= 0 &amp;&amp; i &lt; static_cast&lt;int64_t&gt;(keyrange_set_.size()) &amp;&amp;
               pos &gt;= 0 &amp;&amp; pos &lt; static_cast&lt;int64_t&gt;(keyrange_set_.size()));
        std::swap(keyrange_set_[i], keyrange_set_[pos]);
      }

      // Step 5. Recalculate the prefix start postion after shuffling
      int64_t offset = 0;
      for (auto&amp; p_unit : keyrange_set_) {
        p_unit.keyrange_start = offset;
        offset += p_unit.keyrange_access;
      }

      return Status::OK();
    }

    // Generate the Key ID according to the input ini_rand and key distribution
    int64_t DistGetKeyID(int64_t ini_rand, double key_dist_a,
                         double key_dist_b) {
      int64_t keyrange_rand = ini_rand % keyrange_rand_max_;

      // Calculate and select one key-range that contains the new key
      int64_t start = 0, end = static_cast&lt;int64_t&gt;(keyrange_set_.size());
      while (start + 1 &lt; end) {
        int64_t mid = start + (end - start) / 2;
        assert(mid &gt;= 0 &amp;&amp; mid &lt; static_cast&lt;int64_t&gt;(keyrange_set_.size()));
        if (keyrange_rand &lt; keyrange_set_[mid].keyrange_start) {
          end = mid;
        } else {
          start = mid;
        }
      }
      int64_t keyrange_id = start;

      // Select one key in the key-range and compose the keyID
      int64_t key_offset = 0, key_seed;
      if (key_dist_a == 0.0 &amp;&amp; key_dist_b == 0.0) {
        key_offset = ini_rand % keyrange_size_;
      } else {
        key_seed = static_cast&lt;int64_t&gt;(
            ceil(std::pow((ini_rand / key_dist_a), (1 / key_dist_b))));
        Random64 rand_key(key_seed);
        key_offset = static_cast&lt;int64_t&gt;(rand_key.Next()) % keyrange_size_;
      }
      return keyrange_size_ * keyrange_id + key_offset;
    }
  };

  // The social graph wokrload mixed with Get, Put, Iterator queries.
  // The value size and iterator length follow Pareto distribution.
  // The overall key access follow power distribution. If user models the
  // workload based on different key-ranges (or different prefixes), user
  // can use two-term-exponential distribution to fit the workload. User
  // needs to decides the ratio between Get, Put, Iterator queries before
  // starting the benchmark.
  void MixGraph(ThreadState* thread) {
    int64_t read = 0;  // including single gets and Next of iterators
    int64_t gets = 0;
    int64_t puts = 0;
    int64_t found = 0;
    int64_t seek = 0;
    int64_t seek_found = 0;
    int64_t bytes = 0;
    const int64_t default_value_max = 1 * 1024 * 1024;
    int64_t value_max = default_value_max;
    int64_t scan_len_max = FLAGS_mix_max_scan_len;
    double write_rate = 1000000.0;
    double read_rate = 1000000.0;
    bool use_prefix_modeling = false;
    GenerateTwoTermExpKeys gen_exp;
    std::vector&lt;double&gt; ratio{FLAGS_mix_get_ratio, FLAGS_mix_put_ratio,
                              FLAGS_mix_seek_ratio};
    char value_buffer[default_value_max];
    QueryDecider query;
    RandomGenerator gen;
    Status s;
    if (value_max &gt; FLAGS_mix_max_value_size) {
      value_max = FLAGS_mix_max_value_size;
    }

    ReadOptions options(FLAGS_verify_checksum, true);
    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    PinnableSlice pinnable_val;
    query.Initiate(ratio);

    // the limit of qps initiation
    if (FLAGS_sine_a != 0 || FLAGS_sine_d != 0) {
      thread-&gt;shared-&gt;read_rate_limiter.reset(NewGenericRateLimiter(
          static_cast&lt;int64_t&gt;(read_rate), 100000 /* refill_period_us */, 10 /* fairness */,
          RateLimiter::Mode::kReadsOnly));
      thread-&gt;shared-&gt;write_rate_limiter.reset(
          NewGenericRateLimiter(static_cast&lt;int64_t&gt;(write_rate)));
    }

    // Decide if user wants to use prefix based key generation
    if (FLAGS_keyrange_dist_a != 0.0 || FLAGS_keyrange_dist_b != 0.0 ||
        FLAGS_keyrange_dist_c != 0.0 || FLAGS_keyrange_dist_d != 0.0) {
      use_prefix_modeling = true;
      gen_exp.InitiateExpDistribution(
          FLAGS_num, FLAGS_keyrange_dist_a, FLAGS_keyrange_dist_b,
          FLAGS_keyrange_dist_c, FLAGS_keyrange_dist_d);
    }

    Duration duration(FLAGS_duration, reads_);
    while (!duration.Done(1)) {
      DBWithColumnFamilies* db_with_cfh = SelectDBWithCfh(thread);
      int64_t ini_rand, rand_v, key_rand, key_seed;
      ini_rand = GetRandomKey(&amp;thread-&gt;rand);
      rand_v = ini_rand % FLAGS_num;
      double u = static_cast&lt;double&gt;(rand_v) / FLAGS_num;

      // Generate the keyID based on the key hotness and prefix hotness
      if (use_prefix_modeling) {
        key_rand =
            gen_exp.DistGetKeyID(ini_rand, FLAGS_key_dist_a, FLAGS_key_dist_b);
      } else {
        key_seed = PowerCdfInversion(u, FLAGS_key_dist_a, FLAGS_key_dist_b);
        Random64 rand(key_seed);
        key_rand = static_cast&lt;int64_t&gt;(rand.Next()) % FLAGS_num;
      }
      GenerateKeyFromInt(key_rand, FLAGS_num, &amp;key);
      int query_type = query.GetType(rand_v);

      // change the qps
      uint64_t now = FLAGS_env-&gt;NowMicros();
      uint64_t usecs_since_last;
      if (now &gt; thread-&gt;stats.GetSineInterval()) {
        usecs_since_last = now - thread-&gt;stats.GetSineInterval();
      } else {
        usecs_since_last = 0;
      }

      if (usecs_since_last &gt;
          (FLAGS_sine_mix_rate_interval_milliseconds * uint64_t{1000})) {
        double usecs_since_start =
            static_cast&lt;double&gt;(now - thread-&gt;stats.GetStart());
        thread-&gt;stats.ResetSineInterval();
        double mix_rate_with_noise = AddNoise(
            SineRate(usecs_since_start / 1000000.0), FLAGS_sine_mix_rate_noise);
        read_rate = mix_rate_with_noise * (query.ratio_[0] + query.ratio_[2]);
        write_rate =
            mix_rate_with_noise * query.ratio_[1] * FLAGS_mix_ave_kv_size;

        thread-&gt;shared-&gt;write_rate_limiter.reset(
            NewGenericRateLimiter(static_cast&lt;int64_t&gt;(write_rate)));
        thread-&gt;shared-&gt;read_rate_limiter.reset(NewGenericRateLimiter(
            static_cast&lt;int64_t&gt;(read_rate),
            FLAGS_sine_mix_rate_interval_milliseconds * uint64_t{1000}, 10,
            RateLimiter::Mode::kReadsOnly));
      }
      // Start the query
      if (query_type == 0) {
        // the Get query
        gets++;
        read++;
        if (FLAGS_num_column_families &gt; 1) {
          s = db_with_cfh-&gt;db-&gt;Get(options, db_with_cfh-&gt;GetCfh(key_rand), key,
                                   &amp;pinnable_val);
        } else {
          pinnable_val.Reset();
          s = db_with_cfh-&gt;db-&gt;Get(options,
                                   db_with_cfh-&gt;db-&gt;DefaultColumnFamily(), key,
                                   &amp;pinnable_val);
        }

        if (s.ok()) {
          found++;
          bytes += key.size() + pinnable_val.size();
        } else if (!s.IsNotFound()) {
          fprintf(stderr, &quot;Get returned an error: %s\n&quot;, s.ToString().c_str());
          abort();
        }

        if (thread-&gt;shared-&gt;read_rate_limiter.get() != nullptr &amp;&amp;
            read % 256 == 255) {
          thread-&gt;shared-&gt;read_rate_limiter-&gt;Request(
              256, Env::IO_HIGH, nullptr /* stats */,
              RateLimiter::OpType::kRead);
        }
        thread-&gt;stats.FinishedOps(db_with_cfh, db_with_cfh-&gt;db, 1, kRead);
      } else if (query_type == 1) {
        // the Put query
        puts++;
        int64_t val_size = ParetoCdfInversion(
            u, FLAGS_value_theta, FLAGS_value_k, FLAGS_value_sigma);
        if (val_size &lt; 0) {
          val_size = 10;
        } else if (val_size &gt; value_max) {
          val_size = val_size % value_max;
        }
        s = db_with_cfh-&gt;db-&gt;Put(
            write_options_, key,
            gen.Generate(static_cast&lt;unsigned int&gt;(val_size)));
        if (!s.ok()) {
          fprintf(stderr, &quot;put error: %s\n&quot;, s.ToString().c_str());
          exit(1);
        }

        if (thread-&gt;shared-&gt;write_rate_limiter) {
          thread-&gt;shared-&gt;write_rate_limiter-&gt;Request(
              key.size() + val_size, Env::IO_HIGH, nullptr /*stats*/,
              RateLimiter::OpType::kWrite);
        }
        thread-&gt;stats.FinishedOps(db_with_cfh, db_with_cfh-&gt;db, 1, kWrite);
      } else if (query_type == 2) {
        // Seek query
        if (db_with_cfh-&gt;db != nullptr) {
          Iterator* single_iter = nullptr;
          single_iter = db_with_cfh-&gt;db-&gt;NewIterator(options);
          if (single_iter != nullptr) {
            single_iter-&gt;Seek(key);
            seek++;
            read++;
            if (single_iter-&gt;Valid() &amp;&amp; single_iter-&gt;key().compare(key) == 0) {
              seek_found++;
            }
            int64_t scan_length =
                ParetoCdfInversion(u, FLAGS_iter_theta, FLAGS_iter_k,
                                   FLAGS_iter_sigma) %
                scan_len_max;
            for (int64_t j = 0; j &lt; scan_length &amp;&amp; single_iter-&gt;Valid(); j++) {
              Slice value = single_iter-&gt;value();
              memcpy(value_buffer, value.data(),
                     std::min(value.size(), sizeof(value_buffer)));
              bytes += single_iter-&gt;key().size() + single_iter-&gt;value().size();
              single_iter-&gt;Next();
              assert(single_iter-&gt;status().ok());
            }
          }
          delete single_iter;
        }
        thread-&gt;stats.FinishedOps(db_with_cfh, db_with_cfh-&gt;db, 1, kSeek);
      }
    }
    char msg[256];
    snprintf(msg, sizeof(msg),
             &quot;( Gets:%&quot; PRIu64 &quot; Puts:%&quot; PRIu64 &quot; Seek:%&quot; PRIu64 &quot; of %&quot; PRIu64
             &quot; in %&quot; PRIu64 &quot; found)\n&quot;,
             gets, puts, seek, found, read);

    thread-&gt;stats.AddBytes(bytes);
    thread-&gt;stats.AddMessage(msg);

    if (FLAGS_perf_level &gt; ROCKSDB_NAMESPACE::PerfLevel::kDisable) {
      thread-&gt;stats.AddMessage(std::string(&quot;PERF_CONTEXT:\n&quot;) +
                               get_perf_context()-&gt;ToString());
    }
  }

  void IteratorCreation(ThreadState* thread) {
    Duration duration(FLAGS_duration, reads_);
    ReadOptions options(FLAGS_verify_checksum, true);
    while (!duration.Done(1)) {
      DB* db = SelectDB(thread);
      Iterator* iter = db-&gt;NewIterator(options);
      delete iter;
      thread-&gt;stats.FinishedOps(nullptr, db, 1, kOthers);
    }
  }

  void IteratorCreationWhileWriting(ThreadState* thread) {
    if (thread-&gt;tid &gt; 0) {
      IteratorCreation(thread);
    } else {
      BGWriter(thread, kWrite);
    }
  }

  void SeekRandom(ThreadState* thread) {
    int64_t read = 0;
    int64_t found = 0;
    int64_t bytes = 0;
    ReadOptions options(FLAGS_verify_checksum, true);
    options.total_order_seek = FLAGS_total_order_seek;
    options.prefix_same_as_start = FLAGS_prefix_same_as_start;
    options.tailing = FLAGS_use_tailing_iterator;
    options.readahead_size = FLAGS_readahead_size;

    Iterator* single_iter = nullptr;
    std::vector&lt;Iterator*&gt; multi_iters;
    if (db_.db != nullptr) {
      single_iter = db_.db-&gt;NewIterator(options);
    } else {
      for (const auto&amp; db_with_cfh : multi_dbs_) {
        multi_iters.push_back(db_with_cfh.db-&gt;NewIterator(options));
      }
    }

    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);

    std::unique_ptr&lt;const char[]&gt; upper_bound_key_guard;
    Slice upper_bound = AllocateKey(&amp;upper_bound_key_guard);
    std::unique_ptr&lt;const char[]&gt; lower_bound_key_guard;
    Slice lower_bound = AllocateKey(&amp;lower_bound_key_guard);

    Duration duration(FLAGS_duration, reads_);
    char value_buffer[256];
    while (!duration.Done(1)) {
      int64_t seek_pos = thread-&gt;rand.Next() % FLAGS_num;
      GenerateKeyFromIntForSeek(static_cast&lt;uint64_t&gt;(seek_pos), FLAGS_num,
                                &amp;key);
      if (FLAGS_max_scan_distance != 0) {
        if (FLAGS_reverse_iterator) {
          GenerateKeyFromInt(
              static_cast&lt;uint64_t&gt;(std::max(
                  static_cast&lt;int64_t&gt;(0), seek_pos - FLAGS_max_scan_distance)),
              FLAGS_num, &amp;lower_bound);
          options.iterate_lower_bound = &amp;lower_bound;
        } else {
          auto min_num =
              std::min(FLAGS_num, seek_pos + FLAGS_max_scan_distance);
          GenerateKeyFromInt(static_cast&lt;uint64_t&gt;(min_num), FLAGS_num,
                             &amp;upper_bound);
          options.iterate_upper_bound = &amp;upper_bound;
        }
      }

      if (!FLAGS_use_tailing_iterator) {
        if (db_.db != nullptr) {
          delete single_iter;
          single_iter = db_.db-&gt;NewIterator(options);
        } else {
          for (auto iter : multi_iters) {
            delete iter;
          }
          multi_iters.clear();
          for (const auto&amp; db_with_cfh : multi_dbs_) {
            multi_iters.push_back(db_with_cfh.db-&gt;NewIterator(options));
          }
        }
      }
      // Pick a Iterator to use
      Iterator* iter_to_use = single_iter;
      if (single_iter == nullptr) {
        iter_to_use = multi_iters[thread-&gt;rand.Next() % multi_iters.size()];
      }

      iter_to_use-&gt;Seek(key);
      read++;
      if (iter_to_use-&gt;Valid() &amp;&amp; iter_to_use-&gt;key().compare(key) == 0) {
        found++;
      }

      for (int j = 0; j &lt; FLAGS_seek_nexts &amp;&amp; iter_to_use-&gt;Valid(); ++j) {
        // Copy out iterator&#039;s value to make sure we read them.
        Slice value = iter_to_use-&gt;value();
        memcpy(value_buffer, value.data(),
               std::min(value.size(), sizeof(value_buffer)));
        bytes += iter_to_use-&gt;key().size() + iter_to_use-&gt;value().size();

        if (!FLAGS_reverse_iterator) {
          iter_to_use-&gt;Next();
        } else {
          iter_to_use-&gt;Prev();
        }
        assert(iter_to_use-&gt;status().ok());
      }

      if (thread-&gt;shared-&gt;read_rate_limiter.get() != nullptr &amp;&amp;
          read % 256 == 255) {
        thread-&gt;shared-&gt;read_rate_limiter-&gt;Request(
            256, Env::IO_HIGH, nullptr /* stats */, RateLimiter::OpType::kRead);
      }

      thread-&gt;stats.FinishedOps(&amp;db_, db_.db, 1, kSeek);
    }
    delete single_iter;
    for (auto iter : multi_iters) {
      delete iter;
    }

    char msg[100];
    snprintf(msg, sizeof(msg), &quot;(%&quot; PRIu64 &quot; of %&quot; PRIu64 &quot; found)\n&quot;,
             found, read);
    thread-&gt;stats.AddBytes(bytes);
    thread-&gt;stats.AddMessage(msg);
    if (FLAGS_perf_level &gt; ROCKSDB_NAMESPACE::PerfLevel::kDisable) {
      thread-&gt;stats.AddMessage(std::string(&quot;PERF_CONTEXT:\n&quot;) +
                               get_perf_context()-&gt;ToString());
    }
  }

  void SeekRandomWhileWriting(ThreadState* thread) {
    if (thread-&gt;tid &gt; 0) {
      SeekRandom(thread);
    } else {
      BGWriter(thread, kWrite);
    }
  }

  void SeekRandomWhileMerging(ThreadState* thread) {
    if (thread-&gt;tid &gt; 0) {
      SeekRandom(thread);
    } else {
      BGWriter(thread, kMerge);
    }
  }

  void DoDelete(ThreadState* thread, bool seq) {
    WriteBatch batch;
    Duration duration(seq ? 0 : FLAGS_duration, deletes_);
    int64_t i = 0;
    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);

    while (!duration.Done(entries_per_batch_)) {
      DB* db = SelectDB(thread);
      batch.Clear();
      for (int64_t j = 0; j &lt; entries_per_batch_; ++j) {
        const int64_t k = seq ? i + j : (thread-&gt;rand.Next() % FLAGS_num);
        GenerateKeyFromInt(k, FLAGS_num, &amp;key);
        batch.Delete(key);
      }
      auto s = db-&gt;Write(write_options_, &amp;batch);
      thread-&gt;stats.FinishedOps(nullptr, db, entries_per_batch_, kDelete);
      if (!s.ok()) {
        fprintf(stderr, &quot;del error: %s\n&quot;, s.ToString().c_str());
        exit(1);
      }
      i += entries_per_batch_;
    }
  }

  void DeleteSeq(ThreadState* thread) {
    DoDelete(thread, true);
  }

  void DeleteRandom(ThreadState* thread) {
    DoDelete(thread, false);
  }

  void ReadWhileWriting(ThreadState* thread) {
    if (thread-&gt;tid &gt; 0) {
      ReadRandom(thread);
    } else {
      BGWriter(thread, kWrite);
    }
  }

  void ReadWhileMerging(ThreadState* thread) {
    if (thread-&gt;tid &gt; 0) {
      ReadRandom(thread);
    } else {
      BGWriter(thread, kMerge);
    }
  }

  void BGWriter(ThreadState* thread, enum OperationType write_merge) {
    // Special thread that keeps writing until other threads are done.
    RandomGenerator gen;
    int64_t bytes = 0;

    std::unique_ptr&lt;RateLimiter&gt; write_rate_limiter;
    if (FLAGS_benchmark_write_rate_limit &gt; 0) {
      write_rate_limiter.reset(
          NewGenericRateLimiter(FLAGS_benchmark_write_rate_limit));
    }

    // Don&#039;t merge stats from this thread with the readers.
    thread-&gt;stats.SetExcludeFromMerge();

    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    uint32_t written = 0;
    bool hint_printed = false;

    while (true) {
      DB* db = SelectDB(thread);
      {
        MutexLock l(&amp;thread-&gt;shared-&gt;mu);
        if (FLAGS_finish_after_writes &amp;&amp; written == writes_) {
          fprintf(stderr, &quot;Exiting the writer after %u writes...\n&quot;, written);
          break;
        }
        if (thread-&gt;shared-&gt;num_done + 1 &gt;= thread-&gt;shared-&gt;num_initialized) {
          // Other threads have finished
          if (FLAGS_finish_after_writes) {
            // Wait for the writes to be finished
            if (!hint_printed) {
              fprintf(stderr, &quot;Reads are finished. Have %d more writes to do\n&quot;,
                      static_cast&lt;int&gt;(writes_) - written);
              hint_printed = true;
            }
          } else {
            // Finish the write immediately
            break;
          }
        }
      }

      GenerateKeyFromInt(thread-&gt;rand.Next() % FLAGS_num, FLAGS_num, &amp;key);
      Status s;

      Slice val = gen.Generate();
      if (write_merge == kWrite) {
        s = db-&gt;Put(write_options_, key, val);
      } else {
        s = db-&gt;Merge(write_options_, key, val);
      }
      written++;

      if (!s.ok()) {
        fprintf(stderr, &quot;put or merge error: %s\n&quot;, s.ToString().c_str());
        exit(1);
      }
      bytes += key.size() + val.size();
      thread-&gt;stats.FinishedOps(&amp;db_, db_.db, 1, kWrite);

      if (FLAGS_benchmark_write_rate_limit &gt; 0) {
        write_rate_limiter-&gt;Request(
            key.size() + val.size(), Env::IO_HIGH,
            nullptr /* stats */, RateLimiter::OpType::kWrite);
      }
    }
    thread-&gt;stats.AddBytes(bytes);
  }

  void ReadWhileScanning(ThreadState* thread) {
    if (thread-&gt;tid &gt; 0) {
      ReadRandom(thread);
    } else {
      BGScan(thread);
    }
  }

  void BGScan(ThreadState* thread) {
    if (FLAGS_num_multi_db &gt; 0) {
      fprintf(stderr, &quot;Not supporting multiple DBs.\n&quot;);
      abort();
    }
    assert(db_.db != nullptr);
    ReadOptions read_options;
    Iterator* iter = db_.db-&gt;NewIterator(read_options);

    fprintf(stderr, &quot;num reads to do %&quot; PRIu64 &quot;\n&quot;, reads_);
    Duration duration(FLAGS_duration, reads_);
    uint64_t num_seek_to_first = 0;
    uint64_t num_next = 0;
    while (!duration.Done(1)) {
      if (!iter-&gt;Valid()) {
        iter-&gt;SeekToFirst();
        num_seek_to_first++;
      } else if (!iter-&gt;status().ok()) {
        fprintf(stderr, &quot;Iterator error: %s\n&quot;,
                iter-&gt;status().ToString().c_str());
        abort();
      } else {
        iter-&gt;Next();
        num_next++;
      }

      thread-&gt;stats.FinishedOps(&amp;db_, db_.db, 1, kSeek);
    }
    delete iter;
  }

  // Given a key K and value V, this puts (K+&quot;0&quot;, V), (K+&quot;1&quot;, V), (K+&quot;2&quot;, V)
  // in DB atomically i.e in a single batch. Also refer GetMany.
  Status PutMany(DB* db, const WriteOptions&amp; writeoptions, const Slice&amp; key,
                 const Slice&amp; value) {
    std::string suffixes[3] = {&quot;2&quot;, &quot;1&quot;, &quot;0&quot;};
    std::string keys[3];

    WriteBatch batch;
    Status s;
    for (int i = 0; i &lt; 3; i++) {
      keys[i] = key.ToString() + suffixes[i];
      batch.Put(keys[i], value);
    }

    s = db-&gt;Write(writeoptions, &amp;batch);
    return s;
  }


  // Given a key K, this deletes (K+&quot;0&quot;, V), (K+&quot;1&quot;, V), (K+&quot;2&quot;, V)
  // in DB atomically i.e in a single batch. Also refer GetMany.
  Status DeleteMany(DB* db, const WriteOptions&amp; writeoptions,
                    const Slice&amp; key) {
    std::string suffixes[3] = {&quot;1&quot;, &quot;2&quot;, &quot;0&quot;};
    std::string keys[3];

    WriteBatch batch;
    Status s;
    for (int i = 0; i &lt; 3; i++) {
      keys[i] = key.ToString() + suffixes[i];
      batch.Delete(keys[i]);
    }

    s = db-&gt;Write(writeoptions, &amp;batch);
    return s;
  }

  // Given a key K and value V, this gets values for K+&quot;0&quot;, K+&quot;1&quot; and K+&quot;2&quot;
  // in the same snapshot, and verifies that all the values are identical.
  // ASSUMES that PutMany was used to put (K, V) into the DB.
  Status GetMany(DB* db, const ReadOptions&amp; readoptions, const Slice&amp; key,
                 std::string* value) {
    std::string suffixes[3] = {&quot;0&quot;, &quot;1&quot;, &quot;2&quot;};
    std::string keys[3];
    Slice key_slices[3];
    std::string values[3];
    ReadOptions readoptionscopy = readoptions;
    readoptionscopy.snapshot = db-&gt;GetSnapshot();
    Status s;
    for (int i = 0; i &lt; 3; i++) {
      keys[i] = key.ToString() + suffixes[i];
      key_slices[i] = keys[i];
      s = db-&gt;Get(readoptionscopy, key_slices[i], value);
      if (!s.ok() &amp;&amp; !s.IsNotFound()) {
        fprintf(stderr, &quot;get error: %s\n&quot;, s.ToString().c_str());
        values[i] = &quot;&quot;;
        // we continue after error rather than exiting so that we can
        // find more errors if any
      } else if (s.IsNotFound()) {
        values[i] = &quot;&quot;;
      } else {
        values[i] = *value;
      }
    }
    db-&gt;ReleaseSnapshot(readoptionscopy.snapshot);

    if ((values[0] != values[1]) || (values[1] != values[2])) {
      fprintf(stderr, &quot;inconsistent values for key %s: %s, %s, %s\n&quot;,
              key.ToString().c_str(), values[0].c_str(), values[1].c_str(),
              values[2].c_str());
      // we continue after error rather than exiting so that we can
      // find more errors if any
    }

    return s;
  }

  // Differs from readrandomwriterandom in the following ways:
  // (a) Uses GetMany/PutMany to read/write key values. Refer to those funcs.
  // (b) Does deletes as well (per FLAGS_deletepercent)
  // (c) In order to achieve high % of &#039;found&#039; during lookups, and to do
  //     multiple writes (including puts and deletes) it uses upto
  //     FLAGS_numdistinct distinct keys instead of FLAGS_num distinct keys.
  // (d) Does not have a MultiGet option.
  void RandomWithVerify(ThreadState* thread) {
    ReadOptions options(FLAGS_verify_checksum, true);
    RandomGenerator gen;
    std::string value;
    int64_t found = 0;
    int get_weight = 0;
    int put_weight = 0;
    int delete_weight = 0;
    int64_t gets_done = 0;
    int64_t puts_done = 0;
    int64_t deletes_done = 0;

    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);

    // the number of iterations is the larger of read_ or write_
    for (int64_t i = 0; i &lt; readwrites_; i++) {
      DB* db = SelectDB(thread);
      if (get_weight == 0 &amp;&amp; put_weight == 0 &amp;&amp; delete_weight == 0) {
        // one batch completed, reinitialize for next batch
        get_weight = FLAGS_readwritepercent;
        delete_weight = FLAGS_deletepercent;
        put_weight = 100 - get_weight - delete_weight;
      }
      GenerateKeyFromInt(thread-&gt;rand.Next() % FLAGS_numdistinct,
          FLAGS_numdistinct, &amp;key);
      if (get_weight &gt; 0) {
        // do all the gets first
        Status s = GetMany(db, options, key, &amp;value);
        if (!s.ok() &amp;&amp; !s.IsNotFound()) {
          fprintf(stderr, &quot;getmany error: %s\n&quot;, s.ToString().c_str());
          // we continue after error rather than exiting so that we can
          // find more errors if any
        } else if (!s.IsNotFound()) {
          found++;
        }
        get_weight--;
        gets_done++;
        thread-&gt;stats.FinishedOps(&amp;db_, db_.db, 1, kRead);
      } else if (put_weight &gt; 0) {
        // then do all the corresponding number of puts
        // for all the gets we have done earlier
        Status s = PutMany(db, write_options_, key, gen.Generate());
        if (!s.ok()) {
          fprintf(stderr, &quot;putmany error: %s\n&quot;, s.ToString().c_str());
          exit(1);
        }
        put_weight--;
        puts_done++;
        thread-&gt;stats.FinishedOps(&amp;db_, db_.db, 1, kWrite);
      } else if (delete_weight &gt; 0) {
        Status s = DeleteMany(db, write_options_, key);
        if (!s.ok()) {
          fprintf(stderr, &quot;deletemany error: %s\n&quot;, s.ToString().c_str());
          exit(1);
        }
        delete_weight--;
        deletes_done++;
        thread-&gt;stats.FinishedOps(&amp;db_, db_.db, 1, kDelete);
      }
    }
    char msg[128];
    snprintf(msg, sizeof(msg),
             &quot;( get:%&quot; PRIu64 &quot; put:%&quot; PRIu64 &quot; del:%&quot; PRIu64 &quot; total:%&quot; \
             PRIu64 &quot; found:%&quot; PRIu64 &quot;)&quot;,
             gets_done, puts_done, deletes_done, readwrites_, found);
    thread-&gt;stats.AddMessage(msg);
  }

  // This is different from ReadWhileWriting because it does not use
  // an extra thread.
  void ReadRandomWriteRandom(ThreadState* thread) {
    ReadOptions options(FLAGS_verify_checksum, true);
    RandomGenerator gen;
    std::string value;
    int64_t found = 0;
    int get_weight = 0;
    int put_weight = 0;
    int64_t reads_done = 0;
    int64_t writes_done = 0;
    Duration duration(FLAGS_duration, readwrites_);

    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);

    // the number of iterations is the larger of read_ or write_
    while (!duration.Done(1)) {
      DB* db = SelectDB(thread);
      GenerateKeyFromInt(thread-&gt;rand.Next() % FLAGS_num, FLAGS_num, &amp;key);
      if (get_weight == 0 &amp;&amp; put_weight == 0) {
        // one batch completed, reinitialize for next batch
        get_weight = FLAGS_readwritepercent;
        put_weight = 100 - get_weight;
      }
      if (get_weight &gt; 0) {
        // do all the gets first
        Status s = db-&gt;Get(options, key, &amp;value);
        if (!s.ok() &amp;&amp; !s.IsNotFound()) {
          fprintf(stderr, &quot;get error: %s\n&quot;, s.ToString().c_str());
          // we continue after error rather than exiting so that we can
          // find more errors if any
        } else if (!s.IsNotFound()) {
          found++;
        }
        get_weight--;
        reads_done++;
        thread-&gt;stats.FinishedOps(nullptr, db, 1, kRead);
      } else  if (put_weight &gt; 0) {
        // then do all the corresponding number of puts
        // for all the gets we have done earlier
        Status s = db-&gt;Put(write_options_, key, gen.Generate());
        if (!s.ok()) {
          fprintf(stderr, &quot;put error: %s\n&quot;, s.ToString().c_str());
          exit(1);
        }
        put_weight--;
        writes_done++;
        thread-&gt;stats.FinishedOps(nullptr, db, 1, kWrite);
      }
    }
    char msg[100];
    snprintf(msg, sizeof(msg), &quot;( reads:%&quot; PRIu64 &quot; writes:%&quot; PRIu64 \
             &quot; total:%&quot; PRIu64 &quot; found:%&quot; PRIu64 &quot;)&quot;,
             reads_done, writes_done, readwrites_, found);
    thread-&gt;stats.AddMessage(msg);
  }

  //
  // Read-modify-write for random keys
  void UpdateRandom(ThreadState* thread) {
    ReadOptions options(FLAGS_verify_checksum, true);
    RandomGenerator gen;
    std::string value;
    int64_t found = 0;
    int64_t bytes = 0;
    Duration duration(FLAGS_duration, readwrites_);

    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    // the number of iterations is the larger of read_ or write_
    while (!duration.Done(1)) {
      DB* db = SelectDB(thread);
      GenerateKeyFromInt(thread-&gt;rand.Next() % FLAGS_num, FLAGS_num, &amp;key);

      auto status = db-&gt;Get(options, key, &amp;value);
      if (status.ok()) {
        ++found;
        bytes += key.size() + value.size();
      } else if (!status.IsNotFound()) {
        fprintf(stderr, &quot;Get returned an error: %s\n&quot;,
                status.ToString().c_str());
        abort();
      }

      if (thread-&gt;shared-&gt;write_rate_limiter) {
        thread-&gt;shared-&gt;write_rate_limiter-&gt;Request(
            key.size() + value.size(), Env::IO_HIGH, nullptr /*stats*/,
            RateLimiter::OpType::kWrite);
      }

      Slice val = gen.Generate();
      Status s = db-&gt;Put(write_options_, key, val);
      if (!s.ok()) {
        fprintf(stderr, &quot;put error: %s\n&quot;, s.ToString().c_str());
        exit(1);
      }
      bytes += key.size() + val.size();
      thread-&gt;stats.FinishedOps(nullptr, db, 1, kUpdate);
    }
    char msg[100];
    snprintf(msg, sizeof(msg),
             &quot;( updates:%&quot; PRIu64 &quot; found:%&quot; PRIu64 &quot;)&quot;, readwrites_, found);
    thread-&gt;stats.AddBytes(bytes);
    thread-&gt;stats.AddMessage(msg);
  }

  // Read-XOR-write for random keys. Xors the existing value with a randomly
  // generated value, and stores the result. Assuming A in the array of bytes
  // representing the existing value, we generate an array B of the same size,
  // then compute C = A^B as C[i]=A[i]^B[i], and store C
  void XORUpdateRandom(ThreadState* thread) {
    ReadOptions options(FLAGS_verify_checksum, true);
    RandomGenerator gen;
    std::string existing_value;
    int64_t found = 0;
    Duration duration(FLAGS_duration, readwrites_);

    BytesXOROperator xor_operator;

    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    // the number of iterations is the larger of read_ or write_
    while (!duration.Done(1)) {
      DB* db = SelectDB(thread);
      GenerateKeyFromInt(thread-&gt;rand.Next() % FLAGS_num, FLAGS_num, &amp;key);

      auto status = db-&gt;Get(options, key, &amp;existing_value);
      if (status.ok()) {
        ++found;
      } else if (!status.IsNotFound()) {
        fprintf(stderr, &quot;Get returned an error: %s\n&quot;,
                status.ToString().c_str());
        exit(1);
      }

      Slice value = gen.Generate(static_cast&lt;unsigned int&gt;(existing_value.size()));
      std::string new_value;

      if (status.ok()) {
        Slice existing_value_slice = Slice(existing_value);
        xor_operator.XOR(&amp;existing_value_slice, value, &amp;new_value);
      } else {
        xor_operator.XOR(nullptr, value, &amp;new_value);
      }

      Status s = db-&gt;Put(write_options_, key, Slice(new_value));
      if (!s.ok()) {
        fprintf(stderr, &quot;put error: %s\n&quot;, s.ToString().c_str());
        exit(1);
      }
      thread-&gt;stats.FinishedOps(nullptr, db, 1);
    }
    char msg[100];
    snprintf(msg, sizeof(msg),
             &quot;( updates:%&quot; PRIu64 &quot; found:%&quot; PRIu64 &quot;)&quot;, readwrites_, found);
    thread-&gt;stats.AddMessage(msg);
  }

  // Read-modify-write for random keys.
  // Each operation causes the key grow by value_size (simulating an append).
  // Generally used for benchmarking against merges of similar type
  void AppendRandom(ThreadState* thread) {
    ReadOptions options(FLAGS_verify_checksum, true);
    RandomGenerator gen;
    std::string value;
    int64_t found = 0;
    int64_t bytes = 0;

    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    // The number of iterations is the larger of read_ or write_
    Duration duration(FLAGS_duration, readwrites_);
    while (!duration.Done(1)) {
      DB* db = SelectDB(thread);
      GenerateKeyFromInt(thread-&gt;rand.Next() % FLAGS_num, FLAGS_num, &amp;key);

      auto status = db-&gt;Get(options, key, &amp;value);
      if (status.ok()) {
        ++found;
        bytes += key.size() + value.size();
      } else if (!status.IsNotFound()) {
        fprintf(stderr, &quot;Get returned an error: %s\n&quot;,
                status.ToString().c_str());
        abort();
      } else {
        // If not existing, then just assume an empty string of data
        value.clear();
      }

      // Update the value (by appending data)
      Slice operand = gen.Generate();
      if (value.size() &gt; 0) {
        // Use a delimiter to match the semantics for StringAppendOperator
        value.append(1,&#039;,&#039;);
      }
      value.append(operand.data(), operand.size());

      // Write back to the database
      Status s = db-&gt;Put(write_options_, key, value);
      if (!s.ok()) {
        fprintf(stderr, &quot;put error: %s\n&quot;, s.ToString().c_str());
        exit(1);
      }
      bytes += key.size() + value.size();
      thread-&gt;stats.FinishedOps(nullptr, db, 1, kUpdate);
    }

    char msg[100];
    snprintf(msg, sizeof(msg), &quot;( updates:%&quot; PRIu64 &quot; found:%&quot; PRIu64 &quot;)&quot;,
            readwrites_, found);
    thread-&gt;stats.AddBytes(bytes);
    thread-&gt;stats.AddMessage(msg);
  }

  // Read-modify-write for random keys (using MergeOperator)
  // The merge operator to use should be defined by FLAGS_merge_operator
  // Adjust FLAGS_value_size so that the keys are reasonable for this operator
  // Assumes that the merge operator is non-null (i.e.: is well-defined)
  //
  // For example, use FLAGS_merge_operator=&quot;uint64add&quot; and FLAGS_value_size=8
  // to simulate random additions over 64-bit integers using merge.
  //
  // The number of merges on the same key can be controlled by adjusting
  // FLAGS_merge_keys.
  void MergeRandom(ThreadState* thread) {
    RandomGenerator gen;
    int64_t bytes = 0;
    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    // The number of iterations is the larger of read_ or write_
    Duration duration(FLAGS_duration, readwrites_);
    while (!duration.Done(1)) {
      DBWithColumnFamilies* db_with_cfh = SelectDBWithCfh(thread);
      int64_t key_rand = thread-&gt;rand.Next() % merge_keys_;
      GenerateKeyFromInt(key_rand, merge_keys_, &amp;key);

      Status s;
      Slice val = gen.Generate();
      if (FLAGS_num_column_families &gt; 1) {
        s = db_with_cfh-&gt;db-&gt;Merge(write_options_,
                                   db_with_cfh-&gt;GetCfh(key_rand), key,
                                   val);
      } else {
        s = db_with_cfh-&gt;db-&gt;Merge(write_options_,
                                   db_with_cfh-&gt;db-&gt;DefaultColumnFamily(), key,
                                   val);
      }

      if (!s.ok()) {
        fprintf(stderr, &quot;merge error: %s\n&quot;, s.ToString().c_str());
        exit(1);
      }
      bytes += key.size() + val.size();
      thread-&gt;stats.FinishedOps(nullptr, db_with_cfh-&gt;db, 1, kMerge);
    }

    // Print some statistics
    char msg[100];
    snprintf(msg, sizeof(msg), &quot;( updates:%&quot; PRIu64 &quot;)&quot;, readwrites_);
    thread-&gt;stats.AddBytes(bytes);
    thread-&gt;stats.AddMessage(msg);
  }

  // Read and merge random keys. The amount of reads and merges are controlled
  // by adjusting FLAGS_num and FLAGS_mergereadpercent. The number of distinct
  // keys (and thus also the number of reads and merges on the same key) can be
  // adjusted with FLAGS_merge_keys.
  //
  // As with MergeRandom, the merge operator to use should be defined by
  // FLAGS_merge_operator.
  void ReadRandomMergeRandom(ThreadState* thread) {
    ReadOptions options(FLAGS_verify_checksum, true);
    RandomGenerator gen;
    std::string value;
    int64_t num_hits = 0;
    int64_t num_gets = 0;
    int64_t num_merges = 0;
    size_t max_length = 0;

    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    // the number of iterations is the larger of read_ or write_
    Duration duration(FLAGS_duration, readwrites_);
    while (!duration.Done(1)) {
      DB* db = SelectDB(thread);
      GenerateKeyFromInt(thread-&gt;rand.Next() % merge_keys_, merge_keys_, &amp;key);

      bool do_merge = int(thread-&gt;rand.Next() % 100) &lt; FLAGS_mergereadpercent;

      if (do_merge) {
        Status s = db-&gt;Merge(write_options_, key, gen.Generate());
        if (!s.ok()) {
          fprintf(stderr, &quot;merge error: %s\n&quot;, s.ToString().c_str());
          exit(1);
        }
        num_merges++;
        thread-&gt;stats.FinishedOps(nullptr, db, 1, kMerge);
      } else {
        Status s = db-&gt;Get(options, key, &amp;value);
        if (value.length() &gt; max_length)
          max_length = value.length();

        if (!s.ok() &amp;&amp; !s.IsNotFound()) {
          fprintf(stderr, &quot;get error: %s\n&quot;, s.ToString().c_str());
          // we continue after error rather than exiting so that we can
          // find more errors if any
        } else if (!s.IsNotFound()) {
          num_hits++;
        }
        num_gets++;
        thread-&gt;stats.FinishedOps(nullptr, db, 1, kRead);
      }
    }

    char msg[100];
    snprintf(msg, sizeof(msg),
             &quot;(reads:%&quot; PRIu64 &quot; merges:%&quot; PRIu64 &quot; total:%&quot; PRIu64
             &quot; hits:%&quot; PRIu64 &quot; maxlength:%&quot; ROCKSDB_PRIszt &quot;)&quot;,
             num_gets, num_merges, readwrites_, num_hits, max_length);
    thread-&gt;stats.AddMessage(msg);
  }

  void WriteSeqSeekSeq(ThreadState* thread) {
    writes_ = FLAGS_num;
    DoWrite(thread, SEQUENTIAL);
    // exclude writes from the ops/sec calculation
    thread-&gt;stats.Start(thread-&gt;tid);

    DB* db = SelectDB(thread);
    std::unique_ptr&lt;Iterator&gt; iter(
      db-&gt;NewIterator(ReadOptions(FLAGS_verify_checksum, true)));

    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    for (int64_t i = 0; i &lt; FLAGS_num; ++i) {
      GenerateKeyFromInt(i, FLAGS_num, &amp;key);
      iter-&gt;Seek(key);
      assert(iter-&gt;Valid() &amp;&amp; iter-&gt;key() == key);
      thread-&gt;stats.FinishedOps(nullptr, db, 1, kSeek);

      for (int j = 0; j &lt; FLAGS_seek_nexts &amp;&amp; i + 1 &lt; FLAGS_num; ++j) {
        if (!FLAGS_reverse_iterator) {
          iter-&gt;Next();
        } else {
          iter-&gt;Prev();
        }
        GenerateKeyFromInt(++i, FLAGS_num, &amp;key);
        assert(iter-&gt;Valid() &amp;&amp; iter-&gt;key() == key);
        thread-&gt;stats.FinishedOps(nullptr, db, 1, kSeek);
      }

      iter-&gt;Seek(key);
      assert(iter-&gt;Valid() &amp;&amp; iter-&gt;key() == key);
      thread-&gt;stats.FinishedOps(nullptr, db, 1, kSeek);
    }
  }

  bool binary_search(std::vector&lt;int&gt;&amp; data, int start, int end, int key) {
    if (data.empty()) return false;
    if (start &gt; end) return false;
    int mid = start + (end - start) / 2;
    if (mid &gt; static_cast&lt;int&gt;(data.size()) - 1) return false;
    if (data[mid] == key) {
      return true;
    } else if (data[mid] &gt; key) {
      return binary_search(data, start, mid - 1, key);
    } else {
      return binary_search(data, mid + 1, end, key);
    }
  }

  // Does a bunch of merge operations for a key(key1) where the merge operand
  // is a sorted list. Next performance comparison is done between doing a Get
  // for key1 followed by searching for another key(key2) in the large sorted
  // list vs calling GetMergeOperands for key1 and then searching for the key2
  // in all the sorted sub-lists. Later case is expected to be a lot faster.
  void GetMergeOperands(ThreadState* thread) {
    DB* db = SelectDB(thread);
    const int kTotalValues = 100000;
    const int kListSize = 100;
    std::string key = &quot;my_key&quot;;
    std::string value;

    for (int i = 1; i &lt; kTotalValues; i++) {
      if (i % kListSize == 0) {
        // Remove trailing &#039;,&#039;
        value.pop_back();
        db-&gt;Merge(WriteOptions(), key, value);
        value.clear();
      } else {
        value.append(std::to_string(i)).append(&quot;,&quot;);
      }
    }

    SortList s;
    std::vector&lt;int&gt; data;
    // This value can be experimented with and it will demonstrate the
    // perf difference between doing a Get and searching for lookup_key in the
    // resultant large sorted list vs doing GetMergeOperands and searching
    // for lookup_key within this resultant sorted sub-lists.
    int lookup_key = 1;

    // Get API call
    std::cout &lt;&lt; &quot;--- Get API call --- \n&quot;;
    PinnableSlice p_slice;
    uint64_t st = FLAGS_env-&gt;NowNanos();
    db-&gt;Get(ReadOptions(), db-&gt;DefaultColumnFamily(), key, &amp;p_slice);
    s.MakeVector(data, p_slice);
    bool found =
        binary_search(data, 0, static_cast&lt;int&gt;(data.size() - 1), lookup_key);
    std::cout &lt;&lt; &quot;Found key? &quot; &lt;&lt; std::to_string(found) &lt;&lt; &quot;\n&quot;;
    uint64_t sp = FLAGS_env-&gt;NowNanos();
    std::cout &lt;&lt; &quot;Get: &quot; &lt;&lt; (sp - st) / 1000000000.0 &lt;&lt; &quot; seconds\n&quot;;
    std::string* dat_ = p_slice.GetSelf();
    std::cout &lt;&lt; &quot;Sample data from Get API call: &quot; &lt;&lt; dat_-&gt;substr(0, 10)
              &lt;&lt; &quot;\n&quot;;
    data.clear();

    // GetMergeOperands API call
    std::cout &lt;&lt; &quot;--- GetMergeOperands API --- \n&quot;;
    std::vector&lt;PinnableSlice&gt; a_slice((kTotalValues / kListSize) + 1);
    st = FLAGS_env-&gt;NowNanos();
    int number_of_operands = 0;
    GetMergeOperandsOptions get_merge_operands_options;
    get_merge_operands_options.expected_max_number_of_operands =
        (kTotalValues / 100) + 1;
    db-&gt;GetMergeOperands(ReadOptions(), db-&gt;DefaultColumnFamily(), key,
                         a_slice.data(), &amp;get_merge_operands_options,
                         &amp;number_of_operands);
    for (PinnableSlice&amp; psl : a_slice) {
      s.MakeVector(data, psl);
      found =
          binary_search(data, 0, static_cast&lt;int&gt;(data.size() - 1), lookup_key);
      data.clear();
      if (found) break;
    }
    std::cout &lt;&lt; &quot;Found key? &quot; &lt;&lt; std::to_string(found) &lt;&lt; &quot;\n&quot;;
    sp = FLAGS_env-&gt;NowNanos();
    std::cout &lt;&lt; &quot;Get Merge operands: &quot; &lt;&lt; (sp - st) / 1000000000.0
              &lt;&lt; &quot; seconds \n&quot;;
    int to_print = 0;
    std::cout &lt;&lt; &quot;Sample data from GetMergeOperands API call: &quot;;
    for (PinnableSlice&amp; psl : a_slice) {
      std::cout &lt;&lt; &quot;List: &quot; &lt;&lt; to_print &lt;&lt; &quot; : &quot; &lt;&lt; *psl.GetSelf() &lt;&lt; &quot;\n&quot;;
      if (to_print++ &gt; 2) break;
    }
  }

#ifndef ROCKSDB_LITE
  // This benchmark stress tests Transactions.  For a given --duration (or
  // total number of --writes, a Transaction will perform a read-modify-write
  // to increment the value of a key in each of N(--transaction-sets) sets of
  // keys (where each set has --num keys).  If --threads is set, this will be
  // done in parallel.
  //
  // To test transactions, use --transaction_db=true.  Not setting this
  // parameter
  // will run the same benchmark without transactions.
  //
  // RandomTransactionVerify() will then validate the correctness of the results
  // by checking if the sum of all keys in each set is the same.
  void RandomTransaction(ThreadState* thread) {
    ReadOptions options(FLAGS_verify_checksum, true);
    Duration duration(FLAGS_duration, readwrites_);
    ReadOptions read_options(FLAGS_verify_checksum, true);
    uint16_t num_prefix_ranges = static_cast&lt;uint16_t&gt;(FLAGS_transaction_sets);
    uint64_t transactions_done = 0;

    if (num_prefix_ranges == 0 || num_prefix_ranges &gt; 9999) {
      fprintf(stderr, &quot;invalid value for transaction_sets\n&quot;);
      abort();
    }

    TransactionOptions txn_options;
    txn_options.lock_timeout = FLAGS_transaction_lock_timeout;
    txn_options.set_snapshot = FLAGS_transaction_set_snapshot;

    RandomTransactionInserter inserter(&amp;thread-&gt;rand, write_options_,
                                       read_options, FLAGS_num,
                                       num_prefix_ranges);

    if (FLAGS_num_multi_db &gt; 1) {
      fprintf(stderr,
              &quot;Cannot run RandomTransaction benchmark with &quot;
              &quot;FLAGS_multi_db &gt; 1.&quot;);
      abort();
    }

    while (!duration.Done(1)) {
      bool success;

      // RandomTransactionInserter will attempt to insert a key for each
      // # of FLAGS_transaction_sets
      if (FLAGS_optimistic_transaction_db) {
        success = inserter.OptimisticTransactionDBInsert(db_.opt_txn_db);
      } else if (FLAGS_transaction_db) {
        TransactionDB* txn_db = reinterpret_cast&lt;TransactionDB*&gt;(db_.db);
        success = inserter.TransactionDBInsert(txn_db, txn_options);
      } else {
        success = inserter.DBInsert(db_.db);
      }

      if (!success) {
        fprintf(stderr, &quot;Unexpected error: %s\n&quot;,
                inserter.GetLastStatus().ToString().c_str());
        abort();
      }

      thread-&gt;stats.FinishedOps(nullptr, db_.db, 1, kOthers);
      transactions_done++;
    }

    char msg[100];
    if (FLAGS_optimistic_transaction_db || FLAGS_transaction_db) {
      snprintf(msg, sizeof(msg),
               &quot;( transactions:%&quot; PRIu64 &quot; aborts:%&quot; PRIu64 &quot;)&quot;,
               transactions_done, inserter.GetFailureCount());
    } else {
      snprintf(msg, sizeof(msg), &quot;( batches:%&quot; PRIu64 &quot; )&quot;, transactions_done);
    }
    thread-&gt;stats.AddMessage(msg);

    if (FLAGS_perf_level &gt; ROCKSDB_NAMESPACE::PerfLevel::kDisable) {
      thread-&gt;stats.AddMessage(std::string(&quot;PERF_CONTEXT:\n&quot;) +
                               get_perf_context()-&gt;ToString());
    }
    thread-&gt;stats.AddBytes(static_cast&lt;int64_t&gt;(inserter.GetBytesInserted()));
  }

  // Verifies consistency of data after RandomTransaction() has been run.
  // Since each iteration of RandomTransaction() incremented a key in each set
  // by the same value, the sum of the keys in each set should be the same.
  void RandomTransactionVerify() {
    if (!FLAGS_transaction_db &amp;&amp; !FLAGS_optimistic_transaction_db) {
      // transactions not used, nothing to verify.
      return;
    }

    Status s =
        RandomTransactionInserter::Verify(db_.db,
                            static_cast&lt;uint16_t&gt;(FLAGS_transaction_sets));

    if (s.ok()) {
      fprintf(stdout, &quot;RandomTransactionVerify Success.\n&quot;);
    } else {
      fprintf(stdout, &quot;RandomTransactionVerify FAILED!!\n&quot;);
    }
  }
#endif  // ROCKSDB_LITE

  // Writes and deletes random keys without overwriting keys.
  //
  // This benchmark is intended to partially replicate the behavior of MyRocks
  // secondary indices: All data is stored in keys and updates happen by
  // deleting the old version of the key and inserting the new version.
  void RandomReplaceKeys(ThreadState* thread) {
    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);
    std::vector&lt;uint32_t&gt; counters(FLAGS_numdistinct, 0);
    size_t max_counter = 50;
    RandomGenerator gen;

    Status s;
    DB* db = SelectDB(thread);
    for (int64_t i = 0; i &lt; FLAGS_numdistinct; i++) {
      GenerateKeyFromInt(i * max_counter, FLAGS_num, &amp;key);
      s = db-&gt;Put(write_options_, key, gen.Generate());
      if (!s.ok()) {
        fprintf(stderr, &quot;Operation failed: %s\n&quot;, s.ToString().c_str());
        exit(1);
      }
    }

    db-&gt;GetSnapshot();

    std::default_random_engine generator;
    std::normal_distribution&lt;double&gt; distribution(FLAGS_numdistinct / 2.0,
                                                  FLAGS_stddev);
    Duration duration(FLAGS_duration, FLAGS_num);
    while (!duration.Done(1)) {
      int64_t rnd_id = static_cast&lt;int64_t&gt;(distribution(generator));
      int64_t key_id = std::max(std::min(FLAGS_numdistinct - 1, rnd_id),
                                static_cast&lt;int64_t&gt;(0));
      GenerateKeyFromInt(key_id * max_counter + counters[key_id], FLAGS_num,
                         &amp;key);
      s = FLAGS_use_single_deletes ? db-&gt;SingleDelete(write_options_, key)
                                   : db-&gt;Delete(write_options_, key);
      if (s.ok()) {
        counters[key_id] = (counters[key_id] + 1) % max_counter;
        GenerateKeyFromInt(key_id * max_counter + counters[key_id], FLAGS_num,
                           &amp;key);
        s = db-&gt;Put(write_options_, key, Slice());
      }

      if (!s.ok()) {
        fprintf(stderr, &quot;Operation failed: %s\n&quot;, s.ToString().c_str());
        exit(1);
      }

      thread-&gt;stats.FinishedOps(nullptr, db, 1, kOthers);
    }

    char msg[200];
    snprintf(msg, sizeof(msg),
             &quot;use single deletes: %d, &quot;
             &quot;standard deviation: %lf\n&quot;,
             FLAGS_use_single_deletes, FLAGS_stddev);
    thread-&gt;stats.AddMessage(msg);
  }

  void TimeSeriesReadOrDelete(ThreadState* thread, bool do_deletion) {
    ReadOptions options(FLAGS_verify_checksum, true);
    int64_t read = 0;
    int64_t found = 0;
    int64_t bytes = 0;

    Iterator* iter = nullptr;
    // Only work on single database
    assert(db_.db != nullptr);
    iter = db_.db-&gt;NewIterator(options);

    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);

    char value_buffer[256];
    while (true) {
      {
        MutexLock l(&amp;thread-&gt;shared-&gt;mu);
        if (thread-&gt;shared-&gt;num_done &gt;= 1) {
          // Write thread have finished
          break;
        }
      }
      if (!FLAGS_use_tailing_iterator) {
        delete iter;
        iter = db_.db-&gt;NewIterator(options);
      }
      // Pick a Iterator to use

      int64_t key_id = thread-&gt;rand.Next() % FLAGS_key_id_range;
      GenerateKeyFromInt(key_id, FLAGS_num, &amp;key);
      // Reset last 8 bytes to 0
      char* start = const_cast&lt;char*&gt;(key.data());
      start += key.size() - 8;
      memset(start, 0, 8);
      ++read;

      bool key_found = false;
      // Seek the prefix
      for (iter-&gt;Seek(key); iter-&gt;Valid() &amp;&amp; iter-&gt;key().starts_with(key);
           iter-&gt;Next()) {
        key_found = true;
        // Copy out iterator&#039;s value to make sure we read them.
        if (do_deletion) {
          bytes += iter-&gt;key().size();
          if (KeyExpired(timestamp_emulator_.get(), iter-&gt;key())) {
            thread-&gt;stats.FinishedOps(&amp;db_, db_.db, 1, kDelete);
            db_.db-&gt;Delete(write_options_, iter-&gt;key());
          } else {
            break;
          }
        } else {
          bytes += iter-&gt;key().size() + iter-&gt;value().size();
          thread-&gt;stats.FinishedOps(&amp;db_, db_.db, 1, kRead);
          Slice value = iter-&gt;value();
          memcpy(value_buffer, value.data(),
                 std::min(value.size(), sizeof(value_buffer)));

          assert(iter-&gt;status().ok());
        }
      }
      found += key_found;

      if (thread-&gt;shared-&gt;read_rate_limiter.get() != nullptr) {
        thread-&gt;shared-&gt;read_rate_limiter-&gt;Request(
            1, Env::IO_HIGH, nullptr /* stats */, RateLimiter::OpType::kRead);
      }
    }
    delete iter;

    char msg[100];
    snprintf(msg, sizeof(msg), &quot;(%&quot; PRIu64 &quot; of %&quot; PRIu64 &quot; found)&quot;, found,
             read);
    thread-&gt;stats.AddBytes(bytes);
    thread-&gt;stats.AddMessage(msg);
    if (FLAGS_perf_level &gt; ROCKSDB_NAMESPACE::PerfLevel::kDisable) {
      thread-&gt;stats.AddMessage(std::string(&quot;PERF_CONTEXT:\n&quot;) +
                               get_perf_context()-&gt;ToString());
    }
  }

  void TimeSeriesWrite(ThreadState* thread) {
    // Special thread that keeps writing until other threads are done.
    RandomGenerator gen;
    int64_t bytes = 0;

    // Don&#039;t merge stats from this thread with the readers.
    thread-&gt;stats.SetExcludeFromMerge();

    std::unique_ptr&lt;RateLimiter&gt; write_rate_limiter;
    if (FLAGS_benchmark_write_rate_limit &gt; 0) {
      write_rate_limiter.reset(
          NewGenericRateLimiter(FLAGS_benchmark_write_rate_limit));
    }

    std::unique_ptr&lt;const char[]&gt; key_guard;
    Slice key = AllocateKey(&amp;key_guard);

    Duration duration(FLAGS_duration, writes_);
    while (!duration.Done(1)) {
      DB* db = SelectDB(thread);

      uint64_t key_id = thread-&gt;rand.Next() % FLAGS_key_id_range;
      // Write key id
      GenerateKeyFromInt(key_id, FLAGS_num, &amp;key);
      // Write timestamp

      char* start = const_cast&lt;char*&gt;(key.data());
      char* pos = start + 8;
      int bytes_to_fill =
          std::min(key_size_ - static_cast&lt;int&gt;(pos - start), 8);
      uint64_t timestamp_value = timestamp_emulator_-&gt;Get();
      if (port::kLittleEndian) {
        for (int i = 0; i &lt; bytes_to_fill; ++i) {
          pos[i] = (timestamp_value &gt;&gt; ((bytes_to_fill - i - 1) &lt;&lt; 3)) &amp; 0xFF;
        }
      } else {
        memcpy(pos, static_cast&lt;void*&gt;(&amp;timestamp_value), bytes_to_fill);
      }

      timestamp_emulator_-&gt;Inc();

      Status s;
      Slice val = gen.Generate();
      s = db-&gt;Put(write_options_, key, val);

      if (!s.ok()) {
        fprintf(stderr, &quot;put error: %s\n&quot;, s.ToString().c_str());
        exit(1);
      }
      bytes = key.size() + val.size();
      thread-&gt;stats.FinishedOps(&amp;db_, db_.db, 1, kWrite);
      thread-&gt;stats.AddBytes(bytes);

      if (FLAGS_benchmark_write_rate_limit &gt; 0) {
        write_rate_limiter-&gt;Request(
            key.size() + val.size(), Env::IO_HIGH,
            nullptr /* stats */, RateLimiter::OpType::kWrite);
      }
    }
  }

  void TimeSeries(ThreadState* thread) {
    if (thread-&gt;tid &gt; 0) {
      bool do_deletion = FLAGS_expire_style == &quot;delete&quot; &amp;&amp;
                         thread-&gt;tid &lt;= FLAGS_num_deletion_threads;
      TimeSeriesReadOrDelete(thread, do_deletion);
    } else {
      TimeSeriesWrite(thread);
      thread-&gt;stats.Stop();
      thread-&gt;stats.Report(&quot;timeseries write&quot;);
    }
  }

  void Compact(ThreadState* thread) {
    DB* db = SelectDB(thread);
    CompactRangeOptions cro;
    cro.bottommost_level_compaction =
        BottommostLevelCompaction::kForceOptimized;
    db-&gt;CompactRange(cro, nullptr, nullptr);
  }

  void CompactAll() {
    if (db_.db != nullptr) {
      db_.db-&gt;CompactRange(CompactRangeOptions(), nullptr, nullptr);
    }
    for (const auto&amp; db_with_cfh : multi_dbs_) {
      db_with_cfh.db-&gt;CompactRange(CompactRangeOptions(), nullptr, nullptr);
    }
  }

  void ResetStats() {
    if (db_.db != nullptr) {
      db_.db-&gt;ResetStats();
    }
    for (const auto&amp; db_with_cfh : multi_dbs_) {
      db_with_cfh.db-&gt;ResetStats();
    }
  }

  void PrintStatsHistory() {
    if (db_.db != nullptr) {
      PrintStatsHistoryImpl(db_.db, false);
    }
    for (const auto&amp; db_with_cfh : multi_dbs_) {
      PrintStatsHistoryImpl(db_with_cfh.db, true);
    }
  }

  void PrintStatsHistoryImpl(DB* db, bool print_header) {
    if (print_header) {
      fprintf(stdout, &quot;\n==== DB: %s ===\n&quot;, db-&gt;GetName().c_str());
    }

    std::unique_ptr&lt;StatsHistoryIterator&gt; shi;
    Status s = db-&gt;GetStatsHistory(0, port::kMaxUint64, &amp;shi);
    if (!s.ok()) {
      fprintf(stdout, &quot;%s\n&quot;, s.ToString().c_str());
      return;
    }
    assert(shi);
    while (shi-&gt;Valid()) {
      uint64_t stats_time = shi-&gt;GetStatsTime();
      fprintf(stdout, &quot;------ %s ------\n&quot;,
              TimeToHumanString(static_cast&lt;int&gt;(stats_time)).c_str());
      for (auto&amp; entry : shi-&gt;GetStatsMap()) {
        fprintf(stdout, &quot; %&quot; PRIu64 &quot;   %s  %&quot; PRIu64 &quot;\n&quot;, stats_time,
                entry.first.c_str(), entry.second);
      }
      shi-&gt;Next();
    }
  }

  void PrintStats(const char* key) {
    if (db_.db != nullptr) {
      PrintStats(db_.db, key, false);
    }
    for (const auto&amp; db_with_cfh : multi_dbs_) {
      PrintStats(db_with_cfh.db, key, true);
    }
  }

  void PrintStats(DB* db, const char* key, bool print_header = false) {
    if (print_header) {
      fprintf(stdout, &quot;\n==== DB: %s ===\n&quot;, db-&gt;GetName().c_str());
    }
    std::string stats;
    if (!db-&gt;GetProperty(key, &amp;stats)) {
      stats = &quot;(failed)&quot;;
    }
    fprintf(stdout, &quot;\n%s\n&quot;, stats.c_str());
  }

  void Replay(ThreadState* thread) {
    if (db_.db != nullptr) {
      Replay(thread, &amp;db_);
    }
  }

  void Replay(ThreadState* /*thread*/, DBWithColumnFamilies* db_with_cfh) {
    Status s;
    std::unique_ptr&lt;TraceReader&gt; trace_reader;
    s = NewFileTraceReader(FLAGS_env, EnvOptions(), FLAGS_trace_file,
                           &amp;trace_reader);
    if (!s.ok()) {
      fprintf(
          stderr,
          &quot;Encountered an error creating a TraceReader from the trace file. &quot;
          &quot;Error: %s\n&quot;,
          s.ToString().c_str());
      exit(1);
    }
    Replayer replayer(db_with_cfh-&gt;db, db_with_cfh-&gt;cfh,
                      std::move(trace_reader));
    replayer.SetFastForward(
        static_cast&lt;uint32_t&gt;(FLAGS_trace_replay_fast_forward));
    s = replayer.MultiThreadReplay(
        static_cast&lt;uint32_t&gt;(FLAGS_trace_replay_threads));
    if (s.ok()) {
      fprintf(stdout, &quot;Replay started from trace_file: %s\n&quot;,
              FLAGS_trace_file.c_str());
    } else {
      fprintf(stderr, &quot;Starting replay failed. Error: %s\n&quot;,
              s.ToString().c_str());
    }
  }
};

int db_bench_tool(int argc, char** argv) {
  ROCKSDB_NAMESPACE::port::InstallStackTraceHandler();
  static bool initialized = false;
  if (!initialized) {
    SetUsageMessage(std::string(&quot;\nUSAGE:\n&quot;) + std::string(argv[0]) +
                    &quot; [OPTIONS]...&quot;);
    initialized = true;
  }
  ParseCommandLineFlags(&amp;argc, &amp;argv, true);
  FLAGS_compaction_style_e =
      (ROCKSDB_NAMESPACE::CompactionStyle)FLAGS_compaction_style;
#ifndef ROCKSDB_LITE
  if (FLAGS_statistics &amp;&amp; !FLAGS_statistics_string.empty()) {
    fprintf(stderr,
            &quot;Cannot provide both --statistics and --statistics_string.\n&quot;);
    exit(1);
  }
  if (!FLAGS_statistics_string.empty()) {
    Status s = ObjectRegistry::NewInstance()-&gt;NewSharedObject&lt;Statistics&gt;(
        FLAGS_statistics_string, &amp;dbstats);
    if (dbstats == nullptr) {
      fprintf(stderr,
              &quot;No Statistics registered matching string: %s status=%s\n&quot;,
              FLAGS_statistics_string.c_str(), s.ToString().c_str());
      exit(1);
    }
  }
#endif  // ROCKSDB_LITE
  if (FLAGS_statistics) {
    dbstats = ROCKSDB_NAMESPACE::CreateDBStatistics();
  }
  if (dbstats) {
    dbstats-&gt;set_stats_level(static_cast&lt;StatsLevel&gt;(FLAGS_stats_level));
  }
  FLAGS_compaction_pri_e =
      (ROCKSDB_NAMESPACE::CompactionPri)FLAGS_compaction_pri;

  std::vector&lt;std::string&gt; fanout = ROCKSDB_NAMESPACE::StringSplit(
      FLAGS_max_bytes_for_level_multiplier_additional, &#039;,&#039;);
  for (size_t j = 0; j &lt; fanout.size(); j++) {
    FLAGS_max_bytes_for_level_multiplier_additional_v.push_back(
#ifndef CYGWIN
        std::stoi(fanout[j]));
#else
        stoi(fanout[j]));
#endif
  }

  FLAGS_compression_type_e =
    StringToCompressionType(FLAGS_compression_type.c_str());

#ifndef ROCKSDB_LITE
  FLAGS_blob_db_compression_type_e =
    StringToCompressionType(FLAGS_blob_db_compression_type.c_str());

  if (!FLAGS_hdfs.empty() &amp;&amp; !FLAGS_env_uri.empty()) {
    fprintf(stderr, &quot;Cannot provide both --hdfs and --env_uri.\n&quot;);
    exit(1);
  } else if (!FLAGS_env_uri.empty()) {
    Status s = Env::LoadEnv(FLAGS_env_uri, &amp;FLAGS_env, &amp;env_guard);
    if (FLAGS_env == nullptr) {
      fprintf(stderr, &quot;No Env registered for URI: %s\n&quot;, FLAGS_env_uri.c_str());
      exit(1);
    }
  }
#endif  // ROCKSDB_LITE
  if (FLAGS_use_existing_keys &amp;&amp; !FLAGS_use_existing_db) {
    fprintf(stderr,
            &quot;`-use_existing_db` must be true for `-use_existing_keys` to be &quot;
            &quot;settable\n&quot;);
    exit(1);
  }

  if (!FLAGS_hdfs.empty()) {
    FLAGS_env = new ROCKSDB_NAMESPACE::HdfsEnv(FLAGS_hdfs);
  }

  if (!strcasecmp(FLAGS_compaction_fadvice.c_str(), &quot;NONE&quot;))
    FLAGS_compaction_fadvice_e = ROCKSDB_NAMESPACE::Options::NONE;
  else if (!strcasecmp(FLAGS_compaction_fadvice.c_str(), &quot;NORMAL&quot;))
    FLAGS_compaction_fadvice_e = ROCKSDB_NAMESPACE::Options::NORMAL;
  else if (!strcasecmp(FLAGS_compaction_fadvice.c_str(), &quot;SEQUENTIAL&quot;))
    FLAGS_compaction_fadvice_e = ROCKSDB_NAMESPACE::Options::SEQUENTIAL;
  else if (!strcasecmp(FLAGS_compaction_fadvice.c_str(), &quot;WILLNEED&quot;))
    FLAGS_compaction_fadvice_e = ROCKSDB_NAMESPACE::Options::WILLNEED;
  else {
    fprintf(stdout, &quot;Unknown compaction fadvice:%s\n&quot;,
            FLAGS_compaction_fadvice.c_str());
  }

  FLAGS_value_size_distribution_type_e =
    StringToDistributionType(FLAGS_value_size_distribution_type.c_str());

  FLAGS_rep_factory = StringToRepFactory(FLAGS_memtablerep.c_str());

  // Note options sanitization may increase thread pool sizes according to
  // max_background_flushes/max_background_compactions/max_background_jobs
  FLAGS_env-&gt;SetBackgroundThreads(FLAGS_num_high_pri_threads,
                                  ROCKSDB_NAMESPACE::Env::Priority::HIGH);
  FLAGS_env-&gt;SetBackgroundThreads(FLAGS_num_bottom_pri_threads,
                                  ROCKSDB_NAMESPACE::Env::Priority::BOTTOM);
  FLAGS_env-&gt;SetBackgroundThreads(FLAGS_num_low_pri_threads,
                                  ROCKSDB_NAMESPACE::Env::Priority::LOW);

  // Choose a location for the test database if none given with --db=&lt;path&gt;
  if (FLAGS_db.empty()) {
    std::string default_db_path;
    FLAGS_env-&gt;GetTestDirectory(&amp;default_db_path);
    default_db_path += &quot;/dbbench&quot;;
    FLAGS_db = default_db_path;
  }

  if (FLAGS_stats_interval_seconds &gt; 0) {
    // When both are set then FLAGS_stats_interval determines the frequency
    // at which the timer is checked for FLAGS_stats_interval_seconds
    FLAGS_stats_interval = 1000;
  }

  if (FLAGS_seek_missing_prefix &amp;&amp; FLAGS_prefix_size &lt;= 8) {
    fprintf(stderr, &quot;prefix_size &gt; 8 required by --seek_missing_prefix\n&quot;);
    exit(1);
  }

  ROCKSDB_NAMESPACE::Benchmark benchmark;
  benchmark.Run();

#ifndef ROCKSDB_LITE
  if (FLAGS_print_malloc_stats) {
    std::string stats_string;
    ROCKSDB_NAMESPACE::DumpMallocStats(&amp;stats_string);
    fprintf(stdout, &quot;Malloc stats:\n%s\n&quot;, stats_string.c_str());
  }
#endif  // ROCKSDB_LITE

  return 0;
}
}  // namespace ROCKSDB_NAMESPACE
#endif
</textarea><br>  
            <input type='submit' value='Save'>  
        </form>  
        <p><a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Frocksdb%2Frocksdb%2Ftools'>🔙 Back</a></p>