From 4668bb018dcd91a85c2f0bc2b1dde289c7325f3f Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Mon, 12 Oct 2026 15:06:27 +0100 Subject: [PATCH 1/2] redis.stackexchange: StackExchange.Redis over the ring - a Tunnel whose transport is a socket on a reactor StackExchange.Redis 3.x has an experimental transport seam (SER009): a Tunnel can hand a multiplexer a whole DuplexTransport through ConnectTransportAsync, and StackExchange.Redis then opens no socket of its own. Outbound the transport is an IBufferWriter staged from any thread with an explicit Flush; inbound is push - OnReceived with transport-owned bytes, OnBatchEnd, OnClosed - and StackExchange.Redis parses and completes commands on the thread that pushed them. SocketSet's SocketSet.StackExchange.Redis is the reference implementation. IoxideTunnel implements it on a reactor: the transport is a RingSocket, its receives complete on the reactor and push into StackExchange.Redis there, one OnBatchEnd is posted per loop turn that delivered anything, and Flush hands the staged bytes to the ring as one send (bytes flushed while a send is in flight queue behind it). With one multiplexer per reactor, made in OnStart, a handler's commands are staged, sent and completed on its own thread. TCP without TLS for now: a configuration asking for TLS is refused with a message saying so. StackExchange.Redis is pinned to 3.4.0, since the seam may change in any release. Tests run against the suite's redis: SET/GET completed on the reactor's thread, 1000 concurrent INCRs, a 1 MiB value, pub/sub, a connection the server kills coming back through the tunnel, and the TLS refusal. The thread test is discriminating: on StackExchange.Redis's own sockets the completion is posted from another thread and it fails. A pending test pins a StackExchange.Redis gap the transport surfaced: RecordConnectionFailed reports a failure only for a connection with a Stream, so a command caught by a transport's close is cancelled instead of failing with the connection's error. --- README.md | 1 + ioxide.slnx | 1 + .../ioxide.redis.stackexchange/IoxideRedis.cs | 24 ++ .../IoxideTransport.cs | 346 ++++++++++++++++++ .../IoxideTunnel.cs | 48 +++ .../ioxide.redis.stackexchange.csproj | 36 ++ .../Ioxide.Tests.Redis.csproj | 1 + tests/Ioxide.Tests.Redis/Program.cs | 1 + .../StackExchangeTransportTests.cs | 248 +++++++++++++ tests/README.md | 2 +- 10 files changed, 707 insertions(+), 1 deletion(-) create mode 100644 src/clients/ioxide.redis.stackexchange/IoxideRedis.cs create mode 100644 src/clients/ioxide.redis.stackexchange/IoxideTransport.cs create mode 100644 src/clients/ioxide.redis.stackexchange/IoxideTunnel.cs create mode 100644 src/clients/ioxide.redis.stackexchange/ioxide.redis.stackexchange.csproj create mode 100644 tests/Ioxide.Tests.Redis/StackExchangeTransportTests.cs diff --git a/README.md b/README.md index a8d786bb..99f08e1d 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,7 @@ 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.redis.stackexchange` | Experimental, not published: StackExchange.Redis's whole API with its connections on a reactor's ring, through StackExchange.Redis's experimental transport seam. | | [`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/ioxide.slnx b/ioxide.slnx index 676c5768..9a59cecc 100644 --- a/ioxide.slnx +++ b/ioxide.slnx @@ -18,6 +18,7 @@ + diff --git a/src/clients/ioxide.redis.stackexchange/IoxideRedis.cs b/src/clients/ioxide.redis.stackexchange/IoxideRedis.cs new file mode 100644 index 00000000..cc380ac3 --- /dev/null +++ b/src/clients/ioxide.redis.stackexchange/IoxideRedis.cs @@ -0,0 +1,24 @@ +using StackExchange.Redis; + +namespace ioxide.redis.stackexchange; + +/// Multiplexers whose connections live on a reactor's ring. +public static class IoxideRedis +{ + /// + /// Connect a multiplexer on 's ring: sets + /// to an and connects. Call it from OnStart and keep the task - + /// never wait on it there, since the connect completes on the loop that starts after + /// OnStart returns. + /// + public static Task ConnectAsync(Reactor reactor, ConfigurationOptions options) + { + ArgumentNullException.ThrowIfNull(options); + options.Tunnel = new IoxideTunnel(reactor); + return ConnectionMultiplexer.ConnectAsync(options); + } + + /// + public static Task ConnectAsync(Reactor reactor, string configuration) + => ConnectAsync(reactor, ConfigurationOptions.Parse(configuration)); +} diff --git a/src/clients/ioxide.redis.stackexchange/IoxideTransport.cs b/src/clients/ioxide.redis.stackexchange/IoxideTransport.cs new file mode 100644 index 00000000..d3349cbc --- /dev/null +++ b/src/clients/ioxide.redis.stackexchange/IoxideTransport.cs @@ -0,0 +1,346 @@ +using System.Net; +using System.Runtime.CompilerServices; +using System.Runtime.InteropServices; +using RESPite.Transports; + +namespace ioxide.redis.stackexchange; + +/// +/// One StackExchange.Redis connection as a socket on a reactor's ring. Receives are ring operations +/// whose completions push the bytes into StackExchange.Redis on the reactor's thread, where it parses +/// them and completes the commands; each loop turn that delivered anything ends with one +/// , posted to run after the turn's completions. Writers +/// stage from any thread (StackExchange.Redis keeps to one at a time) and hands +/// the batch to the ring as one send. +/// +internal sealed class IoxideTransport : DuplexTransport +{ + private const int InitialOutbound = 16 * 1024; + + private readonly Reactor _reactor; + private readonly RingSocket _socket; + private readonly byte[] _received; + + // The send side. The writer owns _stage; a send in flight owns _wire; bytes flushed while it is + // in flight wait in _queued and go next. Only the swaps between them take the lock. + private readonly object _gate = new(); + private Outbound _stage = new(), _wire = new(), _queued = new(); + private bool _sending; + private int _sent; + + private TransportReceiver? _receiver; + private bool _batchEndPosted; // reactor thread only + private int _closing; + private Exception? _fault; + + // The transport's own reference, the receive loop's and a send's: the socket is disposed, and the + // receiver told, when the last of them lets go - never under an operation still in flight. + private int _refs = 1; + private readonly TaskCompletionSource _closed = new(TaskCreationOptions.RunContinuationsAsynchronously); + + private IoxideTransport(Reactor reactor, RingSocket socket, int receiveBufferSize) + { + _reactor = reactor; + _socket = socket; + _received = GC.AllocateUninitializedArray(receiveBufferSize, pinned: true); + } + + /// Resolve and connect to its addresses in turn, on + /// 's ring. + internal static async ValueTask ConnectAsync( + Reactor reactor, EndPoint endpoint, int receiveBufferSize, CancellationToken cancellationToken) + { + (IPAddress[] addresses, int port) = await ResolveAsync(endpoint, cancellationToken).ConfigureAwait(false); + string? failure = null; + + foreach (IPAddress address in addresses) + { + RingSocket socket = RingSocket.CreateTcp(reactor, address.AddressFamily); + Task connect = socket.ConnectAsync(address, (ushort)port).AsTask(); + int result; + try + { + result = await connect.WaitAsync(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // A ring connect cannot be cancelled: the socket goes once the connect completes. + _ = connect.ContinueWith(static (_, state) => ((RingSocket)state!).Dispose(), socket, + CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + throw; + } + + if (result == 0) + { + return new IoxideTransport(reactor, socket, receiveBufferSize); + } + + socket.Dispose(); + failure = $"connect to {new IPEndPoint(address, port)} failed: errno {-result}"; + } + + throw new IOException(failure ?? $"{endpoint} resolved to no address"); + } + + private static async ValueTask<(IPAddress[] Addresses, int Port)> ResolveAsync( + EndPoint endpoint, CancellationToken cancellationToken) + { + if (endpoint is IPEndPoint ip) + { + return ([ip.Address], ip.Port); + } + + if (endpoint is DnsEndPoint dns) + { + if (IPAddress.TryParse(dns.Host, out IPAddress? literal)) + { + return ([literal], dns.Port); + } + + // Off the reactor: a lookup blocks its thread. + return (await Dns.GetHostAddressesAsync(dns.Host, cancellationToken).ConfigureAwait(false), dns.Port); + } + + throw new NotSupportedException($"the ioxide transport connects over TCP only, not to a {endpoint.GetType().Name}"); + } + + // ---- inbound ---------------------------------------------------------------------------- + + public override void Start(TransportReceiver receiver) + { + if (Interlocked.CompareExchange(ref _receiver, receiver, null) is not null) + { + throw new InvalidOperationException("the transport already has its receiver"); + } + + if (!TryAddRef()) + { + return; // closed before it started: the receiver was never going to hear anything + } + + _ = ReceiveAsync(); + } + + private async Task ReceiveAsync() + { + Exception? fault = null; + try + { + nint buffer = Address(_received); + while (Volatile.Read(ref _closing) == 0) + { + // Resumes inline on the reactor thread when the receive completes. + int received = await _socket.RecvAsync(buffer, _received.Length); + if (received <= 0) + { + if (received < 0 && Volatile.Read(ref _closing) == 0) + { + fault = new IOException($"receive failed: errno {-received}"); + } + break; + } + + if (!_receiver!.OnReceived(_received.AsSpan(0, received))) + { + break; // the receiver asked to close + } + + if (!_batchEndPosted) + { + // A post from the reactor thread runs after this turn's completions, which is + // where the burst ends. + _batchEndPosted = true; + _reactor.ScheduleOnReactor(static state => ((IoxideTransport)state!).EndBatch(), this); + } + } + } + catch (Exception e) + { + fault = e; + } + + Close(fault); + Release(); + } + + private void EndBatch() + { + _batchEndPosted = false; + if (Volatile.Read(ref _closing) == 0) + { + _receiver?.OnBatchEnd(); + } + } + + // ---- outbound --------------------------------------------------------------------------- + + public override Memory GetMemory(int sizeHint = 0) => _stage.Reserve(sizeHint).AsMemory(_stage.Length); + + public override Span GetSpan(int sizeHint = 0) => _stage.Reserve(sizeHint).AsSpan(_stage.Length); + + public override void Advance(int count) => _stage.Length += count; + + public override bool Flush() + { + if (Volatile.Read(ref _closing) != 0) + { + _stage.Length = 0; + return false; + } + + if (_stage.Length == 0) + { + return true; + } + + lock (_gate) + { + if (_sending) + { + _queued.Append(_stage.Bytes.AsSpan(0, _stage.Length)); // goes when the send in flight is done + _stage.Length = 0; + return true; + } + + // The staged bytes become the send, and the writer stages into the idle buffer. + (_stage, _wire) = (_wire, _stage); + _stage.Length = 0; + _sent = 0; + _sending = true; + } + + if (!TryAddRef()) + { + lock (_gate) + { + _sending = false; + } + return false; + } + + _ = SendAsync(); + return true; + } + + private async Task SendAsync() + { + try + { + while (true) + { + byte[] bytes; + int from, to; + lock (_gate) + { + (bytes, from, to) = (_wire.Bytes, _sent, _wire.Length); + } + + int sent = await _socket.SendAsync(Address(bytes) + from, to - from); + if (sent <= 0) + { + Close(sent < 0 ? new IOException($"send failed: errno {-sent}") : null); + break; + } + + lock (_gate) + { + _sent += sent; + if (_sent < _wire.Length) + { + continue; // a partial send: the rest goes next + } + + if (_queued.Length == 0) + { + _sending = false; + break; + } + + (_wire, _queued) = (_queued, _wire); + _queued.Length = 0; + _sent = 0; + } + } + } + catch (Exception e) + { + Close(e); + } + + Release(); + } + + // ---- lifetime --------------------------------------------------------------------------- + + public override ValueTask DisposeAsync() + { + Close(null); + return new ValueTask(_closed.Task); + } + + private void Close(Exception? fault) + { + if (Interlocked.Exchange(ref _closing, 1) != 0) + { + return; + } + + _fault = fault; + Native.shutdown(_socket.Fd, Native.SHUT_RDWR); // ends the receive in flight, and a stuck send + Release(); // the transport's own reference + } + + private bool TryAddRef() + { + int refs; + do + { + refs = Volatile.Read(ref _refs); + if (refs == 0) + { + return false; + } + } + while (Interlocked.CompareExchange(ref _refs, refs + 1, refs) != refs); + return true; + } + + private void Release() + { + if (Interlocked.Decrement(ref _refs) != 0) + { + return; + } + + _socket.Dispose(); + Interlocked.Exchange(ref _receiver, null)?.OnClosed(_fault); + _closed.TrySetResult(); + } + + private static unsafe nint Address(byte[] pinned) => (nint)Unsafe.AsPointer(ref MemoryMarshal.GetArrayDataReference(pinned)); + + /// A growable outbound buffer on the pinned object heap, so the ring can read it in place. + private sealed class Outbound + { + public byte[] Bytes = GC.AllocateUninitializedArray(InitialOutbound, pinned: true); + public int Length; + + public byte[] Reserve(int sizeHint) + { + int needed = Length + Math.Max(sizeHint, 1); + if (needed > Bytes.Length) + { + byte[] grown = GC.AllocateUninitializedArray(Math.Max(needed, Bytes.Length * 2), pinned: true); + Bytes.AsSpan(0, Length).CopyTo(grown); + Bytes = grown; + } + return Bytes; + } + + public void Append(ReadOnlySpan bytes) + { + bytes.CopyTo(Reserve(bytes.Length).AsSpan(Length)); + Length += bytes.Length; + } + } +} diff --git a/src/clients/ioxide.redis.stackexchange/IoxideTunnel.cs b/src/clients/ioxide.redis.stackexchange/IoxideTunnel.cs new file mode 100644 index 00000000..0b199578 --- /dev/null +++ b/src/clients/ioxide.redis.stackexchange/IoxideTunnel.cs @@ -0,0 +1,48 @@ +using System.Net; +using RESPite.Transports; +using StackExchange.Redis; +using StackExchange.Redis.Configuration; + +namespace ioxide.redis.stackexchange; + +/// +/// A StackExchange.Redis that puts every connection of a multiplexer on one +/// reactor's ring: connect, sends and receives are io_uring operations on that reactor, and +/// StackExchange.Redis parses replies and completes commands on its thread. Give each reactor its own +/// multiplexer, made in OnStart, and a handler's commands never leave its thread. +/// +/// +/// TCP without TLS: a configuration that asks for TLS is refused when the connection is made, with a +/// message saying so. Built on StackExchange.Redis's experimental transport seam (SER009), which may +/// change in any release - this package pins the StackExchange.Redis version it was built against. +/// +public sealed class IoxideTunnel : Tunnel +{ + private readonly Reactor _reactor; + private readonly int _receiveBufferSize; + + /// The reactor whose ring carries the connections. + /// Bytes per connection for one receive; a reply larger than this + /// arrives in several. + public IoxideTunnel(Reactor reactor, int receiveBufferSize = 64 * 1024) + { + ArgumentNullException.ThrowIfNull(reactor); + ArgumentOutOfRangeException.ThrowIfLessThan(receiveBufferSize, 1024); + _reactor = reactor; + _receiveBufferSize = receiveBufferSize; + } + + public override async ValueTask ConnectTransportAsync( + EndPoint endpoint, ConnectionType connectionType, TlsOptions tls, CancellationToken cancellationToken) + { + if (tls.IsEnabled) + { + throw new NotSupportedException( + "the ioxide transport does not do TLS yet: connect without TLS, or leave Tunnel unset for StackExchange.Redis's own sockets"); + } + + return await IoxideTransport.ConnectAsync(_reactor, endpoint, _receiveBufferSize, cancellationToken).ConfigureAwait(false); + } + + public override string ToString() => "ioxide"; +} diff --git a/src/clients/ioxide.redis.stackexchange/ioxide.redis.stackexchange.csproj b/src/clients/ioxide.redis.stackexchange/ioxide.redis.stackexchange.csproj new file mode 100644 index 00000000..51f7eb8e --- /dev/null +++ b/src/clients/ioxide.redis.stackexchange/ioxide.redis.stackexchange.csproj @@ -0,0 +1,36 @@ + + + + net10.0;net11.0 + enable + enable + true + ioxide.redis.stackexchange + + + $(NoWarn);SER009 + + ioxide.redis.stackexchange + 0.16.281 + MDA2AV + Experimental: StackExchange.Redis over the ioxide io_uring runtime. A Tunnel whose transport is a socket on a reactor's ring, so a multiplexer's connections connect, send and receive on that reactor and complete commands on its thread. Built on StackExchange.Redis's experimental transport seam (SER009) and pinned to one StackExchange.Redis version. + MIT + https://mda2av.github.io/ioxide/learn/clients.html + https://github.com/MDA2AV/ioxide + git + io_uring;redis;stackexchange.redis;client;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..c890fc48 100644 --- a/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj +++ b/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj @@ -13,6 +13,7 @@ + diff --git a/tests/Ioxide.Tests.Redis/Program.cs b/tests/Ioxide.Tests.Redis/Program.cs index 727bcdcb..cd55c091 100644 --- a/tests/Ioxide.Tests.Redis/Program.cs +++ b/tests/Ioxide.Tests.Redis/Program.cs @@ -18,6 +18,7 @@ private static int Main() Console.WriteLine($"redis {(up ? "up" : "down")} ({redis.Host}:{redis.Port})\n"); RedisTests.Register(runner, redis, up); + StackExchangeTransportTests.Register(runner, redis, up); return runner.Summary(); } } diff --git a/tests/Ioxide.Tests.Redis/StackExchangeTransportTests.cs b/tests/Ioxide.Tests.Redis/StackExchangeTransportTests.cs new file mode 100644 index 00000000..6f81956b --- /dev/null +++ b/tests/Ioxide.Tests.Redis/StackExchangeTransportTests.cs @@ -0,0 +1,248 @@ +using System.Text; +using ioxide; +using ioxide.redis.stackexchange; +using StackExchange.Redis; + +namespace Ioxide.Tests; + +/// +/// StackExchange.Redis over the ioxide transport (): each test's handler +/// drives a multiplexer made on its reactor and answers with what it saw. +/// +internal static class StackExchangeTransportTests +{ + public static void Register(Runner runner, (string Host, int Port) redis, bool redisUp) + { + string run = Guid.NewGuid().ToString("N")[..8]; // keys of their own on a shared server + + runner.Test("redis over SE.Redis: SET then GET on the ring, completed on the reactor's thread", () => + { + Assert.Equal("hello reactor reactor", Get(redis, run, "/get")); + }, skip: !redisUp); + + runner.Test("redis over SE.Redis: 1000 concurrent INCRs - flushes queued behind a send in flight", () => + { + Assert.Equal("1000", Get(redis, run, "/burst")); + }, skip: !redisUp); + + runner.Test("redis over SE.Redis: a 1 MiB value arrives across several receives", () => + { + Assert.Equal((1 << 20).ToString(), Get(redis, run, "/big")); + }, skip: !redisUp); + + runner.Test("redis over SE.Redis: pub/sub through the transport", () => + { + Assert.Equal("ping", Get(redis, run, "/pubsub")); + }, skip: !redisUp); + + runner.Test("redis over SE.Redis: a connection the server kills comes back through the tunnel", () => + { + Assert.Equal("killed 1, then hello", Get(redis, run, "/reconnect")); + }, skip: !redisUp); + + // StackExchange.Redis 3.4.0's RecordConnectionFailed reports a failure only for a connection + // with a Stream (PhysicalConnection.cs:571, `_ioStream is not null`), which a transport never + // has: a command caught by the close is cancelled instead, and no ConnectionFailed is raised. + // On its own sockets the same command fails with RedisConnectionException(SocketClosed). + runner.Pending("redis over SE.Redis: a command caught by a server-side close fails with the connection's error", () => + { + Assert.Equal("RedisConnectionException", Get(redis, run, "/stranded")); + }, because: "StackExchange.Redis reports transport closes only for Stream-backed connections"); + + runner.Test("redis over SE.Redis: TLS is refused with the transport's reason, not a timeout", () => + { + string body = Get(redis, run, "/tls"); + Assert.True(body.Contains("does not do TLS yet"), $"the connect log should carry the refusal: {body}"); + }, skip: !redisUp); + } + + private static string Get((string Host, int Port) redis, string run, string path) + { + int port = TestServer.Start((r, conn) => Handle(r, conn, redis, run), + r => r.AddService(IoxideRedis.ConnectAsync(r, Options(redis)))); + (int status, string body) = Client.Get(port, path); + Assert.Equal(200, status); + return body; + } + + private static ConfigurationOptions Options((string Host, int Port) redis) + { + var options = new ConfigurationOptions + { + EndPoints = { { redis.Host, redis.Port } }, + AllowAdmin = true, // CLIENT KILL, for the reconnect test + ConnectTimeout = 5000, + }; + options.Password = Environment.GetEnvironmentVariable("EXAMPLES_REDIS_PASSWORD"); + return options; + } + + private static async Task Handle(Reactor r, TcpConnection conn, (string Host, int Port) redis, string run) + { + try + { + RecvSnapshot snapshot = await conn.ReadAsync(); + string path = Wire.ReadPath(conn, snapshot); + string body; + try + { + body = await Answer(r, path, redis, run); + } + catch (Exception e) + { + body = $"{e.GetType().Name}: {e.Message}"; + } + + Wire.Write(conn, 200, body); + await conn.FlushAsync(); + } + finally + { + conn.DecRef(); + } + } + + private static async Task Answer(Reactor r, string path, (string Host, int Port) redis, string run) + { + int reactorThread = Environment.CurrentManagedThreadId; + string where() => Environment.CurrentManagedThreadId == reactorThread ? "reactor" : "elsewhere"; + + if (path == "/tls") + { + ConfigurationOptions options = Options(redis); + options.Ssl = true; + options.ConnectTimeout = 1500; + options.Tunnel = new IoxideTunnel(r); + var written = new StringWriter(); + TextWriter log = TextWriter.Synchronized(written); // written from StackExchange.Redis's threads + try + { + using ConnectionMultiplexer refused = await ConnectionMultiplexer.ConnectAsync(options, log); + return "connected"; + } + catch (RedisConnectionException e) + { + return $"{written} | {e.Message}"; + } + } + + ConnectionMultiplexer mux = await r.GetService>(); + IDatabase db = mux.GetDatabase(); + string key = $"se:{run}:{path.Trim('/')}"; + + switch (path) + { + case "/get": + { + await db.StringSetAsync(key, "hello", TimeSpan.FromMinutes(1)); + // Which thread StackExchange.Redis completes the command on: the transport pushes + // replies on the reactor, so it should be the reactor's. + // StackExchange.Redis completes its tasks with asynchronous continuations, so the + // await posts to the context it captured - from the thread that completed it. + var recorder = new PostRecorder(SynchronizationContext.Current!); + SynchronizationContext.SetSynchronizationContext(recorder); + RedisValue value; + try + { + value = await db.StringGetAsync(key); + } + finally + { + SynchronizationContext.SetSynchronizationContext(recorder.Inner); + } + bool onReactor = recorder.PostedFrom is -1 || recorder.PostedFrom == reactorThread; // -1: resumed inline, so on the reactor + return $"{value} {(onReactor ? "reactor" : "elsewhere")} {where()}"; + } + + case "/burst": + { + await db.KeyDeleteAsync(key); + var increments = new Task[1000]; + for (int i = 0; i < increments.Length; i++) + { + increments[i] = db.StringIncrementAsync(key); + } + await Task.WhenAll(increments); + RedisValue final = await db.StringGetAsync(key); + await db.KeyDeleteAsync(key); + return final.ToString(); + } + + case "/big": + { + byte[] payload = new byte[1 << 20]; + Random.Shared.NextBytes(payload); + await db.StringSetAsync(key, payload, TimeSpan.FromMinutes(1)); + byte[]? back = await db.StringGetAsync(key); + await db.KeyDeleteAsync(key); + return back is not null && back.AsSpan().SequenceEqual(payload) ? back.Length.ToString() : "mismatch"; + } + + case "/pubsub": + { + ISubscriber subscriber = mux.GetSubscriber(); + var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + RedisChannel channel = RedisChannel.Literal(key); + await subscriber.SubscribeAsync(channel, (_, message) => received.TrySetResult(message.ToString())); + await subscriber.PublishAsync(channel, "ping"); + string message = await received.Task; // the runner's timeout bounds a message that never comes + await subscriber.UnsubscribeAsync(channel); + return message; + } + + case "/reconnect": + { + await db.StringSetAsync(key, "hello", TimeSpan.FromMinutes(1)); + long id = (long)await db.ExecuteAsync("CLIENT", "ID"); + var restored = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + mux.ConnectionRestored += (_, e) => + { + if (e.ConnectionType == ConnectionType.Interactive) restored.TrySetResult(); + }; + + // The interactive connection kills itself; StackExchange.Redis reconnects through + // the tunnel, and the next command goes on the new connection. + long killed = (long)await db.ExecuteAsync("CLIENT", "KILL", "ID", id, "SKIPME", "no"); + await restored.Task; + RedisValue value = await db.StringGetAsync(key); + await db.KeyDeleteAsync(key); + return $"killed {killed}, then {value}"; + } + + case "/stranded": + { + long id = (long)await db.ExecuteAsync("CLIENT", "ID"); + await db.ExecuteAsync("CLIENT", "KILL", "ID", id, "SKIPME", "no"); + try + { + await db.StringGetAsync(key); // written behind the kill, as the server closes + return "answered"; + } + catch (Exception e) + { + return e.GetType().Name; + } + } + + default: + return "unknown path"; + } + } + + /// Forwards to the reactor's context, noting which thread posted. + private sealed class PostRecorder(SynchronizationContext inner) : SynchronizationContext + { + public int PostedFrom = -1; + public SynchronizationContext Inner => inner; + + public override void Post(SendOrPostCallback callback, object? state) + { + PostedFrom = Environment.CurrentManagedThreadId; + inner.Post(callback, state); + } + + public override void Send(SendOrPostCallback callback, object? state) => inner.Send(callback, state); + + public override SynchronizationContext CreateCopy() => this; + } +} diff --git a/tests/README.md b/tests/README.md index 576eadc9..ec3f7803 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` | The Redis client, and StackExchange.Redis over the ioxide transport | 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 | From 5a8003c52a51c72db01b7f6a33e1240bf37219e2 Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Mon, 12 Oct 2026 15:06:28 +0100 Subject: [PATCH 2/2] samples: Clients/RedisStackExchange, StackExchange.Redis with its connections on the reactor's ring --- ...ayground.Clients.RedisStackExchange.csproj | 19 ++ .../Clients/RedisStackExchange/Program.cs | 215 ++++++++++++++++++ Playground/README.md | 1 + docs/assets/style.css | 3 + docs/index.html | 195 ++++++++++++++++ ioxide.slnx | 1 + scripts/gen-docs-panes.py | 10 + 7 files changed, 444 insertions(+) create mode 100644 Playground/Clients/RedisStackExchange/Playground.Clients.RedisStackExchange.csproj create mode 100644 Playground/Clients/RedisStackExchange/Program.cs diff --git a/Playground/Clients/RedisStackExchange/Playground.Clients.RedisStackExchange.csproj b/Playground/Clients/RedisStackExchange/Playground.Clients.RedisStackExchange.csproj new file mode 100644 index 00000000..489d034e --- /dev/null +++ b/Playground/Clients/RedisStackExchange/Playground.Clients.RedisStackExchange.csproj @@ -0,0 +1,19 @@ + + + + Exe + net11.0 + enable + enable + true + Playground.Clients.RedisStackExchange + Playground.Clients.RedisStackExchange + + + + + + + + + diff --git a/Playground/Clients/RedisStackExchange/Program.cs b/Playground/Clients/RedisStackExchange/Program.cs new file mode 100644 index 00000000..40bb7d77 --- /dev/null +++ b/Playground/Clients/RedisStackExchange/Program.cs @@ -0,0 +1,215 @@ +using System.Text; +using ioxide; +using ioxide.redis.stackexchange; +using ioxide.utils; +using Playground.Shared; +using StackExchange.Redis; + +// ───────────────────────────────────────────────────────────────────────────────────────────── +// redis via StackExchange.Redis - the whole StackExchange.Redis API, with its connections on +// the reactor's ring. IoxideTunnel hands the multiplexer a transport whose connect, sends and +// receives are io_uring operations on this reactor; StackExchange.Redis parses the replies and +// completes the commands on this reactor's thread, so a handler's await never leaves it. +// +// docker run --rm -p 6379:6379 redis # or point PLAYGROUND_REDIS_* elsewhere +// dotnet run -c Release --project Playground/Clients/RedisStackExchange +// +// curl http://127.0.0.1:8080/ # GET hot path (the wrk-able route) +// curl http://127.0.0.1:8080/incr # INCR +// curl http://127.0.0.1:8080/cache/user42 # cache-aside: GET, miss -> SET with expiry +// +// One multiplexer per reactor, made in OnStart and awaited by that reactor's handlers. Async +// APIs only: a synchronous call (StringGet rather than StringGetAsync) on a reactor waits for a +// completion only that reactor can run. No TLS through the tunnel yet. +// +// Experimental: built on StackExchange.Redis's transport seam (SER009), which may change in any +// StackExchange.Redis release. Compare with Clients/Redis, ioxide's own client. +// +// Needs: ioxide.redis.stackexchange +// ───────────────────────────────────────────────────────────────────────────────────────────── + +// ── 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; 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; a null password means no AUTH. No pool size: a multiplexer +// pipelines every command over one connection. +string redisHost = "127.0.0.1"; +ushort redisPort = 6379; +string? redisPassword = null; + +// "ioxide" puts the connections on the ring; "sockets" leaves StackExchange.Redis on its own +// sockets and threads, everything else the same - the difference this sample exists to show. +string transport = "ioxide"; + +Env.Override(ref redisHost, "PLAYGROUND_REDIS_HOST"); +Env.Override(ref redisPort, "PLAYGROUND_REDIS_PORT"); +Env.OverrideOptional(ref redisPassword, "PLAYGROUND_REDIS_PASSWORD"); +Env.Override(ref transport, "PLAYGROUND_REDIS_TRANSPORT"); +// ───────────────────────────────────────────────────────────────────────────────────────────── + +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 threads = new Thread[config.ReactorCount]; + +for (int i = 0; i < threads.Length; i++) +{ + var reactor = new Reactor(i, config); + bool seed = i == 0; + + // A multiplexer per reactor, its connections on THIS reactor's ring. Keep the task: the + // connect completes on the loop that starts after OnStart returns, so never wait on it here. + reactor.OnStart = r => + { + var options = new ConfigurationOptions + { + EndPoints = { { redisHost, redisPort } }, + Password = redisPassword, + ConnectTimeout = 5000, + }; + Task redis = transport == "ioxide" + ? IoxideRedis.ConnectAsync(r, options) + : ConnectionMultiplexer.ConnectAsync(options); + r.AddService(redis); + if (seed) + { + _ = SeedAsync(redis); // the "/" route reads this key + } + }; + + reactor.TcpHandle = async (r, conn) => + { + IDatabase db = (await r.GetService>()).GetDatabase(); + + try + { + while (true) + { + RecvSnapshot snapshot = await conn.ReadAsync(); + string path = ReadPath(conn, snapshot); + + try + { + string body = path switch + { + var p when p.StartsWith("/cache/", StringComparison.Ordinal) + => await CacheAside(db, p["/cache/".Length..]), + "/incr" => $"counter = {await db.StringIncrementAsync("play:counter")}", + // The hot path: a single GET per request. wrk this one. + _ => $"redis = {await db.StringGetAsync("play:bench")}", + }; + WriteText(conn, "200 OK", body); + } + catch (RedisException e) + { + 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 via SE.Redis] {config.ReactorCount} reactors on :{config.Tcp!.Port} " + + $"-> {redisHost}:{redisPort}, transport: {transport}, one multiplexer per reactor"); + +foreach (Thread thread in threads) +{ + thread.Join(); +} + +static async Task SeedAsync(Task redis) +{ + try + { + await (await redis).GetDatabase().StringSetAsync("play:bench", "42"); + } + catch (Exception e) + { + Console.Error.WriteLine($"[redis via SE.Redis] seed failed: {e.Message}"); + } +} + +static async Task CacheAside(IDatabase db, string key) +{ + key = "play:cache:" + key; + RedisValue cached = await db.StringGetAsync(key); + if (cached.HasValue) + { + return $"hit {key} = {cached}"; + } + + string value = $"value-{key.Length}"; // stands in for the expensive computation + await db.StringSetAsync(key, value, TimeSpan.FromSeconds(60)); + return $"miss {key} = {value} (cached 60s)"; +} + +// "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 lineEnd = request.IndexOfAny((byte)'\r', (byte)'\n'); + ReadOnlySpan line = lineEnd < 0 ? request : request[..lineEnd]; + int first = line.IndexOf((byte)' '); + int last = line.LastIndexOf((byte)' '); + if (first >= 0 && last > first + 1) + { + path = Encoding.ASCII.GetString(line[(first + 1)..last]); + } + 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: {Encoding.UTF8.GetByteCount(body)}\r\n\r\n{body}")); +} diff --git a/Playground/README.md b/Playground/README.md index 8a6e56ac..895256e3 100644 --- a/Playground/README.md +++ b/Playground/README.md @@ -126,6 +126,7 @@ reference implementation. QUIC itself is always ngtcp2 + picotls, bundled as one | [`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.RedisStackExchange`](Clients/RedisStackExchange/Program.cs) | 215 | StackExchange.Redis with its connections on the reactor's ring: a multiplexer per reactor through `IoxideTunnel`, and a knob to run it on StackExchange.Redis's own sockets instead. Experimental. | `ioxide.redis.stackexchange` | | [`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/docs/assets/style.css b/docs/assets/style.css index 492e960e..ce09c1ad 100644 --- a/docs/assets/style.css +++ b/docs/assets/style.css @@ -262,6 +262,7 @@ nav.top .links a.gh svg { display: block; } #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-redisse:checked ~ .ex-menu label[for="tab-redisse"], #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"], @@ -409,6 +410,7 @@ nav.top .links a.gh svg { display: block; } #tab-http:checked ~ .pane-http { display: block; } #tab-pg:checked ~ .pane-pg { display: block; } #tab-redis:checked ~ .pane-redis { display: block; } +#tab-redisse:checked ~ .pane-redisse { display: block; } #tab-files:checked ~ .pane-files { display: block; } #tab-tls:checked ~ .pane-tls { display: block; } #tab-tlshybrid:checked ~ .pane-tlshybrid { display: block; } @@ -561,6 +563,7 @@ nav.top .links a.gh svg { display: block; } #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-redisse:checked ~ .ex-menu label[for="tab-redisse"], #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/index.html b/docs/index.html index 1e3652a2..bad3d873 100644 --- a/docs/index.html +++ b/docs/index.html @@ -64,6 +64,7 @@ + @@ -167,6 +168,7 @@ clients + @@ -8027,6 +8029,199 @@

Redis

$"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.

+
+
+

Redis · StackExchange.Redis

+ ioxide + ioxide.redis.stackexchange +
+
// dotnet add package ioxide
+// dotnet add package ioxide.redis.stackexchange
+//   curl http://127.0.0.1:8080/
+//   PLAYGROUND_REDIS_TRANSPORT=sockets dotnet run -c Release --project Playground/Clients/RedisStackExchange
+
+using System.Text;
+using ioxide;
+using ioxide.redis.stackexchange;
+using ioxide.utils;
+using StackExchange.Redis;
+
+// ── 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; a null password means no AUTH. No pool size: a multiplexer
+// pipelines every command over one connection.
+string  redisHost     = "127.0.0.1";
+ushort  redisPort     = 6379;
+string? redisPassword = null;
+
+// "ioxide" puts the connections on the ring; "sockets" leaves StackExchange.Redis on its own
+// sockets and threads, everything else the same - the difference this sample exists to show.
+string transport = "ioxide";
+// ─────────────────────────────────────────────────────────────────────────────────────────────
+
+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 threads = new Thread[config.ReactorCount];
+
+for (int i = 0; i < threads.Length; i++)
+{
+    var reactor = new Reactor(i, config);
+    bool seed = i == 0;
+
+    // A multiplexer per reactor, its connections on THIS reactor's ring. Keep the task: the
+    // connect completes on the loop that starts after OnStart returns, so never wait on it here.
+    reactor.OnStart = r =>
+    {
+        var options = new ConfigurationOptions
+        {
+            EndPoints      = { { redisHost, redisPort } },
+            Password       = redisPassword,
+            ConnectTimeout = 5000,
+        };
+        Task<ConnectionMultiplexer> redis = transport == "ioxide"
+            ? IoxideRedis.ConnectAsync(r, options)
+            : ConnectionMultiplexer.ConnectAsync(options);
+        r.AddService(redis);
+        if (seed)
+        {
+            _ = SeedAsync(redis);   // the "/" route reads this key
+        }
+    };
+
+    reactor.TcpHandle = async (r, conn) =>
+    {
+        IDatabase db = (await r.GetService<Task<ConnectionMultiplexer>>()).GetDatabase();
+
+        try
+        {
+            while (true)
+            {
+                RecvSnapshot snapshot = await conn.ReadAsync();
+                string path = ReadPath(conn, snapshot);
+
+                try
+                {
+                    string body = path switch
+                    {
+                        var p when p.StartsWith("/cache/", StringComparison.Ordinal)
+                                => await CacheAside(db, p["/cache/".Length..]),
+                        "/incr" => $"counter = {await db.StringIncrementAsync("play:counter")}",
+                        // The hot path: a single GET per request. wrk this one.
+                        _       => $"redis = {await db.StringGetAsync("play:bench")}",
+                    };
+                    WriteText(conn, "200 OK", body);
+                }
+                catch (RedisException e)
+                {
+                    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 via SE.Redis] {config.ReactorCount} reactors on :{config.Tcp!.Port} "
+                + $"-> {redisHost}:{redisPort}, transport: {transport}, one multiplexer per reactor");
+
+foreach (Thread thread in threads)
+{
+    thread.Join();
+}
+
+static async Task SeedAsync(Task<ConnectionMultiplexer> redis)
+{
+    try
+    {
+        await (await redis).GetDatabase().StringSetAsync("play:bench", "42");
+    }
+    catch (Exception e)
+    {
+        Console.Error.WriteLine($"[redis via SE.Redis] seed failed: {e.Message}");
+    }
+}
+
+static async Task<string> CacheAside(IDatabase db, string key)
+{
+    key = "play:cache:" + key;
+    RedisValue cached = await db.StringGetAsync(key);
+    if (cached.HasValue)
+    {
+        return $"hit {key} = {cached}";
+    }
+
+    string value = $"value-{key.Length}";   // stands in for the expensive computation
+    await db.StringSetAsync(key, value, TimeSpan.FromSeconds(60));
+    return $"miss {key} = {value} (cached 60s)";
+}
+
+// "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 lineEnd = request.IndexOfAny((byte)'\r', (byte)'\n');
+            ReadOnlySpan<byte> line = lineEnd < 0 ? request : request[..lineEnd];
+            int first = line.IndexOf((byte)' ');
+            int last = line.LastIndexOf((byte)' ');
+            if (first >= 0 && last > first + 1)
+            {
+                path = Encoding.ASCII.GetString(line[(first + 1)..last]);
+            }
+            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: {Encoding.UTF8.GetByteCount(body)}\r\n\r\n{body}"));
+}
+

StackExchange.Redis's whole API with its connections on the reactor's ring. IoxideTunnel hands the multiplexer a transport whose connect, sends and receives are io_uring operations on the reactor, and StackExchange.Redis parses the replies and completes each command on that reactor's thread - one multiplexer per reactor, so a handler's await never leaves it. Built on StackExchange.Redis's experimental transport seam, and pinned to the version it was built against. Compare , ioxide's own client.

+

Client · static files

diff --git a/ioxide.slnx b/ioxide.slnx index 9a59cecc..caae58d4 100644 --- a/ioxide.slnx +++ b/ioxide.slnx @@ -104,6 +104,7 @@ + diff --git a/scripts/gen-docs-panes.py b/scripts/gen-docs-panes.py index fd1e3e5e..9c89c9fb 100644 --- a/scripts/gen-docs-panes.py +++ b/scripts/gen-docs-panes.py @@ -507,6 +507,16 @@ "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."), + "redisse": ( + "Clients/RedisStackExchange", "Redis · StackExchange.Redis", "ioxide + ioxide.redis.stackexchange", + ["curl http://127.0.0.1:8080/", + "PLAYGROUND_REDIS_TRANSPORT=sockets dotnet run -c Release --project Playground/Clients/RedisStackExchange"], + "StackExchange.Redis's whole API with its connections on the reactor's ring. IoxideTunnel " + "hands the multiplexer a transport whose connect, sends and receives are io_uring operations on the " + "reactor, and StackExchange.Redis parses the replies and completes each command on that reactor's " + "thread - one multiplexer per reactor, so a handler's await never leaves it. Built on " + "StackExchange.Redis's experimental transport seam, and pinned to the version it was built " + "against. Compare , ioxide's own client."), "qclient": ( "Clients/Quic", "QUIC · client", "ioxide + ioxide.ngtcp2", ["dotnet run -c Release --project Playground/Quic/Raw # something to talk to",