Minor nits

pkg/events/events.go

@@ -31,14 +31,12 @@ import (
 var log = logger.Get()
 
 type Events struct {
-	RingBuffers          []*RingBuffer
-	PageSize             int
-	RingCnt              int
-	stopRingBufferChan   chan struct{}
-	updateRingBufferChan chan *RingBuffer
-	eventsStopChannel    chan struct{}
-	wg                   sync.WaitGroup
-	eventsDataChannel    chan []byte
+	RingBuffers       []*RingBuffer
+	PageSize          int
+	RingCnt           int
+	eventsStopChannel chan struct{}
+	wg                sync.WaitGroup
+	eventsDataChannel chan []byte
 
 	epoller *poller.EventPoller
 }
@@ -165,13 +163,7 @@ func (ev *Events) CleanupRingBuffer() {
 	return
 }
 
-func (ev *Events) reconcileEventsDataChannel() {
-
-	pollerCh := ev.epoller.EpollStart()
-	defer func() {
-		ev.wg.Done()
-	}()
-
+func (ev *Events) reconcileEventsDataChannelHandler(pollerCh <-chan int) {
 	for {
 		select {
 		case bufferPtr, ok := <-pollerCh:
@@ -187,6 +179,16 @@ func (ev *Events) reconcileEventsDataChannel() {
 	}
 }
 
+func (ev *Events) reconcileEventsDataChannel() {
+
+	pollerCh := ev.epoller.EpollStart()
+	defer ev.wg.Done()
+
+	go ev.reconcileEventsDataChannelHandler(pollerCh)
+
+	<-ev.eventsStopChannel
+}
+
 // Similar to libbpf poll ring
 func (ev *Events) ReadRingBuffer(eventRing *RingBuffer) {
 	readDone := true

pkg/events/poll/epoll.go

@@ -17,17 +17,19 @@ type EventPoller struct {
 	epollEvent []unix.EpollEvent
 	bufferCnt  int
 
-	stopEventPollerChan   chan struct{}
-	updateEventPollerChan chan int
+	stopEventPollerChan chan struct{}
+	fdEventPollerChan   chan int
 }
 
 func NewEventPoller() (*EventPoller, error) {
 	epollFD, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
 	if err != nil {
-		return nil, fmt.Errorf("Failed to create epoll instance: %s", err)
+		return nil, fmt.Errorf("failed to create epoll instance: %s", err)
 	}
 	e := &EventPoller{
-		epollFd: epollFD,
+		epollFd:             epollFD,
+		stopEventPollerChan: make(chan struct{}),
+		fdEventPollerChan:   make(chan int),
 	}
 	return e, nil
 }
@@ -52,12 +54,21 @@ func (e *EventPoller) AddEpollCtl(mapFD, eventFD int) error {
 
 func (e *EventPoller) EpollStart() <-chan int {
 
-	e.stopEventPollerChan = make(chan struct{})
-	e.updateEventPollerChan = make(chan int)
 	e.wg.Add(1)
 	go e.eventsPoller()
 
-	return e.updateEventPollerChan
+	return e.fdEventPollerChan
+}
+
+func (e *EventPoller) getEventFDs(totalEvents int) {
+	for _, event := range e.epollEvent[:totalEvents] {
+		select {
+		case e.fdEventPollerChan <- int(event.Fd):
+
+		case <-e.stopEventPollerChan:
+			return
+		}
+	}
 }
 
 func (e *EventPoller) eventsPoller() {
@@ -69,20 +80,12 @@ func (e *EventPoller) eventsPoller() {
 		default:
 			break
 		}
-		numEvents := e.poll(e.epollEvent[:e.bufferCnt])
-		for _, event := range e.epollEvent[:numEvents] {
-			select {
-			case e.updateEventPollerChan <- int(event.Fd):
-
-			case <-e.stopEventPollerChan:
-				return
-			}
-		}
+		totalEvents := e.poll(e.epollEvent[:e.bufferCnt])
+		e.getEventFDs(totalEvents)
 	}
 }
 
 func (e *EventPoller) poll(events []unix.EpollEvent) int {
-
 	timeoutMs := 150
 	n, err := unix.EpollWait(e.epollFd, events, timeoutMs)
 	if err != nil {