From 30d149a17a1adead60c0b37fc8fbe88268c29f6d Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Mon, 12 Oct 2026 15:18:35 +0100 Subject: [PATCH] redis: drop ioxide.redis ioxide.redis is weight the project does not want to carry: a whole Redis client to maintain - five open fixes against it - when StackExchange.Redis can run on the reactor's ring instead (#335, ioxide.redis.stackexchange). Removed: the package, its Playground sample (Clients/Redis) and the Env helper only it used, its tests (the Redis suite stays, empty, for the transport's tests), its CI pack step, its bench run and row, its pane on the docs page, and docs/learn's two Redis pages with their links. The passing Redis mentions in the Http3/Nghttp3Buffered sample's comments go too. The versions already on nuget.org are untouched. --- .github/workflows/build.yml | 3 - .../Redis/Playground.Clients.Redis.csproj | 19 - Playground/Clients/Redis/Program.cs | 238 ----------- Playground/Http3/Nghttp3Buffered/Program.cs | 6 +- Playground/README.md | 1 - Playground/Shared/Env.cs | 9 - README.md | 1 - bench/run.sh | 13 +- bench/samples.tsv | 1 - docs/assets/style.css | 3 - docs/how-ioxide-does-h3.html | 2 - docs/index.html | 231 +---------- docs/learn/architecture.html | 2 - docs/learn/clients.html | 2 - docs/learn/dev-core.html | 8 +- docs/learn/dev-file.html | 2 - docs/learn/dev-pg.html | 4 +- docs/learn/dev-redis.html | 249 ----------- docs/learn/dev-tls.html | 2 - docs/learn/files.html | 2 - docs/learn/multiport.html | 2 - docs/learn/overview.html | 9 +- docs/learn/postgres.html | 4 +- docs/learn/redis.html | 144 ------- docs/learn/tls.html | 2 - ioxide.slnx | 2 - scripts/gen-docs-panes.py | 6 - src/clients/ioxide.redis/README.md | 40 -- src/clients/ioxide.redis/RedisArg.cs | 22 - .../ioxide.redis/RedisConnection.Commands.cs | 180 -------- src/clients/ioxide.redis/RedisConnection.cs | 385 ------------------ src/clients/ioxide.redis/RedisException.cs | 10 - src/clients/ioxide.redis/RedisOptions.cs | 32 -- src/clients/ioxide.redis/RedisPool.cs | 202 --------- src/clients/ioxide.redis/RespProtocol.cs | 232 ----------- src/clients/ioxide.redis/RespValue.cs | 92 ----- src/clients/ioxide.redis/ioxide.redis.csproj | 27 -- .../Ioxide.Tests.Redis.csproj | 1 - tests/Ioxide.Tests.Redis/Program.cs | 3 +- tests/Ioxide.Tests.Redis/RedisHandlers.cs | 56 --- tests/Ioxide.Tests.Redis/RedisTests.cs | 42 -- tests/README.md | 2 +- 42 files changed, 18 insertions(+), 2275 deletions(-) delete mode 100644 Playground/Clients/Redis/Playground.Clients.Redis.csproj delete mode 100644 Playground/Clients/Redis/Program.cs delete mode 100644 docs/learn/dev-redis.html delete mode 100644 docs/learn/redis.html delete mode 100644 src/clients/ioxide.redis/README.md delete mode 100644 src/clients/ioxide.redis/RedisArg.cs delete mode 100644 src/clients/ioxide.redis/RedisConnection.Commands.cs delete mode 100644 src/clients/ioxide.redis/RedisConnection.cs delete mode 100644 src/clients/ioxide.redis/RedisException.cs delete mode 100644 src/clients/ioxide.redis/RedisOptions.cs delete mode 100644 src/clients/ioxide.redis/RedisPool.cs delete mode 100644 src/clients/ioxide.redis/RespProtocol.cs delete mode 100644 src/clients/ioxide.redis/RespValue.cs delete mode 100644 src/clients/ioxide.redis/ioxide.redis.csproj delete mode 100644 tests/Ioxide.Tests.Redis/RedisHandlers.cs delete mode 100644 tests/Ioxide.Tests.Redis/RedisTests.cs diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 2d10077e..e5e278f6 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -115,9 +115,6 @@ jobs: - name: Pack ioxide.pg run: dotnet pack src/clients/ioxide.pg/ioxide.pg.csproj --configuration Release --no-build --output ./artifacts - - name: Pack ioxide.redis - run: dotnet pack src/clients/ioxide.redis/ioxide.redis.csproj --configuration Release --no-build --output ./artifacts - - name: Pack ioxide.timer run: dotnet pack src/clients/ioxide.timer/ioxide.timer.csproj --configuration Release --no-build --output ./artifacts diff --git a/Playground/Clients/Redis/Playground.Clients.Redis.csproj b/Playground/Clients/Redis/Playground.Clients.Redis.csproj deleted file mode 100644 index 2867ba24..00000000 --- a/Playground/Clients/Redis/Playground.Clients.Redis.csproj +++ /dev/null @@ -1,19 +0,0 @@ - - - - Exe - net11.0 - enable - enable - true - Playground.Clients.Redis - Playground.Clients.Redis - - - - - - - - - diff --git a/Playground/Clients/Redis/Program.cs b/Playground/Clients/Redis/Program.cs deleted file mode 100644 index b2325c80..00000000 --- a/Playground/Clients/Redis/Program.cs +++ /dev/null @@ -1,238 +0,0 @@ -using System.Text; -using ioxide; -using ioxide.redis; -using ioxide.utils; -using Playground.Shared; - -// ───────────────────────────────────────────────────────────────────────────────────────────── -// redis - a RedisPool per reactor, driven on the same ring that accepted the request. Every -// await below resumes inline on the reactor thread; commands from concurrent requests share -// the pool's connections and pipeline automatically. -// -// docker run --rm -p 6379:6379 redis # or point PLAYGROUND_REDIS_* elsewhere -// dotnet run -c Release --project Playground/Redis -// -// curl http://127.0.0.1:8080/ # GET hot path (the wrk-able route) -// curl http://127.0.0.1:8080/cache/user42 # cache-aside: GET, miss -> SET EX -// curl http://127.0.0.1:8080/incr # RESP integer reply -// curl http://127.0.0.1:8080/hash # RESP array (HGETALL) -// curl http://127.0.0.1:8080/list # RESP array (LRANGE) -// curl http://127.0.0.1:8080/pipeline # SET+INCR+GET in one round trip -// -// Needs: ioxide.redis -// ───────────────────────────────────────────────────────────────────────────────────────────── - -// ── Knobs ──────────────────────────────────────────────────────────────────────────────────── -// Edit these. That is the whole mechanism - there is no config file and nothing else to find. -// An Env.Override line means the value can also be set from the environment, which is how -// bench/run.sh drives the sample; the literal is what applies otherwise. Delete those lines when -// you copy this out and the literals above them are the entire configuration. - -ushort port = 8080; // http://127.0.0.1:8080/ -int reactors = Environment.ProcessorCount; // one ring per reactor, one reactor per core - -Env.Override(ref port, ref reactors); - -// The Redis this talks to. IPv4 literal for the same reason as everywhere else - a DNS lookup -// would block the reactor - and the pool is per reactor. A null password means no AUTH. -string redisHost = "127.0.0.1"; -ushort redisPort = 6379; -string? redisPassword = null; -int redisPoolSize = 4; - -Env.OverrideRedis(ref redisHost, ref redisPort, ref redisPassword, ref redisPoolSize); -// ───────────────────────────────────────────────────────────────────────────────────────────── - -var config = new ServerConfig -{ - ReactorCount = reactors, // io_uring rings/threads - one per core - RingEntries = 8192, // SQ/CQ depth per ring - DualStack = false, // true = one IPv6 socket also accepts IPv4-mapped - RecvBufferSize = 32 * 1024, // bytes per shared recv buffer - RecvSlots = 4096, // shared recv buffer-ring depth - Incremental = null, // per-connection recv rings (6.12+) - see Tcp/Incremental - Udp = null, // no raw UDP sockets (TCP-only server) - Quic = null, // no QUIC transport - see Http3/* and Quic/Alpn - Tcp = new TcpOptions - { - Port = port, - ExtraPorts = [], // extra listener ports (one handler, several doors) - ListenBacklog = 1024, // accept-queue depth per SO_REUSEPORT listener - WriteSlabSize = 16 * 1024, // per-connection write buffer before overflow kicks in - PoolMax = 1024, // pooled connection objects kept per reactor - WriteOverflow = WriteOverflowStrategy.Grow, // Grow = realloc one slab; Segmented = chain + vectored SENDMSG - ZeroCopySend = false, // SEND_ZC: kernel copies less, wins on large writes - RecvQueueEntries = 4096, // per-connection recv completion queue depth - ReadTimeoutMs = 60_000, // close a connection whose read waits this long on a silent peer; 0 = off - SendTimeoutMs = 60_000, // close a connection whose flush takes this long to go out; 0 = off - }, -}; - -var redisOptions = new RedisOptions -{ - Host = redisHost, // IPv4 literal - resolve names up front, DNS blocks the reactor - Port = redisPort, // Redis wire port - Password = redisPassword, // AUTH password; null = no auth - User = null, // ACL username (Redis 6+); null = default user - Database = 0, // logical DB index to SELECT on connect - PoolSize = redisPoolSize, // per reactor, not global - CommandTimeoutMs = 30_000, // oldest in-flight command past this -> torn down; 0 disables -}; - -var threads = new Thread[config.ReactorCount]; - -for (int i = 0; i < threads.Length; i++) -{ - var reactor = new Reactor(i, config); - - // The pool lives on the reactor: its connections ride this reactor's ring, so a query is - // an io_uring send + recv with the continuation resuming right here. - reactor.OnStart = r => - { - RedisPool pool = RedisPool.Start(r, redisOptions); - _ = SeedAsync(pool); // the "/" route reads this key; seed it so the demo answers - }; - - reactor.TcpHandle = async (r, conn) => - { - RedisPool pool = r.GetService(); - - try - { - while (true) - { - RecvSnapshot snapshot = await conn.ReadAsync(); - string path = ReadPath(conn, snapshot); - - try - { - string body = path switch - { - // Cache-aside: try the key; on a miss compute, SET with a TTL, return. - var p when p.StartsWith("/cache/", StringComparison.Ordinal) - => await CacheAside(pool, p["/cache/".Length..]), - - // One RESP reply type per route, through the generic ExecuteAsync. - "/incr" => $"counter = {(await pool.ExecuteAsync("INCR", "play:counter")).AsInteger()}", - "/hash" => await Hash(pool), - "/list" => await List(pool), - - // Several commands, one round trip, replies in order. - "/pipeline" => await Pipeline(pool), - - // The hot path: a single GET per request. wrk this one. - _ => $"redis = {await pool.GetAsync("play:bench") ?? "(nil)"}", - }; - - WriteText(conn, "200 OK", body); - } - catch (RedisException e) - { - // A server error is an exception, not a broken socket: the connection has - // resynced and goes back to the pool usable. - WriteText(conn, "500 Internal Server Error", e.Message); - } - - await conn.FlushAsync(); - - if (snapshot.IsClosed) return; - conn.ResetRead(); - } - } - finally - { - conn.DecRef(); - } - }; - - threads[i] = new Thread(reactor.Run) { Name = $"reactor-{i}" }; - threads[i].Start(); -} - -Console.WriteLine($"[redis] {config.ReactorCount} reactors on :{config.Tcp!.Port} " - + $"-> {redisOptions.Host}:{redisOptions.Port}, {redisOptions.PoolSize} connections each"); - -foreach (Thread thread in threads) -{ - thread.Join(); -} - -static async Task SeedAsync(RedisPool pool) -{ - try - { - await pool.ExecuteAsync("SET", "play:bench", "42"); - } - catch (Exception e) { Console.Error.WriteLine($"[redis] seed failed: {e.Message}"); } -} - -static async Task CacheAside(RedisPool pool, string key) -{ - key = "play:cache:" + key; - string? cached = await pool.GetAsync(key); - if (cached != null) - { - return $"hit {key} = {cached}"; - } - - string value = $"value-{key.Length}"; // stands in for the expensive computation - await pool.SetExAsync(key, value, seconds: 60); - return $"miss {key} = {value} (cached 60s)"; -} - -// RESP array of bulk strings: field, value, field, value, ... -static async Task Hash(RedisPool pool) -{ - await pool.ExecuteAsync("HSET", "play:hash", "lang", "csharp", "engine", "io_uring"); - RespValue all = await pool.ExecuteAsync("HGETALL", "play:hash"); - return "hash = [" + string.Join(", ", all.Items.Select(v => v.AsString())) + "]"; -} - -static async Task List(RedisPool pool) -{ - await pool.ExecuteAsync("DEL", "play:list"); - await pool.ExecuteAsync("RPUSH", "play:list", "a", "b", "c"); - RespValue items = await pool.ExecuteAsync("LRANGE", "play:list", "0", "-1"); - return "list = [" + string.Join(", ", items.Items.Select(v => v.AsString())) + "]"; -} - -static async Task Pipeline(RedisPool pool) -{ - RespValue[] replies = await pool.PipelineAsync( - new RedisCommand("SET", "play:pipe", "1"), - new RedisCommand("INCR", "play:pipe"), - new RedisCommand("GET", "play:pipe")); - return $"set={replies[0].AsString()} incr={replies[1].AsInteger()} get={replies[2].AsString()}"; -} - -// "GET /cache/x HTTP/1.1" -> "/cache/x", draining the recv on the way past. -static string ReadPath(TcpConnection conn, RecvSnapshot snapshot) -{ - string path = "/"; - while (conn.TryGetItem(snapshot, out SpscRecvRing.Item item)) - { - if (item.HasBuffer) - { - ReadOnlySpan request = item.AsSpan(); - int firstSpace = request.IndexOf((byte)' '); - if (firstSpace >= 0) - { - ReadOnlySpan rest = request[(firstSpace + 1)..]; - int secondSpace = rest.IndexOf((byte)' '); - if (secondSpace > 0) - { - ReadOnlySpan target = rest[..secondSpace]; - int query = target.IndexOf((byte)'?'); - path = Encoding.ASCII.GetString(query >= 0 ? target[..query] : target); - } - } - conn.ReturnBuffer(in item); - } - } - - return path; -} - -static void WriteText(TcpConnection conn, string status, string body) - => conn.Write(Encoding.ASCII.GetBytes( - $"HTTP/1.1 {status}\r\nContent-Type: text/plain\r\nContent-Length: {body.Length}\r\n\r\n{body}")); diff --git a/Playground/Http3/Nghttp3Buffered/Program.cs b/Playground/Http3/Nghttp3Buffered/Program.cs index c501eafc..0d71f67b 100644 --- a/Playground/Http3/Nghttp3Buffered/Program.cs +++ b/Playground/Http3/Nghttp3Buffered/Program.cs @@ -11,7 +11,7 @@ // // BUFFERED: dispatch waits for end-of-stream, so the whole body is already in request.Body when // your handler runs - no BodyReader, no pacing - and the handler may still await (a PgPool query, -// Redis, anything ioxide-native resumes inline on the reactor). +// anything ioxide-native resumes inline on the reactor). // // The trade: memory holds the entire body, so this suits normal-sized requests. Use Playground/Http3/Nghttp3Request // when uploads can be large or hostile. @@ -149,8 +149,8 @@ { // Dispatch waited for end-of-stream, so the body is ALREADY here: request.Body is // complete and request.Body.Length is just a property read. No BodyReader, no pacing. - // This overload is synchronous, but the awaiting one exists too - a PgPool query or a - // Redis command resumes inline on this reactor, so you can await it right here. + // This overload is synchronous, but the awaiting one exists too - a PgPool query + // resumes inline on this reactor, so you can await it right here. _ = request.Body.Length; // One response object, reused for every request: zero allocations on this path. To diff --git a/Playground/README.md b/Playground/README.md index 8a6e56ac..64da19b6 100644 --- a/Playground/README.md +++ b/Playground/README.md @@ -125,7 +125,6 @@ reference implementation. QUIC itself is always ngtcp2 + picotls, bundled as one | [`Clients.Quic`](Clients/Quic/Program.cs) | 166 | The client half of QUIC, and the thing that drives `Quic/Raw`. No listener of any kind. | `ioxide.ngtcp2` | | [`Clients.File`](Clients/File/Program.cs) | 422 | A static file server: every file opened ONCE, descriptors shared across reactors, read positionally off the ring. | `ioxide.file` | | [`Clients.Pg`](Clients/Pg/Program.cs) | 224 | A Postgres-backed server, whole: a pool per reactor, connect/query/row streaming all ring operations. | `ioxide.pg` | -| [`Clients.Redis`](Clients/Redis/Program.cs) | 236 | A `RedisPool` per reactor; concurrent requests share connections and pipeline automatically. | `ioxide.redis` | | [`Clients.Timer`](Clients/Timer/Program.cs) | 184 | A deadline that stays on the ring: `GET /` answers after that long, one `IORING_OP_TIMEOUT` per wait. A knob swaps in `Task.Delay` to price the difference. | `ioxide.timer` | | [`Proxy.H1ToH1`](Proxy/H1ToH1/Program.cs) | 259 | A reverse proxy with TLS on BOTH hops - ioxide TLS in, `TlsClientContext` out. Read this one first. | `ioxide.httpclient` | | [`Proxy.H2ToH1`](Proxy/H2ToH1/Program.cs) | 233 | The classic edge: h2 over TLS in (browsers refuse h2c), h1 origin behind. | `ioxide.http2`, `ioxide.httpclient` | diff --git a/Playground/Shared/Env.cs b/Playground/Shared/Env.cs index a7607977..cff6df4b 100644 --- a/Playground/Shared/Env.cs +++ b/Playground/Shared/Env.cs @@ -147,15 +147,6 @@ public static void OverridePg(ref string host, ref ushort pgPort, ref string use commandTimeoutMs = Int("PLAYGROUND_PG_TIMEOUT", commandTimeoutMs); } - /// The Redis sample's connection settings. - public static void OverrideRedis(ref string host, ref ushort redisPort, ref string? password, ref int poolSize) - { - host = Str("PLAYGROUND_REDIS_HOST", host); - redisPort = Port("PLAYGROUND_REDIS_PORT", redisPort); - password = StrOrNull("PLAYGROUND_REDIS_PASSWORD") ?? password; - poolSize = Int("PLAYGROUND_REDIS_POOL", poolSize); - } - /// Where a proxy sample forwards to, and how it trusts the origin. public static void OverrideUpstream(ref string host, ref ushort upstreamPort, ref string serverName, ref int poolSize, ref string? caFile, ref bool insecure) diff --git a/README.md b/README.md index a8d786bb..99e08321 100644 --- a/README.md +++ b/README.md @@ -60,7 +60,6 @@ accepted the request. | --- | --- | | [`ioxide.httpclient`](https://www.nuget.org/packages/ioxide.httpclient/) | The HTTP/1.1 upstream leg between a proxy and an origin, with client-side TLS and mutual TLS for `https://`. | | [`ioxide.pg`](https://www.nuget.org/packages/ioxide.pg/) | Postgres. Connect, query and stream rows on the owning ring. | -| [`ioxide.redis`](https://www.nuget.org/packages/ioxide.redis/) | Redis. RESP2, pipelining, pub/sub. | | [`ioxide.file`](https://www.nuget.org/packages/ioxide.file/) | Static assets. Immutable snapshots, baked responses, positional ring reads. | | [`ioxide.timer`](https://www.nuget.org/packages/ioxide.timer/) | Deadlines. A wait is one `IORING_OP_TIMEOUT` on the caller's own ring - no timer thread, nothing allocated per wait. | diff --git a/bench/run.sh b/bench/run.sh index 8f6130b6..af836c0a 100755 --- a/bench/run.sh +++ b/bench/run.sh @@ -8,7 +8,7 @@ # Servers run 4 reactors (BENCH_REACTORS to override); the tcp-raw/tcp-pipe pair repeats at 12 # reactors as the io_uring baseline. Loads: wrk for everything TCP (fixed -c 64 -t 4, # time-bounded), h2load built with HTTP/3 for the HTTP/3 server (H2LOAD_H3 to point at it), Bench.Clients for the -# ring-native clients. Sidecars (redis, postgres, an h2c nginx) start via docker when available; +# ring-native clients. Sidecars (postgres, an h2c nginx) start via docker when available; # anything missing skips with a note instead of failing the run. # # Numbers are hardware-specific: compare runs on one machine, never across machines. @@ -126,16 +126,8 @@ else note "client-h2" "${R}r" "SKIP (docker unavailable for the h2c upstream)" fi -# ── redis / pg / file ─────────────────────────────────────────────────────────────────────── +# ── pg / file ─────────────────────────────────────────────────────────────────────── if command -v docker >/dev/null 2>&1 && docker info >/dev/null 2>&1; then - docker rm -f bench-redis >/dev/null 2>&1 - docker run -d --name bench-redis --network host redis:7-alpine >/dev/null 2>&1 - sleep 2 - play Clients/Redis PLAYGROUND_REACTORS=$R PLAYGROUND_PORT=18085 \ - && note "redis" "${R}r" "$(wrk_h1 http://127.0.0.1:18085/) req/s" - stop_play - docker rm -f bench-redis >/dev/null 2>&1 - docker rm -f bench-pg >/dev/null 2>&1 docker run -d --name bench-pg --network host \ -e POSTGRES_USER=bench -e POSTGRES_PASSWORD=bench -e POSTGRES_DB=bench \ @@ -146,7 +138,6 @@ if command -v docker >/dev/null 2>&1 && docker info >/dev/null 2>&1; then stop_play docker rm -f bench-pg >/dev/null 2>&1 else - note "redis" "${R}r" "SKIP (docker unavailable)" note "pg" "${R}r" "SKIP (docker unavailable)" fi diff --git a/bench/samples.tsv b/bench/samples.tsv index d08c414d..a856c8cb 100644 --- a/bench/samples.tsv +++ b/bench/samples.tsv @@ -71,7 +71,6 @@ Quic/Raw echo 8443 / - - # upstream clients and static files Clients/File h1 8080 /f.bin - PLAYGROUND_DIR=/tmp/ioxide-bench-assets Clients/Pg h1 8080 / - - -Clients/Redis h1 8080 / - - Clients/Http h1 8090 / Tcp/Raw - Clients/Https h1 8080 / Tls/OpenSsl PLAYGROUND_UPSTREAM_CA=/tmp/ioxide-playground-quic/quic.crt Clients/Timer h1 8080 / - PLAYGROUND_DELAY_MS=0 diff --git a/docs/assets/style.css b/docs/assets/style.css index 492e960e..3146ac61 100644 --- a/docs/assets/style.css +++ b/docs/assets/style.css @@ -261,7 +261,6 @@ nav.top .links a.gh svg { display: block; } #tab-h3:checked ~ .ex-menu label[for="tab-h3"], #tab-http:checked ~ .ex-menu label[for="tab-http"], #tab-pg:checked ~ .ex-menu label[for="tab-pg"], -#tab-redis:checked ~ .ex-menu label[for="tab-redis"], #tab-files:checked ~ .ex-menu label[for="tab-files"], #tab-tls:checked ~ .ex-menu label[for="tab-tls"], #tab-tlshybrid:checked ~ .ex-menu label[for="tab-tlshybrid"], @@ -408,7 +407,6 @@ nav.top .links a.gh svg { display: block; } #tab-h3:checked ~ .pane-h3 { display: block; } #tab-http:checked ~ .pane-http { display: block; } #tab-pg:checked ~ .pane-pg { display: block; } -#tab-redis:checked ~ .pane-redis { display: block; } #tab-files:checked ~ .pane-files { display: block; } #tab-tls:checked ~ .pane-tls { display: block; } #tab-tlshybrid:checked ~ .pane-tlshybrid { display: block; } @@ -560,7 +558,6 @@ nav.top .links a.gh svg { display: block; } #tab-h3:checked ~ .ex-menu label[for="tab-h3"], #tab-http:checked ~ .ex-menu label[for="tab-http"], #tab-pg:checked ~ .ex-menu label[for="tab-pg"], - #tab-redis:checked ~ .ex-menu label[for="tab-redis"], #tab-files:checked ~ .ex-menu label[for="tab-files"], #tab-tls:checked ~ .ex-menu label[for="tab-tls"], #tab-tlshybrid:checked ~ .ex-menu label[for="tab-tlshybrid"], diff --git a/docs/how-ioxide-does-h3.html b/docs/how-ioxide-does-h3.html index 5d375733..836a6fcc 100644 --- a/docs/how-ioxide-does-h3.html +++ b/docs/how-ioxide-does-h3.html @@ -47,7 +47,6 @@
Guide
Introduction Postgres - Redis TLS Static files Multi-port @@ -58,7 +57,6 @@
Packages
ioxide core ioxide.pg - ioxide.redis ioxide tls ioxide.file diff --git a/docs/index.html b/docs/index.html index 1e3652a2..d0774ee7 100644 --- a/docs/index.html +++ b/docs/index.html @@ -63,7 +63,6 @@ - @@ -166,7 +165,6 @@
clients - @@ -201,7 +199,7 @@

Nothing scheduled

Both directions

-

A Postgres query, a Redis command, an outbound HTTP call or a file read rides the +

A Postgres query, an outbound HTTP call or a file read rides the same ring that accepted the request.

@@ -3171,8 +3169,8 @@

HTTP/3 · buffered (nghttp3)

{ // Dispatch waited for end-of-stream, so the body is ALREADY here: request.Body is // complete and request.Body.Length is just a property read. No BodyReader, no pacing. - // This overload is synchronous, but the awaiting one exists too - a PgPool query or a - // Redis command resumes inline on this reactor, so you can await it right here. + // This overload is synchronous, but the awaiting one exists too - a PgPool query + // resumes inline on this reactor, so you can await it right here. _ = request.Body.Length; // One response object, reused for every request: zero allocations on this path. To @@ -7804,229 +7802,6 @@

Postgres

}

One pool per reactor, opened on the reactor thread, so a query rides the same ring that accepted the request and resumes the handler inline. The host must be an IPv4 literal - a DNS lookup would block the reactor.

-
-
-

Redis

- ioxide + ioxide.redis -
-
// dotnet add package ioxide
-// dotnet add package ioxide.redis
-//   curl http://127.0.0.1:8080/
-
-using System.Text;
-using ioxide;
-using ioxide.redis;
-using ioxide.utils;
-
-// ── Knobs ────────────────────────────────────────────────────────────────────────────────────
-
-ushort port     = 8080;                        // http://127.0.0.1:8080/
-int    reactors = Environment.ProcessorCount;  // one ring per reactor, one reactor per core
-
-
-// The Redis this talks to. IPv4 literal for the same reason as everywhere else - a DNS lookup
-// would block the reactor - and the pool is per reactor. A null password means no AUTH.
-string  redisHost     = "127.0.0.1";
-ushort  redisPort     = 6379;
-string? redisPassword = null;
-int     redisPoolSize = 4;
-// ─────────────────────────────────────────────────────────────────────────────────────────────
-
-var config = new ServerConfig
-{
-    ReactorCount   = reactors,  // io_uring rings/threads - one per core
-    RingEntries    = 8192,                                                        // SQ/CQ depth per ring
-    DualStack      = false,                                                       // true = one IPv6 socket also accepts IPv4-mapped
-    RecvBufferSize = 32 * 1024,                                                   // bytes per shared recv buffer
-    RecvSlots      = 4096,                                                        // shared recv buffer-ring depth
-    Incremental    = null,                                                        // per-connection recv rings (6.12+) - see Tcp/Incremental
-    Udp            = null,                                                        // no raw UDP sockets (TCP-only server)
-    Quic           = null,                                                        // no QUIC transport - see Http3/* and Quic/Alpn
-    Tcp = new TcpOptions
-    {
-        Port             = port,
-        ExtraPorts       = [],                                                    // extra listener ports (one handler, several doors)
-        ListenBacklog    = 1024,                                                  // accept-queue depth per SO_REUSEPORT listener
-        WriteSlabSize    = 16 * 1024,                                             // per-connection write buffer before overflow kicks in
-        PoolMax          = 1024,                                                  // pooled connection objects kept per reactor
-        WriteOverflow    = WriteOverflowStrategy.Grow,                            // Grow = realloc one slab; Segmented = chain + vectored SENDMSG
-        ZeroCopySend     = false,                                                 // SEND_ZC: kernel copies less, wins on large writes
-        RecvQueueEntries = 4096,                                                  // per-connection recv completion queue depth
-        ReadTimeoutMs    = 60_000,                                                // close a connection whose read waits this long on a silent peer; 0 = off
-        SendTimeoutMs    = 60_000,                                                // close a connection whose flush takes this long to go out; 0 = off
-    },
-};
-
-var redisOptions = new RedisOptions
-{
-    Host             = redisHost,  // IPv4 literal - resolve names up front, DNS blocks the reactor
-    Port             = redisPort,        // Redis wire port
-    Password         = redisPassword,     // AUTH password; null = no auth
-    User             = null,                                           // ACL username (Redis 6+); null = default user
-    Database         = 0,                                              // logical DB index to SELECT on connect
-    PoolSize         = redisPoolSize,            // per reactor, not global
-    CommandTimeoutMs = 30_000,                                         // oldest in-flight command past this -> torn down; 0 disables
-};
-
-var threads = new Thread[config.ReactorCount];
-
-for (int i = 0; i < threads.Length; i++)
-{
-    var reactor = new Reactor(i, config);
-
-    // The pool lives on the reactor: its connections ride this reactor's ring, so a query is
-    // an io_uring send + recv with the continuation resuming right here.
-    reactor.OnStart = r =>
-    {
-        RedisPool pool = RedisPool.Start(r, redisOptions);
-        _ = SeedAsync(pool);   // the "/" route reads this key; seed it so the demo answers
-    };
-
-    reactor.TcpHandle = async (r, conn) =>
-    {
-        RedisPool pool = r.GetService<RedisPool>();
-
-        try
-        {
-            while (true)
-            {
-                RecvSnapshot snapshot = await conn.ReadAsync();
-                string path = ReadPath(conn, snapshot);
-
-                try
-                {
-                    string body = path switch
-                    {
-                        // Cache-aside: try the key; on a miss compute, SET with a TTL, return.
-                        var p when p.StartsWith("/cache/", StringComparison.Ordinal)
-                            => await CacheAside(pool, p["/cache/".Length..]),
-
-                        // One RESP reply type per route, through the generic ExecuteAsync.
-                        "/incr"     => $"counter = {(await pool.ExecuteAsync("INCR", "play:counter")).AsInteger()}",
-                        "/hash"     => await Hash(pool),
-                        "/list"     => await List(pool),
-
-                        // Several commands, one round trip, replies in order.
-                        "/pipeline" => await Pipeline(pool),
-
-                        // The hot path: a single GET per request. wrk this one.
-                        _           => $"redis = {await pool.GetAsync("play:bench") ?? "(nil)"}",
-                    };
-
-                    WriteText(conn, "200 OK", body);
-                }
-                catch (RedisException e)
-                {
-                    // A server error is an exception, not a broken socket: the connection has
-                    // resynced and goes back to the pool usable.
-                    WriteText(conn, "500 Internal Server Error", e.Message);
-                }
-
-                await conn.FlushAsync();
-
-                if (snapshot.IsClosed) return;
-                conn.ResetRead();
-            }
-        }
-        finally
-        {
-            conn.DecRef();
-        }
-    };
-
-    threads[i] = new Thread(reactor.Run) { Name = $"reactor-{i}" };
-    threads[i].Start();
-}
-
-Console.WriteLine($"[redis] {config.ReactorCount} reactors on :{config.Tcp!.Port} "
-                + $"-> {redisOptions.Host}:{redisOptions.Port}, {redisOptions.PoolSize} connections each");
-
-foreach (Thread thread in threads)
-{
-    thread.Join();
-}
-
-static async Task SeedAsync(RedisPool pool)
-{
-    try
-    {
-        await pool.ExecuteAsync("SET", "play:bench", "42");
-    }
-    catch (Exception e) { Console.Error.WriteLine($"[redis] seed failed: {e.Message}"); }
-}
-
-static async Task<string> CacheAside(RedisPool pool, string key)
-{
-    key = "play:cache:" + key;
-    string? cached = await pool.GetAsync(key);
-    if (cached != null)
-    {
-        return $"hit {key} = {cached}";
-    }
-
-    string value = $"value-{key.Length}";           // stands in for the expensive computation
-    await pool.SetExAsync(key, value, seconds: 60);
-    return $"miss {key} = {value} (cached 60s)";
-}
-
-// RESP array of bulk strings: field, value, field, value, ...
-static async Task<string> Hash(RedisPool pool)
-{
-    await pool.ExecuteAsync("HSET", "play:hash", "lang", "csharp", "engine", "io_uring");
-    RespValue all = await pool.ExecuteAsync("HGETALL", "play:hash");
-    return "hash = [" + string.Join(", ", all.Items.Select(v => v.AsString())) + "]";
-}
-
-static async Task<string> List(RedisPool pool)
-{
-    await pool.ExecuteAsync("DEL", "play:list");
-    await pool.ExecuteAsync("RPUSH", "play:list", "a", "b", "c");
-    RespValue items = await pool.ExecuteAsync("LRANGE", "play:list", "0", "-1");
-    return "list = [" + string.Join(", ", items.Items.Select(v => v.AsString())) + "]";
-}
-
-static async Task<string> Pipeline(RedisPool pool)
-{
-    RespValue[] replies = await pool.PipelineAsync(
-        new RedisCommand("SET", "play:pipe", "1"),
-        new RedisCommand("INCR", "play:pipe"),
-        new RedisCommand("GET", "play:pipe"));
-    return $"set={replies[0].AsString()} incr={replies[1].AsInteger()} get={replies[2].AsString()}";
-}
-
-// "GET /cache/x HTTP/1.1" -> "/cache/x", draining the recv on the way past.
-static string ReadPath(TcpConnection conn, RecvSnapshot snapshot)
-{
-    string path = "/";
-    while (conn.TryGetItem(snapshot, out SpscRecvRing.Item item))
-    {
-        if (item.HasBuffer)
-        {
-            ReadOnlySpan<byte> request = item.AsSpan();
-            int firstSpace = request.IndexOf((byte)' ');
-            if (firstSpace >= 0)
-            {
-                ReadOnlySpan<byte> rest = request[(firstSpace + 1)..];
-                int secondSpace = rest.IndexOf((byte)' ');
-                if (secondSpace > 0)
-                {
-                    ReadOnlySpan<byte> target = rest[..secondSpace];
-                    int query = target.IndexOf((byte)'?');
-                    path = Encoding.ASCII.GetString(query >= 0 ? target[..query] : target);
-                }
-            }
-            conn.ReturnBuffer(in item);
-        }
-    }
-
-    return path;
-}
-
-static void WriteText(TcpConnection conn, string status, string body)
-    => conn.Write(Encoding.ASCII.GetBytes(
-        $"HTTP/1.1 {status}\r\nContent-Type: text/plain\r\nContent-Length: {body.Length}\r\n\r\n{body}"));
-

RESP2 with pipelining, one pool per reactor. Same shape as - the client is ring-native, so the round trip never leaves the core the request landed on.

-

Client · static files

diff --git a/docs/learn/architecture.html b/docs/learn/architecture.html index 83ac1cd9..189f02d4 100644 --- a/docs/learn/architecture.html +++ b/docs/learn/architecture.html @@ -23,7 +23,6 @@
Guide
Introduction Postgres - Redis TLS Static files Multi-port @@ -34,7 +33,6 @@
Packages
ioxide core ioxide.pg - ioxide.redis ioxide tls ioxide.file diff --git a/docs/learn/clients.html b/docs/learn/clients.html index 4c4cdf1c..473b6af1 100644 --- a/docs/learn/clients.html +++ b/docs/learn/clients.html @@ -23,7 +23,6 @@
Guide
Introduction Postgres - Redis TLS Static files Multi-port @@ -34,7 +33,6 @@
Packages
ioxide core ioxide.pg - ioxide.redis ioxide tls ioxide.file diff --git a/docs/learn/dev-core.html b/docs/learn/dev-core.html index 8838f68c..181ab148 100644 --- a/docs/learn/dev-core.html +++ b/docs/learn/dev-core.html @@ -23,7 +23,6 @@
Guide
Introduction Postgres - Redis TLS Static files Multi-port @@ -34,7 +33,6 @@
Packages
ioxide core ioxide.pg - ioxide.redis ioxide tls ioxide.file @@ -196,7 +194,7 @@

The periodic timer and tickers

A single-shot IORING_OP_TIMEOUT (a native __kernel_timespec the reactor owns) fires every ~250 ms, re-armed each time. On fire, OnTimerTick runs every callback registered via AddTicker(Action) - on the reactor thread - then - re-arms. This is how the pg/redis pools sweep per-command timeouts and replenish connections + re-arms. This is how the pg pools sweep per-command timeouts and replenish connections without a thread-per-command or any cross-thread timer: one timeout SQE per reactor, negligible cost, and the tickers see reactor-owned state directly.

Reactor.cs · a single-shot IORING_OP_TIMEOUT, re-armed after every tick

@@ -213,7 +211,7 @@

The periodic timer and tickers

{ for (int i = 0; i < _tickers.Count; i++) { - _tickers[i](); // pg/redis pools sweep timeouts + replenish here + _tickers[i](); // pg pools sweep timeouts + replenish here } ArmTimer(); // single-shot: re-arm for the next interval } @@ -343,7 +341,7 @@

Mpsc · Utils/Mpsc.cs

false sharing. Used for the buffer-return and flush hand-off queues.

The ring-native client plumbing

-

How Postgres, Redis, and file I/O run on the same ring as the server.

+

How Postgres and file I/O run on the same ring as the server.

- +
diff --git a/docs/learn/dev-redis.html b/docs/learn/dev-redis.html deleted file mode 100644 index 552e36d5..00000000 --- a/docs/learn/dev-redis.html +++ /dev/null @@ -1,249 +0,0 @@ - - - - - - Redis internals - ioxide - - - - - - - - -
- - -
-

Redis internals

-

ioxide.redis is a lean RESP2 client on the same ring as everything - else. Its design mirrors the Postgres driver's pipelining and pooling; - the value/parse layer borrows the allocation discipline of StackExchange.Redis. Every class - below.

- -

RedisConnection · RedisConnection.cs (+ .Commands)

-

One Redis connection over a RingSocket. ConnectAsync connects, then - AUTH (ACL-aware) and SELECT inline before returning a ready connection. The generic surface is - ExecuteAsync(command, args); typed helpers (GET/SET/INCR/HSET/LRANGE/ZADD/…) layer on - top in RedisConnection.Commands.cs.

- -

Pipelining

-

Identical shape to the pg driver: SubmitCore appends the encoded command into a - fixed 512 KB native send buffer (overflow throws, never realloc - a pipelined send - holds its address), enqueues a Pending (IValueTaskSource<RespValue>), - and kicks the sender/reader loops. SenderLoopAsync drains the staged bytes; - ReaderLoopAsync pulls one reply per in-flight entry and completes the dequeued - Pending FIFO (Redis guarantees reply order on one connection). A documented invariant: - there is no await between the sender loop's final drain check and clearing - _sending, and completions resume inline, so no command can be stranded - it depends on - RingOpSource keeping synchronous continuations. PipelineAsync sends several - commands back to back and awaits all replies - one round trip.

-

RedisConnection.cs · staging a command into the fixed send buffer (overflow throws, never realloc)

-
private unsafe void AppendCommand(ReadOnlySpan<byte> nameToken, RedisArg[] args)
-{
-    int size = RespProtocol.CommandSize(nameToken, args);
-    if (size > _sendCapacity)
-    {
-        throw new RedisException($"command exceeds send buffer ({_sendCapacity} bytes)");
-    }
-    if (_sendEnd + size > _sendCapacity)
-    {
-        throw new RedisException("pipelined send buffer full");   // an in-flight send holds _send's address
-    }
-    int written = RespProtocol.WriteCommand(
-        new Span<byte>((void*)(_send + _sendEnd), _sendCapacity - _sendEnd), nameToken, args);
-    _sendEnd += written;
-}
- -

Command encoding

-

Command names are a tiny fixed set, so their full pre-framed RESP token - ($3\r\nGET\r\n) is computed once and cached in a static - ConcurrentDictionary<string, byte[]> - the hot path memcpy's the whole token in - one shot, no per-call encoding or framing allocation.

-

RedisConnection.cs · the pre-framed command token is computed once and memoized

-
private static readonly ConcurrentDictionary<string, byte[]> CommandCache = new();
-
-private static byte[] EncodeCommand(string command) =>
-    CommandCache.GetOrAdd(command, static c => RespProtocol.FrameName(Encoding.ASCII.GetBytes(c)));
- -

Receiving with an examined cursor

-

ReceiveReplyAsync reads into a growable native recv buffer (up to 64 MB), - compacting the unparsed tail before growth. TryParse reports, on an incomplete reply, - how many bytes are needed before a re-parse can make progress (the examined cursor), so a large - reply arriving over many recvs isn't re-scanned from the start each time.

-

RespProtocol.cs · the examined cursor - on an incomplete reply, report how many more bytes are needed

-
public static bool TryParse(ReadOnlySpan<byte> buffer, out RespValue value, out int consumed, out int needed)
-{
-    int pos = 0;
-    needed = 0;
-    if (TryParseAt(buffer, ref pos, out value, ref needed))
-    {
-        consumed = pos;
-        return true;
-    }
-    consumed = 0;
-    return false;   // 'needed' lets the caller wait for that many bytes before re-parsing
-}
- -

Timeouts and breakage

-

Each Pending stamps EnqueuedAtMs; the pool's ticker calls - CheckTimeout to tear down a connection whose oldest in-flight command has aged past - CommandTimeoutMs (diagnostic RedisException; closing the fd cancels the - stuck recv). A transport failure marks IsBroken and the pool evicts.

- -

RespProtocol · RespProtocol.cs

-

The RESP2 codec, a pure static. CommandSize/WriteCommand write a - command as a RESP array of bulk strings (the pre-framed name token + each arg). FrameName - builds the cached name token. TryParse recursively parses one reply - - + simple string, - error, : integer, $ bulk - (with $-1 null), * array (with *-1 null), and _ - (RESP3 null, tolerated) - returning false with a needed-bytes count when the reply - isn't fully buffered. ParseLong is strict: optional sign then ASCII digits, with - overflow-checked accumulation, so a malformed length breaks the connection cleanly instead of - computing a garbage size and wedging. Bulk/array lengths are range-guarded.

-

RespProtocol.cs · WriteCommand - a RESP array of bulk strings, name token memcpy'd in

-
public static int WriteCommand(Span<byte> dst, ReadOnlySpan<byte> nameToken, ReadOnlySpan<RedisArg> args)
-{
-    int p = 0;
-    dst[p++] = (byte)'*';
-    p += WriteInt(dst[p..], args.Length + 1);
-    dst[p++] = (byte)'\r';
-    dst[p++] = (byte)'\n';
-
-    nameToken.CopyTo(dst[p..]);   // pre-framed $len\r\nNAME\r\n - one memcpy, no per-call framing
-    p += nameToken.Length;
-
-    foreach (RedisArg arg in args)
-    {
-        p += WriteBulk(dst[p..], arg.Bytes);
-    }
-    return p;
-}
-

RespProtocol.cs · ParseLong - strict ASCII digits, overflow-checked, malformed input breaks the connection

-
private static long ParseLong(ReadOnlySpan<byte> s)
-{
-    bool neg = s.Length > 0 && s[0] == (byte)'-';
-    if (neg) s = s[1..];
-    if (s.IsEmpty)
-    {
-        throw new RedisException("malformed RESP integer");
-    }
-    long n = 0;
-    foreach (byte c in s)
-    {
-        if (c < (byte)'0' || c > (byte)'9')
-        {
-            throw new RedisException($"malformed RESP integer (byte 0x{c:x2})");
-        }
-        n = checked(n * 10 + (c - '0'));
-    }
-    return neg ? -n : n;
-}
- -

RespValue · RespValue.cs

-

A parsed reply, as a value type (struct), so parsing allocates no per-node object - - integers, nulls, and booleans are entirely allocation-free; bulk/simple strings and arrays carry - the bytes (or element array) they materialized during parse, which outlive the recv buffer. - Accessors convert on demand: AsString, Bytes, AsInteger - (accepts integer replies and numeric strings, length-checked), AsDouble (accepts - integers, numeric strings, and the non-finite tokens inf/-inf/nan - that Utf8Parser won't parse - e.g. a ZSCORE of +inf), AsBool, - Items, IsNull, IsError.

-

RespValue.cs · a readonly struct - integers/nulls/bools cost zero allocation

-
public readonly struct RespValue
-{
-    public RespKind Kind { get; }
-    public long Integer { get; }
-    private readonly byte[]? _bytes;
-    private readonly RespValue[]? _items;
-
-    private RespValue(RespKind kind, long integer, byte[]? bytes, RespValue[]? items)
-    {
-        Kind = kind;
-        Integer = integer;
-        _bytes = bytes;     // bulk/simple strings materialize here at parse, outliving the recv buffer
-        _items = items;     // arrays only
-    }
-    // ...accessors below convert on demand
-}
-

RespValue.cs · AsDouble - falls back to the non-finite tokens Utf8Parser won't take

-
public double AsDouble()
-{
-    if (Kind == RespKind.Integer)
-    {
-        return Integer;
-    }
-    if (_bytes is not null && (Kind == RespKind.BulkString || Kind == RespKind.SimpleString))
-    {
-        if (Utf8Parser.TryParse(_bytes, out double d, out int n) && n == _bytes.Length)
-        {
-            return d;
-        }
-        ReadOnlySpan<byte> b = _bytes;
-        if (b.SequenceEqual("inf"u8) || b.SequenceEqual("+inf"u8)) return double.PositiveInfinity;
-        if (b.SequenceEqual("-inf"u8)) return double.NegativeInfinity;
-        if (b.SequenceEqual("nan"u8)) return double.NaN;   // e.g. a ZSCORE of +inf
-    }
-    throw new RedisException($"reply ({Kind}) is not a double");
-}
- -

RedisPool · RedisPool.cs

-

N connections per reactor with round-robin and pipelining, mirroring PgPool: - bounded fail-fast opens (WaitForConnectionAsync throws after - MaxConnectAttempts instead of hanging), a reactor ticker that sweeps command timeouts - and replenishes with jittered backoff, and the cache-aside surface (GetAsync, - SetExAsync, DelAsync) plus ExecuteAsync and - PipelineAsync (the latter picks one connection so the batch is a real single round - trip). One pool per reactor, reactor-thread-only.

- -

The value types

-
    -
  • RedisArg (RedisArg.cs) - a value-type command argument with implicit - conversions from string/byte[]/int/long/double/bool, rendered to bytes for the wire (double via - round-trip "R", bool as "1"/"0").
    • -
  • RedisCommand (in RedisConnection.cs) - a (name, args) pair for - PipelineAsync; the name is pre-framed to its RESP token at construction.
    • -
  • RedisOptions (RedisOptions.cs) - host/port, password + optional ACL user, - database index, PoolSize, CommandTimeoutMs.
    • -
  • RedisException (RedisException.cs) - a -ERR reply or a - transport failure.
    • -
- -

Note: RESP2 only (no HELLO), and the FIFO - one-reply-per-request model has no out-of-band/push path - pub/sub on a RESP2 connection would - need a dedicated connection, as StackExchange.Redis does.

- - -
-
- - - - - - diff --git a/docs/learn/dev-tls.html b/docs/learn/dev-tls.html index 92d1c4fd..dd121094 100644 --- a/docs/learn/dev-tls.html +++ b/docs/learn/dev-tls.html @@ -23,7 +23,6 @@
Guide
Introduction Postgres - Redis TLS Static files Multi-port @@ -34,7 +33,6 @@
Packages
ioxide core ioxide.pg - ioxide.redis ioxide tls ioxide.file diff --git a/docs/learn/files.html b/docs/learn/files.html index 695571e6..7e697ce5 100644 --- a/docs/learn/files.html +++ b/docs/learn/files.html @@ -23,7 +23,6 @@
Guide
Introduction Postgres - Redis TLS Static files Multi-port @@ -34,7 +33,6 @@
Packages
ioxide core ioxide.pg - ioxide.redis ioxide tls ioxide.file diff --git a/docs/learn/multiport.html b/docs/learn/multiport.html index 5a1fb6cb..5bb7c34d 100644 --- a/docs/learn/multiport.html +++ b/docs/learn/multiport.html @@ -23,7 +23,6 @@
Guide
Introduction Postgres - Redis TLS Static files Multi-port @@ -34,7 +33,6 @@
Packages
ioxide core ioxide.pg - ioxide.redis ioxide tls ioxide.file diff --git a/docs/learn/overview.html b/docs/learn/overview.html index 776b9f20..d66b6471 100644 --- a/docs/learn/overview.html +++ b/docs/learn/overview.html @@ -23,7 +23,6 @@
Guide
Introduction Postgres - Redis TLS Static files Multi-port @@ -34,7 +33,6 @@
Packages
ioxide core ioxide.pg - ioxide.redis ioxide tls ioxide.file @@ -108,8 +106,7 @@

Plugging in backends

engine never names a client type, so adding one never touches the reactor:

- + diff --git a/docs/learn/redis.html b/docs/learn/redis.html deleted file mode 100644 index 78a2dd6b..00000000 --- a/docs/learn/redis.html +++ /dev/null @@ -1,144 +0,0 @@ - - - - - - Redis - ioxide - - - - - - - - -
- - -
-

Redis over the ring

-

ioxide.redis is a Redis client where connect, every command, and - pipelines are io_uring ops on the host reactor's ring, resumed inline. Full RESP2, a generic - command API that runs anything, typed helpers across every data type, and pipelining - pooled - per reactor, shared-nothing.

- -
Using the client. How it works inside - pipelining, the RESP codec, the - allocation-free value struct - is in ioxide.redis internals.
- -

The pool

-

Open connections per reactor from OnStart and register the pool as a service; - handlers fetch it and rent per operation. One connection carries one in-flight command (or - pipeline) at a time, so the pool size is the reactor's Redis concurrency.

-
reactor.OnStart = r => RedisPool.Start(r, new RedisOptions
-{
-    Host = "127.0.0.1", Port = 6379, PoolSize = 8,
-    // Password, User (ACL), Database (SELECT on connect) are all optional
-});
- -

Two API levels

-

Generic - ExecuteAsync runs any command and returns a - RespValue, the full RESP2 reply taxonomy. Use it for anything the typed helpers do - not cover:

-
var redis = reactor.GetService<RedisPool>();
-
-RespValue pong = await redis.ExecuteAsync("PING");
-RespValue n    = await redis.ExecuteAsync("INCRBY", "counter", 5);
-RespValue info = await redis.ExecuteAsync("COMMAND", "COUNT");
-

Typed - convenience methods over the generic call, covering strings, keys, hashes, - lists, sets, sorted sets, pub/sub, and scripting. Rent a connection for a sequence, or use the - pool's one-shot shortcuts:

-
var c = await redis.RentAsync();
-try
-{
-    await c.SetExAsync("session:42", token, seconds: 300);
-    string? v   = await c.GetAsync("session:42");
-    long count  = await c.IncrAsync("hits");
-    var fields  = await c.HGetAllAsync("user:42");
-    var range   = await c.LRangeAsync("queue", 0, -1);
-    double? sc  = await c.ZScoreAsync("board", "alice");
-}
-finally { redis.Return(c); }
- -

The reply type

-

RespValue represents any RESP2 reply - null, simple string, error, integer, bulk - string, or a (possibly nested) array - and converts on demand. Errors are surfaced two ways: - ExecuteAsync throws a RedisException on a top-level error reply, while - array elements that are errors stay inspectable (for MULTI/EXEC).

-
RespValue r = await redis.ExecuteAsync("HGETALL", "user:42");
-foreach (RespValue field in r.Items)        // arrays
-    Console.WriteLine(field.AsString());
-
-long size = (await redis.ExecuteAsync("DBSIZE")).AsInteger();   // integers
-bool gone = (await redis.ExecuteAsync("GET", "missing")).IsNull;
- -

Pipelining

-

Send several commands back to back and read all replies in one round trip. Errors are - returned per-command rather than thrown, so you can inspect each outcome.

-
RespValue[] replies = await c.PipelineAsync(
-    new RedisCommand("SET", "k", "10"),
-    new RedisCommand("INCR", "k"),
-    new RedisCommand("GET", "k"));
-
-long current = replies[2].AsInteger();   // 11
- -

Cache-aside

-

The everyday pattern - check the cache, fall back to the source, populate with a TTL, and - invalidate on write. The pool exposes GetAsync / SetExAsync / - DelAsync directly so a read is one rented round trip:

-
string key = $"item:{id}";
-string? cached = await redis.GetAsync(key);
-if (cached is null)
-{
-    cached = await LoadFromDatabase(id);          // ioxide.pg, also on the ring
-    await redis.SetExAsync(key, cached, seconds: 1);
-}
-// ... on update:
-await redis.DelAsync($"item:{id}");
-

Shared-nothing, but consistent. Each reactor owns its own Redis connections, so there - is no shared managed state and no lock on the hot path. Because Redis itself is shared across - reactors, a write that invalidates a key is seen by every reactor's next read - cache-aside - stays correct even though connections land on different reactors. This is the reason a shared - cache (Redis) fits the model better than a per-reactor in-process cache, which would serve stale - reads across reactors.

- -

Connection

-

RedisOptions: Host (IPv4 literal - resolve names up front, DNS - would block the reactor), Port, Password with optional - User for ACL AUTH, Database to SELECT on - connect, and PoolSize per reactor. Total server-side connections are - PoolSize × ReactorCount. Broken connections are discarded on return and - replaced in the background, exactly like the Postgres pool.

- - -
-
- - - - - - diff --git a/docs/learn/tls.html b/docs/learn/tls.html index ee5b6059..1f74d47a 100644 --- a/docs/learn/tls.html +++ b/docs/learn/tls.html @@ -23,7 +23,6 @@
Guide
Introduction Postgres - Redis TLS Static files Multi-port @@ -34,7 +33,6 @@
Packages
ioxide core ioxide.pg - ioxide.redis ioxide tls ioxide.file diff --git a/ioxide.slnx b/ioxide.slnx index 676c5768..609fc517 100644 --- a/ioxide.slnx +++ b/ioxide.slnx @@ -17,7 +17,6 @@ - @@ -102,7 +101,6 @@ - diff --git a/scripts/gen-docs-panes.py b/scripts/gen-docs-panes.py index fd1e3e5e..9b353acd 100644 --- a/scripts/gen-docs-panes.py +++ b/scripts/gen-docs-panes.py @@ -501,12 +501,6 @@ "response - so the hop it costs is measurable rather than asserted; " " is that hop on its " "own."), - "redis": ( - "Clients/Redis", "Redis", "ioxide + ioxide.redis", - ["curl http://127.0.0.1:8080/"], - "RESP2 with pipelining, one pool per reactor. Same shape as " - " - the client is ring-native, so " - "the round trip never leaves the core the request landed on."), "qclient": ( "Clients/Quic", "QUIC · client", "ioxide + ioxide.ngtcp2", ["dotnet run -c Release --project Playground/Quic/Raw # something to talk to", diff --git a/src/clients/ioxide.redis/README.md b/src/clients/ioxide.redis/README.md deleted file mode 100644 index d3d633aa..00000000 --- a/src/clients/ioxide.redis/README.md +++ /dev/null @@ -1,40 +0,0 @@ -# ioxide.redis - -Redis client for the [ioxide](https://github.com/MDA2AV/ioxide) io_uring runtime. Pooled -ring-native connections per reactor, full RESP2 protocol, a generic command API plus typed helpers -across every data type, and pipelining - all with inline completion resume. - -```csharp -reactor.OnStart = r => RedisPool.Start(r, new RedisOptions -{ - Host = "127.0.0.1", Port = 6379, PoolSize = 8, -}); - -// in a handler - cache-aside: -var redis = reactor.GetService(); -string? cached = await redis.GetAsync($"item:{id}"); -if (cached is null) -{ - cached = await LoadFromDb(id); - await redis.SetExAsync($"item:{id}", cached, seconds: 1); // 1s TTL -} -``` - -## Surface - -- **Generic**: `ExecuteAsync(command, args...)` runs *any* Redis command and returns a `RespValue` - (the full RESP2 taxonomy - null, simple/bulk string, integer, error, nested array). -- **Typed helpers**: strings, keys, hashes, lists, sets, sorted sets, pub/sub, scripting - layered - over `ExecuteAsync`, so anything not wrapped still works through it. -- **Pipelining**: `PipelineAsync(cmd1, cmd2, ...)` sends a batch and reads all replies in one round - trip. -- **Pool**: rent for a multi-command sequence (MULTI/EXEC, pipelines) or use the pool's - `GetAsync` / `SetExAsync` / `DelAsync` / `ExecuteAsync` shortcuts for one-offs. - -## Connection - -`RedisOptions`: `Host` (IPv4 literal), `Port`, `Password` (+ optional `User` for ACL AUTH), -`Database` (SELECT on connect), `PoolSize` (connections per reactor). One connection carries one -in-flight command or pipeline; total server connections = `PoolSize × ReactorCount`. - -Requires `ioxide`. RESP2 (Redis 6+/7). MIT. diff --git a/src/clients/ioxide.redis/RedisArg.cs b/src/clients/ioxide.redis/RedisArg.cs deleted file mode 100644 index 2e2e947e..00000000 --- a/src/clients/ioxide.redis/RedisArg.cs +++ /dev/null @@ -1,22 +0,0 @@ -using System.Globalization; -using System.Text; - -namespace ioxide.redis; - -/// -/// A single command argument. Implicit conversions cover the common types; everything is sent as a -/// RESP bulk string, so a value's type only matters for how it's rendered to bytes. -/// -public readonly struct RedisArg -{ - internal readonly byte[] Bytes; - - private RedisArg(byte[] bytes) => Bytes = bytes; - - public static implicit operator RedisArg(string value) => new(Encoding.UTF8.GetBytes(value)); - public static implicit operator RedisArg(byte[] value) => new(value); - public static implicit operator RedisArg(int value) => new(Encoding.ASCII.GetBytes(value.ToString(CultureInfo.InvariantCulture))); - public static implicit operator RedisArg(long value) => new(Encoding.ASCII.GetBytes(value.ToString(CultureInfo.InvariantCulture))); - public static implicit operator RedisArg(double value) => new(Encoding.ASCII.GetBytes(value.ToString("R", CultureInfo.InvariantCulture))); - public static implicit operator RedisArg(bool value) => new(value ? "1"u8.ToArray() : "0"u8.ToArray()); -} diff --git a/src/clients/ioxide.redis/RedisConnection.Commands.cs b/src/clients/ioxide.redis/RedisConnection.Commands.cs deleted file mode 100644 index 4618e0dc..00000000 --- a/src/clients/ioxide.redis/RedisConnection.Commands.cs +++ /dev/null @@ -1,180 +0,0 @@ -namespace ioxide.redis; - -/// -/// Typed convenience commands over . Not exhaustive - -/// anything missing runs through ExecuteAsync directly - but it covers the everyday surface of each -/// data type. -/// -public sealed partial class RedisConnection -{ - // -- strings ---------------------------------------------------------- - - public async ValueTask GetAsync(string key) => - (await ExecuteAsync("GET", key)).AsString(); - - public async ValueTask GetBytesAsync(string key) => - (await ExecuteAsync("GET", key)).Bytes; - - public async ValueTask SetAsync(string key, RedisArg value) => - await ExecuteAsync("SET", key, value); - - /// SET with an expiry in seconds (the cache-aside populate path). - public async ValueTask SetExAsync(string key, RedisArg value, int seconds) => - await ExecuteAsync("SET", key, value, "EX", seconds); - - /// SET only if absent; returns true if the key was set. - public async ValueTask SetNxAsync(string key, RedisArg value) => - !(await ExecuteAsync("SET", key, value, "NX")).IsNull; - - public async ValueTask GetSetAsync(string key, RedisArg value) => - (await ExecuteAsync("GETSET", key, value)).AsString(); - - public async ValueTask GetDelAsync(string key) => - (await ExecuteAsync("GETDEL", key)).AsString(); - - public async ValueTask IncrAsync(string key) => (await ExecuteAsync("INCR", key)).AsInteger(); - public async ValueTask DecrAsync(string key) => (await ExecuteAsync("DECR", key)).AsInteger(); - public async ValueTask IncrByAsync(string key, long by) => (await ExecuteAsync("INCRBY", key, by)).AsInteger(); - public async ValueTask AppendAsync(string key, RedisArg value) => (await ExecuteAsync("APPEND", key, value)).AsInteger(); - public async ValueTask StrLenAsync(string key) => (await ExecuteAsync("STRLEN", key)).AsInteger(); - - public async ValueTask MGetAsync(params string[] keys) - { - var args = Array.ConvertAll(keys, k => (RedisArg)k); - RespValue reply = await ExecuteAsync("MGET", args); - return Array.ConvertAll(reply.Items, v => v.AsString()); - } - - public async ValueTask MSetAsync(params (string Key, RedisArg Value)[] pairs) - { - var args = new RedisArg[pairs.Length * 2]; - for (int i = 0; i < pairs.Length; i++) - { - args[i * 2] = pairs[i].Key; - args[i * 2 + 1] = pairs[i].Value; - } - await ExecuteAsync("MSET", args); - } - - // -- keys ------------------------------------------------------------- - - public async ValueTask DelAsync(params string[] keys) => - (await ExecuteAsync("DEL", Array.ConvertAll(keys, k => (RedisArg)k))).AsInteger(); - - public async ValueTask UnlinkAsync(params string[] keys) => - (await ExecuteAsync("UNLINK", Array.ConvertAll(keys, k => (RedisArg)k))).AsInteger(); - - public async ValueTask ExistsAsync(string key) => (await ExecuteAsync("EXISTS", key)).AsInteger() > 0; - public async ValueTask ExpireAsync(string key, int seconds) => (await ExecuteAsync("EXPIRE", key, seconds)).AsBool(); - public async ValueTask PExpireAsync(string key, long ms) => (await ExecuteAsync("PEXPIRE", key, ms)).AsBool(); - public async ValueTask PersistAsync(string key) => (await ExecuteAsync("PERSIST", key)).AsBool(); - public async ValueTask TtlAsync(string key) => (await ExecuteAsync("TTL", key)).AsInteger(); - public async ValueTask TypeAsync(string key) => (await ExecuteAsync("TYPE", key)).AsString(); - public async ValueTask RenameAsync(string key, string newKey) => await ExecuteAsync("RENAME", key, newKey); - - // -- hashes ----------------------------------------------------------- - - public async ValueTask HGetAsync(string key, string field) => (await ExecuteAsync("HGET", key, field)).AsString(); - public async ValueTask HSetAsync(string key, string field, RedisArg value) => (await ExecuteAsync("HSET", key, field, value)).AsInteger(); - public async ValueTask HExistsAsync(string key, string field) => (await ExecuteAsync("HEXISTS", key, field)).AsBool(); - public async ValueTask HDelAsync(string key, params string[] fields) - { - var args = new RedisArg[fields.Length + 1]; - args[0] = key; - for (int i = 0; i < fields.Length; i++) args[i + 1] = fields[i]; - return (await ExecuteAsync("HDEL", args)).AsInteger(); - } - public async ValueTask HLenAsync(string key) => (await ExecuteAsync("HLEN", key)).AsInteger(); - public async ValueTask HIncrByAsync(string key, string field, long by) => (await ExecuteAsync("HINCRBY", key, field, by)).AsInteger(); - - public async ValueTask> HGetAllAsync(string key) - { - RespValue reply = await ExecuteAsync("HGETALL", key); - var map = new Dictionary(reply.Items.Length / 2); - for (int i = 0; i + 1 < reply.Items.Length; i += 2) - { - map[reply.Items[i].AsString()!] = reply.Items[i + 1].AsString()!; - } - return map; - } - - // -- lists ------------------------------------------------------------ - - public async ValueTask LPushAsync(string key, params RedisArg[] values) => (await ExecuteAsync("LPUSH", Prepend(key, values))).AsInteger(); - public async ValueTask RPushAsync(string key, params RedisArg[] values) => (await ExecuteAsync("RPUSH", Prepend(key, values))).AsInteger(); - public async ValueTask LPopAsync(string key) => (await ExecuteAsync("LPOP", key)).AsString(); - public async ValueTask RPopAsync(string key) => (await ExecuteAsync("RPOP", key)).AsString(); - public async ValueTask LLenAsync(string key) => (await ExecuteAsync("LLEN", key)).AsInteger(); - public async ValueTask LIndexAsync(string key, long index) => (await ExecuteAsync("LINDEX", key, index)).AsString(); - public async ValueTask LTrimAsync(string key, long start, long stop) => await ExecuteAsync("LTRIM", key, start, stop); - - public async ValueTask LRangeAsync(string key, long start, long stop) - { - RespValue reply = await ExecuteAsync("LRANGE", key, start, stop); - return Array.ConvertAll(reply.Items, v => v.AsString()); - } - - // -- sets ------------------------------------------------------------- - - public async ValueTask SAddAsync(string key, params RedisArg[] members) => (await ExecuteAsync("SADD", Prepend(key, members))).AsInteger(); - public async ValueTask SRemAsync(string key, params RedisArg[] members) => (await ExecuteAsync("SREM", Prepend(key, members))).AsInteger(); - public async ValueTask SIsMemberAsync(string key, RedisArg member) => (await ExecuteAsync("SISMEMBER", key, member)).AsBool(); - public async ValueTask SCardAsync(string key) => (await ExecuteAsync("SCARD", key)).AsInteger(); - - public async ValueTask SMembersAsync(string key) - { - RespValue reply = await ExecuteAsync("SMEMBERS", key); - return Array.ConvertAll(reply.Items, v => v.AsString()); - } - - // -- sorted sets ------------------------------------------------------ - - public async ValueTask ZAddAsync(string key, double score, RedisArg member) => (await ExecuteAsync("ZADD", key, score, member)).AsInteger(); - public async ValueTask ZRemAsync(string key, params RedisArg[] members) => (await ExecuteAsync("ZREM", Prepend(key, members))).AsInteger(); - public async ValueTask ZCardAsync(string key) => (await ExecuteAsync("ZCARD", key)).AsInteger(); - public async ValueTask ZScoreAsync(string key, RedisArg member) - { - RespValue reply = await ExecuteAsync("ZSCORE", key, member); - return reply.IsNull ? null : reply.AsDouble(); - } - public async ValueTask ZRankAsync(string key, RedisArg member) - { - RespValue reply = await ExecuteAsync("ZRANK", key, member); - return reply.IsNull ? null : reply.AsInteger(); - } - public async ValueTask ZIncrByAsync(string key, double by, RedisArg member) => (await ExecuteAsync("ZINCRBY", key, by, member)).AsDouble(); - - public async ValueTask ZRangeAsync(string key, long start, long stop) - { - RespValue reply = await ExecuteAsync("ZRANGE", key, start, stop); - return Array.ConvertAll(reply.Items, v => v.AsString()); - } - - // -- pub/sub & server ------------------------------------------------- - - public async ValueTask PublishAsync(string channel, RedisArg message) => (await ExecuteAsync("PUBLISH", channel, message)).AsInteger(); - public async ValueTask PingAsync() => (await ExecuteAsync("PING")).AsString(); - public async ValueTask EchoAsync(RedisArg message) => (await ExecuteAsync("ECHO", message)).AsString(); - public async ValueTask FlushDbAsync() => await ExecuteAsync("FLUSHDB"); - public async ValueTask DbSizeAsync() => (await ExecuteAsync("DBSIZE")).AsInteger(); - - // -- scripting -------------------------------------------------------- - - public async ValueTask EvalAsync(string script, string[] keys, params RedisArg[] args) - { - var all = new RedisArg[2 + keys.Length + args.Length]; - all[0] = script; - all[1] = keys.Length; - for (int i = 0; i < keys.Length; i++) all[2 + i] = keys[i]; - for (int i = 0; i < args.Length; i++) all[2 + keys.Length + i] = args[i]; - return await ExecuteAsync("EVAL", all); - } - - private static RedisArg[] Prepend(string key, RedisArg[] rest) - { - var args = new RedisArg[rest.Length + 1]; - args[0] = key; - Array.Copy(rest, 0, args, 1, rest.Length); - return args; - } -} diff --git a/src/clients/ioxide.redis/RedisConnection.cs b/src/clients/ioxide.redis/RedisConnection.cs deleted file mode 100644 index 62a80acc..00000000 --- a/src/clients/ioxide.redis/RedisConnection.cs +++ /dev/null @@ -1,385 +0,0 @@ -using System.Collections.Concurrent; -using System.Runtime.InteropServices; -using System.Text; -using System.Threading.Tasks.Sources; - -namespace ioxide.redis; - -/// -/// One Redis connection on a reactor's ring. Connect, auth, and every command run as ring ops with -/// inline completion resume. The protocol surface is generic - -/// runs any command - with typed helpers (in RedisConnection.Commands.cs) layered on top. Commands are -/// pipelined: many in flight at once, replies routed FIFO, so one connection stays busy under load. -/// -public sealed partial class RedisConnection : IDisposable -{ - private const int InitialBufferSize = 16 * 1024; - private const int SendBufferSize = 512 * 1024; // fixed: a pipelined send holds _send's address, so never realloc'd - private const int MaxBufferSize = 64 * 1024 * 1024; - - private readonly RingSocket _socket; - - private nint _send; - private int _sendCapacity; - private nint _recv; - private int _recvCapacity; - private int _received; // valid bytes in _recv - private int _scan; // start of the first unparsed reply - private int _need; // examined-cursor: bytes required in [_scan,_received) before the front reply can parse (0 = unknown) - - // Pipelining: commands sent back to back, one RESP reply each, routed FIFO. Single-threaded - // per reactor, so no locks. _send staged at [_sendOffset, _sendEnd). - private readonly Queue _inflight = new(); - private bool _sending; - private bool _reading; - private int _sendOffset; - private int _sendEnd; - - public bool IsBroken { get; private set; } - - private unsafe RedisConnection(RingSocket socket) - { - _socket = socket; - _send = (nint)NativeMemory.Alloc(SendBufferSize); - _sendCapacity = SendBufferSize; - _recv = (nint)NativeMemory.Alloc(InitialBufferSize); - _recvCapacity = InitialBufferSize; - } - - public static async Task ConnectAsync(IRingHost host, RedisOptions options) - { - RingSocket socket = RingSocket.CreateTcp(host); - var connection = new RedisConnection(socket); - - try - { - int rc = await socket.ConnectAsync(options.Host, options.Port); - if (rc < 0) - { - throw RedisException.Transport($"connect to {options.Host}:{options.Port}", rc); - } - - if (!string.IsNullOrEmpty(options.Password)) - { - _ = options.User is { } user - ? await connection.ExecuteAsync("AUTH", user, options.Password) - : await connection.ExecuteAsync("AUTH", options.Password); - } - - if (options.Database != 0) - { - await connection.ExecuteAsync("SELECT", options.Database); - } - - return connection; - } - catch - { - connection.Dispose(); - throw; - } - } - - // Command names are a tiny fixed set; cache the full pre-framed RESP token ($len\r\nNAME\r\n) so - // the hot path memcpy's it in one shot - no per-call encoding or framing, no byte[] alloc. - // Static ConcurrentDictionary: shared and thread-safe across reactors. - private static readonly ConcurrentDictionary CommandCache = new(); - private static byte[] EncodeCommand(string command) => - CommandCache.GetOrAdd(command, static c => RespProtocol.FrameName(Encoding.ASCII.GetBytes(c))); - - /// Run any command and return its reply. Throws on a top-level error reply. - public async ValueTask ExecuteAsync(string command, params RedisArg[] args) - { - RespValue reply = await SubmitCore(EncodeCommand(command), args); - if (reply.IsError) - { - throw new RedisException(reply.AsString() ?? "redis error"); - } - - return reply; - } - - // Stage a command, enqueue its waiter, ensure the sender and reader are running. - private ValueTask SubmitCore(ReadOnlySpan nameToken, RedisArg[] args) - { - if (IsBroken) - { - return ValueTask.FromException(new RedisException("connection is broken")); - } - - try - { - AppendCommand(nameToken, args); - } - catch (RedisException ex) - { - return ValueTask.FromException(ex); - } - - var pending = new Pending { EnqueuedAtMs = Native.MonotonicMs }; - _inflight.Enqueue(pending); - - if (!_sending) - { - _sending = true; - _ = SenderLoopAsync(); - } - - if (!_reading) - { - _reading = true; - _ = ReaderLoopAsync(); - } - - return new ValueTask(pending, pending.Version); - } - - /// - /// Send several commands back to back and read all replies in one round trip. Each element of - /// is (command, args). Error replies are returned, not thrown, so - /// the caller can inspect per-command outcomes. - /// - public async ValueTask PipelineAsync(params RedisCommand[] commands) - { - if (IsBroken) - { - throw new RedisException("connection is broken"); - } - - var pending = new ValueTask[commands.Length]; - for (int i = 0; i < commands.Length; i++) - { - pending[i] = SubmitCore(commands[i].NameToken, commands[i].Args); - } - - var replies = new RespValue[commands.Length]; - for (int i = 0; i < commands.Length; i++) - { - replies[i] = await pending[i]; - } - - return replies; - } - - // -- wire I/O --------------------------------------------------------- - - // Append a RESP command to the pipelined send buffer. Fixed buffer (an in-flight send holds its - // address), so this never realloc's - it throws on overflow instead. - private unsafe void AppendCommand(ReadOnlySpan nameToken, RedisArg[] args) - { - int size = RespProtocol.CommandSize(nameToken, args); - if (size > _sendCapacity) - { - throw new RedisException($"command exceeds send buffer ({_sendCapacity} bytes)"); - } - - if (_sendEnd + size > _sendCapacity) - { - throw new RedisException("pipelined send buffer full"); - } - - int written = RespProtocol.WriteCommand(new Span((void*)(_send + _sendEnd), _sendCapacity - _sendEnd), nameToken, args); - _sendEnd += written; - } - - private async Task SenderLoopAsync() - { - try - { - while (_sendOffset < _sendEnd) - { - int n = await _socket.SendAsync(_send + _sendOffset, _sendEnd - _sendOffset); - if (n <= 0) - { - IsBroken = true; - throw RedisException.Transport("send", n); - } - _sendOffset += n; - } - // No re-check is needed before clearing _sending: there is no await between the final - // drain check above and here, and send completions resume inline (synchronously) on the - // reactor, so no SubmitCore can interleave between them. A command appended while a send - // was in flight already advanced _sendEnd and was picked up by the loop condition. (This - // invariant depends on RingOpSource keeping RunContinuationsAsynchronously = false.) - _sendOffset = 0; // all staged bytes sent; reuse the buffer from the front - _sendEnd = 0; - _sending = false; - } - catch (Exception ex) - { - IsBroken = true; - _sending = false; - FailAll(ex); - } - } - - private async Task ReaderLoopAsync() - { - try - { - while (_inflight.Count > 0) - { - RespValue reply = await ReceiveReplyAsync(); - _inflight.Dequeue().Complete(reply); - } - _reading = false; - } - catch (Exception ex) - { - IsBroken = true; - _reading = false; - FailAll(ex); - } - } - - private void FailAll(Exception ex) - { - RedisException re = ex as RedisException ?? new RedisException(ex.Message); - while (_inflight.TryDequeue(out Pending? p)) - { - p.Fail(re); - } - } - - // Reactor-thread (pool ticker): tear the connection down if its oldest in-flight command is - // overdue - fail waiters with a diagnostic error and mark it broken; the pool disposes it, and - // closing the fd cancels the stuck ring recv/send. Returns true if it timed out. - internal bool CheckTimeout(long nowMs, int timeoutMs, string host, ushort port) - { - if (timeoutMs <= 0 || _inflight.Count == 0) - { - return false; - } - - long age = nowMs - _inflight.Peek().EnqueuedAtMs; - if (age <= timeoutMs) - { - return false; - } - - int inflight = _inflight.Count; - IsBroken = true; - FailAll(new RedisException( - $"redis command timed out after {timeoutMs} ms ({inflight} in flight on {host}:{port}, oldest sent ~{age} ms ago)")); - return true; - } - - // One pipelined command: completes its waiter with the RESP reply (inline on the reactor). - private sealed class Pending : IValueTaskSource - { - private ManualResetValueTaskSourceCore _core = new() { RunContinuationsAsynchronously = false }; - - public short Version => _core.Version; - public long EnqueuedAtMs; // Native.MonotonicMs at enqueue - drives the command-timeout sweep - public void Complete(RespValue reply) => _core.SetResult(reply); - public void Fail(RedisException ex) => _core.SetException(ex); - - public RespValue GetResult(short token) => _core.GetResult(token); - public ValueTaskSourceStatus GetStatus(short token) => _core.GetStatus(token); - // Completes on the reactor thread only - strip the context-post so resumes stay inline - // (see ReactorSynchronizationContext). - public void OnCompleted(Action continuation, object? state, short token, ValueTaskSourceOnCompletedFlags flags) - => _core.OnCompleted(continuation, state, token, flags & ~ValueTaskSourceOnCompletedFlags.UseSchedulingContext); - } - - private async ValueTask ReceiveReplyAsync() - { - while (true) - { - // Only re-parse once enough bytes for the current sticking point have arrived; below that - // threshold a re-scan can't make progress (examined-cursor). _need == 0 means "unknown" - - // parse eagerly. - if (_received - _scan >= _need) - { - if (TryParseReply(out RespValue reply, out int need)) - { - _need = 0; - return reply; - } - _need = need; - } - - if (_scan == _received) - { - _scan = 0; - _received = 0; - _need = 0; - } - - EnsureRecvSpace(); - - int n = await _socket.RecvAsync(_recv + _received, _recvCapacity - _received); - if (n <= 0) - { - IsBroken = true; - throw RedisException.Transport("recv", n); - } - _received += n; - } - } - - private unsafe bool TryParseReply(out RespValue reply, out int needed) - { - var buffered = new ReadOnlySpan((void*)(_recv + _scan), _received - _scan); - if (RespProtocol.TryParse(buffered, out reply, out int consumed, out needed)) - { - _scan += consumed; - return true; - } - - return false; - } - - private unsafe void EnsureRecvSpace() - { - if (_received < _recvCapacity) - { - return; - } - - if (_scan > 0) - { - Buffer.MemoryCopy((void*)(_recv + _scan), (void*)_recv, _recvCapacity, _received - _scan); - _received -= _scan; - _scan = 0; - return; - } - - if (_recvCapacity >= MaxBufferSize) - { - IsBroken = true; - throw new RedisException($"reply exceeds {MaxBufferSize} bytes"); - } - _recvCapacity *= 2; - _recv = (nint)NativeMemory.Realloc((void*)_recv, (nuint)_recvCapacity); - } - - public unsafe void Dispose() - { - _socket.Dispose(); - if (_send != 0) - { - NativeMemory.Free((void*)_send); - _send = 0; - } - - if (_recv != 0) - { - NativeMemory.Free((void*)_recv); - _recv = 0; - } - } -} - -/// A command for : a name and its arguments. -public readonly struct RedisCommand -{ - internal readonly byte[] NameToken; // pre-framed $len\r\nNAME\r\n - internal readonly RedisArg[] Args; - - public RedisCommand(string name, params RedisArg[] args) - { - NameToken = RespProtocol.FrameName(Encoding.ASCII.GetBytes(name)); - Args = args; - } -} diff --git a/src/clients/ioxide.redis/RedisException.cs b/src/clients/ioxide.redis/RedisException.cs deleted file mode 100644 index fa3178e3..00000000 --- a/src/clients/ioxide.redis/RedisException.cs +++ /dev/null @@ -1,10 +0,0 @@ -namespace ioxide.redis; - -/// A Redis error reply (-ERR ...) or a transport failure. -public sealed class RedisException : Exception -{ - public RedisException(string message) : base(message) { } - - internal static RedisException Transport(string what, int errno) => - new($"redis {what} failed: errno {-errno}"); -} diff --git a/src/clients/ioxide.redis/RedisOptions.cs b/src/clients/ioxide.redis/RedisOptions.cs deleted file mode 100644 index 35e9b098..00000000 --- a/src/clients/ioxide.redis/RedisOptions.cs +++ /dev/null @@ -1,32 +0,0 @@ -namespace ioxide.redis; - -/// How to reach Redis and how much capacity to provision per reactor. -public sealed record RedisOptions -{ - /// IPv4 literal. Resolve hostnames up front - DNS would block the reactor. - public string Host { get; init; } = "127.0.0.1"; - - public ushort Port { get; init; } = 6379; - - /// AUTH password (or null for no auth). With set, uses ACL AUTH. - public string? Password { get; init; } - - /// ACL username (Redis 6+); null uses the default user with . - public string? User { get; init; } - - /// Logical database index to SELECT on connect. - public int Database { get; init; } - - /// - /// Connections opened per reactor. One connection carries one in-flight command (or pipeline) - /// at a time, so this is the reactor's Redis concurrency; total = PoolSize × ReactorCount. - /// - public int PoolSize { get; init; } = 8; - - /// - /// Per-command timeout in milliseconds: a connection whose oldest in-flight command exceeds this - /// is torn down and its waiters fail with a diagnostic error, so a silent server can't park a - /// command forever. 0 disables. Default 30000. - /// - public int CommandTimeoutMs { get; init; } = 30_000; -} diff --git a/src/clients/ioxide.redis/RedisPool.cs b/src/clients/ioxide.redis/RedisPool.cs deleted file mode 100644 index a7f27bd6..00000000 --- a/src/clients/ioxide.redis/RedisPool.cs +++ /dev/null @@ -1,202 +0,0 @@ -namespace ioxide.redis; - -/// -/// N Redis connections on one reactor's ring, with pipelining. Commands round-robin across the -/// connections and each connection multiplexes many in flight, so the connections stay busy under -/// load instead of one round trip at a time. Broken connections are evicted and replaced. One pool -/// per reactor - create from Reactor.OnStart. -/// -/// -/// Not thread-safe by design: every member must run on the owning reactor thread - commands arrive -/// from reactor-thread handlers and connect completions resume there too. Do not call the pool from -/// off-reactor code. -/// -public sealed class RedisPool -{ - // Bound the slow-path wait so a pool that can't reach Redis fails fast with a clear error - // instead of parking commands forever; each attempt drives one coalesced connect try. - private const int MaxConnectAttempts = 10; - - private readonly List _connections = new(); - private readonly IRingHost _host; - private readonly RedisOptions _options; - private int _next; - private int _opening; // in-flight connect attempts (coalesces opens) - private long _reopenAtMs; // jittered backoff gate for background replenishment - private TaskCompletionSource? _opened; // completed when any open attempt finishes - - private RedisPool(IRingHost host, RedisOptions options) - { - _host = host; - _options = options; - } - - /// Open connections and register the pool as a service. - public static RedisPool Start(Reactor reactor, RedisOptions options) - { - var pool = new RedisPool(reactor, options); - for (int i = 0; i < options.PoolSize; i++) - { - pool.StartOpen(); - } - reactor.AddService(pool); - reactor.AddTicker(pool.Sweep); // per-command timeout sweep + background replenishment - return pool; - } - - // Begin one connect attempt; OpenOneAsync clears the in-flight count and wakes waiters when done. - private void StartOpen() - { - _opening++; - _ = OpenOneAsync(); - } - - private async Task OpenOneAsync() - { - try - { - RedisConnection connection = await RedisConnection.ConnectAsync(_host, _options); - _connections.Add(connection); - _reopenAtMs = 0; // healthy - allow immediate replenishment - } - catch (Exception e) - { - Console.Error.WriteLine($"[redis] connect to {_options.Host}:{_options.Port} failed: {e.Message}"); - _reopenAtMs = Native.MonotonicMs + BackoffMs(); // jittered backoff before the next attempt - } - finally - { - _opening--; - // Wake every slow-path waiter whether the attempt succeeded or failed, so a run of - // failures surfaces as a fast error instead of a permanent hang. Swap the field out - // before completing so a fresh waiter arms a new signal. - TaskCompletionSource? signal = _opened; - _opened = null; - signal?.TrySetResult(true); - } - } - - /// Live connections (for diagnostics). - public int ConnectionCount => _connections.Count; - - /// Run any command on the next connection (pipelined). - public ValueTask ExecuteAsync(string command, params RedisArg[] args) - { - RedisConnection? c = Pick(); - return c != null ? c.ExecuteAsync(command, args) : ExecuteSlowAsync(command, args); - } - - /// Send several commands on one connection in a single round trip; replies returned in order. - public ValueTask PipelineAsync(params RedisCommand[] commands) - { - RedisConnection? c = Pick(); - return c != null ? c.PipelineAsync(commands) : PipelineSlowAsync(commands); - } - - /// GET (the cache-aside read path). - public ValueTask GetAsync(string key) - { - RedisConnection? c = Pick(); - return c != null ? c.GetAsync(key) : GetSlowAsync(key); - } - - /// SET key value EX seconds (the cache-aside populate path). - public ValueTask SetExAsync(string key, RedisArg value, int seconds) - { - RedisConnection? c = Pick(); - return c != null ? c.SetExAsync(key, value, seconds) : SetExSlowAsync(key, value, seconds); - } - - /// DEL (cache invalidation). - public ValueTask DelAsync(params string[] keys) - { - RedisConnection? c = Pick(); - return c != null ? c.DelAsync(keys) : DelSlowAsync(keys); - } - - private async ValueTask ExecuteSlowAsync(string command, RedisArg[] args) - => await (await WaitForConnectionAsync()).ExecuteAsync(command, args); - private async ValueTask GetSlowAsync(string key) - => await (await WaitForConnectionAsync()).GetAsync(key); - private async ValueTask SetExSlowAsync(string key, RedisArg value, int seconds) - => await (await WaitForConnectionAsync()).SetExAsync(key, value, seconds); - private async ValueTask DelSlowAsync(string[] keys) - => await (await WaitForConnectionAsync()).DelAsync(keys); - private async ValueTask PipelineSlowAsync(RedisCommand[] commands) - => await (await WaitForConnectionAsync()).PipelineAsync(commands); - - // Round-robin over healthy connections; evict and replace any found broken. - private RedisConnection? Pick() - { - while (_connections.Count > 0) - { - int index = (_next++ & 0x7fffffff) % _connections.Count; - RedisConnection c = _connections[index]; - if (!c.IsBroken) - { - return c; - } - _connections.RemoveAt(index); - c.Dispose(); - StartOpen(); - } - return null; - } - - // No live connection (cold start, or every connection broke). Drive coalesced connect attempts - // and wait for one to finish; fail fast after MaxConnectAttempts so commands never hang. - private async ValueTask WaitForConnectionAsync() - { - for (int attempt = 1; ; attempt++) - { - RedisConnection? c = Pick(); - if (c != null) - { - return c; - } - - if (attempt > MaxConnectAttempts) - { - throw new RedisException( - $"redis pool: no connection to {_options.Host}:{_options.Port} after {MaxConnectAttempts} attempts"); - } - - _opened ??= new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - Task wait = _opened.Task; - if (_opening == 0) - { - StartOpen(); // nothing live and nothing opening - kick one off - } - await wait; - } - } - - // Reactor-thread ticker (~250 ms): time out stuck connections and replenish toward PoolSize. - private void Sweep() - { - long now = Native.MonotonicMs; - - int timeoutMs = _options.CommandTimeoutMs; - if (timeoutMs > 0) - { - for (int i = _connections.Count - 1; i >= 0; i--) - { - if (_connections[i].CheckTimeout(now, timeoutMs, _options.Host, _options.Port)) - { - RedisConnection c = _connections[i]; - _connections.RemoveAt(i); - c.Dispose(); // closes the fd - unsticks the connection's stuck send/recv loops - } - } - } - - // Background replenishment toward PoolSize, gated by a jittered backoff so a recovering - // server isn't hammered by every reactor at once. - if (_connections.Count < _options.PoolSize && _opening == 0 && now >= _reopenAtMs) - { - StartOpen(); - } - } - - private static int BackoffMs() => 200 + Random.Shared.Next(400); // jittered ~200-600 ms -} diff --git a/src/clients/ioxide.redis/RespProtocol.cs b/src/clients/ioxide.redis/RespProtocol.cs deleted file mode 100644 index 8b79544e..00000000 --- a/src/clients/ioxide.redis/RespProtocol.cs +++ /dev/null @@ -1,232 +0,0 @@ -using System.Buffers.Text; - -namespace ioxide.redis; - -/// RESP2 wire format: write a command as an array of bulk strings; parse one reply. -internal static class RespProtocol -{ - /// - /// Bytes needed to encode a command as a RESP array of bulk strings, given a pre-framed name - /// token ($len\r\nNAME\r\n, see ) plus the argument list. - /// - public static int CommandSize(ReadOnlySpan nameToken, ReadOnlySpan args) - { - int size = 1 + IntLen(args.Length + 1) + 2; // *\r\n - size += nameToken.Length; // name is already framed - foreach (RedisArg arg in args) - { - size += BulkSize(arg.Bytes.Length); - } - - return size; - } - - /// Write a command into (sized by ). - public static int WriteCommand(Span dst, ReadOnlySpan nameToken, ReadOnlySpan args) - { - int p = 0; - dst[p++] = (byte)'*'; - p += WriteInt(dst[p..], args.Length + 1); - dst[p++] = (byte)'\r'; - dst[p++] = (byte)'\n'; - - nameToken.CopyTo(dst[p..]); // pre-framed $len\r\nNAME\r\n - one memcpy, no per-call framing - p += nameToken.Length; - - foreach (RedisArg arg in args) - { - p += WriteBulk(dst[p..], arg.Bytes); - } - - return p; - } - - /// - /// Frame a command name as a RESP bulk-string token ($len\r\nNAME\r\n). Computed once per - /// command name and cached, so the hot path memcpy's the whole token instead of re-framing it. - /// - public static byte[] FrameName(ReadOnlySpan name) - { - byte[] token = new byte[BulkSize(name.Length)]; - WriteBulk(token, name); - return token; - } - - private static int WriteBulk(Span dst, ReadOnlySpan payload) - { - int p = 0; - dst[p++] = (byte)'$'; - p += WriteInt(dst[p..], payload.Length); - dst[p++] = (byte)'\r'; - dst[p++] = (byte)'\n'; - payload.CopyTo(dst[p..]); - p += payload.Length; - dst[p++] = (byte)'\r'; - dst[p++] = (byte)'\n'; - return p; - } - - private static int BulkSize(int payloadLength) => 1 + IntLen(payloadLength) + 2 + payloadLength + 2; - - private static int WriteInt(Span dst, int value) - { - Utf8Formatter.TryFormat(value, dst, out int written); - return written; - } - - private static int IntLen(int value) - { - int len = value < 0 ? 1 : 0; - do - { - len++; - value /= 10; - } while (value != 0); - return len; - } - - /// - /// Parse one complete reply (recursively, for arrays) from . False if - /// the reply isn't fully buffered yet; on true, is its byte length. - /// On false, is the buffer length required before a re-parse can make - /// progress (0 = unknown), so the caller can avoid re-scanning until enough bytes arrive. - /// - public static bool TryParse(ReadOnlySpan buffer, out RespValue value, out int consumed, out int needed) - { - int pos = 0; - needed = 0; - if (TryParseAt(buffer, ref pos, out value, ref needed)) - { - consumed = pos; - return true; - } - consumed = 0; - return false; - } - - private static bool TryParseAt(ReadOnlySpan buffer, ref int pos, out RespValue value, ref int needed) - { - value = default; - if (pos >= buffer.Length) - { - return false; - } - - byte type = buffer[pos]; - int lineEnd = buffer[(pos + 1)..].IndexOf("\r\n"u8); - if (lineEnd < 0) - { - return false; - } - ReadOnlySpan line = buffer.Slice(pos + 1, lineEnd); - int afterLine = pos + 1 + lineEnd + 2; - - switch (type) - { - case (byte)'+': - value = RespValue.Simple(line.ToArray()); - pos = afterLine; - return true; - - case (byte)'-': - value = RespValue.Err(line.ToArray()); - pos = afterLine; - return true; - - case (byte)':': - value = RespValue.Int(ParseLong(line)); - pos = afterLine; - return true; - - case (byte)'$': - { - long len = ParseLong(line); - if (len == -1) - { - value = RespValue.Null; - pos = afterLine; - return true; - } - - if (len < -1) - { - throw new RedisException($"invalid bulk length {len}"); - } - - if (afterLine + len + 2 > buffer.Length) - { - // Body not fully buffered: report how many bytes are needed so the reader can - // skip re-parsing until they arrive, instead of re-scanning on every recv. - needed = afterLine + (int)len + 2; - return false; - } - value = RespValue.Bulk(buffer.Slice(afterLine, (int)len).ToArray()); - pos = afterLine + (int)len + 2; - return true; - } - - case (byte)'*': - { - long count = ParseLong(line); - if (count == -1) - { - value = RespValue.Null; - pos = afterLine; - return true; - } - - if (count < -1 || count > int.MaxValue) - { - throw new RedisException($"invalid array length {count}"); - } - - var items = new RespValue[count]; - int p = afterLine; - for (int i = 0; i < count; i++) - { - if (!TryParseAt(buffer, ref p, out RespValue item, ref needed)) - { - return false; - } - items[i] = item; - } - value = RespValue.Arr(items); - pos = p; - return true; - } - - case (byte)'_': // RESP3 null; tolerated even though we negotiate RESP2 - value = RespValue.Null; - pos = afterLine; - return true; - - default: - throw new RedisException($"unexpected RESP type byte 0x{type:x2} (RESP2 expected)"); - } - } - - // Strict: RESP integers and bulk/array lengths are an optional '-' then ASCII digits. Anything - // else is a malformed reply - throw so the reader breaks the connection cleanly instead of - // computing a garbage length and wedging (or slicing out of range). - private static long ParseLong(ReadOnlySpan s) - { - bool neg = s.Length > 0 && s[0] == (byte)'-'; - if (neg) s = s[1..]; - if (s.IsEmpty) - { - throw new RedisException("malformed RESP integer"); - } - - long n = 0; - foreach (byte c in s) - { - if (c < (byte)'0' || c > (byte)'9') - { - throw new RedisException($"malformed RESP integer (byte 0x{c:x2})"); - } - n = checked(n * 10 + (c - '0')); - } - - return neg ? -n : n; - } -} diff --git a/src/clients/ioxide.redis/RespValue.cs b/src/clients/ioxide.redis/RespValue.cs deleted file mode 100644 index a713788e..00000000 --- a/src/clients/ioxide.redis/RespValue.cs +++ /dev/null @@ -1,92 +0,0 @@ -using System.Buffers.Text; -using System.Text; - -namespace ioxide.redis; - -public enum RespKind { Null, SimpleString, Error, Integer, BulkString, Array } - -/// -/// A parsed RESP2 reply. A value type, so parsing a reply allocates no per-node object: -/// integers, nulls and booleans are entirely allocation-free; bulk/simple strings and arrays carry -/// the bytes (or element array) materialized during parse, which outlive the receive buffer. -/// Accessors convert on demand. covers both null bulk strings ($-1) and -/// null arrays (*-1). (Layout follows StackExchange.Redis's allocation-light RedisValue.) -/// -public readonly struct RespValue -{ - /// The RESP null (a null bulk string or array); equivalent to default. - public static readonly RespValue Null = default; - - public RespKind Kind { get; } - public long Integer { get; } - private readonly byte[]? _bytes; - private readonly RespValue[]? _items; - - private RespValue(RespKind kind, long integer, byte[]? bytes, RespValue[]? items) - { - Kind = kind; - Integer = integer; - _bytes = bytes; - _items = items; - } - - internal static RespValue Simple(byte[] b) => new(RespKind.SimpleString, 0, b, null); - internal static RespValue Err(byte[] b) => new(RespKind.Error, 0, b, null); - internal static RespValue Int(long n) => new(RespKind.Integer, n, null, null); - internal static RespValue Bulk(byte[] b) => new(RespKind.BulkString, 0, b, null); - internal static RespValue Arr(RespValue[] items) => new(RespKind.Array, 0, null, items); - - public bool IsNull => Kind == RespKind.Null; - public bool IsError => Kind == RespKind.Error; - - /// Raw bytes of a bulk/simple string/error; null for other kinds. - public byte[]? Bytes => _bytes; - - /// Array elements (empty for non-arrays). - public RespValue[] Items => _items ?? []; - - public string? AsString() => _bytes is null ? null : Encoding.UTF8.GetString(_bytes); - - public long AsInteger() => Kind switch - { - RespKind.Integer => Integer, - RespKind.BulkString or RespKind.SimpleString when _bytes is not null - && Utf8Parser.TryParse(_bytes, out long n, out int c) && c == _bytes.Length => n, - _ => throw new RedisException($"reply ({Kind}) is not an integer"), - }; - - /// - /// A numeric reply as a double. Accepts integers, and the bulk/simple-string forms Redis uses for - /// scores - including the non-finite tokens inf/+inf/-inf/nan that - /// won't parse (e.g. a ZSCORE of +inf). - /// - public double AsDouble() - { - if (Kind == RespKind.Integer) - { - return Integer; - } - if (_bytes is not null && (Kind == RespKind.BulkString || Kind == RespKind.SimpleString)) - { - if (Utf8Parser.TryParse(_bytes, out double d, out int n) && n == _bytes.Length) - { - return d; - } - ReadOnlySpan b = _bytes; - if (b.SequenceEqual("inf"u8) || b.SequenceEqual("+inf"u8)) return double.PositiveInfinity; - if (b.SequenceEqual("-inf"u8)) return double.NegativeInfinity; - if (b.SequenceEqual("nan"u8)) return double.NaN; - } - throw new RedisException($"reply ({Kind}) is not a double"); - } - - public bool AsBool() => AsInteger() != 0; - - public override string ToString() => Kind switch - { - RespKind.Null => "(nil)", - RespKind.Integer => Integer.ToString(), - RespKind.Array => $"[{string.Join(", ", Items.Select(i => i.ToString()))}]", - _ => AsString() ?? "(nil)", - }; -} diff --git a/src/clients/ioxide.redis/ioxide.redis.csproj b/src/clients/ioxide.redis/ioxide.redis.csproj deleted file mode 100644 index 8947ad8e..00000000 --- a/src/clients/ioxide.redis/ioxide.redis.csproj +++ /dev/null @@ -1,27 +0,0 @@ - - - - net10.0;net11.0 - enable - enable - true - ioxide.redis - - ioxide.redis - 0.16.281 - MDA2AV - Redis client for the ioxide io_uring runtime: pooled ring-native connections per reactor, full RESP2 protocol, a generic command API plus typed helpers (strings, keys, hashes, lists, sets, sorted sets, pub/sub, transactions, scripting), and pipelining. Inline completion resume. - MIT - https://mda2av.github.io/ioxide/learn/clients.html - https://github.com/MDA2AV/ioxide - git - io_uring;redis;resp;client;cache;linux;ioxide - README.md - - - - - - - - diff --git a/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj b/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj index 6bc9a7ce..858dc5cc 100644 --- a/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj +++ b/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj @@ -12,7 +12,6 @@ - diff --git a/tests/Ioxide.Tests.Redis/Program.cs b/tests/Ioxide.Tests.Redis/Program.cs index 727bcdcb..9217e330 100644 --- a/tests/Ioxide.Tests.Redis/Program.cs +++ b/tests/Ioxide.Tests.Redis/Program.cs @@ -1,7 +1,7 @@ namespace Ioxide.Tests; /// -/// The Redis client against a real server. Skips entirely when none is reachable: +/// Redis over the ring, against a real server. Skips entirely when none is reachable: /// /// docker run -d --name redis -p 6379:6379 redis:7-alpine /// @@ -17,7 +17,6 @@ private static int Main() bool up = Sidecars.Reachable(redis.Host, redis.Port); Console.WriteLine($"redis {(up ? "up" : "down")} ({redis.Host}:{redis.Port})\n"); - RedisTests.Register(runner, redis, up); return runner.Summary(); } } diff --git a/tests/Ioxide.Tests.Redis/RedisHandlers.cs b/tests/Ioxide.Tests.Redis/RedisHandlers.cs deleted file mode 100644 index 81cc93d4..00000000 --- a/tests/Ioxide.Tests.Redis/RedisHandlers.cs +++ /dev/null @@ -1,56 +0,0 @@ -using System.Text; -using ioxide; -using ioxide.redis; - -namespace Ioxide.Tests; - -/// Handlers that drive a RedisPool, kept beside the suite that exercises them. -internal static class RedisHandlers -{ - public static async Task Redis(Reactor r, TcpConnection conn) - { - RedisPool pool = r.GetService(); - - try - { - while (true) - { - RecvSnapshot snapshot = await conn.ReadAsync(); - string path = Wire.ReadPath(conn, snapshot); - - if (path == "/incr") - { - long n = (await pool.ExecuteAsync("INCR", "e2e:n")).AsInteger(); - Wire.Write(conn, 200, n.ToString()); - } - else if (path == "/pipe") - { - RespValue[] replies = await pool.PipelineAsync( - new RedisCommand("SET", "e2e:p", "1"), - new RedisCommand("INCR", "e2e:p"), - new RedisCommand("GET", "e2e:p")); - Wire.Write(conn, 200, replies[2].AsString() ?? ""); - } - else - { - await pool.SetExAsync("e2e:k", "hello", 60); - string? value = await pool.GetAsync("e2e:k"); - Wire.Write(conn, 200, value ?? ""); - } - - await conn.FlushAsync(); - - if (snapshot.IsClosed) - { - return; - } - - conn.ResetRead(); - } - } - finally - { - conn.DecRef(); - } - } -} diff --git a/tests/Ioxide.Tests.Redis/RedisTests.cs b/tests/Ioxide.Tests.Redis/RedisTests.cs deleted file mode 100644 index bd686e23..00000000 --- a/tests/Ioxide.Tests.Redis/RedisTests.cs +++ /dev/null @@ -1,42 +0,0 @@ -using ioxide.redis; - -namespace Ioxide.Tests; - -/// Redis client over the ring: RESP2 strings/integers and pipelining. -internal static class RedisTests -{ - public static void Register(Runner runner, (string Host, int Port) redis, bool redisUp) - { - runner.Test("redis: SET then GET", () => - { - int port = TestServer.Start(RedisHandlers.Redis, r => RedisPool.Start(r, RedisOpts(redis))); - (int status, string body) = Client.Get(port, "/"); - Assert.Equal(200, status); - Assert.Equal("hello", body); - }, skip: !redisUp); - - runner.Test("redis: INCR (RESP integer)", () => - { - int port = TestServer.Start(RedisHandlers.Redis, r => RedisPool.Start(r, RedisOpts(redis))); - (int status, string body) = Client.Get(port, "/incr"); - Assert.Equal(200, status); - Assert.True(long.TryParse(body, out long n) && n >= 1, $"expected a positive integer, got [{body}]"); - }, skip: !redisUp); - - runner.Test("redis: pipeline SET/INCR/GET", () => - { - int port = TestServer.Start(RedisHandlers.Redis, r => RedisPool.Start(r, RedisOpts(redis))); - (int status, string body) = Client.Get(port, "/pipe"); - Assert.Equal(200, status); - Assert.Equal("2", body); - }, skip: !redisUp); - } - - private static RedisOptions RedisOpts((string Host, int Port) redis) => new() - { - Host = redis.Host, - Port = (ushort)redis.Port, - Password = Environment.GetEnvironmentVariable("EXAMPLES_REDIS_PASSWORD"), - PoolSize = 2, - }; -} diff --git a/tests/README.md b/tests/README.md index 576eadc9..be0ae5dd 100644 --- a/tests/README.md +++ b/tests/README.md @@ -16,7 +16,7 @@ done | `E2E` | Reactor lifecycle, TCP read/write paths, hardening, UDP, **QUIC and HTTP/3 including their TLS** | nothing | | `Http` | The HTTP client: h1, h2c, h3 and the negotiating layer | h2c tests need an HTTP/2 server | | `Pg` | The Postgres driver | a postgres | -| `Redis` | The Redis client | a redis | +| `Redis` | Redis over the ring | a redis | | `Tls` | **TLS over TCP**: handshake, pipes, mutual TLS, SNI, rotation, posture, PEM formats | the `tls` kernel module for the kTLS cases (`sudo modprobe tls`) | | `File` | Static assets: cache, ring reads, revalidation | nothing | | `Chaos` | Malformed and hostile inputs across TCP, TLS, h2c and QUIC/HTTP/3 | nothing |