Make FileSystemDebug generic, and use for all debug reporting.

example/benchmark.sh

@@ -31,7 +31,7 @@ sleep ${DELAY}
 find ${MP} > /tmp/zipfiles.txt
 fusermount -u ${MP}
 
-# The command below should be profiled.
+# Run vanilla: block box measurement.
 ${ZIPFS} ${MP} ${ZIPFILE} >& zipfs.log &
 
 # Wait for zipfs to unpack and serve the file.
@@ -41,8 +41,8 @@ sleep ${DELAY}
 ${BULKSTAT} -runs 1 /tmp/zipfiles.txt
 
 # Performance number without 6prof running
+echo -e "\n\n"
 ${BULKSTAT} -runs 5 /tmp/zipfiles.txt
-
 echo -e "\n\n"
 
 # Run 6prof
@@ -54,6 +54,21 @@ ${BULKSTAT} -runs 3 /tmp/zipfiles.txt
 
 echo -e "\n\n"
 
-fusermount -u /tmp/zipbench
+fusermount -u ${MP}
+
+# Now run with internal monitoring.
+${ZIPFS} -latencies ${MP} ${ZIPFILE} >& zipfs.log &
+
+sleep ${DELAY}
+
+# Warm caches.
+${BULKSTAT} -runs 1 /tmp/zipfiles.txt
+
+# Measurements.
+${BULKSTAT} -runs 5 /tmp/zipfiles.txt
+
+# Dump internal measurements.
+cat ${MP}/.debug/*
+
+
 
-cat zipfs.log

example/loopback/loopback.go

@@ -27,6 +27,7 @@ func (me *PathPrintingFs) GetAttr(name string) (*fuse.Attr, fuse.Status) {
 func main() {
 	// Scans the arg list and sets up flags
 	debug := flag.Bool("debug", false, "print debugging messages.")
+	latencies := flag.Bool("latencies", false, "record latencies.")
 	threaded := flag.Bool("threaded", true, "switch off threading; print debugging messages.")
 	flag.Parse()
 	if flag.NArg() < 2 {
@@ -35,30 +36,44 @@ func main() {
 		os.Exit(2)
 	}
 
-	var fs fuse.FileSystem
+	var finalFs fuse.FileSystem
 	orig := flag.Arg(1)
 	loopbackfs := fuse.NewLoopbackFileSystem(orig)
-	fs = loopbackfs
-	debugFs := new(fuse.FileSystemDebug)
-	debugFs.Original = fs
-	fs = debugFs
+	finalFs = loopbackfs
 	
-	timing := fuse.NewTimingFileSystem(fs)
-	fs = timing
+	debugFs := fuse.NewFileSystemDebug()
+	if *latencies {
+		timing := fuse.NewTimingFileSystem(finalFs)
+		debugFs.AddTimingFileSystem(timing)
+		finalFs = timing
+	}
 
 	var opts fuse.FileSystemConnectorOptions
 
 	loopbackfs.FillOptions(&opts)
 
-	conn := fuse.NewFileSystemConnector(fs)
-	debugFs.Connector = conn
-	
-	rawTiming := fuse.NewTimingRawFileSystem(conn)
+	if *latencies {
+		debugFs.Original = finalFs
+		finalFs = debugFs
+	}
+
+	conn := fuse.NewFileSystemConnector(finalFs)
+	var finalRawFs fuse.RawFileSystem = conn
+	if *latencies {
+		rawTiming := fuse.NewTimingRawFileSystem(conn)
+		debugFs.AddRawTimingFileSystem(rawTiming)
+		finalRawFs = rawTiming
+	}
 	conn.SetOptions(opts)
 
-	state := fuse.NewMountState(rawTiming)
+	state := fuse.NewMountState(finalRawFs)
 	state.Debug = *debug
 
+	if *latencies {
+		state.RecordStatistics = true
+		debugFs.AddMountState(state)
+		debugFs.AddFileSystemConnector(conn)
+	}
 	mountPoint := flag.Arg(0)
 	err := state.Mount(mountPoint)
 	if err != nil {
@@ -66,33 +81,5 @@ func main() {
 		os.Exit(1)
 	}
 
-	fmt.Printf("Mounted %s on %s (threaded=%v, debug=%v)\n", orig, mountPoint, *threaded, *debug)
 	state.Loop(*threaded)
-	fmt.Println("Finished", state.Stats())
-
-	fmt.Println("\n\nMountState statistics\n")
-	counts := state.OperationCounts()
-	fmt.Println("Counts: ", counts)
-
-	latency := state.Latencies()
-	fmt.Println("Operation latency (ms):")
-	fuse.PrintMap(latency)
-
-	latency = rawTiming.Latencies()
-	fmt.Println("\n\nRaw FS (ms):", latency)
-
-	fmt.Println("\n\nLoopback FS statistics\n")
-	latency = timing.Latencies()
-	fmt.Println("Latency (ms):", latency)
-
-	fmt.Println("Operation counts:", timing.OperationCounts())
-
-	hot, unique := timing.HotPaths("GetAttr")
-	top := 20
-	start := len(hot) - top
-	if start < 0 {
-		start = 0
-	}
-	fmt.Printf("Unique GetAttr paths: %d\n", unique)
-	fmt.Printf("Top %d GetAttr paths: %v", top, hot[start:])
 }

example/multizip/multizip.go

@@ -29,5 +29,4 @@ func main() {
 
 	fmt.Printf("Mounted %s\n", mountPoint)
 	state.Loop(true)
-	fmt.Println(state.Stats())
 }

example/zipfs/main.go

@@ -14,6 +14,8 @@ var _ = log.Printf
 func main() {
 	// Scans the arg list and sets up flags
 	debug := flag.Bool("debug", false, "print debugging messages.")
+	latencies := flag.Bool("latencies", false, "record operation latencies.")
+	
 	flag.Parse()
 	if flag.NArg() < 2 {
 		// TODO - where to get program name?
@@ -21,22 +23,28 @@ func main() {
 		os.Exit(2)
 	}
 
-	fs := zipfs.NewZipArchiveFileSystem(flag.Arg(1))
+	var fs fuse.FileSystem
+	fs = zipfs.NewZipArchiveFileSystem(flag.Arg(1))
+	debugFs := fuse.NewFileSystemDebug()
+
+	if *latencies {
+		debugFs.Original = fs
+		fs = debugFs
+	}
+	
 	conn := fuse.NewFileSystemConnector(fs)
 	state := fuse.NewMountState(conn)
 
+	if *latencies {
+		debugFs.AddFileSystemConnector(conn)
+		debugFs.AddMountState(state)
+	}
+	
 	mountPoint := flag.Arg(0)
+	state.RecordStatistics = *latencies
 	state.Debug = *debug
 	state.Mount(mountPoint)
 
 	fmt.Printf("Mounted %s - PID %s\n", mountPoint, fuse.MyPID())
 	state.Loop(true)
-
-	latency := state.Latencies()
-	fmt.Println("Operation latency (ms):")
-	fuse.PrintMap(latency)
-	counts := state.OperationCounts()
-	fmt.Println("Counts: ", counts)
-
-	fmt.Println("Memory stats", state.Stats())
 }

fuse/fuse.go

@@ -103,6 +103,10 @@ func (me *MountState) Unmount() os.Error {
 }
 
 func (me *MountState) Write(req *request) {
+	if me.RecordStatistics {
+		req.preWriteNs = time.Nanoseconds()
+	}
+	
 	if req.outHeaderBytes == nil {
 		return
 	}
@@ -153,7 +157,7 @@ func (me *MountState) OperationCounts() map[string]int64 {
 	return r
 }
 
-func (me *MountState) Stats() string {
+func (me *MountState) BufferPoolStats() string {
 	return fmt.Sprintf("buffer alloc count %d\nbuffers %v",
 		me.buffers.AllocCount(), me.buffers.String())
 }
@@ -172,7 +176,9 @@ func (me *MountState) readRequest(req *request) os.Error {
 	n, err := me.mountFile.Read(req.inputBuf)
 	// If we start timing before the read, we may take into
 	// account waiting for input into the timing.
-	req.startNs = time.Nanoseconds()
+	if me.RecordStatistics {
+		req.startNs = time.Nanoseconds()
+	}
 	req.inputBuf = req.inputBuf[0:n]
 	return err
 }
@@ -197,6 +203,11 @@ func (me *MountState) discardRequest(req *request) {
 		key = opname + "-write"
 		me.operationLatencies[key] += (endNs - req.preWriteNs)
 		me.operationCounts[key] += 1
+
+		recDt := time.Nanoseconds() - endNs
+		key = "measurement"
+		me.operationCounts[key] += 1
+		me.operationLatencies[key] += recDt
 	}
 
 	me.buffers.FreeBuffer(req.inputBuf)
@@ -294,14 +305,15 @@ func (me *MountState) handle(req *request) {
 	// failed, err: writev: no such file or directory
 	if req.inHeader.Opcode != FUSE_FORGET {
 		serialize(req, handler, me.Debug)
-		req.preWriteNs = time.Nanoseconds()
 		me.Write(req)
 	}
 }
 
 func (me *MountState) dispatch(req *request, handler *operationHandler) {
-	req.dispatchNs = time.Nanoseconds()
-
+	if me.RecordStatistics {
+		req.dispatchNs = time.Nanoseconds()
+	}
+	
 	if me.Debug {
 		nm := ""
 		// TODO - reinstate filename printing.

fuse/pathdebug.go

@@ -2,30 +2,174 @@ package fuse
 
 import (
 	"fmt"
+	"log"
+	"path/filepath"
+	"sort"
+	"strings"
+	"sync"
+	"syscall"
 )
 
 var _ = fmt.Println
+var _ = log.Println
 
-type FileSystemDebug struct {
-	// TODO - use a generic callback system instead.
-	Connector *FileSystemConnector
+const (
+	DebugDir = ".debug"
+)
+
+type getter func() []byte
 
+type FileSystemDebug struct {
+	sync.RWMutex
+	callbacks map[string]getter
 	WrappingFileSystem
 }
 
+func NewFileSystemDebug() *FileSystemDebug {
+	return &FileSystemDebug{
+		callbacks: make(map[string]getter),
+	}
+}
+
+func (me *FileSystemDebug) Add(name string, callback getter) {
+	me.RWMutex.Lock()
+	defer me.RWMutex.Unlock()
+	me.callbacks[name] = callback
+}
+
 func (me *FileSystemDebug) Open(path string, flags uint32) (fuseFile File, status Status) {
-	if path == ".debug" && me.Connector != nil {
-		return NewReadOnlyFile([]byte(me.Connector.DebugString())), OK
+	
+	content := me.getContent(path)
+	if content != nil {
+		return NewReadOnlyFile(content), OK
 	}
 	return me.Original.Open(path, flags)
 }
 
+func (me *FileSystemDebug) getContent(path string) []byte {
+	comps := strings.Split(path, filepath.SeparatorString, -1)
+	if comps[0] == DebugDir {
+		me.RWMutex.RLock()
+		defer me.RWMutex.RUnlock()
+		f := me.callbacks[comps[1]]
+		if f != nil {
+			return f()
+		}
+	}
+	return nil
+}
+
+func (me *FileSystemDebug) GetXAttr(name string, attr string) ([]byte, Status) {
+	if strings.HasPrefix(name, DebugDir) {
+		return nil, syscall.ENODATA
+	}
+	return me.Original.GetXAttr(name, attr)
+}
+
 func (me *FileSystemDebug) GetAttr(path string) (*Attr, Status) {
-	if path == ".debug" && me.Connector != nil {
+	if !strings.HasPrefix(path, DebugDir) {
+		return me.Original.GetAttr(path)
+	}		
+	if path == DebugDir {
+		return &Attr{
+			Mode: S_IFDIR | 0755,
+		}, OK
+	}
+	c := me.getContent(path)
+	if c != nil {
 		return &Attr{
-			Mode: S_IFREG,
-			Size: uint64(len(me.Connector.DebugString())),
+			Mode: S_IFREG | 0644,
+			Size: uint64(len(c)),
 		},OK
 	}
-	return me.Original.GetAttr(path)
+	return nil, ENOENT
+}
+
+func FloatMapToBytes(m map[string]float64) []byte {
+	keys := make([]string, 0, len(m))
+	for k, _ := range m {
+		keys = append(keys, k)
+	}
+
+	sort.SortStrings(keys)
+
+	var r []string
+	for _, k := range keys {
+		r = append(r, fmt.Sprintf("%v %v", k, m[k]))
+	}
+	return []byte(strings.Join(r, "\n"))
+}
+
+// Ugh - generics.
+func IntMapToBytes(m map[string]int64) []byte {
+	keys := make([]string, 0, len(m))
+	for k, _ := range m {
+		keys = append(keys, k)
+	}
+
+	sort.SortStrings(keys)
+
+	var r []string
+	for _, k := range keys {
+		r = append(r, fmt.Sprintf("%v %v", k, m[k]))
+	}
+	return []byte(strings.Join(r, "\n"))
+}
+
+func (me *FileSystemDebug) OpenDir(name string) (stream chan DirEntry, status Status) {
+	if name == DebugDir {
+		me.RWMutex.RLock()
+		defer me.RWMutex.RUnlock()
+	
+		stream = make(chan DirEntry, len(me.callbacks))
+		for k, _ := range me.callbacks {
+			stream <- DirEntry {
+				Name: k,
+				Mode: S_IFREG,
+			}
+		}
+		close(stream)
+		return stream, OK
+		
+	}
+	return me.Original.OpenDir(name)
+}
+
+func (me *FileSystemDebug) AddMountState(state *MountState) {
+	me.Add("mountstate-latencies",
+		func() []byte { return FloatMapToBytes(state.Latencies()) })
+	me.Add("mountstate-opcounts",
+		func() []byte { return IntMapToBytes(state.OperationCounts()) })
+	me.Add("mountstate-bufferpool",
+		func() []byte { return []byte(state.BufferPoolStats()) })
+}
+
+func (me *FileSystemDebug) AddFileSystemConnector(conn *FileSystemConnector) {
+	me.Add("filesystemconnector-stats",
+		func() []byte { return []byte(conn.Statistics()) })
+}
+
+func hotPaths(timing *TimingFileSystem) []byte {
+	hot, unique := timing.HotPaths("GetAttr")
+	top := 20
+	start := len(hot) - top
+	if start < 0 {
+		start = 0
+	}
+	return []byte(fmt.Sprintf("Unique GetAttr paths: %d\nTop %d GetAttr paths: %v",
+		unique, top, hot[start:]))
+}
+
+func (me *FileSystemDebug) AddTimingFileSystem(tfs *TimingFileSystem) {
+	me.Add("timingfs-latencies",
+		func() []byte { return FloatMapToBytes(tfs.Latencies()) })
+	me.Add("timingfs-opcounts",
+		func() []byte { return IntMapToBytes(tfs.OperationCounts()) })
+	me.Add("timingfs-hotpaths",
+		func() []byte { return hotPaths(tfs) })
+}
+
+func (me *FileSystemDebug) AddRawTimingFileSystem(tfs *TimingRawFileSystem) {
+	me.Add("rawtimingfs-latencies",
+		func() []byte { return FloatMapToBytes(tfs.Latencies()) })
 }

fuse/pathdebug_test.go

+package fuse
+
+import (
+	"io/ioutil"
+	"path/filepath"
+	"os"
+	"testing"
+)
+
+func TestPathDebug(t *testing.T) {
+	fs := &DefaultFileSystem{}
+	debugFs := NewFileSystemDebug(fs)
+	debugFs.Add("test-entry", func()[]byte { return []byte("test-content"); })
+
+	connector := NewFileSystemConnector(debugFs)
+	mountPoint := MakeTempDir()
+
+	state := NewMountState(connector)
+	state.Mount(mountPoint)
+	state.Debug = true
+	defer state.Unmount()
+
+	go state.Loop(false)
+
+	dir := filepath.Join(mountPoint, ".debug")
+	_, err := os.Lstat(dir)
+	CheckSuccess(err)
+
+	names, err := ioutil.ReadDir(dir)
+	CheckSuccess(err)
+
+	if len(names) != 1 || names[0].Name != "test-entry" {
+		t.Error("unexpected readdir out:", names) 
+	}
+
+	c, err := ioutil.ReadFile(filepath.Join(dir, "test-entry"))
+	CheckSuccess(err)
+
+	if string(c) != "test-content" {
+		t.Error("unexpected content", c)
+	}
+}

fuse/pathfilesystem.go

@@ -185,7 +185,7 @@ type interfaceBridge struct {
 	Iface interface{}
 }
 
-func (me *FileSystemConnector) DebugString() string {
+func (me *FileSystemConnector) Statistics() string {
 	me.treeLock.RLock()
 	defer me.treeLock.RUnlock()
 

fuse/timingfs.go

@@ -67,7 +67,8 @@ func (me *TimingFileSystem) Latencies() map[string]float64 {
 
 	r := make(map[string]float64)
 	for k, v := range me.counts {
-		r[k] = float64(me.latencies[k]) / float64(v)
+		lat := float64(me.latencies[k]) / float64(v)
+		r[k] = lat
 	}
 	return r
 }