Repository: cilium/ebpf
Pull request #475 (2021-11-04 17:59:12)

epoll: factor out poller
The ringbuf package uses a copy of the perf epoll code. Move the code into
a shared package.


Patch:
diff --git a/internal/epoll/poller.go b/internal/epoll/poller.go
new file mode 100644
index 0000000..26e707b
--- /dev/null
+++ b/internal/epoll/poller.go
@@ -0,0 +1,207 @@
+package epoll
+
+import (
+	"fmt"
+	"math"
+	"os"
+	"runtime"
+	"sync"
+
+	"github.com/cilium/ebpf/internal"
+	"github.com/cilium/ebpf/internal/unix"
+)
+
+// Poller waits for readiness notifications from multiple file descriptors.
+//
+// The wait can be interrupted by calling Close.
+type Poller struct {
+	// mutexes protect the fields declared below them. If you need to
+	// acquire both at once you must lock epollMu before eventMu.
+	epollMu sync.Mutex
+	epollFd int
+
+	eventMu sync.Mutex
+	event   *eventFd
+}
+
+func New() (*Poller, error) {
+	epollFd, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
+	if err != nil {
+		return nil, fmt.Errorf("create epoll fd: %v", err)
+	}
+
+	p := &Poller{epollFd: epollFd}
+	p.event, err = newEventFd()
+	if err != nil {
+		unix.Close(epollFd)
+		return nil, err
+	}
+
+	if err := p.Add(p.event.raw, 0); err != nil {
+		unix.Close(epollFd)
+		p.event.close()
+		return nil, fmt.Errorf("add eventfd: %w", err)
+	}
+
+	runtime.SetFinalizer(p, (*Poller).Close)
+	return p, nil
+}
+
+// Close the poller.
+//
+// Interrupts any calls to Wait. Multiple calls to Close are valid, but subsequent
+// calls will return os.ErrClosed.
+func (p *Poller) Close() error {
+	runtime.SetFinalizer(p, nil)
+
+	// Interrupt Wait() via the event fd if it's currently blocked.
+	if err := p.wakeWait(); err != nil {
+		return err
+	}
+
+	// Acquire the lock. This ensures that Wait isn't running.
+	p.epollMu.Lock()
+	defer p.epollMu.Unlock()
+
+	// Prevent other calls to Close().
+	p.eventMu.Lock()
+	defer p.eventMu.Unlock()
+
+	if p.epollFd != -1 {
+		unix.Close(p.epollFd)
+		p.epollFd = -1
+	}
+
+	if p.event != nil {
+		p.event.close()
+		p.event = nil
+	}
+
+	return nil
+}
+
+// Add an fd to the poller.
+//
+// id is returned by Wait in the unix.EpollEvent.Pad field any may be zero. It
+// must not exceed math.MaxInt32.
+//
+// Add is blocked by Wait.
+func (p *Poller) Add(fd int, id int) error {
+	if int64(id) > math.MaxInt32 {
+		return fmt.Errorf("unsupported id: %d", id)
+	}
+
+	p.epollMu.Lock()
+	defer p.epollMu.Unlock()
+
+	if p.epollFd == -1 {
+		return fmt.Errorf("epoll add: %w", os.ErrClosed)
+	}
+
+	// The representation of EpollEvent isn't entirely accurate.
+	// Pad is fully useable, not just padding. Hence we stuff the
+	// id in there, which allows us to identify the event later (e.g.,
+	// in case of perf events, which CPU sent it).
+	event := unix.EpollEvent{
+		Events: unix.EPOLLIN,
+		Fd:     int32(fd),
+		Pad:    int32(id),
+	}
+
+	if err := unix.EpollCtl(p.epollFd, unix.EPOLL_CTL_ADD, fd, &event); err != nil {
+		return fmt.Errorf("add fd to epoll: %v", err)
+	}
+
+	return nil
+}
+
+// Wait for events.
+//
+// Returns the number of pending events or an error wrapping os.ErrClosed if
+// Close is called.
+func (p *Poller) Wait(events []unix.EpollEvent) (int, error) {
+	p.epollMu.Lock()
+	defer p.epollMu.Unlock()
+
+	if p.epollFd == -1 {
+		return 0, fmt.Errorf("epoll wait: %w", os.ErrClosed)
+	}
+
+	for {
+		n, err := unix.EpollWait(p.epollFd, events, -1)
+		if temp, ok := err.(temporaryError); ok && temp.Temporary() {
+			// Retry the syscall if we were interrupted, see https://github.com/golang/go/issues/20400
+			continue
+		}
+
+		if err != nil {
+			return 0, err
+		}
+
+		for _, event := range events[:n] {
+			if int(event.Fd) == p.event.raw {
+				// Since we don't read p.event the event is never cleared and
+				// we'll keep getting this wakeup until Close() acquires the
+				// lock and sets p.epollFd = -1.
+				return 0, fmt.Errorf("epoll wait: %w", os.ErrClosed)
+			}
+		}
+
+		return n, nil
+	}
+}
+
+type temporaryError interface {
+	Temporary() bool
+}
+
+// waitWait unblocks Wait if it's epoll_wait.
+func (p *Poller) wakeWait() error {
+	p.eventMu.Lock()
+	defer p.eventMu.Unlock()
+
+	if p.event == nil {
+		return fmt.Errorf("epoll wake: %w", os.ErrClosed)
+	}
+
+	return p.event.add(1)
+}
+
+// eventFd wraps a Linux eventfd.
+//
+// An eventfd acts like a counter: writes add to the counter, reads retrieve
+// the counter and reset it to zero. Reads also block if the counter is zero.
+//
+// See man 2 eventfd.
+type eventFd struct {
+	file *os.File
+	// prefer raw over file.Fd(), since the latter puts the file into blocking
+	// mode.
+	raw int
+}
+
+func newEventFd() (*eventFd, error) {
+	fd, err := unix.Eventfd(0, unix.O_CLOEXEC|unix.O_NONBLOCK)
+	if err != nil {
+		return nil, err
+	}
+	file := os.NewFile(uintptr(fd), "event")
+	return &eventFd{file, fd}, nil
+}
+
+func (efd *eventFd) close() error {
+	return efd.file.Close()
+}
+
+func (efd *eventFd) add(n uint64) error {
+	var buf [8]byte
+	internal.NativeEndian.PutUint64(buf[:], 1)
+	_, err := efd.file.Write(buf[:])
+	return err
+}
+
+func (efd *eventFd) read() (uint64, error) {
+	var buf [8]byte
+	_, err := efd.file.Read(buf[:])
+	return internal.NativeEndian.Uint64(buf[:]), err
+}
diff --git a/perf/reader.go b/perf/reader.go
index 8cc4401..d472ac4 100644
--- a/perf/reader.go
+++ b/perf/reader.go
@@ -5,17 +5,18 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"math"
+	"os"
 	"runtime"
 	"sync"
 
 	"github.com/cilium/ebpf"
 	"github.com/cilium/ebpf/internal"
+	"github.com/cilium/ebpf/internal/epoll"
 	"github.com/cilium/ebpf/internal/unix"
 )
 
 var (
-	ErrClosed = errors.New("perf reader was closed")
+	ErrClosed = os.ErrClosed
 	errEOR    = errors.New("end of ring")
 )
 
@@ -26,27 +27,6 @@ type perfEventHeader struct {
 	Size uint16
 }
 
-func addToEpoll(epollfd, fd int, cpu int) error {
-	if int64(cpu) > math.MaxInt32 {
-		return fmt.Errorf("unsupported CPU number: %d", cpu)
-	}
-
-	// The representation of EpollEvent isn't entirely accurate.
-	// Pad is fully useable, not just padding. Hence we stuff the
-	// CPU in there, which allows us to use a slice to access
-	// the correct perf ring.
-	event := unix.EpollEvent{
-		Events: unix.EPOLLIN,
-		Fd:     int32(fd),
-		Pad:    int32(cpu),
-	}
-
-	if err := unix.EpollCtl(epollfd, unix.EPOLL_CTL_ADD, fd, &event); err != nil {
-		return fmt.Errorf("can't add fd to epoll: %v", err)
-	}
-	return nil
-}
-
 func cpuForEvent(event *unix.EpollEvent) int {
 	return int(event.Pad)
 }
@@ -130,6 +110,8 @@ func readRawSample(rd io.Reader) ([]byte, error) {
 // Reader allows reading bpf_perf_event_output
 // from user space.
 type Reader struct {
+	poller *epoll.Poller
+
 	// mu protects read/write access to the Reader structure with the
 	// exception of 'pauseFds', which is protected by 'pauseMu'.
 	// If locking both 'mu' and 'pauseMu', 'mu' must be locked first.
@@ -137,16 +119,10 @@ type Reader struct {
 
 	// Closing a PERF_EVENT_ARRAY removes all event fds
 	// stored in it, so we keep a reference alive.
-	array *ebpf.Map
-	rings []*perfEventRing
-
-	epollFd     int
+	array       *ebpf.Map
+	rings       []*perfEventRing
 	epollEvents []unix.EpollEvent
 	epollRings  []*perfEventRing
-	// Eventfds for closing
-	closeFd int
-	// Ensure we only close once
-	closeOnce sync.Once
 
 	// pauseFds are a copy of the fds in 'rings', protected by 'pauseMu'.
 	// These allow Pause/Resume to be executed independently of any ongoing
@@ -179,20 +155,21 @@ func NewReaderWithOptions(array *ebpf.Map, perCPUBuffer int, opts ReaderOptions)
 		return nil, errors.New("perCPUBuffer must be larger than 0")
 	}
 
-	epollFd, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
-	if err != nil {
-		return nil, fmt.Errorf("can't create epoll fd: %v", err)
-	}
-
 	var (
-		fds      = []int{epollFd}
+		fds      []int
 		nCPU     = int(array.MaxEntries())
 		rings    = make([]*perfEventRing, 0, nCPU)
 		pauseFds = make([]int, 0, nCPU)
 	)
 
+	poller, err := epoll.New()
+	if err != nil {
+		return nil, err
+	}
+
 	defer func() {
 		if err != nil {
+			poller.Close()
 			for _, fd := range fds {
 				unix.Close(fd)
 			}
@@ -222,34 +199,22 @@ func NewReaderWithOptions(array *ebpf.Map, perCPUBuffer int, opts ReaderOptions)
 		rings = append(rings, ring)
 		pauseFds = append(pauseFds, ring.fd)
 
-		if err := addToEpoll(epollFd, ring.fd, len(rings)-1); err != nil {
+		if err := poller.Add(ring.fd, i); err != nil {
 			return nil, err
 		}
 	}
 
-	closeFd, err := unix.Eventfd(0, unix.O_CLOEXEC|unix.O_NONBLOCK)
-	if err != nil {
-		return nil, err
-	}
-	fds = append(fds, closeFd)
-
-	if err := addToEpoll(epollFd, closeFd, -1); err != nil {
-		return nil, err
-	}
-
 	array, err = array.Clone()
 	if err != nil {
 		return nil, err
 	}
 
 	pr = &Reader{
-		array:   array,
-		rings:   rings,
-		epollFd: epollFd,
-		// Allocate extra event for closeFd
-		epollEvents: make([]unix.EpollEvent, len(rings)+1),
+		array:       array,
+		rings:       rings,
+		poller:      poller,
+		epollEvents: make([]unix.EpollEvent, len(rings)),
 		epollRings:  make([]*perfEventRing, 0, len(rings)),
-		closeFd:     closeFd,
 		pauseFds:    pauseFds,
 	}
 	if err = pr.Resume(); err != nil {
@@ -266,44 +231,27 @@ func NewReaderWithOptions(array *ebpf.Map, perCPUBuffer int, opts ReaderOptions)
 // Calls to perf_event_output from eBPF programs will return
 // ENOENT after calling this method.
 func (pr *Reader) Close() error {
-	var err error
-	pr.closeOnce.Do(func() {
-		runtime.SetFinalizer(pr, nil)
-
-		// Interrupt Read() via the event fd.
-		var value [8]byte
-		internal.NativeEndian.PutUint64(value[:], 1)
-		_, err = unix.Write(pr.closeFd, value[:])
-		if err != nil {
-			err = fmt.Errorf("can't write event fd: %v", err)
-			return
+	if err := pr.poller.Close(); err != nil {
+		if errors.Is(err, os.ErrClosed) {
+			return nil
 		}
+		return fmt.Errorf("close poller: %w", err)
+	}
 
-		// Acquire the locks. This ensures that Read, Pause and Resume
-		// aren't running.
-		pr.mu.Lock()
-		defer pr.mu.Unlock()
-		pr.pauseMu.Lock()
-		defer pr.pauseMu.Unlock()
-
-		unix.Close(pr.epollFd)
-		unix.Close(pr.closeFd)
-		pr.epollFd, pr.closeFd = -1, -1
-
-		// Close rings
-		for _, ring := range pr.rings {
-			if ring != nil {
-				ring.Close()
-			}
-		}
-		pr.rings = nil
-		pr.pauseFds = nil
+	// Trying to poll will now fail, so Read() can't block anymore. Acquire the
+	// lock so that we can clean up.
+	pr.mu.Lock()
+	defer pr.mu.Unlock()
 
-		pr.array.Close()
-	})
-	if err != nil {
-		return fmt.Errorf("close PerfReader: %w", err)
+	for _, ring := range pr.rings {
+		if ring != nil {
+			ring.Close()
+		}
 	}
+	pr.rings = nil
+	pr.pauseFds = nil
+	pr.array.Close()
+
 	return nil
 }
 
@@ -321,27 +269,18 @@ func (pr *Reader) Read() (Record, error) {
 	pr.mu.Lock()
 	defer pr.mu.Unlock()
 
-	if pr.epollFd == -1 {
-		return Record{}, fmt.Errorf("%w", ErrClosed)
+	if pr.rings == nil {
+		return Record{}, fmt.Errorf("perf ringbuffer: %w", ErrClosed)
 	}
 
 	for {
 		if len(pr.epollRings) == 0 {
-			nEvents, err := unix.EpollWait(pr.epollFd, pr.epollEvents, -1)
-			if temp, ok := err.(temporaryError); ok && temp.Temporary() {
-				// Retry the syscall if we we're interrupted, see https://github.com/golang/go/issues/20400
-				continue
-			}
-
+			nEvents, err := pr.poller.Wait(pr.epollEvents)
 			if err != nil {
 				return Record{}, err
 			}
 
 			for _, event := range pr.epollEvents[:nEvents] {
-				if int(event.Fd) == pr.closeFd {
-					return Record{}, fmt.Errorf("%w", ErrClosed)
-				}
-
 				ring := pr.rings[cpuForEvent(&event)]
 				pr.epollRings = append(pr.epollRings, ring)
 
@@ -414,10 +353,6 @@ func (pr *Reader) Resume() error {
 	return nil
 }
 
-type temporaryError interface {
-	Temporary() bool
-}
-
 // IsClosed returns true if the error occurred because
 // a Reader was closed.
 //
diff --git a/ringbuf/reader.go b/ringbuf/reader.go
index 6efec76..64e5fec 100644
--- a/ringbuf/reader.go
+++ b/ringbuf/reader.go
@@ -5,33 +5,21 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"runtime"
+	"os"
 	"sync"
 
 	"github.com/cilium/ebpf"
 	"github.com/cilium/ebpf/internal"
+	"github.com/cilium/ebpf/internal/epoll"
 	"github.com/cilium/ebpf/internal/unix"
 )
 
 var (
-	ErrClosed  = errors.New("ringbuf reader was closed")
+	ErrClosed  = os.ErrClosed
 	errDiscard = errors.New("sample discarded")
 	errBusy    = errors.New("sample not committed yet")
 )
 
-func addToEpoll(epollfd, fd int) error {
-
-	event := unix.EpollEvent{
-		Events: unix.EPOLLIN,
-		Fd:     int32(fd),
-	}
-
-	if err := unix.EpollCtl(epollfd, unix.EPOLL_CTL_ADD, fd, &event); err != nil {
-		return fmt.Errorf("can't add fd to epoll: %w", err)
-	}
-	return nil
-}
-
 // ringbufHeader from 'struct bpf_ringbuf_hdr' in kernel/bpf/ringbuf.c
 type ringbufHeader struct {
 	Len   uint32
@@ -102,116 +90,68 @@ func readRecord(rd *ringbufEventRing) (r Record, err error) {
 // Reader allows reading bpf_ringbuf_output
 // from user space.
 type Reader struct {
-	// mu protects read/write access to the Reader structure
-	mu sync.Mutex
-
-	ring *ringbufEventRing
+	poller *epoll.Poller
 
-	epollFd     int
+	// mu protects read/write access to the Reader structure
+	mu          sync.Mutex
+	ring        *ringbufEventRing
 	epollEvents []unix.EpollEvent
-
-	closeFd int
-	// Ensure we only close once
-	closeOnce sync.Once
 }
 
 // NewReader creates a new BPF ringbuf reader.
-func NewReader(ringbufMap *ebpf.Map) (r *Reader, err error) {
+func NewReader(ringbufMap *ebpf.Map) (*Reader, error) {
 	if ringbufMap.Type() != ebpf.RingBuf {
 		return nil, fmt.Errorf("invalid Map type: %s", ringbufMap.Type())
 	}
 
 	maxEntries := int(ringbufMap.MaxEntries())
 	if maxEntries == 0 || (maxEntries&(maxEntries-1)) != 0 {
-		return nil, fmt.Errorf("Ringbuffer map size %d is zero or not a power of two", maxEntries)
+		return nil, fmt.Errorf("ringbuffer map size %d is zero or not a power of two", maxEntries)
 	}
 
-	epollFd, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
+	poller, err := epoll.New()
 	if err != nil {
-		return nil, fmt.Errorf("can't create epoll fd: %w", err)
-	}
-
-	var (
-		fds  = []int{epollFd}
-		ring *ringbufEventRing
-	)
-
-	defer func() {
-		if err != nil {
-			// close epollFd and closeFd
-			for _, fd := range fds {
-				unix.Close(fd)
-			}
-			if ring != nil {
-				ring.Close()
-			}
-		}
-	}()
-
-	ring, err = newRingBufEventRing(ringbufMap.FD(), maxEntries)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create ringbuf ring: %w", err)
-	}
-
-	if err := addToEpoll(epollFd, ringbufMap.FD()); err != nil {
 		return nil, err
 	}
 
-	closeFd, err := unix.Eventfd(0, unix.O_CLOEXEC|unix.O_NONBLOCK)
-	if err != nil {
+	if err := poller.Add(ringbufMap.FD(), 0); err != nil {
+		poller.Close()
 		return nil, err
 	}
-	fds = append(fds, closeFd)
 
-	if err := addToEpoll(epollFd, closeFd); err != nil {
-		return nil, err
+	ring, err := newRingBufEventRing(ringbufMap.FD(), maxEntries)
+	if err != nil {
+		poller.Close()
+		return nil, fmt.Errorf("failed to create ringbuf ring: %w", err)
 	}
 
-	r = &Reader{
-		ring:    ring,
-		epollFd: epollFd,
-		// Allocate extra event for closeFd
-		epollEvents: make([]unix.EpollEvent, 2),
-		closeFd:     closeFd,
-	}
-	runtime.SetFinalizer(r, (*Reader).Close)
-	return r, nil
+	return &Reader{
+		poller:      poller,
+		ring:        ring,
+		epollEvents: make([]unix.EpollEvent, 1),
+	}, nil
 }
 
 // Close frees resources used by the reader.
 //
 // It interrupts calls to Read.
 func (r *Reader) Close() error {
-	var err error
-	r.closeOnce.Do(func() {
-		runtime.SetFinalizer(r, nil)
-
-		// Interrupt Read() via the closeFd event fd.
-		var value [8]byte
-		internal.NativeEndian.PutUint64(value[:], 1)
-
-		if _, err = unix.Write(r.closeFd, value[:]); err != nil {
-			err = fmt.Errorf("can't write event fd: %w", err)
-			return
+	if err := r.poller.Close(); err != nil {
+		if errors.Is(err, os.ErrClosed) {
+			return nil
 		}
+		return err
+	}
 
-		// Acquire the lock. This ensures that Read isn't running.
-		r.mu.Lock()
-		defer r.mu.Unlock()
-
-		unix.Close(r.epollFd)
-		unix.Close(r.closeFd)
-		r.epollFd, r.closeFd = -1, -1
+	// Acquire the lock. This ensures that Read isn't running.
+	r.mu.Lock()
+	defer r.mu.Unlock()
 
-		if r.ring != nil {
-			r.ring.Close()
-		}
+	if r.ring != nil {
+		r.ring.Close()
 		r.ring = nil
-
-	})
-	if err != nil {
-		return fmt.Errorf("close Reader: %w", err)
 	}
+
 	return nil
 }
 
@@ -222,29 +162,17 @@ func (r *Reader) Read() (Record, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
-	if r.epollFd == -1 {
-		return Record{}, fmt.Errorf("%w", ErrClosed)
+	if r.ring == nil {
+		return Record{}, fmt.Errorf("ringbuffer: %w", ErrClosed)
 	}
 
 	for {
-		nEvents, err := unix.EpollWait(r.epollFd, r.epollEvents, -1)
-		if temp, ok := err.(temporaryError); ok && temp.Temporary() {
-			// Retry the syscall if we we're interrupted, see https://github.com/golang/go/issues/20400
-			continue
-		}
-
+		_, err := r.poller.Wait(r.epollEvents)
 		if err != nil {
 			return Record{}, err
 		}
 
-		for _, event := range r.epollEvents[:nEvents] {
-			if int(event.Fd) == r.closeFd {
-				return Record{}, fmt.Errorf("%w", ErrClosed)
-			}
-		}
-
 		record, err := readRecord(r.ring)
-
 		if errors.Is(err, errBusy) || errors.Is(err, errDiscard) {
 			continue
 		}
@@ -252,7 +180,3 @@ func (r *Reader) Read() (Record, error) {
 		return record, err
 	}
 }
-
-type temporaryError interface {
-	Temporary() bool
-}


Test patch:
diff --git a/internal/epoll/poller_test.go b/internal/epoll/poller_test.go
new file mode 100644
index 0000000..272abec
--- /dev/null
+++ b/internal/epoll/poller_test.go
@@ -0,0 +1,91 @@
+package epoll
+
+import (
+	"errors"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/cilium/ebpf/internal/unix"
+)
+
+func TestPoller(t *testing.T) {
+	event, err := newEventFd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer event.close()
+
+	poller, err := New()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer poller.Close()
+
+	if err := poller.Add(event.raw, 42); err != nil {
+		t.Fatal("Can't add fd:", err)
+	}
+
+	done := make(chan struct{}, 1)
+	read := func() {
+		defer func() {
+			done <- struct{}{}
+		}()
+
+		events := make([]unix.EpollEvent, 1)
+
+		n, err := poller.Wait(events)
+		if errors.Is(err, os.ErrClosed) {
+			return
+		}
+
+		if err != nil {
+			t.Error("Error from wait:", err)
+			return
+		}
+
+		if n != 1 {
+			t.Errorf("Got %d instead of 1 events", n)
+		}
+
+		if e := events[0]; e.Pad != 42 {
+			t.Errorf("Incorrect value in EpollEvent.Pad: %d != 42", e.Pad)
+		}
+	}
+
+	if err := event.add(1); err != nil {
+		t.Fatal(err)
+	}
+
+	go read()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("Timed out")
+	}
+
+	if _, err := event.read(); err != nil {
+		t.Fatal(err)
+	}
+
+	go read()
+	select {
+	case <-done:
+		t.Fatal("Wait doesn't block")
+	case <-time.After(time.Second):
+	}
+
+	if err := poller.Close(); err != nil {
+		t.Fatal("Close returns an error:", err)
+	}
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("Close doesn't unblock Wait")
+	}
+
+	if err := poller.Close(); !errors.Is(err, os.ErrClosed) {
+		t.Fatal("Closing a second time doesn't return ErrClosed:", err)
+	}
+}