Changes after PR review

Author: mkaruza

src/server/db_slice.cc

@@ -356,7 +356,7 @@ SliceEvents& SliceEvents::operator+=(const SliceEvents& o) {
 class DbSlice::PrimeBumpPolicy {
  public:
   bool CanBump(const CompactObj& obj) const {
-    return obj.IsSticky() ? false : true;
+    return !obj.IsSticky();
   }
 };
 
@@ -1691,9 +1691,11 @@ void DbSlice::PerformDeletion(Iterator del_it, DbTable* table) {
   PerformDeletionAtomic(del_it, exp_it, table);
 }
 
-void DbSlice::OnCbFinish() {
+void DbSlice::OnCbFinishBlocking() {
   if (IsCacheMode()) {
-    for (const auto& [key_hash, db_index, key] : fetched_items_) {
+    // move fetched items to local variable
+    auto moved_fetched_items_ = std::move(fetched_items_);
+    for (const auto& [key_hash, db_index, key] : moved_fetched_items_) {
       auto& db = *db_arr_[db_index];
 
       auto predicate = [&key](const PrimeKey& key_) { return key_ == key; };

src/server/db_slice.h

@@ -357,7 +357,7 @@ class DbSlice {
     return shard_id_;
   }
 
-  void OnCbFinish();
+  void OnCbFinishBlocking();
 
   bool Acquire(IntentLock::Mode m, const KeyLockArgs& lock_args);
   void Release(IntentLock::Mode m, const KeyLockArgs& lock_args);

src/server/debugcmd.cc

@@ -868,11 +868,11 @@ void DebugCmd::PopulateRangeFiber(uint64_t from, uint64_t num_of_keys,
 
   ess.AwaitRunningOnShardQueue([&](EngineShard* shard) {
     DoPopulateBatch(options, ps[shard->shard_id()]);
-    // Debug populate does not use transaction framework therefore we call OnCbFinish manually
-    // after running the callback
-    // Note that running debug populate while running flushall/db can cause dcheck fail because the
-    // finish cb is executed just when we finish populating the database.
-    cntx_->ns->GetDbSlice(shard->shard_id()).OnCbFinish();
+    // Debug populate does not use transaction framework therefore we call OnCbFinishBlocking
+    // manually after running the callback Note that running debug populate while running
+    // flushall/db can cause dcheck fail because the finish cb is executed just when we finish
+    // populating the database.
+    cntx_->ns->GetDbSlice(shard->shard_id()).OnCbFinishBlocking();
   });
 }
 

src/server/transaction.cc

@@ -695,7 +695,7 @@ void Transaction::RunCallback(EngineShard* shard) {
   }
 
   auto& db_slice = GetDbSlice(shard->shard_id());
-  db_slice.OnCbFinish();
+  db_slice.OnCbFinishBlocking();
 
   // Handle result flags to alter behaviour.
   if (result.flags & RunnableResult::AVOID_CONCLUDING) {
@@ -1364,7 +1364,7 @@ OpStatus Transaction::RunSquashedMultiCb(RunnableType cb) {
   auto& db_slice = GetDbSlice(shard->shard_id());
 
   auto result = cb(this, shard);
-  db_slice.OnCbFinish();
+  db_slice.OnCbFinishBlocking();
 
   LogAutoJournalOnShard(shard, result);
   MaybeInvokeTrackingCb();

tests/dragonfly/generic_test.py

@@ -114,19 +114,19 @@ async def block(id):
             tasks.append(block(i))
         await asyncio.gather(*tasks)
 
-
     # produce is constantly waking up consumers. It is used to trigger the
     # flow that creates wake ups on a differrent database in the
     # middle of continuation transaction.
     async def tasks_produce(num, iters):
         LPUSH_SCRIPT = """
             redis.call('LPUSH', KEYS[1], "val")
         """
+
         async def produce(id):
             c = df_server.client(db=1)  # important to be on a different db
             for i in range(iters):
                 # Must be a lua script and not multi-exec for some reason.
-                await c.eval(LPUSH_SCRIPT, 1,  f"list{{{id}}}")
+                await c.eval(LPUSH_SCRIPT, 1, f"list{{{id}}}")
 
         tasks = []
         for i in range(num):
@@ -151,7 +151,6 @@ async def drain(id, iters):
         await asyncio.gather(*tasks)
         logging.info("Finished consuming")
 
-
     num_keys = 32
     num_iters = 200
     async_task1 = asyncio.create_task(blmove_task_loose(num_keys))
@@ -288,3 +287,58 @@ async def test_rename_huge_values(df_factory, type):
     target_data = await StaticSeeder.capture(client)
 
     assert source_data == target_data
+
+
+@pytest.mark.asyncio
+async def test_key_bump_ups(df_factory):
+    master = df_factory.create(
+        proactor_threads=2,
+        cache_mode="true",
+    )
+    df_factory.start_all([master])
+    c_master = master.client()
+
+    await c_master.execute_command("DEBUG POPULATE 18000 KEY 32 RAND")
+
+    info = await c_master.info("stats")
+    assert info["bump_ups"] == 0
+
+    keys = await c_master.execute_command("SCAN 0")
+    keys = keys[1][0:10]
+
+    # Bump keys
+    for key in keys:
+        await c_master.execute_command("GET " + key)
+    info = await c_master.info("stats")
+    assert info["bump_ups"] <= 10
+
+    # Multi get bump
+    await c_master.execute_command("MGET " + " ".join(keys))
+    info = await c_master.info("stats")
+    assert info["bump_ups"] >= 10 and info["bump_ups"] <= 20
+    last_bump_ups = info["bump_ups"]
+
+    for key in keys:
+        await c_master.execute_command("DEL " + key)
+
+    # DEL should not bump up any key
+    info = await c_master.info("stats")
+    assert last_bump_ups == info["bump_ups"]
+
+    #  Find key that has slot > 0 and bump it
+    while True:
+        keys = await c_master.execute_command("SCAN 0")
+        key = keys[1][0]
+
+        debug_key_info = await c_master.execute_command("DEBUG OBJECT " + key)
+        slot_id = int(dict(map(lambda s: s.split(":"), debug_key_info.split()))["slot"])
+        if slot_id == 0:
+            # delete the key and continue
+            await c_master.execute_command("DEL " + key)
+            continue
+
+        await c_master.execute_command("GET " + key)
+        debug_key_info = await c_master.execute_command("DEBUG OBJECT " + key)
+        new_slot_id = int(dict(map(lambda s: s.split(":"), debug_key_info.split()))["slot"])
+        assert new_slot_id + 1 == slot_id
+        break