chore(tiering): Make lower bound configurable (#5810)

Author: dranikpg

helio

@@ -14,6 +14,7 @@
 #include <variant>
 
 #include "base/cycle_clock.h"
+#include "base/flag_utils.h"
 #include "base/flags.h"
 #include "base/histogram.h"
 #include "base/io_buf.h"
@@ -22,7 +23,6 @@
 #include "core/heap_size.h"
 #include "facade/conn_context.h"
 #include "facade/dragonfly_listener.h"
-#include "facade/flag_utils.h"
 #include "facade/memcache_parser.h"
 #include "facade/redis_parser.h"
 #include "facade/service_interface.h"
@@ -2178,9 +2178,9 @@ void Connection::UpdateFromFlags() {
 }
 
 std::vector<std::string> Connection::GetMutableFlagNames() {
-  return GetFlagNames(FLAGS_pipeline_queue_limit, FLAGS_pipeline_buffer_limit,
-                      FLAGS_max_busy_read_usec, FLAGS_always_flush_pipeline,
-                      FLAGS_pipeline_squash_limit, FLAGS_pipeline_wait_batch_usec);
+  return base::GetFlagNames(FLAGS_pipeline_queue_limit, FLAGS_pipeline_buffer_limit,
+                            FLAGS_max_busy_read_usec, FLAGS_always_flush_pipeline,
+                            FLAGS_pipeline_squash_limit, FLAGS_pipeline_wait_batch_usec);
 }
 
 void Connection::GetRequestSizeHistogramThreadLocal(std::string* hist) {

src/facade/dragonfly_connection.cc

@@ -31,11 +31,11 @@ extern "C" {
 #include <filesystem>
 
 #include "base/cycle_clock.h"
+#include "base/flag_utils.h"
 #include "base/flags.h"
 #include "base/logging.h"
 #include "facade/dragonfly_connection.h"
 #include "facade/error.h"
-#include "facade/flag_utils.h"
 #include "facade/reply_builder.h"
 #include "facade/reply_capture.h"
 #include "server/acl/acl_commands_def.h"
@@ -742,8 +742,8 @@ void UpdateFromFlagsOnThread() {
 }
 
 std::vector<std::string> GetMutableFlagNames() {
-  return facade::GetFlagNames(FLAGS_shard_thread_busy_polling_usec,
-                              FLAGS_squash_stats_latency_lower_limit);
+  return base::GetFlagNames(FLAGS_shard_thread_busy_polling_usec,
+                            FLAGS_squash_stats_latency_lower_limit);
 }
 
 void UpdateUringFlagsOnThread() {
@@ -951,13 +951,13 @@ void Service::Init(util::AcceptServer* acceptor, std::vector<facade::Listener*>
                        []() { MultiCommandSquasher::UpdateFromFlags(); });
   // Register uring proactor flags
   RegisterMutableFlags(&config_registry,
-                       facade::GetFlagNames(FLAGS_uring_wake_mode, FLAGS_uring_submit_threshold),
+                       base::GetFlagNames(FLAGS_uring_wake_mode, FLAGS_uring_submit_threshold),
                        []() { UpdateUringFlagsOnThread(); });
   // Register scheduler flags
   RegisterMutableFlags(
       &config_registry,
-      facade::GetFlagNames(FLAGS_scheduler_background_budget, FLAGS_scheduler_background_sleep_prob,
-                           FLAGS_scheduler_background_warrant),
+      base::GetFlagNames(FLAGS_scheduler_background_budget, FLAGS_scheduler_background_sleep_prob,
+                         FLAGS_scheduler_background_warrant),
       []() { UpdateSchedulerFlagsOnThread(); });
 
   config_registry.RegisterSetter<MemoryBytesFlag>("maxmemory", [](const MemoryBytesFlag& flag) {

src/facade/flag_utils.h

@@ -7,11 +7,11 @@
 #include <absl/container/inlined_vector.h>
 
 #include "base/cycle_clock.h"
+#include "base/flag_utils.h"
 #include "base/flags.h"
 #include "base/logging.h"
 #include "core/overloaded.h"
 #include "facade/dragonfly_connection.h"
-#include "facade/flag_utils.h"
 #include "server/command_registry.h"
 #include "server/conn_context.h"
 #include "server/engine_shard_set.h"
@@ -426,7 +426,7 @@ void MultiCommandSquasher::UpdateFromFlags() {
 }
 
 vector<string> MultiCommandSquasher::GetMutableFlagNames() {
-  return facade::GetFlagNames(FLAGS_max_busy_squash_usec, FLAGS_log_squash_info_threshold_usec);
+  return base::GetFlagNames(FLAGS_max_busy_squash_usec, FLAGS_log_squash_info_threshold_usec);
 }
 
 }  // namespace dfly

src/server/main_service.cc

@@ -12,11 +12,11 @@ extern "C" {
 #include "redis/zmalloc.h"
 }
 
+#include "base/flag_utils.h"
 #include "base/flags.h"
 #include "base/logging.h"
 #include "facade/conn_context.h"
 #include "facade/dragonfly_connection.h"
-#include "facade/flag_utils.h"
 #include "server/channel_store.h"
 #include "server/journal/journal.h"
 #include "util/listener_interface.h"
@@ -237,8 +237,8 @@ void ServerState::UpdateFromFlags() {
 }
 
 vector<string> ServerState::GetMutableFlagNames() {
-  return facade::GetFlagNames(FLAGS_rss_oom_deny_ratio, FLAGS_serialization_max_chunk_size,
-                              FLAGS_max_squashed_cmd_num);
+  return base::GetFlagNames(FLAGS_rss_oom_deny_ratio, FLAGS_serialization_max_chunk_size,
+                            FLAGS_max_squashed_cmd_num);
 }
 
 Interpreter* ServerState::BorrowInterpreter() {

src/server/multi_command_squasher.cc

@@ -14,9 +14,9 @@
 
 #include "absl/cleanup/cleanup.h"
 #include "absl/flags/internal/flag.h"
+#include "base/flag_utils.h"
 #include "base/flags.h"
 #include "base/logging.h"
-#include "facade/flag_utils.h"
 #include "server/common.h"
 #include "server/db_slice.h"
 #include "server/engine_shard_set.h"
@@ -29,6 +29,10 @@
 
 using namespace facade;
 
+using AtLeast64 = base::ConstrainedNumericFlagValue<size_t, 64>;  // ABSL_FLAG breaks with commas
+ABSL_FLAG(AtLeast64, tiered_min_value_size, 64,
+          "Minimum size of values eligible for offloading. Must be at least 64");
+
 ABSL_FLAG(bool, tiered_experimental_cooling, true,
           "If true, uses intermediate cooling layer "
           "when offloading values to storage");
@@ -467,6 +471,7 @@ float TieredStorage::WriteDepthUsage() const {
 
 void TieredStorage::UpdateFromFlags() {
   config_ = {
+      .min_value_size = absl::GetFlag(FLAGS_tiered_min_value_size),
       .experimental_cooling = absl::GetFlag(FLAGS_tiered_experimental_cooling),
       .write_depth_limit = absl::GetFlag(FLAGS_tiered_storage_write_depth),
       .offload_threshold = absl::GetFlag(FLAGS_tiered_offload_threshold),
@@ -475,8 +480,9 @@ void TieredStorage::UpdateFromFlags() {
 }
 
 std::vector<std::string> TieredStorage::GetMutableFlagNames() {
-  return facade::GetFlagNames(FLAGS_tiered_experimental_cooling, FLAGS_tiered_storage_write_depth,
-                              FLAGS_tiered_offload_threshold, FLAGS_tiered_upload_threshold);
+  return base::GetFlagNames(FLAGS_tiered_min_value_size, FLAGS_tiered_experimental_cooling,
+                            FLAGS_tiered_storage_write_depth, FLAGS_tiered_offload_threshold,
+                            FLAGS_tiered_upload_threshold);
 }
 
 bool TieredStorage::ShouldOffload() const {
@@ -564,7 +570,7 @@ size_t TieredStorage::ReclaimMemory(size_t goal) {
 bool TieredStorage::ShouldStash(const PrimeValue& pv) const {
   const auto& disk_stats = op_manager_->GetStats().disk_stats;
   return !pv.IsExternal() && !pv.HasStashPending() && pv.ObjType() == OBJ_STRING &&
-         pv.Size() >= kMinValueSize &&
+         pv.Size() >= config_.min_value_size &&
          disk_stats.allocated_bytes + tiering::kPageSize + pv.Size() < disk_stats.max_file_size;
 }
 

src/server/server_state.cc

@@ -31,8 +31,6 @@ class TieredStorage {
   class ShardOpManager;
 
  public:
-  const static size_t kMinValueSize = 64;
-
   // Min sizes of values taking up full page on their own
   const static size_t kMinOccupancySize = tiering::kPageSize / 2;
 
@@ -114,6 +112,7 @@ class TieredStorage {
   CoolQueue cool_queue_;
 
   struct {
+    size_t min_value_size;
     bool experimental_cooling;
     unsigned write_depth_limit;
     float offload_threshold;
@@ -140,8 +139,6 @@ class TieredStorage {
   class ShardOpManager;
 
  public:
-  const static size_t kMinValueSize = 64;
-
   // Min sizes of values taking up full page on their own
   const static size_t kMinOccupancySize = tiering::kPageSize / 2;