From 0aafade8f9954828e50b8246466367c21f9dbd61 Mon Sep 17 00:00:00 2001 From: Moses Narrow <36607567+0pcom@users.noreply.github.com> Date: Tue, 14 Jul 2026 18:26:30 -0500 Subject: [PATCH 1/3] net: non-blocking host netdev I/O with an epoll poller The host netdev now uses non-blocking sockets and a small epoll poller in netpoll_native.go. A syscall that would block parks the goroutine on a channel until the fd is ready, its deadline passes, or the fd is closed. Closing an fd wakes every goroutine parked on it, so a blocked Accept, Recv or Send returns on shutdown instead of pinning an OS thread. Deadline changes also interrupt I/O that is already parked, which net/http needs for abortPendingRead. --- netdev.go | 12 ++ netdev_native.go | 92 +++++++++++---- netpoll_native.go | 281 ++++++++++++++++++++++++++++++++++++++++++++++ tcpsock.go | 32 ++++-- udpsock.go | 16 +++ 5 files changed, 403 insertions(+), 30 deletions(-) create mode 100644 netpoll_native.go diff --git a/netdev.go b/netdev.go index 8822262..ab98cd6 100644 --- a/netdev.go +++ b/netdev.go @@ -33,6 +33,18 @@ const ( var netdev netdever = &nopNetdev{} // (useNetdev is go:linkname'd from tinygo/drivers package) +// errPollInterrupted is returned from Recv and Send when a deadline change +// interrupted a blocked operation. The net package retries with the new one. +var errPollInterrupted = errors.New("net: I/O interrupted by deadline change") + +// pollInterrupt wakes goroutines blocked in Recv (write=false) or Send +// (write=true) on sockfd, for netdevs that support it. +func pollInterrupt(sockfd int, write bool) { + if p, ok := netdev.(interface{ PollInterrupt(sockfd int, write bool) }); ok { + p.PollInterrupt(sockfd, write) + } +} + func useNetdev(dev netdever) { netdev = dev } diff --git a/netdev_native.go b/netdev_native.go index e522a08..d7c9448 100644 --- a/netdev_native.go +++ b/netdev_native.go @@ -90,7 +90,9 @@ func (*hostNetdev) Socket(domain, stype, protocol int) (int, error) { protocol = syscall.IPPROTO_TCP } - fd, err := syscall.Socket(domain, stype, protocol) + // Non-blocking so the poller (netpoll_native.go) can park goroutines on + // EAGAIN instead of pinning a thread in a blocking syscall. + fd, err := syscall.Socket(domain, stype|syscall.SOCK_NONBLOCK|syscall.SOCK_CLOEXEC, protocol) if err != nil { return -1, err } @@ -121,30 +123,69 @@ func (n *hostNetdev) Connect(sockfd int, host string, ip netip.AddrPort) error { sa := sockaddrFromParts(addr, ip.Port()) for { err := syscall.Connect(sockfd, sa) - if err == syscall.EINTR { + switch err { + case nil: + return nil + case syscall.EINTR: continue + case syscall.EINPROGRESS, syscall.EALREADY, syscall.EAGAIN: + // A non-blocking connect is in progress. Wait for the socket to + // become writable, then read the pending error via SO_ERROR. + if werr := poller.wait(sockfd, true, time.Time{}); werr != nil { + return werr + } + soErr, gerr := syscall.GetsockoptInt(sockfd, syscall.SOL_SOCKET, syscall.SO_ERROR) + if gerr != nil { + return gerr + } + if soErr != 0 { + return syscall.Errno(soErr) + } + return nil + default: + return err } - return err } } +// PollInterrupt wakes any goroutine parked in Recv or Send on sockfd so it +// re-evaluates its deadline. +func (*hostNetdev) PollInterrupt(sockfd int, write bool) { + poller.interrupt(sockfd, write) +} + func (*hostNetdev) Listen(sockfd int, backlog int) error { return syscall.Listen(sockfd, backlog) } func (*hostNetdev) Accept(sockfd int) (int, netip.AddrPort, error) { - nfd, sa, err := syscall.Accept(sockfd) - if err != nil { - return -1, netip.AddrPort{}, err - } - var raddr netip.AddrPort - switch s := sa.(type) { - case *syscall.SockaddrInet4: - raddr = netip.AddrPortFrom(netip.AddrFrom4(s.Addr), uint16(s.Port)) - case *syscall.SockaddrInet6: - raddr = netip.AddrPortFrom(netip.AddrFrom16(s.Addr), uint16(s.Port)) + for { + // Accept4 with SOCK_NONBLOCK|SOCK_CLOEXEC keeps accepted sockets + // non-blocking too, so their reads/writes also go through the poller. + nfd, sa, err := syscall.Accept4(sockfd, syscall.SOCK_NONBLOCK|syscall.SOCK_CLOEXEC) + switch err { + case nil: + case syscall.EINTR: + continue + case syscall.EAGAIN: // == EWOULDBLOCK on Linux + // No pending connection. Park until the listener is readable or + // the listening fd is closed, which unblocks accept for shutdown. + if werr := poller.wait(sockfd, false, time.Time{}); werr != nil { + return -1, netip.AddrPort{}, werr + } + continue + default: + return -1, netip.AddrPort{}, err + } + var raddr netip.AddrPort + switch s := sa.(type) { + case *syscall.SockaddrInet4: + raddr = netip.AddrPortFrom(netip.AddrFrom4(s.Addr), uint16(s.Port)) + case *syscall.SockaddrInet6: + raddr = netip.AddrPortFrom(netip.AddrFrom16(s.Addr), uint16(s.Port)) + } + return nfd, raddr, nil } - return nfd, raddr, nil } func (*hostNetdev) Send(sockfd int, buf []byte, flags int, deadline time.Time) (int, error) { @@ -156,16 +197,18 @@ func (*hostNetdev) Send(sockfd int, buf []byte, flags int, deadline time.Time) ( if expired(deadline) { return total, timeoutError{} } - if err := setSockTimeout(sockfd, syscall.SO_SNDTIMEO, deadline); err != nil { - return total, err - } n, err := syscall.Write(sockfd, buf[total:]) if err != nil { if err == syscall.EINTR { continue } if err == syscall.EAGAIN || err == syscall.EWOULDBLOCK { - return total, timeoutError{} + // The send buffer is full. Park until writable, the deadline + // expires, or the fd is closed. + if werr := poller.wait(sockfd, true, deadline); werr != nil { + return total, werr + } + continue } return total, err } @@ -181,9 +224,6 @@ func (*hostNetdev) Recv(sockfd int, buf []byte, flags int, deadline time.Time) ( if expired(deadline) { return 0, timeoutError{} } - if err := setSockTimeout(sockfd, syscall.SO_RCVTIMEO, deadline); err != nil { - return 0, err - } for { n, err := syscall.Read(sockfd, buf) @@ -192,7 +232,12 @@ func (*hostNetdev) Recv(sockfd int, buf []byte, flags int, deadline time.Time) ( continue } if err == syscall.EAGAIN || err == syscall.EWOULDBLOCK { - return 0, timeoutError{} + // Nothing to read yet. Park until readable, the deadline + // expires, or the fd is closed. + if werr := poller.wait(sockfd, false, deadline); werr != nil { + return 0, werr + } + continue } return n, err } @@ -208,6 +253,9 @@ func (*hostNetdev) Recv(sockfd int, buf []byte, flags int, deadline time.Time) ( } func (*hostNetdev) Close(sockfd int) error { + // Wake goroutines parked on this fd before closing it, so a blocked + // Accept, Recv or Send returns on shutdown. + poller.close(sockfd) return syscall.Close(sockfd) } diff --git a/netpoll_native.go b/netpoll_native.go new file mode 100644 index 0000000..84c01f0 --- /dev/null +++ b/netpoll_native.go @@ -0,0 +1,281 @@ +//go:build linux && !baremetal && !nintendoswitch && !wasm_unknown && !tinygo.wasm + +// TINYGO: a small epoll poller for the host netdev. A goroutine that would +// block parks here until the fd is ready, its deadline passes, or it closes. + +package net + +import ( + "errors" + "sync" + "syscall" + "time" +) + +// errPollClosed is returned to goroutines parked on an fd when that fd is closed. +var errPollClosed = errors.New("net: use of closed network connection") + +// pollDesc holds the goroutines currently waiting on one fd, split by direction. +type pollDesc struct { + fd int + readers []chan error + writers []chan error + inEpoll bool + + // Records an interrupt that arrived with no waiter parked, so the next + // wait in that direction returns at once instead of using a stale deadline. + readInterrupt bool + writeInterrupt bool +} + +type netPoller struct { + once sync.Once + err error // set if the poller failed to initialize + epfd int + + mu sync.Mutex + fds map[int]*pollDesc +} + +var poller netPoller + +func (p *netPoller) init() { + p.once.Do(func() { + epfd, err := syscall.EpollCreate1(syscall.EPOLL_CLOEXEC) + if err != nil { + p.err = err + return + } + p.epfd = epfd + p.fds = make(map[int]*pollDesc) + go p.loop() + }) +} + +// events returns the epoll interest mask for pd given its current waiters. +func (pd *pollDesc) events() uint32 { + var ev uint32 + if len(pd.readers) != 0 { + ev |= syscall.EPOLLIN | syscall.EPOLLRDHUP + } + if len(pd.writers) != 0 { + ev |= syscall.EPOLLOUT + } + if ev != 0 { + ev |= syscall.EPOLLONESHOT + } + return ev +} + +// arm (re)programs epoll for pd. Must be called with p.mu held. +func (p *netPoller) arm(pd *pollDesc) { + ev := pd.events() + if ev == 0 { + if pd.inEpoll { + syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_DEL, pd.fd, nil) + pd.inEpoll = false + } + if !pd.readInterrupt && !pd.writeInterrupt { + delete(p.fds, pd.fd) + } + return + } + event := &syscall.EpollEvent{Events: ev, Fd: int32(pd.fd)} + if pd.inEpoll { + syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_MOD, pd.fd, event) + } else { + if err := syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_ADD, pd.fd, event); err == nil { + pd.inEpoll = true + } + } +} + +// wait blocks until fd is ready in the requested direction (write=EPOLLOUT, +// otherwise EPOLLIN), the deadline expires, or the fd is closed. +func (p *netPoller) wait(fd int, write bool, deadline time.Time) error { + p.init() + if p.err != nil { + return p.err + } + + ch := make(chan error, 1) + + p.mu.Lock() + pd := p.fds[fd] + if pd == nil { + pd = &pollDesc{fd: fd} + p.fds[fd] = pd + } + if (write && pd.writeInterrupt) || (!write && pd.readInterrupt) { + // An interrupt arrived before we parked. Consume it and let the caller + // re-evaluate its deadline. + if write { + pd.writeInterrupt = false + } else { + pd.readInterrupt = false + } + p.arm(pd) + p.mu.Unlock() + return errPollInterrupted + } + if write { + pd.writers = append(pd.writers, ch) + } else { + pd.readers = append(pd.readers, ch) + } + p.arm(pd) + p.mu.Unlock() + + var timeout <-chan time.Time + if !deadline.IsZero() { + d := time.Until(deadline) + if d <= 0 { + p.cancelWaiter(fd, write, ch) + return timeoutError{} + } + t := time.NewTimer(d) + defer t.Stop() + timeout = t.C + } + + select { + case err := <-ch: + return err + case <-timeout: + p.cancelWaiter(fd, write, ch) + return timeoutError{} + } +} + +// cancelWaiter removes a single waiter channel that gave up (deadline expired) +// before the poller signalled it. +func (p *netPoller) cancelWaiter(fd int, write bool, ch chan error) { + p.mu.Lock() + defer p.mu.Unlock() + pd := p.fds[fd] + if pd == nil { + return + } + if write { + pd.writers = removeChan(pd.writers, ch) + } else { + pd.readers = removeChan(pd.readers, ch) + } + p.arm(pd) +} + +// interrupt wakes every goroutine parked on fd in the given direction so it +// re-evaluates its deadline. net/http's abortPendingRead relies on this. +func (p *netPoller) interrupt(fd int, write bool) { + if p.err != nil { + return + } + p.mu.Lock() + if p.fds == nil { + // The poller never started, so nothing can be parked. Any later + // wait() captures the new deadline anyway. + p.mu.Unlock() + return + } + pd := p.fds[fd] + if pd == nil { + pd = &pollDesc{fd: fd} + p.fds[fd] = pd + } + var chs []chan error + if write { + chs, pd.writers = pd.writers, nil + if len(chs) == 0 { + pd.writeInterrupt = true + } + } else { + chs, pd.readers = pd.readers, nil + if len(chs) == 0 { + pd.readInterrupt = true + } + } + p.arm(pd) + p.mu.Unlock() + + for _, ch := range chs { + ch <- errPollInterrupted + } +} + +// close wakes every goroutine parked on fd with errPollClosed and stops polling +// it. It must be called just before the fd is actually closed. +func (p *netPoller) close(fd int) { + if p.err != nil { + return + } + p.mu.Lock() + pd := p.fds[fd] + if pd == nil { + p.mu.Unlock() + return + } + if pd.inEpoll { + syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_DEL, fd, nil) + } + delete(p.fds, fd) + readers, writers := pd.readers, pd.writers + pd.readers, pd.writers = nil, nil + p.mu.Unlock() + + for _, ch := range readers { + ch <- errPollClosed + } + for _, ch := range writers { + ch <- errPollClosed + } +} + +// loop is the poller's background goroutine. It waits for epoll events +// and wakes the corresponding parked goroutines. +func (p *netPoller) loop() { + events := make([]syscall.EpollEvent, 64) + for { + n, err := syscall.EpollWait(p.epfd, events, -1) + if err != nil { + if err == syscall.EINTR { + continue + } + return + } + p.mu.Lock() + for i := 0; i < n; i++ { + e := events[i] + pd := p.fds[int(e.Fd)] + if pd == nil { + continue + } + // On error/hangup, wake everyone so they observe the real result. + hup := e.Events&(syscall.EPOLLERR|syscall.EPOLLHUP|syscall.EPOLLRDHUP) != 0 + if hup || e.Events&syscall.EPOLLIN != 0 { + for _, ch := range pd.readers { + ch <- nil + } + pd.readers = nil + } + if hup || e.Events&syscall.EPOLLOUT != 0 { + for _, ch := range pd.writers { + ch <- nil + } + pd.writers = nil + } + // EPOLLONESHOT disabled the fd, so re-arm for any remaining waiters. + pd.inEpoll = true // it is still registered, just disarmed + p.arm(pd) + } + p.mu.Unlock() + } +} + +func removeChan(s []chan error, ch chan error) []chan error { + for i, c := range s { + if c == ch { + return append(s[:i], s[i+1:]...) + } + } + return s +} diff --git a/tcpsock.go b/tcpsock.go index 4e3c864..51964c8 100644 --- a/tcpsock.go +++ b/tcpsock.go @@ -208,6 +208,11 @@ func (c *TCPConn) SyscallConn() (syscall.RawConn, error) { func (c *TCPConn) Read(b []byte) (int, error) { n, err := netdev.Recv(c.fd, b, 0, c.readDeadline) + for err == errPollInterrupted { + // A concurrent deadline change interrupted the wait. Retry with the + // fresh deadline, since nothing was read. + n, err = netdev.Recv(c.fd, b, 0, c.readDeadline) + } // Turn the -1 socket error into 0 and let err speak for error if n < 0 { n = 0 @@ -219,15 +224,22 @@ func (c *TCPConn) Read(b []byte) (int, error) { } func (c *TCPConn) Write(b []byte) (int, error) { - n, err := netdev.Send(c.fd, b, 0, c.writeDeadline) - // Turn the -1 socket error into 0 and let err speak for error - if n < 0 { - n = 0 + total := 0 + for { + n, err := netdev.Send(c.fd, b[total:], 0, c.writeDeadline) + if n > 0 { + total += n + } + if err == errPollInterrupted { + // A concurrent deadline change interrupted the wait. Keep sending + // the remainder under the fresh deadline. + continue + } + if err != nil { + err = &OpError{Op: "write", Net: c.net, Source: c.laddr, Addr: c.raddr, Err: err} + } + return total, err } - if err != nil { - err = &OpError{Op: "write", Net: c.net, Source: c.laddr, Addr: c.raddr, Err: err} - } - return n, err } func (c *TCPConn) Close() error { @@ -245,6 +257,8 @@ func (c *TCPConn) RemoteAddr() Addr { func (c *TCPConn) SetDeadline(t time.Time) error { c.readDeadline = t c.writeDeadline = t + pollInterrupt(c.fd, false) + pollInterrupt(c.fd, true) return nil } @@ -284,11 +298,13 @@ func (c *TCPConn) SetKeepAlivePeriod(d time.Duration) error { func (c *TCPConn) SetReadDeadline(t time.Time) error { c.readDeadline = t + pollInterrupt(c.fd, false) return nil } func (c *TCPConn) SetWriteDeadline(t time.Time) error { c.writeDeadline = t + pollInterrupt(c.fd, true) return nil } diff --git a/udpsock.go b/udpsock.go index bf69aee..a2e9006 100644 --- a/udpsock.go +++ b/udpsock.go @@ -269,6 +269,9 @@ func (c *UDPConn) SyscallConn() (syscall.RawConn, error) { func (c *UDPConn) Read(b []byte) (int, error) { n, err := netdev.Recv(c.fd, b, 0, c.readDeadline) + for err == errPollInterrupted { + n, err = netdev.Recv(c.fd, b, 0, c.readDeadline) + } // Turn the -1 socket error into 0 and let err speak for error if n < 0 { n = 0 @@ -281,6 +284,9 @@ func (c *UDPConn) Read(b []byte) (int, error) { func (c *UDPConn) Write(b []byte) (int, error) { n, err := netdev.Send(c.fd, b, 0, c.writeDeadline) + for err == errPollInterrupted { + n, err = netdev.Send(c.fd, b, 0, c.writeDeadline) + } // Turn the -1 socket error into 0 and let err speak for error if n < 0 { n = 0 @@ -297,6 +303,9 @@ func (c *UDPConn) Write(b []byte) (int, error) { // remote address (c.raddr) is returned as the source. func (c *UDPConn) ReadFromUDP(b []byte) (int, *UDPAddr, error) { n, err := netdev.Recv(c.fd, b, 0, c.readDeadline) + for err == errPollInterrupted { + n, err = netdev.Recv(c.fd, b, 0, c.readDeadline) + } if n < 0 { n = 0 } @@ -321,6 +330,9 @@ func (c *UDPConn) ReadFromUDPAddrPort(b []byte) (n int, addr netip.AddrPort, err // connected remote regardless of addr, mirroring netdev.Send. func (c *UDPConn) WriteToUDP(b []byte, addr *UDPAddr) (int, error) { n, err := netdev.Send(c.fd, b, 0, c.writeDeadline) + for err == errPollInterrupted { + n, err = netdev.Send(c.fd, b, 0, c.writeDeadline) + } if n < 0 { n = 0 } @@ -400,15 +412,19 @@ func (c *UDPConn) RemoteAddr() Addr { func (c *UDPConn) SetDeadline(t time.Time) error { c.readDeadline = t c.writeDeadline = t + pollInterrupt(c.fd, false) + pollInterrupt(c.fd, true) return nil } func (c *UDPConn) SetReadDeadline(t time.Time) error { c.readDeadline = t + pollInterrupt(c.fd, false) return nil } func (c *UDPConn) SetWriteDeadline(t time.Time) error { c.writeDeadline = t + pollInterrupt(c.fd, true) return nil } From f416c1f3d7653b52ff28d1be03d7508b7635420e Mon Sep 17 00:00:00 2001 From: yohimik Date: Sat, 5 Sep 2026 06:26:54 +0400 Subject: [PATCH 2/3] net: prevent repeated Close from closing a reused socket (cherry picked from commit a0c3ff5e037ce1f9ec34f1d270659050e360278f) --- close_native_test.go | 176 +++++++++++++++++++++++++++++++++++++++++++ net.go | 16 ++++ tcpsock.go | 10 ++- udpsock.go | 3 +- 4 files changed, 200 insertions(+), 5 deletions(-) create mode 100644 close_native_test.go diff --git a/close_native_test.go b/close_native_test.go new file mode 100644 index 0000000..f13ce6d --- /dev/null +++ b/close_native_test.go @@ -0,0 +1,176 @@ +//go:build linux && !baremetal && !tinygo.wasm + +package net + +import ( + "errors" + "sync" + "syscall" + "testing" + "time" +) + +type closeCountNetdev struct { + nopNetdev + calls int +} + +func (d *closeCountNetdev) Close(int) error { + d.calls++ + return nil +} + +func TestCloseConcurrent(t *testing.T) { + previous := netdev + defer func() { netdev = previous }() + d := &closeCountNetdev{} + netdev = d + c := &TCPConn{fd: 42} + var wg sync.WaitGroup + results := make(chan error, 32) + for i := 0; i < 32; i++ { + wg.Add(1) + go func() { defer wg.Done(); results <- c.Close() }() + } + wg.Wait() + close(results) + success := 0 + for err := range results { + if err == nil { + success++ + } else if !errors.Is(err, ErrClosed) { + t.Fatal(err) + } + } + if success != 1 || d.calls != 1 { + t.Fatalf("success=%d device calls=%d", success, d.calls) + } +} + +func TestCloseDescriptorReuse(t *testing.T) { + for _, kind := range []string{"listener", "tcp", "udp"} { + t.Run(kind, func(t *testing.T) { + fd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_STREAM, 0) + if err != nil { + t.Fatal(err) + } + var closeSocket func() error + switch kind { + case "listener": + closeSocket = (&listener{fd: fd}).Close + case "tcp": + closeSocket = (&TCPConn{fd: fd}).Close + case "udp": + closeSocket = (&UDPConn{fd: fd}).Close + } + if err := closeSocket(); err != nil { + t.Fatal(err) + } + replacement, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_STREAM, 0) + if err != nil { + t.Fatal(err) + } + if replacement != fd { + if err := syscall.Dup3(replacement, fd, 0); err != nil { + syscall.Close(replacement) + t.Fatal(err) + } + syscall.Close(replacement) + } + defer syscall.Close(fd) + err = closeSocket() + if !errors.Is(err, ErrClosed) { + t.Errorf("second Close = %v, want ErrClosed", err) + } + if _, err := syscall.Getsockname(fd); err != nil { + t.Fatalf("second Close damaged replacement socket: %v", err) + } + }) + } +} + +func TestCloseBlockedAccept(t *testing.T) { + l, err := Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { + c, err := l.Accept() + if c != nil { + c.Close() + } + done <- err + }() + time.Sleep(25 * time.Millisecond) + if err := l.Close(); err != nil { + t.Fatal(err) + } + select { + case err := <-done: + if err == nil { + t.Fatal("Accept succeeded after close") + } + case <-time.After(time.Second): + t.Fatal("Accept did not stop after close") + } +} + +func TestCloseBlockedRead(t *testing.T) { + pair, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0) + if err != nil { + t.Fatal(err) + } + defer syscall.Close(pair[1]) + // Every socket this netdev creates is SOCK_NONBLOCK, so a hand-made fd must + // match, or Read blocks in the kernel where the poller cannot wake it. + if err := syscall.SetNonblock(pair[0], true); err != nil { + t.Fatal(err) + } + c := &TCPConn{fd: pair[0], net: "tcp"} + done := make(chan error, 1) + go func() { _, err := c.Read(make([]byte, 1)); done <- err }() + time.Sleep(25 * time.Millisecond) + if err := c.Close(); err != nil { + t.Fatal(err) + } + select { + case err := <-done: + if err == nil { + t.Fatal("Read succeeded after close") + } + case <-time.After(time.Second): + t.Fatal("Read did not stop after close") + } +} + +func TestCloseBlockedWrite(t *testing.T) { + pair, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0) + if err != nil { + t.Fatal(err) + } + defer syscall.Close(pair[1]) + if err := syscall.SetsockoptInt(pair[0], syscall.SOL_SOCKET, syscall.SO_SNDBUF, 4096); err != nil { + t.Fatal(err) + } + // See TestCloseBlockedRead. The fd must be non-blocking to park on the + // poller rather than in the kernel. + if err := syscall.SetNonblock(pair[0], true); err != nil { + t.Fatal(err) + } + c := &TCPConn{fd: pair[0], net: "tcp"} + done := make(chan error, 1) + go func() { _, err := c.Write(make([]byte, 1<<20)); done <- err }() + time.Sleep(25 * time.Millisecond) + if err := c.Close(); err != nil { + t.Fatal(err) + } + select { + case err := <-done: + if err == nil { + t.Fatal("Write succeeded after close") + } + case <-time.After(time.Second): + t.Fatal("Write did not stop after close") + } +} diff --git a/net.go b/net.go index 8da88b0..44a594a 100644 --- a/net.go +++ b/net.go @@ -9,9 +9,25 @@ package net import ( "errors" "io" + "sync" "time" ) +type closeGuard struct { + mu sync.Mutex + closed bool +} + +func (c *closeGuard) close(fd int) error { + c.mu.Lock() + defer c.mu.Unlock() + if c.closed { + return ErrClosed + } + c.closed = true + return netdev.Close(fd) +} + // Addr represents a network end point address. // // The two methods [Addr.Network] and [Addr.String] conventionally return strings diff --git a/tcpsock.go b/tcpsock.go index 51964c8..03e7305 100644 --- a/tcpsock.go +++ b/tcpsock.go @@ -143,6 +143,7 @@ func TCPAddrFromAddrPort(addr netip.AddrPort) *TCPAddr { // TCPConn is an implementation of the [Conn] interface for TCP network // connections. type TCPConn struct { + closer closeGuard fd int net string laddr *TCPAddr @@ -243,7 +244,7 @@ func (c *TCPConn) Write(b []byte) (int, error) { } func (c *TCPConn) Close() error { - return netdev.Close(c.fd) + return c.closer.close(c.fd) } func (c *TCPConn) LocalAddr() Addr { @@ -368,8 +369,9 @@ type onlyWriter struct { } type listener struct { - fd int - laddr *TCPAddr + closer closeGuard + fd int + laddr *TCPAddr } func (l *listener) Accept() (Conn, error) { @@ -387,7 +389,7 @@ func (l *listener) Accept() (Conn, error) { } func (l *listener) Close() error { - return netdev.Close(l.fd) + return l.closer.close(l.fd) } func (l *listener) Addr() Addr { diff --git a/udpsock.go b/udpsock.go index a2e9006..bcf1af0 100644 --- a/udpsock.go +++ b/udpsock.go @@ -127,6 +127,7 @@ func UDPAddrFromAddrPort(addr netip.AddrPort) *UDPAddr { // UDPConn is the implementation of the Conn and PacketConn interfaces // for UDP network connections. type UDPConn struct { + closer closeGuard fd int net string laddr *UDPAddr @@ -398,7 +399,7 @@ func (c *UDPConn) WriteMsgUDP(b, oob []byte, addr *UDPAddr) (n, oobn int, err er } func (c *UDPConn) Close() error { - return netdev.Close(c.fd) + return c.closer.close(c.fd) } func (c *UDPConn) LocalAddr() Addr { From 97788609b0fa734403bd9cfedf8a691ac3a2ee8e Mon Sep 17 00:00:00 2001 From: Moses Narrow <36607567+0pcom@users.noreply.github.com> Date: Tue, 27 Oct 2026 10:56:37 -0500 Subject: [PATCH 3/3] net: close the poller race with Close, and run without threads A Read or Write that saw EAGAIN could park forever if Close ran before it reached the poller. poller.close found no waiter, the fd was closed, the later EPOLL_CTL_ADD failed with EBADF and arm ignored it. If the fd number was reused, the call parked on another socket. Recv, Send, Accept and Connect now hold a reference on the fd while they run. Close marks the fd closing, wakes its waiters and leaves close(2) to the last call that releases it, so the number cannot be reused while a call may still wait on it. A wait after Close returns at once. arm passes an epoll error to the waiters instead of dropping it. All poller state is now under one mutex, which fixes the race between interrupt and the lazy init. Without the threads scheduler a blocking epoll_wait stops the whole program, so the poller polls and sleeps 1ms instead. Also removes the unused setSockTimeout and moves the useNetdev comment back. --- close_native_test.go | 61 +++++++++ netdev.go | 2 +- netdev_native.go | 70 ++++------- netpoll_native.go | 289 +++++++++++++++++++++++-------------------- netpoll_nothreads.go | 12 ++ netpoll_threads.go | 11 ++ 6 files changed, 264 insertions(+), 181 deletions(-) create mode 100644 netpoll_nothreads.go create mode 100644 netpoll_threads.go diff --git a/close_native_test.go b/close_native_test.go index f13ce6d..d0a7418 100644 --- a/close_native_test.go +++ b/close_native_test.go @@ -174,3 +174,64 @@ func TestCloseBlockedWrite(t *testing.T) { t.Fatal("Write did not stop after close") } } + +// Close lands after a call saw EAGAIN but before it parked on the poller. +func TestCloseBeforeWait(t *testing.T) { + pair, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0) + if err != nil { + t.Fatal(err) + } + defer syscall.Close(pair[1]) + fd := pair[0] + if err := syscall.SetNonblock(fd, true); err != nil { + t.Fatal(err) + } + if err := poller.acquire(fd); err != nil { + t.Fatal(err) + } + if err := netdev.Close(fd); err != nil { + t.Fatal(err) + } + if _, err := syscall.Getsockname(fd); err != nil { + t.Fatalf("fd closed while a call still holds it: %v", err) + } + done := make(chan error, 1) + go func() { done <- poller.wait(fd, false, time.Time{}) }() + select { + case err := <-done: + if err != errPollClosed { + t.Fatalf("wait = %v, want errPollClosed", err) + } + case <-time.After(time.Second): + t.Fatal("wait parked after close") + } + poller.release(fd) + if _, err := syscall.Getsockname(fd); err != syscall.EBADF { + t.Fatalf("fd still open after the last call released it: %v", err) + } +} + +func TestDeadlineInterruptsRead(t *testing.T) { + pair, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0) + if err != nil { + t.Fatal(err) + } + defer syscall.Close(pair[1]) + if err := syscall.SetNonblock(pair[0], true); err != nil { + t.Fatal(err) + } + c := &TCPConn{fd: pair[0], net: "tcp"} + defer c.Close() + done := make(chan error, 1) + go func() { _, err := c.Read(make([]byte, 1)); done <- err }() + time.Sleep(25 * time.Millisecond) + c.SetReadDeadline(time.Now()) + select { + case err := <-done: + if !errors.As(err, new(timeoutError)) { + t.Fatalf("Read = %v, want a timeout", err) + } + case <-time.After(time.Second): + t.Fatal("Read did not see the new deadline") + } +} diff --git a/netdev.go b/netdev.go index ab98cd6..51af68d 100644 --- a/netdev.go +++ b/netdev.go @@ -32,7 +32,6 @@ const ( // explicitly set with useNetdev(). var netdev netdever = &nopNetdev{} -// (useNetdev is go:linkname'd from tinygo/drivers package) // errPollInterrupted is returned from Recv and Send when a deadline change // interrupted a blocked operation. The net package retries with the new one. var errPollInterrupted = errors.New("net: I/O interrupted by deadline change") @@ -45,6 +44,7 @@ func pollInterrupt(sockfd int, write bool) { } } +// (useNetdev is go:linkname'd from tinygo/drivers package) func useNetdev(dev netdever) { netdev = dev } diff --git a/netdev_native.go b/netdev_native.go index d7c9448..a8a9301 100644 --- a/netdev_native.go +++ b/netdev_native.go @@ -34,11 +34,8 @@ func init() { // hostNetdev implements netdever using raw Linux sockets via the syscall // package. The "sockfd" values it returns are plain OS file descriptors. // -// Deadlines are implemented with the per-socket SO_RCVTIMEO/SO_SNDTIMEO -// options rather than a runtime poller. As a consequence, issuing concurrent -// reads (or concurrent writes) with different deadlines on the same connection -// is not supported; this matches the typical net.Conn usage of one reader and -// one writer goroutine. +// Sockets are non-blocking. A call that would block parks on the poller in +// netpoll_native.go until the fd is ready, its deadline passes, or it closes. type hostNetdev struct{} // timeoutError is returned from Send/Recv when a deadline expires. It satisfies @@ -90,8 +87,6 @@ func (*hostNetdev) Socket(domain, stype, protocol int) (int, error) { protocol = syscall.IPPROTO_TCP } - // Non-blocking so the poller (netpoll_native.go) can park goroutines on - // EAGAIN instead of pinning a thread in a blocking syscall. fd, err := syscall.Socket(domain, stype|syscall.SOCK_NONBLOCK|syscall.SOCK_CLOEXEC, protocol) if err != nil { return -1, err @@ -121,6 +116,10 @@ func (n *hostNetdev) Connect(sockfd int, host string, ip netip.AddrPort) error { addr = resolved } sa := sockaddrFromParts(addr, ip.Port()) + if err := poller.acquire(sockfd); err != nil { + return err + } + defer poller.release(sockfd) for { err := syscall.Connect(sockfd, sa) switch err { @@ -129,8 +128,8 @@ func (n *hostNetdev) Connect(sockfd int, host string, ip netip.AddrPort) error { case syscall.EINTR: continue case syscall.EINPROGRESS, syscall.EALREADY, syscall.EAGAIN: - // A non-blocking connect is in progress. Wait for the socket to - // become writable, then read the pending error via SO_ERROR. + // The connect finishes when the socket becomes writable, and + // SO_ERROR holds its result. if werr := poller.wait(sockfd, true, time.Time{}); werr != nil { return werr } @@ -148,8 +147,8 @@ func (n *hostNetdev) Connect(sockfd int, host string, ip netip.AddrPort) error { } } -// PollInterrupt wakes any goroutine parked in Recv or Send on sockfd so it -// re-evaluates its deadline. +// PollInterrupt wakes the goroutines parked in Recv or Send on sockfd so they +// pick up a new deadline. func (*hostNetdev) PollInterrupt(sockfd int, write bool) { poller.interrupt(sockfd, write) } @@ -159,17 +158,17 @@ func (*hostNetdev) Listen(sockfd int, backlog int) error { } func (*hostNetdev) Accept(sockfd int) (int, netip.AddrPort, error) { + if err := poller.acquire(sockfd); err != nil { + return -1, netip.AddrPort{}, err + } + defer poller.release(sockfd) for { - // Accept4 with SOCK_NONBLOCK|SOCK_CLOEXEC keeps accepted sockets - // non-blocking too, so their reads/writes also go through the poller. nfd, sa, err := syscall.Accept4(sockfd, syscall.SOCK_NONBLOCK|syscall.SOCK_CLOEXEC) switch err { case nil: case syscall.EINTR: continue - case syscall.EAGAIN: // == EWOULDBLOCK on Linux - // No pending connection. Park until the listener is readable or - // the listening fd is closed, which unblocks accept for shutdown. + case syscall.EAGAIN: if werr := poller.wait(sockfd, false, time.Time{}); werr != nil { return -1, netip.AddrPort{}, werr } @@ -189,9 +188,11 @@ func (*hostNetdev) Accept(sockfd int) (int, netip.AddrPort, error) { } func (*hostNetdev) Send(sockfd int, buf []byte, flags int, deadline time.Time) (int, error) { - // net.Conn.Write must send the whole buffer (or report an error), so loop - // over short writes. The send timeout is (re)programmed each iteration so a - // deadline bounds the whole operation, not each individual write. + // net.Conn.Write must send the whole buffer or report an error. + if err := poller.acquire(sockfd); err != nil { + return 0, err + } + defer poller.release(sockfd) total := 0 for total < len(buf) { if expired(deadline) { @@ -202,9 +203,7 @@ func (*hostNetdev) Send(sockfd int, buf []byte, flags int, deadline time.Time) ( if err == syscall.EINTR { continue } - if err == syscall.EAGAIN || err == syscall.EWOULDBLOCK { - // The send buffer is full. Park until writable, the deadline - // expires, or the fd is closed. + if err == syscall.EAGAIN { if werr := poller.wait(sockfd, true, deadline); werr != nil { return total, werr } @@ -224,6 +223,10 @@ func (*hostNetdev) Recv(sockfd int, buf []byte, flags int, deadline time.Time) ( if expired(deadline) { return 0, timeoutError{} } + if err := poller.acquire(sockfd); err != nil { + return 0, err + } + defer poller.release(sockfd) for { n, err := syscall.Read(sockfd, buf) @@ -231,9 +234,7 @@ func (*hostNetdev) Recv(sockfd int, buf []byte, flags int, deadline time.Time) ( if err == syscall.EINTR { continue } - if err == syscall.EAGAIN || err == syscall.EWOULDBLOCK { - // Nothing to read yet. Park until readable, the deadline - // expires, or the fd is closed. + if err == syscall.EAGAIN { if werr := poller.wait(sockfd, false, deadline); werr != nil { return 0, werr } @@ -253,10 +254,7 @@ func (*hostNetdev) Recv(sockfd int, buf []byte, flags int, deadline time.Time) ( } func (*hostNetdev) Close(sockfd int) error { - // Wake goroutines parked on this fd before closing it, so a blocked - // Accept, Recv or Send returns on shutdown. - poller.close(sockfd) - return syscall.Close(sockfd) + return poller.close(sockfd) } func (*hostNetdev) SetSockOpt(sockfd int, level int, opt int, value interface{}) error { @@ -319,20 +317,6 @@ func expired(deadline time.Time) bool { return !deadline.IsZero() && !deadline.After(time.Now()) } -// setSockTimeout programs SO_RCVTIMEO/SO_SNDTIMEO so a blocking recv/send -// returns EAGAIN at the deadline. A zero deadline disables the timeout. -func setSockTimeout(sockfd int, opt int, deadline time.Time) error { - var tv syscall.Timeval - if !deadline.IsZero() { - d := time.Until(deadline) - if d < time.Microsecond { - d = time.Microsecond - } - tv = syscall.NsecToTimeval(d.Nanoseconds()) - } - return syscall.SetsockoptTimeval(sockfd, syscall.SOL_SOCKET, opt, &tv) -} - // --- Name resolution -------------------------------------------------------- // hostLookup resolves a host name (or IP literal) to a single IPv4 address, diff --git a/netpoll_native.go b/netpoll_native.go index 84c01f0..5e1d896 100644 --- a/netpoll_native.go +++ b/netpoll_native.go @@ -12,47 +12,105 @@ import ( "time" ) -// errPollClosed is returned to goroutines parked on an fd when that fd is closed. var errPollClosed = errors.New("net: use of closed network connection") -// pollDesc holds the goroutines currently waiting on one fd, split by direction. type pollDesc struct { - fd int readers []chan error writers []chan error inEpoll bool - // Records an interrupt that arrived with no waiter parked, so the next - // wait in that direction returns at once instead of using a stale deadline. + // An interrupt that found no parked waiter, so the next wait returns at once. readInterrupt bool writeInterrupt bool + + // Calls in progress on the fd. Close defers close(2) until this drops to 0, + // so the fd number cannot be reused while a call may still wait on it. + refs int + closing bool } type netPoller struct { - once sync.Once - err error // set if the poller failed to initialize - epfd int - - mu sync.Mutex - fds map[int]*pollDesc + mu sync.Mutex + started bool + err error + epfd int + fds map[int]*pollDesc } var poller netPoller -func (p *netPoller) init() { - p.once.Do(func() { - epfd, err := syscall.EpollCreate1(syscall.EPOLL_CLOEXEC) - if err != nil { - p.err = err - return +// start creates the epoll instance on first use. Must be called with p.mu held. +func (p *netPoller) start() error { + if !p.started { + p.started = true + p.epfd, p.err = syscall.EpollCreate1(syscall.EPOLL_CLOEXEC) + if p.err == nil { + go p.loop(p.epfd) } - p.epfd = epfd + } + return p.err +} + +// acquire registers a call on fd. It fails once Close has run. +func (p *netPoller) acquire(fd int) error { + p.mu.Lock() + defer p.mu.Unlock() + if p.fds == nil { p.fds = make(map[int]*pollDesc) - go p.loop() - }) + } + pd := p.fds[fd] + if pd == nil { + pd = &pollDesc{} + p.fds[fd] = pd + } + if pd.closing { + return errPollClosed + } + pd.refs++ + return nil +} + +// release ends a call started with acquire, and closes fd if it was the last +// call after Close. +func (p *netPoller) release(fd int) { + p.mu.Lock() + pd := p.fds[fd] + pd.refs-- + if pd.refs > 0 { + p.mu.Unlock() + return + } + closing := pd.closing + delete(p.fds, fd) + p.mu.Unlock() + if closing { + syscall.Close(fd) + } +} + +// close wakes every waiter on fd with errPollClosed and closes fd, at once if +// no call is in progress and otherwise when the last one releases it. +func (p *netPoller) close(fd int) error { + p.mu.Lock() + pd := p.fds[fd] + if pd == nil { + p.mu.Unlock() + return syscall.Close(fd) + } + pd.closing = true + if pd.inEpoll { + syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_DEL, fd, nil) + pd.inEpoll = false + } + chs := append(pd.readers, pd.writers...) + pd.readers, pd.writers = nil, nil + p.mu.Unlock() + for _, ch := range chs { + ch <- errPollClosed + } + return nil } -// events returns the epoll interest mask for pd given its current waiters. func (pd *pollDesc) events() uint32 { var ev uint32 if len(pd.readers) != 0 { @@ -67,71 +125,71 @@ func (pd *pollDesc) events() uint32 { return ev } -// arm (re)programs epoll for pd. Must be called with p.mu held. -func (p *netPoller) arm(pd *pollDesc) { +// arm programs epoll for the waiters on fd. If that fails, it wakes them with +// the error. Must be called with p.mu held. +func (p *netPoller) arm(fd int, pd *pollDesc) { ev := pd.events() if ev == 0 { if pd.inEpoll { - syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_DEL, pd.fd, nil) + syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_DEL, fd, nil) pd.inEpoll = false } - if !pd.readInterrupt && !pd.writeInterrupt { - delete(p.fds, pd.fd) - } return } - event := &syscall.EpollEvent{Events: ev, Fd: int32(pd.fd)} + op := syscall.EPOLL_CTL_ADD if pd.inEpoll { - syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_MOD, pd.fd, event) - } else { - if err := syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_ADD, pd.fd, event); err == nil { - pd.inEpoll = true - } + op = syscall.EPOLL_CTL_MOD + } + err := syscall.EpollCtl(p.epfd, op, fd, &syscall.EpollEvent{Events: ev, Fd: int32(fd)}) + if err == nil { + pd.inEpoll = true + return + } + chs := append(pd.readers, pd.writers...) + pd.readers, pd.writers = nil, nil + for _, ch := range chs { + ch <- err } } -// wait blocks until fd is ready in the requested direction (write=EPOLLOUT, -// otherwise EPOLLIN), the deadline expires, or the fd is closed. +// wait blocks until fd is ready for writing (write) or reading, the deadline +// passes, or fd is closed. The caller must hold a reference from acquire. func (p *netPoller) wait(fd int, write bool, deadline time.Time) error { - p.init() - if p.err != nil { - return p.err - } - ch := make(chan error, 1) p.mu.Lock() - pd := p.fds[fd] - if pd == nil { - pd = &pollDesc{fd: fd} - p.fds[fd] = pd + if err := p.start(); err != nil { + p.mu.Unlock() + return err } - if (write && pd.writeInterrupt) || (!write && pd.readInterrupt) { - // An interrupt arrived before we parked. Consume it and let the caller - // re-evaluate its deadline. - if write { - pd.writeInterrupt = false - } else { - pd.readInterrupt = false - } - p.arm(pd) + pd := p.fds[fd] + if pd.closing { p.mu.Unlock() - return errPollInterrupted + return errPollClosed } if write { + if pd.writeInterrupt { + pd.writeInterrupt = false + p.mu.Unlock() + return errPollInterrupted + } pd.writers = append(pd.writers, ch) } else { + if pd.readInterrupt { + pd.readInterrupt = false + p.mu.Unlock() + return errPollInterrupted + } pd.readers = append(pd.readers, ch) } - p.arm(pd) + p.arm(fd, pd) p.mu.Unlock() var timeout <-chan time.Time if !deadline.IsZero() { d := time.Until(deadline) if d <= 0 { - p.cancelWaiter(fd, write, ch) - return timeoutError{} + return p.cancelWaiter(fd, write, ch) } t := time.NewTimer(d) defer t.Stop() @@ -142,130 +200,87 @@ func (p *netPoller) wait(fd int, write bool, deadline time.Time) error { case err := <-ch: return err case <-timeout: - p.cancelWaiter(fd, write, ch) - return timeoutError{} + return p.cancelWaiter(fd, write, ch) } } -// cancelWaiter removes a single waiter channel that gave up (deadline expired) -// before the poller signalled it. -func (p *netPoller) cancelWaiter(fd int, write bool, ch chan error) { +// cancelWaiter removes a waiter whose deadline passed. If it was woken in the +// meantime, that result wins over the timeout. +func (p *netPoller) cancelWaiter(fd int, write bool, ch chan error) error { p.mu.Lock() - defer p.mu.Unlock() pd := p.fds[fd] - if pd == nil { - return - } if write { pd.writers = removeChan(pd.writers, ch) } else { pd.readers = removeChan(pd.readers, ch) } - p.arm(pd) + p.arm(fd, pd) + p.mu.Unlock() + select { + case err := <-ch: + return err + default: + return timeoutError{} + } } -// interrupt wakes every goroutine parked on fd in the given direction so it -// re-evaluates its deadline. net/http's abortPendingRead relies on this. +// interrupt wakes the goroutines parked on fd in one direction so they pick up +// a new deadline. net/http's abortPendingRead relies on this. func (p *netPoller) interrupt(fd int, write bool) { - if p.err != nil { - return - } p.mu.Lock() - if p.fds == nil { - // The poller never started, so nothing can be parked. Any later - // wait() captures the new deadline anyway. + pd := p.fds[fd] + if pd == nil || pd.refs == 0 { p.mu.Unlock() return } - pd := p.fds[fd] - if pd == nil { - pd = &pollDesc{fd: fd} - p.fds[fd] = pd - } var chs []chan error if write { chs, pd.writers = pd.writers, nil - if len(chs) == 0 { - pd.writeInterrupt = true - } + pd.writeInterrupt = len(chs) == 0 } else { chs, pd.readers = pd.readers, nil - if len(chs) == 0 { - pd.readInterrupt = true - } + pd.readInterrupt = len(chs) == 0 } - p.arm(pd) + p.arm(fd, pd) p.mu.Unlock() - for _, ch := range chs { ch <- errPollInterrupted } } -// close wakes every goroutine parked on fd with errPollClosed and stops polling -// it. It must be called just before the fd is actually closed. -func (p *netPoller) close(fd int) { - if p.err != nil { - return - } - p.mu.Lock() - pd := p.fds[fd] - if pd == nil { - p.mu.Unlock() - return - } - if pd.inEpoll { - syscall.EpollCtl(p.epfd, syscall.EPOLL_CTL_DEL, fd, nil) - } - delete(p.fds, fd) - readers, writers := pd.readers, pd.writers - pd.readers, pd.writers = nil, nil - p.mu.Unlock() - - for _, ch := range readers { - ch <- errPollClosed - } - for _, ch := range writers { - ch <- errPollClosed - } -} - -// loop is the poller's background goroutine. It waits for epoll events -// and wakes the corresponding parked goroutines. -func (p *netPoller) loop() { +func (p *netPoller) loop(epfd int) { events := make([]syscall.EpollEvent, 64) for { - n, err := syscall.EpollWait(p.epfd, events, -1) + n, err := syscall.EpollWait(epfd, events, pollWaitMsec) + if err == syscall.EINTR { + continue + } if err != nil { - if err == syscall.EINTR { - continue - } return } + if n == 0 { + time.Sleep(pollIdle) + continue + } p.mu.Lock() - for i := 0; i < n; i++ { - e := events[i] + for _, e := range events[:n] { pd := p.fds[int(e.Fd)] if pd == nil { continue } - // On error/hangup, wake everyone so they observe the real result. + // On an error or hangup, wake everyone so they see the real result. hup := e.Events&(syscall.EPOLLERR|syscall.EPOLLHUP|syscall.EPOLLRDHUP) != 0 + var chs []chan error if hup || e.Events&syscall.EPOLLIN != 0 { - for _, ch := range pd.readers { - ch <- nil - } - pd.readers = nil + chs, pd.readers = append(chs, pd.readers...), nil } if hup || e.Events&syscall.EPOLLOUT != 0 { - for _, ch := range pd.writers { - ch <- nil - } - pd.writers = nil + chs, pd.writers = append(chs, pd.writers...), nil + } + for _, ch := range chs { + ch <- nil } - // EPOLLONESHOT disabled the fd, so re-arm for any remaining waiters. - pd.inEpoll = true // it is still registered, just disarmed - p.arm(pd) + p.arm(int(e.Fd), pd) } p.mu.Unlock() } diff --git a/netpoll_nothreads.go b/netpoll_nothreads.go new file mode 100644 index 0000000..a4079a3 --- /dev/null +++ b/netpoll_nothreads.go @@ -0,0 +1,12 @@ +//go:build linux && !baremetal && !nintendoswitch && !wasm_unknown && !tinygo.wasm && !scheduler.threads + +package net + +import "time" + +// Other schedulers run every goroutine on one thread, so a blocking epoll_wait +// would stop the program. The poller checks without blocking and sleeps instead. +const ( + pollWaitMsec = 0 + pollIdle = time.Millisecond +) diff --git a/netpoll_threads.go b/netpoll_threads.go new file mode 100644 index 0000000..a5f794e --- /dev/null +++ b/netpoll_threads.go @@ -0,0 +1,11 @@ +//go:build linux && !baremetal && !nintendoswitch && !wasm_unknown && !tinygo.wasm && scheduler.threads + +package net + +import "time" + +// The poller goroutine has its own thread, so it can block in epoll_wait. +const ( + pollWaitMsec = -1 + pollIdle = time.Duration(0) +)