Fix SessionPool blocking indefinitely after a server outage(#62)
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 @@
| -------------- | ------------------------- | ------------------------ | ----------------------------- |
| 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..99039de 100644
--- a/docs/SessionPool_Exception_Handling.md
+++ b/docs/SessionPool_Exception_Handling.md
@@ -34,9 +34,9 @@
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 @@
}
```
+## `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 @@
```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 @@
// Check pool health
Console.WriteLine($"Available Clients: {sessionPool.AvailableClients}");
Console.WriteLine($"Total Pool Size: {sessionPool.TotalPoolSize}");
+Console.WriteLine($"Unrealized Capacity: {sessionPool.UnrealizedCapacity}");
Console.WriteLine($"Failed Reconnections: {sessionPool.FailedReconnections}");
```
@@ -81,8 +128,27 @@
| -------------------- | --------------------- | ------------------------------------------------ | --------------------------- |
| 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) |
+| 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. `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 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
+ `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
### Scenario 1: Pool Exhaustion (High Load)
@@ -101,9 +167,9 @@
```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 +203,25 @@
**Symptoms:**
- `SessionPoolDepletedException` with reason "Reconnection failed"
-- `AvailableClients` decreases over time
-- `FailedReconnections` > 0 and increasing
+- `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
**Recovery Strategies:**
+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)
+{
+ // Capacity is retained and refilled on demand - just back off and try again
+ await Task.Delay(2000);
+}
+```
+
1. **Reinitialize SessionPool:**
```csharp
@@ -159,9 +237,9 @@
// 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 +323,9 @@
```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 +459,10 @@
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 +559,11 @@
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`, 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
- 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..f59b631 100644
--- a/src/Apache.IoTDB/ConcurrentClientQueue.cs
+++ b/src/Apache.IoTDB/ConcurrentClientQueue.cs
@@ -58,26 +58,52 @@
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;
+
+ /// <summary>
+ /// The maximum time, in milliseconds, that <see cref="Take"/> waits for a client to be
+ /// returned to the pool before throwing. Defaults to 10000 (10 seconds).
+ /// </summary>
+ public int TimeoutInMs { get; set; } = DefaultTimeoutInMs;
+
+ internal const int DefaultTimeoutInMs = 10_000;
+
+ /// <summary>
+ /// The wait timeout expressed in seconds. Kept for backward compatibility only; it is a thin
+ /// wrapper over <see cref="TimeoutInMs"/>. Prefer <see cref="TimeoutInMs"/>, which avoids the
+ /// unit ambiguity that previously caused millisecond values to be interpreted as seconds.
+ /// </summary>
+ [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;
+ // 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)
- {
- timeout = !Monitor.Wait(ClientQueue, TimeSpan.FromSeconds(Timeout));
- }
- ClientQueue.TryDequeue(out client);
-
- if (client != null || timeout)
+ if (ClientQueue.TryDequeue(out client))
{
break;
}
+
+ var remainingMs = budgetMs - (int)elapsed.ElapsedMilliseconds;
+ if (remainingMs <= 0)
+ {
+ break;
+ }
+
+ Monitor.Wait(ClientQueue, TimeSpan.FromMilliseconds(remainingMs));
}
}
finally
@@ -86,7 +112,7 @@
}
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({budgetMs}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 @@
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 @@
return this;
}
+ /// <summary>
+ /// Sets how long, in milliseconds, an operation waits for a client to become available in the pool
+ /// before a <see cref="SessionPoolDepletedException"/> is thrown. Defaults to
+ /// <see cref="DefaultPoolWaitTimeoutInMs"/> (10 seconds). This is independent of
+ /// <see cref="SetConnectionTimeoutInMs"/>, which controls the socket-level timeout.
+ /// </summary>
+ public Builder SetPoolWaitTimeoutInMs(int poolWaitTimeoutInMs)
+ {
+ _poolWaitTimeoutInMs = poolWaitTimeoutInMs;
+ return this;
+ }
+
public Builder SetUseSsl(bool useSsl)
{
_useSsl = useSsl;
@@ -135,6 +148,7 @@
_poolSize = 8;
_enableRpcCompression = false;
_connectionTimeoutInMs = 500;
+ _poolWaitTimeoutInMs = DefaultPoolWaitTimeoutInMs;
_useSsl = false;
_certificatePath = null;
_sqlDialect = IoTDBConstant.TREE_SQL_DIALECT;
@@ -146,9 +160,9 @@
// 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..fff1ea3 100644
--- a/src/Apache.IoTDB/SessionPool.cs
+++ b/src/Apache.IoTDB/SessionPool.cs
@@ -40,6 +40,12 @@
private static readonly TSProtocolVersion ProtocolVersion = TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3;
private const string DepletionReasonReconnectFailed = "Reconnection failed";
+ /// <summary>
+ /// Default time, in milliseconds, that an operation waits for a client to become available
+ /// in the pool before a <see cref="SessionPoolDepletedException"/> is raised.
+ /// </summary>
+ public const int DefaultPoolWaitTimeoutInMs = 10_000;
+
private readonly string _username;
private readonly string _password;
private bool _enableRpcCompression;
@@ -55,6 +61,11 @@
/// _timeout is the amount of time a Session will wait for a send operation to complete successfully.
/// </summary>
private readonly int _timeout;
+ /// <summary>
+ /// _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.
+ /// </summary>
+ private readonly int _poolWaitTimeoutInMs = DefaultPoolWaitTimeoutInMs;
private readonly int _poolSize = 4;
private readonly string _sqlDialect = IoTDBConstant.TREE_SQL_DIALECT;
private string _database;
@@ -65,6 +76,14 @@
private ConcurrentClientQueue _clients;
private ILogger _logger;
private PoolHealthMetrics _healthMetrics;
+ /// <summary>
+ /// 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.
+ /// </summary>
+ private int _unrealizedCapacity;
public delegate Task<TResult> AsyncOperation<TResult>(Client client);
@@ -83,6 +102,14 @@
/// </summary>
public int FailedReconnections => _healthMetrics?.GetReconnectionFailureTally() ?? 0;
+ /// <summary>
+ /// 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 <see cref="FailedReconnections"/>
+ /// to reason about server availability.
+ /// </summary>
+ public int UnrealizedCapacity => Volatile.Read(ref _unrealizedCapacity);
+
[Obsolete("This method is deprecated, please use new SessionPool.Builder().")]
public SessionPool(string host, int port, int poolSize)
@@ -110,6 +137,10 @@
{
}
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 +156,7 @@
_certificatePath = certificatePath;
_sqlDialect = sqlDialect;
_database = database;
+ _poolWaitTimeoutInMs = poolWaitTimeoutInMs;
}
/// <summary>
/// Initializes a new instance of the <see cref="SessionPool"/> class.
@@ -153,6 +185,10 @@
}
protected internal SessionPool(List<string> 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<string> 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 +208,61 @@
_certificatePath = certificatePath;
_sqlDialect = sqlDialect;
_database = database;
+ _poolWaitTimeoutInMs = poolWaitTimeoutInMs;
}
+ /// <summary>
+ /// 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.
+ /// </summary>
+ private async Task<Client> AcquireClientAsync(CancellationToken cancellationToken = default)
+ {
+ if (!_isClose && _clients.ClientQueue.IsEmpty && TryReserveUnrealizedCapacity())
+ {
+ try
+ {
+ return await Reconnect(cancellationToken: cancellationToken);
+ }
+ catch (ReconnectionFailedException reconnectEx)
+ {
+ // Still unreachable - hand the slot back so a later call can retry.
+ 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 _unrealizedCapacity);
+ throw;
+ }
+ }
+
+ return _clients.Take();
+ }
+
+ /// <summary>
+ /// Atomically claims one unit of unrealized capacity, returning false when none is left.
+ /// </summary>
+ private bool TryReserveUnrealizedCapacity()
+ {
+ while (true)
+ {
+ int current = Volatile.Read(ref _unrealizedCapacity);
+ if (current <= 0)
+ {
+ return false;
+ }
+ if (Interlocked.CompareExchange(ref _unrealizedCapacity, current - 1, current) == current)
+ {
+ return true;
+ }
+ }
+ }
+
public async Task<TResult> ExecuteClientOperationAsync<TResult>(AsyncOperation<TResult> 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 +283,11 @@
}
catch (ReconnectionFailedException reconnectEx)
{
- // Reconnection failed - original client was closed by Reconnect
+ // Reconnection failed - original client was closed by Reconnect. Record the now-empty
+ // slot so the pool keeps its configured capacity and can rebuild it later, instead of
+ // shrinking by one on every failure until every caller blocks forever.
shouldReturnClient = false;
+ Interlocked.Increment(ref _unrealizedCapacity);
throw new SessionPoolDepletedException(DepletionReasonReconnectFailed, AvailableClients, TotalPoolSize, FailedReconnections, reconnectEx);
}
@@ -268,8 +358,9 @@
{
_healthMetrics = new PoolHealthMetrics(_poolSize);
_clients = new ConcurrentClientQueue();
- _clients.Timeout = _timeout * 5;
+ _clients.TimeoutInMs = _poolWaitTimeoutInMs;
_clients.DiagnosticReporter = this;
+ Volatile.Write(ref _unrealizedCapacity, 0);
if (_nodeUrls.Count == 0)
{
@@ -344,7 +435,9 @@
}
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 +465,16 @@
throw new ReconnectionFailedException("Error occurs when reconnecting session pool. Could not connect to any server");
}
+ /// <summary>
+ /// Indicates whether this pool has been opened and not yet closed by the caller.
+ /// </summary>
+ /// <remarks>
+ /// 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 <see cref="Close"/> is called. To reason about connectivity, use
+ /// <see cref="AvailableClients"/>, <see cref="UnrealizedCapacity"/> and <see cref="FailedReconnections"/>,
+ /// or simply let an operation throw <see cref="SessionPoolDepletedException"/>.
+ /// </remarks>
public bool IsOpen() => !_isClose;
public async Task Close()
@@ -381,6 +484,13 @@
return;
}
+ // Flip the lifecycle state first. It must not depend on how many clients happen to be queued:
+ // 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 _unrealizedCapacity, 0);
+
foreach (var client in _clients.ClientQueue.AsEnumerable())
{
var closeSessionRequest = new TSCloseSessionReq(client.SessionId);
@@ -394,8 +504,6 @@
}
finally
{
- _isClose = true;
-
client.Transport?.Close();
}
}
@@ -430,7 +538,7 @@
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 @@
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 @@
return this;
}
+ /// <summary>
+ /// Sets how long, in milliseconds, an operation waits for a client to become available in the pool
+ /// before a <see cref="SessionPoolDepletedException"/> is thrown. Defaults to
+ /// <see cref="SessionPool.DefaultPoolWaitTimeoutInMs"/> (10 seconds). This is independent of
+ /// <see cref="SetConnectionTimeoutInMs"/>, which controls the socket-level timeout.
+ /// </summary>
+ public Builder SetPoolWaitTimeoutInMs(int poolWaitTimeoutInMs)
+ {
+ _poolWaitTimeoutInMs = poolWaitTimeoutInMs;
+ return this;
+ }
+
public Builder SetUseSsl(bool useSsl)
{
_useSsl = useSsl;
@@ -138,6 +151,7 @@
_poolSize = 8;
_enableRpcCompression = false;
_connectionTimeoutInMs = 500;
+ _poolWaitTimeoutInMs = SessionPool.DefaultPoolWaitTimeoutInMs;
_sqlDialect = IoTDBConstant.TABLE_SQL_DIALECT;
_database = "";
}
@@ -148,11 +162,11 @@
// 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..8511020
--- /dev/null
+++ b/tests/Apache.IoTDB.Tests/ConcurrentClientQueueTests.cs
@@ -0,0 +1,178 @@
+/*
+ * 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<TimeoutException>(() => 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<TimeoutException>(() => 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
+
+ [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<TimeoutException>(() => 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
new file mode 100644
index 0000000..e27cf96
--- /dev/null
+++ b/tests/Apache.IoTDB.Tests/SessionPoolConfigurationTests.cs
@@ -0,0 +1,135 @@
+/*
+ * 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 System.Threading.Tasks;
+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<string> { "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_StartsWithNoUnrealizedCapacityAndIsNotOpen()
+ {
+ var pool = new SessionPool.Builder().SetHost("127.0.0.1").SetPort(6667).Build();
+
+ Assert.That(pool.UnrealizedCapacity, 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 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, "_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.UnrealizedCapacity, 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);
+ }
+ }
+}
