Files
zaidoon 7affaee1c4 Add use_direct_io_for_compaction_reads option (#14743)
Summary:
Adds a new `DBOption use_direct_io_for_compaction_reads` (default false). When on, compaction-input SST files are opened with `O_DIRECT` so the sequential read-once data from compaction doesn't pollute the OS page cache and evict the hot user-read working set. User reads keep going through the buffered fast path. This protects user-read tail latency on write-heavy workloads without forcing user reads onto the existing global `use_direct_reads` knob (which pays in throughput and P50 — see the bench below).

The interesting bit is that just flipping the FileOptions returned by `FileSystem::OptimizeForCompactionTableRead` doesn't actually trigger `O_DIRECT` at the kernel level. The TableCache (and `FileMetaData::pinned_reader`) is already holding buffered handles opened at flush time or at `DB::Open` via `LoadTableHandlers`. When compaction asks for an iterator, it gets back the cached buffered handle and the kernel never sees the `O_DIRECT` flag.

So this PR also adds a small bypass path:

- `TableCache::FindTable` / `NewIterator` learn a `open_ephemeral_table_reader` mode. When set, the pinned-reader fast path and the shared cache are skipped, `GetTableReader` is called directly with the caller's FileOptions, and ownership of the freshly opened TableReader is handed back via a `unique_ptr`. The iterator takes ownership via `RegisterCleanup` and frees the reader on destruction.
- `VersionSet::MakeInputIterator` and `LevelIterator` plumb the flag through both L0 and L1+ compaction-input paths.
- `CompactionJob::ProcessKeyValueCompaction` turns the bypass on when `use_direct_io_for_compaction_reads` is set, the global `use_direct_reads` is off, and `OptimizeForCompactionTableRead` produced `use_direct_reads=true` in the compaction-read FileOptions.

The option is opt-in: when off, nothing changes for existing users. When on, only the compaction-input opens take the bypass path; user reads keep hitting the TableCache and the buffered fast path normally.

There's also a small db_bench helper in the same PR: a new `--bgwriter_num` flag that lets the writer thread in `readwhilewriting` (and the other "while writing" variants) spread its puts across `[0, bgwriter_num)` instead of `[0, num)`. Without this the readers and writer share a key range and you can't have both a hot read subset and meaningful compaction work — this lets you have both.

### Benchmark

Setup: Ubuntu 24.04 (kernel 7.0.5, OrbStack Linux VM on Apple Silicon), 14 vCPUs, virtio-blk disk, btrfs. MGLRU disabled (`echo 0 > /sys/kernel/mm/lru_gen/enabled`) so the kernel uses the classic active/inactive LRU. 14 GB DB (3.5M keys × 4 KB values), no compression. Each measurement run is pinned to a 1 GB cgroup via `systemd-run --scope -p MemoryMax=1G -p MemorySwapMax=0`. Page cache is dropped between configs. db_bench is Release build.

Workload: `readwhilewriting` for 120s. 4 reader threads doing random reads over a hot key subset, plus 1 writer thread spreading overwrites across the full 3.5M-key keyspace (via `--bgwriter_num=3500000`) throttled at 200 MB/s, so there's continuous compaction running while the readers go.

The size of the hot reader subset relative to available page cache controls how visible the optimization is. The Cassandra blog ([Lightfoot 2026](https://lightfoot.dev/direct-i-o-for-cassandra-compaction-cutting-p99-read-latency-by-5x/)) documented the same thing: biggest wins when the hot set is big enough to actually compete for cache, smaller wins when the hot set trivially fits, neutral when the hot set is way bigger than cache. So I ran two hot-set sizes.

#### Small hot set: ~30 MB (~3% of the 1 GB cgroup) — N=5 iterations, mean (CV)

`--num=7500`. The hot set is small enough that the page cache holds it without much trouble even under compaction, so the wins here are real but on the modest side.

| Config | Throughput (ops/s) | Read P50 (µs) | Read P99 (µs) | Read P99.9 (µs) | Read P99.99 (µs) |
|---|---|---|---|---|---|
| buffered (default) | 233,477 (8.2%) | 16.09 | 82.24 | 721.0 | 2,102.5 |
| direct_compaction_writes_only (existing knob alone) | 287,405 (2.8%) — **+23.1%** | 13.00 (−19.2%) | **66.77 (−18.8%)** | 553.9 (−23.2%) | 1,787.6 (−15.0%) |
| direct_compaction_read_only (new knob alone) | 250,669 (2.4%) — +7.4% | 14.16 (−12.0%) | 102.99 (+25.2%) | 689.8 (−4.3%) | 1,801.3 (−14.3%) |
| direct_compaction_read_write (new + existing, recommended) | 277,920 (3.3%) — **+19.0%** | **12.99 (−19.3%)** | 84.23 (+2.4%) | 613.4 (−14.9%) | **1,738.2 (−17.3%)** |
| use_direct_reads=true (existing global) + write-side | 249,014 (2.5%) — +6.7% | 15.95 (−0.9%) | 68.78 (−16.4%) | **450.8 (−37.5%)** | 1,814.5 (−13.7%) |

CV is 2.4–3.3% on the optimized configs (8.2% on buffered), so the deltas are real. With a hot set this small, the existing `use_direct_io_for_flush_and_compaction` knob is already doing most of the work — the new flag's main extra contribution here is P99.99 (combined wins it by ~2 points vs writes-only-alone). Worth noting: the new flag *alone* (without the existing write-side flag) improves P99.99 but regresses P99 by 25% on this small-hot-set workload, because direct compaction reads lose kernel readahead and compaction-output writes are still hitting the page cache. That regression goes away once you combine with the existing write-side flag, or once the hot set is bigger (see next table). So if you're using just one knob, use the existing one. If you're using this PR's flag, pair it with `use_direct_io_for_flush_and_compaction=true`.

#### Larger hot set: ~400 MB (~40% of cache) — N=5 iterations, mean (CV)

`--num=100000`. This is the case the Cassandra blog calls out — hot set big enough to actually fight compaction for cache. Their analogous setup (1M hot partitions, ~33% hot/cache) reported 1.93× p99 improvement. Numbers here are the headline:

| Config | Throughput (ops/s) | Read P50 (µs) | Read P99 (µs) | Read P99.9 (µs) | Read P99.99 (µs) |
|---|---|---|---|---|---|
| buffered (default) | 68,959 (7.7%) | 44.81 | 541.22 | 2,225.2 | 11,334.5 |
| direct_compaction_writes_only (existing knob alone) | 73,973 (10.3%) — +7.3% | 42.22 (−5.8%) | 456.27 (−15.7%) | 2,016.9 (−9.4%) | 9,190.0 (−18.9%) |
| direct_compaction_read_only (new knob alone) | 84,337 (2.3%) — +22.3% | 38.66 (−13.7%) | 386.97 (−28.5%) | 1,644.8 (−26.1%) | 4,837.9 (−57.3%, 2.34×) |
| direct_compaction_read_write (new + existing, recommended) | **104,923 (8.4%) — +52.2%** | **34.26 (−23.5%)** | **290.97 (−46.2%)** | **1,143.4 (−48.6%)** | **3,080.3 (−72.8%, 3.68×)** |
| use_direct_reads=true (existing global) + write-side | 71,598 (9.1%) — +3.8% | 51.33 (+14.5%) | 297.91 (−45.0%) | 1,663.6 (−25.2%) | 6,530.0 (−42.4%) |

Combined config gets a 3.68× p99.99 win, 1.86× p99, p50 down 23%, throughput up 52%. Same shape as the Cassandra blog's 1.93× p99 result — the improvement just lands at deeper percentiles for us because RocksDB's baseline data path is roughly 40× faster than Cassandra's (their buffered p99 was 35 ms, ours is 0.54 ms), so the cache-miss tail is further out.

A few things worth calling out from this table:

- The new flag is doing real work on top of the existing write-side flag here, not just shifting things around. Combined throughput is +42% over `direct_compaction_writes_only` alone, and combined p99.99 is 3× better. The existing knob alone gives a fairly modest +7% throughput / -19% p99.99 in this case — there's a clear gap that the new flag fills.
- The new flag *alone* (no existing write-side flag) is also a real improvement here: +22% throughput, p99.99 down 57%. The P99 regression we saw in the small-hot-set case is gone, because the cache-protection effect now dominates the lost-readahead cost.
- `use_direct_reads=true` (the existing global flag) actually regresses P50 by 14.5% in this workload — taking user reads off the page cache hurts you when the hot data could have been cached. It also gets the worst throughput of any direct config. It's not an equivalent way to get these gains.

### `compaction_readahead_size` matters when this flag is on

Direct I/O bypasses kernel readahead, so RocksDB's own `DBOptions::compaction_readahead_size` becomes the only prefetch the iterator has. The default of 2 MB is enough and real users will get it automatically. **But `db_bench`'s `--compaction_readahead_size` CLI default is 0**, which defeats prefetch and makes direct compaction look slower than it actually is. If you're reproducing the numbers above, pass `--compaction_readahead_size=2097152` (or larger).

- Recommended production config is `use_direct_io_for_compaction_reads=true` + `use_direct_io_for_flush_and_compaction=true`. Strongest configuration at every percentile and throughput in both benches.
- The new flag is the read-side counterpart to `use_direct_io_for_flush_and_compaction`, which handles compaction-write cache pollution. They address different sources of pollution and compose. The gap between "combined" and "writes-only-alone" is 17 percentage points on p99.99 in the small-hot-set bench and 54 points in the larger one, so the new flag is contributing real value, especially as the hot set grows.
- The new flag alone is also a real improvement when the hot set is big enough to compete with cache (+22% throughput, 2.34× p99.99 in the larger-hot-set bench). On a very small hot set it improves p99.99 but regresses p99, so pairing with the existing write-side flag is safer.
- The benefit is workload-dependent. Small hot sets get modest tail-latency wins. Hot sets sized to actually compete for cache get the big multi-percentile wins shown above. Hot sets bigger than cache (not benched here but covered in the Cassandra blog) see no change either way — every read misses regardless.

### Reproducing

Any Linux host (or a Linux VM on macOS via OrbStack / Multipass / lima):

```bash
sudo apt-get install -y build-essential clang cmake git pkg-config \
  libgflags-dev libsnappy-dev zlib1g-dev libbz2-dev liblz4-dev libzstd-dev

cmake -DCMAKE_BUILD_TYPE=Release -DPORTABLE=1 -DWITH_GFLAGS=1 -DWITH_TESTS=0 ..
make -j db_bench

echo 0 | sudo tee /sys/kernel/mm/lru_gen/enabled
```

Build the source DB once, unrestricted memory:

```bash
./db_bench --benchmarks=fillrandom,compact,waitforcompaction,stats \
  --db=/path/to/source_db --num=3500000 --key_size=16 --value_size=4096 \
  --write_buffer_size=16777216 --target_file_size_base=16777216 \
  --max_background_jobs=4 --compression_type=none --cache_size=4194304 \
  --max_bytes_for_level_base=67108864 --disable_wal=1 --sync=0
```

For each config, copy `source_db -> scratch_db`, run `sync && echo 3 > /proc/sys/vm/drop_caches`, then:

```bash
sudo systemd-run --scope -p MemoryMax=1G -p MemorySwapMax=0 \
  ./db_bench --use_existing_db=1 \
    --benchmarks=readwhilewriting,stats --db=/path/to/scratch_db \
    --threads=5 --duration=120 --statistics=true --histogram=1 \
    --num=7500 --bgwriter_num=3500000 \
    --key_size=16 --value_size=4096 \
    --write_buffer_size=16777216 --target_file_size_base=16777216 \
    --max_background_jobs=4 --compression_type=none \
    --cache_size=4194304 --open_files=200 \
    --skip_stats_update_on_db_open=true \
    --max_bytes_for_level_base=67108864 \
    --benchmark_write_rate_limit=209715200 \
    --compaction_readahead_size=2097152 \
    --rate_limiter_bytes_per_sec=0 \
    --use_direct_reads={true|false} \
    --use_direct_io_for_compaction_reads={true|false} \
    --use_direct_io_for_flush_and_compaction={true|false}
```

For the larger hot-set table, change `--num=7500` to `--num=100000`.

The five configs in the tables:
- `buffered`: all three flags false.
- `direct_compaction_writes_only`: `use_direct_io_for_flush_and_compaction=true`, the other two false. This is what users have today without this PR.
- `direct_compaction_read_only`: `use_direct_io_for_compaction_reads=true`, the other two false.
- `direct_compaction_read_write`: `use_direct_io_for_compaction_reads=true`, `use_direct_io_for_flush_and_compaction=true`, `use_direct_reads=false`. **Recommended.**
- `direct_all`: `use_direct_reads=true`, `use_direct_io_for_flush_and_compaction=true`, `use_direct_io_for_compaction_reads=false`.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/14743

Reviewed By: pdillinger

Differential Revision: D108017601

Pulled By: xingbowang

fbshipit-source-id: 4039d490d7e77b476db7a477a2f3d24738db6336
2026-06-09 17:02:53 -07:00

780 lines
26 KiB
C++

// 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.
#include "test_util/testutil.h"
#include <fcntl.h>
#include <sys/stat.h>
#include <array>
#include <cctype>
#include <fstream>
#include <sstream>
#include "db/memtable_list.h"
#include "env/composite_env_wrapper.h"
#include "file/random_access_file_reader.h"
#include "file/sequence_file_reader.h"
#include "file/writable_file_writer.h"
#include "port/port.h"
#include "rocksdb/convenience.h"
#include "rocksdb/system_clock.h"
#include "rocksdb/utilities/object_registry.h"
#include "test_util/mock_time_env.h"
#include "test_util/sync_point.h"
#include "util/random.h"
#include "util/string_util.h"
#ifndef ROCKSDB_UNITTESTS_WITH_CUSTOM_OBJECTS_FROM_STATIC_LIBS
void RegisterCustomObjects(int /*argc*/, char** /*argv*/) {}
#endif
namespace ROCKSDB_NAMESPACE::test {
const uint32_t kDefaultFormatVersion = BlockBasedTableOptions().format_version;
const std::set<uint32_t> kFooterFormatVersionsToTest{
// Non-legacy, before big footer changes
5U,
// After big footer changes
6U,
// In case any interesting future changes
kDefaultFormatVersion,
kLatestBbtFormatVersion,
};
const ReadOptionsNoIo kReadOptionsNoIo;
std::string RandomKey(Random* rnd, int len, RandomKeyType type) {
// Make sure to generate a wide variety of characters so we
// test the boundary conditions for short-key optimizations.
static const char kTestChars[] = {'\0', '\1', 'a', 'b', 'c',
'd', 'e', '\xfd', '\xfe', '\xff'};
std::string result;
for (int i = 0; i < len; i++) {
std::size_t indx = 0;
switch (type) {
case RandomKeyType::RANDOM:
indx = rnd->Uniform(sizeof(kTestChars));
break;
case RandomKeyType::LARGEST:
indx = sizeof(kTestChars) - 1;
break;
case RandomKeyType::MIDDLE:
indx = sizeof(kTestChars) / 2;
break;
case RandomKeyType::SMALLEST:
indx = 0;
break;
}
result += kTestChars[indx];
}
return result;
}
const std::vector<UserDefinedTimestampTestMode>& GetUDTTestModes() {
static std::vector<UserDefinedTimestampTestMode> udt_test_modes = {
UserDefinedTimestampTestMode::kStripUserDefinedTimestamp,
UserDefinedTimestampTestMode::kNormal,
UserDefinedTimestampTestMode::kNone};
return udt_test_modes;
}
bool IsUDTEnabled(const UserDefinedTimestampTestMode& test_mode) {
return test_mode != UserDefinedTimestampTestMode::kNone;
}
bool ShouldPersistUDT(const UserDefinedTimestampTestMode& test_mode) {
return test_mode != UserDefinedTimestampTestMode::kStripUserDefinedTimestamp;
}
Slice CompressibleString(Random* rnd, double compressed_to_fraction, int len,
std::string* dst) {
int raw = static_cast<int>(len * compressed_to_fraction);
if (raw < 1) {
raw = 1;
}
std::string raw_data = rnd->RandomBinaryString(raw);
// Duplicate the random data until we have filled "len" bytes
dst->clear();
while (dst->size() < (unsigned int)len) {
dst->append(raw_data);
}
dst->resize(len);
return Slice(*dst);
}
namespace {
class Uint64ComparatorImpl : public Comparator {
public:
Uint64ComparatorImpl() = default;
const char* Name() const override { return "rocksdb.Uint64Comparator"; }
int Compare(const Slice& a, const Slice& b) const override {
assert(a.size() == sizeof(uint64_t) && b.size() == sizeof(uint64_t));
const uint64_t* left = reinterpret_cast<const uint64_t*>(a.data());
const uint64_t* right = reinterpret_cast<const uint64_t*>(b.data());
uint64_t leftValue;
uint64_t rightValue;
GetUnaligned(left, &leftValue);
GetUnaligned(right, &rightValue);
if (leftValue == rightValue) {
return 0;
} else if (leftValue < rightValue) {
return -1;
} else {
return 1;
}
}
void FindShortestSeparator(std::string* /*start*/,
const Slice& /*limit*/) const override {}
void FindShortSuccessor(std::string* /*key*/) const override {}
};
} // namespace
const Comparator* Uint64Comparator() {
static Uint64ComparatorImpl uint64comp;
return &uint64comp;
}
const Comparator* BytewiseComparatorWithU64TsWrapper() {
ConfigOptions config_options;
const Comparator* user_comparator = nullptr;
Status s = Comparator::CreateFromString(
config_options, "leveldb.BytewiseComparator.u64ts", &user_comparator);
s.PermitUncheckedError();
return user_comparator;
}
const Comparator* ReverseBytewiseComparatorWithU64TsWrapper() {
ConfigOptions config_options;
const Comparator* user_comparator = nullptr;
Status s = Comparator::CreateFromString(
config_options, "rocksdb.ReverseBytewiseComparator.u64ts",
&user_comparator);
s.PermitUncheckedError();
return user_comparator;
}
void CorruptKeyType(InternalKey* ikey) {
std::string keystr = ikey->Encode().ToString();
keystr[keystr.size() - 8] = kTypeLogData;
ikey->DecodeFrom(Slice(keystr.data(), keystr.size()));
}
std::string KeyStr(const std::string& user_key, const SequenceNumber& seq,
const ValueType& t, bool corrupt) {
InternalKey k(user_key, seq, t);
if (corrupt) {
CorruptKeyType(&k);
}
return k.Encode().ToString();
}
std::string KeyStr(uint64_t ts, const std::string& user_key,
const SequenceNumber& seq, const ValueType& t,
bool corrupt) {
std::string user_key_with_ts(user_key);
std::string ts_str;
PutFixed64(&ts_str, ts);
user_key_with_ts.append(ts_str);
return KeyStr(user_key_with_ts, seq, t, corrupt);
}
bool SleepingBackgroundTask::TimedWaitUntilSleeping(uint64_t wait_time) {
auto abs_time = SystemClock::Default()->NowMicros() + wait_time;
MutexLock l(&mutex_);
while (!sleeping_ || !should_sleep_) {
if (bg_cv_.TimedWait(abs_time)) {
return true;
}
}
return false;
}
bool SleepingBackgroundTask::TimedWaitUntilDone(uint64_t wait_time) {
auto abs_time = SystemClock::Default()->NowMicros() + wait_time;
MutexLock l(&mutex_);
while (!done_with_sleep_) {
if (bg_cv_.TimedWait(abs_time)) {
return true;
}
}
return false;
}
std::string RandomName(Random* rnd, const size_t len) {
std::stringstream ss;
for (size_t i = 0; i < len; ++i) {
ss << static_cast<char>(rnd->Uniform(26) + 'a');
}
return ss.str();
}
CompressionType RandomCompressionType(Random* rnd) {
auto ret = static_cast<CompressionType>(rnd->Uniform(6));
while (!CompressionTypeSupported(ret)) {
ret = static_cast<CompressionType>((static_cast<int>(ret) + 1) % 6);
}
return ret;
}
void RandomCompressionTypeVector(const size_t count,
std::vector<CompressionType>* types,
Random* rnd) {
types->clear();
for (size_t i = 0; i < count; ++i) {
types->emplace_back(RandomCompressionType(rnd));
}
}
const SliceTransform* RandomSliceTransform(Random* rnd, int pre_defined) {
int random_num = pre_defined >= 0 ? pre_defined : rnd->Uniform(4);
switch (random_num) {
case 0:
return NewFixedPrefixTransform(rnd->Uniform(20) + 1);
case 1:
return NewCappedPrefixTransform(rnd->Uniform(20) + 1);
case 2:
return NewNoopTransform();
default:
return nullptr;
}
}
BlockBasedTableOptions RandomBlockBasedTableOptions(Random* rnd) {
BlockBasedTableOptions opt;
opt.cache_index_and_filter_blocks = rnd->Uniform(2);
opt.pin_l0_filter_and_index_blocks_in_cache = rnd->Uniform(2);
opt.pin_top_level_index_and_filter = rnd->Uniform(2);
using IndexType = BlockBasedTableOptions::IndexType;
const std::array<IndexType, 4> index_types = {
{IndexType::kBinarySearch, IndexType::kHashSearch,
IndexType::kTwoLevelIndexSearch, IndexType::kBinarySearchWithFirstKey}};
opt.index_type =
index_types[rnd->Uniform(static_cast<int>(index_types.size()))];
opt.checksum = static_cast<ChecksumType>(rnd->Uniform(3));
opt.block_size = rnd->Uniform(10000000);
opt.block_size_deviation = rnd->Uniform(100);
opt.block_restart_interval = rnd->Uniform(100);
opt.index_block_restart_interval = rnd->Uniform(100);
opt.whole_key_filtering = rnd->Uniform(2);
return opt;
}
TableFactory* RandomTableFactory(Random* rnd, int pre_defined) {
int random_num = pre_defined >= 0 ? pre_defined : rnd->Uniform(4);
switch (random_num) {
case 0:
return NewPlainTableFactory();
case 1:
return NewCuckooTableFactory();
default:
return NewBlockBasedTableFactory();
}
}
MergeOperator* RandomMergeOperator(Random* rnd) {
return new ChanglingMergeOperator(RandomName(rnd, 10));
}
CompactionFilter* RandomCompactionFilter(Random* rnd) {
return new ChanglingCompactionFilter(RandomName(rnd, 10));
}
CompactionFilterFactory* RandomCompactionFilterFactory(Random* rnd) {
return new ChanglingCompactionFilterFactory(RandomName(rnd, 10));
}
void RandomInitDBOptions(DBOptions* db_opt, Random* rnd) {
// boolean options
db_opt->advise_random_on_open = rnd->Uniform(2);
db_opt->allow_mmap_reads = rnd->Uniform(2);
db_opt->allow_mmap_writes = rnd->Uniform(2);
db_opt->use_direct_reads = rnd->Uniform(2);
db_opt->use_direct_io_for_compaction_reads = rnd->Uniform(2);
db_opt->use_direct_io_for_flush_and_compaction = rnd->Uniform(2);
db_opt->create_if_missing = rnd->Uniform(2);
db_opt->create_missing_column_families = rnd->Uniform(2);
db_opt->enable_thread_tracking = rnd->Uniform(2);
db_opt->error_if_exists = rnd->Uniform(2);
db_opt->is_fd_close_on_exec = rnd->Uniform(2);
db_opt->paranoid_checks = rnd->Uniform(2);
db_opt->track_and_verify_wals_in_manifest = rnd->Uniform(2);
db_opt->track_and_verify_wals = rnd->Uniform(2);
db_opt->verify_sst_unique_id_in_manifest = rnd->Uniform(2);
db_opt->skip_stats_update_on_db_open = rnd->Uniform(2);
db_opt->use_adaptive_mutex = rnd->Uniform(2);
db_opt->use_fsync = rnd->Uniform(2);
db_opt->recycle_log_file_num = rnd->Uniform(2);
db_opt->async_wal_precreate = rnd->Uniform(2);
db_opt->avoid_flush_during_recovery = rnd->Uniform(2);
db_opt->avoid_flush_during_shutdown = rnd->Uniform(2);
db_opt->enforce_single_del_contracts = rnd->Uniform(2);
// int options
db_opt->max_background_compactions = rnd->Uniform(100);
db_opt->max_background_flushes = rnd->Uniform(100);
db_opt->max_file_opening_threads = rnd->Uniform(100);
db_opt->max_open_files = rnd->Uniform(100);
db_opt->table_cache_numshardbits = rnd->Uniform(100);
// size_t options
db_opt->db_write_buffer_size = rnd->Uniform(10000);
db_opt->keep_log_file_num = rnd->Uniform(10000);
db_opt->log_file_time_to_roll = rnd->Uniform(10000);
db_opt->manifest_preallocation_size = rnd->Uniform(10000);
db_opt->max_log_file_size = rnd->Uniform(10000);
// std::string options
db_opt->db_log_dir = "path/to/db_log_dir";
db_opt->wal_dir = "path/to/wal_dir";
// uint32_t options
db_opt->max_subcompactions = rnd->Uniform(100000);
// uint64_t options
static const uint64_t uint_max = static_cast<uint64_t>(UINT_MAX);
db_opt->WAL_size_limit_MB = uint_max + rnd->Uniform(100000);
db_opt->WAL_ttl_seconds = uint_max + rnd->Uniform(100000);
db_opt->bytes_per_sync = uint_max + rnd->Uniform(100000);
db_opt->delayed_write_rate = uint_max + rnd->Uniform(100000);
db_opt->delete_obsolete_files_period_micros = uint_max + rnd->Uniform(100000);
db_opt->max_manifest_file_size = uint_max + rnd->Uniform(100000);
db_opt->max_total_wal_size = uint_max + rnd->Uniform(100000);
db_opt->wal_bytes_per_sync = uint_max + rnd->Uniform(100000);
// unsigned int options
db_opt->stats_dump_period_sec = rnd->Uniform(100000);
}
void RandomInitCFOptions(ColumnFamilyOptions* cf_opt, DBOptions& db_options,
Random* rnd) {
cf_opt->compaction_style = (CompactionStyle)(rnd->Uniform(4));
// boolean options
cf_opt->report_bg_io_stats = rnd->Uniform(2);
cf_opt->disable_auto_compactions = rnd->Uniform(2);
cf_opt->inplace_update_support = rnd->Uniform(2);
cf_opt->level_compaction_dynamic_level_bytes = rnd->Uniform(2);
cf_opt->optimize_filters_for_hits = rnd->Uniform(2);
cf_opt->paranoid_file_checks = rnd->Uniform(2);
cf_opt->force_consistency_checks = rnd->Uniform(2);
cf_opt->compaction_options_fifo.allow_compaction = rnd->Uniform(2);
cf_opt->memtable_whole_key_filtering = rnd->Uniform(2);
cf_opt->enable_blob_files = rnd->Uniform(2);
cf_opt->enable_blob_garbage_collection = rnd->Uniform(2);
cf_opt->strict_max_successive_merges = rnd->Uniform(2);
// double options
cf_opt->memtable_prefix_bloom_size_ratio =
static_cast<double>(rnd->Uniform(10000)) / 20000.0;
cf_opt->blob_garbage_collection_age_cutoff = rnd->Uniform(10000) / 10000.0;
cf_opt->blob_garbage_collection_force_threshold =
rnd->Uniform(10000) / 10000.0;
// int options
cf_opt->level0_file_num_compaction_trigger = rnd->Uniform(100);
cf_opt->level0_slowdown_writes_trigger = rnd->Uniform(100);
cf_opt->level0_stop_writes_trigger = rnd->Uniform(100);
cf_opt->max_bytes_for_level_multiplier = rnd->Uniform(100);
cf_opt->max_write_buffer_number = rnd->Uniform(100);
cf_opt->max_write_buffer_size_to_maintain = rnd->Uniform(10000);
cf_opt->min_write_buffer_number_to_merge = rnd->Uniform(100);
cf_opt->num_levels = rnd->Uniform(100);
cf_opt->target_file_size_multiplier = rnd->Uniform(100);
// vector int options
cf_opt->max_bytes_for_level_multiplier_additional.resize(cf_opt->num_levels);
for (int i = 0; i < cf_opt->num_levels; i++) {
cf_opt->max_bytes_for_level_multiplier_additional[i] = rnd->Uniform(100);
}
// size_t options
cf_opt->arena_block_size = rnd->Uniform(10000);
cf_opt->inplace_update_num_locks = rnd->Uniform(10000);
cf_opt->max_successive_merges = rnd->Uniform(10000);
cf_opt->memtable_huge_page_size = rnd->Uniform(10000);
cf_opt->write_buffer_size = rnd->Uniform(10000);
// uint32_t options
cf_opt->bloom_locality = rnd->Uniform(10000);
cf_opt->max_bytes_for_level_base = rnd->Uniform(10000);
// uint64_t options
static const uint64_t uint_max = static_cast<uint64_t>(UINT_MAX);
cf_opt->ttl =
db_options.max_open_files == -1 ? uint_max + rnd->Uniform(10000) : 0;
cf_opt->periodic_compaction_seconds =
db_options.max_open_files == -1 ? uint_max + rnd->Uniform(10000) : 0;
cf_opt->max_sequential_skip_in_iterations = uint_max + rnd->Uniform(10000);
cf_opt->target_file_size_base = uint_max + rnd->Uniform(10000);
cf_opt->max_compaction_bytes =
cf_opt->target_file_size_base * rnd->Uniform(100);
cf_opt->compaction_options_fifo.max_table_files_size =
uint_max + rnd->Uniform(10000);
cf_opt->min_blob_size = uint_max + rnd->Uniform(10000);
cf_opt->blob_file_size = uint_max + rnd->Uniform(10000);
cf_opt->blob_compaction_readahead_size = uint_max + rnd->Uniform(10000);
// pointer typed options
cf_opt->prefix_extractor.reset(RandomSliceTransform(rnd));
cf_opt->table_factory.reset(RandomTableFactory(rnd));
cf_opt->merge_operator.reset(RandomMergeOperator(rnd));
if (cf_opt->compaction_filter) {
delete cf_opt->compaction_filter;
}
cf_opt->compaction_filter = RandomCompactionFilter(rnd);
cf_opt->compaction_filter_factory.reset(RandomCompactionFilterFactory(rnd));
// custom typed options
cf_opt->compression = RandomCompressionType(rnd);
RandomCompressionTypeVector(cf_opt->num_levels,
&cf_opt->compression_per_level, rnd);
cf_opt->blob_compression_type = RandomCompressionType(rnd);
}
bool IsDirectIOSupported(Env* env, const std::string& dir) {
EnvOptions env_options;
env_options.use_mmap_writes = false;
env_options.use_direct_writes = true;
std::string tmp = TempFileName(dir, 999);
Status s;
{
std::unique_ptr<WritableFile> file;
s = env->NewWritableFile(tmp, &file, env_options);
}
if (s.ok()) {
s = env->DeleteFile(tmp);
}
return s.ok();
}
bool IsPrefetchSupported(const std::shared_ptr<FileSystem>& fs,
const std::string& dir) {
bool supported = false;
std::string tmp = TempFileName(dir, 999);
Random rnd(301);
std::string test_string = rnd.RandomString(4096);
Slice data(test_string);
IOOptions opts;
Status s = WriteStringToFile(fs.get(), data, tmp, true, opts);
if (s.ok()) {
std::unique_ptr<FSRandomAccessFile> file;
auto io_s = fs->NewRandomAccessFile(tmp, FileOptions(), &file, nullptr);
if (io_s.ok()) {
supported =
!(file->Prefetch(0, data.size(), opts, nullptr).IsNotSupported());
}
s = fs->DeleteFile(tmp, opts, nullptr);
}
return s.ok() && supported;
}
size_t GetLinesCount(const std::string& fname, const std::string& pattern) {
std::stringstream ssbuf;
std::string line;
size_t count = 0;
std::ifstream inFile(fname.c_str());
ssbuf << inFile.rdbuf();
while (getline(ssbuf, line)) {
if (line.find(pattern) != std::string::npos) {
count++;
}
}
return count;
}
Status CorruptFile(Env* env, const std::string& fname, int offset,
int bytes_to_corrupt, bool verify_checksum /*=true*/) {
uint64_t size;
Status s = env->GetFileSize(fname, &size);
if (!s.ok()) {
return s;
} else if (offset < 0) {
// Relative to end of file; make it absolute
if (-offset > static_cast<int>(size)) {
offset = 0;
} else {
offset = static_cast<int>(size + offset);
}
}
if (offset > static_cast<int>(size)) {
offset = static_cast<int>(size);
}
if (offset + bytes_to_corrupt > static_cast<int>(size)) {
bytes_to_corrupt = static_cast<int>(size - offset);
}
// Do it
std::string contents;
s = ReadFileToString(env, fname, &contents);
if (s.ok()) {
for (int i = 0; i < bytes_to_corrupt; i++) {
contents[i + offset] ^= 0x80;
}
s = WriteStringToFile(env, contents, fname, false /* should_sync */);
}
if (s.ok() && verify_checksum) {
Options options;
options.env = env;
EnvOptions env_options;
Status v = VerifySstFileChecksum(options, env_options, fname);
assert(!v.ok());
}
return s;
}
Status TruncateFile(Env* env, const std::string& fname, uint64_t new_length) {
uint64_t old_length;
Status s = env->GetFileSize(fname, &old_length);
if (!s.ok() || new_length == old_length) {
return s;
}
// Do it
std::string contents;
s = ReadFileToString(env, fname, &contents);
if (s.ok()) {
contents.resize(static_cast<size_t>(new_length), 'b');
s = WriteStringToFile(env, contents, fname, false /* should_sync */);
}
return s;
}
// Try and delete a directory if it exists
Status TryDeleteDir(Env* env, const std::string& dirname) {
bool is_dir = false;
Status s = env->IsDirectory(dirname, &is_dir);
if (s.ok() && is_dir) {
s = env->DeleteDir(dirname);
}
return s;
}
// Delete a directory if it exists
void DeleteDir(Env* env, const std::string& dirname) {
TryDeleteDir(env, dirname).PermitUncheckedError();
}
FileType GetFileType(const std::string& path) {
FileType type = kTempFile;
std::size_t found = path.find_last_of('/');
if (found == std::string::npos) {
found = 0;
}
std::string file_name = path.substr(found);
uint64_t number = 0;
ParseFileName(file_name, &number, &type);
return type;
}
uint64_t GetFileNumber(const std::string& path) {
FileType type = kTempFile;
std::size_t found = path.find_last_of('/');
if (found == std::string::npos) {
found = 0;
}
std::string file_name = path.substr(found);
uint64_t number = 0;
ParseFileName(file_name, &number, &type);
return number;
}
Status CreateEnvFromSystem(const ConfigOptions& config_options, Env** result,
std::shared_ptr<Env>* guard) {
const char* env_uri = getenv("TEST_ENV_URI");
const char* fs_uri = getenv("TEST_FS_URI");
if (env_uri || fs_uri) {
return Env::CreateFromUri(config_options,
(env_uri != nullptr) ? env_uri : "",
(fs_uri != nullptr) ? fs_uri : "", result, guard);
} else {
// Neither specified. Use the default
*result = config_options.env;
guard->reset();
return Status::OK();
}
}
namespace {
// A hacky skip list mem table that triggers flush after number of entries.
class SpecialMemTableRep : public MemTableRep {
public:
explicit SpecialMemTableRep(Allocator* allocator, MemTableRep* memtable,
int num_entries_flush)
: MemTableRep(allocator),
memtable_(memtable),
num_entries_flush_(num_entries_flush),
num_entries_(0) {}
KeyHandle Allocate(const size_t len, char** buf) override {
return memtable_->Allocate(len, buf);
}
// Insert key into the list.
// REQUIRES: nothing that compares equal to key is currently in the list.
void Insert(KeyHandle handle) override {
num_entries_++;
memtable_->Insert(handle);
}
void InsertConcurrently(KeyHandle handle) override {
num_entries_++;
memtable_->Insert(handle);
}
// Returns true iff an entry that compares equal to key is in the list.
bool Contains(const char* key) const override {
return memtable_->Contains(key);
}
size_t ApproximateMemoryUsage() override {
// Return a high memory usage when number of entries exceeds the threshold
// to trigger a flush.
return (num_entries_ < num_entries_flush_) ? 0 : 1024 * 1024 * 1024;
}
void Get(const LookupKey& k, void* callback_args,
bool (*callback_func)(void* arg, const char* entry)) override {
memtable_->Get(k, callback_args, callback_func);
}
uint64_t ApproximateNumEntries(const Slice& start_ikey,
const Slice& end_ikey) override {
return memtable_->ApproximateNumEntries(start_ikey, end_ikey);
}
MemTableRep::Iterator* GetIterator(Arena* arena = nullptr) override {
return memtable_->GetIterator(arena);
}
~SpecialMemTableRep() override = default;
private:
std::unique_ptr<MemTableRep> memtable_;
int num_entries_flush_;
int num_entries_;
};
class SpecialSkipListFactory : public MemTableRepFactory {
public:
static bool Register(ObjectLibrary& library, const std::string& /*arg*/) {
library.AddFactory<MemTableRepFactory>(
ObjectLibrary::PatternEntry(SpecialSkipListFactory::kClassName(), true)
.AddNumber(":"),
[](const std::string& uri, std::unique_ptr<MemTableRepFactory>* guard,
std::string* /* errmsg */) {
auto colon = uri.find(':');
if (colon != std::string::npos) {
auto count = ParseInt(uri.substr(colon + 1));
guard->reset(new SpecialSkipListFactory(count));
} else {
guard->reset(new SpecialSkipListFactory(2));
}
return guard->get();
});
return true;
}
// After number of inserts >= `num_entries_flush` in a mem table, trigger
// flush.
explicit SpecialSkipListFactory(int num_entries_flush)
: num_entries_flush_(num_entries_flush) {}
using MemTableRepFactory::CreateMemTableRep;
MemTableRep* CreateMemTableRep(const MemTableRep::KeyComparator& compare,
Allocator* allocator,
const SliceTransform* transform,
Logger* /*logger*/) override {
return new SpecialMemTableRep(
allocator,
factory_.CreateMemTableRep(compare, allocator, transform, nullptr),
num_entries_flush_);
}
static const char* kClassName() { return "SpecialSkipListFactory"; }
const char* Name() const override { return kClassName(); }
std::string GetId() const override {
std::string id = Name();
if (num_entries_flush_ > 0) {
id.append(":").append(std::to_string(num_entries_flush_));
}
return id;
}
bool IsInsertConcurrentlySupported() const override {
return factory_.IsInsertConcurrentlySupported();
}
private:
SkipListFactory factory_;
int num_entries_flush_;
};
} // namespace
MemTableRepFactory* NewSpecialSkipListFactory(int num_entries_per_flush) {
RegisterTestLibrary();
return new SpecialSkipListFactory(num_entries_per_flush);
}
// This method loads existing test classes into the ObjectRegistry
int RegisterTestObjects(ObjectLibrary& library, const std::string& arg) {
size_t num_types;
library.AddFactory<const Comparator>(
test::SimpleSuffixReverseComparator::kClassName(),
[](const std::string& /*uri*/,
std::unique_ptr<const Comparator>* /*guard*/,
std::string* /* errmsg */) {
static test::SimpleSuffixReverseComparator ssrc;
return &ssrc;
});
SpecialSkipListFactory::Register(library, arg);
library.AddFactory<MergeOperator>(
"Changling",
[](const std::string& uri, std::unique_ptr<MergeOperator>* guard,
std::string* /* errmsg */) {
guard->reset(new test::ChanglingMergeOperator(uri));
return guard->get();
});
library.AddFactory<CompactionFilter>(
"Changling",
[](const std::string& uri, std::unique_ptr<CompactionFilter>* /*guard*/,
std::string* /* errmsg */) {
return new test::ChanglingCompactionFilter(uri);
});
library.AddFactory<CompactionFilterFactory>(
"Changling", [](const std::string& uri,
std::unique_ptr<CompactionFilterFactory>* guard,
std::string* /* errmsg */) {
guard->reset(new test::ChanglingCompactionFilterFactory(uri));
return guard->get();
});
library.AddFactory<SystemClock>(
MockSystemClock::kClassName(),
[](const std::string& /*uri*/, std::unique_ptr<SystemClock>* guard,
std::string* /* errmsg */) {
guard->reset(new MockSystemClock(SystemClock::Default()));
return guard->get();
});
return static_cast<int>(library.GetFactoryCount(&num_types));
}
void RegisterTestLibrary(const std::string& arg) {
static bool registered = false;
if (!registered) {
registered = true;
ObjectRegistry::Default()->AddLibrary("test", RegisterTestObjects, arg);
}
}
const std::string kUnitTestDbId = "UnitTest";
} // namespace ROCKSDB_NAMESPACE::test