diff --git a/astro.config.mjs b/astro.config.mjs index 9bda7bc7..9b36b494 100755 --- a/astro.config.mjs +++ b/astro.config.mjs @@ -145,6 +145,8 @@ export default defineConfig({ "how-to/connection-management", "how-to/publish-and-subscribe-messages", "how-to/send-batch-commands", + "how-to/client-pool", + "how-to/isolated-scope", "how-to/synchronous-connection", "how-to/execute-custom-scripts", "how-to/execute-custom-commands", diff --git a/src/content/docs/how-to/client-pool.mdx b/src/content/docs/how-to/client-pool.mdx new file mode 100644 index 00000000..dc772ec3 --- /dev/null +++ b/src/content/docs/how-to/client-pool.mdx @@ -0,0 +1,230 @@ +--- +title: Use Client Pool +description: Manage a pool of reusable GLIDE client connections for high-concurrency workloads, blocking commands, and connection isolation. +--- + +import { Tabs, TabItem } from '@astrojs/starlight/components'; + +The Client Pool manages a set of reusable GLIDE client instances that can be borrowed and returned on demand. Use it when your application needs high concurrency, runs blocking commands like `BLPOP`, `BRPOP`, or `XREAD BLOCK` that tie up a connection, or requires connection isolation between independent workflows. + +## Configuration + +The pool is configured with a `PoolConfig` object. All parameters are optional and have sensible defaults. + +| Parameter | Type | Default | Description | +|-----------|------|---------|-------------| +| `max_size` / `maxSize` / `MaxSize` | int | 10 | Maximum number of client instances in the pool | +| `min_idle` / `minIdle` / `MinIdle` | int | 1 | Number of clients to pre-warm at creation | +| `idle_timeout_ms` / `idleTimeout` / `IdleTimeout` | duration | 5 min | How long an idle client stays in the pool before eviction | +| `request_timeout_ms` / `requestTimeout` / `RequestTimeout` | duration | 5 sec | Timeout for state reset (DISCARD, UNWATCH) when returning a client | +| `acquire_timeout_s` / `acquireTimeoutS` / `AcquireTimeout` | duration | 5 sec | How long to wait for a client when the pool is exhausted | +| `abandon_timeout_ms` / `abandonTimeoutMs` / `AbandonTimeout` | duration | 5 min | Max inactivity time before the pool discards an unreturned client. Set to 0 to disable (Go: use a negative value to disable, 0 means default). | + +## Create a Pool + + + + ```python + from glide import GlideClientConfiguration, NodeAddress + from glide import AsyncClientPool, PoolConfig + + pool_config = PoolConfig( + max_size=10, + min_idle=2, + acquire_timeout_s=5.0, + ) + client_config = GlideClientConfiguration([NodeAddress("localhost", 6379)]) + pool = await AsyncClientPool.create(client_config, pool_config) + ``` + + + + ```java + import glide.api.models.pool.ClientPool; + import glide.api.models.pool.ClientPoolConfig; + import glide.api.models.configuration.GlideClientConfiguration; + import glide.api.models.configuration.NodeAddress; + + ClientPoolConfig poolConfig = ClientPoolConfig.builder() + .maxSize(10) + .minIdle(2) + .acquireTimeout(Duration.ofSeconds(5)) + .clientConfig(GlideClientConfiguration.builder() + .address(NodeAddress.builder().host("localhost").port(6379).build()) + .build()) + .build(); + + ClientPool pool = ClientPool.create(poolConfig); + ``` + + + + ```typescript + import { ClientPool, GlideClientConfiguration } from '@valkey/valkey-glide'; + + const pool = await ClientPool.create( + { addresses: [{ host: 'localhost', port: 6379 }] }, + { maxSize: 10, minIdle: 2, acquireTimeoutS: 5 } + ); + ``` + + + + ```go + import ( + glide "github.com/valkey-io/valkey-glide/go/v2" + "github.com/valkey-io/valkey-glide/go/v2/config" + "time" + ) + + poolCfg := glide.PoolConfig{ + MaxSize: 10, + MinIdle: 2, + AcquireTimeout: 5 * time.Second, + } + clientCfg := config.NewClientConfiguration(). + WithAddress(&config.NodeAddress{Host: "localhost", Port: 6379}) + + pool, err := glide.NewClientPool(clientCfg, poolCfg) + ``` + + + +## Borrow and Return Clients + +Acquire a client from the pool before use, and always release it when done. Use context managers or try/finally blocks to guarantee the client is returned even if an error occurs. + + + + ```python + async with pool.borrow() as client: + await client.set("key", "value") + result = await client.get("key") + ``` + + + + ```java + try (PooledGlideClient client = pool.acquire().get()) { + client.set("key", "value").get(); + String val = client.get("key").get(); + } // automatically returned to pool + ``` + + + + ```typescript + const client = await pool.acquire(); + try { + await client.set('key', 'value'); + const result = await client.get('key'); + } finally { + await pool.release(client); + } + ``` + + + + ```go + clientID, err := pool.Acquire(ctx) + if err != nil { + log.Fatal(err) + } + defer pool.Release(clientID) + + client, err := pool.GetClient(clientID) + if err != nil { + log.Fatal(err) + } + client.Set(ctx, "key", "value") + result, _ := client.Get(ctx, "key") + ``` + + + +## Pool Metrics + +Check the current state of the pool to monitor utilization and diagnose capacity issues. + + + + ```python + metrics = pool.metrics() + print(f"Idle: {metrics['idle']}, Active: {metrics['active']}, Total: {metrics['total']}") + ``` + + + + ```java + System.out.println("Idle: " + pool.getIdleCount() + + ", Active: " + pool.getActiveCount() + + ", Total: " + pool.getTotalCount()); + ``` + + + + ```typescript + const metrics = pool.getMetrics(); + console.log(`Idle: ${metrics.idle}, Active: ${metrics.active}, Total: ${metrics.total}`); + ``` + + + + ```go + fmt.Printf("Idle: %d, Active: %d, Total: %d\n", pool.IdleCount(), pool.ActiveCount(), pool.TotalCount()) + ``` + + + +## Close the Pool + +When your application shuts down, close the pool to release all connections. + + + + ```python + pool.close() + + # Or use as an async context manager: + async with AsyncClientPool.create(config, pool_config) as pool: + async with pool.borrow() as client: + await client.set("key", "value") + ``` + + + + ```java + pool.close(); + ``` + + + + ```typescript + pool.close(); + ``` + + + + ```go + pool.Close() + ``` + + + +## Abandon Detection + +The pool runs a background monitor that detects borrowed clients left idle (no commands sent) for longer than `abandon_timeout`. When triggered: + +1. The idle client's connection is discarded (closed). +2. The pool slot is freed for a new connection on the next `acquire()`. +3. A warning is logged identifying the abandoned client. + +The inactivity timer resets on every command sent through the borrowed client. Blocking commands (`BLPOP`, `BRPOP`, `XREAD BLOCK`, etc.) are automatically exempt — the monitor skips clients currently executing a blocking command. + +Set `abandon_timeout` to 0 (or negative in Go) to disable the monitor entirely. + +## Limitations + +- **No Pub/Sub subscriptions:** Clients with active subscriptions are rejected at pool creation because `UNSUBSCRIBE` cannot be reliably sent during state reset. +- **No background eviction:** Idle clients are evicted lazily when `acquire()` is called, not on a background timer. +- **State reset always sends DISCARD:** When a client is returned, the pool sends `DISCARD` and `UNWATCH` even if no `MULTI` transaction was active. This adds a small round-trip on each release. diff --git a/src/content/docs/how-to/isolated-scope.mdx b/src/content/docs/how-to/isolated-scope.mdx new file mode 100644 index 00000000..c3bc4d30 --- /dev/null +++ b/src/content/docs/how-to/isolated-scope.mdx @@ -0,0 +1,222 @@ +--- +title: Use Isolated Execution Scopes +description: Run WATCH/MULTI/EXEC transactions on a dedicated connection with optimistic concurrency control. +--- + +import { Tabs, TabItem } from '@astrojs/starlight/components'; + +An Isolated Execution Scope provides a dedicated, per-connection state for running `WATCH`/`MULTI`/`EXEC` transactions. GLIDE clients normally multiplex commands over shared connections, which means per-connection state like `WATCH` cannot be held safely for a single caller. A scope gives you an exclusive connection where your `WATCH` keys remain valid until you call `EXEC`. + +Use isolated scopes when you need optimistic concurrency control (OCC) — for example, incrementing a counter only if no other client modified it, or implementing compare-and-swap patterns. + +## Acquire a Scope + + + + ```python + from glide import GlideClient + + client = await GlideClient.create(config) + + async with await client.scoped_connection(routing_key="my-key") as scope: + # Use scope for WATCH/MULTI/EXEC + pass + ``` + + + + ```java + import glide.api.GlideClient; + import glide.api.models.scope.IsolatedScope; + + GlideClient client = GlideClient.createClient(config).get(); + + try (IsolatedScope scope = client.scopedConnection( + Duration.ofSeconds(5), "my-key").get()) { + // Use scope for WATCH/MULTI/EXEC + } + ``` + + + + ```typescript + import { GlideClient } from '@valkey/valkey-glide'; + + const client = await GlideClient.createClient(config); + + const scope = await client.scopedConnection({ routingKey: 'my-key' }); + try { + // Use scope for WATCH/MULTI/EXEC + } finally { + await scope.close(); + } + ``` + + + + ```go + scope, err := client.ScopedConnection(ctx, 10*time.Second, "my-key") + if err != nil { + log.Fatal(err) + } + defer scope.Close() + // Use scope for WATCH/MULTI/EXEC + ``` + + + +## Optimistic Locking with WATCH/MULTI/EXEC + +The classic optimistic concurrency control pattern works as follows: + +1. Acquire a scope +2. `WATCH` the key(s) you intend to modify +3. Read the current value +4. Begin a `MULTI` transaction +5. Write the new value +6. `EXEC` — returns `null` if any watched key was modified by another client +7. Retry if `EXEC` returned `null` + + + + ```python + from glide import GlideClient + + client = await GlideClient.create(config) + + max_retries = 10 + for attempt in range(max_retries): + async with await client.scoped_connection(routing_key="counter") as scope: + await scope.watch("counter") + val = await scope.get("counter") + new_val = str(int(val or "0") + 1) + await scope.multi() + await scope.set("counter", new_val) + result = await scope.exec() + if result is not None: + break # Success + # WATCH conflict — retry + ``` + + + + ```java + import glide.api.GlideClient; + import glide.api.models.scope.IsolatedScope; + + GlideClient client = GlideClient.createClient(config).get(); + + for (int attempt = 0; attempt < 10; attempt++) { + try (IsolatedScope scope = client.scopedConnection( + Duration.ofSeconds(5), "counter").get()) { + scope.watch("counter").get(); + String val = scope.get("counter").get(); + int newVal = (val != null ? Integer.parseInt(val) : 0) + 1; + scope.multi().get(); + scope.set("counter", String.valueOf(newVal)).get(); + Object result = scope.exec().get(); + if (result != null) break; // Success + } + } + ``` + + + + ```typescript + import { GlideClient } from '@valkey/valkey-glide'; + + const client = await GlideClient.createClient(config); + + for (let attempt = 0; attempt < 10; attempt++) { + const scope = await client.scopedConnection({ routingKey: 'counter' }); + try { + await scope.watch('counter'); + const val = await scope.get('counter'); + const newVal = String((parseInt(val ?? '0', 10)) + 1); + await scope.multi(); + await scope.set('counter', newVal); + const result = await scope.exec(); + if (result !== null) break; // Success + } finally { + await scope.close(); + } + } + ``` + + + + ```go + for attempt := 0; attempt < 10; attempt++ { + scope, err := client.ScopedConnection(ctx, 10*time.Second, "counter") + if err != nil { + log.Fatal(err) + } + + scope.Watch(ctx, "counter") + val, _ := scope.Get(ctx, "counter") + newVal := strconv.Itoa(atoi(val) + 1) + scope.Multi(ctx) + scope.Set(ctx, "counter", newVal) + result, _ := scope.Exec(ctx) + scope.Close() + if result != nil { + break // Success + } + } + ``` + + + +## Cluster Mode + +When using isolated scopes with a cluster client, keep the following in mind: + +### Routing Key + +Pass a `routing_key` when acquiring the scope. The key is used to determine which cluster node (via CRC16 slot hashing) the scope's dedicated connection targets. + + + + ```python + async with await client.scoped_connection(routing_key="user:123") as scope: + await scope.watch("user:123") + # ... + ``` + + + + ```java + try (IsolatedScope scope = client.scopedConnection( + Duration.ofSeconds(5), "user:123").get()) { + scope.watch("user:123").get(); + // ... + } + ``` + + + + ```typescript + const scope = await client.scopedConnection({ routingKey: 'user:123' }); + ``` + + + + ```go + scope, _ := client.ScopedConnection(ctx, 10*time.Second, "user:123") + ``` + + + +### Same-Slot Requirement + +All keys used within a scope must hash to the same slot. If you attempt a cross-slot operation, the command will be rejected. Use hash tags (e.g., `{user:123}:name` and `{user:123}:email`) to ensure related keys land in the same slot. + +### MOVED Handling + +If a `MOVED` redirect occurs during an active scope (for example, during a cluster rebalance), the `WATCH` detects this as a conflict and `EXEC` returns `null`. Your retry loop will handle this naturally — simply retry and the scope will connect to the new node for that slot. + +## Limitations + +- **No Pub/Sub:** Scoped connections do not support `SUBSCRIBE` or `PSUBSCRIBE`. Use the main client for Pub/Sub. +- **No MOVED/ASK redirects:** By design, scoped connections hold per-connection state and cannot follow redirects. Conflicts are surfaced as `EXEC` returning `null`. +- **Single caller per scope:** Only one coroutine or thread should use a scope at a time. Concurrent access to the same scope leads to undefined behavior.