refactor: address comments

Signed-off-by: Stepan Bagritsevich <[email protected]>

Author: BagritsevichStepan

src/server/dfly_main.cc

@@ -823,12 +823,7 @@ Usage: dragonfly [FLAGS]
   // export MIMALLOC_VERBOSE=1 to see the options before the override.
   mi_option_enable(mi_option_show_errors);
   mi_option_set(mi_option_max_warnings, 0);
-
   mi_option_enable(mi_option_purge_decommits);
-  DCHECK(mi_option_get(mi_option_reset_decommits) == 1);
-
-  mi_option_set(mi_option_purge_delay, 0);
-  DCHECK(!mi_option_get(mi_option_reset_delay));
 
   fb2::SetDefaultStackResource(&fb2::std_malloc_resource, kFiberDefaultStackSize);
 

src/server/dragonfly_test.cc

@@ -26,6 +26,7 @@ ABSL_DECLARE_FLAG(float, mem_defrag_waste_threshold);
 ABSL_DECLARE_FLAG(uint32_t, mem_defrag_check_sec_interval);
 ABSL_DECLARE_FLAG(std::vector<std::string>, rename_command);
 ABSL_DECLARE_FLAG(bool, lua_resp2_legacy_float);
+ABSL_DECLARE_FLAG(bool, enable_heartbeat_eviction);
 ABSL_DECLARE_FLAG(double, eviction_memory_budget_threshold);
 ABSL_DECLARE_FLAG(std::vector<std::string>, command_alias);
 
@@ -688,6 +689,8 @@ TEST_F(DefragDflyEngineTest, TestDefragOption) {
   absl::SetFlag(&FLAGS_mem_defrag_threshold, 0.0);
   absl::SetFlag(&FLAGS_mem_defrag_check_sec_interval, 0);
   absl::SetFlag(&FLAGS_mem_defrag_waste_threshold, 0.1);
+  // We need to disable heartbeat eviction because it enfluences the defragmentation
+  absl::SetFlag(&FLAGS_enable_heartbeat_eviction, false);
 
   //  Fill data into dragonfly and then check if we have
   //  any location in memory to defrag. See issue #448 for details about this.

src/server/engine_shard.cc

@@ -67,9 +67,6 @@ ABSL_FLAG(double, eviction_memory_budget_threshold, 0.1,
           "Eviction starts when the free memory (including RSS memory) drops below "
           "eviction_memory_budget_threshold * max_memory_limit.");
 
-ABSL_FLAG(uint64_t, force_decommit_threshold, 8_MB,
-          "The threshold of memory to force decommit when memory is under pressure.");
-
 ABSL_DECLARE_FLAG(uint32_t, max_eviction_per_heartbeat);
 
 namespace dfly {
@@ -325,41 +322,55 @@ bool EngineShard::DefragTaskState::CheckRequired() {
     return false;
   }
 
-  std::optional<ShardMemUsage> shard_mem_usage;
-
-  if (GetFlag(FLAGS_enable_heartbeat_eviction)) {
-    shard_mem_usage = ReadShardMemUsage(GetFlag(FLAGS_mem_defrag_page_utilization_threshold));
-    const static double eviction_waste_threshold = 0.05;
-    if (shard_mem_usage->wasted_mem >
-        (uint64_t(shard_mem_usage->commited * eviction_waste_threshold))) {
-      VLOG(1) << "memory issue found for memory " << shard_mem_usage.value();
-      return true;
-    }
-  }
-
-  const std::size_t global_threshold = max_memory_limit * GetFlag(FLAGS_mem_defrag_threshold);
+  /*
+   If the eviction is enabled, we want to run the defrag task more frequently and more aggressively.
+   For global threshold rss we use the rss memory minus the eviction budget threshold and minus 3%
+    - For example if rss_deny_oom_ratio is 0.8 and eviction_memory_budget_threshold is 0.1,
+      we will start eviction when rss memory is above 0.8 - 0.1 = 0.7. And defragmentation
+      should still working if used rss memory is above 0.7 - 0.03 = 0.67.
+   For defrag interval we use the EvictionTaskState::kMemDefragCheckSecInterval
+   For waste threshold we use the EvictionTaskState::kEvictionWasteThreshold
+  */
+  const bool is_eviction_enabled = GetFlag(FLAGS_enable_heartbeat_eviction);
+
+  const double mem_defrag_threshold_flag = GetFlag(FLAGS_mem_defrag_threshold);
+  const double defrag_threshold =
+      !is_eviction_enabled ? mem_defrag_threshold_flag
+                           : std::min(mem_defrag_threshold_flag,
+                                      ServerState::tlocal()->rss_oom_deny_ratio -
+                                          GetFlag(FLAGS_eviction_memory_budget_threshold) -
+                                          EvictionTaskState::kDefragRssMemoryDelta);
+
+  const std::size_t global_threshold = max_memory_limit * defrag_threshold;
   if (global_threshold > rss_mem_current.load(memory_order_relaxed)) {
     return false;
   }
 
   const auto now = time(nullptr);
   const auto seconds_from_prev_check = now - last_check_time;
-  const auto mem_defrag_interval = GetFlag(FLAGS_mem_defrag_check_sec_interval);
+
+  const uint32_t check_sec_interval_flag = GetFlag(FLAGS_mem_defrag_check_sec_interval);
+  const auto mem_defrag_interval =
+      !is_eviction_enabled
+          ? check_sec_interval_flag
+          : std::min(check_sec_interval_flag, EvictionTaskState::kDefragCheckSecInterval);
 
   if (seconds_from_prev_check < mem_defrag_interval) {
     return false;
   }
-  last_check_time = now;
 
-  if (!shard_mem_usage) {
-    shard_mem_usage = ReadShardMemUsage(GetFlag(FLAGS_mem_defrag_page_utilization_threshold));
-  }
+  last_check_time = now;
 
-  DCHECK(shard_mem_usage.has_value());
+  ShardMemUsage shard_mem_usage =
+      ReadShardMemUsage(GetFlag(FLAGS_mem_defrag_page_utilization_threshold));
 
-  const double waste_threshold = GetFlag(FLAGS_mem_defrag_waste_threshold);
-  if (shard_mem_usage->wasted_mem > (uint64_t(shard_mem_usage->commited * waste_threshold))) {
-    VLOG(1) << "memory issue found for memory " << shard_mem_usage.value();
+  const float waste_threshold_flag = GetFlag(FLAGS_mem_defrag_waste_threshold);
+  const double waste_threshold =
+      !is_eviction_enabled
+          ? waste_threshold_flag
+          : std::min(waste_threshold_flag, EvictionTaskState::kDefragWasteThreshold);
+  if (shard_mem_usage.wasted_mem > (uint64_t(shard_mem_usage.commited * waste_threshold))) {
+    VLOG(1) << "memory issue found for memory " << shard_mem_usage;
     return true;
   }
 
@@ -858,7 +869,7 @@ size_t EngineShard::CalculateEvictionBytes() {
   size_t goal_bytes = CalculateHowManyBytesToEvictOnShard(max_memory_limit, global_used_memory,
                                                           shard_memory_budget_threshold);
 
-  LOG_IF_EVERY_N(INFO, goal_bytes > 0, 50)
+  VLOG_IF_EVERY_N(1, goal_bytes > 0, 50)
       << "Memory goal bytes: " << goal_bytes << ", used memory: " << global_used_memory
       << ", memory limit: " << max_memory_limit;
 
@@ -889,7 +900,7 @@ size_t EngineShard::CalculateEvictionBytes() {
         max_rss_memory, global_used_rss_memory - deleted_bytes_before_rss_update * shards_count,
         shard_rss_memory_budget_threshold);
 
-    LOG_IF_EVERY_N(INFO, rss_goal_bytes > 0, 50)
+    VLOG_IF_EVERY_N(1, rss_goal_bytes > 0, 50)
         << "Rss memory goal bytes: " << rss_goal_bytes
         << ", rss used memory: " << global_used_rss_memory
         << ", rss memory limit: " << max_rss_memory

src/server/engine_shard.h

@@ -220,6 +220,14 @@ class EngineShard {
   };
 
   struct EvictionTaskState {
+    /* This constant are used to control the defragmentation task when eviction is enabled.
+     The task is run periodically to check whether we need to do memory defragmentation.
+     When eviction is enabled, we want to make defragment task run more frequently
+     and also we want to make waste threshold lower to allow more aggressive defragmentation. */
+    static constexpr uint32_t kDefragCheckSecInterval = 2;
+    static constexpr float kDefragWasteThreshold = 0.05;
+    static constexpr double kDefragRssMemoryDelta = 0.03;
+
     size_t deleted_bytes_before_rss_update = 0;
     size_t global_rss_memory_at_prev_eviction = 0;
   };

tests/dragonfly/memory_test.py

@@ -256,7 +256,7 @@ async def test_cache_eviction_with_rss_deny_oom_simple_case(
             break
 
     # Wait for some time
-    await asyncio.sleep(2)
+    await asyncio.sleep(3)
 
     memory_arena = await async_client.execute_command("MEMORY", "ARENA")
     fragmentation_waste = extract_fragmentation_waste(memory_arena)
@@ -319,10 +319,10 @@ async def test_cache_eviction_with_rss_deny_oom_two_waves(
     rss_eviction_threshold = max_memory * (rss_oom_deny_ratio - eviction_memory_budget_threshold)
 
     # first wave fills 85% of max memory
-    # second wave fills 20% of max memory
+    # second wave fills 17% of max memory
     data_fill_size = [
         int((rss_oom_deny_ratio + 0.05) * max_memory),
-        int((1 - rss_oom_deny_ratio) * max_memory),
+        int((1 - rss_oom_deny_ratio - 0.03) * max_memory),
     ]
 
     val_size = 1024 * 5  # 5 kb
@@ -372,7 +372,7 @@ async def test_cache_eviction_with_rss_deny_oom_two_waves(
                 break
 
         # Wait for some time
-        await asyncio.sleep(2)
+        await asyncio.sleep(3)
 
         memory_arena = await async_client.execute_command("MEMORY", "ARENA")
         fragmentation_waste = extract_fragmentation_waste(memory_arena)
@@ -387,7 +387,7 @@ async def test_cache_eviction_with_rss_deny_oom_two_waves(
             rss_oom_deny_ratio - eviction_memory_budget_threshold - 0.05
         ), "We should not evict all items."
         assert memory_info["used_memory"] < max_memory * (
-            rss_oom_deny_ratio - eviction_memory_budget_threshold + 0.05
+            rss_oom_deny_ratio - eviction_memory_budget_threshold + 0.08
         ), "Used memory should be smaller than threshold."
         assert memory_info["used_memory_rss"] > max_memory * (
             rss_oom_deny_ratio - eviction_memory_budget_threshold - 0.05