From 9c53653997af24798a052f23d411e981320e3ed4 Mon Sep 17 00:00:00 2001 From: Alban Crequy Date: Thu, 20 Jul 2017 16:41:56 +0200 Subject: [PATCH 1/3] EbpfTracker: restart it when it dies EbpfTracker can die when the tcp events are received out of order. This can happen with a buggy kernel or apparently in other cases, see: https://github.com/weaveworks/scope/issues/2650 As a workaround, restart EbpfTracker when an event is received out of order. This does not seem to happen often, but as a precaution, EbpfTracker will not restart if the last failure is less than 5 minutes ago. This is not easy to test but I added instrumentation to trigger a restart: - Start Scope with: $ sudo WEAVESCOPE_DOCKER_ARGS="-e SCOPE_DEBUG_BPF=1" ./scope launch - Request a stop with: $ echo stop | sudo tee /proc/$(pidof scope-probe)/root/var/run/scope/debug-bpf --- probe/endpoint/connection_tracker.go | 29 ++++++++- probe/endpoint/ebpf.go | 88 ++++++++++++++++++++++++---- probe/endpoint/ebpf_test.go | 9 +++ 3 files changed, 111 insertions(+), 15 deletions(-) diff --git a/probe/endpoint/connection_tracker.go b/probe/endpoint/connection_tracker.go index 136ee59f5..cd8dac7a0 100644 --- a/probe/endpoint/connection_tracker.go +++ b/probe/endpoint/connection_tracker.go @@ -2,6 +2,7 @@ package endpoint import ( "strconv" + "time" log "github.com/Sirupsen/logrus" "github.com/weaveworks/scope/probe/endpoint/procspy" @@ -29,6 +30,9 @@ type connectionTracker struct { flowWalker flowWalker // Interface ebpfTracker *EbpfTracker reverseResolver *reverseResolver + + // time of the previous ebpf failure, or zero if it didn't fail + ebpfLastFailureTime time.Time } func newConnectionTracker(conf connectionTrackerConfig) connectionTracker { @@ -87,8 +91,29 @@ func (t *connectionTracker) ReportConnections(rpt *report.Report) { t.performEbpfTrack(rpt, hostNodeID) return } - log.Warnf("ebpf tracker died, gently falling back to proc scanning") - t.useProcfs() + + // We only restart the EbpfTracker if the failures are not too frequent to + // avoid repeatitive restarts. + + ebpfLastFailureTime := t.ebpfLastFailureTime + t.ebpfLastFailureTime = time.Now() + + if ebpfLastFailureTime.After(time.Now().Add(-5 * time.Minute)) { + // Multiple failures in the last 5 minutes, fall back to proc parsing + log.Warnf("ebpf tracker died again, gently falling back to proc scanning") + t.useProcfs() + } else { + // Tolerable failure rate, restart the tracker + log.Warnf("ebpf tracker died, restarting it") + err := t.ebpfTracker.restart() + if err == nil { + go t.getInitialState() + t.performEbpfTrack(rpt, hostNodeID) + return + } + log.Warnf("could not restart ebpf tracker, falling back to proc scanning: %v", err) + t.useProcfs() + } } // consult the flowWalker for short-lived (conntracked) connections diff --git a/probe/endpoint/ebpf.go b/probe/endpoint/ebpf.go index 290e0b3f2..7b4b10fde 100644 --- a/probe/endpoint/ebpf.go +++ b/probe/endpoint/ebpf.go @@ -3,8 +3,11 @@ package endpoint import ( "bytes" "fmt" + "io/ioutil" + "os" "regexp" "strconv" + "strings" "sync" "syscall" @@ -30,9 +33,21 @@ type EbpfTracker struct { sync.Mutex tracer *tracer.Tracer ready bool + stopping bool dead bool lastTimestampV4 uint64 + // debugBPF specifies if EbpfTracker must be started in debug mode. This + // allows to easily debug issues like: + // https://github.com/weaveworks/scope/issues/2650 + // + // Scope could be started this way: + // $ sudo WEAVESCOPE_DOCKER_ARGS="-e SCOPE_DEBUG_BPF=1" ./scope launch + // + // Then, EbpfTracker could be tricked into restarting with: + // $ echo stop | sudo tee /proc/$(pidof scope-probe)/root/var/run/scope/debug-bpf + debugBPF bool + openConnections map[fourTuple]ebpfConnection closedConnections []ebpfConnection closedDuringInit map[fourTuple]struct{} @@ -77,24 +92,35 @@ func newEbpfTracker() (*EbpfTracker, error) { return nil, fmt.Errorf("kernel not supported: %v", err) } - tracker := &EbpfTracker{ - openConnections: map[fourTuple]ebpfConnection{}, - closedDuringInit: map[fourTuple]struct{}{}, + var debugBPF bool + if os.Getenv("SCOPE_DEBUG_BPF") != "" { + log.Infof("ebpf tracker started in debug mode") + debugBPF = true } - tracer, err := tracer.NewTracer(tracker) - if err != nil { + tracker := &EbpfTracker{ + debugBPF: debugBPF, + } + if err := tracker.restart(); err != nil { return nil, err } - tracker.tracer = tracer - tracer.Start() - return tracker, nil } // TCPEventV4 handles IPv4 TCP events from the eBPF tracer func (t *EbpfTracker) TCPEventV4(e tracer.TcpV4) { + if t.debugBPF { + debugBPFFile := "/var/run/scope/debug-bpf" + b, err := ioutil.ReadFile("/var/run/scope/debug-bpf") + if err == nil && strings.TrimSpace(string(b[:])) == "stop" { + os.Remove(debugBPFFile) + log.Warnf("ebpf tracker stopped as requested by user") + t.stop() + return + } + } + if t.lastTimestampV4 > e.Timestamp { // A kernel bug can cause the timestamps to be wrong (e.g. on Ubuntu with Linux 4.4.0-47.68) // Upgrading the kernel will fix the problem. For further info see: @@ -102,6 +128,7 @@ func (t *EbpfTracker) TCPEventV4(e tracer.TcpV4) { // https://github.com/weaveworks/scope/issues/2334 log.Errorf("tcp tracer received event with timestamp %v even though the last timestamp was %v. Stopping the eBPF tracker.", e.Timestamp, t.lastTimestampV4) t.stop() + return } t.lastTimestampV4 = e.Timestamp @@ -287,10 +314,45 @@ func (t *EbpfTracker) isDead() bool { func (t *EbpfTracker) stop() { t.Lock() - alreadyDead := t.dead - t.dead = true + alreadyDead := t.dead || t.stopping + t.stopping = true t.Unlock() - if !alreadyDead && t.tracer != nil { - t.tracer.Stop() - } + + // Do not call tracer.Stop() in this thread, otherwise tracer.Stop() will + // deadlock waiting for this thread to pick up the next event. + go func() { + if !alreadyDead && t.tracer != nil { + t.tracer.Stop() + t.tracer = nil + } + + // Only advertise the tracer as dead after the tracer is fully stopped so that + // restart() is not called in parallel in another thread. + t.Lock() + t.stopping = false + t.dead = true + t.Unlock() + }() +} + +func (t *EbpfTracker) restart() error { + t.Lock() + defer t.Unlock() + + t.dead = false + t.ready = false + + t.openConnections = map[fourTuple]ebpfConnection{} + t.closedDuringInit = map[fourTuple]struct{}{} + t.closedConnections = []ebpfConnection{} + + tracer, err := tracer.NewTracer(t) + if err != nil { + return err + } + + t.tracer = tracer + tracer.Start() + + return nil } diff --git a/probe/endpoint/ebpf_test.go b/probe/endpoint/ebpf_test.go index 14f9c8064..1bf182940 100644 --- a/probe/endpoint/ebpf_test.go +++ b/probe/endpoint/ebpf_test.go @@ -5,6 +5,7 @@ import ( "reflect" "strconv" "testing" + "time" "github.com/weaveworks/tcptracer-bpf/pkg/tracer" ) @@ -220,6 +221,14 @@ func TestInvalidTimeStampDead(t *testing.T) { if cnt != 2 { t.Errorf("walkConnections found %v instead of 2 connections", cnt) } + // EbpfTracker is marked as dead asynchronously. + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if mockEbpfTracker.isDead() { + break + } + time.Sleep(100 * time.Millisecond) + } if !mockEbpfTracker.isDead() { t.Errorf("expected ebpfTracker to be set to dead after events with wrong order") } From af14cf796de448c8e135fa5bb6640903c29178b7 Mon Sep 17 00:00:00 2001 From: Alban Crequy Date: Tue, 25 Jul 2017 16:48:39 +0200 Subject: [PATCH 2/3] integration test: restart EbpfTracker --- integration/315_ebpf_restart_test.sh | 55 ++++++++++++++++++++++++++++ integration/config.sh | 13 +++++-- 2 files changed, 64 insertions(+), 4 deletions(-) create mode 100755 integration/315_ebpf_restart_test.sh diff --git a/integration/315_ebpf_restart_test.sh b/integration/315_ebpf_restart_test.sh new file mode 100755 index 000000000..6235a6df7 --- /dev/null +++ b/integration/315_ebpf_restart_test.sh @@ -0,0 +1,55 @@ +#! /bin/bash + +# shellcheck disable=SC1091 +. ./config.sh + +start_suite "Test with ebpf restarts and proc fallback" + +weave_on "$HOST1" launch +# Manually start scope in order to start EbpfTracker in debug mode +DOCKER_HOST=tcp://${HOST1}:${DOCKER_PORT} CHECKPOINT_DISABLE=true \ + WEAVESCOPE_DOCKER_ARGS="-e SCOPE_DEBUG_BPF=1" \ + "${SCOPE}" launch + +server_on "$HOST1" +client_on "$HOST1" + +wait_for_containers "$HOST1" 60 nginx client + +has_container "$HOST1" nginx +has_container "$HOST1" client +has_connection containers "$HOST1" client nginx + +# shellcheck disable=SC2016 +run_on "$HOST1" 'echo stop | sudo tee /proc/$(pidof scope-probe)/root/var/run/scope/debug-bpf' +sleep 5 + +server_on "$HOST1" "nginx2" +client_on "$HOST1" "client2" "nginx2" + +wait_for_containers "$HOST1" 60 nginx2 client2 + +has_container "$HOST1" nginx2 +has_container "$HOST1" client2 +has_connection containers "$HOST1" client2 nginx2 + +# Save stdout for debugging output +exec 3>&1 +assert_raises "docker_on $HOST1 logs weavescope 2>&1 | grep 'ebpf tracker died, restarting it' || (docker_on $HOST1 logs weavescope 2>&3 ; false)" + +# shellcheck disable=SC2016 +run_on "$HOST1" 'echo stop | sudo tee /proc/$(pidof scope-probe)/root/var/run/scope/debug-bpf' +sleep 5 + +server_on "$HOST1" "nginx3" +client_on "$HOST1" "client3" "nginx3" + +wait_for_containers "$HOST1" 60 nginx3 client3 + +has_container "$HOST1" nginx3 +has_container "$HOST1" client3 +has_connection containers "$HOST1" client3 nginx3 + +assert_raises "docker_on $HOST1 logs weavescope 2>&1 | grep 'ebpf tracker died again, gently falling back to proc scanning' || (docker_on $HOST1 logs weavescope 2>&3 ; false)" + +scope_end_suite diff --git a/integration/config.sh b/integration/config.sh index 3fb8c4ccb..01b2eb5bc 100644 --- a/integration/config.sh +++ b/integration/config.sh @@ -36,13 +36,18 @@ weave_proxy_on() { } server_on() { - weave_proxy_on "$1" run -d --name nginx nginx + local host=$1 + local name=${2:-nginx} + weave_proxy_on "$1" run -d --name "$name" nginx } client_on() { - weave_proxy_on "$1" run -d --name client alpine /bin/sh -c "while true; do \ - wget http://nginx.weave.local:80/ -O - >/dev/null || true; \ - sleep 1; \ + local host=$1 + local name=${2:-client} + local server=${3:-nginx} + weave_proxy_on "$1" run -d --name "$name" alpine /bin/sh -c "while true; do \ + wget http://$server.weave.local:80/ -O - >/dev/null || true; \ + sleep 1; \ done" } From 93ca8b83ede738189f1c5c935d5f403c94a69cfd Mon Sep 17 00:00:00 2001 From: Alban Crequy Date: Wed, 9 Aug 2017 15:20:46 +0200 Subject: [PATCH 3/3] vendor: update tcptracer-bpf and gobpf This includes: - https://github.com/iovisor/gobpf/pull/70 perf: close go channels idiomatically - https://github.com/iovisor/gobpf/pull/70 close channels on the sender side & fix closing race - https://github.com/weaveworks/tcptracer-bpf/pull/50 vendor: update gobpf --- .../tcptracer-bpf/pkg/tracer/tracer.go | 23 +++- .../github.com/iovisor/gobpf/elf/elf.go | 22 +++- .../github.com/iovisor/gobpf/elf/module.go | 118 ++++++++++++++---- .../github.com/iovisor/gobpf/elf/perf.go | 33 +++-- vendor/manifest | 2 +- 5 files changed, 163 insertions(+), 35 deletions(-) diff --git a/vendor/github.com/weaveworks/tcptracer-bpf/pkg/tracer/tracer.go b/vendor/github.com/weaveworks/tcptracer-bpf/pkg/tracer/tracer.go index 8d725a36f..ad21afdbb 100644 --- a/vendor/github.com/weaveworks/tcptracer-bpf/pkg/tracer/tracer.go +++ b/vendor/github.com/weaveworks/tcptracer-bpf/pkg/tracer/tracer.go @@ -79,10 +79,19 @@ func NewTracer(cb Callback) (*Tracer, error) { for { select { case <-stopChan: + // On stop, stopChan will be closed but the other channels will + // also be closed shortly after. The select{} has no priorities, + // therefore, the "ok" value must be checked below. return - case data := <-channelV4: + case data, ok := <-channelV4: + if !ok { + return // see explanation above + } cb.TCPEventV4(tcpV4ToGo(&data)) - case lost := <-lostChanV4: + case lost, ok := <-lostChanV4: + if !ok { + return // see explanation above + } cb.LostV4(lost) } } @@ -93,9 +102,15 @@ func NewTracer(cb Callback) (*Tracer, error) { select { case <-stopChan: return - case data := <-channelV6: + case data, ok := <-channelV6: + if !ok { + return // see explanation above + } cb.TCPEventV6(tcpV6ToGo(&data)) - case lost := <-lostChanV6: + case lost, ok := <-lostChanV6: + if !ok { + return // see explanation above + } cb.LostV6(lost) } } diff --git a/vendor/github.com/weaveworks/tcptracer-bpf/vendor/github.com/iovisor/gobpf/elf/elf.go b/vendor/github.com/weaveworks/tcptracer-bpf/vendor/github.com/iovisor/gobpf/elf/elf.go index 5f9926e9c..c5ce0438e 100644 --- a/vendor/github.com/weaveworks/tcptracer-bpf/vendor/github.com/iovisor/gobpf/elf/elf.go +++ b/vendor/github.com/weaveworks/tcptracer-bpf/vendor/github.com/iovisor/gobpf/elf/elf.go @@ -515,6 +515,7 @@ func (b *Module) Load(parameters map[string]SectionParams) error { isCgroupSkb := strings.HasPrefix(secName, "cgroup/skb") isCgroupSock := strings.HasPrefix(secName, "cgroup/sock") isSocketFilter := strings.HasPrefix(secName, "socket") + isTracepoint := strings.HasPrefix(secName, "tracepoint/") var progType uint32 switch { @@ -528,9 +529,11 @@ func (b *Module) Load(parameters map[string]SectionParams) error { progType = uint32(C.BPF_PROG_TYPE_CGROUP_SOCK) case isSocketFilter: progType = uint32(C.BPF_PROG_TYPE_SOCKET_FILTER) + case isTracepoint: + progType = uint32(C.BPF_PROG_TYPE_TRACEPOINT) } - if isKprobe || isKretprobe || isCgroupSkb || isCgroupSock || isSocketFilter { + if isKprobe || isKretprobe || isCgroupSkb || isCgroupSock || isSocketFilter || isTracepoint { rdata, err := rsection.Data() if err != nil { return err @@ -579,6 +582,12 @@ func (b *Module) Load(parameters map[string]SectionParams) error { insns: insns, fd: int(progFd), } + case isTracepoint: + b.tracepointPrograms[secName] = &TracepointProgram{ + Name: secName, + insns: insns, + fd: int(progFd), + } } } } @@ -596,6 +605,7 @@ func (b *Module) Load(parameters map[string]SectionParams) error { isCgroupSkb := strings.HasPrefix(secName, "cgroup/skb") isCgroupSock := strings.HasPrefix(secName, "cgroup/sock") isSocketFilter := strings.HasPrefix(secName, "socket") + isTracepoint := strings.HasPrefix(secName, "tracepoint/") var progType uint32 switch { @@ -609,9 +619,11 @@ func (b *Module) Load(parameters map[string]SectionParams) error { progType = uint32(C.BPF_PROG_TYPE_CGROUP_SOCK) case isSocketFilter: progType = uint32(C.BPF_PROG_TYPE_SOCKET_FILTER) + case isTracepoint: + progType = uint32(C.BPF_PROG_TYPE_TRACEPOINT) } - if isKprobe || isKretprobe || isCgroupSkb || isCgroupSock || isSocketFilter { + if isKprobe || isKretprobe || isCgroupSkb || isCgroupSock || isSocketFilter || isTracepoint { data, err := section.Data() if err != nil { return err @@ -655,6 +667,12 @@ func (b *Module) Load(parameters map[string]SectionParams) error { insns: insns, fd: int(progFd), } + case isTracepoint: + b.tracepointPrograms[secName] = &TracepointProgram{ + Name: secName, + insns: insns, + fd: int(progFd), + } } } } diff --git a/vendor/github.com/weaveworks/tcptracer-bpf/vendor/github.com/iovisor/gobpf/elf/module.go b/vendor/github.com/weaveworks/tcptracer-bpf/vendor/github.com/iovisor/gobpf/elf/module.go index 4270806be..9b850d311 100644 --- a/vendor/github.com/weaveworks/tcptracer-bpf/vendor/github.com/iovisor/gobpf/elf/module.go +++ b/vendor/github.com/weaveworks/tcptracer-bpf/vendor/github.com/iovisor/gobpf/elf/module.go @@ -96,11 +96,12 @@ type Module struct { fileReader io.ReaderAt file *elf.File - log []byte - maps map[string]*Map - probes map[string]*Kprobe - cgroupPrograms map[string]*CgroupProgram - socketFilters map[string]*SocketFilter + log []byte + maps map[string]*Map + probes map[string]*Kprobe + cgroupPrograms map[string]*CgroupProgram + socketFilters map[string]*SocketFilter + tracepointPrograms map[string]*TracepointProgram } // Kprobe represents a kprobe or kretprobe and has to be declared @@ -134,13 +135,22 @@ type SocketFilter struct { fd int } +// TracepointProgram represents a tracepoint program +type TracepointProgram struct { + Name string + insns *C.struct_bpf_insn + fd int + efd int +} + func NewModule(fileName string) *Module { return &Module{ - fileName: fileName, - probes: make(map[string]*Kprobe), - cgroupPrograms: make(map[string]*CgroupProgram), - socketFilters: make(map[string]*SocketFilter), - log: make([]byte, 65536), + fileName: fileName, + probes: make(map[string]*Kprobe), + cgroupPrograms: make(map[string]*CgroupProgram), + socketFilters: make(map[string]*SocketFilter), + tracepointPrograms: make(map[string]*TracepointProgram), + log: make([]byte, 65536), } } @@ -186,6 +196,22 @@ func writeKprobeEvent(probeType, eventName, funcName, maxactiveStr string) (int, return kprobeId, nil } +func perfEventOpenTracepoint(id int, progFd int) (int, error) { + efd, err := C.perf_event_open_tracepoint(C.int(id), -1 /* pid */, 0 /* cpu */, -1 /* group_fd */, C.PERF_FLAG_FD_CLOEXEC) + if efd < 0 { + return -1, fmt.Errorf("perf_event_open error: %v", err) + } + + if _, _, err := syscall.Syscall(syscall.SYS_IOCTL, uintptr(efd), C.PERF_EVENT_IOC_ENABLE, 0); err != 0 { + return -1, fmt.Errorf("error enabling perf event: %v", err) + } + + if _, _, err := syscall.Syscall(syscall.SYS_IOCTL, uintptr(efd), C.PERF_EVENT_IOC_SET_BPF, uintptr(progFd)); err != 0 { + return -1, fmt.Errorf("error attaching bpf program to perf event: %v", err) + } + return int(efd), nil +} + // EnableKprobe enables a kprobe/kretprobe identified by secName. // For kretprobes, you can configure the maximum number of instances // of the function that can be probed simultaneously with maxactive. @@ -222,22 +248,43 @@ func (b *Module) EnableKprobe(secName string, maxactive int) error { return err } - efd := C.perf_event_open_tracepoint(C.int(kprobeId), -1 /* pid */, 0 /* cpu */, -1 /* group_fd */, C.PERF_FLAG_FD_CLOEXEC) - if efd < 0 { - return fmt.Errorf("perf_event_open for kprobe error") + probe.efd, err = perfEventOpenTracepoint(kprobeId, progFd) + return err +} + +func writeTracepointEvent(category, name string) (int, error) { + tracepointIdFile := fmt.Sprintf("/sys/kernel/debug/tracing/events/%s/%s/id", category, name) + tracepointIdBytes, err := ioutil.ReadFile(tracepointIdFile) + if err != nil { + return -1, fmt.Errorf("cannot read tracepoint id %q: %v", tracepointIdFile, err) } - _, _, err2 := syscall.Syscall(syscall.SYS_IOCTL, uintptr(efd), C.PERF_EVENT_IOC_ENABLE, 0) - if err2 != 0 { - return fmt.Errorf("error enabling perf event: %v", err2) + tracepointId, err := strconv.Atoi(strings.TrimSpace(string(tracepointIdBytes))) + if err != nil { + return -1, fmt.Errorf("invalid tracepoint id: %v\n", err) } - _, _, err2 = syscall.Syscall(syscall.SYS_IOCTL, uintptr(efd), C.PERF_EVENT_IOC_SET_BPF, uintptr(progFd)) - if err2 != 0 { - return fmt.Errorf("error enabling perf event: %v", err2) + return tracepointId, nil +} + +func (b *Module) EnableTracepoint(secName string) error { + prog, ok := b.tracepointPrograms[secName] + if !ok { + return fmt.Errorf("no such tracepoint program %q", secName) } - probe.efd = int(efd) - return nil + progFd := prog.fd + + tracepointGroup := strings.SplitN(secName, "/", 3) + category := tracepointGroup[1] + name := tracepointGroup[2] + + tracepointId, err := writeTracepointEvent(category, name) + if err != nil { + return err + } + + prog.efd, err = perfEventOpenTracepoint(tracepointId, progFd) + return err } // IterKprobes returns a channel that emits the kprobes that included in the @@ -277,6 +324,17 @@ func (b *Module) IterCgroupProgram() <-chan *CgroupProgram { return ch } +func (b *Module) IterTracepointProgram() <-chan *TracepointProgram { + ch := make(chan *TracepointProgram) + go func() { + for name := range b.tracepointPrograms { + ch <- b.tracepointPrograms[name] + } + close(ch) + }() + return ch +} + func (b *Module) CgroupProgram(name string) *CgroupProgram { return b.cgroupPrograms[name] } @@ -412,6 +470,21 @@ func (b *Module) closeProbes() error { return nil } +func (b *Module) closeTracepointPrograms() error { + for _, program := range b.tracepointPrograms { + if program.efd != -1 { + if err := syscall.Close(program.efd); err != nil { + return fmt.Errorf("error closing perf event fd: %v", err) + } + program.efd = -1 + } + if err := syscall.Close(program.fd); err != nil { + return fmt.Errorf("error closing tracepoint program fd: %v", err) + } + } + return nil +} + func (b *Module) closeCgroupPrograms() error { for _, program := range b.cgroupPrograms { if err := syscall.Close(program.fd); err != nil { @@ -490,6 +563,9 @@ func (b *Module) CloseExt(options map[string]CloseOptions) error { if err := b.closeCgroupPrograms(); err != nil { return err } + if err := b.closeTracepointPrograms(); err != nil { + return err + } if err := b.closeSocketFilters(); err != nil { return err } diff --git a/vendor/github.com/weaveworks/tcptracer-bpf/vendor/github.com/iovisor/gobpf/elf/perf.go b/vendor/github.com/weaveworks/tcptracer-bpf/vendor/github.com/iovisor/gobpf/elf/perf.go index 75ba3f8e6..1ae543a16 100644 --- a/vendor/github.com/weaveworks/tcptracer-bpf/vendor/github.com/iovisor/gobpf/elf/perf.go +++ b/vendor/github.com/weaveworks/tcptracer-bpf/vendor/github.com/iovisor/gobpf/elf/perf.go @@ -109,7 +109,7 @@ type PerfMap struct { pageCount int receiverChan chan []byte lostChan chan uint64 - pollStop chan bool + pollStop chan struct{} timestamp func(*[]byte) uint64 } @@ -125,6 +125,9 @@ func InitPerfMap(b *Module, mapName string, receiverChan chan []byte, lostChan c if !ok { return nil, fmt.Errorf("no map with name %s", mapName) } + if receiverChan == nil { + return nil, fmt.Errorf("receiverChan is nil") + } // Maps are initialized in b.Load(), nothing to do here return &PerfMap{ name: mapName, @@ -132,7 +135,7 @@ func InitPerfMap(b *Module, mapName string, receiverChan chan []byte, lostChan c pageCount: m.pageCount, receiverChan: receiverChan, lostChan: lostChan, - pollStop: make(chan bool), + pollStop: make(chan struct{}), }, nil } @@ -163,6 +166,13 @@ func (pm *PerfMap) PollStart() { pageSize := os.Getpagesize() state := C.struct_read_state{} + defer func() { + close(pm.receiverChan) + if pm.lostChan != nil { + close(pm.lostChan) + } + }() + for { select { case <-pm.pollStop: @@ -208,7 +218,11 @@ func (pm *PerfMap) PollStart() { } case C.PERF_RECORD_LOST: if pm.lostChan != nil { - pm.lostChan <- lost.Lost + select { + case pm.lostChan <- lost.Lost: + case <-pm.pollStop: + return + } } default: // ignore unknown events @@ -226,7 +240,11 @@ func (pm *PerfMap) PollStart() { // elements also must not be processed now. break harvestLoop } - pm.receiverChan <- incoming.bytesArray[0] + select { + case pm.receiverChan <- incoming.bytesArray[0]: + case <-pm.pollStop: + return + } // remove first element incoming.bytesArray = incoming.bytesArray[1:] } @@ -238,10 +256,11 @@ func (pm *PerfMap) PollStart() { }() } -// PollStop stops the goroutine that polls the perf event map. Make -// sure to close the receiverChan only *after* calling PollStop. +// PollStop stops the goroutine that polls the perf event map. +// Callers must not close receiverChan or lostChan: they will be automatically +// closed on the sender side. func (pm *PerfMap) PollStop() { - pm.pollStop <- true + close(pm.pollStop) } func perfEventPoll(fds []C.int) error { diff --git a/vendor/manifest b/vendor/manifest index 63d1395ec..a595c8e7e 100644 --- a/vendor/manifest +++ b/vendor/manifest @@ -1020,7 +1020,7 @@ "importpath": "github.com/weaveworks/tcptracer-bpf", "repository": "https://github.com/weaveworks/tcptracer-bpf", "vcs": "git", - "revision": "53a889d82fbd8a4fc4ef3f0013a9f88a3e9f4ccc", + "revision": "e080bd747dc6b62d4ed3ed2b7f0be4801bef8faf", "branch": "master", "notests": true },