ringbuf: add zero-copy consumer APIs

The current ringbuf consumer APIs Read/ReadInto always copy samples into
a user-provided buffer, see
https://github.com/cilium/ebpf/blob/v0.20.0/ringbuf/ring.go#L96. For
callers that consume data synchronizely, this extra copy is unnecessary.

This change adds a zero-copy consumer API:

- Peek / PeekInto – return a view into the mmap’d ring buffer without
  advancing the consumer position.
- Consume – advance the consumer position once the view has been
  processed.

Existing Read/ReadInto semantics are unchanged and continue to work as
before.

A preliminary microbenchmark [1] shows zero-copy advantage grows with
larger records because copy throughput falls while view throughput stays
roughly flat.

Single-core run on CPU 2, ring size 1 GiB, Go 1.24.4. Throughput is in
million events/second (Mev/s); "speedup" is zero-copy / copy.

| event-size (B) | events/run | copy (Mev/s) | zero-copy (Mev/s) | speedup |
| --- | --- | --- | --- | --- |
| 128 | 7,895,160 | 45.63 | 49.83 | 1.09x |
| 512 | 2,064,888 | 25.03 | 34.94 | 1.40x |
| 1024 | 1,040,447 | 8.90 | 34.94 | 3.93x |
| 2048 | 522,247 | 4.57 | 29.56 | 6.47x |

[1] https://github.com/jschwinger233/bpf_ringbuf_zc_benchmark

Signed-off-by: graymon <[email protected]>

Author: jschwinger233

ringbuf/reader.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 	"sync"
+	"sync/atomic"
 	"time"
 	"unsafe"
 
@@ -46,6 +47,14 @@ type Record struct {
 	Remaining int
 }
 
+// View is a zero-copy record returned by Peek/PeekInto until Consume is called.
+type View struct {
+	Sample    []byte
+	Remaining int
+
+	nextCons uintptr
+}
+
 // Reader allows reading bpf_ringbuf_output
 // from user space.
 type Reader struct {
@@ -144,6 +153,32 @@ func (r *Reader) Read() (Record, error) {
 
 // ReadInto is like Read except that it allows reusing Record and associated buffers.
 func (r *Reader) ReadInto(rec *Record) error {
+	return r.readLocked(func() error {
+		return r.ring.readRecord(rec)
+	})
+}
+
+// Peek returns the next sample without advancing the consumer position.
+func (r *Reader) Peek() (View, error) {
+	var v View
+	err := r.PeekInto(&v)
+	return v, err
+}
+
+// PeekInto is like Peek but stores the sample in view for reuse.
+func (r *Reader) PeekInto(view *View) error {
+	return r.readLocked(func() error {
+		return r.ring.readView(view)
+	})
+}
+
+// Consume advances the consumer position after a successful Peek/PeekInto.
+func (r *Reader) Consume(view *View) {
+	atomic.StoreUintptr(r.ring.cons_pos, view.nextCons)
+}
+
+// readLocked drives the polling / data-availability loop shared by Record and View reads.
+func (r *Reader) readLocked(read func() error) error {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
@@ -171,7 +206,7 @@ func (r *Reader) ReadInto(rec *Record) error {
 		}
 
 		for {
-			err := r.ring.readRecord(rec)
+			err := read()
 			// Not using errors.Is which is quite a bit slower
 			// For a tight loop it might make a difference
 			if err == errBusy {

ringbuf/reader_test.go

@@ -384,3 +384,56 @@ func BenchmarkReadInto(b *testing.B) {
 		}
 	}
 }
+
+func TestPeek(t *testing.T) {
+	testutils.SkipOnOldKernel(t, "5.8", "BPF ring buffer")
+
+	prog, events := mustOutputSamplesProg(t,
+		sampleMessage{size: 5, flags: 0},
+		sampleMessage{size: 7, flags: 0},
+	)
+	mustRun(t, prog)
+
+	rd, err := NewReader(events)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rd.Close()
+
+	view1, err := rd.Peek()
+	qt.Assert(t, qt.IsNil(err))
+	qt.Assert(t, qt.Equals(len(view1.Sample), 5))
+	rd.Consume(&view1)
+
+	view2, err := rd.Peek()
+	qt.Assert(t, qt.IsNil(err))
+	qt.Assert(t, qt.Equals(len(view2.Sample), 7))
+	rd.Consume(&view2)
+}
+
+func TestPeekInto(t *testing.T) {
+	testutils.SkipOnOldKernel(t, "5.8", "BPF ring buffer")
+
+	prog, events := mustOutputSamplesProg(t,
+		sampleMessage{size: 5, flags: 0},
+		sampleMessage{size: 7, flags: 0},
+	)
+	mustRun(t, prog)
+
+	rd, err := NewReader(events)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rd.Close()
+
+	var view View
+	err = rd.PeekInto(&view)
+	qt.Assert(t, qt.IsNil(err))
+	qt.Assert(t, qt.Equals(len(view.Sample), 5))
+	rd.Consume(&view)
+
+	err = rd.PeekInto(&view)
+	qt.Assert(t, qt.IsNil(err))
+	qt.Assert(t, qt.Equals(len(view.Sample), 7))
+	rd.Consume(&view)
+}

ringbuf/ring.go

@@ -41,16 +41,17 @@ func (rr *ringReader) AvailableBytes() uint64 {
 	return uint64(prod - cons)
 }
 
-// Read a record from an event ring.
-func (rr *ringReader) readRecord(rec *Record) error {
+// readSample returns a zero-copy view into the next sample, together with the
+// consumer position that should be stored to release the data.
+func (rr *ringReader) readSample() (sample []byte, remaining int, nextCons uintptr, err error) {
 	prod := atomic.LoadUintptr(rr.prod_pos)
 	cons := atomic.LoadUintptr(rr.cons_pos)
 
 	for {
 		if remaining := prod - cons; remaining == 0 {
-			return errEOR
+			return nil, 0, 0, errEOR
 		} else if remaining < sys.BPF_RINGBUF_HDR_SZ {
-			return fmt.Errorf("read record header: %w", io.ErrUnexpectedEOF)
+			return nil, 0, 0, fmt.Errorf("read record header: %w", io.ErrUnexpectedEOF)
 		}
 
 		// read the len field of the header atomically to ensure a happens before
@@ -65,15 +66,15 @@ func (rr *ringReader) readRecord(rec *Record) error {
 			// the next sample in the ring is not committed yet so we
 			// exit without storing the reader/consumer position
 			// and start again from the same position.
-			return errBusy
+			return nil, 0, 0, errBusy
 		}
 
 		cons += sys.BPF_RINGBUF_HDR_SZ
 
 		// Data is always padded to 8 byte alignment.
 		dataLenAligned := uintptr(internal.Align(header.dataLen(), 8))
 		if remaining := prod - cons; remaining < dataLenAligned {
-			return fmt.Errorf("read sample data: %w", io.ErrUnexpectedEOF)
+			return nil, 0, 0, fmt.Errorf("read sample data: %w", io.ErrUnexpectedEOF)
 		}
 
 		start = cons & rr.mask
@@ -87,15 +88,40 @@ func (rr *ringReader) readRecord(rec *Record) error {
 			continue
 		}
 
-		if n := header.dataLen(); cap(rec.RawSample) < n {
-			rec.RawSample = make([]byte, n)
-		} else {
-			rec.RawSample = rec.RawSample[:n]
-		}
+		end := int(start) + header.dataLen()
+		return rr.ring[start:end], int(prod - cons), cons, nil
+	}
+}
+
+// Read a record from an event ring.
+func (rr *ringReader) readRecord(rec *Record) error {
+	sample, remaining, nextCons, err := rr.readSample()
+	if err != nil {
+		return err
+	}
 
-		copy(rec.RawSample, rr.ring[start:])
-		rec.Remaining = int(prod - cons)
-		atomic.StoreUintptr(rr.cons_pos, cons)
-		return nil
+	if n := len(sample); cap(rec.RawSample) < n {
+		rec.RawSample = make([]byte, n)
+	} else {
+		rec.RawSample = rec.RawSample[:n]
 	}
+
+	copy(rec.RawSample, sample)
+	rec.Remaining = remaining
+	atomic.StoreUintptr(rr.cons_pos, nextCons)
+	return nil
+}
+
+// readView is the zero-copy variant of readRecord. It leaves advancing the
+// consumer position to the caller.
+func (rr *ringReader) readView(view *View) error {
+	sample, remaining, nextCons, err := rr.readSample()
+	if err != nil {
+		return err
+	}
+
+	view.Sample = sample
+	view.Remaining = remaining
+	view.nextCons = nextCons
+	return nil
 }