[C++ SDK] Added PreferPrimaryPile balancing policy (#24283)

Author: Gazizonoki

.github/last_commit.txt

@@ -1 +1 @@
-730ae3fcb84a9a6b29a233c52ee6c519cac1e144
+110937321b45fb3be9ca39ff3bb83bcdfcf52c97

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

@@ -66,15 +66,7 @@ struct TEndpointInfo {
 };
 
 struct TPileState {
-    enum EState {
-        UNSPECIFIED = 0 /* "unspecified" */,
-        PRIMARY = 1 /* "primary" */,
-        PROMOTED = 2 /* "promoted" */,
-        SYNCHRONIZED = 3 /* "synchronized" */,
-        NOT_SYNCHRONIZED = 4 /* "not_synchronized" */,
-        SUSPENDED = 5 /* "suspended" */,
-        DISCONNECTED = 6 /* "disconnected" */
-    };
+    using EState = NYdb::EPileState;
 
     EState State;
     std::string PileName;

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

@@ -26,16 +26,20 @@ class TDriverConfig {
     //! where "<protocol>://" can be "grpc://" or "grpcs://" or be absent, "<hostname:port>" is endpoint,
     //! "/?database=<database-path>" is optional
     TDriverConfig(const std::string& connectionString = "");
+
     //! Endpoint to initiate connections with Ydb cluster,
     //! client will connect to others nodes according to client loadbalancing
     TDriverConfig& SetEndpoint(const std::string& endpoint);
+
     //! Set number of network threads, default: 2
     TDriverConfig& SetNetworkThreadsNum(size_t sz);
+
     //! Set number of client pool threads, if 0 adaptive thread pool will be used.
     //! NOTE: in case of no zero value it is possible to get deadlock if all threads
     //! of this pool is blocked somewhere in user code.
     //! default: 0
     TDriverConfig& SetClientThreadsNum(size_t sz);
+
     //! Warning: not recommended to change
     //! Set max number of queued responses. 0 - no limit
     //! There is a queue to perform async calls to user code,
@@ -46,29 +50,37 @@ class TDriverConfig {
     //! This value doesn't make sense if SetClientThreadsNum is 0
     //! default: 0
     TDriverConfig& SetMaxClientQueueSize(size_t sz);
+
     //! 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& SetUsePerChannelTcpConnection(bool usePerChannel);
     TDriverConfig& UseClientCertificate(const std::string& clientCert, const std::string& clientPrivateKey);
+
     //! Set token, this option can be overridden for client by ClientSettings
     TDriverConfig& SetAuthToken(const std::string& token);
+
     //! Set database, this option can be overridden for client by ClientSettings
     TDriverConfig& SetDatabase(const std::string& database);
+
     //! Set credentials data, this option can be overridden for client by ClientSettings
     TDriverConfig& SetCredentialsProviderFactory(std::shared_ptr<ICredentialsProviderFactory> credentialsProviderFactory);
+
     //! Set behaviour of discovery routine
     //! See EDiscoveryMode enum comments
     //! default: EDiscoveryMode::Sync
     TDriverConfig& SetDiscoveryMode(EDiscoveryMode discoveryMode);
+
     //! Max number of requests in queue waiting for discovery if "Async" mode chosen
     //! default: 100
     TDriverConfig& SetMaxQueuedRequests(size_t sz);
+
     //! Limit using of memory for grpc buffer pool. 0 means disabled.
     //! If enabled the size must be greater than size of recieved message.
     //! default: 0
     TDriverConfig& SetGrpcMemoryQuota(uint64_t bytes);
+
     //! Specify tcp keep alive settings
     //! This option allows to adjust tcp keep alive settings, useful to work
     //! with balancers or to detect unexpected connectivity problem.
@@ -83,12 +95,20 @@ class TDriverConfig {
     //! NOTE: Please read OS documentation and investigate your network topology before touching this option.
     //! default: true, 30, 5, 10 for linux, and true and OS default for others POSIX
     TDriverConfig& SetTcpKeepAliveSettings(bool enable, size_t idle, size_t count, size_t interval);
+
     //! Enable or disable drain of client logic (e.g. session pool drain) during dtor call
     TDriverConfig& SetDrainOnDtors(bool allowed);
+
+    //! Set policy for balancing
+    //! default: TBalancingPolicy::UsePreferableLocation()
+    TDriverConfig& SetBalancingPolicy(TBalancingPolicy&& policy);
+
+    //! DEPRECATED
     //! 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());
+
     //! 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
@@ -97,6 +117,7 @@ class TDriverConfig {
     //! default: enabled, 10 seconds
     TDriverConfig& SetGRpcKeepAliveTimeout(TDuration timeout);
     TDriverConfig& SetGRpcKeepAlivePermitWithoutCalls(bool permitWithoutCalls);
+
     //! Set inactive socket timeout.
     //! Used to close connections, that were inactive for given time.
     //! Closes unused connections every 1/10 of timeout, so deletion time is approximate.
@@ -107,11 +128,14 @@ class TDriverConfig {
     //! Set maximum incoming message size.
     //! Note: this option overrides MaxMessageSize for incoming messages.
     //! default: 0
+
     TDriverConfig& SetMaxInboundMessageSize(uint64_t maxInboundMessageSize);
+
     //! Set maximum outgoing message size.
     //! Note: this option overrides MaxMessageSize for outgoing messages.
     //! default: 0
     TDriverConfig& SetMaxOutboundMessageSize(uint64_t maxOutboundMessageSize);
+
     //! Note: if this option is unset, default 64_MB message size will be used.
     //! default: 0
     TDriverConfig& SetMaxMessageSize(uint64_t maxMessageSize);

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

@@ -3,6 +3,10 @@
 #include "fwd.h"
 #include "status_codes.h"
 
+#include <memory>
+#include <string>
+
+
 namespace NYdb::inline V3 {
 
 enum class EDiscoveryMode {
@@ -22,12 +26,47 @@ enum class EDiscoveryMode {
     Off
 };
 
+//! @deprecated Use TBalancingPolicy instead
 enum class EBalancingPolicy {
     //! Use all available cluster nodes regardless datacenter locality
     UseAllNodes,
     //! Use preferable location,
     //! params is a name of location (VLA, MAN), if params is empty local datacenter is used
-    UsePreferableLocation
+    UsePreferableLocation,
+};
+
+enum EPileState {
+    UNSPECIFIED = 0 /* "unspecified" */,
+    PRIMARY = 1 /* "primary" */,
+    PROMOTED = 2 /* "promoted" */,
+    SYNCHRONIZED = 3 /* "synchronized" */,
+    NOT_SYNCHRONIZED = 4 /* "not_synchronized" */,
+    SUSPENDED = 5 /* "suspended" */,
+    DISCONNECTED = 6 /* "disconnected" */
+};
+
+class TBalancingPolicy {
+    friend class TDriverConfig;
+    friend class TDriver;
+public:
+    //! Use preferable location,
+    //! location is a name of datacenter (VLA, MAN), if location is empty local datacenter is used
+    static TBalancingPolicy UsePreferableLocation(const std::string& location = {});
+
+    //! Use all available cluster nodes regardless datacenter locality
+    static TBalancingPolicy UseAllNodes();
+
+    //! EXPERIMENTAL
+    //! Use pile with preferable state
+    static TBalancingPolicy UsePreferablePileState(EPileState pileState = EPileState::PRIMARY);
+
+    class TImpl;
+private:
+    TBalancingPolicy(std::unique_ptr<TImpl>&& impl);
+
+    TBalancingPolicy(EBalancingPolicy policy, const std::string& params);
+
+    std::unique_ptr<TImpl> Impl_;
 };
 
 } // namespace NYdb

src/client/common_client/impl/client.h

@@ -3,7 +3,7 @@
 #include "iface.h"
 
 #define INCLUDE_YDB_INTERNAL_H
-#include <src/client/impl/ydb_internal/grpc_connections/grpc_connections.h>
+#include <src/client/impl/internal/grpc_connections/grpc_connections.h>
 #undef INCLUDE_YDB_INTERNAL_H
 
 #include <ydb-cpp-sdk/client/types/exceptions/exceptions.h>

src/client/common_client/settings.cpp

@@ -1,6 +1,6 @@
 #include <ydb-cpp-sdk/client/common_client/settings.h>
 
-#include <src/client/impl/ydb_internal/common/parser.h>
+#include <src/client/impl/internal/common/parser.h>
 
 namespace NYdb::inline V3 {
 

src/client/coordination/coordination.cpp

@@ -1,8 +1,8 @@
 #include <ydb-cpp-sdk/client/coordination/coordination.h>
 
 #define INCLUDE_YDB_INTERNAL_H
-#include <src/client/impl/ydb_internal/make_request/make.h>
-#include <src/client/impl/ydb_internal/scheme_helpers/helpers.h>
+#include <src/client/impl/internal/make_request/make.h>
+#include <src/client/impl/internal/scheme_helpers/helpers.h>
 #undef INCLUDE_YDB_INTERNAL_H
 
 #include <src/api/grpc/ydb_coordination_v1.grpc.pb.h>

src/client/datastreams/datastreams.cpp

@@ -1,7 +1,7 @@
 #include <ydb-cpp-sdk/client/datastreams/datastreams.h>
 
 #define INCLUDE_YDB_INTERNAL_H
-#include <src/client/impl/ydb_internal/make_request/make.h>
+#include <src/client/impl/internal/make_request/make.h>
 #undef INCLUDE_YDB_INTERNAL_H
 
 #include <ydb-cpp-sdk/library/issue/yql_issue.h>

src/client/driver/driver.cpp

@@ -1,14 +1,14 @@
 #include <ydb-cpp-sdk/client/driver/driver.h>
 
 #define INCLUDE_YDB_INTERNAL_H
-#include <src/client/impl/ydb_internal/driver/constants.h>
-#include <src/client/impl/ydb_internal/grpc_connections/grpc_connections.h>
-#include <src/client/impl/ydb_internal/logger/log.h>
+#include <src/client/impl/internal/driver/constants.h>
+#include <src/client/impl/internal/grpc_connections/grpc_connections.h>
+#include <src/client/impl/internal/logger/log.h>
 #undef INCLUDE_YDB_INTERNAL_H
 
 #include <library/cpp/logger/log.h>
-#include <src/client/impl/ydb_internal/common/parser.h>
-#include <src/client/impl/ydb_internal/common/getenv.h>
+#include <src/client/impl/internal/common/parser.h>
+#include <src/client/impl/internal/common/getenv.h>
 #include <ydb-cpp-sdk/client/common_client/ssl_credentials.h>
 #include <util/stream/file.h>
 #include <ydb-cpp-sdk/client/resources/ydb_ca.h>
@@ -41,7 +41,7 @@ class TDriverConfig::TImpl : public IConnectionsParams {
     size_t GetMaxQueuedRequests() const override { return MaxQueuedRequests; }
     TTcpKeepAliveSettings GetTcpKeepAliveSettings() const override { return TcpKeepAliveSettings; }
     bool GetDrinOnDtors() const override { return DrainOnDtors; }
-    TBalancingSettings GetBalancingSettings() const override { return BalancingSettings; }
+    TBalancingPolicy::TImpl GetBalancingSettings() const override { return BalancingSettings; }
     TDuration GetGRpcKeepAliveTimeout() const override { return GRpcKeepAliveTimeout; }
     bool GetGRpcKeepAlivePermitWithoutCalls() const override { return GRpcKeepAlivePermitWithoutCalls; }
     TDuration GetSocketIdleTimeout() const override { return SocketIdleTimeout; }
@@ -69,7 +69,7 @@ class TDriverConfig::TImpl : public IConnectionsParams {
             TCP_KEEPALIVE_INTERVAL
         };
     bool DrainOnDtors = true;
-    TBalancingSettings BalancingSettings = TBalancingSettings{EBalancingPolicy::UsePreferableLocation, std::string()};
+    TBalancingPolicy::TImpl BalancingSettings = TBalancingPolicy::TImpl("");
     TDuration GRpcKeepAliveTimeout = TDuration::Seconds(10);
     bool GRpcKeepAlivePermitWithoutCalls = true;
     TDuration SocketIdleTimeout = TDuration::Minutes(6);
@@ -170,11 +170,15 @@ TDriverConfig& TDriverConfig::SetDrainOnDtors(bool allowed) {
     return *this;
 }
 
-TDriverConfig& TDriverConfig::SetBalancingPolicy(EBalancingPolicy policy, const std::string& params) {
-    Impl_->BalancingSettings = TBalancingSettings{policy, params};
+TDriverConfig& TDriverConfig::SetBalancingPolicy(TBalancingPolicy&& policy) {
+    Impl_->BalancingSettings = std::move(*policy.Impl_);
     return *this;
 }
 
+TDriverConfig& TDriverConfig::SetBalancingPolicy(EBalancingPolicy policy, const std::string& params) {
+    return SetBalancingPolicy(TBalancingPolicy(policy, params));
+}
+
 TDriverConfig& TDriverConfig::SetGRpcKeepAliveTimeout(TDuration timeout) {
     Impl_->GRpcKeepAliveTimeout = timeout;
     return *this;
@@ -253,7 +257,7 @@ TDriverConfig TDriver::GetConfig() const {
         Impl_->TcpKeepAliveSettings_.Interval
     );
     config.SetDrainOnDtors(Impl_->DrainOnDtors_);
-    config.SetBalancingPolicy(Impl_->BalancingSettings_.Policy, Impl_->BalancingSettings_.PolicyParams);
+    config.SetBalancingPolicy(std::make_unique<TBalancingPolicy::TImpl>(Impl_->BalancingSettings_));
     config.SetGRpcKeepAliveTimeout(Impl_->GRpcKeepAliveTimeout_);
     config.SetGRpcKeepAlivePermitWithoutCalls(Impl_->GRpcKeepAlivePermitWithoutCalls_);
     config.SetSocketIdleTimeout(Impl_->SocketIdleTimeout_);

src/client/export/export.cpp

@@ -1,7 +1,7 @@
 #include <ydb-cpp-sdk/client/export/export.h>
 
 #define INCLUDE_YDB_INTERNAL_H
-#include <src/client/impl/ydb_internal/make_request/make.h>
+#include <src/client/impl/internal/make_request/make.h>
 #undef INCLUDE_YDB_INTERNAL_H
 
 #include <src/api/grpc/ydb_discovery_v1.grpc.pb.h>