feat: improve target node latency during migration (#5715)

* feat: improve target node latency during migration

Author: BorysTheDev

src/server/cluster/incoming_slot_migration.cc

@@ -19,6 +19,10 @@
 #include "util/fibers/synchronization.h"
 
 ABSL_DECLARE_FLAG(int, migration_finalization_timeout_ms);
+ABSL_FLAG(uint32_t, slot_migration_throttle_us, 0,
+          "Incoming migration throttle time in us, we throttle every 100us of migration commands "
+          "processing, 0 to disable. Recommended value is 20. Values more than 50 can "
+          "significantly reduce migration speed.");
 
 namespace dfly::cluster {
 
@@ -60,6 +64,9 @@ class ClusterShardMigration {
     });
     JournalReader reader{source, 0};
     TransactionReader tx_reader;
+    uint64_t last_sleep = fb2::ProactorBase::GetMonotonicTimeNs();
+
+    const uint64_t throttle_us = absl::GetFlag(FLAGS_slot_migration_throttle_us);
 
     while (cntx->IsRunning()) {
       if (pause_) {
@@ -112,6 +119,13 @@ class ClusterShardMigration {
           break;
         }
       }
+      if (throttle_us > 0) {
+        // every 100us we do sleep for 20us to allow other commands to be processed
+        if (uint64_t now = fb2::ProactorBase::GetMonotonicTimeNs(); now - last_sleep > 100000) {
+          ThisFiber::SleepFor(std::chrono::microseconds(throttle_us));
+          last_sleep = now;
+        }
+      }
     }
 
     VLOG(2) << "Flow " << source_shard_id_ << " canceled";

src/server/main_service.cc

@@ -886,6 +886,7 @@ void Service::Init(util::AcceptServer* acceptor, std::vector<facade::Listener*>
   config_registry.RegisterMutable("replica_partial_sync");
   config_registry.RegisterMutable("replication_timeout");
   config_registry.RegisterMutable("migration_finalization_timeout_ms");
+  config_registry.RegisterMutable("slot_migration_throttle_us");
   config_registry.RegisterMutable("table_growth_margin");
   config_registry.RegisterMutable("tcp_keepalive");
   config_registry.RegisterMutable("timeout");