diff --git a/src/clients/ioxide.pg/PgConnection.cs b/src/clients/ioxide.pg/PgConnection.cs index 0abca19f..d4496e7d 100644 --- a/src/clients/ioxide.pg/PgConnection.cs +++ b/src/clients/ioxide.pg/PgConnection.cs @@ -305,6 +305,11 @@ private async Task SenderLoopAsync() _sending = false; FailAll(ex); } + + if (IsBroken) + { + Dispose(); // a Dispose that found this send in flight left the teardown to here + } } // Move [from, from+length) to the front of the send buffer. CopyTo is memmove-safe for the @@ -387,6 +392,11 @@ private async Task ReaderLoopAsync() _reading = false; FailAll(ex is PgException p ? p : new PgException(ex.Message)); } + + if (IsBroken) + { + Dispose(); // a Dispose that found this recv in flight left the teardown to here + } } private void FailAll(Exception ex) @@ -400,7 +410,7 @@ private void FailAll(Exception ex) // Reactor-thread (pool ticker): if the oldest in-flight command is older than the timeout, tear // the connection down - fail its waiters with a diagnostic error and mark it broken. The pool then - // disposes it, and closing the fd cancels the stuck ring recv/send. Returns true if it timed out. + // disposes it, and shutting the socket down completes 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) @@ -677,6 +687,15 @@ private PgException ReadServerError(in Message message) public unsafe void Dispose() { + IsBroken = true; + + // close() would leave an armed op holding the socket open and using the buffers freed below. + if (_reading || _sending) + { + Native.shutdown(_socket.Fd, Native.SHUT_RDWR); + return; + } + _socket.Dispose(); if (_send != 0) diff --git a/src/clients/ioxide.pg/PgPool.cs b/src/clients/ioxide.pg/PgPool.cs index 7c108e28..5e022514 100644 --- a/src/clients/ioxide.pg/PgPool.cs +++ b/src/clients/ioxide.pg/PgPool.cs @@ -181,7 +181,7 @@ private void Sweep() { PgConnection c = _connections[i]; _connections.RemoveAt(i); - c.Dispose(); // closes the fd - unsticks the connection's stuck send/recv loops + c.Dispose(); // shuts the socket down - unsticks the connection's stuck send/recv loops } } } diff --git a/tests/Ioxide.Tests.Harness/TcpSockets.cs b/tests/Ioxide.Tests.Harness/TcpSockets.cs new file mode 100644 index 00000000..e1a55bdd --- /dev/null +++ b/tests/Ioxide.Tests.Harness/TcpSockets.cs @@ -0,0 +1,101 @@ +namespace Ioxide.Tests; + +/// This process's TCP connections from /proc/net/tcp, which unlike a descriptor count shows a socket outliving its close(). +public static class TcpSockets +{ + private const string Established = "01"; // the "st" column + + /// One IPv4 connection, named by both of its ends exactly as /proc/net/tcp prints them. + public readonly record struct Connection(string Local, string Remote); + + /// This process's IPv4 connections that are ESTABLISHED to . + public static HashSet EstablishedTo(int remotePort) + { + HashSet mine = OwnSocketInodes(); + var found = new HashSet(); + + foreach (string[] row in Rows()) + { + if (row[3] == Established && PortOf(row[2]) == remotePort && mine.Contains(row[9])) + { + found.Add(new Connection(row[1], row[2])); + } + } + + return found; + } + + /// Whether is still ESTABLISHED, whether or not a descriptor still holds it. + public static bool StillEstablished(Connection connection) + => Rows().Any(row => row[1] == connection.Local && row[2] == connection.Remote && row[3] == Established); + + /// Waits for connections to beyond , and returns what it found. + public static HashSet WaitForNew(int remotePort, HashSet earlier, int count, int timeoutMs = 10_000) + { + long deadlineMs = Environment.TickCount64 + timeoutMs; + + while (true) + { + HashSet fresh = EstablishedTo(remotePort); + fresh.ExceptWith(earlier); + + if (fresh.Count >= count || Environment.TickCount64 > deadlineMs) + { + return fresh; + } + + Thread.Sleep(20); + } + } + + /// Waits until any of stops being ESTABLISHED, and returns the ones that have. + public static List WaitForAnyToEnd(IReadOnlyCollection connections, int timeoutMs) + { + long deadlineMs = Environment.TickCount64 + timeoutMs; + + while (true) + { + List ended = connections.Where(c => !StillEstablished(c)).ToList(); + + if (ended.Count > 0 || Environment.TickCount64 > deadlineMs) + { + return ended; + } + + Thread.Sleep(20); + } + } + + // Columns: sl, local_address, rem_address, st, tx:rx queues, tr:when, retrnsmt, uid, timeout, inode. + private static IEnumerable Rows() + => File.ReadAllLines("/proc/net/tcp") + .Skip(1) + .Select(line => line.Split(' ', StringSplitOptions.RemoveEmptyEntries)) + .Where(row => row.Length > 9); + + private static int PortOf(string endpoint) => Convert.ToInt32(endpoint[(endpoint.IndexOf(':') + 1)..], 16); + + // A socket descriptor links to "socket:[inode]", the number in the inode column. + private static HashSet OwnSocketInodes() + { + var inodes = new HashSet(); + + foreach (string fd in Directory.EnumerateFileSystemEntries("/proc/self/fd")) + { + try + { + string? target = new FileInfo(fd).LinkTarget; + if (target is not null && target.StartsWith("socket:[", StringComparison.Ordinal)) + { + inodes.Add(target["socket:[".Length..^1]); + } + } + catch (IOException) + { + // Closed between the listing and the read. + } + } + + return inodes; + } +} diff --git a/tests/Ioxide.Tests.Pg/PgHandlers.cs b/tests/Ioxide.Tests.Pg/PgHandlers.cs index 153e9d5b..114cf0a5 100644 --- a/tests/Ioxide.Tests.Pg/PgHandlers.cs +++ b/tests/Ioxide.Tests.Pg/PgHandlers.cs @@ -35,6 +35,11 @@ public static async Task Pg(Reactor r, TcpConnection conn) await pool.QueryAsync("SELECT * FROM no_such_table"); Wire.Write(conn, 200, "unreachable"); } + else if (path.StartsWith("/sleep/", StringComparison.Ordinal) && int.TryParse(path["/sleep/".Length..], out int seconds)) + { + await pool.QueryAsync($"SELECT pg_sleep({seconds})"); + Wire.Write(conn, 200, "slept"); + } else { PgResult result = await pool.QueryAsync("SELECT 42"); @@ -43,7 +48,7 @@ public static async Task Pg(Reactor r, TcpConnection conn) } catch (PgException e) { - Wire.Write(conn, 500, e.SqlState ?? "error"); + Wire.Write(conn, 500, e.SqlState ?? e.Message); } await conn.FlushAsync(); diff --git a/tests/Ioxide.Tests.Pg/PgTests.cs b/tests/Ioxide.Tests.Pg/PgTests.cs index 772eb7a1..99c94502 100644 --- a/tests/Ioxide.Tests.Pg/PgTests.cs +++ b/tests/Ioxide.Tests.Pg/PgTests.cs @@ -59,6 +59,36 @@ public static void Register(Runner runner, (string Host, int Port) pg, bool pgUp (int status, _) = Client.Get(port, "/slow", timeoutMs: 8000); Assert.Equal(503, status); }, skip: !pgUp); + + // A socket outliving its close() is the visible half of a recv still armed into the freed buffer. + runner.Test("pg: a timed-out connection's socket ends at the timeout, not when the late reply comes", () => + { + HashSet earlier = TcpSockets.EstablishedTo(pg.Port); + int port = TestServer.Start(PgHandlers.Pg, r => PgPool.Start(r, PgOpts(pg, commandTimeoutMs: 1000))); + + (int okStatus, string okBody) = Client.Get(port, "/"); + Assert.Equal(200, okStatus); + Assert.Equal("42", okBody); + + HashSet pool = TcpSockets.WaitForNew(pg.Port, earlier, count: 2); + Assert.True(pool.Count == 2, $"expected the pool's 2 connections ESTABLISHED to :{pg.Port}, found {pool.Count}"); + + // The sleep outlasts the wait below, so its reply cannot be what ends the socket. + (int status, string body) = Client.Get(port, "/sleep/30", timeoutMs: 15_000); + Assert.Equal(500, status); + Assert.True(body.Contains("timed out"), $"expected the command timeout, got [{body}]"); + + List ended = TcpSockets.WaitForAnyToEnd(pool, timeoutMs: 10_000); + Assert.True(ended.Count == 1, + $"{ended.Count} of the pool's 2 connections ended within 10 s of the timeout; the timed-out one is still ESTABLISHED"); + + // Control: the connection that did not time out is untouched, and the pool still answers. + Assert.True(pool.Where(c => !ended.Contains(c)).All(TcpSockets.StillEstablished), + "the connection that did not time out ended too"); + (int afterStatus, string afterBody) = Client.Get(port, "/"); + Assert.Equal(200, afterStatus); + Assert.Equal("42", afterBody); + }, skip: !pgUp); } private static PgOptions PgOpts((string Host, int Port) pg, int commandTimeoutMs = 30_000) => new()