fix(fdwait): remove epoll watches before descriptors can close
- commit
- 464d66cb9a69d04f6c942036dc3dfd59436d27a6
- author
- brice · · 2026-10-13 20:40 UTC
- parent
- ea0e1c7
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email - refs
- proto/netpoll-waits
3 files · +88 -40
A internal/adapters/fdwait/duplicate_test.go +60 -0
| @@ -0,0 +1,60 @@ | |||
| 1 | + | package fdwait | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "golang.org/x/sys/unix" | |
| 5 | + | "testing" | |
| 6 | + | "time" | |
| 7 | + | ) | |
| 8 | + | ||
| 9 | + | // Epoll retains a registration after its fd closes if another descriptor | |
| 10 | + | // still refers to the same open file description. | |
| 11 | + | func TestRemovedDuplicatedDescriptorDoesNotKeepPollReadable(t *testing.T) { | |
| 12 | + | p, err := New() | |
| 13 | + | if err != nil { | |
| 14 | + | t.Fatal(err) | |
| 15 | + | } | |
| 16 | + | defer p.Close() | |
| 17 | + | old, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC) | |
| 18 | + | if err != nil { | |
| 19 | + | t.Fatal(err) | |
| 20 | + | } | |
| 21 | + | dup, err := unix.Dup(old) | |
| 22 | + | if err != nil { | |
| 23 | + | unix.Close(old) | |
| 24 | + | t.Fatal(err) | |
| 25 | + | } | |
| 26 | + | defer unix.Close(dup) | |
| 27 | + | if _, err := p.Poll([]unix.PollFd{{Fd: int32(old), Events: unix.POLLIN}}, 1); err != nil { | |
| 28 | + | t.Fatal(err) | |
| 29 | + | } | |
| 30 | + | // Create the next descriptor before closing old so its number is not | |
| 31 | + | // reused. This leaves an old epoll registration with no usable fd. | |
| 32 | + | fresh, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC) | |
| 33 | + | if err != nil { | |
| 34 | + | t.Fatal(err) | |
| 35 | + | } | |
| 36 | + | defer unix.Close(fresh) | |
| 37 | + | unix.Close(old) | |
| 38 | + | if _, err := unix.Write(dup, []byte{1, 0, 0, 0, 0, 0, 0, 0}); err != nil { | |
| 39 | + | t.Fatal(err) | |
| 40 | + | } | |
| 41 | + | if err := p.raw.Control(func(ep uintptr) { | |
| 42 | + | var events [1]unix.EpollEvent | |
| 43 | + | if n, err := unix.EpollWait(int(ep), events[:], 0); n != 0 || err != nil { | |
| 44 | + | t.Errorf("closed watched descriptor still makes epoll readable: %d %v", n, err) | |
| 45 | + | } | |
| 46 | + | }); err != nil { | |
| 47 | + | t.Fatal(err) | |
| 48 | + | } | |
| 49 | + | done := make(chan error, 1) | |
| 50 | + | go func() { _, err := p.Poll([]unix.PollFd{{Fd: int32(fresh), Events: unix.POLLIN}}, 20); done <- err }() | |
| 51 | + | select { | |
| 52 | + | case err := <-done: | |
| 53 | + | if err != nil { | |
| 54 | + | t.Fatal(err) | |
| 55 | + | } | |
| 56 | + | case <-time.After(time.Second): | |
| 57 | + | p.Close() | |
| 58 | + | t.Fatal("stale epoll readiness prevents timeout") | |
| 59 | + | } | |
| 60 | + | } | |
| @@ -0,0 +1,60 @@ | |||||
| 1 | + | package fdwait | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "golang.org/x/sys/unix" | |||
| 5 | + | "testing" | |||
| 6 | + | "time" | |||
| 7 | + | ) | |||
| 8 | + | ||||
| 9 | + | // Epoll retains a registration after its fd closes if another descriptor | |||
| 10 | + | // still refers to the same open file description. | |||
| 11 | + | func TestRemovedDuplicatedDescriptorDoesNotKeepPollReadable(t *testing.T) { | |||
| 12 | + | p, err := New() | |||
| 13 | + | if err != nil { | |||
| 14 | + | t.Fatal(err) | |||
| 15 | + | } | |||
| 16 | + | defer p.Close() | |||
| 17 | + | old, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC) | |||
| 18 | + | if err != nil { | |||
| 19 | + | t.Fatal(err) | |||
| 20 | + | } | |||
| 21 | + | dup, err := unix.Dup(old) | |||
| 22 | + | if err != nil { | |||
| 23 | + | unix.Close(old) | |||
| 24 | + | t.Fatal(err) | |||
| 25 | + | } | |||
| 26 | + | defer unix.Close(dup) | |||
| 27 | + | if _, err := p.Poll([]unix.PollFd{{Fd: int32(old), Events: unix.POLLIN}}, 1); err != nil { | |||
| 28 | + | t.Fatal(err) | |||
| 29 | + | } | |||
| 30 | + | // Create the next descriptor before closing old so its number is not | |||
| 31 | + | // reused. This leaves an old epoll registration with no usable fd. | |||
| 32 | + | fresh, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC) | |||
| 33 | + | if err != nil { | |||
| 34 | + | t.Fatal(err) | |||
| 35 | + | } | |||
| 36 | + | defer unix.Close(fresh) | |||
| 37 | + | unix.Close(old) | |||
| 38 | + | if _, err := unix.Write(dup, []byte{1, 0, 0, 0, 0, 0, 0, 0}); err != nil { | |||
| 39 | + | t.Fatal(err) | |||
| 40 | + | } | |||
| 41 | + | if err := p.raw.Control(func(ep uintptr) { | |||
| 42 | + | var events [1]unix.EpollEvent | |||
| 43 | + | if n, err := unix.EpollWait(int(ep), events[:], 0); n != 0 || err != nil { | |||
| 44 | + | t.Errorf("closed watched descriptor still makes epoll readable: %d %v", n, err) | |||
| 45 | + | } | |||
| 46 | + | }); err != nil { | |||
| 47 | + | t.Fatal(err) | |||
| 48 | + | } | |||
| 49 | + | done := make(chan error, 1) | |||
| 50 | + | go func() { _, err := p.Poll([]unix.PollFd{{Fd: int32(fresh), Events: unix.POLLIN}}, 20); done <- err }() | |||
| 51 | + | select { | |||
| 52 | + | case err := <-done: | |||
| 53 | + | if err != nil { | |||
| 54 | + | t.Fatal(err) | |||
| 55 | + | } | |||
| 56 | + | case <-time.After(time.Second): | |||
| 57 | + | p.Close() | |||
| 58 | + | t.Fatal("stale epoll readiness prevents timeout") | |||
| 59 | + | } | |||
| 60 | + | } | |||
M internal/adapters/fdwait/poller.go +21 -38
| @@ -13,9 +13,8 @@ import ( | |||
| 13 | 13 | // Poller owns an epoll descriptor and is used by one goroutine. | |
| 14 | 14 | // Watched descriptors remain owned by the caller. | |
| 15 | 15 | type Poller struct { | |
| 16 | - | file *os.File | |
| 17 | - | raw syscall.RawConn | |
| 18 | - | registered map[int]struct{} | |
| 16 | + | file *os.File | |
| 17 | + | raw syscall.RawConn | |
| 19 | 18 | } | |
| 20 | 19 | ||
| 21 | 20 | func New() (*Poller, error) { | |
| @@ -33,52 +32,36 @@ func New() (*Poller, error) { | |||
| 33 | 32 | f.Close() | |
| 34 | 33 | return nil, err | |
| 35 | 34 | } | |
| 36 | - | return &Poller{file: f, raw: raw, registered: make(map[int]struct{})}, nil | |
| 35 | + | return &Poller{file: f, raw: raw}, nil | |
| 37 | 36 | } | |
| 38 | - | ||
| 39 | 37 | func (p *Poller) Close() error { return p.file.Close() } | |
| 40 | 38 | ||
| 41 | - | // Poll has the readable/error semantics of unix.Poll. A negative timeout | |
| 42 | - | // waits indefinitely. Timeout expiration returns zero, not an error. | |
| 39 | + | // Poll waits for readable/error events. It never consumes caller events. | |
| 40 | + | // Each call explicitly removes its epoll registrations before the caller can | |
| 41 | + | // close or recycle watched fds. Relying on close to remove them is unsafe: | |
| 42 | + | // another fd or a kernel fence may retain the open file description. | |
| 43 | 43 | func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) { | |
| 44 | - | // Remove registrations no longer requested. Closing a watched fd already | |
| 45 | - | // removes it from epoll, so ENOENT and EBADF are harmless here. | |
| 46 | - | err := p.raw.Control(func(ep uintptr) { | |
| 47 | - | for fd := range p.registered { | |
| 48 | - | found := false | |
| 49 | - | for _, f := range fds { | |
| 50 | - | if int(f.Fd) == fd { | |
| 51 | - | found = true | |
| 52 | - | break | |
| 53 | - | } | |
| 54 | - | } | |
| 55 | - | if !found { | |
| 56 | - | unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, fd, nil) | |
| 57 | - | delete(p.registered, fd) | |
| 58 | - | } | |
| 59 | - | } | |
| 60 | - | }) | |
| 61 | - | if err != nil { | |
| 62 | - | return 0, err | |
| 63 | - | } | |
| 64 | 44 | var ctlErr error | |
| 65 | - | err = p.raw.Control(func(ep uintptr) { | |
| 45 | + | added := 0 | |
| 46 | + | err := p.raw.Control(func(ep uintptr) { | |
| 66 | 47 | for _, f := range fds { | |
| 67 | - | fd := int(f.Fd) | |
| 68 | - | if fd < 0 { | |
| 48 | + | if f.Fd < 0 { | |
| 49 | + | added++ | |
| 69 | 50 | continue | |
| 70 | 51 | } | |
| 71 | - | // MOD detects a descriptor number recycled after its old file closed. | |
| 72 | - | ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: int32(fd)} | |
| 73 | - | e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_MOD, fd, &ev) | |
| 74 | - | if errors.Is(e, unix.ENOENT) { | |
| 75 | - | e = unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, fd, &ev) | |
| 76 | - | } | |
| 77 | - | if e != nil { | |
| 52 | + | ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: f.Fd} | |
| 53 | + | if e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, int(f.Fd), &ev); e != nil { | |
| 78 | 54 | ctlErr = e | |
| 79 | 55 | return | |
| 80 | 56 | } | |
| 81 | - | p.registered[fd] = struct{}{} | |
| 57 | + | added++ | |
| 58 | + | } | |
| 59 | + | }) | |
| 60 | + | defer p.raw.Control(func(ep uintptr) { | |
| 61 | + | for _, f := range fds[:added] { | |
| 62 | + | if f.Fd >= 0 { | |
| 63 | + | unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, int(f.Fd), nil) | |
| 64 | + | } | |
| 82 | 65 | } | |
| 83 | 66 | }) | |
| 84 | 67 | if err != nil { | |
| @@ -13,9 +13,8 @@ import ( | |||||
| 13 | // Poller owns an epoll descriptor and is used by one goroutine. | 13 | // Poller owns an epoll descriptor and is used by one goroutine. | ||
| 14 | // Watched descriptors remain owned by the caller. | 14 | // Watched descriptors remain owned by the caller. | ||
| 15 | type Poller struct { | 15 | type Poller struct { | ||
| 16 | - | file *os.File | 16 | + | file *os.File |
| 17 | - | raw syscall.RawConn | 17 | + | raw syscall.RawConn |
| 18 | - | registered map[int]struct{} | |||
| 19 | } | 18 | } | ||
| 20 | 19 | ||||
| 21 | func New() (*Poller, error) { | 20 | func New() (*Poller, error) { | ||
| @@ -33,52 +32,36 @@ func New() (*Poller, error) { | |||||
| 33 | f.Close() | 32 | f.Close() | ||
| 34 | return nil, err | 33 | return nil, err | ||
| 35 | } | 34 | } | ||
| 36 | - | return &Poller{file: f, raw: raw, registered: make(map[int]struct{})}, nil | 35 | + | return &Poller{file: f, raw: raw}, nil |
| 37 | } | 36 | } | ||
| 38 | - | ||||
| 39 | func (p *Poller) Close() error { return p.file.Close() } | 37 | func (p *Poller) Close() error { return p.file.Close() } | ||
| 40 | 38 | ||||
| 41 | - | // Poll has the readable/error semantics of unix.Poll. A negative timeout | 39 | + | // Poll waits for readable/error events. It never consumes caller events. |
| 42 | - | // waits indefinitely. Timeout expiration returns zero, not an error. | 40 | + | // Each call explicitly removes its epoll registrations before the caller can |
| 41 | + | // close or recycle watched fds. Relying on close to remove them is unsafe: | |||
| 42 | + | // another fd or a kernel fence may retain the open file description. | |||
| 43 | func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) { | 43 | func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) { | ||
| 44 | - | // Remove registrations no longer requested. Closing a watched fd already | |||
| 45 | - | // removes it from epoll, so ENOENT and EBADF are harmless here. | |||
| 46 | - | err := p.raw.Control(func(ep uintptr) { | |||
| 47 | - | for fd := range p.registered { | |||
| 48 | - | found := false | |||
| 49 | - | for _, f := range fds { | |||
| 50 | - | if int(f.Fd) == fd { | |||
| 51 | - | found = true | |||
| 52 | - | break | |||
| 53 | - | } | |||
| 54 | - | } | |||
| 55 | - | if !found { | |||
| 56 | - | unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, fd, nil) | |||
| 57 | - | delete(p.registered, fd) | |||
| 58 | - | } | |||
| 59 | - | } | |||
| 60 | - | }) | |||
| 61 | - | if err != nil { | |||
| 62 | - | return 0, err | |||
| 63 | - | } | |||
| 64 | var ctlErr error | 44 | var ctlErr error | ||
| 65 | - | err = p.raw.Control(func(ep uintptr) { | 45 | + | added := 0 |
| 46 | + | err := p.raw.Control(func(ep uintptr) { | |||
| 66 | for _, f := range fds { | 47 | for _, f := range fds { | ||
| 67 | - | fd := int(f.Fd) | 48 | + | if f.Fd < 0 { |
| 68 | - | if fd < 0 { | 49 | + | added++ |
| 69 | continue | 50 | continue | ||
| 70 | } | 51 | } | ||
| 71 | - | // MOD detects a descriptor number recycled after its old file closed. | 52 | + | ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: f.Fd} |
| 72 | - | ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: int32(fd)} | 53 | + | if e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, int(f.Fd), &ev); e != nil { |
| 73 | - | e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_MOD, fd, &ev) | |||
| 74 | - | if errors.Is(e, unix.ENOENT) { | |||
| 75 | - | e = unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, fd, &ev) | |||
| 76 | - | } | |||
| 77 | - | if e != nil { | |||
| 78 | ctlErr = e | 54 | ctlErr = e | ||
| 79 | return | 55 | return | ||
| 80 | } | 56 | } | ||
| 81 | - | p.registered[fd] = struct{}{} | 57 | + | added++ |
| 58 | + | } | |||
| 59 | + | }) | |||
| 60 | + | defer p.raw.Control(func(ep uintptr) { | |||
| 61 | + | for _, f := range fds[:added] { | |||
| 62 | + | if f.Fd >= 0 { | |||
| 63 | + | unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, int(f.Fd), nil) | |||
| 64 | + | } | |||
| 82 | } | 65 | } | ||
| 83 | }) | 66 | }) | ||
| 84 | if err != nil { | 67 | if err != nil { | ||
M internal/adapters/fdwait/poller_test.go +7 -2
| @@ -52,8 +52,13 @@ func TestPollReadinessTimeoutAndReplacement(t *testing.T) { | |||
| 52 | 52 | if n, err := p.Poll(fds, 1000); n != 1 || err != nil { | |
| 53 | 53 | t.Fatalf("replacement: %d %v", n, err) | |
| 54 | 54 | } | |
| 55 | - | if _, ok := p.registered[fd]; ok { | |
| 56 | - | t.Fatal("old registration retained") | |
| 55 | + | if err := p.raw.Control(func(ep uintptr) { | |
| 56 | + | var events [1]unix.EpollEvent | |
| 57 | + | if n, err := unix.EpollWait(int(ep), events[:], 0); n != 0 || err != nil { | |
| 58 | + | t.Errorf("registration retained after Poll returned: %d %v", n, err) | |
| 59 | + | } | |
| 60 | + | }); err != nil { | |
| 61 | + | t.Fatal(err) | |
| 57 | 62 | } | |
| 58 | 63 | } | |
| 59 | 64 | ||
| @@ -52,8 +52,13 @@ func TestPollReadinessTimeoutAndReplacement(t *testing.T) { | |||||
| 52 | if n, err := p.Poll(fds, 1000); n != 1 || err != nil { | 52 | if n, err := p.Poll(fds, 1000); n != 1 || err != nil { | ||
| 53 | t.Fatalf("replacement: %d %v", n, err) | 53 | t.Fatalf("replacement: %d %v", n, err) | ||
| 54 | } | 54 | } | ||
| 55 | - | if _, ok := p.registered[fd]; ok { | 55 | + | if err := p.raw.Control(func(ep uintptr) { |
| 56 | - | t.Fatal("old registration retained") | 56 | + | var events [1]unix.EpollEvent |
| 57 | + | if n, err := unix.EpollWait(int(ep), events[:], 0); n != 0 || err != nil { | |||
| 58 | + | t.Errorf("registration retained after Poll returned: %d %v", n, err) | |||
| 59 | + | } | |||
| 60 | + | }); err != nil { | |||
| 61 | + | t.Fatal(err) | |||
| 57 | } | 62 | } | ||
| 58 | } | 63 | } | ||
| 59 | 64 | ||||