From 0e68fe918c6a26ae160f1e53e909c7818e3b8767 Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Wed, 7 Oct 2026 21:45:29 +0100 Subject: [PATCH] redis: the send buffer reclaims what it has sent, even when it never drains fully The sender sent [_sendOffset, _sendEnd) and moved both back to 0 only after a send completed with nothing appended behind it. A connection that always had a command appended while a send was in flight never reached that point, so _sendEnd only grew: after 512 KiB of traffic every new command was refused with "pipelined send buffer full", with one command actually waiting to go. pg had the same loop and fixed it in ce82399. The sender now sends a snapshot of the staged bytes and, once it is out, moves whatever was appended meanwhile to the front, as PgConnection does. No new state. New test, RedisSendBufferTests, drives a RedisConnection through a scripted IRingHost, so the interleaving is exact: each PING is appended before the previous send completes, and every reply is delivered. It pushes 4 x 512 KiB through one connection and checks every command is answered and the bytes sent are exactly the PINGs in order. Confirmed both ways: on f298af0 command 37450 is refused with 524,272 bytes sent; with the fix all 149,798 go through. No redis needed. --- src/clients/ioxide.redis/RedisConnection.cs | 32 +++-- tests/Ioxide.Tests.Redis/Program.cs | 1 + .../RedisSendBufferTests.cs | 118 ++++++++++++++++++ 3 files changed, 144 insertions(+), 7 deletions(-) create mode 100644 tests/Ioxide.Tests.Redis/RedisSendBufferTests.cs diff --git a/src/clients/ioxide.redis/RedisConnection.cs b/src/clients/ioxide.redis/RedisConnection.cs index 34799ea3..018e5980 100644 --- a/src/clients/ioxide.redis/RedisConnection.cs +++ b/src/clients/ioxide.redis/RedisConnection.cs @@ -182,27 +182,39 @@ private unsafe void AppendCommand(ReadOnlySpan nameToken, RedisArg[] args) _sendEnd += written; } + // Reclaims the sent prefix after every snapshot: a command appended during each send means it never drains fully. private async Task SenderLoopAsync() { try { while (_sendOffset < _sendEnd) { - int n = await _socket.SendAsync(_send + _sendOffset, _sendEnd - _sendOffset); - if (n <= 0) + int end = _sendEnd; // snapshot; more may be appended (at _sendEnd) while we send + while (_sendOffset < end) { - IsBroken = true; - throw RedisException.Transport("send", n); + int n = await _socket.SendAsync(_send + _sendOffset, end - _sendOffset); + if (n <= 0) + { + IsBroken = true; + throw RedisException.Transport("send", n); + } + _sendOffset += n; } - _sendOffset += n; + + // Nothing is in flight, so slide what was appended during the send to the front. + int tail = _sendEnd - end; + if (tail > 0) + { + CompactSend(end, tail); + } + _sendOffset = 0; + _sendEnd = tail; } // 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) @@ -213,6 +225,12 @@ private async Task SenderLoopAsync() } } + // Move [from, from+length) to the front of the send buffer; CopyTo is memmove-safe if they overlap. + private unsafe void CompactSend(int from, int length) + { + new Span((void*)(_send + from), length).CopyTo(new Span((void*)_send, length)); + } + private async Task ReaderLoopAsync() { try diff --git a/tests/Ioxide.Tests.Redis/Program.cs b/tests/Ioxide.Tests.Redis/Program.cs index 727bcdcb..af808ce3 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); + RedisSendBufferTests.Register(runner); return runner.Summary(); } } diff --git a/tests/Ioxide.Tests.Redis/RedisSendBufferTests.cs b/tests/Ioxide.Tests.Redis/RedisSendBufferTests.cs new file mode 100644 index 00000000..f763d981 --- /dev/null +++ b/tests/Ioxide.Tests.Redis/RedisSendBufferTests.cs @@ -0,0 +1,118 @@ +using System.Text; +using ioxide; +using ioxide.redis; + +namespace Ioxide.Tests; + +/// +/// The pipelined send buffer, driven through a scripted ring so the order of appends and send +/// completions is exact rather than left to the network. +/// +internal static class RedisSendBufferTests +{ + private const int SendBufferSize = 512 * 1024; // RedisConnection's fixed send buffer + + public static void Register(Runner runner) + { + // A command is appended while every send is still in flight, so the buffer never empties out on its own. + runner.Test("redis: a connection whose sends never fully drain still reuses its send buffer", () => + { + var ring = new ScriptedRing(); + RedisConnection connection = RedisConnection.ConnectAsync(ring, new RedisOptions()).GetAwaiter().GetResult(); + + try + { + byte[] ping = Encoding.ASCII.GetBytes("*1\r\n$4\r\nPING\r\n"); + ValueTask previous = connection.ExecuteAsync("PING"); + int commands = 1; + + while (ring.Sent.Length < 4L * SendBufferSize) + { + ValueTask next = connection.ExecuteAsync("PING"); + commands++; + if (next.IsFaulted) + { + throw new Exception($"command {commands} was refused with {ring.Sent.Length} bytes sent: {Error(next)}"); + } + + ring.CompleteSend(); + ring.Reply("+PONG\r\n"u8); + ExpectPong(previous, commands - 1); + previous = next; + } + + ring.CompleteSend(); + ring.Reply("+PONG\r\n"u8); + ExpectPong(previous, commands); + + byte[] wire = ring.Sent.ToArray(); + Assert.Equal((long)commands * ping.Length, wire.LongLength); + for (int at = 0; at < wire.Length; at += ping.Length) + { + Assert.True(wire.AsSpan(at, ping.Length).SequenceEqual(ping), $"the bytes sent at {at} are not the next PING"); + } + } + finally + { + connection.Dispose(); + } + }); + } + + private static void ExpectPong(ValueTask command, int number) + { + if (!command.IsCompletedSuccessfully || command.Result.AsString() != "PONG") + { + throw new Exception($"command {number} was not answered PONG: {Error(command)}"); + } + } + + private static string Error(ValueTask command) + => command.IsFaulted ? command.AsTask().Exception!.InnerException!.Message : "still pending"; + + /// + /// A ring that completes a connect at once and holds every send and recv until the test + /// completes it. Completions run inline on the test thread, the way a reactor runs them. + /// + private sealed unsafe class ScriptedRing : IRingHost + { + private (nint Buffer, int Length, IRingCompletion Completion)? _send; + private (nint Buffer, int Length, IRingCompletion Completion)? _recv; + + /// Every byte the connection has sent, in order. + public MemoryStream Sent { get; } = new(); + + public void SubmitConnect(int fd, nint sockaddr, int sockaddrLen, IRingCompletion completion) => completion.Complete(0); + + public void SubmitSend(int fd, nint buffer, int length, IRingCompletion completion) => _send = (buffer, length, completion); + + public void SubmitRecv(int fd, nint buffer, int length, IRingCompletion completion) => _recv = (buffer, length, completion); + + /// Completes the send in flight in full; the connection may submit its next one before this returns. + public void CompleteSend() + { + (nint buffer, int length, IRingCompletion completion) = _send ?? throw new Exception("no send in flight"); + _send = null; + Sent.Write(new ReadOnlySpan((void*)buffer, length)); + completion.Complete(length); + } + + /// Delivers to the recv in flight. + public void Reply(ReadOnlySpan reply) + { + (nint buffer, int length, IRingCompletion completion) = _recv ?? throw new Exception("no recv in flight"); + _recv = null; + reply.CopyTo(new Span((void*)buffer, length)); + completion.Complete(reply.Length); + } + + public void SubmitRead(int fd, nint buffer, int length, long offset, IRingCompletion completion) => throw new NotSupportedException(); + + public void SubmitWrite(int fd, nint buffer, int length, long offset, IRingCompletion completion) => throw new NotSupportedException(); + + // The test owns the clock: a timeout is accepted and never expires. + public void SubmitTimeout(long nanoseconds, IRingCompletion completion) + { + } + } +}