syz-manager: collect more stats about exchange calls

Collect total number of calls, average number of requested progs,
average server latency, and average end-to-end latency.

5 files changed, 54 insertions, 26 deletions

diff --git a/pkg/rpctype/rpctype.go b/pkg/rpctype/rpctype.go
index cb50ecfa6..b31f9ec96 100644
--- a/pkg/rpctype/rpctype.go
+++ b/pkg/rpctype/rpctype.go
@@ -7,6 +7,7 @@ package rpctype
 
 import (
 	"math"
+	"time"
 
 	"github.com/google/syzkaller/pkg/host"
 	"github.com/google/syzkaller/pkg/ipc"
@@ -45,6 +46,7 @@ type ExchangeInfoRequest struct {
 	NeedProgs  int
 	StatsDelta map[string]uint64
 	Results    []ExecutionResult
+	Latency    time.Duration // latency of the previous ExchangeInfo request
 }
 
 // ExchangeInfoReply is a reply to ExchangeInfoRequest.
diff --git a/syz-fuzzer/fuzzer.go b/syz-fuzzer/fuzzer.go
index 56b3321e5..f539de6f6 100644
--- a/syz-fuzzer/fuzzer.go
+++ b/syz-fuzzer/fuzzer.go
@@ -259,7 +259,7 @@ func main() {
 		execOpts.Flags |= ipc.FlagEnableCoverageFilter
 	}
 	// Query enough inputs at the beginning.
-	fuzzerTool.exchangeDataCall(inputsCount, nil)
+	fuzzerTool.exchangeDataCall(inputsCount, nil, 0)
 	log.Logf(0, "starting %v executor processes", *flagProcs)
 	for pid := 0; pid < *flagProcs; pid++ {
 		proc, err := newProc(fuzzerTool, execOpts, pid)
@@ -335,19 +335,26 @@ func (tool *FuzzerTool) filterDataRaceFrames(frames []string) {
 	log.Logf(0, "%s", output)
 }
 
-func (tool *FuzzerTool) exchangeDataCall(needProgs int, results []executionResult) {
+func (tool *FuzzerTool) exchangeDataCall(needProgs int, results []executionResult,
+	latency time.Duration) time.Duration {
 	a := &rpctype.ExchangeInfoRequest{
 		Name:       tool.name,
 		NeedProgs:  needProgs,
 		StatsDelta: tool.grabStats(),
+		Latency:    latency,
 	}
 	for _, result := range results {
 		a.Results = append(a.Results, tool.convertExecutionResult(result))
 	}
 	r := &rpctype.ExchangeInfoReply{}
+	start := time.Now()
 	if err := tool.manager.Call("Manager.ExchangeInfo", a, r); err != nil {
 		log.SyzFatalf("Manager.ExchangeInfo call failed: %v", err)
 	}
+	latency = time.Since(start)
+	if needProgs != len(r.Requests) {
+		log.SyzFatalf("manager returned wrong number of requests: %v/%v", needProgs, len(r.Requests))
+	}
 	tool.updateMaxSignal(r.NewMaxSignal, r.DropMaxSignal)
 	for _, req := range r.Requests {
 		p := tool.deserializeInput(req.ProgData)
@@ -359,9 +366,11 @@ func (tool *FuzzerTool) exchangeDataCall(needProgs int, results []executionResul
 			prog:             p,
 		}
 	}
+	return latency
 }
 
 func (tool *FuzzerTool) exchangeDataWorker() {
+	var latency time.Duration
 	for result := range tool.results {
 		results := []executionResult{
 			result,
@@ -377,7 +386,7 @@ func (tool *FuzzerTool) exchangeDataWorker() {
 			}
 		}
 		// Replenish exactly the finished requests.
-		tool.exchangeDataCall(len(results), results)
+		latency = tool.exchangeDataCall(len(results), results, latency)
 	}
 }
 
diff --git a/syz-manager/http.go b/syz-manager/http.go
index 4477c2a52..edc167bcf 100644
--- a/syz-manager/http.go
+++ b/syz-manager/http.go
@@ -175,8 +175,8 @@ func (mgr *Manager) collectStats() []UIStat {
 		var intStats []UIStat
 		for k, v := range rawStats {
 			val := ""
-			switch k {
-			case "fuzzer jobs":
+			switch {
+			case k == "fuzzer jobs" || strings.HasPrefix(k, "rpc exchange"):
 				val = fmt.Sprint(v)
 			default:
 				val = rateStat(v, secs)
diff --git a/syz-manager/rpc.go b/syz-manager/rpc.go
index fa11dc558..0c94a5793 100644
--- a/syz-manager/rpc.go
+++ b/syz-manager/rpc.go
@@ -8,6 +8,7 @@ import (
 	"net"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/google/syzkaller/pkg/cover"
 	"github.com/google/syzkaller/pkg/fuzzer"
@@ -177,6 +178,7 @@ func (serv *RPCServer) Check(a *rpctype.CheckArgs, r *int) error {
 }
 
 func (serv *RPCServer) ExchangeInfo(a *rpctype.ExchangeInfoRequest, r *rpctype.ExchangeInfoReply) error {
+	start := time.Now()
 	var runner *Runner
 	if val, _ := serv.runners.Load(a.Name); val != nil {
 		runner = val.(*Runner)
@@ -231,6 +233,11 @@ func (serv *RPCServer) ExchangeInfo(a *rpctype.ExchangeInfoRequest, r *rpctype.E
 
 	log.Logf(2, "exchange with %s: %d done, %d new requests, %d new max signal, %d drop signal",
 		a.Name, len(a.Results), len(r.Requests), len(r.NewMaxSignal), len(r.DropMaxSignal))
+
+	serv.stats.rpcExchangeCalls.inc()
+	serv.stats.rpcExchangeProgs.add(a.NeedProgs)
+	serv.stats.rpcExchangeClientLatency.add(int(a.Latency))
+	serv.stats.rpcExchangeServerLatency.add(int(time.Since(start).Nanoseconds()))
 	return nil
 }
 
diff --git a/syz-manager/stats.go b/syz-manager/stats.go
index d85a1a78b..88db8ed17 100644
--- a/syz-manager/stats.go
+++ b/syz-manager/stats.go
@@ -14,26 +14,30 @@ import (
 type Stat uint64
 
 type Stats struct {
-	crashes             Stat
-	crashTypes          Stat
-	crashSuppressed     Stat
-	vmRestarts          Stat
-	newInputs           Stat
-	execTotal           Stat
-	rpcTraffic          Stat
-	hubSendProgAdd      Stat
-	hubSendProgDel      Stat
-	hubSendRepro        Stat
-	hubRecvProg         Stat
-	hubRecvProgDrop     Stat
-	hubRecvRepro        Stat
-	hubRecvReproDrop    Stat
-	corpusCover         Stat
-	corpusCoverFiltered Stat
-	corpusSignal        Stat
-	maxSignal           Stat
-	triageQueueLen      Stat
-	fuzzerJobs          Stat
+	crashes                  Stat
+	crashTypes               Stat
+	crashSuppressed          Stat
+	vmRestarts               Stat
+	newInputs                Stat
+	execTotal                Stat
+	rpcTraffic               Stat
+	rpcExchangeCalls         Stat
+	rpcExchangeProgs         Stat
+	rpcExchangeServerLatency Stat
+	rpcExchangeClientLatency Stat
+	hubSendProgAdd           Stat
+	hubSendProgDel           Stat
+	hubSendRepro             Stat
+	hubRecvProg              Stat
+	hubRecvProgDrop          Stat
+	hubRecvRepro             Stat
+	hubRecvReproDrop         Stat
+	corpusCover              Stat
+	corpusCoverFiltered      Stat
+	corpusSignal             Stat
+	maxSignal                Stat
+	triageQueueLen           Stat
+	fuzzerJobs               Stat
 
 	mu         sync.Mutex
 	namedStats map[string]uint64
@@ -74,9 +78,15 @@ func (stats *Stats) all() map[string]uint64 {
 		"filtered coverage": stats.corpusCoverFiltered.get(),
 		"signal":            stats.corpusSignal.get(),
 		"max signal":        stats.maxSignal.get(),
-		"rpc traffic (MB)":  stats.rpcTraffic.get() / 1e6,
+		"rpc traffic (MB)":  stats.rpcTraffic.get() >> 20,
 		"fuzzer jobs":       stats.fuzzerJobs.get(),
 	}
+	if exchanges := stats.rpcExchangeCalls.get(); exchanges != 0 {
+		m["exchange calls"] = exchanges
+		m["exchange progs"] = uint64(float64(stats.rpcExchangeProgs.get())/float64(exchanges) + 0.5)
+		m["exchange lat server (us)"] = stats.rpcExchangeServerLatency.get() / exchanges / 1e3
+		m["exchange lat client (us)"] = stats.rpcExchangeClientLatency.get() / exchanges / 1e3
+	}
 	if stats.haveHub {
 		m["hub: send prog add"] = stats.hubSendProgAdd.get()
 		m["hub: send prog del"] = stats.hubSendProgDel.get()