Fix dublicates messages after stopping ydbd (#26014)

Author: kruall

ydb/core/driver_lib/run/run.cpp

@@ -173,10 +173,29 @@
 
 namespace NKikimr {
 
+namespace {
+
+    void StopGRpcServers(std::weak_ptr<TGRpcServersWrapper> grpcServersWrapper) {
+        auto wrapper = grpcServersWrapper.lock();
+        if (!wrapper) {
+            return;
+        }
+        TGuard<TMutex> guard = wrapper->Guard();
+        for (auto& [name, server] : wrapper->Servers) {
+            if (!server) {
+                continue;
+            }
+            server->Stop();
+        }
+        wrapper->Servers.clear();
+    }
+
+}
+
 class TGRpcServersManager : public TActorBootstrapped<TGRpcServersManager> {
-    TGRpcServersFactory GRpcServersFactory;
-    TGRpcServers GRpcServers;
+    std::weak_ptr<TGRpcServersWrapper> GRpcServersWrapper;
     TIntrusivePtr<NMemory::IProcessMemoryInfoProvider> ProcessMemoryInfoProvider;
+    bool Started = false;
 
 public:
     enum {
@@ -192,9 +211,9 @@ class TGRpcServersManager : public TActorBootstrapped<TGRpcServersManager> {
     };
 
 public:
-    TGRpcServersManager(TGRpcServersFactory grpcServersFactory,
+    TGRpcServersManager(std::weak_ptr<TGRpcServersWrapper> grpcServersWrapper,
             TIntrusivePtr<NMemory::IProcessMemoryInfoProvider> processMemoryInfoProvider)
-        : GRpcServersFactory(std::move(grpcServersFactory))
+        : GRpcServersWrapper(std::move(grpcServersWrapper))
         , ProcessMemoryInfoProvider(std::move(processMemoryInfoProvider))
     {}
 
@@ -215,11 +234,17 @@ class TGRpcServersManager : public TActorBootstrapped<TGRpcServersManager> {
     }
 
     void Start() {
-        if (GRpcServers) {
+        if (Started) {
+            return;
+        }
+        Started = true;
+        auto wrapper = GRpcServersWrapper.lock();
+        if (!wrapper) {
             return;
         }
-        GRpcServers = GRpcServersFactory();
-        for (auto& [name, server] : GRpcServers) {
+        TGuard<TMutex> guard = wrapper->Guard();
+        wrapper->Servers = wrapper->GrpcServersFactory();
+        for (auto& [name, server] : wrapper->Servers) {
             if (!server) {
                 continue;
             }
@@ -251,12 +276,11 @@ class TGRpcServersManager : public TActorBootstrapped<TGRpcServersManager> {
     }
 
     void Stop() {
-        for (auto& [name, server] : GRpcServers) {
-            if (server) {
-                server->Stop();
-            }
+        if (!Started) {
+            return;
         }
-        GRpcServers.clear();
+        Started = false;
+        StopGRpcServers(GRpcServersWrapper);
     }
 
     void HandleStop(TEvStop::TPtr ev) {
@@ -645,7 +669,10 @@ void TKikimrRunner::InitializeKqpController(const TKikimrRunConfig& runConfig) {
 }
 
 void TKikimrRunner::InitializeGRpc(const TKikimrRunConfig& runConfig) {
-    GRpcServersFactory = [runConfig, this] { return CreateGRpcServers(runConfig); };
+    if (!GRpcServersWrapper) {
+        GRpcServersWrapper = std::make_shared<TGRpcServersWrapper>();
+    }
+    GRpcServersWrapper->GrpcServersFactory = [runConfig, this] { return CreateGRpcServers(runConfig); };
 }
 
 TGRpcServers TKikimrRunner::CreateGRpcServers(const TKikimrRunConfig& runConfig) {
@@ -1990,8 +2017,9 @@ void TKikimrRunner::KikimrStart() {
         Monitoring->Start(ActorSystem.Get());
     }
 
-    if (GRpcServersFactory) {
-        GRpcServersManager = ActorSystem->Register(new TGRpcServersManager(std::move(GRpcServersFactory), ProcessMemoryInfoProvider));
+    if (GRpcServersWrapper) {
+        GRpcServersWrapper->Servers = GRpcServersWrapper->GrpcServersFactory();
+        GRpcServersManager = ActorSystem->Register(new TGRpcServersManager(GRpcServersWrapper, ProcessMemoryInfoProvider));
     }
 
     if (SqsHttp) {
@@ -2096,23 +2124,18 @@ void TKikimrRunner::KikimrStop(bool graceful) {
         SqsHttp.Destroy();
     }
 
-    // stop processing grpc requests/response - we must stop feeding ActorSystem
-    if (GRpcServersManager) {
-        TManualEvent event;
-        ActorSystem->Send(new IEventHandle(GRpcServersManager, {}, new TGRpcServersManager::TEvStop(&event)));
-        event.WaitI();
-    }
-
     if (ActorSystem) {
         ActorSystem->Stop();
     }
 
-    if (YqSharedResources) {
-        YqSharedResources->Stop();
+    // stop processing grpc requests/response - we must stop feeding ActorSystem
+    if (GRpcServersManager) {
+        StopGRpcServers(GRpcServersWrapper);
+        GRpcServersWrapper->Servers.clear();
     }
 
-    if (ActorSystem) {
-        ActorSystem->Cleanup();
+    if (YqSharedResources) {
+        YqSharedResources->Stop();
     }
 
     if (ModuleFactories) {
@@ -2121,12 +2144,17 @@ void TKikimrRunner::KikimrStop(bool graceful) {
         }
     }
 
-    if (YdbDriver) {
-        YdbDriver->Stop(true);
-    }
     for (auto plugin: Plugins) {
         plugin->Stop();
     }
+
+    if (ActorSystem) {
+        ActorSystem->Cleanup();
+    }
+
+    if (YdbDriver) {
+        YdbDriver->Stop(true);
+    }
 }
 
 void TKikimrRunner::BusyLoop() {

ydb/core/driver_lib/run/run.h

@@ -29,6 +29,16 @@ namespace NKikimr {
 using TGRpcServers = TVector<std::pair<TString, TAutoPtr<NYdbGrpc::TGRpcServer>>>;
 using TGRpcServersFactory = std::function<TGRpcServers()>;
 
+struct TGRpcServersWrapper {
+    TGRpcServers Servers;
+    TGRpcServersFactory GrpcServersFactory;
+    TMutex Mutex;
+
+    TGuard<TMutex> Guard() {
+        return TGuard<TMutex>(Mutex);
+    }
+};
+
 class TKikimrRunner : public virtual TThrRefBase, private IGlobalObjectStorage {
 protected:
     static TProgramShouldContinue KikimrShouldContinue;
@@ -69,7 +79,7 @@ class TKikimrRunner : public virtual TThrRefBase, private IGlobalObjectStorage {
 
     TKikimrRunner(std::shared_ptr<TModuleFactories> factories = {});
 
-    TGRpcServersFactory GRpcServersFactory;
+    std::shared_ptr<TGRpcServersWrapper> GRpcServersWrapper;
     TActorId GRpcServersManager;
 
     virtual ~TKikimrRunner();