ringbuf: fix corrupt samples on arm64

Reading the len field of the event header needs to be atomic to match
the release memory barrier from
https://github.com/torvalds/linux/blob/eb26cbb1a754ccde5d4d74527dad5ba051808fad/kernel/bpf/ringbuf.c#L487

We also don't care about the PgOff field which is only used by ringbuf
internals.

[ Additional context by Lorenz ]

Based on a suggestion by Daniel Borkmann I wrote a litmus test against
the Linux Kernel Memory Model:

    C ringbuf

    {}

    P0(int *hdr_len, int *prod_pos, int *data) // Producer
    {
        // __bpf_ringbuf_reserve
        *hdr_len = 1; // BUSY
        smp_store_release(prod_pos, 1);

        // store some data.
        *data = 1;

        // bpf_ringbuf_commit
        int r0 = xchg(hdr_len, 2); // COMMIT
    }

    P1(int *hdr_len, int *prod_pos, int *data) // Consumer
    {
        int p = smp_load_acquire(prod_pos);
        if (p > 0) {
            int l = *hdr_len;
            if (l == 2) {
                int d = *data;
            }
        }
    }

    exists (prod_pos=1 /\ 1:l=2 /\ 1:d=0)

Note that hdr_len is read via a plain read. Executing the litmus test
shows that we can indeed observe a COMMIT (l=2) before data is visible:

    $ herd7 -conf linux-kernel.cfg litmus-tests/ringbuf.litmus
    Test ringbuf Allowed
    States 4
    1:d=0; 1:l=0; [prod_pos]=1;
    1:d=0; 1:l=1; [prod_pos]=1;
    1:d=0; 1:l=2; [prod_pos]=1;
    1:d=1; 1:l=2; [prod_pos]=1;
    Ok
    Witnesses
    Positive: 1 Negative: 3
    Flag data-race
    Condition exists ([prod_pos]=1 /\ 1:l=2 /\ 1:d=0)
    Observation ringbuf Sometimes 1 3
    Time ringbuf 0.01
    Hash=7aed947102aa203443a5b9b5884f956f

This is fixed by using an smp_load_acquire or equivalent to read hdr_len.

Signed-off-by: Paul Cacheux <[email protected]>
Co-developed-by: Lorenz Bauer <[email protected]>

Author: paulcacheux

ringbuf/reader.go

@@ -23,8 +23,8 @@ var (
 
 // ringbufHeader from 'struct bpf_ringbuf_hdr' in kernel/bpf/ringbuf.c
 type ringbufHeader struct {
-	Len   uint32
-	PgOff uint32
+	Len uint32
+	_   uint32 // pg_off, only used by kernel internals
 }
 
 func (rh *ringbufHeader) isBusy() bool {
@@ -47,23 +47,16 @@ type Record struct {
 }
 
 // Read a record from an event ring.
-//
-// buf must be at least BPF_RINGBUF_HDR_SZ bytes long.
-func readRecord(rd *ringbufEventRing, rec *Record, buf []byte) error {
+func readRecord(rd *ringbufEventRing, rec *Record) error {
 	rd.loadConsumer()
 
-	buf = buf[:unix.BPF_RINGBUF_HDR_SZ]
-	if _, err := io.ReadFull(rd, buf); err == io.EOF {
+	header, err := rd.readHeader()
+	if err == io.EOF {
 		return errEOR
 	} else if err != nil {
 		return fmt.Errorf("read event header: %w", err)
 	}
 
-	header := ringbufHeader{
-		internal.NativeEndian.Uint32(buf[0:4]),
-		internal.NativeEndian.Uint32(buf[4:8]),
-	}
-
 	if header.isBusy() {
 		// the next sample in the ring is not committed yet so we
 		// exit without storing the reader/consumer position
@@ -111,7 +104,6 @@ type Reader struct {
 	mu          sync.Mutex
 	ring        *ringbufEventRing
 	epollEvents []unix.EpollEvent
-	header      []byte
 	haveData    bool
 	deadline    time.Time
 	bufferSize  int
@@ -148,7 +140,6 @@ func NewReader(ringbufMap *ebpf.Map) (*Reader, error) {
 		poller:      poller,
 		ring:        ring,
 		epollEvents: make([]unix.EpollEvent, 1),
-		header:      make([]byte, unix.BPF_RINGBUF_HDR_SZ),
 		bufferSize:  ring.size(),
 	}, nil
 }
@@ -220,7 +211,7 @@ func (r *Reader) ReadInto(rec *Record) error {
 		}
 
 		for {
-			err := readRecord(r.ring, rec, r.header)
+			err := readRecord(r.ring, rec)
 			// Not using errors.Is which is quite a bit slower
 			// For a tight loop it might make a difference
 			if err == errBusy || err == errDiscard {

ringbuf/ring.go

@@ -102,6 +102,26 @@ func (rr *ringReader) remaining() int {
 	return int((prod - cons) & rr.mask)
 }
 
+func (rr *ringReader) readHeader() (ringbufHeader, error) {
+	prod := atomic.LoadUint64(rr.prod_pos)
+
+	if remaining := prod - rr.cons; remaining == 0 {
+		return ringbufHeader{}, io.EOF
+	} else if remaining < unix.BPF_RINGBUF_HDR_SZ {
+		return ringbufHeader{}, io.ErrUnexpectedEOF
+	}
+
+	// read the len field of the header atomically to ensure a happens before
+	// relationship with the xchg in the kernel. Without this we may see len
+	// without BPF_RINGBUF_BUSY_BIT before the written data is visible.
+	// See https://github.com/torvalds/linux/blob/v6.8/kernel/bpf/ringbuf.c#L484
+	len := atomic.LoadUint32((*uint32)((unsafe.Pointer)(&rr.ring[rr.cons&rr.mask])))
+	header := ringbufHeader{Len: len}
+
+	rr.cons += unix.BPF_RINGBUF_HDR_SZ
+	return header, nil
+}
+
 func (rr *ringReader) Read(p []byte) (int, error) {
 	prod := atomic.LoadUint64(rr.prod_pos)
 	n := min(prod-rr.cons, uint64(len(p)))