runtime: for kqueue treat EVFILT_READ with EV_EOF as permitting a write

On systems that use kqueue, we always register descriptors for both
EVFILT_READ and EVFILT_WRITE. On at least FreeBSD and OpenBSD, when
the write end of a pipe is registered for EVFILT_READ and EVFILT_WRITE
events, and the read end of the pipe is closed, kqueue reports an
EVFILT_READ event with EV_EOF set, but does not report an EVFILT_WRITE
event. Since the write to the pipe is waiting for an EVFILT_WRITE
event, closing the read end of a pipe can cause the write end to hang
rather than attempt another write which will fail with EPIPE.

Fix this by treating EVFILT_READ with EV_EOF set as making both reads
and writes ready to proceed.

The real test for this is in CL 71770, which tests using various
timeouts with pipes.

Updates #22114

Change-Id: Ib23fbaaddbccd8eee77bdf18f27a7f0aa50e2742
Reviewed-on: https://go-review.googlesource.com/71973Reviewed-by: Matthew Dempsky <mdempsky@google.com>

src/internal/poll/fd_unix.go

@@ -411,6 +411,15 @@ func (fd *FD) WaitWrite() error {
 	return fd.pd.waitWrite(fd.isFile)
 }
 
+// WriteOnce is for testing only. It makes a single write call.
+func (fd *FD) WriteOnce(p []byte) (int, error) {
+	if err := fd.writeLock(); err != nil {
+		return 0, err
+	}
+	defer fd.writeUnlock()
+	return syscall.Write(fd.Sysfd, p)
+}
+
 // RawControl invokes the user-defined function f for a non-IO
 // operation.
 func (fd *FD) RawControl(f func(uintptr)) error {

src/net/write_unix_test.go

+// Copyright 2017 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// +build darwin dragonfly freebsd linux netbsd openbsd solaris
+
+package net
+
+import (
+	"bytes"
+	"syscall"
+	"testing"
+	"time"
+)
+
+// Test that a client can't trigger an endless loop of write system
+// calls on the server by shutting down the write side on the client.
+// Possibility raised in the discussion of https://golang.org/cl/71973.
+func TestEndlessWrite(t *testing.T) {
+	t.Parallel()
+	c := make(chan bool)
+	server := func(cs *TCPConn) error {
+		cs.CloseWrite()
+		<-c
+		return nil
+	}
+	client := func(ss *TCPConn) error {
+		// Tell the server to return when we return.
+		defer close(c)
+
+		// Loop writing to the server. The server is not reading
+		// anything, so this will eventually block, and then time out.
+		b := bytes.Repeat([]byte{'a'}, 8192)
+		cagain := 0
+		for {
+			n, err := ss.conn.fd.pfd.WriteOnce(b)
+			if n > 0 {
+				cagain = 0
+			}
+			switch err {
+			case nil:
+			case syscall.EAGAIN:
+				if cagain == 0 {
+					// We've written enough data to
+					// start blocking. Set a deadline
+					// so that we will stop.
+					ss.SetWriteDeadline(time.Now().Add(5 * time.Millisecond))
+				}
+				cagain++
+				if cagain > 20 {
+					t.Error("looping on EAGAIN")
+					return nil
+				}
+				if err = ss.conn.fd.pfd.WaitWrite(); err != nil {
+					t.Logf("client WaitWrite: %v", err)
+					return nil
+				}
+			default:
+				// We expect to eventually get an error.
+				t.Logf("client WriteOnce: %v", err)
+				return nil
+			}
+		}
+	}
+	withTCPConnPair(t, client, server)
+}

src/runtime/defs1_netbsd_386.go

@@ -79,6 +79,7 @@ const (
 	_EV_CLEAR     = 0x20
 	_EV_RECEIPT   = 0
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = 0x0
 	_EVFILT_WRITE = 0x1
 )

src/runtime/defs1_netbsd_amd64.go

@@ -79,6 +79,7 @@ const (
 	_EV_CLEAR     = 0x20
 	_EV_RECEIPT   = 0
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = 0x0
 	_EVFILT_WRITE = 0x1
 )

src/runtime/defs1_netbsd_arm.go

@@ -79,6 +79,7 @@ const (
 	_EV_CLEAR     = 0x20
 	_EV_RECEIPT   = 0
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = 0x0
 	_EVFILT_WRITE = 0x1
 )

src/runtime/defs_darwin.go

@@ -139,6 +139,7 @@ const (
 	EV_CLEAR     = C.EV_CLEAR
 	EV_RECEIPT   = C.EV_RECEIPT
 	EV_ERROR     = C.EV_ERROR
+	EV_EOF       = C.EV_EOF
 	EVFILT_READ  = C.EVFILT_READ
 	EVFILT_WRITE = C.EVFILT_WRITE
 )

src/runtime/defs_darwin_386.go

@@ -118,6 +118,7 @@ const (
 	_EV_CLEAR     = 0x20
 	_EV_RECEIPT   = 0x40
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = -0x1
 	_EVFILT_WRITE = -0x2
 )

src/runtime/defs_darwin_amd64.go

@@ -118,6 +118,7 @@ const (
 	_EV_CLEAR     = 0x20
 	_EV_RECEIPT   = 0x40
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = -0x1
 	_EVFILT_WRITE = -0x2
 )

src/runtime/defs_darwin_arm.go

@@ -120,6 +120,7 @@ const (
 	_EV_CLEAR     = 0x20
 	_EV_RECEIPT   = 0x40
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = -0x1
 	_EVFILT_WRITE = -0x2
 )

src/runtime/defs_darwin_arm64.go

@@ -118,6 +118,7 @@ const (
 	_EV_CLEAR     = 0x20
 	_EV_RECEIPT   = 0x40
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = -0x1
 	_EVFILT_WRITE = -0x2
 )

src/runtime/defs_dragonfly.go

@@ -103,6 +103,7 @@ const (
 	EV_DELETE    = C.EV_DELETE
 	EV_CLEAR     = C.EV_CLEAR
 	EV_ERROR     = C.EV_ERROR
+	EV_EOF       = C.EV_EOF
 	EVFILT_READ  = C.EVFILT_READ
 	EVFILT_WRITE = C.EVFILT_WRITE
 )

src/runtime/defs_dragonfly_amd64.go

@@ -82,6 +82,7 @@ const (
 	_EV_DELETE    = 0x2
 	_EV_CLEAR     = 0x20
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = -0x1
 	_EVFILT_WRITE = -0x2
 )

src/runtime/defs_freebsd.go

@@ -125,6 +125,7 @@ const (
 	EV_CLEAR     = C.EV_CLEAR
 	EV_RECEIPT   = C.EV_RECEIPT
 	EV_ERROR     = C.EV_ERROR
+	EV_EOF       = C.EV_EOF
 	EVFILT_READ  = C.EVFILT_READ
 	EVFILT_WRITE = C.EVFILT_WRITE
 )

src/runtime/defs_freebsd_386.go

@@ -95,6 +95,7 @@ const (
 	_EV_CLEAR     = 0x20
 	_EV_RECEIPT   = 0x40
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = -0x1
 	_EVFILT_WRITE = -0x2
 )

src/runtime/defs_freebsd_amd64.go

@@ -95,6 +95,7 @@ const (
 	_EV_CLEAR     = 0x20
 	_EV_RECEIPT   = 0x40
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = -0x1
 	_EVFILT_WRITE = -0x2
 )

src/runtime/defs_freebsd_arm.go

@@ -95,6 +95,7 @@ const (
 	_EV_CLEAR     = 0x20
 	_EV_RECEIPT   = 0x40
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = -0x1
 	_EVFILT_WRITE = -0x2
 )

src/runtime/defs_netbsd.go

@@ -105,6 +105,7 @@ const (
 	EV_CLEAR     = C.EV_CLEAR
 	EV_RECEIPT   = 0
 	EV_ERROR     = C.EV_ERROR
+	EV_EOF       = C.EV_EOF
 	EVFILT_READ  = C.EVFILT_READ
 	EVFILT_WRITE = C.EVFILT_WRITE
 )

src/runtime/defs_openbsd.go

@@ -100,6 +100,7 @@ const (
 	EV_DELETE    = C.EV_DELETE
 	EV_CLEAR     = C.EV_CLEAR
 	EV_ERROR     = C.EV_ERROR
+	EV_EOF       = C.EV_EOF
 	EVFILT_READ  = C.EVFILT_READ
 	EVFILT_WRITE = C.EVFILT_WRITE
 )

src/runtime/defs_openbsd_386.go

@@ -80,6 +80,7 @@ const (
 	_EV_DELETE    = 0x2
 	_EV_CLEAR     = 0x20
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = -0x1
 	_EVFILT_WRITE = -0x2
 )

src/runtime/defs_openbsd_amd64.go

@@ -80,6 +80,7 @@ const (
 	_EV_DELETE    = 0x2
 	_EV_CLEAR     = 0x20
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = -0x1
 	_EVFILT_WRITE = -0x2
 )

src/runtime/defs_openbsd_arm.go

@@ -80,6 +80,7 @@ const (
 	_EV_DELETE    = 0x2
 	_EV_CLEAR     = 0x20
 	_EV_ERROR     = 0x4000
+	_EV_EOF       = 0x8000
 	_EVFILT_READ  = -0x1
 	_EVFILT_WRITE = -0x2
 )

src/runtime/netpoll_kqueue.go

@@ -88,10 +88,23 @@ retry:
 	for i := 0; i < int(n); i++ {
 		ev := &events[i]
 		var mode int32
-		if ev.filter == _EVFILT_READ {
+		switch ev.filter {
+		case _EVFILT_READ:
 			mode += 'r'
-		}
-		if ev.filter == _EVFILT_WRITE {
+
+			// On some systems when the read end of a pipe
+			// is closed the write end will not get a
+			// _EVFILT_WRITE event, but will get a
+			// _EVFILT_READ event with EV_EOF set.
+			// Note that setting 'w' here just means that we
+			// will wake up a goroutine waiting to write;
+			// that goroutine will try the write again,
+			// and the appropriate thing will happen based
+			// on what that write returns (success, EPIPE, EAGAIN).
+			if ev.flags&_EV_EOF != 0 {
+				mode += 'w'
+			}
+		case _EVFILT_WRITE:
 			mode += 'w'
 		}
 		if mode != 0 {