fix(evicition): Don't accumulate deleted bytes if there is no RSS evi… (#5908)

fix(evicition): Don't accumulate deleted bytes if there is no RSS eviction (#5894)

We start to accumulate number of deleted bytes only if there is expected
that we evict on RSS threshold. Once we stop expecting to evict and we
have cleared number of tracked deleted bytes we stop with tracking.

Fixes #5891

Co-authored-by: mkaruza <[email protected]>

Author: BorysTheDev

src/server/engine_shard.cc

@@ -690,17 +690,19 @@ void EngineShard::RetireExpiredAndEvict() {
       auto [evicted_items, evicted_bytes] =
           db_slice.FreeMemWithEvictionStepAtomic(i, starting_segment_id, eviction_goal);
 
-      DVLOG(2) << "Heartbeat eviction: Expected to evict " << eviction_goal
-               << " bytes. Actually evicted " << evicted_items << " items, " << evicted_bytes
-               << " bytes. Max eviction per heartbeat: "
-               << GetFlag(FLAGS_max_eviction_per_heartbeat);
+      VLOG(2) << "Heartbeat eviction: Expected to evict " << eviction_goal
+              << " bytes. Actually evicted " << evicted_items << " items, " << evicted_bytes
+              << " bytes. Max eviction per heartbeat: "
+              << GetFlag(FLAGS_max_eviction_per_heartbeat);
 
       deleted_bytes += evicted_bytes;
       eviction_goal -= std::min(eviction_goal, evicted_bytes);
     }
   }
 
-  eviction_state_.deleted_bytes_before_rss_update += deleted_bytes;
+  // Track deleted bytes only if we expect to lower memory
+  if (eviction_state_.track_deleted_bytes)
+    eviction_state_.deleted_bytes_before_rss_update += deleted_bytes;
 }
 
 size_t EngineShard::CalculateEvictionBytes() {
@@ -718,9 +720,9 @@ size_t EngineShard::CalculateEvictionBytes() {
   size_t goal_bytes =
       CalculateHowManyBytesToEvictOnShard(limit, global_used_memory, shard_memory_budget_threshold);
 
-  VLOG_IF_EVERY_N(1, goal_bytes > 0, 50)
-      << "Used memory goal bytes: " << goal_bytes << ", used memory: " << global_used_memory
-      << ", memory limit: " << max_memory_limit;
+  VLOG_IF(2, goal_bytes > 0) << "Used memory goal bytes: " << goal_bytes
+                             << ", used memory: " << global_used_memory
+                             << ", memory limit: " << max_memory_limit;
 
   // Check for `enable_heartbeat_rss_eviction` flag since it dynamic. And reset
   // state if flag has changed.
@@ -739,26 +741,45 @@ size_t EngineShard::CalculateEvictionBytes() {
       auto decrease_delete_bytes_before_rss_update =
           std::min(deleted_bytes_before_rss_update,
                    (global_rss_memory_at_prev_eviction - global_used_rss_memory) / shards_count);
-      VLOG_EVERY_N(1, 50) << "deleted_bytes_before_rss_update: " << deleted_bytes_before_rss_update
-                          << " decrease_delete_bytes_before_rss_update: "
-                          << decrease_delete_bytes_before_rss_update;
+      VLOG(2) << "deleted_bytes_before_rss_update: " << deleted_bytes_before_rss_update
+              << " decrease_delete_bytes_before_rss_update: "
+              << decrease_delete_bytes_before_rss_update;
       deleted_bytes_before_rss_update -= decrease_delete_bytes_before_rss_update;
     }
 
     global_rss_memory_at_prev_eviction = global_used_rss_memory;
 
+    LOG_IF(DFATAL, global_used_rss_memory < (deleted_bytes_before_rss_update * shards_count))
+        << "RSS evicition underflow "
+        << "global_used_rss_memory: " << global_used_rss_memory
+        << " deleted_bytes_before_rss_update: " << deleted_bytes_before_rss_update;
+
+    // If we underflow use limit as used_memory
+    size_t used_rss_memory_with_deleted_bytes =
+        std::min(global_used_rss_memory - deleted_bytes_before_rss_update * shards_count, limit);
+
     // Try to evict more bytes if we are close to the rss memory limit
     const size_t rss_goal_bytes = CalculateHowManyBytesToEvictOnShard(
-        limit, global_used_rss_memory - deleted_bytes_before_rss_update * shards_count,
-        shard_memory_budget_threshold);
+        limit, used_rss_memory_with_deleted_bytes, shard_memory_budget_threshold);
+
+    // RSS evictions starts so we should start tracking deleted_bytes
+    if (rss_goal_bytes) {
+      eviction_state_.track_deleted_bytes = true;
+    } else {
+      // There is no RSS eviction goal and we have cleared tracked deleted bytes
+      if (!deleted_bytes_before_rss_update) {
+        eviction_state_.track_deleted_bytes = false;
+      }
+    }
 
-    VLOG_IF_EVERY_N(1, rss_goal_bytes > 0, 50)
+    VLOG_IF(2, rss_goal_bytes > 0)
         << "Rss memory goal bytes: " << rss_goal_bytes
         << ", rss used memory: " << global_used_rss_memory << ", rss memory limit: " << limit
         << ", deleted_bytes_before_rss_update: " << deleted_bytes_before_rss_update;
 
     goal_bytes = std::max(goal_bytes, rss_goal_bytes);
   }
+
   return goal_bytes;
 }
 

src/server/engine_shard.h

@@ -227,6 +227,7 @@ class EngineShard {
 
   struct EvictionTaskState {
     bool rss_eviction_enabled_ = true;
+    bool track_deleted_bytes = false;
     size_t deleted_bytes_before_rss_update = 0;
     size_t global_rss_memory_at_prev_eviction = 0;
   };

tests/dragonfly/memory_test.py

@@ -243,6 +243,44 @@ async def test_eviction_on_rss_treshold(df_factory: DflyInstanceFactory, heartbe
         assert stats_info_after["evicted_keys"] == 0
 
 
+# Github issue #5891
+async def test_no_rss_eviction_overflow_on_expired_keys(df_factory: DflyInstanceFactory):
+    max_memory = 256 * 1024**2  # 256MB
+    df_server = df_factory.create(
+        proactor_threads=1, cache_mode="yes", maxmemory=max_memory, vmodule="engine_shard=2"
+    )
+    df_server.start()
+    client = df_server.client()
+
+    data_fill_size = int(0.20 * max_memory)  # 20% of max_memory
+
+    val_size = 1024 * 50  # 50 kb for key
+    num_keys = data_fill_size // val_size
+
+    for i in range(0, 5):
+        pipe = client.pipeline(transaction=False)
+        step_keys = num_keys + i * 10
+        await pipe.execute_command("DEBUG", "POPULATE", step_keys, "key_1", val_size)
+        await pipe.execute_command("DEBUG", "POPULATE", step_keys + i * 10, "key_2", val_size)
+        for i in range(step_keys):
+            if i % 2 == 0:
+                await pipe.execute_command(f"EXPIRE key_1:{i} 1")
+            else:
+                await pipe.execute_command(f"EXPIRE key_2:{i} 1")
+        await pipe.execute()
+        await asyncio.sleep(2)
+
+    await client.execute_command("FLUSHALL")
+
+    # New keys should be added
+    await client.execute_command("DEBUG", "POPULATE", num_keys, "key", val_size)
+    # Wait so heartbeat eviction
+    await asyncio.sleep(5)
+
+    keyspace_info = await client.info("keyspace")
+    assert keyspace_info["db0"]["keys"] == num_keys
+
+
 @pytest.mark.asyncio
 async def test_throttle_on_commands_squashing_replies_bytes(df_factory: DflyInstanceFactory):
     df = df_factory.create(