Add explicit timeout to perf + ring buffer polling (#309)

Right now it's hardcoded to 300ms which might not work for every
use-case. Having it as a parameter allows users to decide their tradeoff
between latency and overhead.

Signed-off-by: Francisco Javier Honduvilla Coto <javierhonduco@gmail.com>

Readme.md

@@ -99,7 +99,7 @@ mymap.Update(key, value)
 
 // ring buffer
 rb, _ := bpfModule.InitRingBuffer("events", eventsChannel, buffSize)
-rb.Start()
+rb.Poll(300)
 e := <-eventsChannel
 ```
 

libbpfgo.go

@@ -1740,10 +1740,17 @@ func (m *Module) InitRingBuf(mapName string, eventsChan chan []byte) (*RingBuffe
  return ringBuf, nil
 }
 
-func (rb *RingBuffer) Start() {
+// Poll will wait until timeout in milliseconds to gather
+// data from the ring buffer.
+func (rb *RingBuffer) Poll(timeout int) {
  rb.stop = make(chan struct{})
  rb.wg.Add(1)
- go rb.poll()
+ go rb.poll(timeout)
+}
+
+// Deprecated: use RingBuffer.Poll() instead.
+func (rb *RingBuffer) Start() {
+ rb.Poll(300)
 }
 
 func (rb *RingBuffer) Stop() {
@@ -1792,11 +1799,11 @@ func (rb *RingBuffer) isStopped() bool {
  }
 }
 
-func (rb *RingBuffer) poll() error {
+func (rb *RingBuffer) poll(timeout int) error {
  defer rb.wg.Done()
 
  for {
- err := C.ring_buffer__poll(rb.rb, 300)
+ err := C.ring_buffer__poll(rb.rb, C.int(timeout))
  if rb.isStopped() {
  break
  }
@@ -1844,10 +1851,17 @@ func (m *Module) InitPerfBuf(mapName string, eventsChan chan []byte, lostChan ch
  return perfBuf, nil
 }
 
-func (pb *PerfBuffer) Start() {
+// Poll will wait until timeout in milliseconds to gather
+// data from the perf buffer.
+func (pb *PerfBuffer) Poll(timeout int) {
  pb.stop = make(chan struct{})
  pb.wg.Add(1)
- go pb.poll()
+ go pb.poll(timeout)
+}
+
+// Deprecated: use PerfBuffer.Poll() instead.
+func (pb *PerfBuffer) Start() {
+ pb.Poll(300)
 }
 
 func (pb *PerfBuffer) Stop() {
@@ -1895,15 +1909,15 @@ func (pb *PerfBuffer) Close() {
 }
 
 // todo: consider writing the perf polling in go as c to go calls (callback) are expensive
-func (pb *PerfBuffer) poll() error {
+func (pb *PerfBuffer) poll(timeout int) error {
  defer pb.wg.Done()
 
  for {
  select {
  case <-pb.stop:
  return nil
  default:
- err := C.perf_buffer__poll(pb.pb, 300)
+ err := C.perf_buffer__poll(pb.pb, C.int(timeout))
  if err < 0 {
  if syscall.Errno(-err) == syscall.EINTR {
  continue

selftest/cgroup-legacy/main.go

@@ -63,7 +63,7 @@ func main() {
  defer stop()
 
  // start eBPF perf buffer event polling
- bpfPerfBuffer.Start()
+ bpfPerfBuffer.Poll(300)
 
  go func() {
  _, err := exec.Command("ping", "127.0.0.1", "-c 5", "-w 10").Output()

selftest/cgroup/main.go

@@ -52,7 +52,7 @@ func main() {
  os.Exit(-1)
  }
 
- rb.Start()
+ rb.Poll(300)
  numberOfEventsReceived := 0
  go func() {
  for i := 0; i < 10; i++ {

selftest/global-variable/main.go

@@ -70,7 +70,7 @@ func main() {
  exitWithErr(err)
  }
 
- rb.Start()
+ rb.Poll(300)
  go func() {
  time.Sleep(time.Second)
  syscall.Mmap(999, 999, 999, 1, 1)

selftest/map-update/main.go

@@ -95,7 +95,7 @@ func main() {
  os.Exit(-1)
  }
 
- pb.Start()
+ pb.Poll(300)
 
  go func() {
  time.Sleep(time.Second)

selftest/multiple-objects/main.go

@@ -82,7 +82,7 @@ func main() {
  fmt.Println("couldn't init ringbuffer")
  os.Exit(-1)
  }
- ringBuf.Start()
+ ringBuf.Poll(300)
  gotOne, gotTwo := false, false
 
 thisloop:

selftest/netns/main.go

@@ -47,7 +47,7 @@ func main() {
  os.Exit(-1)
  }
 
- rb.Start()
+ rb.Poll(300)
  numberOfEventsReceived := 0
  go func() {
  l, err := net.Listen("tcp", "127.0.0.1:")

selftest/perfbuffers/main.go

@@ -64,7 +64,7 @@ func main() {
  os.Exit(-1)
  }
 
- pb.Start()
+ pb.Poll(300)
 
  numberOfEventsReceived := 0
 

selftest/ringbuffers/main.go

@@ -64,7 +64,7 @@ func main() {
  os.Exit(-1)
  }
 
- rb.Start()
+ rb.Poll(300)
 
  numberOfEventsReceived := 0
  go func() {

selftest/set-attach/main.go

@@ -53,7 +53,7 @@ func main() {
  os.Exit(-1)
  }
 
- rb.Start()
+ rb.Poll(300)
  numberOfEventsReceived := 0
  go func() {
  for {

selftest/spinlocks/main.go

@@ -85,7 +85,7 @@ func main() {
  var zero uint32
  lostEventCounterMap.Update(unsafe.Pointer(&lostEventCounterKey), unsafe.Pointer(&zero))
 
- rb.Start()
+ rb.Poll(300)
 
  numberOfEventsReceived := 0
  go func() {

selftest/tc/main.go

@@ -64,7 +64,7 @@ func main() {
  os.Exit(-1)
  }
 
- rb.Start()
+ rb.Poll(300)
  numberOfEventsReceived := 0
  go func() {
  _, err := exec.Command("ping", "localhost", "-c 10").Output()

selftest/tracing/main.go

@@ -67,7 +67,7 @@ func main() {
  os.Exit(-1)
  }
 
- rb.Start()
+ rb.Poll(300)
  numberOfEventsReceived := 0
  go func() {
  for {

selftest/uprobe/main.go

@@ -81,7 +81,7 @@ func main() {
  os.Exit(-1)
  }
 
- rb.Start()
+ rb.Poll(300)
 
  numberOfEventsReceived := 0
 

selftest/xdp/main.go

@@ -49,7 +49,7 @@ func main() {
  os.Exit(-1)
  }
 
- rb.Start()
+ rb.Poll(300)
  numberOfEventsReceived := 0
  go func() {
  _, err := exec.Command("ping", "localhost", "-c 10").Output()