Repository: cilium/ebpf
Pull request #29 (2019-11-05 00:31:23)

Add support for pausing/resuming perf readers
This implementation is inspired by the upstream Cilium commit
[ce6244109bd0](http://github.com/cilium/cilium/commit/ce6244109bd0) ("cilium: add Mute/Unmute function for perf RB") by Daniel
Borkmann, to support enabling and disabling perf ringbuffer
notifications at runtime.

On the way, I fixed running the perf unit tests via `sudo go test ./...`.

joestringer: > The naming of Mute / Unmute: it seems weird to be able to do this to a Reader. Maybe Pause and Resume? This is a nit really.

I was thinking the same thing, but no ideas came to mind about better alternatives. I like `Pause` and `Resume`.

> Could we use PERF_EVENT_IOC_ENABLE with PERF_IOC_FLAG_GROUP to work around the multiple errors issue? See http://www.man7.org/linux/man-pages/man2/perf_event_open.2.html under perf_event ioctl calls.

I don't think that this helps, we actually had support for this in Cilium as well but it operates on a per-ring basis, which is the same as the fds for the map. So at least at a kernel interface level there's the same tradeoff. I suspect that in practice neither approach is likely to generate an error for these paths anyway though...  ¯\_(ツ)_/¯

Another option I noticed while exploring this was `PERF_EVENT_IOC_PAUSE_OUTPUT` (v4.7+), but again practically I'm not sure that there's any difference.
lmb: Yeah, I had misunderstood the purpose of `group_fd` in `perf_event_open`. It's used for to group events on the same cpu, not across multiple like we do.

`PERF_EVENT_IOC_PAUSE_OUTPUT` is interesting:
```
The discarded samples are considered
lost, and cause a PERF_RECORD_LOST sample to be generated when
possible.  An overflow signal may still be triggered by the
discarded sample even though the ring-buffer remains empty.
```

Should `Pause` generate lost samples or not? I guess that will determine whether to use the ioctl or just removing entries from the map.
joestringer: > Should `Pause` generate lost samples or not? I guess that will determine whether to use the ioctl or just removing entries from the map.

The `lost` samples to me are a sign that the current structure of your event handling (page size, work based on each event, etc.) is not able to keep up, and you may not intend that. Pausing the event output says that you're deliberately choosing not to handle the events, so counting lost events in that case means that the lost events counter now counts both expected and unexpected event discards.

Looking on the kernel side, if we use `PERF_EVENT_IOC_PAUSE_OUTPUT` then it will end up returning `-ENOSPC` from the `bpf_perf_event_output()` helper, meaning it is indistinguishable from actually running out of space on the buffer:

https://github.com/torvalds/linux/blob/26bc672134241a080a83b2ab9aa8abede8d30e1c/kernel/events/ring_buffer.c#L173-L176

If you contrast this with the "fds removal" approach, that will return `-ENOENT`:

https://github.com/torvalds/linux/blob/26bc672134241a080a83b2ab9aa8abede8d30e1c/kernel/trace/bpf_trace.c#L436-L438

This gives the BPF side a clearer signal that the events buffer is not currently listening, so the BPF code can be written to handle this particular case differently from generic event drops. Plus, there's a bit less actual code being executed if we delete the fds from the map---kernel-side the bpf helper first does the map lookup, then proceeds towards the generic event handling code.
lmb: SGTM, can you push your latest changes?
joestringer: Weird, not sure why github didn't pick it up before. I re-pushed.
joestringer: @lorenz Thanks for the suggestion, I've pushed the changes out as a brand new patch. PTAL.

Patch:
diff --git a/internal/unix/types_linux.go b/internal/unix/types_linux.go
index 49c6be5..003907b 100644
--- a/internal/unix/types_linux.go
+++ b/internal/unix/types_linux.go
@@ -30,6 +30,7 @@ const (
 	PerfBitWatermark         = linux.PerfBitWatermark
 	PERF_SAMPLE_RAW          = linux.PERF_SAMPLE_RAW
 	PERF_FLAG_FD_CLOEXEC     = linux.PERF_FLAG_FD_CLOEXEC
+	RLIM_INFINITY            = linux.RLIM_INFINITY
 )
 
 // Statfs_t is a wrapper
diff --git a/perf/reader.go b/perf/reader.go
index 3534e99..70cae4c 100644
--- a/perf/reader.go
+++ b/perf/reader.go
@@ -133,6 +133,9 @@ func readRawSample(rd io.Reader) ([]byte, error) {
 // Reader allows reading bpf_perf_event_output
 // from user space.
 type Reader struct {
+	// 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.
 	mu sync.Mutex
 
 	// Closing a PERF_EVENT_ARRAY removes all event fds
@@ -147,6 +150,12 @@ type Reader struct {
 	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
+	// Read calls, which would otherwise need to be interrupted.
+	pauseMu  sync.Mutex
+	pauseFds []int
 }
 
 // ReaderOptions control the behaviour of the user
@@ -179,9 +188,10 @@ func NewReaderWithOptions(array *ebpf.Map, perCPUBuffer int, opts ReaderOptions)
 	}
 
 	var (
-		fds   = []int{epollFd}
-		nCPU  = int(array.ABI().MaxEntries)
-		rings = make([]*perfEventRing, 0, nCPU)
+		fds      = []int{epollFd}
+		nCPU     = int(array.ABI().MaxEntries)
+		rings    = make([]*perfEventRing, 0, nCPU)
+		pauseFds = make([]int, 0, nCPU)
 	)
 
 	defer func() {
@@ -204,10 +214,7 @@ func NewReaderWithOptions(array *ebpf.Map, perCPUBuffer int, opts ReaderOptions)
 			return nil, errors.Wrapf(err, "failed to create perf ring for CPU %d", i)
 		}
 		rings = append(rings, ring)
-
-		if err := array.Put(uint32(i), uint32(ring.fd)); err != nil {
-			return nil, errors.Wrapf(err, "could't put event fd for CPU %d", i)
-		}
+		pauseFds = append(pauseFds, ring.fd)
 
 		if err := addToEpoll(epollFd, ring.fd, len(rings)-1); err != nil {
 			return nil, err
@@ -237,6 +244,13 @@ func NewReaderWithOptions(array *ebpf.Map, perCPUBuffer int, opts ReaderOptions)
 		epollEvents: make([]unix.EpollEvent, len(rings)+1),
 		epollRings:  make([]*perfEventRing, 0, len(rings)),
 		closeFd:     closeFd,
+		pauseFds:    pauseFds,
+	}
+	if err = pr.Resume(); err != nil {
+		return nil, err
+	}
+	if err = pr.Resume(); err != nil {
+		return nil, err
 	}
 	runtime.SetFinalizer(pr, (*Reader).Close)
 	return pr, nil
@@ -262,10 +276,12 @@ func (pr *Reader) Close() error {
 			return
 		}
 
-		// Acquire the lock. This ensures that Read
-		// isn't running.
+		// 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)
@@ -276,6 +292,7 @@ func (pr *Reader) Close() error {
 			ring.Close()
 		}
 		pr.rings = nil
+		pr.pauseFds = nil
 
 		pr.array.Close()
 	})
@@ -341,6 +358,50 @@ func (pr *Reader) Read() (Record, error) {
 	}
 }
 
+// Pause stops all notifications from this Reader.
+//
+// While the Reader is paused, any attempts to write to the event buffer from
+// BPF programs will return -ENOENT.
+//
+// Subsequent calls to Read will block until a call to Resume.
+func (pr *Reader) Pause() error {
+	pr.pauseMu.Lock()
+	defer pr.pauseMu.Unlock()
+
+	if pr.pauseFds == nil {
+		return errClosed
+	}
+
+	for i := 0; i < len(pr.pauseFds); i++ {
+		if err := pr.array.Delete(uint32(i)); err != nil && !ebpf.IsNotExist(err) {
+			return errors.Wrapf(err, "could't delete event fd for CPU %d", i)
+		}
+	}
+
+	return nil
+}
+
+// Resume allows this perf reader to emit notifications.
+//
+// Subsequent calls to Read will block until the next event notification.
+func (pr *Reader) Resume() error {
+	pr.pauseMu.Lock()
+	defer pr.pauseMu.Unlock()
+
+	if pr.pauseFds == nil {
+		return errClosed
+	}
+
+	for i := 0; i < len(pr.pauseFds); i++ {
+		fd := uint32(pr.pauseFds[i])
+		if err := pr.array.Put(uint32(i), fd); err != nil {
+			return errors.Wrapf(err, "could't put event fd %d for CPU %d", fd, i)
+		}
+	}
+
+	return nil
+}
+
 type temporaryError interface {
 	Temporary() bool
 }


Test patch:
diff --git a/perf/reader_test.go b/perf/reader_test.go
index 2015f0f..0056a79 100644
--- a/perf/reader_test.go
+++ b/perf/reader_test.go
@@ -15,6 +15,21 @@ import (
 	"github.com/cilium/ebpf/internal/unix"
 )
 
+var (
+	readTimeout = 250 * time.Millisecond
+)
+
+func TestMain(m *testing.M) {
+	err := unix.Setrlimit(8, &unix.Rlimit{
+		Cur: unix.RLIM_INFINITY,
+		Max: unix.RLIM_INFINITY,
+	})
+	if err != nil {
+		fmt.Println("WARNING: Failed to adjust rlimit, tests may fail")
+	}
+	os.Exit(m.Run())
+}
+
 func TestPerfReader(t *testing.T) {
 	prog, events := mustOutputSamplesProg(t, 5)
 	defer prog.Close()
@@ -287,6 +302,90 @@ func TestReadRecord(t *testing.T) {
 	}
 }
 
+func TestPause(t *testing.T) {
+	t.Parallel()
+
+	prog, events := mustOutputSamplesProg(t, 5)
+	defer prog.Close()
+	defer events.Close()
+
+	rd, err := NewReader(events, 4096)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rd.Close()
+
+	// Reader is already unpaused by default. It should be idempotent.
+	if err = rd.Resume(); err != nil {
+		t.Fatal(err)
+	}
+
+	// Write a sample. The reader should read it.
+	ret, _, err := prog.Test(make([]byte, 14))
+	if err != nil || ret != 0 {
+		t.Fatal("Can't write sample")
+	}
+	if _, err := rd.Read(); err != nil {
+		t.Fatal(err)
+	}
+
+	// Pause. No notification should trigger.
+	if err = rd.Pause(); err != nil {
+		t.Fatal(err)
+	}
+	errChan := make(chan error, 1)
+	go func() {
+		// Read one notification then send any errors and exit.
+		_, err := rd.Read()
+		errChan <- err
+	}()
+	ret, _, err = prog.Test(make([]byte, 14))
+	if err == nil && ret == 0 {
+		t.Fatal("Unexpectedly wrote sample while paused")
+	} // else Success
+	select {
+	case err := <-errChan:
+		// Failure: Pause was unsuccessful.
+		t.Fatalf("received notification on paused reader: %s", err)
+	case <-time.After(readTimeout):
+		// Success
+	}
+
+	// Pause should be idempotent.
+	if err = rd.Pause(); err != nil {
+		t.Fatal(err)
+	}
+
+	// Resume. Now notifications should continue.
+	if err = rd.Resume(); err != nil {
+		t.Fatal(err)
+	}
+	ret, _, err = prog.Test(make([]byte, 14))
+	if err != nil || ret != 0 {
+		t.Fatal("Can't write sample")
+	}
+	select {
+	case err := <-errChan:
+		if err != nil {
+			t.Fatal(err)
+		} // else Success
+	case <-time.After(readTimeout):
+		t.Fatal("timed out waiting for notification after resume")
+	}
+
+	if err = rd.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	// Pause/Resume after close should be no-op.
+	if err = rd.Pause(); err != errClosed {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	if err = rd.Resume(); err != errClosed {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+}
+
 func BenchmarkReader(b *testing.B) {
 	prog, events := mustOutputSamplesProg(b, 80)
 	defer prog.Close()