shared ring: a few connections parked in a flush can drain the ring and stop every recv on the reactor #526

using System.Net;
using System.Net.Sockets;
using System.Text;
using ioxide;
using ioxide.utils;

namespace Ioxide.Tests;

/// <summary>
/// Connections whose handlers are parked in a flush keep taking buffers for what their peers send, a whole
/// buffer per recv however few bytes it carries. With the shared ring, that is every connection's supply.
/// </summary>
internal static class SharedRingStarvationTests
{
    // Enough parked connections to take the ring without any one of them overflowing its 64-entry queue.
    private const int RingSlots = 256;
    private const int ParkedConnections = 5;
    private const int SendsEach = 60;

    public static void Register(Runner runner)
    {
        runner.Test("shared ring: connections parked in a flush leave another connection able to read", () =>
        {
            Assert.True(AnsweredWhileOthersAreParked(incremental: null),
                "the other connection's request was never answered: its recv is parked for a buffer nobody will return");
        });

        runner.Test("shared ring: control: with a ring per connection the same peers stop nobody", () =>
        {
            Assert.True(AnsweredWhileOthersAreParked(new IncrementalOptions { MaxConnections = 16 }),
                "the other connection's request was never answered");
        }, skip: !TestServer.KernelAtLeast(6, 12));
    }

    private static bool AnsweredWhileOthersAreParked(IncrementalOptions? incremental)
    {
        byte[] big = new byte[8 * 1024 * 1024];   // far past what a 4 KiB window and the socket buffers hold
        byte[] pong = "pong"u8.ToArray();

        int port = TestServer.StartConfigured(async (_, conn) =>
        {
            try
            {
                while (true)
                {
                    RecvSnapshot snapshot = await conn.ReadAsync();
                    while (conn.TryGetItem(snapshot, out SpscRecvRing.Item item))
                    {
                        if (item.HasBuffer)
                        {
                            conn.Write(item.AsSpan()[0] == (byte)'B' ? big : pong);
                            conn.ReturnBuffer(in item);
                        }
                    }

                    await conn.FlushAsync();
                    if (snapshot.IsClosed)
                    {
                        return;
                    }
                }
            }
            finally
            {
                conn.DecRef();
            }
        }, new ServerConfig
        {
            RecvBufferSize = 4096,
            RecvSlots = RingSlots,
            Incremental = incremental,
            Tcp = new TcpOptions { RecvQueueEntries = 64, ReadTimeoutMs = 0, SendTimeoutMs = 0 },
        }).Port;

        using Socket other = Connect(port);
        Assert.True(Asked(other), "the other connection was not answered even before the rest arrived");

        var parked = new List<Socket>();
        try
        {
            for (int i = 0; i < ParkedConnections; i++)
            {
                Socket socket = Connect(port, receiveBuffer: 4096);
                socket.Send("B"u8);   // and never reads the answer, so its handler parks in the flush
                parked.Add(socket);
            }
            Thread.Sleep(300);

            for (int round = 0; round < SendsEach; round++)
            {
                foreach (Socket socket in parked)
                {
                    socket.Send("x"u8);
                }

                Thread.Sleep(1);   // apart, so each is a recv of its own
            }

            return Asked(other);
        }
        finally
        {
            parked.ForEach(socket => socket.Dispose());
        }
    }

    private static bool Asked(Socket socket)
    {
        socket.Send("p"u8);

        var answer = new byte[4];
        try
        {
            for (int got = 0; got < answer.Length;)
            {
                int n = socket.Receive(answer, got, answer.Length - got, SocketFlags.None);
                if (n == 0)
                {
                    return false;
                }

                got += n;
            }
        }
        catch (SocketException e) when (e.SocketErrorCode == SocketError.TimedOut)
        {
            return false;
        }

        return Encoding.ASCII.GetString(answer) == "pong";
    }

    private static Socket Connect(int port, int? receiveBuffer = null)
    {
        var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp)
        {
            NoDelay = true,
            ReceiveTimeout = 5_000,
        };
        if (receiveBuffer is int bytes)
        {
            socket.ReceiveBufferSize = bytes;   // before the connect, so it is the window advertised
        }

        socket.Connect(IPAddress.Loopback, port);
        return socket;
    }
}