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) + { + } + } +}