.

Author: maladetska

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

@@ -105,6 +105,13 @@ class TDriverConfig {
     //! 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);
 
+    //! Set TCP_NODELAY socket option
+    //! enable - if true TCP_NODELAY is enabled (default, no Nagle algorithm, low latency, packet fragmentation)
+    //!        - if false TCP_NODELAY is disabled (Nagle algorithm enabled, reduced packet fragmentation)
+    //! NOTE: This affects network performance. Disable only if you want to reduce packet fragmentation.
+    //! default: true
+    TDriverConfig& SetTcpNoDelay(bool enable);
+
     //! Enable or disable drain of client logic (e.g. session pool drain) during dtor call
     TDriverConfig& SetDrainOnDtors(bool allowed);
 
@@ -117,6 +124,7 @@ class TDriverConfig {
     //! Params is a optionally field to set policy settings
     //! default: EBalancingPolicy::UsePreferableLocation
     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
@@ -126,6 +134,11 @@ class TDriverConfig {
     TDriverConfig& SetGRpcKeepAliveTimeout(TDuration timeout);
     TDriverConfig& SetGRpcKeepAlivePermitWithoutCalls(bool permitWithoutCalls);
 
+    //! Set grpc load balancing policy
+    //! policy - name of the load balancing policy, see grpc documentation for available policies
+    //! default: "round_robin"
+    TDriverConfig& SetGRpcLoadBalancingPolicy(const std::string& policy);
+
     //! 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.

src/client/driver/driver.cpp

@@ -40,10 +40,12 @@ class TDriverConfig::TImpl : public IConnectionsParams {
     EDiscoveryMode GetDiscoveryMode() const override { return DiscoveryMode; }
     size_t GetMaxQueuedRequests() const override { return MaxQueuedRequests; }
     TTcpKeepAliveSettings GetTcpKeepAliveSettings() const override { return TcpKeepAliveSettings; }
+    bool GetTcpNoDelay() const override { return TcpNoDelay; }
     bool GetDrinOnDtors() const override { return DrainOnDtors; }
     TBalancingPolicy::TImpl GetBalancingSettings() const override { return BalancingSettings; }
     TDuration GetGRpcKeepAliveTimeout() const override { return GRpcKeepAliveTimeout; }
     bool GetGRpcKeepAlivePermitWithoutCalls() const override { return GRpcKeepAlivePermitWithoutCalls; }
+    std::string GetGRpcLoadBalancingPolicy() const override { return GRpcLoadBalancingPolicy; }
     TDuration GetSocketIdleTimeout() const override { return SocketIdleTimeout; }
     uint64_t GetMemoryQuota() const override { return MemoryQuota; }
     uint64_t GetMaxInboundMessageSize() const override { return MaxInboundMessageSize; }
@@ -71,10 +73,12 @@ class TDriverConfig::TImpl : public IConnectionsParams {
             TCP_KEEPALIVE_COUNT,
             TCP_KEEPALIVE_INTERVAL
         };
+    bool TcpNoDelay = true;
     bool DrainOnDtors = true;
     TBalancingPolicy::TImpl BalancingSettings = TBalancingPolicy::TImpl::UsePreferableLocation(std::nullopt);
     TDuration GRpcKeepAliveTimeout = TDuration::Seconds(10);
     bool GRpcKeepAlivePermitWithoutCalls = true;
+    std::string GRpcLoadBalancingPolicy = "round_robin";
     TDuration SocketIdleTimeout = TDuration::Minutes(6);
     uint64_t MemoryQuota = 0;
     uint64_t MaxInboundMessageSize = 0;
@@ -174,6 +178,11 @@ TDriverConfig& TDriverConfig::SetTcpKeepAliveSettings(bool enable, size_t idle,
     return *this;
 }
 
+TDriverConfig& TDriverConfig::SetTcpNoDelay(bool enable) {
+    Impl_->TcpNoDelay = enable;
+    return *this;
+}
+
 TDriverConfig& TDriverConfig::SetGrpcMemoryQuota(uint64_t bytes) {
     Impl_->MemoryQuota = bytes;
     return *this;
@@ -203,6 +212,11 @@ TDriverConfig& TDriverConfig::SetGRpcKeepAlivePermitWithoutCalls(bool permitWith
     return *this;
 }
 
+TDriverConfig& TDriverConfig::SetGRpcLoadBalancingPolicy(const std::string& policy) {
+    Impl_->GRpcLoadBalancingPolicy = policy;
+    return *this;
+}
+
 TDriverConfig& TDriverConfig::SetSocketIdleTimeout(TDuration timeout) {
     Impl_->SocketIdleTimeout = timeout;
     return *this;
@@ -285,10 +299,12 @@ TDriverConfig TDriver::GetConfig() const {
         Impl_->TcpKeepAliveSettings_.Count,
         Impl_->TcpKeepAliveSettings_.Interval
     );
+    config.SetTcpNoDelay(Impl_->TcpNoDelay_);
     config.SetDrainOnDtors(Impl_->DrainOnDtors_);
     config.SetBalancingPolicy(std::make_unique<TBalancingPolicy::TImpl>(Impl_->BalancingSettings_));
     config.SetGRpcKeepAliveTimeout(std::chrono::duration_cast<std::chrono::microseconds>(Impl_->GRpcKeepAliveTimeout_));
     config.SetGRpcKeepAlivePermitWithoutCalls(Impl_->GRpcKeepAlivePermitWithoutCalls_);
+    config.SetGRpcLoadBalancingPolicy(Impl_->GRpcLoadBalancingPolicy_);
     config.SetSocketIdleTimeout(std::chrono::duration_cast<std::chrono::microseconds>(Impl_->SocketIdleTimeout_));
     config.SetMaxInboundMessageSize(Impl_->MaxInboundMessageSize_);
     config.SetMaxOutboundMessageSize(Impl_->MaxOutboundMessageSize_);

src/client/impl/internal/grpc_connections/grpc_connections.h

@@ -108,7 +108,7 @@ class TGRpcConnectionsImpl
             clientConfig.MaxOutboundMessageSize = MaxOutboundMessageSize_;
         }
 
-        clientConfig.LoadBalancingPolicy = "round_robin";
+        clientConfig.LoadBalancingPolicy = GRpcLoadBalancingPolicy;
 
         if (dbState->DiscoveryMode != EDiscoveryMode::Off) {
             if (std::is_same<TService,Ydb::Discovery::V1::DiscoveryService>()
@@ -709,13 +709,15 @@ class TGRpcConnectionsImpl
     const TBalancingPolicy::TImpl BalancingSettings_;
     const TDeadline::Duration GRpcKeepAliveTimeout_;
     const bool GRpcKeepAlivePermitWithoutCalls_;
+    const std::string GRpcLoadBalancingPolicy_;
     const std::uint64_t MemoryQuota_;
     const std::uint64_t MaxInboundMessageSize_;
     const std::uint64_t MaxOutboundMessageSize_;
     const std::uint64_t MaxMessageSize_;
 
     std::atomic_int64_t QueuedRequests_;
     const NYdbGrpc::TTcpKeepAliveSettings TcpKeepAliveSettings_;
+    const bool TcpNoDelay_;
     const TDeadline::Duration SocketIdleTimeout_;
 #ifndef YDB_GRPC_BYPASS_CHANNEL_POOL
     NYdbGrpc::TChannelPool ChannelPool_;

src/client/impl/internal/grpc_connections/params.h

@@ -33,10 +33,12 @@ class IConnectionsParams {
     virtual EDiscoveryMode GetDiscoveryMode() const = 0;
     virtual size_t GetMaxQueuedRequests() const = 0;
     virtual NYdbGrpc::TTcpKeepAliveSettings GetTcpKeepAliveSettings() const = 0;
+    virtual bool GetTcpNoDelay() const = 0;
     virtual bool GetDrinOnDtors() const = 0;
     virtual TBalancingPolicy::TImpl GetBalancingSettings() const = 0;
     virtual TDuration GetGRpcKeepAliveTimeout() const = 0;
     virtual bool GetGRpcKeepAlivePermitWithoutCalls() const = 0;
+    virtual std::string GetGRpcLoadBalancingPolicy() const = 0;
     virtual TDuration GetSocketIdleTimeout() const = 0;
     virtual const TLog& GetLog() const = 0;
     virtual uint64_t GetMemoryQuota() const = 0;

src/client/table/impl/table_client.cpp

@@ -42,7 +42,7 @@ TTableClient::TImpl::TImpl(std::shared_ptr<TGRpcConnectionsImpl>&& connections,
 
 TTableClient::TImpl::~TImpl() {
     if (Connections_->GetDrainOnDtors()) {
-        Drain().Wait();
+        Drain().Wait(DRAIN_TIMEOUT);
     }
 }
 

src/client/table/impl/table_client.h

@@ -30,6 +30,9 @@ namespace NTable {
 //How ofter run host scan to perform session balancing
 constexpr TDeadline::Duration HOSTSCAN_PERIODIC_ACTION_INTERVAL = std::chrono::seconds(2);
 constexpr TDuration KEEP_ALIVE_CLIENT_TIMEOUT = TDuration::Seconds(5);
+// Max wait time for Drain() to complete. Sessions are being closed with 2 seconds
+// timeout each (in parallel), plus up to 10 seconds for discovery if needed.
+constexpr TDuration DRAIN_TIMEOUT = TDuration::Seconds(30);
 
 TDuration GetMinTimeToTouch(const TSessionPoolSettings& settings);
 TDuration GetMaxTimeToTouch(const TSessionPoolSettings& settings);

tests/unit/client/CMakeLists.txt

@@ -115,17 +115,3 @@ add_ydb_test(NAME client-ydb_query_metrics_ut GTEST
   LABELS
     unit
 )
-
-add_ydb_test(NAME client-ydb_query_spans_ut GTEST
-  INCLUDE_DIRS
-    ${YDB_SDK_SOURCE_DIR}
-  SOURCES
-    query/query_spans_ut.cpp
-  LINK_LIBRARIES
-    yutil
-    impl-observability
-    client-types
-    client-trace
-  LABELS
-    unit
-)