add convenient wrapper for eventfd

The same create/write/read pattern is copied around several places. It's easier
to understand in a package with names and comments, and we can reuse the smart
blocking code in package rawfile.

PiperOrigin-RevId: 401647108

9 files changed, 45 insertions, 65 deletions

diff --git a/pkg/tcpip/link/sharedmem/BUILD b/pkg/tcpip/link/sharedmem/BUILD
index 6c35aeecf..f8076d83c 100644
--- a/pkg/tcpip/link/sharedmem/BUILD
+++ b/pkg/tcpip/link/sharedmem/BUILD
@@ -17,6 +17,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/cleanup",
+        "//pkg/eventfd",
         "//pkg/log",
         "//pkg/sync",
         "//pkg/tcpip",
diff --git a/pkg/tcpip/link/sharedmem/queuepair.go b/pkg/tcpip/link/sharedmem/queuepair.go
index 5fa6d91f0..b12647fdd 100644
--- a/pkg/tcpip/link/sharedmem/queuepair.go
+++ b/pkg/tcpip/link/sharedmem/queuepair.go
@@ -22,6 +22,7 @@ import (
 	"io/ioutil"
 
 	"golang.org/x/sys/unix"
+	"gvisor.dev/gvisor/pkg/eventfd"
 )
 
 const (
@@ -116,25 +117,25 @@ type queueSizes struct {
 
 func createQueueFDs(s queueSizes) (QueueConfig, error) {
 	success := false
-	var fd uintptr
+	var eventFD eventfd.Eventfd
 	var dataFD, txPipeFD, rxPipeFD, sharedDataFD int
 	defer func() {
 		if success {
 			return
 		}
 		closeFDs(QueueConfig{
-			EventFD:      int(fd),
+			EventFD:      eventFD,
 			DataFD:       dataFD,
 			TxPipeFD:     txPipeFD,
 			RxPipeFD:     rxPipeFD,
 			SharedDataFD: sharedDataFD,
 		})
 	}()
-	eventFD, _, errno := unix.RawSyscall(unix.SYS_EVENTFD2, 0, 0, 0)
-	if errno != 0 {
-		return QueueConfig{}, fmt.Errorf("eventfd failed: %v", error(errno))
+	eventFD, err := eventfd.Create()
+	if err != nil {
+		return QueueConfig{}, fmt.Errorf("eventfd failed: %v", err)
 	}
-	dataFD, err := createFile(s.dataSize, false)
+	dataFD, err = createFile(s.dataSize, false)
 	if err != nil {
 		return QueueConfig{}, fmt.Errorf("failed to create dataFD: %s", err)
 	}
@@ -152,7 +153,7 @@ func createQueueFDs(s queueSizes) (QueueConfig, error) {
 	}
 	success = true
 	return QueueConfig{
-		EventFD:      int(eventFD),
+		EventFD:      eventFD,
 		DataFD:       dataFD,
 		TxPipeFD:     txPipeFD,
 		RxPipeFD:     rxPipeFD,
@@ -191,7 +192,7 @@ func createFile(size int64, initQueue bool) (fd int, err error) {
 
 func closeFDs(c QueueConfig) {
 	unix.Close(c.DataFD)
-	unix.Close(c.EventFD)
+	c.EventFD.Close()
 	unix.Close(c.TxPipeFD)
 	unix.Close(c.RxPipeFD)
 	unix.Close(c.SharedDataFD)
diff --git a/pkg/tcpip/link/sharedmem/rx.go b/pkg/tcpip/link/sharedmem/rx.go
index 399317335..87747dcc7 100644
--- a/pkg/tcpip/link/sharedmem/rx.go
+++ b/pkg/tcpip/link/sharedmem/rx.go
@@ -21,7 +21,7 @@ import (
 	"sync/atomic"
 
 	"golang.org/x/sys/unix"
-	"gvisor.dev/gvisor/pkg/tcpip/link/rawfile"
+	"gvisor.dev/gvisor/pkg/eventfd"
 	"gvisor.dev/gvisor/pkg/tcpip/link/sharedmem/queue"
 )
 
@@ -30,7 +30,7 @@ type rx struct {
 	data       []byte
 	sharedData []byte
 	q          queue.Rx
-	eventFD    int
+	eventFD    eventfd.Eventfd
 }
 
 // init initializes all state needed by the rx queue based on the information
@@ -68,7 +68,7 @@ func (r *rx) init(mtu uint32, c *QueueConfig) error {
 
 	// Duplicate the eventFD so that caller can close it but we can still
 	// use it.
-	efd, err := unix.Dup(c.EventFD)
+	efd, err := c.EventFD.Dup()
 	if err != nil {
 		unix.Munmap(txPipe)
 		unix.Munmap(rxPipe)
@@ -77,16 +77,6 @@ func (r *rx) init(mtu uint32, c *QueueConfig) error {
 		return err
 	}
 
-	// Set the eventfd as non-blocking.
-	if err := unix.SetNonblock(efd, true); err != nil {
-		unix.Munmap(txPipe)
-		unix.Munmap(rxPipe)
-		unix.Munmap(data)
-		unix.Munmap(sharedData)
-		unix.Close(efd)
-		return err
-	}
-
 	// Initialize state based on buffers.
 	r.q.Init(txPipe, rxPipe, sharedDataPointer(sharedData))
 	r.data = data
@@ -105,13 +95,13 @@ func (r *rx) cleanup() {
 
 	unix.Munmap(r.data)
 	unix.Munmap(r.sharedData)
-	unix.Close(r.eventFD)
+	r.eventFD.Close()
 }
 
 // notify writes to the tx.eventFD to indicate to the peer that there is data to
 // be read.
 func (r *rx) notify() {
-	unix.Write(r.eventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
+	r.eventFD.Notify()
 }
 
 // postAndReceive posts the provided buffers (if any), and then tries to read
@@ -128,8 +118,7 @@ func (r *rx) postAndReceive(b []queue.RxBuffer, stopRequested *uint32) ([]queue.
 	if len(b) != 0 && !r.q.PostBuffers(b) {
 		r.q.EnableNotification()
 		for !r.q.PostBuffers(b) {
-			var tmp [8]byte
-			rawfile.BlockingRead(r.eventFD, tmp[:])
+			r.eventFD.Wait()
 			if atomic.LoadUint32(stopRequested) != 0 {
 				r.q.DisableNotification()
 				return nil, 0
@@ -153,8 +142,7 @@ func (r *rx) postAndReceive(b []queue.RxBuffer, stopRequested *uint32) ([]queue.
 		}
 
 		// Wait for notification.
-		var tmp [8]byte
-		rawfile.BlockingRead(r.eventFD, tmp[:])
+		r.eventFD.Wait()
 		if atomic.LoadUint32(stopRequested) != 0 {
 			r.q.DisableNotification()
 			return nil, 0
diff --git a/pkg/tcpip/link/sharedmem/server_rx.go b/pkg/tcpip/link/sharedmem/server_rx.go
index 2ad8bf650..6ea21ffd1 100644
--- a/pkg/tcpip/link/sharedmem/server_rx.go
+++ b/pkg/tcpip/link/sharedmem/server_rx.go
@@ -20,7 +20,7 @@ package sharedmem
 import (
 	"golang.org/x/sys/unix"
 	"gvisor.dev/gvisor/pkg/cleanup"
-	"gvisor.dev/gvisor/pkg/tcpip/link/rawfile"
+	"gvisor.dev/gvisor/pkg/eventfd"
 	"gvisor.dev/gvisor/pkg/tcpip/link/sharedmem/pipe"
 	"gvisor.dev/gvisor/pkg/tcpip/link/sharedmem/queue"
 )
@@ -38,7 +38,7 @@ type serverRx struct {
 	data []byte
 
 	// eventFD is used to notify the peer when transmission is completed.
-	eventFD int
+	eventFD eventfd.Eventfd
 
 	// sharedData the memory region to use to enable/disable notifications.
 	sharedData []byte
@@ -78,16 +78,11 @@ func (s *serverRx) init(c *QueueConfig) error {
 
 	// Duplicate the eventFD so that caller can close it but we can still
 	// use it.
-	efd, err := unix.Dup(c.EventFD)
+	efd, err := c.EventFD.Dup()
 	if err != nil {
 		return err
 	}
-	cu.Add(func() { unix.Close(efd) })
-
-	// Set the eventfd as non-blocking.
-	if err := unix.SetNonblock(efd, true); err != nil {
-		return err
-	}
+	cu.Add(func() { efd.Close() })
 
 	s.packetPipe.Init(packetPipeMem)
 	s.completionPipe.Init(completionPipeMem)
@@ -104,7 +99,7 @@ func (s *serverRx) cleanup() {
 	unix.Munmap(s.completionPipe.Bytes())
 	unix.Munmap(s.data)
 	unix.Munmap(s.sharedData)
-	unix.Close(s.eventFD)
+	s.eventFD.Close()
 }
 
 // completionNotificationSize is size in bytes of a completion notification sent
@@ -143,6 +138,5 @@ func (s *serverRx) receive() []byte {
 }
 
 func (s *serverRx) waitForPackets() {
-	var tmp [8]byte
-	rawfile.BlockingRead(s.eventFD, tmp[:])
+	s.eventFD.Wait()
 }
diff --git a/pkg/tcpip/link/sharedmem/server_tx.go b/pkg/tcpip/link/sharedmem/server_tx.go
index 9370b2a46..13a82903f 100644
--- a/pkg/tcpip/link/sharedmem/server_tx.go
+++ b/pkg/tcpip/link/sharedmem/server_tx.go
@@ -20,6 +20,7 @@ package sharedmem
 import (
 	"golang.org/x/sys/unix"
 	"gvisor.dev/gvisor/pkg/cleanup"
+	"gvisor.dev/gvisor/pkg/eventfd"
 	"gvisor.dev/gvisor/pkg/tcpip/buffer"
 	"gvisor.dev/gvisor/pkg/tcpip/link/sharedmem/pipe"
 	"gvisor.dev/gvisor/pkg/tcpip/link/sharedmem/queue"
@@ -40,7 +41,7 @@ type serverTx struct {
 	data []byte
 
 	// eventFD is used to notify the peer when fill requests are fulfilled.
-	eventFD int
+	eventFD eventfd.Eventfd
 
 	// sharedData the memory region to use to enable/disable notifications.
 	sharedData []byte
@@ -80,16 +81,11 @@ func (s *serverTx) init(c *QueueConfig) error {
 
 	// Duplicate the eventFD so that caller can close it but we can still
 	// use it.
-	efd, err := unix.Dup(c.EventFD)
+	efd, err := c.EventFD.Dup()
 	if err != nil {
 		return err
 	}
-	cu.Add(func() { unix.Close(efd) })
-
-	// Set the eventfd as non-blocking.
-	if err := unix.SetNonblock(efd, true); err != nil {
-		return err
-	}
+	cu.Add(func() { efd.Close() })
 
 	cu.Release()
 
@@ -107,7 +103,7 @@ func (s *serverTx) cleanup() {
 	unix.Munmap(s.completionPipe.Bytes())
 	unix.Munmap(s.data)
 	unix.Munmap(s.sharedData)
-	unix.Close(s.eventFD)
+	s.eventFD.Close()
 }
 
 // fillPacket copies the data in the provided views into buffers pulled from the
@@ -175,5 +171,5 @@ func (s *serverTx) transmit(views []buffer.View) bool {
 }
 
 func (s *serverTx) notify() {
-	unix.Write(s.eventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
+	s.eventFD.Notify()
 }
diff --git a/pkg/tcpip/link/sharedmem/sharedmem.go b/pkg/tcpip/link/sharedmem/sharedmem.go
index e2a8c4863..bcb37a465 100644
--- a/pkg/tcpip/link/sharedmem/sharedmem.go
+++ b/pkg/tcpip/link/sharedmem/sharedmem.go
@@ -27,7 +27,7 @@ import (
 	"fmt"
 	"sync/atomic"
 
-	"golang.org/x/sys/unix"
+	"gvisor.dev/gvisor/pkg/eventfd"
 	"gvisor.dev/gvisor/pkg/log"
 	"gvisor.dev/gvisor/pkg/sync"
 	"gvisor.dev/gvisor/pkg/tcpip"
@@ -49,7 +49,7 @@ type QueueConfig struct {
 
 	// EventFD is a file descriptor for the event that is signaled when
 	// data is becomes available in this queue.
-	EventFD int
+	EventFD eventfd.Eventfd
 
 	// TxPipeFD is a file descriptor for the tx pipe associated with the
 	// queue.
@@ -70,7 +70,7 @@ type QueueConfig struct {
 // of FDs matches when reconstructing the config when serialized or sent
 // as part of control messages.
 func (q *QueueConfig) FDs() []int {
-	return []int{q.DataFD, q.EventFD, q.TxPipeFD, q.RxPipeFD, q.SharedDataFD}
+	return []int{q.DataFD, q.EventFD.FD(), q.TxPipeFD, q.RxPipeFD, q.SharedDataFD}
 }
 
 // QueueConfigFromFDs constructs a QueueConfig out of a slice of ints where each
@@ -84,7 +84,7 @@ func QueueConfigFromFDs(fds []int) (QueueConfig, error) {
 	}
 	return QueueConfig{
 		DataFD:       fds[0],
-		EventFD:      fds[1],
+		EventFD:      eventfd.Wrap(fds[1]),
 		TxPipeFD:     fds[2],
 		RxPipeFD:     fds[3],
 		SharedDataFD: fds[4],
@@ -223,7 +223,7 @@ func (e *endpoint) Close() {
 	// Tell dispatch goroutine to stop, then write to the eventfd so that
 	// it wakes up in case it's sleeping.
 	atomic.StoreUint32(&e.stopRequested, 1)
-	unix.Write(e.rx.eventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
+	e.rx.eventFD.Notify()
 
 	// Cleanup the queues inline if the worker hasn't started yet; we also
 	// know it won't start from now on because stopRequested is set to 1.
diff --git a/pkg/tcpip/link/sharedmem/sharedmem_server.go b/pkg/tcpip/link/sharedmem/sharedmem_server.go
index 16feb64b2..ccc84989d 100644
--- a/pkg/tcpip/link/sharedmem/sharedmem_server.go
+++ b/pkg/tcpip/link/sharedmem/sharedmem_server.go
@@ -20,7 +20,6 @@ package sharedmem
 import (
 	"sync/atomic"
 
-	"golang.org/x/sys/unix"
 	"gvisor.dev/gvisor/pkg/sync"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/buffer"
@@ -122,7 +121,7 @@ func (e *serverEndpoint) Close() {
 	// Tell dispatch goroutine to stop, then write to the eventfd so that it wakes
 	// up in case it's sleeping.
 	atomic.StoreUint32(&e.stopRequested, 1)
-	unix.Write(e.rx.eventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
+	e.rx.eventFD.Notify()
 
 	// Cleanup the queues inline if the worker hasn't started yet; we also know it
 	// won't start from now on because stopRequested is set to 1.
diff --git a/pkg/tcpip/link/sharedmem/sharedmem_test.go b/pkg/tcpip/link/sharedmem/sharedmem_test.go
index bb094da63..66ffc33b8 100644
--- a/pkg/tcpip/link/sharedmem/sharedmem_test.go
+++ b/pkg/tcpip/link/sharedmem/sharedmem_test.go
@@ -619,7 +619,7 @@ func TestSimpleReceive(t *testing.T) {
 		// Push completion.
 		c.pushRxCompletion(uint32(len(contents)), bufs)
 		c.rxq.rx.Flush()
-		unix.Write(c.rxCfg.EventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
+		c.rxCfg.EventFD.Notify()
 
 		// Wait for packet to be received, then check it.
 		c.waitForPackets(1, time.After(5*time.Second), "Timeout waiting for packet")
@@ -665,7 +665,7 @@ func TestRxBuffersReposted(t *testing.T) {
 		// Complete the buffer.
 		c.pushRxCompletion(buffers[i].Size, buffers[i:][:1])
 		c.rxq.rx.Flush()
-		unix.Write(c.rxCfg.EventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
+		c.rxCfg.EventFD.Notify()
 
 		// Wait for it to be reposted.
 		bi := queue.DecodeRxBufferHeader(pollPull(t, &c.rxq.tx, timeout, "Timeout waiting for buffer to be reposted"))
@@ -681,7 +681,7 @@ func TestRxBuffersReposted(t *testing.T) {
 		// Complete with two buffers.
 		c.pushRxCompletion(2*bufferSize, buffers[2*i:][:2])
 		c.rxq.rx.Flush()
-		unix.Write(c.rxCfg.EventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
+		c.rxCfg.EventFD.Notify()
 
 		// Wait for them to be reposted.
 		for j := 0; j < 2; j++ {
@@ -706,7 +706,7 @@ func TestReceivePostingIsFull(t *testing.T) {
 	first := queue.DecodeRxBufferHeader(pollPull(t, &c.rxq.tx, time.After(time.Second), "Timeout waiting for first buffer to be posted"))
 	c.pushRxCompletion(first.Size, []queue.RxBuffer{first})
 	c.rxq.rx.Flush()
-	unix.Write(c.rxCfg.EventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
+	c.rxCfg.EventFD.Notify()
 
 	// Check that packet is received.
 	c.waitForPackets(1, time.After(time.Second), "Timeout waiting for completed packet")
@@ -715,7 +715,7 @@ func TestReceivePostingIsFull(t *testing.T) {
 	second := queue.DecodeRxBufferHeader(pollPull(t, &c.rxq.tx, time.After(time.Second), "Timeout waiting for second buffer to be posted"))
 	c.pushRxCompletion(second.Size, []queue.RxBuffer{second})
 	c.rxq.rx.Flush()
-	unix.Write(c.rxCfg.EventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
+	c.rxCfg.EventFD.Notify()
 
 	// Check that no packet is received yet, as the worker is blocked trying
 	// to repost.
@@ -728,7 +728,7 @@ func TestReceivePostingIsFull(t *testing.T) {
 	// Flush tx queue, which will allow the first buffer to be reposted,
 	// and the second completion to be pulled.
 	c.rxq.tx.Flush()
-	unix.Write(c.rxCfg.EventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
+	c.rxCfg.EventFD.Notify()
 
 	// Check that second packet completes.
 	c.waitForPackets(1, time.After(time.Second), "Timeout waiting for second completed packet")
@@ -750,7 +750,7 @@ func TestCloseWhileWaitingToPost(t *testing.T) {
 	bi := queue.DecodeRxBufferHeader(pollPull(t, &c.rxq.tx, time.After(time.Second), "Timeout waiting for initial buffer to be posted"))
 	c.pushRxCompletion(bi.Size, []queue.RxBuffer{bi})
 	c.rxq.rx.Flush()
-	unix.Write(c.rxCfg.EventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
+	c.rxCfg.EventFD.Notify()
 
 	// Wait for packet to be indicated.
 	c.waitForPackets(1, time.After(time.Second), "Timeout waiting for completed packet")
diff --git a/pkg/tcpip/link/sharedmem/tx.go b/pkg/tcpip/link/sharedmem/tx.go
index 5ffcb8ab4..35e5bff12 100644
--- a/pkg/tcpip/link/sharedmem/tx.go
+++ b/pkg/tcpip/link/sharedmem/tx.go
@@ -18,6 +18,7 @@ import (
 	"math"
 
 	"golang.org/x/sys/unix"
+	"gvisor.dev/gvisor/pkg/eventfd"
 	"gvisor.dev/gvisor/pkg/tcpip/buffer"
 	"gvisor.dev/gvisor/pkg/tcpip/link/sharedmem/queue"
 )
@@ -32,7 +33,7 @@ type tx struct {
 	q            queue.Tx
 	ids          idManager
 	bufs         bufferManager
-	eventFD      int
+	eventFD      eventfd.Eventfd
 	sharedDataFD int
 }
 
@@ -148,7 +149,7 @@ func (t *tx) transmit(bufs ...buffer.View) bool {
 // notify writes to the tx.eventFD to indicate to the peer that there is data to
 // be read.
 func (t *tx) notify() {
-	unix.Write(t.eventFD, []byte{1, 0, 0, 0, 0, 0, 0, 0})
+	t.eventFD.Notify()
 }
 
 // getBuffer returns a memory region mapped to the full contents of the given