feat(rocksdb): add optional direct I/O for reads and flush/compaction

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.

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)},

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;

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

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<std::string> 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);

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())
+	}
+}