better ebpf tasks cancellation

It should improve daemon stopping times.

daemon/procmon/ebpf/ebpf.go

@@ -1,6 +1,7 @@
 package ebpf
 
 import (
+	"context"
 	"encoding/binary"
 	"fmt"
 	"net"
@@ -47,7 +48,7 @@ var (
 		TCP:   make(map[*daemonNetlink.Socket]int),
 		TCPv6: make(map[*daemonNetlink.Socket]int),
 	}
-	stopMonitors = make(chan bool)
+	ctxTasks, cancelTasks = context.WithCancel(context.Background())
 	running               = false
 
 	maxKernelEvents = 32768
@@ -152,15 +153,9 @@ func Stop() {
 	if running == false {
 		return
 	}
-	for i := 0; i < 4; i++ {
-		stopMonitors <- true
-	}
+	cancelTasks()
 	ebpfCache.clear()
 
-	for i := 0; i < eventWorkers; i++ {
-		stopStreamEvents <- true
-	}
-
 	if m != nil {
 		m.Close()
 	}

daemon/procmon/ebpf/events.go

@@ -60,7 +60,6 @@ const (
 
 var (
 	execEvents  = NewEventsStore()
-	stopStreamEvents = make(chan bool)
 	perfMapList = make(map[*elf.PerfMap]*elf.Module)
 	// total workers spawned by the different events PerfMaps
 	eventWorkers = 0
@@ -144,12 +143,11 @@ func initPerfMap(mod *elf.Module) {
 	perfMap.PollStart()
 }
 
-// FIXME: under heavy load these events may arrive AFTER network events
 func streamEventsWorker(id int, chn chan []byte, lost chan uint64, kernelEvents chan interface{}, execEvents *eventsStore) {
 	var event execEvent
 	for {
 		select {
-		case <-stopStreamEvents:
+		case <-ctxTasks.Done():
 			goto Exit
 		case l := <-lost:
 			log.Debug("Lost ebpf events: %d", l)
@@ -176,7 +174,6 @@ func streamEventsWorker(id int, chn chan []byte, lost chan uint64, kernelEvents
 						execEvents.delete(event.PID)
 					}
 				}
-				// TODO: delete old events (by timeout)
 			}
 		}
 	}

daemon/procmon/ebpf/monitor.go

@@ -15,7 +15,7 @@ import (
 func monitorMaps() {
 	for {
 		select {
-		case <-stopMonitors:
+		case <-ctxTasks.Done():
 			goto Exit
 		default:
 			time.Sleep(time.Second * 5)
@@ -35,7 +35,7 @@ Exit:
 func monitorCache() {
 	for {
 		select {
-		case <-stopMonitors:
+		case <-ctxTasks.Done():
 			goto Exit
 		case <-ebpfCacheTicker.C:
 			ebpfCache.DeleteOldItems()
@@ -50,7 +50,7 @@ Exit:
 func monitorLocalAddresses() {
 	for {
 		select {
-		case <-stopMonitors:
+		case <-ctxTasks.Done():
 			goto Exit
 		default:
 			addr, err := netlink.AddrList(nil, netlink.FAMILY_ALL)
@@ -76,7 +76,7 @@ Exit:
 func monitorAlreadyEstablished() {
 	for {
 		select {
-		case <-stopMonitors:
+		case <-ctxTasks.Done():
 			goto Exit
 		default:
 			time.Sleep(time.Second * 1)