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..99039de 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($"Unrealized Capacity: {sessionPool.UnrealizedCapacity}"); Console.WriteLine($"Failed Reconnections: {sessionPool.FailedReconnections}"); ``` @@ -81,8 +128,27 @@ 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) | +| 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 @@ 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 +203,25 @@ for (int i = 0; i < maxRetries; i++) **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 @@ 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 +323,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 +459,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 +559,11 @@ 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`, 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 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; + // 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.FromSeconds(Timeout)); + 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 @@ -86,7 +112,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({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 @@ 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..fff1ea3 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,14 @@ public partial class SessionPool : IDisposable, IPoolDiagnosticReporter private ConcurrentClientQueue _clients; private ILogger _logger; private PoolHealthMetrics _healthMetrics; + /// + /// 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 _unrealizedCapacity; public delegate Task AsyncOperation(Client client); @@ -83,6 +102,14 @@ public partial class SessionPool : IDisposable, IPoolDiagnosticReporter /// public int FailedReconnections => _healthMetrics?.GetReconnectionFailureTally() ?? 0; + /// + /// 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 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 @@ 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 +156,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 +185,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 +208,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 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 (!_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(); + } + + /// + /// Atomically claims one unit of unrealized capacity, returning false when none is left. + /// + 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 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 +283,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 +465,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() @@ -381,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 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 @@ public async Task Close() } finally { - _isClose = true; - client.Transport?.Close(); } } @@ -430,7 +538,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..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(() => 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 + + [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 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 { "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); + } + } +}