iouring: serverFDFor matches both ends of the dialed connection (Refs #1081)

internal/engine/iouring/fd_lifetime_close_test.go

@@ -271,13 +271,17 @@ func TestShutdownEndsOwedOpsBeforeClosing(t *testing.T) {
 		}
 		srv := -1
 		for end := time.Now().Add(time.Second); srv < 0 && time.Now().Before(end); {
-			srv = serverFDFor(c.LocalAddr().String())
+			srv = serverFDFor(c.LocalAddr().String(), c.RemoteAddr().String())
 		}
 		if srv < 0 {
 			t.Fatalf("conn %d: no server-side descriptor found", i)
 		}
+		target := fdTarget(srv)
+		if target == "" {
+			t.Fatalf("conn %d: server-side descriptor %d closed as it was found", i, srv)
+		}
 		servers = append(servers, srv)
-		targets = append(targets, fdTarget(srv))
+		targets = append(targets, target)
 	}
 	defer func() {
 		for _, c := range clients {

internal/engine/iouring/fd_probe_identity_1081_linux_test.go

@@ -0,0 +1,135 @@
+//go:build linux
+
+package iouring
+
+// celeris#1081: TestShutdownEndsOwedOpsBeforeClosing failed in CI with one
+// of its eight server-side descriptors "still open after the engine
+// stopped". The descriptor it checked was not the engine's. serverFDFor
+// took the first descriptor whose peer was the test client's address, and a
+// server-side socket an earlier test had left open names an earlier client's
+// port as its peer: connect(2) hands that port out again once that client is
+// gone, and the test's client can be given it. /proc/self/fd lists names in
+// lexical order, so that socket was found before the engine's whenever its
+// number sorted first. The test below forces the same collision
+// deterministically with a stray of the other kind (a client whose peer, a
+// listener's port, is then given to the test's client): what matters to
+// serverFDFor is only that the stray names the client's address as its peer,
+// at a number that sorts before the server side's.
+
+import (
+	"context"
+	"net"
+	"syscall"
+	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+// dupAbove moves c's descriptor to the lowest free number >= lowest and closes
+// c, returning the new number, which the test closes.
+func dupAbove(t *testing.T, c syscall.Conn, lowest int) int {
+	t.Helper()
+	rc, err := c.SyscallConn()
+	if err != nil {
+		t.Fatal(err)
+	}
+	nfd := -1
+	var derr error
+	if err := rc.Control(func(fd uintptr) { nfd, derr = unix.FcntlInt(fd, unix.F_DUPFD_CLOEXEC, lowest) }); err != nil {
+		t.Fatal(err)
+	}
+	if derr != nil {
+		t.Fatalf("F_DUPFD %d: %v", lowest, derr)
+	}
+	t.Cleanup(func() { _ = unix.Close(nfd) })
+	if cl, ok := c.(interface{ Close() error }); ok {
+		_ = cl.Close()
+	}
+	return nfd
+}
+
+func acceptTCP(t *testing.T, ln net.Listener) *net.TCPConn {
+	t.Helper()
+	c, err := ln.Accept()
+	if err != nil {
+		t.Fatal(err)
+	}
+	return c.(*net.TCPConn)
+}
+
+// TestServerFDForIsTheServerSideOfTheDialedConn1081: a server with two
+// connections, A and B, and a stray socket whose peer is B's client address.
+// The stray sorts first ("1xx"), A's server side next ("2xx"), B's last
+// ("9xxx"). serverFDFor for B must return B's server side: matching the peer
+// alone returns the stray, matching the server address alone returns A's (or
+// the listener).
+func TestServerFDForIsTheServerSideOfTheDialedConn1081(t *testing.T) {
+	// The stray: a client connected to port X, then X's listener closed, so
+	// that B can be given X as its own port while the stray still names it
+	// as its peer.
+	lnX, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	portX := lnX.Addr().(*net.TCPAddr).Port
+	stray, err := net.Dial("tcp", lnX.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	strayAccepted := acceptTCP(t, lnX)
+	defer func() { _ = strayAccepted.Close() }()
+	_ = lnX.Close()
+
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = ln.Close() }()
+	server := ln.Addr().String()
+	a, err := net.Dial("tcp", server)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = a.Close() }()
+	sA := acceptTCP(t, ln)
+	d := net.Dialer{
+		LocalAddr: &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: portX},
+		Control: func(_, _ string, c syscall.RawConn) error {
+			var serr error
+			if err := c.Control(func(fd uintptr) {
+				serr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEADDR, 1)
+			}); err != nil {
+				return err
+			}
+			return serr
+		},
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	b, err := d.DialContext(ctx, "tcp", server)
+	if err != nil {
+		t.Skipf("apparatus: cannot dial from the stray's peer port %d: %v", portX, err)
+	}
+	defer func() { _ = b.Close() }()
+	sB := acceptTCP(t, ln)
+	local := b.LocalAddr().String()
+
+	strayFD := dupAbove(t, stray.(*net.TCPConn), 100)
+	aFD := dupAbove(t, sA, 200)
+	bFD := dupAbove(t, sB, 9000)
+	if !namesPeer(strayFD, local) {
+		t.Fatalf("apparatus: the stray (fd %d) does not name %s as its peer", strayFD, local)
+	}
+	if !namesPeer(bFD, local) {
+		t.Fatalf("apparatus: B's server side (fd %d) does not name %s as its peer", bFD, local)
+	}
+
+	got := serverFDFor(local, server)
+	if got != bFD {
+		lsa, _ := unix.Getsockname(got)
+		psa, _ := unix.Getpeername(got)
+		t.Fatalf("serverFDFor(%s, %s) = %d (local %s, peer %s), want %d, B's server side (stray %d, A's server side %d)",
+			local, server, got, sockaddrString(got, lsa), sockaddrString(got, psa), bFD, strayFD, aFD)
+	}
+}

internal/engine/iouring/fd_probe_linux_test.go

@@ -24,9 +24,21 @@ func fdTarget(fd int) string {
 	return s
 }
 
-// serverFDFor returns this process's descriptor whose peer is local (the
-// server side of a loopback connection the test dialed), or -1.
-func serverFDFor(local string) int {
+// serverFDFor returns this process's descriptor that is the server side of
+// the loopback connection the test dialed from local to server: a socket
+// whose peer is local and whose own address is server. Or -1.
+//
+// Both ends are matched (celeris#1081). The peer alone does not name one
+// connection in a test binary that has run other tests: a socket an earlier
+// test left open (the server side of a connection it accepted and kept, such
+// as the descriptor a hand-built worker was given) has as its peer that
+// test's client port, and connect(2) hands that port out again once the
+// client that had it is gone, possibly to this test's client. /proc/self/fd
+// lists its names in
+// lexical order, so such a socket at "123" was found before this
+// connection's at "45", and a test that then checked that descriptor after
+// the engine stopped found a socket the engine never owned still open.
+func serverFDFor(local, server string) int {
 	ents, err := os.ReadDir("/proc/self/fd")
 	if err != nil {
 		return -1
@@ -36,13 +48,24 @@ func serverFDFor(local string) int {
 		if err != nil {
 			continue
 		}
-		if namesPeer(fd, local) {
+		if namesConn(fd, local, server) {
 			return fd
 		}
 	}
 	return -1
 }
 
+// namesConn reports whether fd is a socket whose peer is local and whose own
+// address is server: the server side of the loopback connection dialed from
+// local to server.
+func namesConn(fd int, local, server string) bool {
+	if !namesPeer(fd, local) {
+		return false
+	}
+	sa, err := unix.Getsockname(fd)
+	return err == nil && sockaddrString(fd, sa) == server
+}
+
 // namesPeer reports whether fd is a socket whose peer is local: the server
 // side of the loopback connection the test dialed from local. It tells a
 // descriptor number still naming that connection's socket from the same

internal/engine/iouring/recv_theft_685_linked_linux_test.go

@@ -231,9 +231,14 @@ func linkedTheftAttempt(t *testing.T, e *Engine, sa *unix.SockaddrInet4, arm str
 		t.Fatalf("getsockname A: %v", err)
 	}
 	aLocal := sockaddrString(a, aLocalSA)
+	aPeerSA, err := unix.Getpeername(a)
+	if err != nil {
+		t.Fatalf("getpeername A: %v", err)
+	}
+	aServer := sockaddrString(a, aPeerSA)
 	srv := -1
 	for deadline := time.Now().Add(2 * time.Second); srv < 0 && time.Now().Before(deadline); {
-		if srv = serverFDFor(aLocal); srv < 0 {
+		if srv = serverFDFor(aLocal, aServer); srv < 0 {
 			time.Sleep(2 * time.Millisecond)
 		}
 	}