syz-manager: factor out rpc serving part

Update #605

9 files changed, 268 insertions, 166 deletions

diff --git a/pkg/mgrconfig/mgrconfig.go b/pkg/mgrconfig/mgrconfig.go
index 8859ac7e8..595b2558b 100644
--- a/pkg/mgrconfig/mgrconfig.go
+++ b/pkg/mgrconfig/mgrconfig.go
@@ -261,7 +261,7 @@ func splitTarget(target string) (string, string, string, error) {
 	return os, vmarch, arch, nil
 }
 
-func ParseEnabledSyscalls(target *prog.Target, enabled, disabled []string) (map[int]bool, error) {
+func ParseEnabledSyscalls(target *prog.Target, enabled, disabled []string) ([]int, error) {
 	syscalls := make(map[int]bool)
 	if len(enabled) != 0 {
 		for _, c := range enabled {
@@ -296,7 +296,11 @@ func ParseEnabledSyscalls(target *prog.Target, enabled, disabled []string) (map[
 	if len(syscalls) == 0 {
 		return nil, fmt.Errorf("all syscalls are disabled by disable_syscalls in config")
 	}
-	return syscalls, nil
+	var arr []int
+	for id := range syscalls {
+		arr = append(arr, id)
+	}
+	return arr, nil
 }
 
 func matchSyscall(name, pattern string) bool {
diff --git a/pkg/rpctype/rpc.go b/pkg/rpctype/rpc.go
index 42cab2969..a7900bb13 100644
--- a/pkg/rpctype/rpc.go
+++ b/pkg/rpctype/rpc.go
@@ -20,13 +20,13 @@ type RPCServer struct {
 	s  *rpc.Server
 }
 
-func NewRPCServer(addr string, receiver interface{}) (*RPCServer, error) {
+func NewRPCServer(addr, name string, receiver interface{}) (*RPCServer, error) {
 	ln, err := net.Listen("tcp", addr)
 	if err != nil {
 		return nil, fmt.Errorf("failed to listen on %v: %v", addr, err)
 	}
 	s := rpc.NewServer()
-	if err := s.Register(receiver); err != nil {
+	if err := s.RegisterName(name, receiver); err != nil {
 		return nil, err
 	}
 	serv := &RPCServer{
diff --git a/syz-hub/hub.go b/syz-hub/hub.go
index 259265576..db8dd506e 100644
--- a/syz-hub/hub.go
+++ b/syz-hub/hub.go
@@ -57,7 +57,7 @@ func main() {
 
 	hub.initHTTP(cfg.HTTP)
 
-	s, err := rpctype.NewRPCServer(cfg.RPC, hub)
+	s, err := rpctype.NewRPCServer(cfg.RPC, "Hub", hub)
 	if err != nil {
 		log.Fatalf("failed to create rpc server: %v", err)
 	}
diff --git a/syz-manager/html.go b/syz-manager/html.go
index b572d59a9..2a05748f7 100644
--- a/syz-manager/html.go
+++ b/syz-manager/html.go
@@ -23,6 +23,7 @@ import (
 	"github.com/google/syzkaller/pkg/html"
 	"github.com/google/syzkaller/pkg/log"
 	"github.com/google/syzkaller/pkg/osutil"
+	"github.com/google/syzkaller/pkg/signal"
 	"github.com/google/syzkaller/prog"
 )
 
@@ -101,14 +102,17 @@ func (mgr *Manager) collectStats() []UIStat {
 	mgr.mu.Lock()
 	defer mgr.mu.Unlock()
 
+	rawStats := mgr.stats.all()
 	stats := []UIStat{
 		{Name: "uptime", Value: fmt.Sprint(time.Since(mgr.startTime) / 1e9 * 1e9)},
 		{Name: "fuzzing", Value: fmt.Sprint(mgr.fuzzingTime / 60e9 * 60e9)},
 		{Name: "corpus", Value: fmt.Sprint(len(mgr.corpus)), Link: "/corpus"},
 		{Name: "triage queue", Value: fmt.Sprint(len(mgr.candidates))},
-		{Name: "cover", Value: fmt.Sprint(len(mgr.corpusCover)), Link: "/cover"},
-		{Name: "signal", Value: fmt.Sprint(mgr.corpusSignal.Len())},
+		{Name: "cover", Value: fmt.Sprint(rawStats["cover"]), Link: "/cover"},
+		{Name: "signal", Value: fmt.Sprint(rawStats["signal"])},
 	}
+	delete(rawStats, "cover")
+	delete(rawStats, "signal")
 	if mgr.checkResult != nil {
 		stats = append(stats, UIStat{
 			Name:  "syscalls",
@@ -121,9 +125,7 @@ func (mgr *Manager) collectStats() []UIStat {
 	if !mgr.firstConnect.IsZero() {
 		secs = uint64(time.Since(mgr.firstConnect))/1e9 + 1
 	}
-
-	intStats := convertStats(mgr.stats.all(), secs)
-	intStats = append(intStats, convertStats(mgr.fuzzerStats, secs)...)
+	intStats := convertStats(rawStats, secs)
 	sort.Slice(intStats, func(i, j int) bool {
 		return intStats[i].Name < intStats[j].Name
 	})
@@ -254,8 +256,12 @@ func (mgr *Manager) httpCoverCover(w http.ResponseWriter, r *http.Request) {
 }
 
 func (mgr *Manager) httpCoverFallback(w http.ResponseWriter, r *http.Request) {
+	var maxSignal signal.Signal
+	for _, inp := range mgr.corpus {
+		maxSignal.Merge(inp.Signal.Deserialize())
+	}
 	calls := make(map[int][]int)
-	for s := range mgr.maxSignal {
+	for s := range maxSignal {
 		id, errno := prog.DecodeFallbackSignal(uint32(s))
 		calls[id] = append(calls[id], errno)
 	}
diff --git a/syz-manager/manager.go b/syz-manager/manager.go
index 37f436c29..943df769f 100644
--- a/syz-manager/manager.go
+++ b/syz-manager/manager.go
@@ -57,7 +57,6 @@ type Manager struct {
 	firstConnect   time.Time
 	fuzzingTime    time.Duration
 	stats          *Stats
-	fuzzerStats    map[string]uint64
 	crashTypes     map[string]bool
 	vmStop         chan bool
 	checkResult    *rpctype.CheckArgs
@@ -74,14 +73,10 @@ type Manager struct {
 	candidates       []rpctype.RPCCandidate // untriaged inputs from corpus and hub
 	disabledHashes   map[string]struct{}
 	corpus           map[string]rpctype.RPCInput
-	corpusCover      cover.Cover
-	corpusSignal     signal.Signal
-	maxSignal        signal.Signal
 	newRepros        [][]byte
 	lastMinCorpus    int
 	memoryLeakFrames map[string]bool
 
-	fuzzers        map[string]*Fuzzer
 	needMoreRepros chan chan bool
 	hubReproQueue  chan *Crash
 	reproRequest   chan chan map[string]bool
@@ -108,12 +103,6 @@ const (
 
 const currentDBVersion = 3
 
-type Fuzzer struct {
-	name         string
-	inputs       []rpctype.RPCInput
-	newMaxSignal signal.Signal
-}
-
 type Crash struct {
 	vmIndex int
 	hub     bool // this crash was created based on a repro from hub
@@ -145,7 +134,7 @@ func main() {
 	RunManager(cfg, target, sysTarget, syscalls)
 }
 
-func RunManager(cfg *mgrconfig.Config, target *prog.Target, sysTarget *targets.Target, syscalls map[int]bool) {
+func RunManager(cfg *mgrconfig.Config, target *prog.Target, sysTarget *targets.Target, syscalls []int) {
 	var vmPool *vm.Pool
 	// Type "none" is a special case for debugging/development when manager
 	// does not start any VMs, but instead you start them manually
@@ -161,11 +150,6 @@ func RunManager(cfg *mgrconfig.Config, target *prog.Target, sysTarget *targets.T
 	crashdir := filepath.Join(cfg.Workdir, "crashes")
 	osutil.MkdirAll(crashdir)
 
-	var enabledSyscalls []int
-	for c := range syscalls {
-		enabledSyscalls = append(enabledSyscalls, c)
-	}
-
 	reporter, err := report.NewReporter(cfg)
 	if err != nil {
 		log.Fatalf("%v", err)
@@ -180,13 +164,11 @@ func RunManager(cfg *mgrconfig.Config, target *prog.Target, sysTarget *targets.T
 		crashdir:         crashdir,
 		startTime:        time.Now(),
 		stats:            new(Stats),
-		fuzzerStats:      make(map[string]uint64),
 		crashTypes:       make(map[string]bool),
-		enabledSyscalls:  enabledSyscalls,
+		enabledSyscalls:  syscalls,
 		corpus:           make(map[string]rpctype.RPCInput),
 		disabledHashes:   make(map[string]struct{}),
 		memoryLeakFrames: make(map[string]bool),
-		fuzzers:          make(map[string]*Fuzzer),
 		fresh:            true,
 		vmStop:           make(chan bool),
 		hubReproQueue:    make(chan *Crash, 10),
@@ -206,13 +188,10 @@ func RunManager(cfg *mgrconfig.Config, target *prog.Target, sysTarget *targets.T
 	mgr.collectUsedFiles()
 
 	// Create RPC server for fuzzers.
-	s, err := rpctype.NewRPCServer(cfg.RPC, mgr)
+	mgr.port, err = startRPCServer(mgr)
 	if err != nil {
 		log.Fatalf("failed to create rpc server: %v", err)
 	}
-	log.Logf(0, "serving rpc on tcp://%v", s.Addr())
-	mgr.port = s.Addr().(*net.TCPAddr).Port
-	go s.Serve()
 
 	if cfg.DashboardAddr != "" {
 		mgr.dash = dashapi.New(cfg.DashboardClient, cfg.DashboardAddr, cfg.DashboardKey)
@@ -232,7 +211,7 @@ func RunManager(cfg *mgrconfig.Config, target *prog.Target, sysTarget *targets.T
 			mgr.fuzzingTime += diff * time.Duration(atomic.LoadUint32(&mgr.numFuzzing))
 			executed := mgr.stats.execTotal.get()
 			crashes := mgr.stats.crashes.get()
-			signal := mgr.corpusSignal.Len()
+			signal := mgr.stats.corpusSignal.get()
 			mgr.mu.Unlock()
 			numReproducing := atomic.LoadUint32(&mgr.numReproducing)
 			numFuzzing := atomic.LoadUint32(&mgr.numFuzzing)
@@ -250,7 +229,7 @@ func RunManager(cfg *mgrconfig.Config, target *prog.Target, sysTarget *targets.T
 		go func() {
 			for {
 				time.Sleep(time.Minute)
-				vals := make(map[string]uint64)
+				vals := mgr.stats.all()
 				mgr.mu.Lock()
 				if mgr.firstConnect.IsZero() {
 					mgr.mu.Unlock()
@@ -260,15 +239,7 @@ func RunManager(cfg *mgrconfig.Config, target *prog.Target, sysTarget *targets.T
 				vals["corpus"] = uint64(len(mgr.corpus))
 				vals["uptime"] = uint64(time.Since(mgr.firstConnect)) / 1e9
 				vals["fuzzing"] = uint64(mgr.fuzzingTime) / 1e9
-				vals["signal"] = uint64(mgr.corpusSignal.Len())
-				vals["coverage"] = uint64(len(mgr.corpusCover))
-				for k, v := range mgr.fuzzerStats {
-					vals[k] = v
-				}
 				mgr.mu.Unlock()
-				for k, v := range mgr.stats.all() {
-					vals[k] = v
-				}
 
 				data, err := json.MarshalIndent(vals, "", "  ")
 				if err != nil {
@@ -914,40 +885,25 @@ func (mgr *Manager) minimizeCorpus() {
 	mgr.corpusDB.BumpVersion(currentDBVersion)
 }
 
-func (mgr *Manager) Connect(a *rpctype.ConnectArgs, r *rpctype.ConnectRes) error {
-	log.Logf(1, "fuzzer %v connected", a.Name)
-	mgr.stats.vmRestarts.inc()
+func (mgr *Manager) fuzzerConnect() ([]rpctype.RPCInput, [][]byte) {
 	mgr.mu.Lock()
 	defer mgr.mu.Unlock()
 
-	f := &Fuzzer{
-		name: a.Name,
-	}
-	mgr.fuzzers[a.Name] = f
 	mgr.minimizeCorpus()
-	f.newMaxSignal = mgr.maxSignal.Copy()
-	f.inputs = make([]rpctype.RPCInput, 0, len(mgr.corpus))
+	corpus := make([]rpctype.RPCInput, 0, len(mgr.corpus))
 	for _, inp := range mgr.corpus {
-		f.inputs = append(f.inputs, inp)
+		corpus = append(corpus, inp)
 	}
-	r.MemoryLeakFrames = make([][]byte, 0, len(mgr.memoryLeakFrames))
+	memoryLeakFrames := make([][]byte, 0, len(mgr.memoryLeakFrames))
 	for frame := range mgr.memoryLeakFrames {
-		r.MemoryLeakFrames = append(r.MemoryLeakFrames, []byte(frame))
+		memoryLeakFrames = append(memoryLeakFrames, []byte(frame))
 	}
-	r.EnabledCalls = mgr.enabledSyscalls
-	r.CheckResult = mgr.checkResult
-	r.GitRevision = sys.GitRevision
-	r.TargetRevision = mgr.target.Revision
-	return nil
+	return corpus, memoryLeakFrames
 }
 
-func (mgr *Manager) Check(a *rpctype.CheckArgs, r *int) error {
+func (mgr *Manager) machineChecked(a *rpctype.CheckArgs) {
 	mgr.mu.Lock()
 	defer mgr.mu.Unlock()
-
-	if mgr.checkResult != nil {
-		return nil
-	}
 	if len(mgr.cfg.EnabledSyscalls) != 0 && len(a.DisabledCalls[mgr.cfg.Sandbox]) != 0 {
 		disabled := make(map[string]string)
 		for _, dc := range a.DisabledCalls[mgr.cfg.Sandbox] {
@@ -969,114 +925,42 @@ func (mgr *Manager) Check(a *rpctype.CheckArgs, r *int) error {
 	for _, feat := range a.Features {
 		log.Logf(0, "%-24v: %v", feat.Name, feat.Reason)
 	}
-	a.DisabledCalls = nil
 	mgr.checkResult = a
 	mgr.loadCorpus()
 	mgr.firstConnect = time.Now()
-	return nil
 }
 
-func (mgr *Manager) NewInput(a *rpctype.NewInputArgs, r *int) error {
-	inputSignal := a.Signal.Deserialize()
-	log.Logf(4, "new input from %v for syscall %v (signal=%v, cover=%v)",
-		a.Name, a.Call, inputSignal.Len(), len(a.Cover))
+func (mgr *Manager) newInput(inp rpctype.RPCInput, sign signal.Signal) {
 	mgr.mu.Lock()
 	defer mgr.mu.Unlock()
-
-	f := mgr.fuzzers[a.Name]
-	if f == nil {
-		log.Fatalf("fuzzer %v is not connected", a.Name)
-	}
-
-	if _, err := mgr.target.Deserialize(a.RPCInput.Prog, prog.NonStrict); err != nil {
-		// This should not happen, but we see such cases episodically, reason unknown.
-		log.Logf(0, "failed to deserialize program from fuzzer: %v\n%s", err, a.RPCInput.Prog)
-		return nil
-	}
-	if mgr.corpusSignal.Diff(inputSignal).Empty() {
-		return nil
-	}
-	mgr.stats.newInputs.inc()
-	mgr.corpusSignal.Merge(inputSignal)
-	mgr.corpusCover.Merge(a.Cover)
-	sig := hash.String(a.RPCInput.Prog)
-	if inp, ok := mgr.corpus[sig]; ok {
+	sig := hash.String(inp.Prog)
+	if old, ok := mgr.corpus[sig]; ok {
 		// The input is already present, but possibly with diffent signal/coverage/call.
-		inputSignal.Merge(inp.Signal.Deserialize())
-		inp.Signal = inputSignal.Serialize()
-		var inputCover cover.Cover
-		inputCover.Merge(inp.Cover)
-		inputCover.Merge(a.RPCInput.Cover)
-		inp.Cover = inputCover.Serialize()
-		mgr.corpus[sig] = inp
+		sign.Merge(old.Signal.Deserialize())
+		old.Signal = sign.Serialize()
+		var cov cover.Cover
+		cov.Merge(old.Cover)
+		cov.Merge(inp.Cover)
+		old.Cover = cov.Serialize()
+		mgr.corpus[sig] = old
 	} else {
-		mgr.corpus[sig] = a.RPCInput
-		mgr.corpusDB.Save(sig, a.RPCInput.Prog, 0)
+		mgr.corpus[sig] = inp
+		mgr.corpusDB.Save(sig, inp.Prog, 0)
 		if err := mgr.corpusDB.Flush(); err != nil {
 			log.Logf(0, "failed to save corpus database: %v", err)
 		}
-		for _, f1 := range mgr.fuzzers {
-			if f1 == f {
-				continue
-			}
-			inp := a.RPCInput
-			inp.Cover = nil // Don't send coverage back to all fuzzers.
-			f1.inputs = append(f1.inputs, inp)
-		}
 	}
-	return nil
 }
 
-func (mgr *Manager) Poll(a *rpctype.PollArgs, r *rpctype.PollRes) error {
+func (mgr *Manager) candidateBatch(size int) []rpctype.RPCCandidate {
 	mgr.mu.Lock()
 	defer mgr.mu.Unlock()
-
-	for k, v := range a.Stats {
-		switch k {
-		case "exec total":
-			mgr.stats.execTotal.add(int(v))
-		default:
-			mgr.fuzzerStats[k] += v
-		}
-	}
-
-	f := mgr.fuzzers[a.Name]
-	if f == nil {
-		log.Fatalf("fuzzer %v is not connected", a.Name)
-	}
-	newMaxSignal := mgr.maxSignal.Diff(a.MaxSignal.Deserialize())
-	if !newMaxSignal.Empty() {
-		mgr.maxSignal.Merge(newMaxSignal)
-		for _, f1 := range mgr.fuzzers {
-			if f1 == f {
-				continue
-			}
-			f1.newMaxSignal.Merge(newMaxSignal)
-		}
-	}
-	r.MaxSignal = f.newMaxSignal.Split(500).Serialize()
-	maxInputs := 5
-	if maxInputs < mgr.cfg.Procs {
-		maxInputs = mgr.cfg.Procs
-	}
-	if a.NeedCandidates {
-		for i := 0; i < maxInputs && len(mgr.candidates) > 0; i++ {
-			last := len(mgr.candidates) - 1
-			r.Candidates = append(r.Candidates, mgr.candidates[last])
-			mgr.candidates[last] = rpctype.RPCCandidate{}
-			mgr.candidates = mgr.candidates[:last]
-		}
-	}
-	if len(r.Candidates) == 0 {
-		for i := 0; i < maxInputs && len(f.inputs) > 0; i++ {
-			last := len(f.inputs) - 1
-			r.NewInputs = append(r.NewInputs, f.inputs[last])
-			f.inputs[last] = rpctype.RPCInput{}
-			f.inputs = f.inputs[:last]
-		}
-		if len(f.inputs) == 0 {
-			f.inputs = nil
-		}
+	var res []rpctype.RPCCandidate
+	for i := 0; i < size && len(mgr.candidates) > 0; i++ {
+		last := len(mgr.candidates) - 1
+		res = append(res, mgr.candidates[last])
+		mgr.candidates[last] = rpctype.RPCCandidate{}
+		mgr.candidates = mgr.candidates[:last]
 	}
 	if len(mgr.candidates) == 0 {
 		mgr.candidates = nil
@@ -1091,9 +975,7 @@ func (mgr *Manager) Poll(a *rpctype.PollArgs, r *rpctype.PollRes) error {
 			mgr.phase = phaseTriagedHub
 		}
 	}
-	log.Logf(4, "poll from %v: candidates=%v inputs=%v maxsignal=%v",
-		a.Name, len(r.Candidates), len(r.NewInputs), len(r.MaxSignal.Elems))
-	return nil
+	return res
 }
 
 func (mgr *Manager) collectUsedFiles() {
@@ -1157,7 +1039,7 @@ func (mgr *Manager) dashboardReporter() {
 			Addr:        webAddr,
 			UpTime:      time.Since(mgr.firstConnect),
 			Corpus:      uint64(len(mgr.corpus)),
-			Cover:       uint64(mgr.corpusSignal.Len()),
+			Cover:       mgr.stats.corpusSignal.get(),
 			FuzzingTime: mgr.fuzzingTime - lastFuzzingTime,
 			Crashes:     crashes - lastCrashes,
 			Execs:       execs - lastExecs,
diff --git a/syz-manager/rpc.go b/syz-manager/rpc.go
new file mode 100644
index 000000000..68d270ea1
--- /dev/null
+++ b/syz-manager/rpc.go
@@ -0,0 +1,175 @@
+// Copyright 2018 syzkaller project authors. All rights reserved.
+// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
+
+package main
+
+import (
+	"net"
+	"sync"
+
+	"github.com/google/syzkaller/pkg/cover"
+	"github.com/google/syzkaller/pkg/log"
+	"github.com/google/syzkaller/pkg/rpctype"
+	"github.com/google/syzkaller/pkg/signal"
+	"github.com/google/syzkaller/prog"
+	"github.com/google/syzkaller/sys"
+)
+
+type RPCServer struct {
+	mgr             RPCManagerView
+	target          *prog.Target
+	enabledSyscalls []int
+	stats           *Stats
+	batchSize       int
+
+	mu           sync.Mutex
+	fuzzers      map[string]*Fuzzer
+	checkResult  *rpctype.CheckArgs
+	maxSignal    signal.Signal
+	corpusSignal signal.Signal
+	corpusCover  cover.Cover
+}
+
+type Fuzzer struct {
+	name         string
+	inputs       []rpctype.RPCInput
+	newMaxSignal signal.Signal
+}
+
+// RPCManagerView restricts interface between RPCServer and Manager.
+type RPCManagerView interface {
+	fuzzerConnect() ([]rpctype.RPCInput, [][]byte)
+	machineChecked(result *rpctype.CheckArgs)
+	newInput(inp rpctype.RPCInput, sign signal.Signal)
+	candidateBatch(size int) []rpctype.RPCCandidate
+}
+
+func startRPCServer(mgr *Manager) (int, error) {
+	serv := &RPCServer{
+		mgr:             mgr,
+		target:          mgr.target,
+		enabledSyscalls: mgr.enabledSyscalls,
+		stats:           mgr.stats,
+		fuzzers:         make(map[string]*Fuzzer),
+	}
+	serv.batchSize = 5
+	if serv.batchSize < mgr.cfg.Procs {
+		serv.batchSize = mgr.cfg.Procs
+	}
+	s, err := rpctype.NewRPCServer(mgr.cfg.RPC, "Manager", serv)
+	if err != nil {
+		return 0, err
+	}
+	log.Logf(0, "serving rpc on tcp://%v", s.Addr())
+	port := s.Addr().(*net.TCPAddr).Port
+	go s.Serve()
+	return port, nil
+}
+
+func (serv *RPCServer) Connect(a *rpctype.ConnectArgs, r *rpctype.ConnectRes) error {
+	log.Logf(1, "fuzzer %v connected", a.Name)
+	serv.stats.vmRestarts.inc()
+
+	corpus, memoryLeakFrames := serv.mgr.fuzzerConnect()
+
+	serv.mu.Lock()
+	defer serv.mu.Unlock()
+
+	serv.fuzzers[a.Name] = &Fuzzer{
+		name:         a.Name,
+		inputs:       corpus,
+		newMaxSignal: serv.maxSignal.Copy(),
+	}
+	r.MemoryLeakFrames = memoryLeakFrames
+	r.EnabledCalls = serv.enabledSyscalls
+	r.CheckResult = serv.checkResult
+	r.GitRevision = sys.GitRevision
+	r.TargetRevision = serv.target.Revision
+	return nil
+}
+
+func (serv *RPCServer) Check(a *rpctype.CheckArgs, r *int) error {
+	serv.mu.Lock()
+	defer serv.mu.Unlock()
+
+	if serv.checkResult != nil {
+		return nil
+	}
+	a.DisabledCalls = nil
+	serv.checkResult = a
+	serv.mgr.machineChecked(a)
+	return nil
+}
+
+func (serv *RPCServer) NewInput(a *rpctype.NewInputArgs, r *int) error {
+	inputSignal := a.Signal.Deserialize()
+	log.Logf(4, "new input from %v for syscall %v (signal=%v, cover=%v)",
+		a.Name, a.Call, inputSignal.Len(), len(a.Cover))
+	if _, err := serv.target.Deserialize(a.RPCInput.Prog, prog.NonStrict); err != nil {
+		// This should not happen, but we see such cases episodically, reason unknown.
+		log.Logf(0, "failed to deserialize program from fuzzer: %v\n%s", err, a.RPCInput.Prog)
+		return nil
+	}
+	serv.mu.Lock()
+	defer serv.mu.Unlock()
+
+	if serv.corpusSignal.Diff(inputSignal).Empty() {
+		return nil
+	}
+	serv.mgr.newInput(a.RPCInput, inputSignal)
+
+	serv.stats.newInputs.inc()
+	serv.corpusSignal.Merge(inputSignal)
+	serv.stats.corpusSignal.set(serv.corpusSignal.Len())
+	serv.corpusCover.Merge(a.Cover)
+	serv.stats.corpusCover.set(len(serv.corpusCover))
+
+	a.RPCInput.Cover = nil // Don't send coverage back to all fuzzers.
+	for _, f := range serv.fuzzers {
+		if f.name == a.Name {
+			continue
+		}
+		f.inputs = append(f.inputs, a.RPCInput)
+	}
+	return nil
+}
+
+func (serv *RPCServer) Poll(a *rpctype.PollArgs, r *rpctype.PollRes) error {
+	serv.stats.mergeNamed(a.Stats)
+
+	serv.mu.Lock()
+	defer serv.mu.Unlock()
+
+	f := serv.fuzzers[a.Name]
+	if f == nil {
+		log.Fatalf("fuzzer %v is not connected", a.Name)
+	}
+	newMaxSignal := serv.maxSignal.Diff(a.MaxSignal.Deserialize())
+	if !newMaxSignal.Empty() {
+		serv.maxSignal.Merge(newMaxSignal)
+		for _, f1 := range serv.fuzzers {
+			if f1 == f {
+				continue
+			}
+			f1.newMaxSignal.Merge(newMaxSignal)
+		}
+	}
+	r.MaxSignal = f.newMaxSignal.Split(500).Serialize()
+	if a.NeedCandidates {
+		r.Candidates = serv.mgr.candidateBatch(serv.batchSize)
+	}
+	if len(r.Candidates) == 0 {
+		for i := 0; i < serv.batchSize && len(f.inputs) > 0; i++ {
+			last := len(f.inputs) - 1
+			r.NewInputs = append(r.NewInputs, f.inputs[last])
+			f.inputs[last] = rpctype.RPCInput{}
+			f.inputs = f.inputs[:last]
+		}
+		if len(f.inputs) == 0 {
+			f.inputs = nil
+		}
+	}
+	log.Logf(4, "poll from %v: candidates=%v inputs=%v maxsignal=%v",
+		a.Name, len(r.Candidates), len(r.NewInputs), len(r.MaxSignal.Elems))
+	return nil
+}
diff --git a/syz-manager/stats.go b/syz-manager/stats.go
index c6098b832..6c48a2047 100644
--- a/syz-manager/stats.go
+++ b/syz-manager/stats.go
@@ -4,6 +4,7 @@
 package main
 
 import (
+	"sync"
 	"sync/atomic"
 )
 
@@ -23,10 +24,15 @@ type Stats struct {
 	hubRecvProgDrop  Stat
 	hubRecvRepro     Stat
 	hubRecvReproDrop Stat
+	corpusCover      Stat
+	corpusSignal     Stat
+
+	mu         sync.Mutex
+	namedStats map[string]uint64
 }
 
 func (stats *Stats) all() map[string]uint64 {
-	return map[string]uint64{
+	m := map[string]uint64{
 		"crashes":              stats.crashes.get(),
 		"crash types":          stats.crashTypes.get(),
 		"suppressed":           stats.crashSuppressed.get(),
@@ -40,6 +46,30 @@ func (stats *Stats) all() map[string]uint64 {
 		"hub: recv prog drop":  stats.hubRecvProgDrop.get(),
 		"hub: recv repro":      stats.hubRecvRepro.get(),
 		"hub: recv repro drop": stats.hubRecvReproDrop.get(),
+		"cover":                stats.corpusCover.get(),
+		"signal":               stats.corpusSignal.get(),
+	}
+	stats.mu.Lock()
+	defer stats.mu.Unlock()
+	for k, v := range stats.namedStats {
+		m[k] = v
+	}
+	return m
+}
+
+func (stats *Stats) mergeNamed(named map[string]uint64) {
+	stats.mu.Lock()
+	defer stats.mu.Unlock()
+	if stats.namedStats == nil {
+		stats.namedStats = make(map[string]uint64)
+	}
+	for k, v := range named {
+		switch k {
+		case "exec total":
+			stats.execTotal.add(int(v))
+		default:
+			stats.namedStats[k] += v
+		}
 	}
 }
 
@@ -54,3 +84,7 @@ func (s *Stat) inc() {
 func (s *Stat) add(v int) {
 	atomic.AddUint64((*uint64)(s), uint64(v))
 }
+
+func (s *Stat) set(v int) {
+	atomic.StoreUint64((*uint64)(s), uint64(v))
+}
diff --git a/tools/syz-mutate/mutate.go b/tools/syz-mutate/mutate.go
index 63bc47ef5..b0361a20b 100644
--- a/tools/syz-mutate/mutate.go
+++ b/tools/syz-mutate/mutate.go
@@ -36,13 +36,14 @@ func main() {
 	}
 	var syscalls map[*prog.Syscall]bool
 	if *flagEnable != "" {
-		syscallsIDs, err := mgrconfig.ParseEnabledSyscalls(target, strings.Split(*flagEnable, ","), nil)
+		enabled := strings.Split(*flagEnable, ",")
+		syscallsIDs, err := mgrconfig.ParseEnabledSyscalls(target, enabled, nil)
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "failed to parse enabled syscalls: %v", err)
 			os.Exit(1)
 		}
 		syscalls = make(map[*prog.Syscall]bool)
-		for id := range syscallsIDs {
+		for _, id := range syscallsIDs {
 			syscalls[target.Syscalls[id]] = true
 		}
 		var disabled map[*prog.Syscall]string
diff --git a/tools/syz-runtest/runtest.go b/tools/syz-runtest/runtest.go
index b21852299..54be07703 100644
--- a/tools/syz-runtest/runtest.go
+++ b/tools/syz-runtest/runtest.go
@@ -71,7 +71,7 @@ func main() {
 		reqMap:           make(map[int]*runtest.RunRequest),
 		lastReq:          make(map[string]int),
 	}
-	s, err := rpctype.NewRPCServer(cfg.RPC, mgr)
+	s, err := rpctype.NewRPCServer(cfg.RPC, "Manager", mgr)
 	if err != nil {
 		log.Fatalf("failed to create rpc server: %v", err)
 	}