"errors"
"internal/syscall/unix"
"internal/testenv"
+ "io"
"io/fs"
"os"
"path/filepath"
"sync"
"syscall"
"testing"
+ "time"
)
func TestFifoEOF(t *testing.T) {
t.Error("pipe blocking after Fd")
}
}
+
+func TestFIFONonBlockingEOF(t *testing.T) {
+ fifoName := filepath.Join(t.TempDir(), "issue-66239-fifo")
+ if err := syscall.Mkfifo(fifoName, 0600); err != nil {
+ t.Fatalf("Error creating fifo: %v", err)
+ }
+
+ r, err := os.OpenFile(fifoName, os.O_RDONLY|syscall.O_NONBLOCK, os.ModeNamedPipe)
+ if err != nil {
+ t.Fatalf("Error opening fifo for read: %v", err)
+ }
+ defer r.Close()
+
+ w, err := os.OpenFile(fifoName, os.O_WRONLY, os.ModeNamedPipe)
+ if err != nil {
+ t.Fatalf("Error opening fifo for write: %v", err)
+ }
+ defer w.Close()
+
+ data := "Hello Gophers!"
+ if _, err := w.WriteString(data); err != nil {
+ t.Fatalf("Error writing to fifo: %v", err)
+ }
+
+ // Close the writer after a short delay to open a gap for the reader
+ // of FIFO to fall into polling. See https://go.dev/issue/66239#issuecomment-1987620476
+ time.AfterFunc(200*time.Millisecond, func() {
+ if err := w.Close(); err != nil {
+ t.Errorf("Error closing writer: %v", err)
+ }
+ })
+
+ buf := make([]byte, len(data))
+ n, err := io.ReadAtLeast(r, buf, len(data))
+ if n != len(data) || string(buf) != data || err != nil {
+ t.Errorf("ReadAtLeast: %v; got %q, want %q", err, buf, data)
+ return
+ }
+
+ // Loop reading from FIFO until EOF to ensure that the reader
+ // is not blocked infinitely, otherwise there is something wrong
+ // with the netpoller.
+ for {
+ _, err = r.Read(buf)
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil && !errors.Is(err, syscall.EAGAIN) {
+ t.Errorf("Error reading bytes from fifo: %v", err)
+ return
+ }
+ }
+}
return nil
}
- kind := kindNewFile
- appendMode := false
- if flags, err := unix.Fcntl(fdi, syscall.F_GETFL, 0); err == nil {
- if unix.HasNonblockFlag(flags) {
- kind = kindNonBlock
- }
- appendMode = flags&syscall.O_APPEND != 0
+ flags, err := unix.Fcntl(fdi, syscall.F_GETFL, 0)
+ if err != nil {
+ flags = 0
}
- f := newFile(fdi, name, kind)
- f.appendMode = appendMode
+ f := newFile(fdi, name, kindNewFile, unix.HasNonblockFlag(flags))
+ f.appendMode = flags&syscall.O_APPEND != 0
return f
}
panic("invalid FD")
}
- f := newFile(fd, name, kindNonBlock)
- f.nonblock = true // tell Fd to return blocking descriptor
- return f
+ return newFile(fd, name, kindSock, true)
}
// newFileKind describes the kind of file to newFile.
// kindNewFile means that the descriptor was passed to us via NewFile.
kindNewFile newFileKind = iota
// kindOpenFile means that the descriptor was opened using
- // Open, Create, or OpenFile (without O_NONBLOCK).
+ // Open, Create, or OpenFile.
kindOpenFile
// kindPipe means that the descriptor was opened using Pipe.
kindPipe
- // kindNonBlock means that the descriptor is already in
- // non-blocking mode.
- kindNonBlock
+ // kindSock means that the descriptor is a network file descriptor
+ // that was created from net package and was opened using net_newUnixFile.
+ kindSock
// kindNoPoll means that we should not put the descriptor into
// non-blocking mode, because we know it is not a pipe or FIFO.
// Used by openFdAt and openDirNolog for directories.
// newFile is like NewFile, but if called from OpenFile or Pipe
// (as passed in the kind parameter) it tries to add the file to
// the runtime poller.
-func newFile(fd int, name string, kind newFileKind) *File {
+func newFile(fd int, name string, kind newFileKind, nonBlocking bool) *File {
f := &File{&file{
pfd: poll.FD{
Sysfd: fd,
stdoutOrErr: fd == 1 || fd == 2,
}}
- pollable := kind == kindOpenFile || kind == kindPipe || kind == kindNonBlock
+ pollable := kind == kindOpenFile || kind == kindPipe || kind == kindSock || nonBlocking
- // If the caller passed a non-blocking filedes (kindNonBlock),
- // we assume they know what they are doing so we allow it to be
- // used with kqueue.
+ // Things like regular files and FIFOs in kqueue on *BSD/Darwin
+ // may not work properly (or accurately according to its manual).
+ // As a result, we should avoid adding those to the kqueue-based
+ // netpoller. Check out #19093, #24164, and #66239 for more contexts.
+ //
+ // If the fd was passed to us via any path other than OpenFile,
+ // we assume those callers know what they were doing, so we won't
+ // perform this check and allow it to be added to the kqueue.
if kind == kindOpenFile {
switch runtime.GOOS {
case "darwin", "ios", "dragonfly", "freebsd", "netbsd", "openbsd":
clearNonBlock := false
if pollable {
- if kind == kindNonBlock {
- // The descriptor is already in non-blocking mode.
- // We only set f.nonblock if we put the file into
- // non-blocking mode.
+ // The descriptor is already in non-blocking mode.
+ // We only set f.nonblock if we put the file into
+ // non-blocking mode.
+ if nonBlocking {
+ // See the comments on net_newUnixFile.
+ if kind == kindSock {
+ f.nonblock = true // tell Fd to return blocking descriptor
+ }
} else if err := syscall.SetNonblock(fd, true); err == nil {
f.nonblock = true
clearNonBlock = true
syscall.CloseOnExec(r)
}
- kind := kindOpenFile
- if unix.HasNonblockFlag(flag) {
- kind = kindNonBlock
- }
-
- f := newFile(r, name, kind)
+ f := newFile(r, name, kindOpenFile, unix.HasNonblockFlag(flag))
f.pfd.SysFile = s
return f, nil
}
syscall.CloseOnExec(r)
}
- f := newFile(r, name, kindNoPoll)
+ f := newFile(r, name, kindNoPoll, false)
f.pfd.SysFile = s
return f, nil
}