From 723b630b332753957399ef732c7bd8ce50dd5712 Mon Sep 17 00:00:00 2001 From: CritasWang Date: Fri, 31 Jul 2026 11:49:00 +0800 Subject: [PATCH 1/3] =?UTF-8?q?Fix=20SessionPool=20hanging=20indefinitely?= =?UTF-8?q?=20after=20server=20outage=20=E4=BF=AE=E5=A4=8D=E6=9C=8D?= =?UTF-8?q?=E5=8A=A1=E7=AB=AF=E6=96=AD=E5=BC=80=E5=90=8E=20SessionPool=20?= =?UTF-8?q?=E6=B0=B8=E4=B9=85=E9=98=BB=E5=A1=9E=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two defects made a pool that survived a server outage unusable, with callers blocking forever instead of receiving an error. 两个缺陷叠加导致服务端断开后连接池不可用，调用方永久阻塞且不抛异常。 - Fix the pool wait timeout being interpreted as seconds while callers assigned a millisecond value. Open() set `_clients.Timeout = _timeout * 5`, and Take() read it back through `TimeSpan.FromSeconds`, so the 500ms default connection timeout produced a ~41 minute block (13.9 hours for the legacy 10000ms constructor argument) before SessionPoolDepletedException surfaced. ConcurrentClientQueue now exposes TimeoutInMs; the seconds-based Timeout property is kept as an obsolete compatibility shim. 修复池等待超时的单位错误：Open() 以毫秒赋值，Take() 却按秒解释，导致 500ms 的 默认连接超时变成约 41 分钟的阻塞（旧构造函数传 10000ms 时为 13.9 小时）。 ConcurrentClientQueue 改用 TimeoutInMs，原秒级 Timeout 属性保留为过时兼容包装。 - Fix pool slots being lost permanently when reconnection failed. The dead client was discarded without being returned and nothing replenished the pool, so capacity shrank by one per failure until the pool was empty; from then on every caller blocked on a queue no one would ever feed, and the pool could not recover even after the server came back. Vacant slots are now tracked and rebuilt lazily on the next acquisition, keeping capacity constant and letting the pool heal itself. 修复重连失败导致的槽位永久丢失：失效连接被丢弃且不归还，池也无补充机制，每失败 一次容量就减一，耗尽后所有调用都阻塞在无人投喂的队列上，且服务恢复后无法自愈。 现在改为记录空槽并在下次获取时按需重建，池容量保持恒定并可自行恢复。 - Add SetPoolWaitTimeoutInMs to SessionPool.Builder and TableSessionPool.Builder (default 10s), independent of the socket-level SetConnectionTimeoutInMs. 为 SessionPool.Builder 和 TableSessionPool.Builder 新增 SetPoolWaitTimeoutInMs （默认 10 秒），与 socket 级的 SetConnectionTimeoutInMs 相互独立。 - Expose SessionPool.VacantSlots and document that IsOpen() is a lifecycle flag, not a connectivity probe - it stays true after the server goes down because the client keeps no heartbeat and reconnects lazily. 暴露 SessionPool.VacantSlots，并明确 IsOpen() 是生命周期标志而非连通性探针： 客户端无心跳、按需重连，服务端断开后该标志仍为 true。 - Allow Reconnect() to run without an original client so the self-heal path works in multi-node deployments. 允许 Reconnect() 在无原始连接时运行，使多节点部署下的自愈路径可用。 - Add regression tests for the timeout unit and builder wiring; update docs/SessionPool_Exception_Handling.md and docs/API.md accordingly. 补充超时单位与 Builder 接线的回归测试；同步更新相关文档。 --- docs/API.md | 2 +- docs/SessionPool_Exception_Handling.md | 117 ++++++++++++++---- src/Apache.IoTDB/ConcurrentClientQueue.cs | 26 +++- src/Apache.IoTDB/SessionPool.Builder.cs | 18 ++- src/Apache.IoTDB/SessionPool.cs | 110 +++++++++++++++- src/Apache.IoTDB/TableSessionPool.Builder.cs | 18 ++- .../ConcurrentClientQueueTests.cs | 106 ++++++++++++++++ .../SessionPoolConfigurationTests.cs | 95 ++++++++++++++ 8 files changed, 458 insertions(+), 34 deletions(-) create mode 100644 tests/Apache.IoTDB.Tests/ConcurrentClientQueueTests.cs create mode 100644 tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs diff --git a/docs/API.md b/docs/API.md index 13bcbc8..43fdd04 100644 --- a/docs/API.md +++ b/docs/API.md @@ -44,7 +44,7 @@ var tablet = | -------------- | ------------------------- | ------------------------ | ----------------------------- | | Open | bool | open session | session_pool.Open(false) | | Close | null | close session | session_pool.Close() | -| IsOpen | null | check if session is open | session_pool.IsOpen() | +| IsOpen | null | check if the pool was opened and not yet closed by the caller. It is a lifecycle flag, **not** a connectivity probe: it stays `true` after the server goes down, because the client keeps no heartbeat and reconnects on demand instead. | session_pool.IsOpen() | | OpenDebugMode | LoggingConfiguration=null | open debug mode | session_pool.OpenDebugMode() | | CloseDebugMode | null | close debug mode | session_pool.CloseDebugMode() | | SetTimeZone | string | set time zone | session_pool.GetTimeZone() | diff --git a/docs/SessionPool_Exception_Handling.md b/docs/SessionPool_Exception_Handling.md index 75fd019..d483d11 100644 --- a/docs/SessionPool_Exception_Handling.md +++ b/docs/SessionPool_Exception_Handling.md @@ -34,9 +34,9 @@ using System; try { var sessionPool = new SessionPool.Builder() - .Host("127.0.0.1") - .Port(6667) - .PoolSize(4) + .SetHost("127.0.0.1") + .SetPort(6667) + .SetPoolSize(4) .Build(); await sessionPool.Open(); @@ -54,6 +54,52 @@ catch (SessionPoolDepletedException ex) } ``` +## `IsOpen()` is a lifecycle flag, not a health check + +`SessionPool.IsOpen()` reports whether **you** have opened the pool and not yet closed it. It is not a +connectivity probe: + +- It becomes `true` after a successful `Open()` and only returns to `false` when you call `Close()`. +- The client runs no heartbeat, so a server that goes down does **not** flip it back to `false`. + Reconnection happens lazily, on the next operation. + +This means the following common guard never re-opens the pool, because the flag stays `true` forever: + +```csharp +// Anti-pattern: this short-circuits even while every connection is dead +if (_pool != null && _pool.IsOpen()) return; +``` + +To reason about actual availability, use the health metrics below, or simply let an operation throw +`SessionPoolDepletedException` and handle it. + +## Pool Wait Timeout + +Two independent timeouts govern a pool operation: + +| Setting | Unit | Default | Controls | +| ----------------------------------------- | ---- | ------- | ------------------------------------------------------------------------- | +| `SetConnectionTimeoutInMs(int)` | ms | 500 | Socket-level send/receive timeout of an individual connection | +| `SetPoolWaitTimeoutInMs(int)` | ms | 10000 | How long an operation waits for a free client before the pool gives up | + +```csharp +var sessionPool = new SessionPool.Builder() + .SetHost("127.0.0.1") + .SetPort(6667) + .SetPoolSize(8) + .SetConnectionTimeoutInMs(500) // socket timeout + .SetPoolWaitTimeoutInMs(10_000) // give up after 10s of waiting for a free client + .Build(); +``` + +When the wait budget is exhausted, the operation throws `SessionPoolDepletedException` with the reason +`Connection pool is empty and wait time out(...ms)`. Raise `SetPoolWaitTimeoutInMs` if your workload +legitimately queues behind long operations; lower it if you would rather fail fast and retry. + +> **Note:** before this setting existed, the wait budget was derived from the connection timeout and then +> misinterpreted as seconds, which turned the 500 ms default into a ~41 minute block. If you are upgrading +> from an older version and relied on that (unintended) long wait, set `SetPoolWaitTimeoutInMs` explicitly. + ## Pool Health Metrics ### Monitoring Pool Status @@ -62,9 +108,9 @@ The `SessionPool` class exposes real-time health metrics that can be used for mo ```csharp var sessionPool = new SessionPool.Builder() - .Host("127.0.0.1") - .Port(6667) - .PoolSize(8) + .SetHost("127.0.0.1") + .SetPort(6667) + .SetPoolSize(8) .Build(); await sessionPool.Open(); @@ -72,6 +118,7 @@ await sessionPool.Open(); // Check pool health Console.WriteLine($"Available Clients: {sessionPool.AvailableClients}"); Console.WriteLine($"Total Pool Size: {sessionPool.TotalPoolSize}"); +Console.WriteLine($"Vacant Slots: {sessionPool.VacantSlots}"); Console.WriteLine($"Failed Reconnections: {sessionPool.FailedReconnections}"); ``` @@ -81,8 +128,21 @@ Console.WriteLine($"Failed Reconnections: {sessionPool.FailedReconnections}"); | -------------------- | --------------------- | ------------------------------------------------ | --------------------------- | | Available Clients | `AvailableClients` | Number of idle clients ready for use | Alert if < 25% of pool size | | Total Pool Size | `TotalPoolSize` | Configured maximum pool size | N/A (constant) | +| Vacant Slots | `VacantSlots` | Slots whose connection was dropped after a failed reconnection, pending rebuild | Alert if > 0 and not returning to 0 | | Failed Reconnections | `FailedReconnections` | Cumulative count of failed reconnection attempts | Alert if > 0 and increasing | +### Self-healing behaviour + +When an operation fails and reconnection also fails, the dead connection is discarded but its **slot is +retained**. `VacantSlots` counts those empty slots, and the next acquisition that finds the pool empty +rebuilds one of them on the spot. Consequences: + +- The pool keeps its configured capacity instead of shrinking by one on every failure. +- Once the server is reachable again, the pool repopulates itself on the next operation - no `Close()` + + `Open()` cycle is required. +- A steady non-zero `VacantSlots` means the server is still unreachable; it should fall back to 0 on its + own after recovery. + ## Failure Scenarios and Recovery Strategies ### Scenario 1: Pool Exhaustion (High Load) @@ -101,9 +161,9 @@ Console.WriteLine($"Failed Reconnections: {sessionPool.FailedReconnections}"); ```csharp var sessionPool = new SessionPool.Builder() - .Host("127.0.0.1") - .Port(6667) - .PoolSize(16) // Increased from 8 + .SetHost("127.0.0.1") + .SetPort(6667) + .SetPoolSize(16) // Increased from 8 .Build(); ``` @@ -137,13 +197,25 @@ for (int i = 0; i < maxRetries; i++) **Symptoms:** - `SessionPoolDepletedException` with reason "Reconnection failed" -- `AvailableClients` decreases over time +- `AvailableClients` drops toward 0 while `VacantSlots` rises - `FailedReconnections` > 0 and increasing **Root Cause:** IoTDB server unreachable or network issues **Recovery Strategies:** +0. **Do nothing but retry.** The pool rebuilds its vacant slots on the next operation, so once the server + comes back a plain retry succeeds. Reinitialising is only needed if you want to change configuration or + drop accumulated state: + +```csharp +catch (SessionPoolDepletedException ex) +{ + // Slots are retained and rebuilt lazily - just back off and try again + await Task.Delay(2000); +} +``` + 1. **Reinitialize SessionPool:** ```csharp @@ -159,9 +231,9 @@ catch (SessionPoolDepletedException ex) when (ex.FailedReconnections > 5) // Create new pool sessionPool = new SessionPool.Builder() - .Host("127.0.0.1") - .Port(6667) - .PoolSize(8) + .SetHost("127.0.0.1") + .SetPort(6667) + .SetPoolSize(8) .Build(); await sessionPool.Open(); @@ -245,9 +317,9 @@ public async Task RateLimitedInsert(string deviceId, RowRecord record) ```csharp var sessionPool = new SessionPool.Builder() - .Host("127.0.0.1") - .Port(6667) - .Timeout(120) // Increased timeout for slow server + .SetHost("127.0.0.1") + .SetPort(6667) + .SetConnectionTimeoutInMs(5000) // Increased socket timeout for a slow server .Build(); ``` @@ -381,10 +453,10 @@ public class ProductionSessionPoolManager public async Task Initialize() { _pool = new SessionPool.Builder() - .Host("127.0.0.1") - .Port(6667) - .PoolSize(8) - .Timeout(60) + .SetHost("127.0.0.1") + .SetPort(6667) + .SetPoolSize(8) + .SetConnectionTimeoutInMs(5000) .Build(); await _pool.Open(); @@ -481,7 +553,10 @@ public class ProductionSessionPoolManager The SessionPool exception handling and health monitoring features provide comprehensive tools for building robust IoTDB applications: - Use `SessionPoolDepletedException` to understand and react to pool issues -- Monitor `AvailableClients`, `TotalPoolSize`, and `FailedReconnections` metrics +- Treat `IsOpen()` as a lifecycle flag, never as a connectivity check +- Tune `SetPoolWaitTimeoutInMs` separately from `SetConnectionTimeoutInMs` +- Monitor `AvailableClients`, `TotalPoolSize`, `VacantSlots`, and `FailedReconnections` metrics +- Rely on lazy slot rebuilding for recovery; reinitialise only when you need to change configuration - Implement appropriate recovery strategies based on failure scenarios - Set up proactive monitoring and alerting to prevent issues - Follow best practices for pool sizing and resource management diff --git a/src/Apache.IoTDB/ConcurrentClientQueue.cs b/src/Apache.IoTDB/ConcurrentClientQueue.cs index 6090c19..596e086 100644 --- a/src/Apache.IoTDB/ConcurrentClientQueue.cs +++ b/src/Apache.IoTDB/ConcurrentClientQueue.cs @@ -58,7 +58,27 @@ public void Return(Client client) public void AddRef() => Interlocked.Increment(ref _ref); public int GetRef() => Volatile.Read(ref _ref); public void RemoveRef() => Interlocked.Decrement(ref _ref); - public int Timeout { get; set; } = 10; + + /// + /// The maximum time, in milliseconds, that waits for a client to be + /// returned to the pool before throwing. Defaults to 10000 (10 seconds). + /// + public int TimeoutInMs { get; set; } = DefaultTimeoutInMs; + + internal const int DefaultTimeoutInMs = 10_000; + + /// + /// The wait timeout expressed in seconds. Kept for backward compatibility only; it is a thin + /// wrapper over . Prefer , which avoids the + /// unit ambiguity that previously caused millisecond values to be interpreted as seconds. + /// + [Obsolete("Use TimeoutInMs instead. This property interprets its value as seconds.")] + public int Timeout + { + get => TimeoutInMs / 1000; + set => TimeoutInMs = value * 1000; + } + public Client Take() { Client client = null; @@ -70,7 +90,7 @@ public Client Take() bool timeout = false; if (ClientQueue.IsEmpty) { - timeout = !Monitor.Wait(ClientQueue, TimeSpan.FromSeconds(Timeout)); + timeout = !Monitor.Wait(ClientQueue, TimeSpan.FromMilliseconds(TimeoutInMs)); } ClientQueue.TryDequeue(out client); @@ -86,7 +106,7 @@ public Client Take() } if (client == null) { - var reasonPhrase = $"Connection pool is empty and wait time out({Timeout}s)"; + var reasonPhrase = $"Connection pool is empty and wait time out({TimeoutInMs}ms)"; if (DiagnosticReporter != null) { throw DiagnosticReporter.BuildDepletionException(reasonPhrase); diff --git a/src/Apache.IoTDB/SessionPool.Builder.cs b/src/Apache.IoTDB/SessionPool.Builder.cs index 9de2874..7f5dcb4 100644 --- a/src/Apache.IoTDB/SessionPool.Builder.cs +++ b/src/Apache.IoTDB/SessionPool.Builder.cs @@ -34,6 +34,7 @@ public class Builder private int _poolSize = 8; private bool _enableRpcCompression = false; private int _connectionTimeoutInMs = 500; + private int _poolWaitTimeoutInMs = DefaultPoolWaitTimeoutInMs; private bool _useSsl = false; private string _certificatePath = null; private string _sqlDialect = IoTDBConstant.TREE_SQL_DIALECT; @@ -94,6 +95,18 @@ public Builder SetConnectionTimeoutInMs(int timeout) return this; } + /// + /// Sets how long, in milliseconds, an operation waits for a client to become available in the pool + /// before a is thrown. Defaults to + /// (10 seconds). This is independent of + /// , which controls the socket-level timeout. + /// + public Builder SetPoolWaitTimeoutInMs(int poolWaitTimeoutInMs) + { + _poolWaitTimeoutInMs = poolWaitTimeoutInMs; + return this; + } + public Builder SetUseSsl(bool useSsl) { _useSsl = useSsl; @@ -135,6 +148,7 @@ public Builder() _poolSize = 8; _enableRpcCompression = false; _connectionTimeoutInMs = 500; + _poolWaitTimeoutInMs = DefaultPoolWaitTimeoutInMs; _useSsl = false; _certificatePath = null; _sqlDialect = IoTDBConstant.TREE_SQL_DIALECT; @@ -146,9 +160,9 @@ public SessionPool Build() // if nodeUrls is not empty, use nodeUrls to create session pool if (_nodeUrls.Count > 0) { - return new SessionPool(_nodeUrls, _username, _password, _fetchSize, _zoneId, _poolSize, _enableRpcCompression, _connectionTimeoutInMs, _useSsl, _certificatePath, _sqlDialect, _database); + return new SessionPool(_nodeUrls, _username, _password, _fetchSize, _zoneId, _poolSize, _enableRpcCompression, _connectionTimeoutInMs, _useSsl, _certificatePath, _sqlDialect, _database, _poolWaitTimeoutInMs); } - return new SessionPool(_host, _port, _username, _password, _fetchSize, _zoneId, _poolSize, _enableRpcCompression, _connectionTimeoutInMs, _useSsl, _certificatePath, _sqlDialect, _database); + return new SessionPool(_host, _port, _username, _password, _fetchSize, _zoneId, _poolSize, _enableRpcCompression, _connectionTimeoutInMs, _useSsl, _certificatePath, _sqlDialect, _database, _poolWaitTimeoutInMs); } } } diff --git a/src/Apache.IoTDB/SessionPool.cs b/src/Apache.IoTDB/SessionPool.cs index 1c12d85..0207692 100644 --- a/src/Apache.IoTDB/SessionPool.cs +++ b/src/Apache.IoTDB/SessionPool.cs @@ -40,6 +40,12 @@ public partial class SessionPool : IDisposable, IPoolDiagnosticReporter private static readonly TSProtocolVersion ProtocolVersion = TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3; private const string DepletionReasonReconnectFailed = "Reconnection failed"; + /// + /// Default time, in milliseconds, that an operation waits for a client to become available + /// in the pool before a is raised. + /// + public const int DefaultPoolWaitTimeoutInMs = 10_000; + private readonly string _username; private readonly string _password; private bool _enableRpcCompression; @@ -55,6 +61,11 @@ public partial class SessionPool : IDisposable, IPoolDiagnosticReporter /// _timeout is the amount of time a Session will wait for a send operation to complete successfully. /// private readonly int _timeout; + /// + /// _poolWaitTimeoutInMs is the amount of time, in milliseconds, an operation waits for a client + /// to become available in the pool. It is unrelated to the socket-level _timeout. + /// + private readonly int _poolWaitTimeoutInMs = DefaultPoolWaitTimeoutInMs; private readonly int _poolSize = 4; private readonly string _sqlDialect = IoTDBConstant.TREE_SQL_DIALECT; private string _database; @@ -65,6 +76,12 @@ public partial class SessionPool : IDisposable, IPoolDiagnosticReporter private ConcurrentClientQueue _clients; private ILogger _logger; private PoolHealthMetrics _healthMetrics; + /// + /// Number of pool slots whose connection was discarded because reconnection failed. The slots are + /// re-materialized on demand by so that the pool keeps its + /// configured capacity and recovers by itself once the server comes back. + /// + private int _vacantSlots; public delegate Task AsyncOperation(Client client); @@ -83,6 +100,13 @@ public partial class SessionPool : IDisposable, IPoolDiagnosticReporter /// public int FailedReconnections => _healthMetrics?.GetReconnectionFailureTally() ?? 0; + /// + /// Number of pool slots that currently hold no connection because a reconnection attempt failed. + /// These slots are rebuilt lazily on the next acquisition, so a non-zero value means the pool is + /// degraded but will heal itself once the server is reachable again. + /// + public int VacantSlots => Volatile.Read(ref _vacantSlots); + [Obsolete("This method is deprecated, please use new SessionPool.Builder().")] public SessionPool(string host, int port, int poolSize) @@ -110,6 +134,10 @@ public SessionPool(string host, int port, string username, string password, int { } protected internal SessionPool(string host, int port, string username, string password, int fetchSize, string zoneId, int poolSize, bool enableRpcCompression, int timeout, bool useSsl, string certificatePath, string sqlDialect, string database) + : this(host, port, username, password, fetchSize, zoneId, poolSize, enableRpcCompression, timeout, useSsl, certificatePath, sqlDialect, database, DefaultPoolWaitTimeoutInMs) + { + } + protected internal SessionPool(string host, int port, string username, string password, int fetchSize, string zoneId, int poolSize, bool enableRpcCompression, int timeout, bool useSsl, string certificatePath, string sqlDialect, string database, int poolWaitTimeoutInMs) { _host = host; _port = port; @@ -125,6 +153,7 @@ protected internal SessionPool(string host, int port, string username, string pa _certificatePath = certificatePath; _sqlDialect = sqlDialect; _database = database; + _poolWaitTimeoutInMs = poolWaitTimeoutInMs; } /// /// Initializes a new instance of the class. @@ -153,6 +182,10 @@ public SessionPool(List nodeUrls, string username, string password, int } protected internal SessionPool(List nodeUrls, string username, string password, int fetchSize, string zoneId, int poolSize, bool enableRpcCompression, int timeout, bool useSsl, string certificatePath, string sqlDialect, string database) + : this(nodeUrls, username, password, fetchSize, zoneId, poolSize, enableRpcCompression, timeout, useSsl, certificatePath, sqlDialect, database, DefaultPoolWaitTimeoutInMs) + { + } + protected internal SessionPool(List nodeUrls, string username, string password, int fetchSize, string zoneId, int poolSize, bool enableRpcCompression, int timeout, bool useSsl, string certificatePath, string sqlDialect, string database, int poolWaitTimeoutInMs) { if (nodeUrls.Count == 0) { @@ -172,10 +205,61 @@ protected internal SessionPool(List nodeUrls, string username, string pa _certificatePath = certificatePath; _sqlDialect = sqlDialect; _database = database; + _poolWaitTimeoutInMs = poolWaitTimeoutInMs; + } + /// + /// Acquires a client from the pool. If the pool has no idle client but owns vacant slots left behind + /// by earlier failed reconnections, one of those slots is re-materialized on the spot instead of + /// blocking on a queue that nobody will ever feed. This is what lets the pool recover on its own + /// after the server has been unreachable for a while. + /// + private async Task AcquireClientAsync(CancellationToken cancellationToken = default) + { + if (_clients.ClientQueue.IsEmpty && TryReserveVacantSlot()) + { + try + { + return await Reconnect(cancellationToken: cancellationToken); + } + catch (ReconnectionFailedException reconnectEx) + { + // Still unreachable - hand the slot back so a later call can retry. + Interlocked.Increment(ref _vacantSlots); + throw new SessionPoolDepletedException(DepletionReasonReconnectFailed, AvailableClients, TotalPoolSize, FailedReconnections, reconnectEx); + } + catch + { + // Any other failure must not swallow the slot either. + Interlocked.Increment(ref _vacantSlots); + throw; + } + } + + return _clients.Take(); } + + /// + /// Atomically claims one vacant slot, returning false when none is left. + /// + private bool TryReserveVacantSlot() + { + while (true) + { + int current = Volatile.Read(ref _vacantSlots); + if (current <= 0) + { + return false; + } + if (Interlocked.CompareExchange(ref _vacantSlots, current - 1, current) == current) + { + return true; + } + } + } + public async Task ExecuteClientOperationAsync(AsyncOperation operation, string errMsg, bool retryOnFailure = true, bool putClientBack = true) { - Client client = _clients.Take(); + Client client = await AcquireClientAsync(); bool shouldReturnClient = true; bool operationSucceeded = false; try @@ -196,8 +280,11 @@ public async Task ExecuteClientOperationAsync(AsyncOperation Reconnect(Client originalClient = null, CancellationTo } else { - int startIndex = _endPoints.FindIndex(x => x.Ip == originalClient.EndPoint.Ip && x.Port == originalClient.EndPoint.Port); + int startIndex = originalClient == null + ? 0 + : _endPoints.FindIndex(x => x.Ip == originalClient.EndPoint.Ip && x.Port == originalClient.EndPoint.Port); if (startIndex == -1) { throw new ArgumentException($"The original client is not in the list of endpoints. Original client: {originalClient.EndPoint.Ip}:{originalClient.EndPoint.Port}"); @@ -372,6 +462,16 @@ public async Task Reconnect(Client originalClient = null, CancellationTo throw new ReconnectionFailedException("Error occurs when reconnecting session pool. Could not connect to any server"); } + /// + /// Indicates whether this pool has been opened and not yet closed by the caller. + /// + /// + /// This reflects the lifecycle of the pool object only - it is NOT a server-connectivity probe. + /// The client performs no heartbeat, so a server going down does not flip this back to false; + /// it stays true until is called. To reason about connectivity, use + /// , and , + /// or simply let an operation throw . + /// public bool IsOpen() => !_isClose; public async Task Close() @@ -430,7 +530,7 @@ public async Task GetTimeZone() return _zoneId; } - var client = _clients.Take(); + var client = await AcquireClientAsync(); try { diff --git a/src/Apache.IoTDB/TableSessionPool.Builder.cs b/src/Apache.IoTDB/TableSessionPool.Builder.cs index 10e24c8..192a677 100644 --- a/src/Apache.IoTDB/TableSessionPool.Builder.cs +++ b/src/Apache.IoTDB/TableSessionPool.Builder.cs @@ -37,6 +37,7 @@ public class Builder private int _poolSize = 8; private bool _enableRpcCompression = false; private int _connectionTimeoutInMs = 500; + private int _poolWaitTimeoutInMs = SessionPool.DefaultPoolWaitTimeoutInMs; private bool _useSsl = false; private string _certificatePath = null; private string _sqlDialect = IoTDBConstant.TREE_SQL_DIALECT; @@ -97,6 +98,18 @@ public Builder SetConnectionTimeoutInMs(int timeout) return this; } + /// + /// Sets how long, in milliseconds, an operation waits for a client to become available in the pool + /// before a is thrown. Defaults to + /// (10 seconds). This is independent of + /// , which controls the socket-level timeout. + /// + public Builder SetPoolWaitTimeoutInMs(int poolWaitTimeoutInMs) + { + _poolWaitTimeoutInMs = poolWaitTimeoutInMs; + return this; + } + public Builder SetUseSsl(bool useSsl) { _useSsl = useSsl; @@ -138,6 +151,7 @@ public Builder() _poolSize = 8; _enableRpcCompression = false; _connectionTimeoutInMs = 500; + _poolWaitTimeoutInMs = SessionPool.DefaultPoolWaitTimeoutInMs; _sqlDialect = IoTDBConstant.TABLE_SQL_DIALECT; _database = ""; } @@ -148,11 +162,11 @@ public TableSessionPool Build() // if nodeUrls is not empty, use nodeUrls to create session pool if (_nodeUrls.Count > 0) { - sessionPool = new SessionPool(_nodeUrls, _username, _password, _fetchSize, _zoneId, _poolSize, _enableRpcCompression, _connectionTimeoutInMs, _useSsl, _certificatePath, _sqlDialect, _database); + sessionPool = new SessionPool(_nodeUrls, _username, _password, _fetchSize, _zoneId, _poolSize, _enableRpcCompression, _connectionTimeoutInMs, _useSsl, _certificatePath, _sqlDialect, _database, _poolWaitTimeoutInMs); } else { - sessionPool = new SessionPool(_host, _port, _username, _password, _fetchSize, _zoneId, _poolSize, _enableRpcCompression, _connectionTimeoutInMs, _useSsl, _certificatePath, _sqlDialect, _database); + sessionPool = new SessionPool(_host, _port, _username, _password, _fetchSize, _zoneId, _poolSize, _enableRpcCompression, _connectionTimeoutInMs, _useSsl, _certificatePath, _sqlDialect, _database, _poolWaitTimeoutInMs); } return new TableSessionPool(sessionPool); } diff --git a/tests/Apache.IoTDB.Tests/ConcurrentClientQueueTests.cs b/tests/Apache.IoTDB.Tests/ConcurrentClientQueueTests.cs new file mode 100644 index 0000000..0bf458a --- /dev/null +++ b/tests/Apache.IoTDB.Tests/ConcurrentClientQueueTests.cs @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +using System; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; +using NUnit.Framework; + +namespace Apache.IoTDB.Tests +{ + [TestFixture] + public class ConcurrentClientQueueTests + { + private static Client NewStubClient() => new Client(null, 1L, 1L, null, null); + + [Test] + public void TimeoutInMs_DefaultsToTenSeconds() + { + var queue = new ConcurrentClientQueue(); + + Assert.That(queue.TimeoutInMs, Is.EqualTo(10_000), "Default pool wait timeout should be 10 seconds."); + } + + [Test] + public void Take_EmptyQueue_ThrowsWithinConfiguredMilliseconds() + { + // Regression guard: the wait timeout used to be interpreted as seconds while callers assigned + // a millisecond value, so a 200ms budget blocked the caller for 200 seconds instead. + var queue = new ConcurrentClientQueue { TimeoutInMs = 200 }; + var stopwatch = Stopwatch.StartNew(); + + Assert.Throws(() => queue.Take()); + + stopwatch.Stop(); + Assert.That(stopwatch.ElapsedMilliseconds, Is.LessThan(5_000), + "Take() must honour TimeoutInMs as milliseconds, not seconds."); + } + + [Test] + public void Take_EmptyQueue_ReportsTimeoutUnitInMessage() + { + var queue = new ConcurrentClientQueue { TimeoutInMs = 150 }; + + var ex = Assert.Throws(() => queue.Take()); + + Assert.That(ex.Message, Does.Contain("150ms"), "The depletion message should state the timeout in milliseconds."); + } + + [Test] + public void Take_ReturnsClientOnceOneIsHandedBack() + { + var queue = new ConcurrentClientQueue { TimeoutInMs = 5_000 }; + var expected = NewStubClient(); + + var taker = Task.Run(() => queue.Take()); + Thread.Sleep(100); // let the taker block on the empty queue + queue.Return(expected); + + Assert.That(taker.Wait(TimeSpan.FromSeconds(5)), Is.True, "Take() should be woken by Return()."); + Assert.That(taker.Result, Is.SameAs(expected)); + } + + [Test] + public void Take_DequeuesWithoutWaitingWhenClientAvailable() + { + var queue = new ConcurrentClientQueue { TimeoutInMs = 60_000 }; + var expected = NewStubClient(); + queue.Add(expected); + + var stopwatch = Stopwatch.StartNew(); + var actual = queue.Take(); + stopwatch.Stop(); + + Assert.That(actual, Is.SameAs(expected)); + Assert.That(stopwatch.ElapsedMilliseconds, Is.LessThan(1_000), "A ready client must be returned immediately."); + } + +#pragma warning disable CS0618 // exercising the obsolete compatibility shim on purpose + [Test] + public void ObsoleteTimeoutProperty_ConvertsBetweenSecondsAndMilliseconds() + { + var queue = new ConcurrentClientQueue { Timeout = 3 }; + + Assert.That(queue.TimeoutInMs, Is.EqualTo(3_000), "Setting Timeout (seconds) should scale to milliseconds."); + Assert.That(queue.Timeout, Is.EqualTo(3), "Reading Timeout should scale back to seconds."); + } +#pragma warning restore CS0618 + } +} diff --git a/tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs b/tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs new file mode 100644 index 0000000..7209b52 --- /dev/null +++ b/tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs @@ -0,0 +1,95 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +using System.Collections.Generic; +using System.Reflection; +using NUnit.Framework; + +namespace Apache.IoTDB.Tests +{ + [TestFixture] + public class SessionPoolConfigurationTests + { + private static int ReadPoolWaitTimeout(SessionPool pool) + { + var field = typeof(SessionPool).GetField("_poolWaitTimeoutInMs", BindingFlags.NonPublic | BindingFlags.Instance); + Assert.That(field, Is.Not.Null, "_poolWaitTimeoutInMs field is expected to exist on SessionPool."); + return (int)field.GetValue(pool); + } + + [Test] + public void Builder_WithoutExplicitPoolWaitTimeout_UsesDefault() + { + var pool = new SessionPool.Builder().SetHost("127.0.0.1").SetPort(6667).Build(); + + Assert.That(ReadPoolWaitTimeout(pool), Is.EqualTo(SessionPool.DefaultPoolWaitTimeoutInMs)); + } + + [Test] + public void Builder_PoolWaitTimeoutIsIndependentOfConnectionTimeout() + { + // The pool wait timeout used to be derived from the connection timeout (_timeout * 5) and then + // read back as seconds, which turned a 500ms connection timeout into a ~41 minute pool wait. + var pool = new SessionPool.Builder() + .SetHost("127.0.0.1") + .SetPort(6667) + .SetConnectionTimeoutInMs(500) + .SetPoolWaitTimeoutInMs(3_000) + .Build(); + + Assert.That(ReadPoolWaitTimeout(pool), Is.EqualTo(3_000)); + } + + [Test] + public void Builder_WithNodeUrls_PropagatesPoolWaitTimeout() + { + var pool = new SessionPool.Builder() + .SetNodeUrl(new List { "127.0.0.1:6667", "127.0.0.1:6668" }) + .SetPoolWaitTimeoutInMs(7_500) + .Build(); + + Assert.That(ReadPoolWaitTimeout(pool), Is.EqualTo(7_500)); + } + + [Test] + public void TableSessionPoolBuilder_PropagatesPoolWaitTimeout() + { + var tablePool = new TableSessionPool.Builder() + .SetHost("127.0.0.1") + .SetPort(6667) + .SetPoolWaitTimeoutInMs(4_200) + .Build(); + + var innerField = typeof(TableSessionPool).GetField("sessionPool", BindingFlags.NonPublic | BindingFlags.Instance); + Assert.That(innerField, Is.Not.Null, "TableSessionPool is expected to wrap a SessionPool."); + + var inner = (SessionPool)innerField.GetValue(tablePool); + Assert.That(ReadPoolWaitTimeout(inner), Is.EqualTo(4_200)); + } + + [Test] + public void NewPool_StartsWithNoVacantSlotsAndIsNotOpen() + { + var pool = new SessionPool.Builder().SetHost("127.0.0.1").SetPort(6667).Build(); + + Assert.That(pool.VacantSlots, Is.Zero); + Assert.That(pool.IsOpen(), Is.False); + } + } +} From f855e96554807dcb9670a576035409db61b2e70c Mon Sep 17 00:00:00 2001 From: CritasWang Date: Fri, 31 Jul 2026 12:32:36 +0800 Subject: [PATCH 2/3] =?UTF-8?q?Address=20review:=20overall=20deadline,=20C?= =?UTF-8?q?lose()=20lifecycle,=20demand-driven=20capacity=20=E5=A4=84?= =?UTF-8?q?=E7=90=86=E8=AF=84=E5=AE=A1=E6=84=8F=E8=A7=81=EF=BC=9A=E6=95=B4?= =?UTF-8?q?=E4=BD=93=E8=B6=85=E6=97=B6=E9=A2=84=E7=AE=97=E3=80=81Close()?= =?UTF-8?q?=20=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F=E3=80=81=E6=8C=89?= =?UTF-8?q?=E9=9C=80=E5=AE=B9=E9=87=8F=E8=AF=AD=E4=B9=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit [P1] Preserve one overall deadline across wake-ups in Take(). [P1] Take() 在多次唤醒之间保持单一的整体超时预算。 Return() uses PulseAll, so every waiter wakes while only one can dequeue. The losing waiters looped and re-armed the full TimeoutInMs, so under steady churn a waiter could exceed its configured bound indefinitely - the original "blocks forever" failure mode was still reachable under contention. The budget is now captured once and the remaining duration recomputed on each wake-up. Return() 使用 PulseAll，所有等待者都会被唤醒但只有一个能出队。落败者会重新进入循环 并再次以完整的 TimeoutInMs 等待，因此在持续周转下等待者可能无限超出配置上限—— 原本"永久阻塞"的失效模式在竞争条件下依然可达。现在预算只捕获一次，每次唤醒重新 计算剩余时长。 [P2] Transition lifecycle state independently of queued clients in Close(). [P2] Close() 的生命周期状态转换不再依赖队列中的连接。 _isClose was assigned only inside the foreach over queued clients. Once every connection had become a vacant slot the queue was empty, the loop ran zero times, and Close() returned while IsOpen() stayed true, leaving the rebuild path armed. _isClose is now set before the loop, vacant slots are cleared, and AcquireClientAsync no longer materializes capacity on a closed pool. _isClose 原先只在遍历队列连接的 foreach 内部赋值。当所有连接都变成空槽后队列为空， 循环零次执行，Close() 返回时 IsOpen() 仍为 true，重建路径依然处于激活状态。现在 _isClose 在循环前设置，空槽被清零，且 AcquireClientAsync 不再在已关闭的池上创建容量。 [P2] Define slot refill as demand-driven capacity and correct the metric docs. [P2] 将槽位补充明确定义为按需容量，并修正指标文档。 A slot is only materialized when an acquisition finds the idle queue empty, so under sequential or light load one connection serves every request and VacantSlots legitimately stays above zero after the server has recovered. The previous documentation claimed a steady non-zero value meant the server was still unreachable, which was wrong. VacantSlots is now documented as unrealized capacity, and the alerting guidance points at FailedReconnections instead. 只有当获取连接时发现空闲队列为空才会实例化槽位，因此在串行或轻负载下一个连接即可 服务所有请求，服务端恢复后 VacantSlots 仍会合理地保持非零。原文档称持续非零表示 服务端仍不可达，这是错误的。现在 VacantSlots 被定义为未实例化的容量，告警指引改为 基于 FailedReconnections。 - Add multi-waiter and repeated-wake-up regression tests for the deadline, and Close()-with-empty-queue regression tests for the lifecycle flag. 新增多等待者与重复唤醒的超时预算回归测试，以及空队列下 Close() 的生命周期回归测试。 --- docs/SessionPool_Exception_Handling.md | 39 +++++----- src/Apache.IoTDB/ConcurrentClientQueue.cs | 18 +++-- src/Apache.IoTDB/SessionPool.cs | 34 +++++---- .../ConcurrentClientQueueTests.cs | 72 +++++++++++++++++++ .../SessionPoolConfigurationTests.cs | 40 +++++++++++ 5 files changed, 168 insertions(+), 35 deletions(-) diff --git a/docs/SessionPool_Exception_Handling.md b/docs/SessionPool_Exception_Handling.md index d483d11..ac25068 100644 --- a/docs/SessionPool_Exception_Handling.md +++ b/docs/SessionPool_Exception_Handling.md @@ -128,20 +128,26 @@ Console.WriteLine($"Failed Reconnections: {sessionPool.FailedReconnections}"); | -------------------- | --------------------- | ------------------------------------------------ | --------------------------- | | Available Clients | `AvailableClients` | Number of idle clients ready for use | Alert if < 25% of pool size | | Total Pool Size | `TotalPoolSize` | Configured maximum pool size | N/A (constant) | -| Vacant Slots | `VacantSlots` | Slots whose connection was dropped after a failed reconnection, pending rebuild | Alert if > 0 and not returning to 0 | +| Vacant Slots | `VacantSlots` | Configured capacity currently holding no connection, refilled on demand | Not an alert signal on its own - see below | | Failed Reconnections | `FailedReconnections` | Cumulative count of failed reconnection attempts | Alert if > 0 and increasing | -### Self-healing behaviour +### Capacity is demand-driven -When an operation fails and reconnection also fails, the dead connection is discarded but its **slot is -retained**. `VacantSlots` counts those empty slots, and the next acquisition that finds the pool empty -rebuilds one of them on the spot. Consequences: +When an operation fails and reconnection also fails, the dead connection is discarded but its **capacity is +retained** rather than lost. `VacantSlots` counts that unrealized capacity, and an acquisition that finds no +idle client materializes one slot before falling back to waiting. Consequences: -- The pool keeps its configured capacity instead of shrinking by one on every failure. -- Once the server is reachable again, the pool repopulates itself on the next operation - no `Close()` + +- The pool no longer shrinks by one on every failure, so it cannot reach the state where every caller blocks + on a queue nobody will feed. +- Once the server is reachable again, the pool repopulates itself as load demands it - no `Close()` + `Open()` cycle is required. -- A steady non-zero `VacantSlots` means the server is still unreachable; it should fall back to 0 on its - own after recovery. +- **Capacity is refilled on demand, not eagerly.** A slot is only materialized when an acquisition finds the + idle queue empty. Under sequential or light workloads one connection is enough to serve every request, so + `VacantSlots` legitimately stays above zero long after the server has fully recovered. It measures + unrealized capacity, not server availability. +- Therefore **do not alert on `VacantSlots` alone.** Use `FailedReconnections` to reason about server + reachability: it only increases when a reconnection actually fails. `VacantSlots` is useful for + understanding how much of the configured pool is currently materialized. ## Failure Scenarios and Recovery Strategies @@ -198,20 +204,20 @@ for (int i = 0; i < maxRetries; i++) - `SessionPoolDepletedException` with reason "Reconnection failed" - `AvailableClients` drops toward 0 while `VacantSlots` rises -- `FailedReconnections` > 0 and increasing +- `FailedReconnections` > 0 and increasing (this, not `VacantSlots`, is the outage signal) **Root Cause:** IoTDB server unreachable or network issues **Recovery Strategies:** -0. **Do nothing but retry.** The pool rebuilds its vacant slots on the next operation, so once the server - comes back a plain retry succeeds. Reinitialising is only needed if you want to change configuration or - drop accumulated state: +0. **Do nothing but retry.** Capacity is retained and refilled on demand, so once the server comes back a + plain retry succeeds. Reinitialising is only needed if you want to change configuration or drop + accumulated state: ```csharp catch (SessionPoolDepletedException ex) { - // Slots are retained and rebuilt lazily - just back off and try again + // Capacity is retained and refilled on demand - just back off and try again await Task.Delay(2000); } ``` @@ -555,8 +561,9 @@ The SessionPool exception handling and health monitoring features provide compre - Use `SessionPoolDepletedException` to understand and react to pool issues - Treat `IsOpen()` as a lifecycle flag, never as a connectivity check - Tune `SetPoolWaitTimeoutInMs` separately from `SetConnectionTimeoutInMs` -- Monitor `AvailableClients`, `TotalPoolSize`, `VacantSlots`, and `FailedReconnections` metrics -- Rely on lazy slot rebuilding for recovery; reinitialise only when you need to change configuration +- Monitor `AvailableClients`, `TotalPoolSize`, and `FailedReconnections`; read `VacantSlots` as unrealized + capacity rather than as an outage signal +- Rely on demand-driven capacity refill for recovery; reinitialise only when you need to change configuration - Implement appropriate recovery strategies based on failure scenarios - Set up proactive monitoring and alerting to prevent issues - Follow best practices for pool sizing and resource management diff --git a/src/Apache.IoTDB/ConcurrentClientQueue.cs b/src/Apache.IoTDB/ConcurrentClientQueue.cs index 596e086..f59b631 100644 --- a/src/Apache.IoTDB/ConcurrentClientQueue.cs +++ b/src/Apache.IoTDB/ConcurrentClientQueue.cs @@ -82,22 +82,28 @@ public int Timeout public Client Take() { Client client = null; + // One overall deadline for the whole call. Return() uses PulseAll, so every waiter wakes up + // while only one of them can dequeue the returned client; re-arming the full timeout on each + // wake-up would let an unlucky waiter exceed the configured bound indefinitely under churn. + var budgetMs = TimeoutInMs; + var elapsed = Stopwatch.StartNew(); Monitor.Enter(ClientQueue); try { while (true) { - bool timeout = false; - if (ClientQueue.IsEmpty) + if (ClientQueue.TryDequeue(out client)) { - timeout = !Monitor.Wait(ClientQueue, TimeSpan.FromMilliseconds(TimeoutInMs)); + break; } - ClientQueue.TryDequeue(out client); - if (client != null || timeout) + var remainingMs = budgetMs - (int)elapsed.ElapsedMilliseconds; + if (remainingMs <= 0) { break; } + + Monitor.Wait(ClientQueue, TimeSpan.FromMilliseconds(remainingMs)); } } finally @@ -106,7 +112,7 @@ public Client Take() } if (client == null) { - var reasonPhrase = $"Connection pool is empty and wait time out({TimeoutInMs}ms)"; + var reasonPhrase = $"Connection pool is empty and wait time out({budgetMs}ms)"; if (DiagnosticReporter != null) { throw DiagnosticReporter.BuildDepletionException(reasonPhrase); diff --git a/src/Apache.IoTDB/SessionPool.cs b/src/Apache.IoTDB/SessionPool.cs index 0207692..d91df89 100644 --- a/src/Apache.IoTDB/SessionPool.cs +++ b/src/Apache.IoTDB/SessionPool.cs @@ -77,9 +77,11 @@ public partial class SessionPool : IDisposable, IPoolDiagnosticReporter private ILogger _logger; private PoolHealthMetrics _healthMetrics; /// - /// Number of pool slots whose connection was discarded because reconnection failed. The slots are - /// re-materialized on demand by so that the pool keeps its - /// configured capacity and recovers by itself once the server comes back. + /// Configured capacity that currently holds no connection, because a reconnection attempt failed and + /// the dead connection was discarded. The pool refills this capacity on demand: a slot is only + /// materialized when an acquisition finds no idle client, so under light or sequential load this + /// stays above zero even after the server has fully recovered. It measures unrealized capacity, + /// not server availability. /// private int _vacantSlots; @@ -101,9 +103,10 @@ public partial class SessionPool : IDisposable, IPoolDiagnosticReporter public int FailedReconnections => _healthMetrics?.GetReconnectionFailureTally() ?? 0; /// - /// Number of pool slots that currently hold no connection because a reconnection attempt failed. - /// These slots are rebuilt lazily on the next acquisition, so a non-zero value means the pool is - /// degraded but will heal itself once the server is reachable again. + /// Configured capacity that currently holds no connection. Capacity is refilled on demand - a slot is + /// materialized only when an acquisition finds no idle client - so a steady non-zero value under light + /// load is normal and does NOT mean the server is unreachable. Use + /// to reason about server availability. /// public int VacantSlots => Volatile.Read(ref _vacantSlots); @@ -208,14 +211,14 @@ protected internal SessionPool(List nodeUrls, string username, string pa _poolWaitTimeoutInMs = poolWaitTimeoutInMs; } /// - /// Acquires a client from the pool. If the pool has no idle client but owns vacant slots left behind - /// by earlier failed reconnections, one of those slots is re-materialized on the spot instead of - /// blocking on a queue that nobody will ever feed. This is what lets the pool recover on its own - /// after the server has been unreachable for a while. + /// Acquires a client from the pool. If the pool has no idle client but still owns unrealized + /// capacity left behind by earlier failed reconnections, one of those slots is materialized on the + /// spot instead of blocking on a queue that nobody will ever feed. This is what lets the pool + /// recover on its own after the server has been unreachable for a while. /// private async Task AcquireClientAsync(CancellationToken cancellationToken = default) { - if (_clients.ClientQueue.IsEmpty && TryReserveVacantSlot()) + if (!_isClose && _clients.ClientQueue.IsEmpty && TryReserveVacantSlot()) { try { @@ -481,6 +484,13 @@ public async Task Close() return; } + // Flip the lifecycle state first. It must not depend on how many clients happen to be queued: + // once every connection has become a vacant slot the queue is empty, and assigning _isClose + // inside the loop below would leave IsOpen() true forever. Clearing the vacant slots also + // disables the rebuild path in AcquireClientAsync while we are tearing down. + _isClose = true; + Volatile.Write(ref _vacantSlots, 0); + foreach (var client in _clients.ClientQueue.AsEnumerable()) { var closeSessionRequest = new TSCloseSessionReq(client.SessionId); @@ -494,8 +504,6 @@ public async Task Close() } finally { - _isClose = true; - client.Transport?.Close(); } } diff --git a/tests/Apache.IoTDB.Tests/ConcurrentClientQueueTests.cs b/tests/Apache.IoTDB.Tests/ConcurrentClientQueueTests.cs index 0bf458a..8511020 100644 --- a/tests/Apache.IoTDB.Tests/ConcurrentClientQueueTests.cs +++ b/tests/Apache.IoTDB.Tests/ConcurrentClientQueueTests.cs @@ -102,5 +102,77 @@ public void ObsoleteTimeoutProperty_ConvertsBetweenSecondsAndMilliseconds() Assert.That(queue.Timeout, Is.EqualTo(3), "Reading Timeout should scale back to seconds."); } #pragma warning restore CS0618 + + [Test] + public void Take_RepeatedWakeUps_StillHonoursTheOverallDeadline() + { + // Return() pulses ALL waiters while only one can dequeue, so a losing waiter re-enters the loop + // and waits again. If the full timeout were re-armed on each wake-up, that waiter could exceed + // its budget indefinitely under steady churn. Here the queue is deliberately never fed: the + // waiter is only ever pulsed, so it must still give up once the overall deadline passes. + var queue = new ConcurrentClientQueue { TimeoutInMs = 200 }; + var stop = new ManualResetEventSlim(false); + + var pulser = Task.Run(() => + { + while (!stop.IsSet) + { + Monitor.Enter(queue.ClientQueue); + try + { + Monitor.PulseAll(queue.ClientQueue); + } + finally + { + Monitor.Exit(queue.ClientQueue); + } + Thread.Sleep(20); + } + }); + + var stopwatch = Stopwatch.StartNew(); + Assert.Throws(() => queue.Take()); + stopwatch.Stop(); + stop.Set(); + pulser.Wait(TimeSpan.FromSeconds(5)); + + Assert.That(stopwatch.ElapsedMilliseconds, Is.LessThan(1_000), + $"Take() must not re-arm its budget on every wake-up (waited {stopwatch.ElapsedMilliseconds}ms for a 200ms budget)."); + } + + [Test] + public void Take_MultipleWaiters_EachHonoursTheOverallDeadline() + { + var queue = new ConcurrentClientQueue { TimeoutInMs = 300 }; + const int waiterCount = 4; + + var stopwatch = Stopwatch.StartNew(); + var waiters = new Task[waiterCount]; + for (int i = 0; i < waiterCount; i++) + { + waiters[i] = Task.Run(() => + { + try + { + queue.Take(); + return true; // won the single client + } + catch (TimeoutException) + { + return false; // timed out within budget + } + }); + } + + Thread.Sleep(100); + queue.Return(NewStubClient()); // wakes all waiters; only one can win + + Assert.That(Task.WaitAll(waiters, TimeSpan.FromSeconds(5)), Is.True, + "Every waiter should settle within its own deadline rather than waiting indefinitely."); + stopwatch.Stop(); + + Assert.That(stopwatch.ElapsedMilliseconds, Is.LessThan(2_000), + $"Losing waiters must not restart their budget (took {stopwatch.ElapsedMilliseconds}ms for a 300ms budget)."); + } } } diff --git a/tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs b/tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs index 7209b52..f728fd5 100644 --- a/tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs +++ b/tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs @@ -19,6 +19,7 @@ using System.Collections.Generic; using System.Reflection; +using System.Threading.Tasks; using NUnit.Framework; namespace Apache.IoTDB.Tests @@ -91,5 +92,44 @@ public void NewPool_StartsWithNoVacantSlotsAndIsNotOpen() Assert.That(pool.VacantSlots, Is.Zero); Assert.That(pool.IsOpen(), Is.False); } + + private static void SetPrivateField(object target, string name, object value) + { + var field = target.GetType().GetField(name, BindingFlags.NonPublic | BindingFlags.Instance); + Assert.That(field, Is.Not.Null, $"{name} field is expected to exist."); + field.SetValue(target, value); + } + + [Test] + public async Task Close_EmptyClientQueue_StillMarksThePoolClosed() + { + // Regression guard: _isClose used to be assigned only inside the foreach over queued clients. + // Once every connection had become a vacant slot the queue was empty, the loop ran zero times, + // and Close() returned while IsOpen() stayed true - leaving the rebuild path armed. + var pool = new SessionPool.Builder().SetHost("127.0.0.1").SetPort(6667).Build(); + SetPrivateField(pool, "_clients", new ConcurrentClientQueue()); + SetPrivateField(pool, "_isClose", false); + SetPrivateField(pool, "_vacantSlots", 8); + + Assert.That(pool.IsOpen(), Is.True, "Precondition: the pool looks open with an empty queue."); + + await pool.Close(); + + Assert.That(pool.IsOpen(), Is.False, "Close() must flip the lifecycle flag regardless of queue contents."); + Assert.That(pool.VacantSlots, Is.Zero, "Close() must disarm capacity refill."); + } + + [Test] + public async Task Close_IsIdempotentWhenQueueIsEmpty() + { + var pool = new SessionPool.Builder().SetHost("127.0.0.1").SetPort(6667).Build(); + SetPrivateField(pool, "_clients", new ConcurrentClientQueue()); + SetPrivateField(pool, "_isClose", false); + + await pool.Close(); + await pool.Close(); + + Assert.That(pool.IsOpen(), Is.False); + } } } From 3ad458fccb31b6d572da5cae80e3731dd5bb460e Mon Sep 17 00:00:00 2001 From: CritasWang Date: Fri, 31 Jul 2026 12:42:12 +0800 Subject: [PATCH 3/3] =?UTF-8?q?Rename=20VacantSlots=20to=20UnrealizedCapac?= =?UTF-8?q?ity=20=E5=B0=86=20VacantSlots=20=E9=87=8D=E5=91=BD=E5=90=8D?= =?UTF-8?q?=E4=B8=BA=20UnrealizedCapacity?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit "Vacant slots" reads as "something is broken", which is exactly the wrong intuition now that refill is documented as demand-driven: a steady non-zero value under light load is normal, not a fault. UnrealizedCapacity states what the number actually measures - configured capacity that has not been materialized yet. "Vacant slots" 读起来像"出了故障"，而在补充策略被明确定义为按需之后这恰恰是错误的 直觉：轻负载下持续非零是正常现象而非故障。UnrealizedCapacity 准确表达了这个数字实际 衡量的内容——尚未实例化的已配置容量。 No compatibility impact: the property is introduced by this PR and has never shipped, so no released API is affected. 无兼容性影响：该属性由本 PR 引入，从未发布，不影响任何已发布的 API。 - SessionPool.VacantSlots -> SessionPool.UnrealizedCapacity - _vacantSlots -> _unrealizedCapacity, TryReserveVacantSlot -> TryReserveUnrealizedCapacity - Update the metrics table, prose and tests to match. 同步更新指标表、说明文字与测试。 --- docs/SessionPool_Exception_Handling.md | 28 ++++++++--------- src/Apache.IoTDB/SessionPool.cs | 30 +++++++++---------- .../SessionPoolConfigurationTests.cs | 10 +++---- 3 files changed, 34 insertions(+), 34 deletions(-) diff --git a/docs/SessionPool_Exception_Handling.md b/docs/SessionPool_Exception_Handling.md index ac25068..99039de 100644 --- a/docs/SessionPool_Exception_Handling.md +++ b/docs/SessionPool_Exception_Handling.md @@ -118,7 +118,7 @@ await sessionPool.Open(); // Check pool health Console.WriteLine($"Available Clients: {sessionPool.AvailableClients}"); Console.WriteLine($"Total Pool Size: {sessionPool.TotalPoolSize}"); -Console.WriteLine($"Vacant Slots: {sessionPool.VacantSlots}"); +Console.WriteLine($"Unrealized Capacity: {sessionPool.UnrealizedCapacity}"); Console.WriteLine($"Failed Reconnections: {sessionPool.FailedReconnections}"); ``` @@ -128,26 +128,26 @@ Console.WriteLine($"Failed Reconnections: {sessionPool.FailedReconnections}"); | -------------------- | --------------------- | ------------------------------------------------ | --------------------------- | | Available Clients | `AvailableClients` | Number of idle clients ready for use | Alert if < 25% of pool size | | Total Pool Size | `TotalPoolSize` | Configured maximum pool size | N/A (constant) | -| Vacant Slots | `VacantSlots` | Configured capacity currently holding no connection, refilled on demand | Not an alert signal on its own - see below | +| Unrealized Capacity | `UnrealizedCapacity` | Configured capacity currently holding no connection, refilled on demand | Not an alert signal on its own - see below | | Failed Reconnections | `FailedReconnections` | Cumulative count of failed reconnection attempts | Alert if > 0 and increasing | ### Capacity is demand-driven When an operation fails and reconnection also fails, the dead connection is discarded but its **capacity is -retained** rather than lost. `VacantSlots` counts that unrealized capacity, and an acquisition that finds no -idle client materializes one slot before falling back to waiting. Consequences: +retained** rather than lost. `UnrealizedCapacity` counts the capacity left without a connection, and an +acquisition that finds no idle client materializes one connection before falling back to waiting. +Consequences: - The pool no longer shrinks by one on every failure, so it cannot reach the state where every caller blocks on a queue nobody will feed. - Once the server is reachable again, the pool repopulates itself as load demands it - no `Close()` + `Open()` cycle is required. -- **Capacity is refilled on demand, not eagerly.** A slot is only materialized when an acquisition finds the +- **Capacity is refilled on demand, not eagerly.** A connection is only created when an acquisition finds the idle queue empty. Under sequential or light workloads one connection is enough to serve every request, so - `VacantSlots` legitimately stays above zero long after the server has fully recovered. It measures - unrealized capacity, not server availability. -- Therefore **do not alert on `VacantSlots` alone.** Use `FailedReconnections` to reason about server - reachability: it only increases when a reconnection actually fails. `VacantSlots` is useful for - understanding how much of the configured pool is currently materialized. + `UnrealizedCapacity` legitimately stays above zero long after the server has fully recovered. It measures + how much of the configured pool has not been materialized, not server availability. +- Therefore **do not alert on `UnrealizedCapacity` alone.** Use `FailedReconnections` to reason about server + reachability: it only increases when a reconnection actually fails. ## Failure Scenarios and Recovery Strategies @@ -203,8 +203,8 @@ for (int i = 0; i < maxRetries; i++) **Symptoms:** - `SessionPoolDepletedException` with reason "Reconnection failed" -- `AvailableClients` drops toward 0 while `VacantSlots` rises -- `FailedReconnections` > 0 and increasing (this, not `VacantSlots`, is the outage signal) +- `AvailableClients` drops toward 0 while `UnrealizedCapacity` rises +- `FailedReconnections` > 0 and increasing (this, not `UnrealizedCapacity`, is the outage signal) **Root Cause:** IoTDB server unreachable or network issues @@ -561,8 +561,8 @@ The SessionPool exception handling and health monitoring features provide compre - Use `SessionPoolDepletedException` to understand and react to pool issues - Treat `IsOpen()` as a lifecycle flag, never as a connectivity check - Tune `SetPoolWaitTimeoutInMs` separately from `SetConnectionTimeoutInMs` -- Monitor `AvailableClients`, `TotalPoolSize`, and `FailedReconnections`; read `VacantSlots` as unrealized - capacity rather than as an outage signal +- Monitor `AvailableClients`, `TotalPoolSize`, and `FailedReconnections`; read `UnrealizedCapacity` as + capacity not yet materialized rather than as an outage signal - Rely on demand-driven capacity refill for recovery; reinitialise only when you need to change configuration - Implement appropriate recovery strategies based on failure scenarios - Set up proactive monitoring and alerting to prevent issues diff --git a/src/Apache.IoTDB/SessionPool.cs b/src/Apache.IoTDB/SessionPool.cs index d91df89..fff1ea3 100644 --- a/src/Apache.IoTDB/SessionPool.cs +++ b/src/Apache.IoTDB/SessionPool.cs @@ -83,7 +83,7 @@ public partial class SessionPool : IDisposable, IPoolDiagnosticReporter /// stays above zero even after the server has fully recovered. It measures unrealized capacity, /// not server availability. /// - private int _vacantSlots; + private int _unrealizedCapacity; public delegate Task AsyncOperation(Client client); @@ -108,7 +108,7 @@ public partial class SessionPool : IDisposable, IPoolDiagnosticReporter /// load is normal and does NOT mean the server is unreachable. Use /// to reason about server availability. /// - public int VacantSlots => Volatile.Read(ref _vacantSlots); + public int UnrealizedCapacity => Volatile.Read(ref _unrealizedCapacity); [Obsolete("This method is deprecated, please use new SessionPool.Builder().")] @@ -218,7 +218,7 @@ protected internal SessionPool(List nodeUrls, string username, string pa /// private async Task AcquireClientAsync(CancellationToken cancellationToken = default) { - if (!_isClose && _clients.ClientQueue.IsEmpty && TryReserveVacantSlot()) + if (!_isClose && _clients.ClientQueue.IsEmpty && TryReserveUnrealizedCapacity()) { try { @@ -227,13 +227,13 @@ private async Task AcquireClientAsync(CancellationToken cancellationToke catch (ReconnectionFailedException reconnectEx) { // Still unreachable - hand the slot back so a later call can retry. - Interlocked.Increment(ref _vacantSlots); + Interlocked.Increment(ref _unrealizedCapacity); throw new SessionPoolDepletedException(DepletionReasonReconnectFailed, AvailableClients, TotalPoolSize, FailedReconnections, reconnectEx); } catch { // Any other failure must not swallow the slot either. - Interlocked.Increment(ref _vacantSlots); + Interlocked.Increment(ref _unrealizedCapacity); throw; } } @@ -242,18 +242,18 @@ private async Task AcquireClientAsync(CancellationToken cancellationToke } /// - /// Atomically claims one vacant slot, returning false when none is left. + /// Atomically claims one unit of unrealized capacity, returning false when none is left. /// - private bool TryReserveVacantSlot() + private bool TryReserveUnrealizedCapacity() { while (true) { - int current = Volatile.Read(ref _vacantSlots); + int current = Volatile.Read(ref _unrealizedCapacity); if (current <= 0) { return false; } - if (Interlocked.CompareExchange(ref _vacantSlots, current - 1, current) == current) + if (Interlocked.CompareExchange(ref _unrealizedCapacity, current - 1, current) == current) { return true; } @@ -287,7 +287,7 @@ public async Task ExecuteClientOperationAsync(AsyncOperation Reconnect(Client originalClient = null, CancellationTo /// This reflects the lifecycle of the pool object only - it is NOT a server-connectivity probe. /// The client performs no heartbeat, so a server going down does not flip this back to false; /// it stays true until is called. To reason about connectivity, use - /// , and , + /// , and , /// or simply let an operation throw . /// public bool IsOpen() => !_isClose; @@ -485,11 +485,11 @@ public async Task Close() } // Flip the lifecycle state first. It must not depend on how many clients happen to be queued: - // once every connection has become a vacant slot the queue is empty, and assigning _isClose - // inside the loop below would leave IsOpen() true forever. Clearing the vacant slots also + // once every connection has been discarded the queue is empty, and assigning _isClose + // inside the loop below would leave IsOpen() true forever. Zeroing the unrealized capacity also // disables the rebuild path in AcquireClientAsync while we are tearing down. _isClose = true; - Volatile.Write(ref _vacantSlots, 0); + Volatile.Write(ref _unrealizedCapacity, 0); foreach (var client in _clients.ClientQueue.AsEnumerable()) { diff --git a/tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs b/tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs index f728fd5..e27cf96 100644 --- a/tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs +++ b/tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs @@ -85,11 +85,11 @@ public void TableSessionPoolBuilder_PropagatesPoolWaitTimeout() } [Test] - public void NewPool_StartsWithNoVacantSlotsAndIsNotOpen() + public void NewPool_StartsWithNoUnrealizedCapacityAndIsNotOpen() { var pool = new SessionPool.Builder().SetHost("127.0.0.1").SetPort(6667).Build(); - Assert.That(pool.VacantSlots, Is.Zero); + Assert.That(pool.UnrealizedCapacity, Is.Zero); Assert.That(pool.IsOpen(), Is.False); } @@ -104,19 +104,19 @@ private static void SetPrivateField(object target, string name, object value) public async Task Close_EmptyClientQueue_StillMarksThePoolClosed() { // Regression guard: _isClose used to be assigned only inside the foreach over queued clients. - // Once every connection had become a vacant slot the queue was empty, the loop ran zero times, + // Once every connection had been discarded the queue was empty, the loop ran zero times, // and Close() returned while IsOpen() stayed true - leaving the rebuild path armed. var pool = new SessionPool.Builder().SetHost("127.0.0.1").SetPort(6667).Build(); SetPrivateField(pool, "_clients", new ConcurrentClientQueue()); SetPrivateField(pool, "_isClose", false); - SetPrivateField(pool, "_vacantSlots", 8); + SetPrivateField(pool, "_unrealizedCapacity", 8); Assert.That(pool.IsOpen(), Is.True, "Precondition: the pool looks open with an empty queue."); await pool.Close(); Assert.That(pool.IsOpen(), Is.False, "Close() must flip the lifecycle flag regardless of queue contents."); - Assert.That(pool.VacantSlots, Is.Zero, "Close() must disarm capacity refill."); + Assert.That(pool.UnrealizedCapacity, Is.Zero, "Close() must disarm capacity refill."); } [Test]