perf add timeout argument

internal/epoll/poller.go

@@ -119,7 +119,7 @@ func (p *Poller) Add(fd int, id int) error {
 //
 // 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) {
+func (p *Poller) Wait(events []unix.EpollEvent, msec int) (int, error) {
 	p.epollMu.Lock()
 	defer p.epollMu.Unlock()
 
@@ -128,7 +128,7 @@ func (p *Poller) Wait(events []unix.EpollEvent) (int, error) {
 	}
 
 	for {
-		n, err := unix.EpollWait(p.epollFd, events, -1)
+		n, err := unix.EpollWait(p.epollFd, events, msec)
 		if temp, ok := err.(temporaryError); ok && temp.Temporary() {
 			// Retry the syscall if we were interrupted, see https://github.com/golang/go/issues/20400
 			continue

internal/epoll/poller_test.go

@@ -34,7 +34,7 @@ func TestPoller(t *testing.T) {
 
 		events := make([]unix.EpollEvent, 1)
 
-		n, err := poller.Wait(events)
+		n, err := poller.Wait(events, -1)
 		if errors.Is(err, os.ErrClosed) {
 			return
 		}

perf/reader.go

@@ -16,8 +16,9 @@ import (
 )
 
 var (
-	ErrClosed = os.ErrClosed
-	errEOR    = errors.New("end of ring")
+	ErrClosed  = os.ErrClosed
+	ErrTimeOut = errors.New("timeout")
+	errEOR     = errors.New("end of ring")
 )
 
 var perfEventHeaderSize = binary.Size(perfEventHeader{})
@@ -162,6 +163,7 @@ type ReaderOptions struct {
 	// Read will process data. Must be smaller than PerCPUBuffer.
 	// The default is to start processing as soon as data is available.
 	Watermark int
+	TimeOut   int
 }
 
 // NewReader creates a new reader with default options.
@@ -290,26 +292,35 @@ func (pr *Reader) Close() error {
 // depending on the input sample's length.
 //
 // Calling Close interrupts the function.
-func (pr *Reader) Read() (Record, error) {
+func (pr *Reader) Read(opts *ReaderOptions) (Record, error) {
 	var r Record
-	return r, pr.ReadInto(&r)
+	return r, pr.ReadInto(&r, opts)
 }
 
 // ReadInto is like Read except that it allows reusing Record and associated buffers.
-func (pr *Reader) ReadInto(rec *Record) error {
+func (pr *Reader) ReadInto(rec *Record, opts *ReaderOptions) error {
 	pr.mu.Lock()
 	defer pr.mu.Unlock()
 
 	if pr.rings == nil {
 		return fmt.Errorf("perf ringbuffer: %w", ErrClosed)
 	}
 
+	msec := -1
+	if opts != nil {
+		msec = opts.TimeOut
+	}
+
 	for {
 		if len(pr.epollRings) == 0 {
-			nEvents, err := pr.poller.Wait(pr.epollEvents)
+			nEvents, err := pr.poller.Wait(pr.epollEvents, msec)
 			if err != nil {
 				return err
 			}
+
+			if nEvents == 0 {
+				return ErrTimeOut
+			}
 
 			for _, event := range pr.epollEvents[:nEvents] {
 				ring := pr.rings[cpuForEvent(&event)]

perf/reader_test.go

@@ -44,7 +44,7 @@ func TestPerfReader(t *testing.T) {
 		t.Fatal("Expected 0 as return value, got", errno)
 	}
 
-	record, err := rd.Read()
+	record, err := rd.Read(nil)
 	if err != nil {
 		t.Fatal("Can't read samples:", err)
 	}
@@ -225,7 +225,7 @@ func TestPerfReaderLostSample(t *testing.T) {
 	}
 
 	for range sampleSizes {
-		record, err := rd.Read()
+		record, err := rd.Read(nil)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -251,7 +251,7 @@ func TestPerfReaderClose(t *testing.T) {
 	waiting := make(chan struct{})
 	go func() {
 		close(waiting)
-		_, err := rd.Read()
+		_, err := rd.Read(nil)
 		errs <- err
 	}()
 
@@ -273,7 +273,7 @@ func TestPerfReaderClose(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if _, err := rd.Read(); err == nil {
+	if _, err := rd.Read(nil); err == nil {
 		t.Fatal("Read on a closed PerfReader doesn't return an error")
 	}
 }
@@ -325,7 +325,7 @@ func TestPause(t *testing.T) {
 	if err != nil || ret != 0 {
 		t.Fatal("Can't write sample")
 	}
-	if _, err := rd.Read(); err != nil {
+	if _, err := rd.Read(nil); err != nil {
 		t.Fatal(err)
 	}
 
@@ -336,7 +336,7 @@ func TestPause(t *testing.T) {
 	errChan := make(chan error, 1)
 	go func() {
 		// Read one notification then send any errors and exit.
-		_, err := rd.Read()
+		_, err := rd.Read(nil)
 		errChan <- err
 	}()
 	ret, _, err = prog.Test(make([]byte, 14))
@@ -410,7 +410,7 @@ func BenchmarkReader(b *testing.B) {
 			b.Fatal("Expected 0 as return value, got", errno)
 		}
 
-		if _, err = rd.Read(); err != nil {
+		if _, err = rd.Read(nil); err != nil {
 			b.Fatal(err)
 		}
 	}
@@ -443,7 +443,7 @@ func BenchmarkReadInto(b *testing.B) {
 			b.Fatal("Expected 0 as return value, got", errno)
 		}
 
-		if err := rd.ReadInto(&rec); err != nil {
+		if err := rd.ReadInto(&rec, nil); err != nil {
 			b.Fatal(err)
 		}
 	}
@@ -494,7 +494,7 @@ func ExampleReader() {
 		panic("Can't write sample")
 	}
 
-	record, err := rd.Read()
+	record, err := rd.Read(nil)
 	if err != nil {
 		panic(err)
 	}
@@ -525,7 +525,7 @@ func ExampleReader_ReadInto() {
 
 	var rec Record
 	for i := 0; i < 2; i++ {
-		if err := rd.ReadInto(&rec); err != nil {
+		if err := rd.ReadInto(&rec, nil); err != nil {
 			panic(err)
 		}
 

ringbuf/reader.go

@@ -190,7 +190,7 @@ func (r *Reader) ReadInto(rec *Record) error {
 
 	for {
 		if !r.haveData {
-			_, err := r.poller.Wait(r.epollEvents[:cap(r.epollEvents)])
+			_, err := r.poller.Wait(r.epollEvents[:cap(r.epollEvents)], -1)
 			if err != nil {
 				return err
 			}