Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/API.md
Original file line number Diff line number Diff line change
Expand Up @@ -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() |
Expand Down
126 changes: 104 additions & 22 deletions docs/SessionPool_Exception_Handling.md
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand All @@ -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
Expand All @@ -62,16 +108,17 @@ 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();

// 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}");
```

Expand All @@ -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)
Expand All @@ -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();
```

Expand Down Expand Up @@ -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
Expand All @@ -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();
Expand Down Expand Up @@ -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();
```

Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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
40 changes: 33 additions & 7 deletions src/Apache.IoTDB/ConcurrentClientQueue.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;

/// <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)
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
Expand All @@ -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);
Expand Down
18 changes: 16 additions & 2 deletions src/Apache.IoTDB/SessionPool.Builder.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -94,6 +95,18 @@ public Builder SetConnectionTimeoutInMs(int timeout)
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;
Expand Down Expand Up @@ -135,6 +148,7 @@ public Builder()
_poolSize = 8;
_enableRpcCompression = false;
_connectionTimeoutInMs = 500;
_poolWaitTimeoutInMs = DefaultPoolWaitTimeoutInMs;
_useSsl = false;
_certificatePath = null;
_sqlDialect = IoTDBConstant.TREE_SQL_DIALECT;
Expand All @@ -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);
}
}
}
Loading
Loading