Merge pull request #742 from weaveworks/284-profiling

Various CPU usage gains from profiling

common/fs/fs.go

@@ -10,6 +10,7 @@ import (
 // Interface is the filesystem interface type.
 type Interface interface {
 	ReadDir(string) ([]os.FileInfo, error)
+	ReadDirNames(string) ([]string, error)
 	ReadFile(string) ([]byte, error)
 	Lstat(string, *syscall.Stat_t) error
 	Stat(string, *syscall.Stat_t) error
@@ -25,6 +26,15 @@ func (realFS) ReadDir(path string) ([]os.FileInfo, error) {
 	return ioutil.ReadDir(path)
 }
 
+func (realFS) ReadDirNames(path string) ([]string, error) {
+	fh, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	defer fh.Close()
+	return fh.Readdirnames(-1)
+}
+
 func (realFS) ReadFile(path string) ([]byte, error) {
 	return ioutil.ReadFile(path)
 }
@@ -48,6 +58,11 @@ func ReadDir(path string) ([]os.FileInfo, error) {
 	return fs.ReadDir(path)
 }
 
+// ReadDirNames see os.File.ReadDirNames
+func ReadDirNames(path string) ([]string, error) {
+	return fs.ReadDirNames(path)
+}
+
 // ReadFile see ioutil.ReadFile
 func ReadFile(path string) ([]byte, error) {
 	return fs.ReadFile(path)

probe/endpoint/procspy/benchmark_internal_test.go

@@ -6,13 +6,16 @@ import (
 )
 
 func BenchmarkParseConnectionsBaseline(b *testing.B) {
-	readFile = func(string, *bytes.Buffer) error { return nil }
+	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) error { _, err := buf.Write(fixture); return err }
+	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
 }

probe/endpoint/procspy/proc.go

@@ -24,38 +24,41 @@ func SetProcRoot(root string) {
 // walkProcPid walks over all numerical (PID) /proc entries, and sees if their
 // ./fd/* files are symlink to sockets. Returns a map from socket ID (inode)
 // to PID. Will return an error if /proc isn't there.
-func walkProcPid(buf *bytes.Buffer, walker process.Walker) (map[uint64]Proc, error) {
+func walkProcPid(buf *bytes.Buffer, walker process.Walker) (map[uint64]*Proc, error) {
 	var (
-		res        = map[uint64]Proc{}
-		namespaces = map[uint64]struct{}{}
+		res        = map[uint64]*Proc{}
+		namespaces = map[uint64]bool{} // map namespace id -> has connections
 		statT      syscall.Stat_t
 	)
 
 	walker.Walk(func(p process.Process) {
 		dirName := strconv.Itoa(p.PID)
 		fdBase := filepath.Join(procRoot, dirName, "fd")
-		fds, err := fs.ReadDir(fdBase)
-		if err != nil {
-			// Process is be gone by now, or we don't have access.
-			return
-		}
 
 		// Read network namespace, and if we haven't seen it before,
 		// read /proc/<pid>/net/tcp
-		err = fs.Lstat(filepath.Join(procRoot, dirName, "/ns/net"), &statT)
-		if err != nil {
+		if err := fs.Lstat(filepath.Join(procRoot, dirName, "/ns/net"), &statT); err != nil {
 			return
 		}
-
-		if _, ok := namespaces[statT.Ino]; !ok {
-			namespaces[statT.Ino] = struct{}{}
-			readFile(filepath.Join(procRoot, dirName, "/net/tcp"), buf)
-			readFile(filepath.Join(procRoot, dirName, "/net/tcp6"), buf)
+		hasConns, ok := namespaces[statT.Ino]
+		if !ok {
+			read, err := readFile(filepath.Join(procRoot, dirName, "/net/tcp"), buf)
+			hasConns = err == nil && read > 0
+			namespaces[statT.Ino] = hasConns
+		}
+		if !hasConns {
+			return
 		}
 
+		fds, err := fs.ReadDirNames(fdBase)
+		if err != nil {
+			// Process is be gone by now, or we don't have access.
+			return
+		}
+		var proc *Proc
 		for _, fd := range fds {
 			// Direct use of syscall.Stat() to save garbage.
-			err = fs.Stat(filepath.Join(fdBase, fd.Name()), &statT)
+			err = fs.Stat(filepath.Join(fdBase, fd), &statT)
 			if err != nil {
 				continue
 			}
@@ -64,11 +67,13 @@ func walkProcPid(buf *bytes.Buffer, walker process.Walker) (map[uint64]Proc, err
 			if statT.Mode&syscall.S_IFMT != syscall.S_IFSOCK {
 				continue
 			}
-
-			res[statT.Ino] = Proc{
-				PID:  uint(p.PID),
-				Name: p.Comm,
+			if proc == nil {
+				proc = &Proc{
+					PID:  uint(p.PID),
+					Name: p.Comm,
+				}
 			}
+			res[statT.Ino] = proc
 		}
 	})
 
@@ -78,12 +83,11 @@ func walkProcPid(buf *bytes.Buffer, walker process.Walker) (map[uint64]Proc, err
 // 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) error {
+var readFile = func(filename string, buf *bytes.Buffer) (int64, error) {
 	f, err := fs.Open(filename)
 	if err != nil {
-		return err
+		return -1, err
 	}
-	_, err = buf.ReadFrom(f)
-	f.Close()
-	return err
+	defer f.Close()
+	return buf.ReadFrom(f)
 }

probe/endpoint/procspy/proc_internal_test.go

@@ -33,6 +33,12 @@ var mockFS = fs.Dir("",
 					FStat: syscall.Stat_t{},
 				},
 			),
+			fs.Dir("net",
+				fs.File{
+					FName:     "tcp",
+					FContents: "I'm a little teapot",
+				},
+			),
 			fs.File{
 				FName:     "stat",
 				FContents: "1 na R 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 1",
@@ -50,7 +56,7 @@ func TestWalkProcPid(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	want := map[uint64]Proc{
+	want := map[uint64]*Proc{
 		45: {
 			PID:  1,
 			Name: "foo",

probe/endpoint/procspy/spy_linux.go

@@ -16,7 +16,7 @@ var bufPool = sync.Pool{
 type pnConnIter struct {
 	pn    *ProcNet
 	buf   *bytes.Buffer
-	procs map[uint64]Proc
+	procs map[uint64]*Proc
 }
 
 func (c *pnConnIter) Next() *Connection {
@@ -27,7 +27,7 @@ func (c *pnConnIter) Next() *Connection {
 		return nil
 	}
 	if proc, ok := c.procs[n.inode]; ok {
-		n.Proc = proc
+		n.Proc = *proc
 	}
 	return n
 }
@@ -38,7 +38,7 @@ var cbConnections = func(processes bool, walker process.Walker) (ConnIter, error
 	buf := bufPool.Get().(*bytes.Buffer)
 	buf.Reset()
 
-	var procs map[uint64]Proc
+	var procs map[uint64]*Proc
 	if processes {
 		var err error
 		if procs, err = walkProcPid(buf, walker); err != nil {

probe/process/cache.go

@@ -0,0 +1,72 @@
+package process
+
+import (
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/armon/go-metrics"
+	"github.com/coocood/freecache"
+
+	"github.com/weaveworks/scope/common/fs"
+)
+
+const (
+	generalTimeout = 30 // seconds
+	statsTimeout   = 10 //seconds
+)
+
+var (
+	hitMetricsKey  = []string{"process", "cache", "hit"}
+	missMetricsKey = []string{"process", "cache", "miss"}
+)
+
+var fileCache = freecache.NewCache(1024 * 16)
+
+type entry struct {
+	buf []byte
+	err error
+	ts  time.Time
+}
+
+func cachedReadFile(path string) ([]byte, error) {
+	key := []byte(path)
+	if v, err := fileCache.Get(key); err == nil {
+		metrics.IncrCounter(hitMetricsKey, 1.0)
+		return v, nil
+	}
+
+	buf, err := fs.ReadFile(path)
+	fileCache.Set(key, buf, generalTimeout)
+	metrics.IncrCounter(missMetricsKey, 1.0)
+	return buf, err
+}
+
+// we cache the stats, but for a shorter period
+func readStats(path string) (int, int, error) {
+	var (
+		key = []byte(path)
+		buf []byte
+		err error
+	)
+	if buf, err = fileCache.Get(key); err == nil {
+		metrics.IncrCounter(hitMetricsKey, 1.0)
+	} else {
+		buf, err = fs.ReadFile(path)
+		if err != nil {
+			return -1, -1, err
+		}
+		fileCache.Set(key, buf, statsTimeout)
+		metrics.IncrCounter(missMetricsKey, 1.0)
+	}
+	splits := strings.Fields(string(buf))
+	ppid, err := strconv.Atoi(splits[3])
+	if err != nil {
+		return -1, -1, err
+	}
+	threads, err := strconv.Atoi(splits[19])
+	if err != nil {
+		return -1, -1, err
+	}
+	return ppid, threads, nil
+}

probe/process/walker_linux.go

@@ -23,41 +23,30 @@ func NewWalker(procRoot string) Walker {
 // passes one-by-one to the supplied function. Walk is only made public
 // so that is can be tested.
 func (w *walker) Walk(f func(Process)) error {
-	dirEntries, err := fs.ReadDir(w.procRoot)
+	dirEntries, err := fs.ReadDirNames(w.procRoot)
 	if err != nil {
 		return err
 	}
 
-	for _, dirEntry := range dirEntries {
-		filename := dirEntry.Name()
+	for _, filename := range dirEntries {
 		pid, err := strconv.Atoi(filename)
 		if err != nil {
 			continue
 		}
 
-		stat, err := fs.ReadFile(path.Join(w.procRoot, filename, "stat"))
+		ppid, threads, err := readStats(path.Join(w.procRoot, filename, "stat"))
 		if err != nil {
 			continue
 		}
-		splits := strings.Fields(string(stat))
-		ppid, err := strconv.Atoi(splits[3])
-		if err != nil {
-			return err
-		}
-
-		threads, err := strconv.Atoi(splits[19])
-		if err != nil {
-			return err
-		}
 
 		cmdline := ""
-		if cmdlineBuf, err := fs.ReadFile(path.Join(w.procRoot, filename, "cmdline")); err == nil {
+		if cmdlineBuf, err := cachedReadFile(path.Join(w.procRoot, filename, "cmdline")); err == nil {
 			cmdlineBuf = bytes.Replace(cmdlineBuf, []byte{'\000'}, []byte{' '}, -1)
 			cmdline = string(cmdlineBuf)
 		}
 
 		comm := "(unknown)"
-		if commBuf, err := fs.ReadFile(path.Join(w.procRoot, filename, "comm")); err == nil {
+		if commBuf, err := cachedReadFile(path.Join(w.procRoot, filename, "comm")); err == nil {
 			comm = strings.TrimSpace(string(commBuf))
 		}
 

test/fs/fs.go

@@ -89,6 +89,24 @@ func (p dir) ReadDir(path string) ([]os.FileInfo, error) {
 	return fs.ReadDir(tail)
 }
 
+func (p dir) ReadDirNames(path string) ([]string, error) {
+	if path == "/" {
+		result := []string{}
+		for _, v := range p.entries {
+			result = append(result, v.Name())
+		}
+		return result, nil
+	}
+
+	head, tail := split(path)
+	fs, ok := p.entries[head]
+	if !ok {
+		return nil, fmt.Errorf("Not found: %s", path)
+	}
+
+	return fs.ReadDirNames(tail)
+}
+
 func (p dir) ReadFile(path string) ([]byte, error) {
 	if path == "/" {
 		return nil, fmt.Errorf("I'm a directory!")
@@ -156,6 +174,11 @@ func (p File) ReadDir(path string) ([]os.FileInfo, error) {
 	return nil, fmt.Errorf("I'm a file!")
 }
 
+// ReadDirNames implements FS
+func (p File) ReadDirNames(path string) ([]string, error) {
+	return nil, fmt.Errorf("I'm a file!")
+}
+
 // ReadFile implements FS
 func (p File) ReadFile(path string) ([]byte, error) {
 	if path != "/" {