Pull helio again, fix heartbeat

Author: dranikpg

helio

@@ -68,6 +68,8 @@ 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(bool, background_heartbeat, false, "Whether to run heartbeat as a background fiber");
+
 ABSL_DECLARE_FLAG(uint32_t, max_eviction_per_heartbeat);
 
 namespace dfly {
@@ -535,10 +537,12 @@ void EngineShard::StartPeriodicHeartbeatFiber(util::ProactorBase* pb) {
     return;
   }
   auto heartbeat = [this]() { Heartbeat(); };
+  auto priority = absl::GetFlag(FLAGS_background_heartbeat) ? fb2::FiberPriority::BACKGROUND
+                                                            : fb2::FiberPriority::NORMAL;
 
   std::chrono::milliseconds period_ms(*cycle_ms);
 
-  fb2::Fiber::Opts fb_opts{.name = "heatbeat"};
+  fb2::Fiber::Opts fb_opts{.priority = priority, .name = "heatbeat"};
   fiber_heartbeat_periodic_ =
       fb2::Fiber(fb_opts, [this, index = pb->GetPoolIndex(), period_ms, heartbeat]() mutable {
         ThisFiber::SetName(absl::StrCat("heartbeat_periodic", index));

src/server/engine_shard.cc

@@ -21,8 +21,7 @@
 #include "util/fibers/synchronization.h"
 
 ABSL_FLAG(bool, point_in_time_snapshot, true, "If true replication uses point in time snapshoting");
-ABSL_FLAG(bool, background_snapshotting, false,
-          "If true, snapshotting uses background priority fibers");
+ABSL_FLAG(bool, background_snapshotting, false, "Whether to run snapshot as a background fiber");
 
 namespace dfly {
 

src/server/snapshot.cc

@@ -126,6 +126,8 @@ def __init__(self, params: DflyParams, args):
                 self.args["num_shards"] = threads - 1
 
     def __del__(self):
+        if self.proc:
+            self.stop()
         assert self.proc == None
 
     def client(self, *args, **kwargs) -> RedisClient:
@@ -461,7 +463,11 @@ async def stop_all(self):
         """Stop all launched instances."""
         exceptions = []  # To collect exceptions
         for instance in self.instances:
-            await instance.close_clients()
+            try:  # ioloop might be no longer running
+                await instance.close_clients()
+            except Exception as e:
+                pass
+
             try:
                 instance.stop()
             except Exception as e:

tests/dragonfly/instance.py

@@ -71,6 +71,7 @@ async def test_replication_all(
         args["maxmemory"] = str(t_master * 256) + "mb"
 
     if background_snapshotting:
+        args["background_heatbeat"] = None
         args["background_snapshotting"] = None
 
     master = df_factory.create(