diff --git a/kvrocks.conf b/kvrocks.conf index d13a13cd120..a0c122924c7 100644 --- a/kvrocks.conf +++ b/kvrocks.conf @@ -1162,6 +1162,22 @@ rocksdb.partition_filters yes # Default: yes # rocksdb.avoid_unnecessary_blocking_io yes +# Use O_DIRECT for reads (bypassing the OS page cache). With buffered reads the page cache +# grows with the working set and, under a hard memory limit (e.g. a cgroup limit), is counted +# against that limit and can trigger the process to be killed. Direct reads keep memory usage +# bounded to the RocksDB block cache, at the cost of losing the page cache as a read cache — so +# size rocksdb.block_cache_size to the hot working set when enabling this. +# see https://github.com/facebook/rocksdb/wiki/Direct-IO +# +# Default: no +# rocksdb.use_direct_reads no + +# Use O_DIRECT for flush and compaction (bypassing the OS page cache on the write path). +# see https://github.com/facebook/rocksdb/wiki/Direct-IO +# +# Default: no +# rocksdb.use_direct_io_for_flush_and_compaction no + # Specifies the maximum size in bytes for a write batch in RocksDB. # If set to 0, there is no size limit for write batches. # This option can help control memory usage and manage large WriteBatch operations more effectively. diff --git a/src/config/config.cc b/src/config/config.cc index 0de6a2c6f47..898dc1fc522 100644 --- a/src/config/config.cc +++ b/src/config/config.cc @@ -328,6 +328,9 @@ Config::Config() { {"rocksdb.rate_limiter_auto_tuned", true, new YesNoField(&rocks_db.rate_limiter_auto_tuned, true)}, {"rocksdb.avoid_unnecessary_blocking_io", true, new YesNoField(&rocks_db.avoid_unnecessary_blocking_io, true)}, {"rocksdb.partition_filters", true, new YesNoField(&rocks_db.partition_filters, true)}, + {"rocksdb.use_direct_reads", true, new YesNoField(&rocks_db.use_direct_reads, false)}, + {"rocksdb.use_direct_io_for_flush_and_compaction", true, + new YesNoField(&rocks_db.use_direct_io_for_flush_and_compaction, false)}, {"rocksdb.max_compaction_bytes", false, new Int64Field(&rocks_db.max_compaction_bytes, 0, 0, INT64_MAX)}, {"rocksdb.sst_file_delete_rate_bytes_per_sec", false, new Int64Field(&rocks_db.sst_file_delete_rate_bytes_per_sec, 0, 0, INT64_MAX)}, diff --git a/src/config/config.h b/src/config/config.h index a6dd19f213c..9b8380c1738 100644 --- a/src/config/config.h +++ b/src/config/config.h @@ -271,6 +271,8 @@ struct Config { bool rate_limiter_auto_tuned; bool avoid_unnecessary_blocking_io = true; bool partition_filters; + bool use_direct_reads = false; + bool use_direct_io_for_flush_and_compaction = false; int64_t max_compaction_bytes; int64_t sst_file_delete_rate_bytes_per_sec = 0; uint64_t periodic_compaction_seconds = kDefaultRocksdbPeriodicCompactionSeconds; diff --git a/src/storage/storage.cc b/src/storage/storage.cc index f328b04f36e..d7cc39712cb 100644 --- a/src/storage/storage.cc +++ b/src/storage/storage.cc @@ -232,6 +232,13 @@ rocksdb::Options Storage::InitRocksDBOptions() { // avoid blocking io on iteration // see https://github.com/facebook/rocksdb/wiki/IO#avoid-blocking-io options.avoid_unnecessary_blocking_io = config_->rocks_db.avoid_unnecessary_blocking_io; + + // Optionally bypass the OS page cache with O_DIRECT. This bounds the process's memory footprint + // to the RocksDB block cache instead of letting the (buffered-read) page cache grow, which is + // useful under a hard memory limit; reads then rely on the block cache, so size it accordingly. + // see https://github.com/facebook/rocksdb/wiki/Direct-IO + options.use_direct_reads = config_->rocks_db.use_direct_reads; + options.use_direct_io_for_flush_and_compaction = config_->rocks_db.use_direct_io_for_flush_and_compaction; return options; } diff --git a/tests/cppunit/config_test.cc b/tests/cppunit/config_test.cc index aff85b86db2..88eb270f87e 100644 --- a/tests/cppunit/config_test.cc +++ b/tests/cppunit/config_test.cc @@ -139,6 +139,8 @@ TEST(Config, GetAndSet) { {"rocksdb.compression_max_dict_bytes", "16384"}, {"rocksdb.compression_zstd_max_train_bytes", "262144"}, {"rocksdb.wal_compression", "no"}, + {"rocksdb.use_direct_reads", "yes"}, + {"rocksdb.use_direct_io_for_flush_and_compaction", "yes"}, {"histogram-bucket-boundaries", "10,100,1000,10000"}, }; @@ -148,6 +150,33 @@ TEST(Config, GetAndSet) { } } +TEST(Config, DirectIO) { + const char *path = "test_direct_io.conf"; + unlink(path); + + Config config; + // Both options are off by default. + EXPECT_FALSE(config.rocks_db.use_direct_reads); + EXPECT_FALSE(config.rocks_db.use_direct_io_for_flush_and_compaction); + + // ... and are parsed from the configuration file. + std::ofstream output_file(path, std::ios::out); + output_file << "rocksdb.use_direct_reads yes" << "\n"; + output_file << "rocksdb.use_direct_io_for_flush_and_compaction yes" << "\n"; + output_file.close(); + ASSERT_TRUE(config.Load(CLIOptions(path)).IsOK()); + EXPECT_TRUE(config.rocks_db.use_direct_reads); + EXPECT_TRUE(config.rocks_db.use_direct_io_for_flush_and_compaction); + + std::vector values; + config.Get("rocksdb.use_direct_reads", &values); + ASSERT_EQ(values.size(), 2); + EXPECT_EQ(values[0], "rocksdb.use_direct_reads"); + EXPECT_EQ(values[1], "yes"); + + unlink(path); +} + TEST(Config, GetRenameCommand) { const char *path = "test.conf"; unlink(path); diff --git a/tests/gocase/unit/config/config_test.go b/tests/gocase/unit/config/config_test.go index d802183447e..1bf3026d0f5 100644 --- a/tests/gocase/unit/config/config_test.go +++ b/tests/gocase/unit/config/config_test.go @@ -522,3 +522,26 @@ func TestConfigDailyOffpeakTimeUTC(t *testing.T) { require.EqualValues(t, "", result[parameter]) }) } + +func TestConfigDirectIO(t *testing.T) { + t.Parallel() + srv := util.StartServer(t, map[string]string{}) + defer srv.Close() + + ctx := context.Background() + rdb := srv.NewClient() + defer func() { require.NoError(t, rdb.Close()) }() + + // The direct-I/O options are off by default and, being applied only when the DB is opened, + // are read-only at runtime (a CONFIG SET must be rejected rather than silently no-op). + for _, parameter := range []string{ + "rocksdb.use_direct_reads", + "rocksdb.use_direct_io_for_flush_and_compaction", + } { + result, err := rdb.ConfigGet(ctx, parameter).Result() + require.NoError(t, err) + require.EqualValues(t, "no", result[parameter]) + + require.Error(t, rdb.ConfigSet(ctx, parameter, "yes").Err()) + } +}