From f922ea19c82088878ac9ebe340fd238dab04b0e9 Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Thu, 4 Feb 2016 12:33:04 +0000
Subject: [PATCH 01/17] Rate-limit reading proc files

Use a reader in the background, dynamically rate-limited, reading the required
files in a loop
---
 .../procspy/background_reader_linux.go        | 110 ++++++++++++++++++
 probe/endpoint/procspy/proc_linux.go          |   4 +-
 probe/endpoint/procspy/spy_linux.go           |   7 +-
 probe/endpoint/reporter.go                    |   1 +
 4 files changed, 118 insertions(+), 4 deletions(-)
 create mode 100644 probe/endpoint/procspy/background_reader_linux.go

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
new file mode 100644
index 000000000..3c5ef0f2f
--- /dev/null
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -0,0 +1,110 @@
+package procspy
+
+import (
+	"bytes"
+	"fmt"
+	"log"
+	"math"
+	"sync"
+	"time"
+
+	"github.com/weaveworks/scope/probe/process"
+)
+
+const (
+	initialRateLimit = 100 * time.Millisecond // read 10 namespaces per second
+	maxRateLimit     = 250 * time.Millisecond // lead at least 4 namespaces per second
+	targetWalkTime   = 10 * time.Second
+
+	maxRateLimitF   = float64(maxRateLimit)
+	targetWalkTimeF = float64(targetWalkTime)
+)
+
+type backgroundReader struct {
+	walker       process.Walker
+	mtx          sync.Mutex
+	walkingBuf   *bytes.Buffer
+	readyBuf     *bytes.Buffer
+	readySockets map[uint64]*Proc
+}
+
+// HACK: Pretty ugly singleton interface (particularly the part of passing
+// the walker to StartBackgroundReader() and ignoring it in in Connections() )
+// experimenting with this for now.
+var singleton *backgroundReader
+
+func getBackgroundReader() (*backgroundReader, error) {
+	var err error
+	if singleton == nil {
+		err = fmt.Errorf("background reader hasn't yet been started")
+	}
+	return singleton, err
+}
+
+// StartBackgroundReader starts a ratelimited background goroutine to
+// read the expensive files from proc.
+func StartBackgroundReader(walker process.Walker) {
+	if singleton != nil {
+		return
+	}
+	singleton = &backgroundReader{
+		walker:     walker,
+		walkingBuf: bytes.NewBuffer(make([]byte, 0, 5000)),
+		readyBuf:   bytes.NewBuffer(make([]byte, 0, 5000)),
+	}
+	go singleton.loop()
+}
+
+func (br *backgroundReader) loop() {
+	rateLimit := initialRateLimit
+
+	namespaceTicker := time.Tick(rateLimit)
+
+	for {
+		start := time.Now()
+		sockets, err := walkProcPid(br.walkingBuf, br.walker, namespaceTicker)
+		if err != nil {
+			log.Printf("background reader: error walking /proc: %s\n", err)
+			continue
+		}
+		walkTime := time.Now().Sub(start)
+		walkTimeF := float64(walkTime)
+
+		log.Printf("debug: background reader: full pass took %s\n", walkTime)
+		if walkTimeF/targetWalkTimeF > 1.5 {
+			log.Printf(
+				"warn: background reader: full pass took %s: 50%% more than expected (%s)\n",
+				walkTime,
+				targetWalkTime,
+			)
+		}
+
+		// Adjust rate limit to more-accurately meet the target walk time in next iteration
+		scaledRateLimit := targetWalkTimeF / walkTimeF * float64(rateLimit)
+		rateLimit = time.Duration(math.Min(scaledRateLimit, maxRateLimitF))
+		log.Printf("debug: background reader: new rate limit %s\n", rateLimit)
+
+		namespaceTicker = time.Tick(rateLimit)
+
+		// Swap buffers
+		br.mtx.Lock()
+		br.readyBuf, br.walkingBuf = br.walkingBuf, br.readyBuf
+		br.readySockets = sockets
+		br.mtx.Unlock()
+
+		br.walkingBuf.Reset()
+
+		// Sleep during spare time
+		time.Sleep(targetWalkTime - walkTime)
+	}
+}
+
+func (br *backgroundReader) getWalkedProcPid(buf *bytes.Buffer) map[uint64]*Proc {
+	br.mtx.Lock()
+	defer br.mtx.Unlock()
+
+	reader := bytes.NewReader(br.readyBuf.Bytes())
+	buf.ReadFrom(reader)
+
+	return br.readySockets
+}
diff --git a/probe/endpoint/procspy/proc_linux.go b/probe/endpoint/procspy/proc_linux.go
index 87ac4bd43..9545415f5 100644
--- a/probe/endpoint/procspy/proc_linux.go
+++ b/probe/endpoint/procspy/proc_linux.go
@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"syscall"
+	"time"
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/armon/go-metrics"
@@ -143,7 +144,7 @@ func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProc
 // /proc/PID/net/tcp{,6} for each namespace and sees if the ./fd/* files of each
 // process in that namespace are symlinks to sockets. Returns a map from socket
 // ID (inode) to PID.
-func walkProcPid(buf *bytes.Buffer, walker process.Walker) (map[uint64]*Proc, error) {
+func walkProcPid(buf *bytes.Buffer, walker process.Walker, namespaceTicker <-chan time.Time) (map[uint64]*Proc, error) {
 	var (
 		sockets    = map[uint64]*Proc{}              // map socket inode -> process
 		namespaces = map[uint64][]*process.Process{} // map network namespace id -> processes
@@ -171,6 +172,7 @@ func walkProcPid(buf *bytes.Buffer, walker process.Walker) (map[uint64]*Proc, er
 	})
 
 	for _, procs := range namespaces {
+		<-namespaceTicker
 		walkNamespacePid(buf, sockets, procs)
 	}
 
diff --git a/probe/endpoint/procspy/spy_linux.go b/probe/endpoint/procspy/spy_linux.go
index 68b0d7c0b..4429aa277 100644
--- a/probe/endpoint/procspy/spy_linux.go
+++ b/probe/endpoint/procspy/spy_linux.go
@@ -33,17 +33,18 @@ func (c *pnConnIter) Next() *Connection {
 }
 
 // cbConnections sets Connections()
-var cbConnections = func(processes bool, walker process.Walker) (ConnIter, error) {
+var cbConnections = func(processes bool, _ process.Walker) (ConnIter, error) {
 	// buffer for contents of /proc/<pid>/net/tcp
 	buf := bufPool.Get().(*bytes.Buffer)
 	buf.Reset()
 
 	var procs map[uint64]*Proc
 	if processes {
-		var err error
-		if procs, err = walkProcPid(buf, walker); err != nil {
+		br, err := getBackgroundReader()
+		if err != nil {
 			return nil, err
 		}
+		procs = br.getWalkedProcPid(buf)
 	}
 
 	if buf.Len() == 0 {
diff --git a/probe/endpoint/reporter.go b/probe/endpoint/reporter.go
index c2f9c83f0..2adf1c9e2 100644
--- a/probe/endpoint/reporter.go
+++ b/probe/endpoint/reporter.go
@@ -49,6 +49,7 @@ var SpyDuration = prometheus.NewSummaryVec(
 // is stored in the Endpoint topology. It optionally enriches that topology
 // with process (PID) information.
 func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bool, procWalker process.Walker) *Reporter {
+	procspy.StartBackgroundReader(procWalker)
 	return &Reporter{
 		hostID:           hostID,
 		hostName:         hostName,

From 87dd43f782ccfe93806c59bfb300e38036ae37bd Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Thu, 4 Feb 2016 15:08:36 +0000
Subject: [PATCH 02/17] Batch rate-limit to stats of /proc/*/fd/*

---
 .../procspy/background_reader_linux.go        | 13 +--
 probe/endpoint/procspy/proc_linux.go          | 86 ++++++++++++++-----
 2 files changed, 70 insertions(+), 29 deletions(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index 3c5ef0f2f..2aea73bbe 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -12,9 +12,10 @@ import (
 )
 
 const (
-	initialRateLimit = 100 * time.Millisecond // read 10 namespaces per second
-	maxRateLimit     = 250 * time.Millisecond // lead at least 4 namespaces per second
-	targetWalkTime   = 10 * time.Second
+	initialRateLimit = 50 * time.Millisecond  // Read 20 * fdBlockSize file descriptors (/proc/PID/fd/*) per namespace per second
+	maxRateLimit     = 250 * time.Millisecond // Read at least 4 * fdBlockSize file descriptors per namespace per second
+	fdBlockSize      = 100
+	targetWalkTime   = 10 * time.Second // Aim at walking all files in 10 seconds
 
 	maxRateLimitF   = float64(maxRateLimit)
 	targetWalkTimeF = float64(targetWalkTime)
@@ -58,11 +59,11 @@ func StartBackgroundReader(walker process.Walker) {
 func (br *backgroundReader) loop() {
 	rateLimit := initialRateLimit
 
-	namespaceTicker := time.Tick(rateLimit)
+	ticker := time.Tick(rateLimit)
 
 	for {
 		start := time.Now()
-		sockets, err := walkProcPid(br.walkingBuf, br.walker, namespaceTicker)
+		sockets, err := walkProcPid(br.walkingBuf, br.walker, ticker, fdBlockSize)
 		if err != nil {
 			log.Printf("background reader: error walking /proc: %s\n", err)
 			continue
@@ -84,7 +85,7 @@ func (br *backgroundReader) loop() {
 		rateLimit = time.Duration(math.Min(scaledRateLimit, maxRateLimitF))
 		log.Printf("debug: background reader: new rate limit %s\n", rateLimit)
 
-		namespaceTicker = time.Tick(rateLimit)
+		ticker = time.Tick(rateLimit)
 
 		// Swap buffers
 		br.mtx.Lock()
diff --git a/probe/endpoint/procspy/proc_linux.go b/probe/endpoint/procspy/proc_linux.go
index 9545415f5..ec0a9d492 100644
--- a/probe/endpoint/procspy/proc_linux.go
+++ b/probe/endpoint/procspy/proc_linux.go
@@ -69,46 +69,82 @@ func getNetNamespacePathSuffix() string {
 	return netNamespacePathSuffix
 }
 
-// walkNamespacePid does the work of walkProcPid for a single namespace
-func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProcs []*process.Process) {
+// Read the connections for a group of processes living in the same namespace,
+// which are found (identically) in /proc/PID/net/tcp{,6} for any of the
+// processes.
+func readProcessConnections(buf *bytes.Buffer, namespaceProcs []*process.Process) (bool, error) {
+	var (
+		errRead  error
+		errRead6 error
+		read     int64
+		read6    int64
+	)
 
-	// Read the connections for the namespace, which are found (identically) in
-	// /proc/PID/net/tcp{,6} for any of the processes in the namespace.
-	var tcpSuccess bool
 	for _, p := range namespaceProcs {
 		dirName := strconv.Itoa(p.PID)
 
-		read, errRead := readFile(filepath.Join(procRoot, dirName, "/net/tcp"), buf)
-		read6, errRead6 := readFile(filepath.Join(procRoot, dirName, "/net/tcp6"), buf)
+		read, errRead = readFile(filepath.Join(procRoot, dirName, "/net/tcp"), buf)
+		read6, errRead6 = readFile(filepath.Join(procRoot, dirName, "/net/tcp6"), buf)
 
 		if errRead != nil || errRead6 != nil {
 			// try next process
 			continue
 		}
-
-		if read+read6 == 0 {
-			// No connections, don't bother reading /fd/*
-			return
-		}
-
-		tcpSuccess = true
-		break
+		return read+read6 > 0, nil
 	}
 
-	if !tcpSuccess {
+	// would be cool to have an or operation between errors
+	if errRead != nil {
+		return false, errRead
+	}
+	if errRead6 != nil {
+		return false, errRead6
+	}
+
+	return false, nil
+
+}
+
+// walkNamespacePid does the work of walkProcPid for a single namespace
+func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProcs []*process.Process, ticker <-chan time.Time, fdBlockSize int) error {
+
+	found, err := readProcessConnections(buf, namespaceProcs)
+	if err != nil {
+		return err
+	}
+	if !found {
 		// There's no point in reading /fd/*
-		return
+		return nil
 	}
 
-	// Get the sockets for all the processes in the namespace
 	var statT syscall.Stat_t
-	for _, p := range namespaceProcs {
+	var fdBlockCount int
+	for i, p := range namespaceProcs {
+
+		// Get the sockets for all the processes in the namespace
 		dirName := strconv.Itoa(p.PID)
 		fdBase := filepath.Join(procRoot, dirName, "fd")
 
+		if fdBlockCount > fdBlockSize {
+			// we surpased the filedescriptor rate limit
+			<-ticker
+			fdBlockCount = 0
+			// read the connections again to
+			// avoid the race between between /net/tcp{,6} and /proc/PID/fd/*
+			// FIXME:refactor this
+			found, err := readProcessConnections(buf, namespaceProcs[i:])
+			if err != nil {
+				return err
+			}
+			if !found {
+				// There's no point in reading /fd/*
+				return nil
+			}
+		}
+
 		fds, err := fs.ReadDirNames(fdBase)
 		if err != nil {
-			// Process is be gone by now, or we don't have access.
+			// Process is gone by now, or we don't have access.
 			continue
 		}
 
@@ -135,16 +171,20 @@ func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProc
 			}
 
 			sockets[statT.Ino] = proc
+
+			fdBlockCount += 1
 		}
 
 	}
+
+	return nil
 }
 
 // walkProcPid walks over all numerical (PID) /proc entries. It reads
 // /proc/PID/net/tcp{,6} for each namespace and sees if the ./fd/* files of each
 // process in that namespace are symlinks to sockets. Returns a map from socket
 // ID (inode) to PID.
-func walkProcPid(buf *bytes.Buffer, walker process.Walker, namespaceTicker <-chan time.Time) (map[uint64]*Proc, error) {
+func walkProcPid(buf *bytes.Buffer, walker process.Walker, ticker <-chan time.Time, fdBlockSize int) (map[uint64]*Proc, error) {
 	var (
 		sockets    = map[uint64]*Proc{}              // map socket inode -> process
 		namespaces = map[uint64][]*process.Process{} // map network namespace id -> processes
@@ -172,8 +212,8 @@ func walkProcPid(buf *bytes.Buffer, walker process.Walker, namespaceTicker <-cha
 	})
 
 	for _, procs := range namespaces {
-		<-namespaceTicker
-		walkNamespacePid(buf, sockets, procs)
+		<-ticker
+		walkNamespacePid(buf, sockets, procs, ticker, fdBlockSize)
 	}
 
 	metrics.SetGauge(namespaceKey, float32(len(namespaces)))

From 6deeca0380ebffb419d4e3dbc3b49d169dd46f2f Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Fri, 5 Feb 2016 15:01:09 +0000
Subject: [PATCH 03/17] Cleanup

---
 .../procspy/background_reader_linux.go        | 10 ++++-----
 probe/endpoint/procspy/proc_linux.go          | 22 +++++--------------
 2 files changed, 11 insertions(+), 21 deletions(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index 2aea73bbe..3007db945 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -16,9 +16,6 @@ const (
 	maxRateLimit     = 250 * time.Millisecond // Read at least 4 * fdBlockSize file descriptors per namespace per second
 	fdBlockSize      = 100
 	targetWalkTime   = 10 * time.Second // Aim at walking all files in 10 seconds
-
-	maxRateLimitF   = float64(maxRateLimit)
-	targetWalkTimeF = float64(targetWalkTime)
 )
 
 type backgroundReader struct {
@@ -57,10 +54,13 @@ func StartBackgroundReader(walker process.Walker) {
 }
 
 func (br *backgroundReader) loop() {
+	const (
+		maxRateLimitF   = float64(maxRateLimit)
+		targetWalkTimeF = float64(targetWalkTime)
+	)
+
 	rateLimit := initialRateLimit
-
 	ticker := time.Tick(rateLimit)
-
 	for {
 		start := time.Now()
 		sockets, err := walkProcPid(br.walkingBuf, br.walker, ticker, fdBlockSize)
diff --git a/probe/endpoint/procspy/proc_linux.go b/probe/endpoint/procspy/proc_linux.go
index ec0a9d492..2af9b999d 100644
--- a/probe/endpoint/procspy/proc_linux.go
+++ b/probe/endpoint/procspy/proc_linux.go
@@ -108,14 +108,9 @@ func readProcessConnections(buf *bytes.Buffer, namespaceProcs []*process.Process
 // walkNamespacePid does the work of walkProcPid for a single namespace
 func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProcs []*process.Process, ticker <-chan time.Time, fdBlockSize int) error {
 
-	found, err := readProcessConnections(buf, namespaceProcs)
-	if err != nil {
+	if found, err := readProcessConnections(buf, namespaceProcs); err != nil || !found {
 		return err
 	}
-	if !found {
-		// There's no point in reading /fd/*
-		return nil
-	}
 
 	var statT syscall.Stat_t
 	var fdBlockCount int
@@ -126,20 +121,15 @@ func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProc
 		fdBase := filepath.Join(procRoot, dirName, "fd")
 
 		if fdBlockCount > fdBlockSize {
-			// we surpased the filedescriptor rate limit
+			// we surpassed the filedescriptor rate limit
 			<-ticker
 			fdBlockCount = 0
+
 			// read the connections again to
 			// avoid the race between between /net/tcp{,6} and /proc/PID/fd/*
-			// FIXME:refactor this
-			found, err := readProcessConnections(buf, namespaceProcs[i:])
-			if err != nil {
+			if found, err := readProcessConnections(buf, namespaceProcs[i:]); err != nil || !found {
 				return err
 			}
-			if !found {
-				// There's no point in reading /fd/*
-				return nil
-			}
 		}
 
 		fds, err := fs.ReadDirNames(fdBase)
@@ -150,6 +140,8 @@ func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProc
 
 		var proc *Proc
 		for _, fd := range fds {
+			fdBlockCount++
+
 			// Direct use of syscall.Stat() to save garbage.
 			err = fs.Stat(filepath.Join(fdBase, fd), &statT)
 			if err != nil {
@@ -171,8 +163,6 @@ func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProc
 			}
 
 			sockets[statT.Ino] = proc
-
-			fdBlockCount += 1
 		}
 
 	}

From d4c68f48fac9393d43108a9c595be4e46c897388 Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Fri, 5 Feb 2016 18:51:30 +0000
Subject: [PATCH 04/17] Get rid of the package-level Connections func

---
 .../procspy/background_reader_linux.go        | 60 ++++++++++++-------
 probe/endpoint/procspy/fixture.go             | 23 +++----
 probe/endpoint/procspy/spy.go                 | 18 +++---
 probe/endpoint/procspy/spy_darwin.go          | 11 +++-
 probe/endpoint/procspy/spy_linux.go           | 23 ++++---
 probe/endpoint/reporter.go                    |  8 +--
 6 files changed, 86 insertions(+), 57 deletions(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index 3007db945..51eb6b7e7 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -21,36 +21,43 @@ const (
 type backgroundReader struct {
 	walker       process.Walker
 	mtx          sync.Mutex
+	running      bool
+	pleaseStop   bool
 	walkingBuf   *bytes.Buffer
 	readyBuf     *bytes.Buffer
 	readySockets map[uint64]*Proc
 }
 
-// HACK: Pretty ugly singleton interface (particularly the part of passing
-// the walker to StartBackgroundReader() and ignoring it in in Connections() )
-// experimenting with this for now.
-var singleton *backgroundReader
-
-func getBackgroundReader() (*backgroundReader, error) {
-	var err error
-	if singleton == nil {
-		err = fmt.Errorf("background reader hasn't yet been started")
-	}
-	return singleton, err
-}
-
-// StartBackgroundReader starts a ratelimited background goroutine to
-// read the expensive files from proc.
-func StartBackgroundReader(walker process.Walker) {
-	if singleton != nil {
-		return
-	}
-	singleton = &backgroundReader{
+// starts a rate-limited background goroutine to read the expensive files from
+// proc.
+func newBackgroundReader(walker process.Walker) *backgroundReader {
+	br := &backgroundReader{
 		walker:     walker,
 		walkingBuf: bytes.NewBuffer(make([]byte, 0, 5000)),
 		readyBuf:   bytes.NewBuffer(make([]byte, 0, 5000)),
 	}
-	go singleton.loop()
+	return br
+}
+
+func (br *backgroundReader) start() error {
+	br.mtx.Lock()
+	defer br.mtx.Unlock()
+	if br.running {
+		return fmt.Errorf("background reader already running")
+	}
+	br.running = true
+	go br.loop()
+	return nil
+}
+
+func (br *backgroundReader) stop() error {
+	br.mtx.Lock()
+	defer br.mtx.Unlock()
+	if !br.running {
+		return fmt.Errorf("background reader already not running")
+	}
+	br.pleaseStop = true
+	return nil
 }
 
 func (br *backgroundReader) loop() {
@@ -87,8 +94,17 @@ func (br *backgroundReader) loop() {
 
 		ticker = time.Tick(rateLimit)
 
-		// Swap buffers
 		br.mtx.Lock()
+
+		// Should we stop?
+		if br.pleaseStop {
+			br.pleaseStop = false
+			br.running = false
+			br.mtx.Unlock()
+			return
+		}
+
+		// Swap buffers
 		br.readyBuf, br.walkingBuf = br.walkingBuf, br.readyBuf
 		br.readySockets = sockets
 		br.mtx.Unlock()
diff --git a/probe/endpoint/procspy/fixture.go b/probe/endpoint/procspy/fixture.go
index 2e53c8190..b8d0ba02d 100644
--- a/probe/endpoint/procspy/fixture.go
+++ b/probe/endpoint/procspy/fixture.go
@@ -1,13 +1,5 @@
 package procspy
 
-import (
-	"github.com/weaveworks/scope/probe/process"
-)
-
-// SetFixtures declares constant Connection and ConnectionProcs which will
-// always be returned by the package-level Connections and Processes
-// functions. It's designed to be used in tests.
-
 type fixedConnIter []Connection
 
 func (f *fixedConnIter) Next() *Connection {
@@ -21,10 +13,13 @@ func (f *fixedConnIter) Next() *Connection {
 	return &car
 }
 
-// SetFixtures is used in test scenarios to have known output.
-func SetFixtures(c []Connection) {
-	cbConnections = func(bool, process.Walker) (ConnIter, error) {
-		f := fixedConnIter(c)
-		return &f, nil
-	}
+// fixedScanner implements ConnectionScanner and uses constant Connection and
+// ConnectionProcs.  It's designed to be used in tests.
+type fixedScanner []Connection
+
+func (s fixedScanner) Connections() (ConnIter, error) {
+	iter := fixedConnIter(s)
+	return &iter, nil
 }
+
+func (s fixedScanner) Stop() {}
diff --git a/probe/endpoint/procspy/spy.go b/probe/endpoint/procspy/spy.go
index 27a748256..78312fe64 100644
--- a/probe/endpoint/procspy/spy.go
+++ b/probe/endpoint/procspy/spy.go
@@ -5,8 +5,6 @@ package procspy
 
 import (
 	"net"
-
-	"github.com/weaveworks/scope/probe/process"
 )
 
 const (
@@ -35,11 +33,13 @@ type ConnIter interface {
 	Next() *Connection
 }
 
-// Connections returns all established (TCP) connections.  If processes is
-// false we'll just list all TCP connections, and there is no need to be root.
-// If processes is true it'll additionally try to lookup the process owning the
-// connection, filling in the Proc field. You will need to run this as root to
-// find all processes.
-func Connections(processes bool, walker process.Walker) (ConnIter, error) {
-	return cbConnections(processes, walker)
+type ConnectionScanner interface {
+	// Connections returns all established (TCP) connections.  If processes is
+	// false we'll just list all TCP connections, and there is no need to be root.
+	// If processes is true it'll additionally try to lookup the process owning the
+	// connection, filling in the Proc field. You will need to run this as root to
+	// find all processes.
+	Connections(processes bool) (ConnIter, error)
+	// Stops the scanning
+	Stop()
 }
diff --git a/probe/endpoint/procspy/spy_darwin.go b/probe/endpoint/procspy/spy_darwin.go
index 56edb52a7..d14400bfe 100644
--- a/probe/endpoint/procspy/spy_darwin.go
+++ b/probe/endpoint/procspy/spy_darwin.go
@@ -13,10 +13,16 @@ const (
 	lsofBinary    = "lsof"
 )
 
+func NewConnectionScanner(_ process.Walker) ConnectionScanner {
+	return &darwinScanner{}
+}
+
+type darwinScanner struct{}
+
 // Connections returns all established (TCP) connections. No need to be root
 // to run this. If processes is true it also tries to fill in the process
 // fields of the connection. You need to be root to find all processes.
-var cbConnections = func(processes bool, walker process.Walker) (ConnIter, error) {
+func (s *darwinScanner) Connections(processes bool) (ConnIter, error) {
 	out, err := exec.Command(
 		netstatBinary,
 		"-n", // no number resolving
@@ -62,3 +68,6 @@ var cbConnections = func(processes bool, walker process.Walker) (ConnIter, error
 	f := fixedConnIter(connections)
 	return &f, nil
 }
+
+// Nothing to stop since there's nothing running in the background
+func (s *darwinScanner) Stop() {}
diff --git a/probe/endpoint/procspy/spy_linux.go b/probe/endpoint/procspy/spy_linux.go
index 4429aa277..65b5829a2 100644
--- a/probe/endpoint/procspy/spy_linux.go
+++ b/probe/endpoint/procspy/spy_linux.go
@@ -32,19 +32,24 @@ func (c *pnConnIter) Next() *Connection {
 	return n
 }
 
-// cbConnections sets Connections()
-var cbConnections = func(processes bool, _ process.Walker) (ConnIter, error) {
+func NewConnectionScanner(walker process.Walker) ConnectionScanner {
+	br := newBackgroundReader(walker)
+	br.start()
+	return &linuxScanner{br}
+}
+
+type linuxScanner struct {
+	br *backgroundReader
+}
+
+func (s *linuxScanner) Connections(processes bool) (ConnIter, error) {
 	// buffer for contents of /proc/<pid>/net/tcp
 	buf := bufPool.Get().(*bytes.Buffer)
 	buf.Reset()
 
 	var procs map[uint64]*Proc
 	if processes {
-		br, err := getBackgroundReader()
-		if err != nil {
-			return nil, err
-		}
-		procs = br.getWalkedProcPid(buf)
+		procs = s.br.getWalkedProcPid(buf)
 	}
 
 	if buf.Len() == 0 {
@@ -58,3 +63,7 @@ var cbConnections = func(processes bool, _ process.Walker) (ConnIter, error) {
 		procs: procs,
 	}, nil
 }
+
+func (s *linuxScanner) Stop() {
+	s.br.stop()
+}
diff --git a/probe/endpoint/reporter.go b/probe/endpoint/reporter.go
index 2adf1c9e2..b88b0c405 100644
--- a/probe/endpoint/reporter.go
+++ b/probe/endpoint/reporter.go
@@ -26,7 +26,7 @@ type Reporter struct {
 	includeProcesses bool
 	includeNAT       bool
 	flowWalker       flowWalker // interface
-	procWalker       process.Walker
+	scanner          procspy.ConnectionScanner
 	natMapper        natMapper
 	reverseResolver  *reverseResolver
 }
@@ -49,7 +49,6 @@ var SpyDuration = prometheus.NewSummaryVec(
 // is stored in the Endpoint topology. It optionally enriches that topology
 // with process (PID) information.
 func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bool, procWalker process.Walker) *Reporter {
-	procspy.StartBackgroundReader(procWalker)
 	return &Reporter{
 		hostID:           hostID,
 		hostName:         hostName,
@@ -57,7 +56,7 @@ func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bo
 		flowWalker:       newConntrackFlowWalker(useConntrack),
 		natMapper:        makeNATMapper(newConntrackFlowWalker(useConntrack, "--any-nat")),
 		reverseResolver:  newReverseResolver(),
-		procWalker:       procWalker,
+		scanner:          procspy.NewConnectionScanner(procWalker),
 	}
 }
 
@@ -69,6 +68,7 @@ func (r *Reporter) Stop() {
 	r.flowWalker.stop()
 	r.natMapper.stop()
 	r.reverseResolver.stop()
+	r.scanner.Stop()
 }
 
 // Report implements Reporter.
@@ -81,7 +81,7 @@ func (r *Reporter) Report() (report.Report, error) {
 	rpt := report.MakeReport()
 
 	{
-		conns, err := procspy.Connections(r.includeProcesses, r.procWalker)
+		conns, err := r.scanner.Connections(r.includeProcesses)
 		if err != nil {
 			return rpt, err
 		}

From b93c3232cd1307ba58d9ff23d37c6211b9c293fb Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Mon, 8 Feb 2016 10:48:55 +0000
Subject: [PATCH 05/17] Make linter happy

---
 probe/endpoint/procspy/spy.go        | 1 +
 probe/endpoint/procspy/spy_darwin.go | 1 +
 probe/endpoint/procspy/spy_linux.go  | 1 +
 3 files changed, 3 insertions(+)

diff --git a/probe/endpoint/procspy/spy.go b/probe/endpoint/procspy/spy.go
index 78312fe64..73d4a560d 100644
--- a/probe/endpoint/procspy/spy.go
+++ b/probe/endpoint/procspy/spy.go
@@ -33,6 +33,7 @@ type ConnIter interface {
 	Next() *Connection
 }
 
+// ConnectionScanner scans the system for established (TCP) connections
 type ConnectionScanner interface {
 	// Connections returns all established (TCP) connections.  If processes is
 	// false we'll just list all TCP connections, and there is no need to be root.
diff --git a/probe/endpoint/procspy/spy_darwin.go b/probe/endpoint/procspy/spy_darwin.go
index d14400bfe..bec6f2a13 100644
--- a/probe/endpoint/procspy/spy_darwin.go
+++ b/probe/endpoint/procspy/spy_darwin.go
@@ -13,6 +13,7 @@ const (
 	lsofBinary    = "lsof"
 )
 
+// NewConnectionScanner creates a new Darwin ConnectionScanner
 func NewConnectionScanner(_ process.Walker) ConnectionScanner {
 	return &darwinScanner{}
 }
diff --git a/probe/endpoint/procspy/spy_linux.go b/probe/endpoint/procspy/spy_linux.go
index 65b5829a2..73ecddf33 100644
--- a/probe/endpoint/procspy/spy_linux.go
+++ b/probe/endpoint/procspy/spy_linux.go
@@ -32,6 +32,7 @@ func (c *pnConnIter) Next() *Connection {
 	return n
 }
 
+// NewConnectionScanner creates a new Linux ConnectionScanner
 func NewConnectionScanner(walker process.Walker) ConnectionScanner {
 	br := newBackgroundReader(walker)
 	br.start()

From 6240187333f83bcef7e772ae644c6abbb066560b Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Mon, 8 Feb 2016 11:50:57 +0000
Subject: [PATCH 06/17] Fix tests

---
 .../procspy/background_reader_linux.go        | 40 ++++++++++---------
 .../procspy/benchmark_internal_test.go        | 36 -----------------
 probe/endpoint/procspy/fixture.go             |  8 ++--
 probe/endpoint/procspy/proc_internal_test.go  |  7 +++-
 probe/endpoint/procspy/proc_linux.go          |  6 +--
 .../procspy/spy_linux_internal_test.go        |  9 ++++-
 probe/endpoint/reporter.go                    |  4 +-
 probe/endpoint/reporter_test.go               | 10 ++---
 prog/probe.go                                 |  3 +-
 9 files changed, 50 insertions(+), 73 deletions(-)
 delete mode 100644 probe/endpoint/procspy/benchmark_internal_test.go

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index 51eb6b7e7..f565662dd 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -67,14 +67,32 @@ func (br *backgroundReader) loop() {
 	)
 
 	rateLimit := initialRateLimit
-	ticker := time.Tick(rateLimit)
+	ticker := time.NewTicker(rateLimit)
 	for {
 		start := time.Now()
-		sockets, err := walkProcPid(br.walkingBuf, br.walker, ticker, fdBlockSize)
+		sockets, err := walkProcPid(br.walkingBuf, br.walker, ticker.C, fdBlockSize)
 		if err != nil {
 			log.Printf("background reader: error walking /proc: %s\n", err)
 			continue
 		}
+
+		br.mtx.Lock()
+
+		// Should we stop?
+		if br.pleaseStop {
+			br.pleaseStop = false
+			br.running = false
+			ticker.Stop()
+			br.mtx.Unlock()
+			return
+		}
+
+		// Swap buffers
+		br.readyBuf, br.walkingBuf = br.walkingBuf, br.readyBuf
+		br.readySockets = sockets
+
+		br.mtx.Unlock()
+
 		walkTime := time.Now().Sub(start)
 		walkTimeF := float64(walkTime)
 
@@ -92,22 +110,8 @@ func (br *backgroundReader) loop() {
 		rateLimit = time.Duration(math.Min(scaledRateLimit, maxRateLimitF))
 		log.Printf("debug: background reader: new rate limit %s\n", rateLimit)
 
-		ticker = time.Tick(rateLimit)
-
-		br.mtx.Lock()
-
-		// Should we stop?
-		if br.pleaseStop {
-			br.pleaseStop = false
-			br.running = false
-			br.mtx.Unlock()
-			return
-		}
-
-		// Swap buffers
-		br.readyBuf, br.walkingBuf = br.walkingBuf, br.readyBuf
-		br.readySockets = sockets
-		br.mtx.Unlock()
+		ticker.Stop()
+		ticker = time.NewTicker(rateLimit)
 
 		br.walkingBuf.Reset()
 
diff --git a/probe/endpoint/procspy/benchmark_internal_test.go b/probe/endpoint/procspy/benchmark_internal_test.go
deleted file mode 100644
index 1e48aab06..000000000
--- a/probe/endpoint/procspy/benchmark_internal_test.go
+++ /dev/null
@@ -1,36 +0,0 @@
-package procspy
-
-import (
-	"bytes"
-	"testing"
-)
-
-func BenchmarkParseConnectionsBaseline(b *testing.B) {
-	readFile = func(string, *bytes.Buffer) (int64, error) { return 0, nil }
-	benchmarkConnections(b)
-	// 333 ns/op, 0 allocs/op
-}
-
-func BenchmarkParseConnectionsFixture(b *testing.B) {
-	readFile = func(_ string, buf *bytes.Buffer) (int64, error) {
-		n, err := buf.Write(fixture)
-		return int64(n), err
-	}
-	benchmarkConnections(b)
-	// 15553 ns/op, 12 allocs/op
-}
-
-func benchmarkConnections(b *testing.B) {
-	b.ReportAllocs()
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		cbConnections(false, nil)
-	}
-}
-
-var fixture = []byte(`  sl  local_address rem_address   st tx_queue rx_queue tr tm->when retrnsmt   uid  timeout inode
-   0: 00000000:A6C0 00000000:0000 01 00000000:00000000 00:00000000 00000000   105        0 5107 1 ffff8800a6aaf040 100 0 0 10 0
-   1: 00000000:006F 00000000:0000 01 00000000:00000000 00:00000000 00000000     0        0 5084 1 ffff8800a6aaf740 100 0 0 10 0
-   2: 0100007F:0019 00000000:0000 01 00000000:00000000 00:00000000 00000000     0        0 10550 1 ffff8800a729b780 100 0 0 10 0
-   3: A12CF62E:E4D7 57FC1EC0:01BB 01 00000000:00000000 02:000006FA 00000000  1000        0 639474 2 ffff88007e75a740 48 4 26 10 -1
-`)
diff --git a/probe/endpoint/procspy/fixture.go b/probe/endpoint/procspy/fixture.go
index b8d0ba02d..baf778fe4 100644
--- a/probe/endpoint/procspy/fixture.go
+++ b/probe/endpoint/procspy/fixture.go
@@ -13,13 +13,13 @@ func (f *fixedConnIter) Next() *Connection {
 	return &car
 }
 
-// fixedScanner implements ConnectionScanner and uses constant Connection and
+// FixedScanner implements ConnectionScanner and uses constant Connection and
 // ConnectionProcs.  It's designed to be used in tests.
-type fixedScanner []Connection
+type FixedScanner []Connection
 
-func (s fixedScanner) Connections() (ConnIter, error) {
+func (s FixedScanner) Connections(_ bool) (ConnIter, error) {
 	iter := fixedConnIter(s)
 	return &iter, nil
 }
 
-func (s fixedScanner) Stop() {}
+func (s FixedScanner) Stop() {}
diff --git a/probe/endpoint/procspy/proc_internal_test.go b/probe/endpoint/procspy/proc_internal_test.go
index 1035e485a..1e3bc7466 100644
--- a/probe/endpoint/procspy/proc_internal_test.go
+++ b/probe/endpoint/procspy/proc_internal_test.go
@@ -5,6 +5,7 @@ import (
 	"reflect"
 	"syscall"
 	"testing"
+	"time"
 
 	fs_hook "github.com/weaveworks/scope/common/fs"
 	"github.com/weaveworks/scope/probe/process"
@@ -57,7 +58,11 @@ func TestWalkProcPid(t *testing.T) {
 	defer fs_hook.Restore()
 
 	buf := bytes.Buffer{}
-	have, err := walkProcPid(&buf, process.NewWalker(procRoot))
+	walker := process.NewWalker(procRoot)
+	ticker := time.NewTicker(time.Millisecond)
+	defer ticker.Stop()
+	fdBlockSize := 1
+	have, err := walkProcPid(&buf, walker, ticker.C, fdBlockSize)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/probe/endpoint/procspy/proc_linux.go b/probe/endpoint/procspy/proc_linux.go
index 2af9b999d..0329fb847 100644
--- a/probe/endpoint/procspy/proc_linux.go
+++ b/probe/endpoint/procspy/proc_linux.go
@@ -210,10 +210,8 @@ func walkProcPid(buf *bytes.Buffer, walker process.Walker, ticker <-chan time.Ti
 	return sockets, nil
 }
 
-// readFile reads an arbitrary file into a buffer. It's a variable so it can
-// be overwritten for benchmarks. That's bad practice and we should change it
-// to be a dependency.
-var readFile = func(filename string, buf *bytes.Buffer) (int64, error) {
+// readFile reads an arbitrary file into a buffer.
+func readFile(filename string, buf *bytes.Buffer) (int64, error) {
 	f, err := fs.Open(filename)
 	if err != nil {
 		return -1, err
diff --git a/probe/endpoint/procspy/spy_linux_internal_test.go b/probe/endpoint/procspy/spy_linux_internal_test.go
index fc5ed3890..bdbf11c8e 100644
--- a/probe/endpoint/procspy/spy_linux_internal_test.go
+++ b/probe/endpoint/procspy/spy_linux_internal_test.go
@@ -4,6 +4,7 @@ import (
 	"net"
 	"reflect"
 	"testing"
+	"time"
 
 	fs_hook "github.com/weaveworks/scope/common/fs"
 	"github.com/weaveworks/scope/probe/process"
@@ -13,8 +14,13 @@ import (
 func TestLinuxConnections(t *testing.T) {
 	fs_hook.Mock(mockFS)
 	defer fs_hook.Restore()
+	scanner := NewConnectionScanner(process.NewWalker("/proc"))
+	defer scanner.Stop()
 
-	iter, err := cbConnections(true, process.NewWalker("/proc"))
+	// let the background scanner finish its first pass
+	time.Sleep(1 * time.Second)
+
+	iter, err := scanner.Connections(true)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -37,4 +43,5 @@ func TestLinuxConnections(t *testing.T) {
 	if have := iter.Next(); have != nil {
 		t.Fatal(have)
 	}
+
 }
diff --git a/probe/endpoint/reporter.go b/probe/endpoint/reporter.go
index b88b0c405..5999312c6 100644
--- a/probe/endpoint/reporter.go
+++ b/probe/endpoint/reporter.go
@@ -48,7 +48,7 @@ var SpyDuration = prometheus.NewSummaryVec(
 // on the host machine, at the granularity of host and port. That information
 // is stored in the Endpoint topology. It optionally enriches that topology
 // with process (PID) information.
-func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bool, procWalker process.Walker) *Reporter {
+func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bool, scanner procspy.ConnectionScanner) *Reporter {
 	return &Reporter{
 		hostID:           hostID,
 		hostName:         hostName,
@@ -56,7 +56,7 @@ func NewReporter(hostID, hostName string, includeProcesses bool, useConntrack bo
 		flowWalker:       newConntrackFlowWalker(useConntrack),
 		natMapper:        makeNATMapper(newConntrackFlowWalker(useConntrack, "--any-nat")),
 		reverseResolver:  newReverseResolver(),
-		scanner:          procspy.NewConnectionScanner(procWalker),
+		scanner:          scanner,
 	}
 }
 
diff --git a/probe/endpoint/reporter_test.go b/probe/endpoint/reporter_test.go
index 4a03636e7..ba3c2e738 100644
--- a/probe/endpoint/reporter_test.go
+++ b/probe/endpoint/reporter_test.go
@@ -64,14 +64,13 @@ var (
 )
 
 func TestSpyNoProcesses(t *testing.T) {
-	procspy.SetFixtures(fixConnections)
-
 	const (
 		nodeID   = "heinz-tomato-ketchup" // TODO rename to hostID
 		nodeName = "frenchs-since-1904"   // TODO rename to hostNmae
 	)
 
-	reporter := endpoint.NewReporter(nodeID, nodeName, false, false, nil)
+	scanner := procspy.FixedScanner(fixConnections)
+	reporter := endpoint.NewReporter(nodeID, nodeName, false, false, scanner)
 	r, _ := reporter.Report()
 	//buf, _ := json.MarshalIndent(r, "", "    ")
 	//t.Logf("\n%s\n", buf)
@@ -101,14 +100,13 @@ func TestSpyNoProcesses(t *testing.T) {
 }
 
 func TestSpyWithProcesses(t *testing.T) {
-	procspy.SetFixtures(fixConnectionsWithProcesses)
-
 	const (
 		nodeID   = "nikon"             // TODO rename to hostID
 		nodeName = "fishermans-friend" // TODO rename to hostNmae
 	)
 
-	reporter := endpoint.NewReporter(nodeID, nodeName, true, false, nil)
+	scanner := procspy.FixedScanner(fixConnectionsWithProcesses)
+	reporter := endpoint.NewReporter(nodeID, nodeName, true, false, scanner)
 	r, _ := reporter.Report()
 	// buf, _ := json.MarshalIndent(r, "", "    ") ; t.Logf("\n%s\n", buf)
 
diff --git a/prog/probe.go b/prog/probe.go
index 51081a6aa..a7a3d6866 100644
--- a/prog/probe.go
+++ b/prog/probe.go
@@ -139,8 +139,9 @@ func probeMain() {
 	defer resolver.Stop()
 
 	processCache := process.NewCachingWalker(process.NewWalker(*procRoot))
+	scanner := procspy.NewConnectionScanner(processCache)
 
-	endpointReporter := endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack, processCache)
+	endpointReporter := endpoint.NewReporter(hostID, hostName, *spyProcs, *useConntrack, scanner)
 	defer endpointReporter.Stop()
 
 	p := probe.New(*spyInterval, *publishInterval, clients)

From 8c3c8994b14864b662f6e082ff5f6497237a5641 Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Mon, 8 Feb 2016 12:00:10 +0000
Subject: [PATCH 07/17] Use levelled logging

---
 .../endpoint/procspy/background_reader_linux.go | 17 +++++++++--------
 prog/probe.go                                   |  1 +
 2 files changed, 10 insertions(+), 8 deletions(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index f565662dd..f21218551 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -3,11 +3,12 @@ package procspy
 import (
 	"bytes"
 	"fmt"
-	"log"
 	"math"
 	"sync"
 	"time"
 
+	log "github.com/Sirupsen/logrus"
+
 	"github.com/weaveworks/scope/probe/process"
 )
 
@@ -28,8 +29,6 @@ type backgroundReader struct {
 	readySockets map[uint64]*Proc
 }
 
-// starts a rate-limited background goroutine to read the expensive files from
-// proc.
 func newBackgroundReader(walker process.Walker) *backgroundReader {
 	br := &backgroundReader{
 		walker:     walker,
@@ -39,6 +38,8 @@ func newBackgroundReader(walker process.Walker) *backgroundReader {
 	return br
 }
 
+// starts a rate-limited background goroutine to read the expensive files from
+// proc.
 func (br *backgroundReader) start() error {
 	br.mtx.Lock()
 	defer br.mtx.Unlock()
@@ -72,7 +73,7 @@ func (br *backgroundReader) loop() {
 		start := time.Now()
 		sockets, err := walkProcPid(br.walkingBuf, br.walker, ticker.C, fdBlockSize)
 		if err != nil {
-			log.Printf("background reader: error walking /proc: %s\n", err)
+			log.Errorf("background /proc reader: error walking /proc: %s", err)
 			continue
 		}
 
@@ -96,10 +97,10 @@ func (br *backgroundReader) loop() {
 		walkTime := time.Now().Sub(start)
 		walkTimeF := float64(walkTime)
 
-		log.Printf("debug: background reader: full pass took %s\n", walkTime)
+		log.Debugf("background /proc reader: full pass took %s", walkTime)
 		if walkTimeF/targetWalkTimeF > 1.5 {
-			log.Printf(
-				"warn: background reader: full pass took %s: 50%% more than expected (%s)\n",
+			log.Warnf(
+				"background /proc reader: full pass took %s: 50%% more than expected (%s)",
 				walkTime,
 				targetWalkTime,
 			)
@@ -108,7 +109,7 @@ func (br *backgroundReader) loop() {
 		// Adjust rate limit to more-accurately meet the target walk time in next iteration
 		scaledRateLimit := targetWalkTimeF / walkTimeF * float64(rateLimit)
 		rateLimit = time.Duration(math.Min(scaledRateLimit, maxRateLimitF))
-		log.Printf("debug: background reader: new rate limit %s\n", rateLimit)
+		log.Debugf("background /proc reader: new rate limit %s", rateLimit)
 
 		ticker.Stop()
 		ticker = time.NewTicker(rateLimit)
diff --git a/prog/probe.go b/prog/probe.go
index a7a3d6866..8a2054d26 100644
--- a/prog/probe.go
+++ b/prog/probe.go
@@ -24,6 +24,7 @@ import (
 	"github.com/weaveworks/scope/probe/controls"
 	"github.com/weaveworks/scope/probe/docker"
 	"github.com/weaveworks/scope/probe/endpoint"
+	"github.com/weaveworks/scope/probe/endpoint/procspy"
 	"github.com/weaveworks/scope/probe/host"
 	"github.com/weaveworks/scope/probe/kubernetes"
 	"github.com/weaveworks/scope/probe/overlay"

From c09ad9e4f69fc479fb4e70b42964e80c5877e584 Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Mon, 8 Feb 2016 12:27:20 +0000
Subject: [PATCH 08/17] Adjust file descript rate-limit block

---
 probe/endpoint/procspy/background_reader_linux.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index f21218551..2dd85285b 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -15,7 +15,7 @@ import (
 const (
 	initialRateLimit = 50 * time.Millisecond  // Read 20 * fdBlockSize file descriptors (/proc/PID/fd/*) per namespace per second
 	maxRateLimit     = 250 * time.Millisecond // Read at least 4 * fdBlockSize file descriptors per namespace per second
-	fdBlockSize      = 100
+	fdBlockSize      = 300
 	targetWalkTime   = 10 * time.Second // Aim at walking all files in 10 seconds
 )
 

From ccabaf5e6a6e0d3d20cadd947890f57debee4535 Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Mon, 8 Feb 2016 12:38:36 +0000
Subject: [PATCH 09/17] Use uint64 for fd counter

---
 probe/endpoint/procspy/background_reader_linux.go | 2 +-
 probe/endpoint/procspy/proc_linux.go              | 6 +++---
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index 2dd85285b..f51bf9f84 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -15,7 +15,7 @@ import (
 const (
 	initialRateLimit = 50 * time.Millisecond  // Read 20 * fdBlockSize file descriptors (/proc/PID/fd/*) per namespace per second
 	maxRateLimit     = 250 * time.Millisecond // Read at least 4 * fdBlockSize file descriptors per namespace per second
-	fdBlockSize      = 300
+	fdBlockSize      = uint64(300)
 	targetWalkTime   = 10 * time.Second // Aim at walking all files in 10 seconds
 )
 
diff --git a/probe/endpoint/procspy/proc_linux.go b/probe/endpoint/procspy/proc_linux.go
index 0329fb847..dec3055e6 100644
--- a/probe/endpoint/procspy/proc_linux.go
+++ b/probe/endpoint/procspy/proc_linux.go
@@ -106,14 +106,14 @@ func readProcessConnections(buf *bytes.Buffer, namespaceProcs []*process.Process
 }
 
 // walkNamespacePid does the work of walkProcPid for a single namespace
-func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProcs []*process.Process, ticker <-chan time.Time, fdBlockSize int) error {
+func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProcs []*process.Process, ticker <-chan time.Time, fdBlockSize uint64) error {
 
 	if found, err := readProcessConnections(buf, namespaceProcs); err != nil || !found {
 		return err
 	}
 
 	var statT syscall.Stat_t
-	var fdBlockCount int
+	var fdBlockCount uint64
 	for i, p := range namespaceProcs {
 
 		// Get the sockets for all the processes in the namespace
@@ -174,7 +174,7 @@ func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProc
 // /proc/PID/net/tcp{,6} for each namespace and sees if the ./fd/* files of each
 // process in that namespace are symlinks to sockets. Returns a map from socket
 // ID (inode) to PID.
-func walkProcPid(buf *bytes.Buffer, walker process.Walker, ticker <-chan time.Time, fdBlockSize int) (map[uint64]*Proc, error) {
+func walkProcPid(buf *bytes.Buffer, walker process.Walker, ticker <-chan time.Time, fdBlockSize uint64) (map[uint64]*Proc, error) {
 	var (
 		sockets    = map[uint64]*Proc{}              // map socket inode -> process
 		namespaces = map[uint64][]*process.Process{} // map network namespace id -> processes

From 08969ec15436ed833d00cb8f7f5247841bb070ed Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Mon, 8 Feb 2016 12:45:48 +0000
Subject: [PATCH 10/17] Clarify rate-limiting further

---
 .../procspy/background_reader_linux.go        | 25 ++++++++++---------
 probe/endpoint/procspy/proc_internal_test.go  |  2 +-
 2 files changed, 14 insertions(+), 13 deletions(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index f51bf9f84..96668c65f 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -13,10 +13,11 @@ import (
 )
 
 const (
-	initialRateLimit = 50 * time.Millisecond  // Read 20 * fdBlockSize file descriptors (/proc/PID/fd/*) per namespace per second
-	maxRateLimit     = 250 * time.Millisecond // Read at least 4 * fdBlockSize file descriptors per namespace per second
-	fdBlockSize      = uint64(300)
-	targetWalkTime   = 10 * time.Second // Aim at walking all files in 10 seconds
+	initialRateLimitPeriod = 50 * time.Millisecond  // Read 20 * fdBlockSize file descriptors (/proc/PID/fd/*) per namespace per second
+	maxRateLimitPeriod     = 250 * time.Millisecond // Read at least 4 * fdBlockSize file descriptors per namespace per second
+	fdBlockSize            = uint64(300)            // Maximum number of /proc/PID/fd/* files to stat per rate-limit period
+	// (as a rule of thumb going through each block should be more expensive than reading /proc/PID/tcp{,6})
+	targetWalkTime = 10 * time.Second // Aim at walking all files in 10 seconds
 )
 
 type backgroundReader struct {
@@ -63,12 +64,12 @@ func (br *backgroundReader) stop() error {
 
 func (br *backgroundReader) loop() {
 	const (
-		maxRateLimitF   = float64(maxRateLimit)
-		targetWalkTimeF = float64(targetWalkTime)
+		maxRateLimitPeriodF = float64(maxRateLimitPeriod)
+		targetWalkTimeF     = float64(targetWalkTime)
 	)
 
-	rateLimit := initialRateLimit
-	ticker := time.NewTicker(rateLimit)
+	rateLimitPeriod := initialRateLimitPeriod
+	ticker := time.NewTicker(rateLimitPeriod)
 	for {
 		start := time.Now()
 		sockets, err := walkProcPid(br.walkingBuf, br.walker, ticker.C, fdBlockSize)
@@ -107,12 +108,12 @@ func (br *backgroundReader) loop() {
 		}
 
 		// Adjust rate limit to more-accurately meet the target walk time in next iteration
-		scaledRateLimit := targetWalkTimeF / walkTimeF * float64(rateLimit)
-		rateLimit = time.Duration(math.Min(scaledRateLimit, maxRateLimitF))
-		log.Debugf("background /proc reader: new rate limit %s", rateLimit)
+		scaledRateLimitPeriod := targetWalkTimeF / walkTimeF * float64(rateLimitPeriod)
+		rateLimitPeriod = time.Duration(math.Min(scaledRateLimitPeriod, maxRateLimitPeriodF))
+		log.Debugf("background /proc reader: new rate limit %s", rateLimitPeriod)
 
 		ticker.Stop()
-		ticker = time.NewTicker(rateLimit)
+		ticker = time.NewTicker(rateLimitPeriod)
 
 		br.walkingBuf.Reset()
 
diff --git a/probe/endpoint/procspy/proc_internal_test.go b/probe/endpoint/procspy/proc_internal_test.go
index 1e3bc7466..362068c41 100644
--- a/probe/endpoint/procspy/proc_internal_test.go
+++ b/probe/endpoint/procspy/proc_internal_test.go
@@ -61,7 +61,7 @@ func TestWalkProcPid(t *testing.T) {
 	walker := process.NewWalker(procRoot)
 	ticker := time.NewTicker(time.Millisecond)
 	defer ticker.Stop()
-	fdBlockSize := 1
+	fdBlockSize := uint64(1)
 	have, err := walkProcPid(&buf, walker, ticker.C, fdBlockSize)
 	if err != nil {
 		t.Fatal(err)

From b08c4276182f4ac96d0797848554b60e9c85ceef Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Mon, 8 Feb 2016 12:48:37 +0000
Subject: [PATCH 11/17] Make linter happy

---
 probe/endpoint/procspy/fixture.go | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/probe/endpoint/procspy/fixture.go b/probe/endpoint/procspy/fixture.go
index baf778fe4..06020dbdb 100644
--- a/probe/endpoint/procspy/fixture.go
+++ b/probe/endpoint/procspy/fixture.go
@@ -17,9 +17,11 @@ func (f *fixedConnIter) Next() *Connection {
 // ConnectionProcs.  It's designed to be used in tests.
 type FixedScanner []Connection
 
+// Connections implements ConnectionsScanner.Connections
 func (s FixedScanner) Connections(_ bool) (ConnIter, error) {
 	iter := fixedConnIter(s)
 	return &iter, nil
 }
 
+// Stop implements ConnectionsScanner.Stop (dummy since there is no background work)
 func (s FixedScanner) Stop() {}

From 3dd2d45fe58040a2e0abd41466af9a7bc7b88751 Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Mon, 8 Feb 2016 13:41:41 +0000
Subject: [PATCH 12/17] Review feedback

---
 probe/endpoint/procspy/background_reader_linux.go | 8 ++++----
 probe/endpoint/procspy/spy_linux.go               | 5 ++++-
 2 files changed, 8 insertions(+), 5 deletions(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index 96668c65f..7b28f78a9 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -3,6 +3,7 @@ package procspy
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"math"
 	"sync"
 	"time"
@@ -122,12 +123,11 @@ func (br *backgroundReader) loop() {
 	}
 }
 
-func (br *backgroundReader) getWalkedProcPid(buf *bytes.Buffer) map[uint64]*Proc {
+func (br *backgroundReader) getWalkedProcPid(buf *bytes.Buffer) (map[uint64]*Proc, error) {
 	br.mtx.Lock()
 	defer br.mtx.Unlock()
 
-	reader := bytes.NewReader(br.readyBuf.Bytes())
-	buf.ReadFrom(reader)
+	_, err := io.Copy(buf, br.readyBuf)
 
-	return br.readySockets
+	return br.readySockets, err
 }
diff --git a/probe/endpoint/procspy/spy_linux.go b/probe/endpoint/procspy/spy_linux.go
index 73ecddf33..9cd1e327e 100644
--- a/probe/endpoint/procspy/spy_linux.go
+++ b/probe/endpoint/procspy/spy_linux.go
@@ -50,7 +50,10 @@ func (s *linuxScanner) Connections(processes bool) (ConnIter, error) {
 
 	var procs map[uint64]*Proc
 	if processes {
-		procs = s.br.getWalkedProcPid(buf)
+		var err error
+		if procs, err = s.br.getWalkedProcPid(buf); err != nil {
+			return nil, err
+		}
 	}
 
 	if buf.Len() == 0 {

From 53bc710c4e912b0f27da55e11c0749024f624dcf Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Mon, 8 Feb 2016 18:01:37 +0000
Subject: [PATCH 13/17] Review feedback

---
 .../procspy/background_reader_linux.go        | 186 +++++++++---------
 probe/endpoint/procspy/fixture.go             |   2 +-
 probe/endpoint/procspy/proc_internal_test.go  |   4 +-
 probe/endpoint/procspy/proc_linux.go          |  44 ++++-
 probe/endpoint/procspy/spy_linux.go           |   1 -
 5 files changed, 129 insertions(+), 108 deletions(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index 7b28f78a9..a7225b3b6 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -2,7 +2,6 @@ package procspy
 
 import (
 	"bytes"
-	"fmt"
 	"io"
 	"math"
 	"sync"
@@ -22,112 +21,109 @@ const (
 )
 
 type backgroundReader struct {
-	walker       process.Walker
-	mtx          sync.Mutex
-	running      bool
-	pleaseStop   bool
-	walkingBuf   *bytes.Buffer
-	readyBuf     *bytes.Buffer
-	readySockets map[uint64]*Proc
-}
-
-func newBackgroundReader(walker process.Walker) *backgroundReader {
-	br := &backgroundReader{
-		walker:     walker,
-		walkingBuf: bytes.NewBuffer(make([]byte, 0, 5000)),
-		readyBuf:   bytes.NewBuffer(make([]byte, 0, 5000)),
-	}
-	return br
+	stopc         chan struct{}
+	mtx           sync.Mutex
+	latestBuf     *bytes.Buffer
+	latestSockets map[uint64]*Proc
 }
 
 // starts a rate-limited background goroutine to read the expensive files from
 // proc.
-func (br *backgroundReader) start() error {
-	br.mtx.Lock()
-	defer br.mtx.Unlock()
-	if br.running {
-		return fmt.Errorf("background reader already running")
+func newBackgroundReader(walker process.Walker) *backgroundReader {
+	br := &backgroundReader{
+		latestBuf: bytes.NewBuffer(make([]byte, 0, 5000)),
+		stopc:     make(chan struct{}),
 	}
-	br.running = true
-	go br.loop()
-	return nil
+	go br.loop(walker)
+	return br
 }
 
-func (br *backgroundReader) stop() error {
-	br.mtx.Lock()
-	defer br.mtx.Unlock()
-	if !br.running {
-		return fmt.Errorf("background reader already not running")
-	}
-	br.pleaseStop = true
-	return nil
-}
-
-func (br *backgroundReader) loop() {
-	const (
-		maxRateLimitPeriodF = float64(maxRateLimitPeriod)
-		targetWalkTimeF     = float64(targetWalkTime)
-	)
-
-	rateLimitPeriod := initialRateLimitPeriod
-	ticker := time.NewTicker(rateLimitPeriod)
-	for {
-		start := time.Now()
-		sockets, err := walkProcPid(br.walkingBuf, br.walker, ticker.C, fdBlockSize)
-		if err != nil {
-			log.Errorf("background /proc reader: error walking /proc: %s", err)
-			continue
-		}
-
-		br.mtx.Lock()
-
-		// Should we stop?
-		if br.pleaseStop {
-			br.pleaseStop = false
-			br.running = false
-			ticker.Stop()
-			br.mtx.Unlock()
-			return
-		}
-
-		// Swap buffers
-		br.readyBuf, br.walkingBuf = br.walkingBuf, br.readyBuf
-		br.readySockets = sockets
-
-		br.mtx.Unlock()
-
-		walkTime := time.Now().Sub(start)
-		walkTimeF := float64(walkTime)
-
-		log.Debugf("background /proc reader: full pass took %s", walkTime)
-		if walkTimeF/targetWalkTimeF > 1.5 {
-			log.Warnf(
-				"background /proc reader: full pass took %s: 50%% more than expected (%s)",
-				walkTime,
-				targetWalkTime,
-			)
-		}
-
-		// Adjust rate limit to more-accurately meet the target walk time in next iteration
-		scaledRateLimitPeriod := targetWalkTimeF / walkTimeF * float64(rateLimitPeriod)
-		rateLimitPeriod = time.Duration(math.Min(scaledRateLimitPeriod, maxRateLimitPeriodF))
-		log.Debugf("background /proc reader: new rate limit %s", rateLimitPeriod)
-
-		ticker.Stop()
-		ticker = time.NewTicker(rateLimitPeriod)
-
-		br.walkingBuf.Reset()
-
-		// Sleep during spare time
-		time.Sleep(targetWalkTime - walkTime)
-	}
+func (br *backgroundReader) stop() {
+	close(br.stopc)
 }
 
 func (br *backgroundReader) getWalkedProcPid(buf *bytes.Buffer) (map[uint64]*Proc, error) {
 	br.mtx.Lock()
 	defer br.mtx.Unlock()
 
-	_, err := io.Copy(buf, br.readyBuf)
+	_, err := io.Copy(buf, br.latestBuf)
 
-	return br.readySockets, err
+	return br.latestSockets, err
+}
+
+func (br *backgroundReader) loop(walker process.Walker) {
+	var (
+		begin           time.Time                      // when we started the last performWalk
+		tickc           = time.After(time.Millisecond) // fire immediately
+		walkc           chan map[uint64]*Proc          // initially nil, i.e. off
+		walkBuf         = bytes.NewBuffer(make([]byte, 0, 5000))
+		rateLimitPeriod = initialRateLimitPeriod
+		ticker          = time.NewTicker(rateLimitPeriod)
+		pWalker         = newPidWalker(walker, ticker.C, fdBlockSize)
+	)
+
+	for {
+		select {
+		case <-tickc:
+			tickc = nil                             // turn off until the next loop
+			walkc = make(chan map[uint64]*Proc, 1)  // turn on (need buffered so we don't leak performWalk)
+			begin = time.Now()                      // reset counter
+			go performWalk(pWalker, walkBuf, walkc) // do work
+
+		case sockets := <-walkc:
+			// Swap buffers
+			br.mtx.Lock()
+			br.latestBuf, walkBuf = walkBuf, br.latestBuf
+			br.latestSockets = sockets
+			br.mtx.Unlock()
+			walkBuf.Reset()
+
+			// Schedule next walk and adjust rate limit
+			walkTime := time.Since(begin)
+			rateLimitPeriod, nextInterval := scheduleNextWalk(rateLimitPeriod, walkTime)
+			ticker.Stop()
+			ticker := time.NewTicker(rateLimitPeriod)
+			pWalker.ticker = ticker.C
+
+			walkc = nil                      // turn off until the next loop
+			tickc = time.After(nextInterval) // turn on
+
+		case <-br.stopc:
+			pWalker.stop()
+			ticker.Stop()
+			return // abort
+		}
+	}
+}
+
+// Adjust rate limit for next walk and calculate how long to wait until it should be started
+func scheduleNextWalk(rateLimitPeriod time.Duration, took time.Duration) (time.Duration, time.Duration) {
+
+	log.Debugf("background /proc reader: full pass took %s", took)
+	if float64(took)/float64(targetWalkTime) > 1.5 {
+		log.Warnf(
+			"background /proc reader: full pass took %s: 50%% more than expected (%s)",
+			took,
+			targetWalkTime,
+		)
+	}
+
+	// Adjust rate limit to more-accurately meet the target walk time in next iteration
+	scaledRateLimitPeriod := float64(targetWalkTime) / float64(took) * float64(rateLimitPeriod)
+	rateLimitPeriod = time.Duration(math.Min(scaledRateLimitPeriod, float64(maxRateLimitPeriod)))
+
+	log.Debugf("background /proc reader: new rate limit %s", rateLimitPeriod)
+
+	return rateLimitPeriod, targetWalkTime - took
+}
+
+func performWalk(w pidWalker, buf *bytes.Buffer, c chan<- map[uint64]*Proc) {
+	sockets, err := w.walk(buf)
+	if err != nil {
+		log.Errorf("background /proc reader: error walking /proc: %s", err)
+		buf.Reset()
+		c <- nil
+		return
+	}
+	c <- sockets
 }
diff --git a/probe/endpoint/procspy/fixture.go b/probe/endpoint/procspy/fixture.go
index 06020dbdb..0731f624a 100644
--- a/probe/endpoint/procspy/fixture.go
+++ b/probe/endpoint/procspy/fixture.go
@@ -14,7 +14,7 @@ func (f *fixedConnIter) Next() *Connection {
 }
 
 // FixedScanner implements ConnectionScanner and uses constant Connection and
-// ConnectionProcs.  It's designed to be used in tests.
+// ConnectionProcs.
 type FixedScanner []Connection
 
 // Connections implements ConnectionsScanner.Connections
diff --git a/probe/endpoint/procspy/proc_internal_test.go b/probe/endpoint/procspy/proc_internal_test.go
index 362068c41..54cfa20f4 100644
--- a/probe/endpoint/procspy/proc_internal_test.go
+++ b/probe/endpoint/procspy/proc_internal_test.go
@@ -61,8 +61,8 @@ func TestWalkProcPid(t *testing.T) {
 	walker := process.NewWalker(procRoot)
 	ticker := time.NewTicker(time.Millisecond)
 	defer ticker.Stop()
-	fdBlockSize := uint64(1)
-	have, err := walkProcPid(&buf, walker, ticker.C, fdBlockSize)
+	pWalker := newPidWalker(walker, ticker.C, 1)
+	have, err := pWalker.walk(&buf)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/probe/endpoint/procspy/proc_linux.go b/probe/endpoint/procspy/proc_linux.go
index dec3055e6..e475aee18 100644
--- a/probe/endpoint/procspy/proc_linux.go
+++ b/probe/endpoint/procspy/proc_linux.go
@@ -24,6 +24,23 @@ var (
 	netNamespacePathSuffix = ""
 )
 
+type pidWalker struct {
+	walker      process.Walker
+	ticker      <-chan time.Time
+	stopc       chan struct{}
+	fdBlockSize uint64
+}
+
+func newPidWalker(walker process.Walker, ticker <-chan time.Time, fdBlockSize uint64) pidWalker {
+	w := pidWalker{
+		walker:      walker,
+		ticker:      ticker,
+		fdBlockSize: fdBlockSize,
+		stopc:       make(chan struct{}),
+	}
+	return w
+}
+
 // SetProcRoot sets the location of the proc filesystem.
 func SetProcRoot(root string) {
 	procRoot = root
@@ -105,8 +122,8 @@ func readProcessConnections(buf *bytes.Buffer, namespaceProcs []*process.Process
 
 }
 
-// walkNamespacePid does the work of walkProcPid for a single namespace
-func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProcs []*process.Process, ticker <-chan time.Time, fdBlockSize uint64) error {
+// walkNamespace does the work of walk for a single namespace
+func (w pidWalker) walkNamespace(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProcs []*process.Process) error {
 
 	if found, err := readProcessConnections(buf, namespaceProcs); err != nil || !found {
 		return err
@@ -120,9 +137,10 @@ func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProc
 		dirName := strconv.Itoa(p.PID)
 		fdBase := filepath.Join(procRoot, dirName, "fd")
 
-		if fdBlockCount > fdBlockSize {
+		if fdBlockCount > w.fdBlockSize {
 			// we surpassed the filedescriptor rate limit
-			<-ticker
+			// TODO: worth selecting on w.stopc?
+			<-w.ticker
 			fdBlockCount = 0
 
 			// read the connections again to
@@ -170,11 +188,11 @@ func walkNamespacePid(buf *bytes.Buffer, sockets map[uint64]*Proc, namespaceProc
 	return nil
 }
 
-// walkProcPid walks over all numerical (PID) /proc entries. It reads
+// walk walks over all numerical (PID) /proc entries. It reads
 // /proc/PID/net/tcp{,6} for each namespace and sees if the ./fd/* files of each
 // process in that namespace are symlinks to sockets. Returns a map from socket
 // ID (inode) to PID.
-func walkProcPid(buf *bytes.Buffer, walker process.Walker, ticker <-chan time.Time, fdBlockSize uint64) (map[uint64]*Proc, error) {
+func (w pidWalker) walk(buf *bytes.Buffer) (map[uint64]*Proc, error) {
 	var (
 		sockets    = map[uint64]*Proc{}              // map socket inode -> process
 		namespaces = map[uint64][]*process.Process{} // map network namespace id -> processes
@@ -189,7 +207,7 @@ func walkProcPid(buf *bytes.Buffer, walker process.Walker, ticker <-chan time.Ti
 	// between reading /net/tcp{,6} of each namespace and /proc/PID/fd/* for
 	// the processes living in that namespace.
 
-	walker.Walk(func(p, _ process.Process) {
+	w.walker.Walk(func(p, _ process.Process) {
 		dirName := strconv.Itoa(p.PID)
 
 		netNamespacePath := filepath.Join(procRoot, dirName, getNetNamespacePathSuffix())
@@ -202,14 +220,22 @@ func walkProcPid(buf *bytes.Buffer, walker process.Walker, ticker <-chan time.Ti
 	})
 
 	for _, procs := range namespaces {
-		<-ticker
-		walkNamespacePid(buf, sockets, procs, ticker, fdBlockSize)
+		select {
+		case <-w.ticker:
+			w.walkNamespace(buf, sockets, procs)
+		case <-w.stopc:
+			break
+		}
 	}
 
 	metrics.SetGauge(namespaceKey, float32(len(namespaces)))
 	return sockets, nil
 }
 
+func (w pidWalker) stop() {
+	close(w.stopc)
+}
+
 // readFile reads an arbitrary file into a buffer.
 func readFile(filename string, buf *bytes.Buffer) (int64, error) {
 	f, err := fs.Open(filename)
diff --git a/probe/endpoint/procspy/spy_linux.go b/probe/endpoint/procspy/spy_linux.go
index 9cd1e327e..61f189881 100644
--- a/probe/endpoint/procspy/spy_linux.go
+++ b/probe/endpoint/procspy/spy_linux.go
@@ -35,7 +35,6 @@ func (c *pnConnIter) Next() *Connection {
 // NewConnectionScanner creates a new Linux ConnectionScanner
 func NewConnectionScanner(walker process.Walker) ConnectionScanner {
 	br := newBackgroundReader(walker)
-	br.start()
 	return &linuxScanner{br}
 }
 

From 0545d9a5c2b7d0b558af7025ebb8de0fd32666e9 Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Mon, 8 Feb 2016 21:03:21 +0000
Subject: [PATCH 14/17] Fix variable scope bug

---
 probe/endpoint/procspy/background_reader_linux.go | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index a7225b3b6..6d4405175 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -58,6 +58,7 @@ func (br *backgroundReader) loop(walker process.Walker) {
 		walkc           chan map[uint64]*Proc          // initially nil, i.e. off
 		walkBuf         = bytes.NewBuffer(make([]byte, 0, 5000))
 		rateLimitPeriod = initialRateLimitPeriod
+		nextInterval    time.Duration
 		ticker          = time.NewTicker(rateLimitPeriod)
 		pWalker         = newPidWalker(walker, ticker.C, fdBlockSize)
 	)
@@ -80,9 +81,9 @@ func (br *backgroundReader) loop(walker process.Walker) {
 
 			// Schedule next walk and adjust rate limit
 			walkTime := time.Since(begin)
-			rateLimitPeriod, nextInterval := scheduleNextWalk(rateLimitPeriod, walkTime)
+			rateLimitPeriod, nextInterval = scheduleNextWalk(rateLimitPeriod, walkTime)
 			ticker.Stop()
-			ticker := time.NewTicker(rateLimitPeriod)
+			ticker = time.NewTicker(rateLimitPeriod)
 			pWalker.ticker = ticker.C
 
 			walkc = nil                      // turn off until the next loop

From eb52adbbec1234366d6dc5ac1c729e0182176c05 Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Mon, 8 Feb 2016 22:29:54 +0000
Subject: [PATCH 15/17] Raise maximum rate limit

---
 probe/endpoint/procspy/background_reader_linux.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index 6d4405175..6199ac3d3 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -14,7 +14,7 @@ import (
 
 const (
 	initialRateLimitPeriod = 50 * time.Millisecond  // Read 20 * fdBlockSize file descriptors (/proc/PID/fd/*) per namespace per second
-	maxRateLimitPeriod     = 250 * time.Millisecond // Read at least 4 * fdBlockSize file descriptors per namespace per second
+	maxRateLimitPeriod     = 500 * time.Millisecond // Read at least 2 * fdBlockSize file descriptors per namespace per second
 	fdBlockSize            = uint64(300)            // Maximum number of /proc/PID/fd/* files to stat per rate-limit period
 	// (as a rule of thumb going through each block should be more expensive than reading /proc/PID/tcp{,6})
 	targetWalkTime = 10 * time.Second // Aim at walking all files in 10 seconds

From dfc136904cebd84ae221bbbb7413e6eb2e711e6c Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Tue, 9 Feb 2016 10:00:04 +0000
Subject: [PATCH 16/17] Review feedback

---
 .../procspy/background_reader_linux.go        | 88 +++++++++++--------
 probe/endpoint/procspy/proc_linux.go          | 25 +++---
 2 files changed, 65 insertions(+), 48 deletions(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index 6199ac3d3..6b1a06858 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -3,7 +3,6 @@ package procspy
 import (
 	"bytes"
 	"io"
-	"math"
 	"sync"
 	"time"
 
@@ -15,7 +14,8 @@ import (
 const (
 	initialRateLimitPeriod = 50 * time.Millisecond  // Read 20 * fdBlockSize file descriptors (/proc/PID/fd/*) per namespace per second
 	maxRateLimitPeriod     = 500 * time.Millisecond // Read at least 2 * fdBlockSize file descriptors per namespace per second
-	fdBlockSize            = uint64(300)            // Maximum number of /proc/PID/fd/* files to stat per rate-limit period
+	minRateLimitPeriod     = initialRateLimitPeriod
+	fdBlockSize            = uint64(300) // Maximum number of /proc/PID/fd/* files to stat per rate-limit period
 	// (as a rule of thumb going through each block should be more expensive than reading /proc/PID/tcp{,6})
 	targetWalkTime = 10 * time.Second // Aim at walking all files in 10 seconds
 )
@@ -51,14 +51,35 @@ func (br *backgroundReader) getWalkedProcPid(buf *bytes.Buffer) (map[uint64]*Pro
 	return br.latestSockets, err
 }
 
+type walkResult struct {
+	buf     *bytes.Buffer
+	sockets map[uint64]*Proc
+}
+
+func performWalk(w pidWalker, c chan<- walkResult) {
+	var (
+		err    error
+		result = walkResult{
+			buf: bytes.NewBuffer(make([]byte, 0, 5000)),
+		}
+	)
+
+	result.sockets, err = w.walk(result.buf)
+	if err != nil {
+		log.Errorf("background /proc reader: error walking /proc: %s", err)
+		result.buf.Reset()
+		result.sockets = nil
+	}
+	c <- result
+}
+
 func (br *backgroundReader) loop(walker process.Walker) {
 	var (
 		begin           time.Time                      // when we started the last performWalk
 		tickc           = time.After(time.Millisecond) // fire immediately
-		walkc           chan map[uint64]*Proc          // initially nil, i.e. off
-		walkBuf         = bytes.NewBuffer(make([]byte, 0, 5000))
+		walkc           chan walkResult                // initially nil, i.e. off
 		rateLimitPeriod = initialRateLimitPeriod
-		nextInterval    time.Duration
+		restInterval    time.Duration
 		ticker          = time.NewTicker(rateLimitPeriod)
 		pWalker         = newPidWalker(walker, ticker.C, fdBlockSize)
 	)
@@ -66,28 +87,27 @@ func (br *backgroundReader) loop(walker process.Walker) {
 	for {
 		select {
 		case <-tickc:
-			tickc = nil                             // turn off until the next loop
-			walkc = make(chan map[uint64]*Proc, 1)  // turn on (need buffered so we don't leak performWalk)
-			begin = time.Now()                      // reset counter
-			go performWalk(pWalker, walkBuf, walkc) // do work
+			tickc = nil                      // turn off until the next loop
+			walkc = make(chan walkResult, 1) // turn on (need buffered so we don't leak performWalk)
+			begin = time.Now()               // reset counter
+			go performWalk(pWalker, walkc)   // do work
 
-		case sockets := <-walkc:
-			// Swap buffers
+		case result := <-walkc:
+			// Expose results
 			br.mtx.Lock()
-			br.latestBuf, walkBuf = walkBuf, br.latestBuf
-			br.latestSockets = sockets
+			br.latestBuf = result.buf
+			br.latestSockets = result.sockets
 			br.mtx.Unlock()
-			walkBuf.Reset()
 
-			// Schedule next walk and adjust rate limit
+			// Schedule next walk and adjust its rate limit
 			walkTime := time.Since(begin)
-			rateLimitPeriod, nextInterval = scheduleNextWalk(rateLimitPeriod, walkTime)
+			rateLimitPeriod, restInterval = scheduleNextWalk(rateLimitPeriod, walkTime)
 			ticker.Stop()
 			ticker = time.NewTicker(rateLimitPeriod)
-			pWalker.ticker = ticker.C
+			pWalker.tickc = ticker.C
 
 			walkc = nil                      // turn off until the next loop
-			tickc = time.After(nextInterval) // turn on
+			tickc = time.After(restInterval) // turn on
 
 		case <-br.stopc:
 			pWalker.stop()
@@ -97,9 +117,8 @@ func (br *backgroundReader) loop(walker process.Walker) {
 	}
 }
 
-// Adjust rate limit for next walk and calculate how long to wait until it should be started
-func scheduleNextWalk(rateLimitPeriod time.Duration, took time.Duration) (time.Duration, time.Duration) {
-
+// Adjust rate limit for next walk and calculate when it should be started
+func scheduleNextWalk(rateLimitPeriod time.Duration, took time.Duration) (newRateLimitPeriod time.Duration, restInterval time.Duration) {
 	log.Debugf("background /proc reader: full pass took %s", took)
 	if float64(took)/float64(targetWalkTime) > 1.5 {
 		log.Warnf(
@@ -110,21 +129,16 @@ func scheduleNextWalk(rateLimitPeriod time.Duration, took time.Duration) (time.D
 	}
 
 	// Adjust rate limit to more-accurately meet the target walk time in next iteration
-	scaledRateLimitPeriod := float64(targetWalkTime) / float64(took) * float64(rateLimitPeriod)
-	rateLimitPeriod = time.Duration(math.Min(scaledRateLimitPeriod, float64(maxRateLimitPeriod)))
-
-	log.Debugf("background /proc reader: new rate limit %s", rateLimitPeriod)
-
-	return rateLimitPeriod, targetWalkTime - took
-}
-
-func performWalk(w pidWalker, buf *bytes.Buffer, c chan<- map[uint64]*Proc) {
-	sockets, err := w.walk(buf)
-	if err != nil {
-		log.Errorf("background /proc reader: error walking /proc: %s", err)
-		buf.Reset()
-		c <- nil
-		return
+	newRateLimitPeriod = time.Duration(float64(targetWalkTime) / float64(took) * float64(rateLimitPeriod))
+	if newRateLimitPeriod > maxRateLimitPeriod {
+		newRateLimitPeriod = maxRateLimitPeriod
+	} else if newRateLimitPeriod < minRateLimitPeriod {
+		newRateLimitPeriod = minRateLimitPeriod
 	}
-	c <- sockets
+	log.Debugf("background /proc reader: new rate limit period %s", newRateLimitPeriod)
+
+	// when to start next walk
+	restInterval = targetWalkTime - took
+
+	return
 }
diff --git a/probe/endpoint/procspy/proc_linux.go b/probe/endpoint/procspy/proc_linux.go
index e475aee18..f0cd8c69b 100644
--- a/probe/endpoint/procspy/proc_linux.go
+++ b/probe/endpoint/procspy/proc_linux.go
@@ -26,15 +26,15 @@ var (
 
 type pidWalker struct {
 	walker      process.Walker
-	ticker      <-chan time.Time
-	stopc       chan struct{}
-	fdBlockSize uint64
+	tickc       <-chan time.Time // Rate-limit clock. Sets the pace when traversing namespaces and /proc/PID/fd/* files.
+	stopc       chan struct{}    // Abort walk
+	fdBlockSize uint64           // Maximum number of /proc/PID/fd/* files to stat() per tick
 }
 
-func newPidWalker(walker process.Walker, ticker <-chan time.Time, fdBlockSize uint64) pidWalker {
+func newPidWalker(walker process.Walker, tickc <-chan time.Time, fdBlockSize uint64) pidWalker {
 	w := pidWalker{
 		walker:      walker,
-		ticker:      ticker,
+		tickc:       tickc,
 		fdBlockSize: fdBlockSize,
 		stopc:       make(chan struct{}),
 	}
@@ -72,7 +72,7 @@ func getNetNamespacePathSuffix() string {
 
 	v, err := getKernelVersion()
 	if err != nil {
-		log.Errorf("getNeNameSpacePath: cannot get kernel version: %s\n", err)
+		log.Errorf("getNamespacePathSuffix: cannot get kernel version: %s\n", err)
 		netNamespacePathSuffix = post38Path
 		return netNamespacePathSuffix
 	}
@@ -139,10 +139,13 @@ func (w pidWalker) walkNamespace(buf *bytes.Buffer, sockets map[uint64]*Proc, na
 
 		if fdBlockCount > w.fdBlockSize {
 			// we surpassed the filedescriptor rate limit
-			// TODO: worth selecting on w.stopc?
-			<-w.ticker
-			fdBlockCount = 0
+			select {
+			case <-w.tickc:
+			case <-w.stopc:
+				return nil // abort
+			}
 
+			fdBlockCount = 0
 			// read the connections again to
 			// avoid the race between between /net/tcp{,6} and /proc/PID/fd/*
 			if found, err := readProcessConnections(buf, namespaceProcs[i:]); err != nil || !found {
@@ -221,10 +224,10 @@ func (w pidWalker) walk(buf *bytes.Buffer) (map[uint64]*Proc, error) {
 
 	for _, procs := range namespaces {
 		select {
-		case <-w.ticker:
+		case <-w.tickc:
 			w.walkNamespace(buf, sockets, procs)
 		case <-w.stopc:
-			break
+			break // abort
 		}
 	}
 

From d4b114daea0c107c935a485ab27f7cfb23e3a5ee Mon Sep 17 00:00:00 2001
From: Alfonso Acosta <fons@weave.works>
Date: Tue, 9 Feb 2016 10:39:51 +0000
Subject: [PATCH 17/17] Review comments

---
 probe/endpoint/procspy/background_reader_linux.go | 9 +++------
 1 file changed, 3 insertions(+), 6 deletions(-)

diff --git a/probe/endpoint/procspy/background_reader_linux.go b/probe/endpoint/procspy/background_reader_linux.go
index 6b1a06858..ce4ee2a4f 100644
--- a/probe/endpoint/procspy/background_reader_linux.go
+++ b/probe/endpoint/procspy/background_reader_linux.go
@@ -31,8 +31,8 @@ type backgroundReader struct {
 // proc.
 func newBackgroundReader(walker process.Walker) *backgroundReader {
 	br := &backgroundReader{
-		latestBuf: bytes.NewBuffer(make([]byte, 0, 5000)),
-		stopc:     make(chan struct{}),
+		stopc:         make(chan struct{}),
+		latestSockets: map[uint64]*Proc{},
 	}
 	go br.loop(walker)
 	return br
@@ -137,8 +137,5 @@ func scheduleNextWalk(rateLimitPeriod time.Duration, took time.Duration) (newRat
 	}
 	log.Debugf("background /proc reader: new rate limit period %s", newRateLimitPeriod)
 
-	// when to start next walk
-	restInterval = targetWalkTime - took
-
-	return
+	return newRateLimitPeriod, targetWalkTime - took
 }