KIKIMR-22403 fix s3 client thread pool creation (#12963)

Author: zinal

ydb/core/wrappers/s3_storage_config.cpp

@@ -6,6 +6,42 @@
 #ifndef KIKIMR_DISABLE_S3_OPS
 namespace NKikimr::NWrappers::NExternalStorage {
 
+class TS3ThreadsPoolByEndpoint {
+private:
+    class TPool {
+    public:
+        std::shared_ptr<Aws::Utils::Threading::PooledThreadExecutor> Executor;
+        ui32 ThreadsCount = 0;
+
+        TPool(const std::shared_ptr<Aws::Utils::Threading::PooledThreadExecutor>& executor, const ui32 threadsCount)
+            : Executor(executor)
+            , ThreadsCount(threadsCount)
+        {
+        }
+    };
+
+    THashMap<TString, TPool> Pools;
+    TMutex Mutex;
+
+    std::shared_ptr<Aws::Utils::Threading::PooledThreadExecutor> GetPoolImpl(const TString& endpoint, const ui32 threadsCount) {
+        TGuard<TMutex> g(Mutex);
+        auto it = Pools.find(endpoint);
+        if (it == Pools.end()) {
+            TPool pool(std::make_shared<Aws::Utils::Threading::PooledThreadExecutor>(threadsCount), threadsCount);
+            it = Pools.emplace(endpoint, std::move(pool)).first;
+        } else if (it->second.ThreadsCount < threadsCount) {
+            TPool pool(std::make_shared<Aws::Utils::Threading::PooledThreadExecutor>(threadsCount), threadsCount);
+            it->second = std::move(pool);
+        }
+        return it->second.Executor;
+    }
+
+public:
+    static std::shared_ptr<Aws::Utils::Threading::PooledThreadExecutor> GetPool(const TString& endpoint, const ui32 threadsCount) {
+        return Singleton<TS3ThreadsPoolByEndpoint>()->GetPoolImpl(endpoint, threadsCount);
+    }
+};
+
 namespace {
 
 namespace NPrivate {
@@ -18,7 +54,8 @@ Aws::Client::ClientConfiguration ConfigFromSettings(const TSettings& settings) {
     auto threadsCount = NKikimrSchemeOp::TS3Settings::default_instance().GetExecutorThreadsCount();
 
     config.endpointOverride = settings.endpoint();
-    config.executor = std::make_shared<Aws::Utils::Threading::PooledThreadExecutor>(threadsCount);
+    config.executor = TS3ThreadsPoolByEndpoint::GetPool(settings.endpoint(), threadsCount);
+    config.enableTcpKeepAlive = true;
     config.verifySSL = false;
     config.connectTimeoutMs = 10000;
     config.maxConnections = threadsCount;
@@ -46,42 +83,6 @@ Aws::Auth::AWSCredentials CredentialsFromSettings(const TSettings& settings) {
 
 } // anonymous
 
-class TS3ThreadsPoolByEndpoint {
-private:
-    class TPool {
-    public:
-        std::shared_ptr<Aws::Utils::Threading::PooledThreadExecutor> Executor;
-        ui32 ThreadsCount = 0;
-
-        TPool(const std::shared_ptr<Aws::Utils::Threading::PooledThreadExecutor>& executor, const ui32 threadsCount)
-            : Executor(executor)
-            , ThreadsCount(threadsCount)
-        {
-        }
-    };
-
-    THashMap<TString, TPool> Pools;
-    TMutex Mutex;
-
-    std::shared_ptr<Aws::Utils::Threading::PooledThreadExecutor> GetPoolImpl(const TString& endpoint, const ui32 threadsCount) {
-        TGuard<TMutex> g(Mutex);
-        auto it = Pools.find(endpoint);
-        if (it == Pools.end()) {
-            TPool pool(std::make_shared<Aws::Utils::Threading::PooledThreadExecutor>(threadsCount), threadsCount);
-            it = Pools.emplace(endpoint, std::move(pool)).first;
-        } else if (it->second.ThreadsCount < threadsCount) {
-            TPool pool(std::make_shared<Aws::Utils::Threading::PooledThreadExecutor>(threadsCount), threadsCount);
-            it->second = std::move(pool);
-        }
-        return it->second.Executor;
-    }
-
-public:
-    static std::shared_ptr<Aws::Utils::Threading::PooledThreadExecutor> GetPool(const TString& endpoint, const ui32 threadsCount) {
-        return Singleton<TS3ThreadsPoolByEndpoint>()->GetPoolImpl(endpoint, threadsCount);
-    }
-};
-
 Aws::Client::ClientConfiguration TS3ExternalStorageConfig::ConfigFromSettings(const NKikimrSchemeOp::TS3Settings& settings) {
     Aws::Client::ClientConfiguration config;