[C++ SDK] Custom executor (#20752)

Author: Gazizonoki

.github/last_commit.txt

@@ -1 +1 @@
-b997686248fd02a04de35b15816f27cf320eb67e
+234f6d55e1e71a242c2767a22b990d68ff7fce72

examples/executor/main.cpp

@@ -0,0 +1,52 @@
+#include <ydb-cpp-sdk/client/driver/driver.h>
+#include <ydb-cpp-sdk/client/helpers/helpers.h>
+#include <ydb-cpp-sdk/client/query/client.h>
+
+#include <library/cpp/getopt/last_getopt.h>
+
+#include <util/thread/pool.h>
+
+#include <thread>
+
+
+void ExecutorExample(const std::string& endpoint, const std::string& database) {
+    auto driverConfig = NYdb::CreateFromEnvironment(endpoint + "/?database=" + database)
+        .SetExecutor(NYdb::CreateThreadPoolExecutorAdapter(
+            std::make_shared<TThreadPool>(TThreadPool::TParams()
+                .SetBlocking(true)
+                .SetCatching(false)
+                .SetForkAware(false)),
+            std::thread::hardware_concurrency())
+        );
+
+    NYdb::TDriver driver(driverConfig);
+    NYdb::NQuery::TQueryClient client(driver);
+
+    try {
+        auto result = client.ExecuteQuery("SELECT 1", NYdb::NQuery::TTxControl::NoTx()).GetValueSync();
+        NYdb::NStatusHelpers::ThrowOnError(result);
+        auto parser = result.GetResultSetParser(0);
+        parser.TryNextRow();
+        std::cout << "Result: " << parser.ColumnParser(0).GetInt32() << std::endl;
+    } catch (const std::exception& e) {
+        std::cerr << "Execution failed: " << e.what() << std::endl;
+    }
+
+    driver.Stop(true);
+}
+
+int main(int argc, char** argv) {
+    std::string endpoint;
+    std::string database;
+
+    NLastGetopt::TOpts opts = NLastGetopt::TOpts::Default();
+
+    opts.AddLongOption('e', "endpoint", "YDB endpoint").Required().RequiredArgument("HOST:PORT").StoreResult(&endpoint);
+    opts.AddLongOption('d', "database", "YDB database").Required().RequiredArgument("DATABASE").StoreResult(&database);
+
+    opts.SetFreeArgsMin(0);
+    NLastGetopt::TOptsParseResult result(&opts, argc, argv);
+
+    ExecutorExample(endpoint, database);
+    return 0;
+}

include/ydb-cpp-sdk/client/driver/driver.h

@@ -8,6 +8,7 @@
 #include <ydb-cpp-sdk/client/types/fatal_error_handlers/handlers.h>
 #include <ydb-cpp-sdk/client/types/request_settings.h>
 #include <ydb-cpp-sdk/client/types/status/status.h>
+#include <ydb-cpp-sdk/client/types/executor/executor.h>
 
 #include <library/cpp/logger/backend.h>
 
@@ -54,7 +55,7 @@ class TDriverConfig {
     //! Enable Ssl.
     //! caCerts  - The buffer containing the PEM encoded root certificates for SSL/TLS connections.
     //!            If this parameter is empty, the default roots will be used.
-    TDriverConfig& UseSecureConnection(const std::string& caCerts = std::string());
+    TDriverConfig& UseSecureConnection(const std::string& caCerts = "");
     TDriverConfig& SetUsePerChannelTcpConnection(bool usePerChannel);
     TDriverConfig& UseClientCertificate(const std::string& clientCert, const std::string& clientPrivateKey);
 
@@ -107,8 +108,7 @@ class TDriverConfig {
     //! Set policy for balancing
     //! Params is a optionally field to set policy settings
     //! default: EBalancingPolicy::UsePreferableLocation
-    TDriverConfig& SetBalancingPolicy(EBalancingPolicy policy, const std::string& params = std::string());
-
+    TDriverConfig& SetBalancingPolicy(EBalancingPolicy policy, const std::string& params = "");
     //! Set grpc level keep alive. If keepalive ping was delayed more than given timeout
     //! internal grpc routine fails request with TRANSIENT_FAILURE or TRANSPORT_UNAVAILABLE error
     //! Note: this timeout should not be too small to prevent fail due to
@@ -142,6 +142,11 @@ class TDriverConfig {
 
     //! Log backend.
     TDriverConfig& SetLog(std::unique_ptr<TLogBackend>&& log);
+
+    //! Set executor for async responses.
+    //! If not set, default executor will be used.
+    TDriverConfig& SetExecutor(std::shared_ptr<IExecutor> executor);
+
 private:
     class TImpl;
     std::shared_ptr<TImpl> Impl_;

include/ydb-cpp-sdk/client/federated_topic/federated_topic.h

@@ -394,7 +394,7 @@ struct TFederatedReadSessionSettings: public NTopic::TReadSessionSettings {
         //! Executor for handlers.
         //! If not set, default single threaded executor will be used.
         //! Shared between subsessions
-        FLUENT_SETTING(NTopic::IExecutor::TPtr, HandlersExecutor);
+        FLUENT_SETTING(IExecutor::TPtr, HandlersExecutor);
     };
 
     //! Federated event handlers.
@@ -492,10 +492,10 @@ struct TFederatedTopicClientSettings : public TCommonClientSettingsBase<TFederat
     using TSelf = TFederatedTopicClientSettings;
 
     //! Default executor for compression tasks.
-    FLUENT_SETTING_DEFAULT(NTopic::IExecutor::TPtr, DefaultCompressionExecutor, NTopic::CreateThreadPoolExecutor(2));
+    FLUENT_SETTING_DEFAULT(IExecutor::TPtr, DefaultCompressionExecutor, CreateThreadPoolExecutor(2));
 
     //! Default executor for callbacks.
-    FLUENT_SETTING_DEFAULT(NTopic::IExecutor::TPtr, DefaultHandlersExecutor, NTopic::CreateThreadPoolExecutor(1));
+    FLUENT_SETTING_DEFAULT(IExecutor::TPtr, DefaultHandlersExecutor, CreateThreadPoolExecutor(1));
 
     //! Connection timeoout for federation discovery.
     FLUENT_SETTING_DEFAULT(TDuration, ConnectionTimeout, TDuration::Seconds(30));

include/ydb-cpp-sdk/client/topic/executor.h

@@ -1,47 +1,9 @@
 #pragma once
 
-#include <util/generic/ptr.h>
-#include <util/system/spinlock.h>
-#include <util/thread/pool.h>
-
-#include <memory>
-#include <mutex>
+#include <ydb-cpp-sdk/client/types/executor/executor.h>
 
 namespace NYdb::inline V3::NTopic {
 
-class IExecutor: public TThrRefBase {
-public:
-    using TPtr = TIntrusivePtr<IExecutor>;
-    using TFunction = std::function<void()>;
-
-    // Is executor asynchronous.
-    virtual bool IsAsync() const = 0;
-
-    // Post function to execute.
-    virtual void Post(TFunction&& f) = 0;
-
-    // Start method.
-    // This method is idempotent.
-    // It can be called many times. Only the first one has effect.
-    void Start() {
-        std::lock_guard guard(StartLock);
-        if (!Started) {
-            DoStart();
-            Started = true;
-        }
-    }
-
-private:
-    virtual void DoStart() = 0;
-
-private:
-    bool Started = false;
-    TAdaptiveLock StartLock;
-};
-IExecutor::TPtr CreateThreadPoolExecutorAdapter(
-    std::shared_ptr<IThreadPool> threadPool); // Thread pool is expected to have been started.
-IExecutor::TPtr CreateThreadPoolExecutor(size_t threads);
-
 IExecutor::TPtr CreateSyncExecutor();
 
 } // namespace NYdb::NTopic

include/ydb-cpp-sdk/client/types/credentials/oauth2_token_exchange/jwt_token_source.h

@@ -36,7 +36,7 @@ struct TJwtTokenSourceParams {
     class ISigningAlgorithm {
     public:
         virtual ~ISigningAlgorithm() = default;
-#ifdef YDB_SDK_USE_NEW_JWT
+#ifdef YDB_SDK_OSS
         virtual std::string sign(const std::string& data, std::error_code& ec) const = 0;
 #else
         virtual std::string sign(const std::string& data) const = 0;
@@ -54,7 +54,7 @@ struct TJwtTokenSourceParams {
         {
         }
 
-#ifdef YDB_SDK_USE_NEW_JWT
+#ifdef YDB_SDK_OSS
         std::string sign(const std::string& data, std::error_code& ec) const override {
             return Alg.sign(data, ec);
         }

include/ydb-cpp-sdk/client/types/executor/executor.h

@@ -0,0 +1,57 @@
+#pragma once
+
+#ifndef YDB_SDK_OSS
+#include <util/thread/pool.h>
+#endif
+
+#include <functional>
+#include <mutex>
+
+namespace NYdb::inline V3 {
+
+class IExecutor {
+public:
+    using TPtr = std::shared_ptr<IExecutor>;
+    using TFunction = std::function<void()>;
+
+    // Start method.
+    // This method is idempotent.
+    void Start() {
+        std::lock_guard guard(StartLock);
+        if (!Started) {
+            DoStart();
+            Started = true;
+        }
+    }
+
+    virtual void Stop() = 0;
+
+    // Post function to execute.
+    virtual void Post(TFunction&& f) = 0;
+
+    // Is executor asynchronous.
+    virtual bool IsAsync() const = 0;
+
+    virtual ~IExecutor() = default;
+
+protected:
+    virtual void DoStart() = 0;
+
+    bool Started = false;
+    std::mutex StartLock;
+};
+
+// Create default executor for thread pool.
+IExecutor::TPtr CreateThreadPoolExecutor(std::size_t threadCount, std::size_t maxQueueSize = 0);
+
+#ifndef YDB_SDK_OSS
+// Create executor adapter for util thread pool.
+// Thread pool is started and stopped by SDK.
+IExecutor::TPtr CreateThreadPoolExecutorAdapter(std::shared_ptr<IThreadPool> threadPool, std::size_t threadCount, std::size_t maxQueueSize = 0);
+
+// Create executor adapter for util thread pool.
+// Thread pool is expected to have been started and stopped by user.
+IExecutor::TPtr CreateExternalThreadPoolExecutorAdapter(std::shared_ptr<IThreadPool> threadPool);
+#endif
+
+} // namespace NYdb

src/client/driver/driver.cpp

@@ -50,6 +50,7 @@ class TDriverConfig::TImpl : public IConnectionsParams {
     uint64_t GetMaxOutboundMessageSize() const override { return MaxOutboundMessageSize; }
     uint64_t GetMaxMessageSize() const override { return MaxMessageSize; }
     const TLog& GetLog() const override { return Log; }
+    std::shared_ptr<IExecutor> GetExecutor() const override { return Executor; }
 
     std::string Endpoint;
     size_t NetworkThreadsNum = 2;
@@ -78,6 +79,7 @@ class TDriverConfig::TImpl : public IConnectionsParams {
     uint64_t MaxOutboundMessageSize = 0;
     uint64_t MaxMessageSize = 0;
     TLog Log; // Null by default.
+    std::shared_ptr<IExecutor> Executor;
 };
 
 TDriverConfig::TDriverConfig(const std::string& connectionString)
@@ -214,6 +216,11 @@ TDriverConfig& TDriverConfig::SetLog(std::unique_ptr<TLogBackend>&& log) {
     return *this;
 }
 
+TDriverConfig& TDriverConfig::SetExecutor(std::shared_ptr<IExecutor> executor) {
+    Impl_->Executor = executor;
+    return *this;
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 std::shared_ptr<TGRpcConnectionsImpl> CreateInternalInterface(const TDriver connection) {

src/client/federated_topic/impl/federated_topic_impl.cpp

@@ -83,10 +83,10 @@ NThreading::TFuture<std::vector<TFederatedTopicClient::TClusterInfo>> TFederated
             });
 }
 
-auto TFederatedTopicClient::TImpl::GetSubsessionHandlersExecutor() -> NTopic::IExecutor::TPtr {
+auto TFederatedTopicClient::TImpl::GetSubsessionHandlersExecutor() -> IExecutor::TPtr {
     with_lock (Lock) {
         if (!SubsessionHandlersExecutor) {
-            SubsessionHandlersExecutor = NTopic::CreateThreadPoolExecutor(1);
+            SubsessionHandlersExecutor = CreateThreadPoolExecutor(1);
         }
         return SubsessionHandlersExecutor;
     }

src/client/federated_topic/impl/federated_topic_impl.h

@@ -76,8 +76,8 @@ class TFederatedTopicClient::TImpl {
 
 private:
 
-     // Use single-threaded executor to prevent deadlocks inside subsession event handlers.
-     NTopic::IExecutor::TPtr GetSubsessionHandlersExecutor();
+    // Use single-threaded executor to prevent deadlocks inside subsession event handlers.
+    IExecutor::TPtr GetSubsessionHandlersExecutor();
 
 private:
     std::shared_ptr<TGRpcConnectionsImpl> Connections;
@@ -86,7 +86,7 @@ class TFederatedTopicClient::TImpl {
     std::shared_ptr<std::unordered_map<NTopic::ECodec, std::unique_ptr<NTopic::ICodec>>> ProvidedCodecs =
          std::make_shared<std::unordered_map<NTopic::ECodec, std::unique_ptr<NTopic::ICodec>>>();
 
-    NTopic::IExecutor::TPtr SubsessionHandlersExecutor;
+    IExecutor::TPtr SubsessionHandlersExecutor;
 
     TAdaptiveLock Lock;
 };