all: move fuzzer to the host

Instead of doing fuzzing in parallel in running VM, make all decisions
in the host syz-manager process.

Instantiate and keep a fuzzer.Fuzzer object in syz-manager and update
the RPC between syz-manager and syz-fuzzer to exchange exact programs to
execute and their resulting signal and coverage.

To optimize the networking traffic, exchange mostly only the difference
between the known max signal and the detected signal.

18 files changed, 683 insertions, 639 deletions

diff --git a/pkg/corpus/corpus.go b/pkg/corpus/corpus.go
index 772037178..8eed1fc63 100644
--- a/pkg/corpus/corpus.go
+++ b/pkg/corpus/corpus.go
@@ -9,7 +9,6 @@ import (
 
 	"github.com/google/syzkaller/pkg/cover"
 	"github.com/google/syzkaller/pkg/hash"
-	"github.com/google/syzkaller/pkg/rpctype"
 	"github.com/google/syzkaller/pkg/signal"
 	"github.com/google/syzkaller/prog"
 )
@@ -63,15 +62,6 @@ func (item Item) StringCall() string {
 	return stringCall(item.Prog, item.Call)
 }
 
-// RPCInputShort() does not include coverage.
-func (item Item) RPCInputShort() rpctype.Input {
-	return rpctype.Input{
-		Call:   item.Call,
-		Prog:   item.ProgData,
-		Signal: item.Signal.Serialize(),
-	}
-}
-
 func stringCall(p *prog.Prog, call int) string {
 	if call != -1 {
 		return p.Calls[call].Meta.Name
@@ -91,16 +81,6 @@ func (item NewInput) StringCall() string {
 	return stringCall(item.Prog, item.Call)
 }
 
-func (item NewInput) RPCInput() rpctype.Input {
-	return rpctype.Input{
-		Call:     item.Call,
-		Prog:     item.Prog.Serialize(),
-		Signal:   item.Signal.Serialize(),
-		Cover:    item.Cover,
-		RawCover: item.RawCover,
-	}
-}
-
 type NewItemEvent struct {
 	Sig      string
 	Exists   bool
diff --git a/pkg/cover/canonicalizer.go b/pkg/cover/canonicalizer.go
index c7c385aed..d3b014af8 100644
--- a/pkg/cover/canonicalizer.go
+++ b/pkg/cover/canonicalizer.go
@@ -9,7 +9,6 @@ import (
 
 	"github.com/google/syzkaller/pkg/host"
 	"github.com/google/syzkaller/pkg/log"
-	"github.com/google/syzkaller/pkg/signal"
 )
 
 type Canonicalizer struct {
@@ -120,18 +119,18 @@ func (can *Canonicalizer) NewInstance(modules []host.KernelModule) *Canonicalize
 	}
 }
 
-func (ci *CanonicalizerInstance) Canonicalize(cov []uint32, sign signal.Serial) ([]uint32, signal.Serial) {
+func (ci *CanonicalizerInstance) Canonicalize(elems []uint32) []uint32 {
 	if ci.canonical.moduleKeys == nil {
-		return cov, sign
+		return elems
 	}
-	return ci.canonicalize.convertPCs(cov, sign)
+	return ci.canonicalize.convertPCs(elems)
 }
 
-func (ci *CanonicalizerInstance) Decanonicalize(cov []uint32, sign signal.Serial) ([]uint32, signal.Serial) {
+func (ci *CanonicalizerInstance) Decanonicalize(elems []uint32) []uint32 {
 	if ci.canonical.moduleKeys == nil {
-		return cov, sign
+		return elems
 	}
-	return ci.decanonicalize.convertPCs(cov, sign)
+	return ci.decanonicalize.convertPCs(elems)
 }
 
 func (ci *CanonicalizerInstance) DecanonicalizeFilter(bitmap map[uint32]uint32) map[uint32]uint32 {
@@ -177,34 +176,21 @@ func findModule(pc uint32, moduleKeys []uint32) (moduleIdx int) {
 	return moduleIdx - 1
 }
 
-func (convert *Convert) convertPCs(cov []uint32, sign signal.Serial) ([]uint32, signal.Serial) {
+func (convert *Convert) convertPCs(pcs []uint32) []uint32 {
 	// Convert coverage.
-	var retCov []uint32
+	var ret []uint32
 	convCtx := &convertContext{convert: convert}
-	for _, pc := range cov {
+	for _, pc := range pcs {
 		if newPC, ok := convert.convertPC(pc); ok {
-			retCov = append(retCov, newPC)
+			ret = append(ret, newPC)
 		} else {
 			convCtx.discard(pc)
 		}
 	}
 	if msg := convCtx.discarded(); msg != "" {
-		log.Logf(4, "error in PC conversion: %v", msg)
-	}
-	// Convert signals.
-	retSign := &signal.Serial{}
-	convCtx = &convertContext{convert: convert}
-	for idx, elem := range sign.Elems {
-		if newSign, ok := convert.convertPC(uint32(elem)); ok {
-			retSign.AddElem(newSign, sign.Prios[idx])
-		} else {
-			convCtx.discard(uint32(elem))
-		}
-	}
-	if msg := convCtx.discarded(); msg != "" {
-		log.Logf(4, "error in signal conversion: %v", msg)
+		log.Logf(4, "error in PC/signal conversion: %v", msg)
 	}
-	return retCov, *retSign
+	return ret
 }
 
 func (convert *Convert) convertPC(pc uint32) (uint32, bool) {
diff --git a/pkg/cover/canonicalizer_test.go b/pkg/cover/canonicalizer_test.go
index db418abe1..aa840a96f 100644
--- a/pkg/cover/canonicalizer_test.go
+++ b/pkg/cover/canonicalizer_test.go
@@ -13,7 +13,6 @@ import (
 
 	"github.com/google/syzkaller/pkg/cover"
 	"github.com/google/syzkaller/pkg/host"
-	"github.com/google/syzkaller/pkg/signal"
 )
 
 type RPCServer struct {
@@ -28,8 +27,8 @@ type Fuzzer struct {
 	goalCov     []uint32
 	bitmap      map[uint32]uint32
 	goalBitmap  map[uint32]uint32
-	sign        signal.Serial
-	goalSign    signal.Serial
+	sign        []uint32
+	goalSign    []uint32
 }
 
 type canonicalizeValue int
@@ -49,13 +48,9 @@ func TestNilModules(t *testing.T) {
 
 	serv.fuzzers["f1"].cov = []uint32{0x00010000, 0x00020000, 0x00030000, 0x00040000}
 	serv.fuzzers["f1"].goalCov = []uint32{0x00010000, 0x00020000, 0x00030000, 0x00040000}
-	serv.fuzzers["f1"].sign = signal.FromRaw(serv.fuzzers["f1"].cov, 0).Serialize()
-	serv.fuzzers["f1"].goalSign = signal.FromRaw(serv.fuzzers["f1"].goalCov, 0).Serialize()
 
 	serv.fuzzers["f2"].cov = []uint32{0x00010000, 0x00020000, 0x00030000, 0x00040000}
 	serv.fuzzers["f2"].goalCov = []uint32{0x00010000, 0x00020000, 0x00030000, 0x00040000}
-	serv.fuzzers["f2"].sign = signal.FromRaw(serv.fuzzers["f2"].cov, 0).Serialize()
-	serv.fuzzers["f2"].goalSign = signal.FromRaw(serv.fuzzers["f2"].goalCov, 0).Serialize()
 
 	serv.fuzzers["f1"].bitmap = map[uint32]uint32{
 		0x00010011: 1,
@@ -87,15 +82,15 @@ func TestNilModules(t *testing.T) {
 	}
 
 	serv.fuzzers["f1"].goalCov = []uint32{0x00010000, 0x00020000, 0x00030000, 0x00040000}
-	serv.fuzzers["f1"].goalSign = signal.FromRaw(serv.fuzzers["f1"].goalCov, 0).Serialize()
+	serv.fuzzers["f1"].goalSign = serv.fuzzers["f1"].goalCov
 	serv.fuzzers["f2"].goalCov = []uint32{0x00010000, 0x00020000, 0x00030000, 0x00040000}
-	serv.fuzzers["f2"].goalSign = signal.FromRaw(serv.fuzzers["f2"].goalCov, 0).Serialize()
+	serv.fuzzers["f2"].goalSign = serv.fuzzers["f2"].goalCov
 	if err := serv.runTest(Decanonicalize); err != "" {
 		t.Fatalf("failed in decanonicalization: %v", err)
 	}
 }
 
-// Confirms there is no change to signals if coverage is disabled and fallback signals are used.
+// Confirms there is no change to PCs if coverage is disabled and fallback signals are used.
 func TestDisabledSignals(t *testing.T) {
 	serv := &RPCServer{
 		fuzzers: make(map[string]*Fuzzer),
@@ -112,18 +107,18 @@ func TestDisabledSignals(t *testing.T) {
 	serv.connect("f2", f2Modules, false)
 
 	pcs := []uint32{0x00010000, 0x00020000, 0x00030000, 0x00040000}
-	serv.fuzzers["f1"].sign = signal.FromRaw(pcs, 0).Serialize()
-	serv.fuzzers["f1"].goalSign = signal.FromRaw(pcs, 0).Serialize()
+	serv.fuzzers["f1"].cov = pcs
+	serv.fuzzers["f1"].goalCov = pcs
 
-	serv.fuzzers["f2"].sign = signal.FromRaw(pcs, 0).Serialize()
-	serv.fuzzers["f2"].goalSign = signal.FromRaw(pcs, 0).Serialize()
+	serv.fuzzers["f2"].sign = pcs
+	serv.fuzzers["f2"].goalSign = pcs
 
 	if err := serv.runTest(Canonicalize); err != "" {
 		t.Fatalf("failed in canonicalization: %v", err)
 	}
 
-	serv.fuzzers["f1"].goalSign = signal.FromRaw(pcs, 0).Serialize()
-	serv.fuzzers["f2"].goalSign = signal.FromRaw(pcs, 0).Serialize()
+	serv.fuzzers["f1"].goalSign = pcs
+	serv.fuzzers["f2"].goalSign = pcs
 	if err := serv.runTest(Decanonicalize); err != "" {
 		t.Fatalf("failed in decanonicalization: %v", err)
 	}
@@ -152,8 +147,6 @@ func TestModules(t *testing.T) {
 		0x00035000, 0x00040000, 0x00045000, 0x00050000, 0x00055000}
 	serv.fuzzers["f1"].goalCov = []uint32{0x00010000, 0x00015000, 0x00020000, 0x00025000, 0x00030000,
 		0x00035000, 0x00040000, 0x00045000, 0x00050000, 0x00055000}
-	serv.fuzzers["f1"].sign = signal.FromRaw(serv.fuzzers["f1"].cov, 0).Serialize()
-	serv.fuzzers["f1"].goalSign = signal.FromRaw(serv.fuzzers["f1"].goalCov, 0).Serialize()
 
 	// The modules addresss are inverted between: (2 and 4), (3 and 5),
 	// affecting the output canonical coverage values in these ranges.
@@ -161,8 +154,6 @@ func TestModules(t *testing.T) {
 		0x00035000, 0x00040000, 0x00045000, 0x00050000, 0x00055000}
 	serv.fuzzers["f2"].goalCov = []uint32{0x00010000, 0x00015000, 0x00040000, 0x00025000, 0x00045000,
 		0x0004a000, 0x00020000, 0x00030000, 0x0003b000, 0x00055000}
-	serv.fuzzers["f2"].sign = signal.FromRaw(serv.fuzzers["f2"].cov, 0).Serialize()
-	serv.fuzzers["f2"].goalSign = signal.FromRaw(serv.fuzzers["f2"].goalCov, 0).Serialize()
 
 	serv.fuzzers["f1"].bitmap = map[uint32]uint32{
 		0x00010011: 1,
@@ -195,10 +186,8 @@ func TestModules(t *testing.T) {
 
 	serv.fuzzers["f1"].goalCov = []uint32{0x00010000, 0x00015000, 0x00020000, 0x00025000, 0x00030000,
 		0x00035000, 0x00040000, 0x00045000, 0x00050000, 0x00055000}
-	serv.fuzzers["f1"].goalSign = signal.FromRaw(serv.fuzzers["f1"].goalCov, 0).Serialize()
 	serv.fuzzers["f2"].goalCov = []uint32{0x00010000, 0x00015000, 0x00020000, 0x00025000, 0x00030000,
 		0x00035000, 0x00040000, 0x00045000, 0x00050000, 0x00055000}
-	serv.fuzzers["f2"].goalSign = signal.FromRaw(serv.fuzzers["f2"].goalCov, 0).Serialize()
 	if err := serv.runTest(Decanonicalize); err != "" {
 		t.Fatalf("failed in decanonicalization: %v", err)
 	}
@@ -225,15 +214,12 @@ func TestChangingModules(t *testing.T) {
 	// in this range should be deleted.
 	serv.fuzzers["f2"].cov = []uint32{0x00010000, 0x00015000, 0x00020000, 0x00025000}
 	serv.fuzzers["f2"].goalCov = []uint32{0x00010000, 0x00015000, 0x00025000}
-	serv.fuzzers["f2"].sign = signal.FromRaw(serv.fuzzers["f2"].cov, 0).Serialize()
-	serv.fuzzers["f2"].goalSign = signal.FromRaw(serv.fuzzers["f2"].goalCov, 0).Serialize()
 
 	if err := serv.runTest(Canonicalize); err != "" {
 		t.Fatalf("failed in canonicalization: %v", err)
 	}
 
 	serv.fuzzers["f2"].goalCov = []uint32{0x00010000, 0x00015000, 0x00025000}
-	serv.fuzzers["f2"].goalSign = signal.FromRaw(serv.fuzzers["f2"].goalCov, 0).Serialize()
 	if err := serv.runTest(Decanonicalize); err != "" {
 		t.Fatalf("failed in decanonicalization: %v", err)
 	}
@@ -241,12 +227,11 @@ func TestChangingModules(t *testing.T) {
 
 func (serv *RPCServer) runTest(val canonicalizeValue) string {
 	var cov []uint32
-	var sign signal.Serial
 	for name, fuzzer := range serv.fuzzers {
 		if val == Canonicalize {
-			cov, sign = fuzzer.instModules.Canonicalize(fuzzer.cov, fuzzer.sign)
+			cov = fuzzer.instModules.Canonicalize(fuzzer.cov)
 		} else {
-			cov, sign = fuzzer.instModules.Decanonicalize(fuzzer.cov, fuzzer.sign)
+			cov = fuzzer.instModules.Decanonicalize(fuzzer.cov)
 			instBitmap := fuzzer.instModules.DecanonicalizeFilter(fuzzer.bitmap)
 			if !reflect.DeepEqual(instBitmap, fuzzer.goalBitmap) {
 				return fmt.Sprintf("failed in bitmap conversion. Fuzzer %v.\nExpected: 0x%x.\nReturned: 0x%x",
@@ -257,12 +242,7 @@ func (serv *RPCServer) runTest(val canonicalizeValue) string {
 			return fmt.Sprintf("failed in coverage conversion. Fuzzer %v.\nExpected: 0x%x.\nReturned: 0x%x",
 				name, fuzzer.goalCov, cov)
 		}
-		if !reflect.DeepEqual(sign.Deserialize(), fuzzer.goalSign.Deserialize()) {
-			return fmt.Sprintf("failed in signal conversion. Fuzzer %v.\nExpected: 0x%x.\nReturned: 0x%x",
-				name, fuzzer.goalSign, sign)
-		}
 		fuzzer.cov = cov
-		fuzzer.sign = sign
 	}
 	return ""
 }
diff --git a/pkg/fuzzer/cover.go b/pkg/fuzzer/cover.go
index 886da004d..5af00f167 100644
--- a/pkg/fuzzer/cover.go
+++ b/pkg/fuzzer/cover.go
@@ -35,6 +35,12 @@ func (cover *Cover) addRawMaxSignal(signal []uint32, prio uint8) signal.Signal {
 	return diff
 }
 
+func (cover *Cover) CopyMaxSignal() signal.Signal {
+	cover.mu.RLock()
+	defer cover.mu.RUnlock()
+	return cover.maxSignal.Copy()
+}
+
 func (cover *Cover) GrabNewSignal() signal.Signal {
 	cover.mu.Lock()
 	defer cover.mu.Unlock()
diff --git a/pkg/fuzzer/fuzzer.go b/pkg/fuzzer/fuzzer.go
index 196e59506..14c3f5902 100644
--- a/pkg/fuzzer/fuzzer.go
+++ b/pkg/fuzzer/fuzzer.go
@@ -13,15 +13,15 @@ import (
 	"time"
 
 	"github.com/google/syzkaller/pkg/corpus"
-	"github.com/google/syzkaller/pkg/hash"
 	"github.com/google/syzkaller/pkg/ipc"
+	"github.com/google/syzkaller/pkg/rpctype"
+	"github.com/google/syzkaller/pkg/signal"
 	"github.com/google/syzkaller/prog"
 )
 
 type Fuzzer struct {
-	Config         *Config
-	Cover          *Cover
-	NeedCandidates chan struct{}
+	Config *Config
+	Cover  *Cover
 
 	ctx    context.Context
 	mu     sync.Mutex
@@ -39,18 +39,16 @@ type Fuzzer struct {
 
 	runningJobs      atomic.Int64
 	queuedCandidates atomic.Int64
-	// If the source of candidates runs out of them, we risk
-	// generating too many needCandidate requests (one for
-	// each Config.MinCandidates). We prevent this with candidatesRequested.
-	candidatesRequested atomic.Bool
+
+	outOfQueue     atomic.Bool
+	outOfQueueNext atomic.Int64
 }
 
 func NewFuzzer(ctx context.Context, cfg *Config, rnd *rand.Rand,
 	target *prog.Target) *Fuzzer {
 	f := &Fuzzer{
-		Config:         cfg,
-		Cover:          &Cover{},
-		NeedCandidates: make(chan struct{}, 1),
+		Config: cfg,
+		Cover:  &Cover{},
 
 		ctx:    ctx,
 		stats:  map[string]uint64{},
@@ -82,18 +80,16 @@ type Config struct {
 	EnabledCalls   map[*prog.Syscall]bool
 	NoMutateCalls  map[int]bool
 	FetchRawCover  bool
-	// If the number of queued candidates is less than MinCandidates,
-	// NeedCandidates is triggered.
-	MinCandidates uint
-	NewInputs     chan corpus.NewInput
+	NewInputFilter func(input *corpus.NewInput) bool
 }
 
 type Request struct {
 	Prog         *prog.Prog
 	NeedCover    bool
 	NeedRawCover bool
-	NeedSignal   bool
+	NeedSignal   rpctype.SignalType
 	NeedHints    bool
+	SignalFilter signal.Signal // If specified, the resulting signal MAY be a subset of it.
 	// Fields that are only relevant within pkg/fuzzer.
 	flags   ProgTypes
 	stat    string
@@ -110,7 +106,7 @@ func (fuzzer *Fuzzer) Done(req *Request, res *Result) {
 	// Triage individual calls.
 	// We do it before unblocking the waiting threads because
 	// it may result it concurrent modification of req.Prog.
-	if req.NeedSignal && res.Info != nil {
+	if req.NeedSignal != rpctype.NoSignal && res.Info != nil {
 		for call, info := range res.Info.Calls {
 			fuzzer.triageProgCall(req.Prog, &info, call, req.flags)
 		}
@@ -166,7 +162,6 @@ func signalPrio(p *prog.Prog, info *ipc.CallInfo, call int) (prio uint8) {
 
 type Candidate struct {
 	Prog      *prog.Prog
-	Hash      hash.Sig
 	Smashed   bool
 	Minimized bool
 }
@@ -178,20 +173,19 @@ func (fuzzer *Fuzzer) NextInput() *Request {
 			panic("queuedCandidates is out of sync")
 		}
 	}
-	if fuzzer.NeedCandidatesNow() &&
-		!fuzzer.candidatesRequested.CompareAndSwap(false, true) {
-		select {
-		case fuzzer.NeedCandidates <- struct{}{}:
-		default:
-		}
-	}
 	return req
 }
 
 func (fuzzer *Fuzzer) nextInput() *Request {
-	nextExec := fuzzer.nextExec.tryPop()
-	if nextExec != nil {
-		return nextExec.value
+	// The fuzzer may get biased to one specific part of the kernel.
+	// Periodically generate random programs to ensure that the coverage
+	// is more uniform.
+	if !fuzzer.outOfQueue.Load() ||
+		fuzzer.outOfQueueNext.Add(1)%400 > 0 {
+		nextExec := fuzzer.nextExec.tryPop()
+		if nextExec != nil {
+			return nextExec.value
+		}
 	}
 	// Either generate a new input or mutate an existing one.
 	mutateRate := 0.95
@@ -227,16 +221,15 @@ func (fuzzer *Fuzzer) Logf(level int, msg string, args ...interface{}) {
 	fuzzer.Config.Logf(level, msg, args...)
 }
 
-func (fuzzer *Fuzzer) NeedCandidatesNow() bool {
-	return fuzzer.queuedCandidates.Load() < int64(fuzzer.Config.MinCandidates)
-}
-
 func (fuzzer *Fuzzer) AddCandidates(candidates []Candidate) {
 	fuzzer.queuedCandidates.Add(int64(len(candidates)))
 	for _, candidate := range candidates {
 		fuzzer.pushExec(candidateRequest(candidate), priority{candidatePrio})
 	}
-	fuzzer.candidatesRequested.Store(false)
+}
+
+func (fuzzer *Fuzzer) EnableOutOfQueue() {
+	fuzzer.outOfQueue.Store(true)
 }
 
 func (fuzzer *Fuzzer) rand() *rand.Rand {
@@ -250,7 +243,7 @@ func (fuzzer *Fuzzer) pushExec(req *Request, prio priority) {
 	if req.stat == "" {
 		panic("Request.Stat field must be set")
 	}
-	if req.NeedHints && (req.NeedCover || req.NeedSignal) {
+	if req.NeedHints && (req.NeedCover || req.NeedSignal != rpctype.NoSignal) {
 		panic("Request.NeedHints is mutually exclusive with other fields")
 	}
 	fuzzer.nextExec.push(&priorityQueueItem[*Request]{
@@ -330,19 +323,3 @@ func (fuzzer *Fuzzer) logCurrentStats() {
 		fuzzer.Logf(0, "%s", str)
 	}
 }
-
-type Stats struct {
-	CoverStats
-	corpus.Stats
-	Candidates  int
-	RunningJobs int
-}
-
-func (fuzzer *Fuzzer) Stats() Stats {
-	return Stats{
-		CoverStats:  fuzzer.Cover.Stats(),
-		Stats:       fuzzer.Config.Corpus.Stats(),
-		Candidates:  int(fuzzer.queuedCandidates.Load()),
-		RunningJobs: int(fuzzer.runningJobs.Load()),
-	}
-}
diff --git a/pkg/fuzzer/fuzzer_test.go b/pkg/fuzzer/fuzzer_test.go
index 4f8cf41c5..bd6d9a8fe 100644
--- a/pkg/fuzzer/fuzzer_test.go
+++ b/pkg/fuzzer/fuzzer_test.go
@@ -22,6 +22,7 @@ import (
 	"github.com/google/syzkaller/pkg/csource"
 	"github.com/google/syzkaller/pkg/ipc"
 	"github.com/google/syzkaller/pkg/ipc/ipcconfig"
+	"github.com/google/syzkaller/pkg/rpctype"
 	"github.com/google/syzkaller/pkg/testutil"
 	"github.com/google/syzkaller/prog"
 	"github.com/google/syzkaller/sys/targets"
@@ -54,7 +55,6 @@ func TestFuzz(t *testing.T) {
 		EnabledCalls: map[*prog.Syscall]bool{
 			target.SyscallMap["syz_test_fuzzer1"]: true,
 		},
-		NewInputs: make(chan corpus.NewInput),
 	}, rand.New(testutil.RandSource(t)), target)
 
 	go func() {
@@ -129,7 +129,7 @@ func emulateExec(req *Request) (*Result, string, error) {
 		if req.NeedCover {
 			callInfo.Cover = []uint32{cover}
 		}
-		if req.NeedSignal {
+		if req.NeedSignal != rpctype.NoSignal {
 			callInfo.Signal = []uint32{cover}
 		}
 		info.Calls = append(info.Calls, callInfo)
@@ -205,7 +205,6 @@ func (f *testFuzzer) wait() {
 	for title, cnt := range f.crashes {
 		t.Logf("%s: %d", title, cnt)
 	}
-	t.Logf("stats:\n%v", f.fuzzer.GrabStats())
 }
 
 // TODO: it's already implemented in syz-fuzzer/proc.go,
@@ -239,7 +238,7 @@ var crashRe = regexp.MustCompile(`{{CRASH: (.*?)}}`)
 func (proc *executorProc) execute(req *Request) (*Result, string, error) {
 	execOpts := proc.execOpts
 	// TODO: it's duplicated from fuzzer.go.
-	if req.NeedSignal {
+	if req.NeedSignal != rpctype.NoSignal {
 		execOpts.Flags |= ipc.FlagCollectSignal
 	}
 	if req.NeedCover {
diff --git a/pkg/fuzzer/job.go b/pkg/fuzzer/job.go
index dfc0b807e..f567fc2cc 100644
--- a/pkg/fuzzer/job.go
+++ b/pkg/fuzzer/job.go
@@ -10,6 +10,7 @@ import (
 	"github.com/google/syzkaller/pkg/corpus"
 	"github.com/google/syzkaller/pkg/cover"
 	"github.com/google/syzkaller/pkg/ipc"
+	"github.com/google/syzkaller/pkg/rpctype"
 	"github.com/google/syzkaller/pkg/signal"
 	"github.com/google/syzkaller/prog"
 )
@@ -64,7 +65,7 @@ func genProgRequest(fuzzer *Fuzzer, rnd *rand.Rand) *Request {
 		fuzzer.ChoiceTable())
 	return &Request{
 		Prog:       p,
-		NeedSignal: true,
+		NeedSignal: rpctype.NewSignal,
 		stat:       statGenerate,
 	}
 }
@@ -83,7 +84,7 @@ func mutateProgRequest(fuzzer *Fuzzer, rnd *rand.Rand) *Request {
 	)
 	return &Request{
 		Prog:       newP,
-		NeedSignal: true,
+		NeedSignal: rpctype.NewSignal,
 		stat:       statFuzz,
 	}
 }
@@ -98,7 +99,7 @@ func candidateRequest(input Candidate) *Request {
 	}
 	return &Request{
 		Prog:       input.Prog,
-		NeedSignal: true,
+		NeedSignal: rpctype.NewSignal,
 		stat:       statCandidate,
 		flags:      flags,
 	}
@@ -157,13 +158,10 @@ func (job *triageJob) run(fuzzer *Fuzzer) {
 		Cover:    info.cover.Serialize(),
 		RawCover: info.rawCover,
 	}
-	fuzzer.Config.Corpus.Save(input)
-	if fuzzer.Config.NewInputs != nil {
-		select {
-		case <-fuzzer.ctx.Done():
-		case fuzzer.Config.NewInputs <- input:
-		}
+	if filter := fuzzer.Config.NewInputFilter; filter != nil && !filter(&input) {
+		return
 	}
+	fuzzer.Config.Corpus.Save(input)
 }
 
 type deflakedCover struct {
@@ -179,7 +177,7 @@ func (job *triageJob) deflake(fuzzer *Fuzzer) (info deflakedCover, stop bool) {
 	for i := 0; i < signalRuns; i++ {
 		result := fuzzer.exec(job, &Request{
 			Prog:         job.p,
-			NeedSignal:   true,
+			NeedSignal:   rpctype.AllSignal,
 			NeedCover:    true,
 			NeedRawCover: fuzzer.Config.FetchRawCover,
 			stat:         statTriage,
@@ -226,9 +224,10 @@ func (job *triageJob) minimize(fuzzer *Fuzzer, newSignal signal.Signal) (stop bo
 			}
 			for i := 0; i < minimizeAttempts; i++ {
 				result := fuzzer.exec(job, &Request{
-					Prog:       p1,
-					NeedSignal: true,
-					stat:       statMinimize,
+					Prog:         p1,
+					NeedSignal:   rpctype.AllSignal,
+					SignalFilter: newSignal,
+					stat:         statMinimize,
 				})
 				if result.Stop {
 					stop = true
@@ -298,7 +297,7 @@ func (job *smashJob) run(fuzzer *Fuzzer) {
 			fuzzer.Config.Corpus.Programs())
 		result := fuzzer.exec(job, &Request{
 			Prog:       p,
-			NeedSignal: true,
+			NeedSignal: rpctype.NewSignal,
 			stat:       statSmash,
 		})
 		if result.Stop {
@@ -386,7 +385,7 @@ func (job *hintsJob) run(fuzzer *Fuzzer) {
 		func(p *prog.Prog) bool {
 			result := fuzzer.exec(job, &Request{
 				Prog:       p,
-				NeedSignal: true,
+				NeedSignal: rpctype.NewSignal,
 				stat:       statHint,
 			})
 			return !result.Stop
diff --git a/pkg/fuzzer/stats.go b/pkg/fuzzer/stats.go
index 17bc6131c..044febc64 100644
--- a/pkg/fuzzer/stats.go
+++ b/pkg/fuzzer/stats.go
@@ -3,6 +3,8 @@
 
 package fuzzer
 
+import "github.com/google/syzkaller/pkg/corpus"
+
 const (
 	statGenerate       = "exec gen"
 	statFuzz           = "exec fuzz"
@@ -17,10 +19,28 @@ const (
 	statBufferTooSmall = "buffer too small"
 )
 
-func (fuzzer *Fuzzer) GrabStats() map[string]uint64 {
+type Stats struct {
+	CoverStats
+	corpus.Stats
+	Candidates  int
+	RunningJobs int
+	// Let's keep stats in Named as long as the rest of the code does not depend
+	// on their specific values.
+	Named map[string]uint64
+}
+
+func (fuzzer *Fuzzer) Stats() Stats {
+	ret := Stats{
+		CoverStats:  fuzzer.Cover.Stats(),
+		Stats:       fuzzer.Config.Corpus.Stats(),
+		Candidates:  int(fuzzer.queuedCandidates.Load()),
+		RunningJobs: int(fuzzer.runningJobs.Load()),
+		Named:       make(map[string]uint64),
+	}
 	fuzzer.mu.Lock()
 	defer fuzzer.mu.Unlock()
-	ret := fuzzer.stats
-	fuzzer.stats = map[string]uint64{}
+	for k, v := range fuzzer.stats {
+		ret.Named[k] = v
+	}
 	return ret
 }
diff --git a/pkg/rpctype/rpctype.go b/pkg/rpctype/rpctype.go
index efd9d6589..9f0b6846e 100644
--- a/pkg/rpctype/rpctype.go
+++ b/pkg/rpctype/rpctype.go
@@ -13,20 +13,47 @@ import (
 	"github.com/google/syzkaller/pkg/signal"
 )
 
-type Input struct {
-	Call     int // seq number of call in the prog to which the item is related (-1 for extra)
-	Prog     []byte
-	Signal   signal.Serial
-	Cover    []uint32
-	RawCover []uint32
+type SignalType int
+
+const (
+	NoSignal  SignalType = 0 // we don't need any signal
+	NewSignal SignalType = 1 // we need the newly seen signal
+	AllSignal SignalType = 2 // we need all signal
+)
+
+// ExecutionRequest describes the task of executing a particular program.
+// Corresponds to Fuzzer.Request.
+type ExecutionRequest struct {
+	ID           int64
+	ProgData     []byte
+	NeedCover    bool
+	NeedRawCover bool
+	NeedHints    bool
+	NeedSignal   SignalType
+	SignalFilter signal.Signal
+}
+
+// ExecutionResult is sent after ExecutionRequest is completed.
+type ExecutionResult struct {
+	ID   int64
+	Info ipc.ProgInfo
+}
+
+// ExchangeInfoRequest is periodically sent by syz-fuzzer to syz-manager.
+type ExchangeInfoRequest struct {
+	Name       string
+	NeedProgs  int
+	StatsDelta map[string]uint64
+	Results    []ExecutionResult
 }
 
-type Candidate struct {
-	Prog      []byte
-	Minimized bool
-	Smashed   bool
+// ExchangeInfoReply is a reply to ExchangeInfoRequest.
+type ExchangeInfoReply struct {
+	Requests     []ExecutionRequest
+	NewMaxSignal []uint32
 }
 
+// TODO: merge ExecutionRequest and ExecTask.
 type ExecTask struct {
 	Prog []byte
 	ID   int64
@@ -40,7 +67,6 @@ type ConnectArgs struct {
 
 type ConnectRes struct {
 	EnabledCalls      []int
-	NoMutateCalls     map[int]bool
 	GitRevision       string
 	TargetRevision    string
 	AllSandboxes      bool
@@ -64,24 +90,6 @@ type SyscallReason struct {
 	Reason string
 }
 
-type NewInputArgs struct {
-	Name string
-	Input
-}
-
-type PollArgs struct {
-	Name           string
-	NeedCandidates bool
-	MaxSignal      signal.Serial
-	Stats          map[string]uint64
-}
-
-type PollRes struct {
-	Candidates []Candidate
-	NewInputs  []Input
-	MaxSignal  signal.Serial
-}
-
 type RunnerConnectArgs struct {
 	Pool, VM int
 }
@@ -199,9 +207,3 @@ type RunTestDoneArgs struct {
 	Info   []*ipc.ProgInfo
 	Error  string
 }
-
-type LogMessageReq struct {
-	Level   int
-	Name    string
-	Message string
-}
diff --git a/pkg/signal/signal.go b/pkg/signal/signal.go
index 7a2a8bd16..2860be95e 100644
--- a/pkg/signal/signal.go
+++ b/pkg/signal/signal.go
@@ -11,11 +11,6 @@ type (
 
 type Signal map[elemType]prioType
 
-type Serial struct {
-	Elems []elemType
-	Prios []prioType
-}
-
 func (s Signal) Len() int {
 	return len(s)
 }
@@ -64,42 +59,6 @@ func FromRaw(raw []uint32, prio uint8) Signal {
 	return s
 }
 
-func (s Signal) Serialize() Serial {
-	if s.Empty() {
-		return Serial{}
-	}
-	res := Serial{
-		Elems: make([]elemType, len(s)),
-		Prios: make([]prioType, len(s)),
-	}
-	i := 0
-	for e, p := range s {
-		res.Elems[i] = e
-		res.Prios[i] = p
-		i++
-	}
-	return res
-}
-
-func (ser *Serial) AddElem(elem uint32, prio prioType) {
-	ser.Elems = append(ser.Elems, elemType(elem))
-	ser.Prios = append(ser.Prios, prio)
-}
-
-func (ser Serial) Deserialize() Signal {
-	if len(ser.Elems) != len(ser.Prios) {
-		panic("corrupted Serial")
-	}
-	if len(ser.Elems) == 0 {
-		return nil
-	}
-	s := make(Signal, len(ser.Elems))
-	for i, e := range ser.Elems {
-		s[e] = ser.Prios[i]
-	}
-	return s
-}
-
 func (s Signal) Diff(s1 Signal) Signal {
 	if s1.Empty() {
 		return nil
@@ -160,6 +119,36 @@ func (s *Signal) Merge(s1 Signal) {
 	}
 }
 
+// FilterRaw returns a subset of original raw elements that coincides with the one in Signal.
+func (s Signal) FilterRaw(raw []uint32) []uint32 {
+	var ret []uint32
+	for _, e := range raw {
+		if _, ok := s[elemType(e)]; ok {
+			ret = append(ret, e)
+		}
+	}
+	return ret
+}
+
+// DiffFromRaw returns a subset of the raw elements that is not present in Signal.
+func (s Signal) DiffFromRaw(raw []uint32) []uint32 {
+	var ret []uint32
+	for _, e := range raw {
+		if _, ok := s[elemType(e)]; !ok {
+			ret = append(ret, e)
+		}
+	}
+	return ret
+}
+
+func (s Signal) ToRaw() []uint32 {
+	var raw []uint32
+	for e := range s {
+		raw = append(raw, uint32(e))
+	}
+	return raw
+}
+
 type Context struct {
 	Signal  Signal
 	Context interface{}
diff --git a/syz-fuzzer/fuzzer.go b/syz-fuzzer/fuzzer.go
index cf83c15d7..17382bd03 100644
--- a/syz-fuzzer/fuzzer.go
+++ b/syz-fuzzer/fuzzer.go
@@ -4,10 +4,8 @@
 package main
 
 import (
-	"context"
 	"flag"
 	"fmt"
-	"math/rand"
 	"net/http"
 	_ "net/http/pprof"
 	"os"
@@ -17,15 +15,14 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/google/syzkaller/pkg/corpus"
 	"github.com/google/syzkaller/pkg/csource"
-	"github.com/google/syzkaller/pkg/fuzzer"
 	"github.com/google/syzkaller/pkg/host"
 	"github.com/google/syzkaller/pkg/ipc"
 	"github.com/google/syzkaller/pkg/ipc/ipcconfig"
 	"github.com/google/syzkaller/pkg/log"
 	"github.com/google/syzkaller/pkg/osutil"
 	"github.com/google/syzkaller/pkg/rpctype"
+	"github.com/google/syzkaller/pkg/signal"
 	"github.com/google/syzkaller/pkg/tool"
 	"github.com/google/syzkaller/prog"
 	_ "github.com/google/syzkaller/sys"
@@ -36,7 +33,6 @@ type FuzzerTool struct {
 	name              string
 	outputType        OutputType
 	config            *ipc.Config
-	fuzzer            *fuzzer.Fuzzer
 	procs             []*Proc
 	gate              *ipc.Gate
 	manager           *rpctype.RPCClient
@@ -47,8 +43,27 @@ type FuzzerTool struct {
 	checkResult *rpctype.CheckArgs
 	logMu       sync.Mutex
 
-	bufferTooSmall uint64
+	bufferTooSmall atomic.Uint64
+	noExecRequests atomic.Uint64
 	resetAccState  bool
+
+	inputs    chan executionRequest
+	results   chan executionResult
+	signalMu  sync.RWMutex
+	maxSignal signal.Signal
+}
+
+// executionResult offloads some computations from the proc loop
+// to the communication thread.
+type executionResult struct {
+	rpctype.ExecutionRequest
+	info *ipc.ProgInfo
+}
+
+// executionRequest offloads prog deseralization to another thread.
+type executionRequest struct {
+	rpctype.ExecutionRequest
+	prog *prog.Prog
 }
 
 type OutputType int
@@ -224,26 +239,8 @@ func main() {
 		runTest(target, manager, *flagName, config.Executor)
 		return
 	}
-	rnd := rand.New(rand.NewSource(time.Now().UnixNano()))
-	calls := make(map[*prog.Syscall]bool)
-	for _, id := range r.CheckResult.EnabledCalls[sandbox] {
-		calls[target.Syscalls[id]] = true
-	}
-	fuzzerObj := fuzzer.NewFuzzer(context.Background(), &fuzzer.Config{
-		Corpus:         corpus.NewCorpus(context.Background()),
-		Coverage:       config.Flags&ipc.FlagSignal > 0,
-		FaultInjection: r.CheckResult.Features[host.FeatureFault].Enabled,
-		Comparisons:    r.CheckResult.Features[host.FeatureComparisons].Enabled,
-		Collide:        execOpts.Flags&ipc.FlagThreaded > 0,
-		EnabledCalls:   calls,
-		NoMutateCalls:  r.NoMutateCalls,
-		FetchRawCover:  *flagRawCover,
-		MinCandidates:  uint(*flagProcs * 2),
-		NewInputs:      make(chan corpus.NewInput),
-	}, rnd, target)
-
+	inputsCount := *flagProcs * 2
 	fuzzerTool := &FuzzerTool{
-		fuzzer:        fuzzerObj,
 		name:          *flagName,
 		outputType:    outputType,
 		manager:       manager,
@@ -252,33 +249,18 @@ func main() {
 		config:        config,
 		checkResult:   r.CheckResult,
 		resetAccState: *flagResetAccState,
-	}
-	fuzzerObj.Config.Logf = func(level int, msg string, args ...interface{}) {
-		// Log 0 messages are most important: send them directly to syz-manager.
-		if level == 0 {
-			fuzzerTool.Logf(level, msg, args...)
-		}
-		// Dump log level 0 and 1 messages into syz-fuzzer output.
-		if level <= 1 {
-			fuzzerTool.logMu.Lock()
-			defer fuzzerTool.logMu.Unlock()
-			log.Logf(0, "fuzzer: "+msg, args...)
-		}
+
+		inputs:  make(chan executionRequest, inputsCount),
+		results: make(chan executionResult, inputsCount),
 	}
 	fuzzerTool.gate = ipc.NewGate(gateSize,
 		fuzzerTool.useBugFrames(r, *flagProcs))
-	for needCandidates, more := true, true; more; needCandidates = false {
-		more = fuzzerTool.poll(needCandidates, nil)
-		// This loop lead to "no output" in qemu emulation, tell manager we are not dead.
-		stat := fuzzerObj.Stats()
-		log.Logf(0, "fetching corpus: %v, signal %v/%v (executing program)",
-			stat.Progs, stat.Signal, stat.MaxSignal)
-	}
 	if r.CoverFilterBitmap != nil {
 		execOpts.Flags |= ipc.FlagEnableCoverageFilter
 	}
-
-	log.Logf(0, "starting %v fuzzer processes", *flagProcs)
+	// Query enough inputs at the beginning.
+	fuzzerTool.exchangeDataCall(inputsCount, nil)
+	log.Logf(0, "starting %v executor processes", *flagProcs)
 	for pid := 0; pid < *flagProcs; pid++ {
 		proc, err := newProc(fuzzerTool, execOpts, pid)
 		if err != nil {
@@ -287,11 +269,8 @@ func main() {
 		fuzzerTool.procs = append(fuzzerTool.procs, proc)
 		go proc.loop()
 	}
-	// Start send input workers.
-	for i := 0; i < *flagProcs*2; i++ {
-		go fuzzerTool.sendInputsWorker(fuzzerObj.Config.NewInputs)
-	}
-	fuzzerTool.pollLoop()
+	go fuzzerTool.exchangeDataWorker()
+	fuzzerTool.exchangeDataWorker()
 }
 
 func collectMachineInfos(target *prog.Target) ([]byte, []host.KernelModule) {
@@ -356,110 +335,75 @@ func (tool *FuzzerTool) filterDataRaceFrames(frames []string) {
 	log.Logf(0, "%s", output)
 }
 
-func (tool *FuzzerTool) pollLoop() {
-	var execTotal uint64
-	var lastPoll time.Time
-	var lastPrint time.Time
-	ticker := time.NewTicker(3 * time.Second * tool.timeouts.Scale).C
-	for {
-		needCandidates := false
-		select {
-		case <-ticker:
-		case <-tool.fuzzer.NeedCandidates:
-			needCandidates = true
-		}
-		if tool.outputType != OutputStdout && time.Since(lastPrint) > 10*time.Second*tool.timeouts.Scale {
-			// Keep-alive for manager.
-			log.Logf(0, "alive, executed %v", execTotal)
-			lastPrint = time.Now()
+func (tool *FuzzerTool) exchangeDataCall(needProgs int, results []executionResult) {
+	a := &rpctype.ExchangeInfoRequest{
+		Name:       tool.name,
+		NeedProgs:  needProgs,
+		StatsDelta: tool.grabStats(),
+	}
+	for _, result := range results {
+		a.Results = append(a.Results, tool.convertExecutionResult(result))
+	}
+	r := &rpctype.ExchangeInfoReply{}
+	if err := tool.manager.Call("Manager.ExchangeInfo", a, r); err != nil {
+		log.SyzFatalf("Manager.ExchangeInfo call failed: %v", err)
+	}
+	tool.addMaxSignal(r.NewMaxSignal)
+	for _, req := range r.Requests {
+		p := tool.deserializeInput(req.ProgData)
+		if p == nil {
+			log.SyzFatalf("failed to deserialize input: %s", req.ProgData)
 		}
-		needCandidates = tool.fuzzer.NeedCandidatesNow()
-		if needCandidates || time.Since(lastPoll) > 10*time.Second*tool.timeouts.Scale {
-			more := tool.poll(needCandidates, tool.grabStats())
-			if !more {
-				lastPoll = time.Now()
-			}
+		tool.inputs <- executionRequest{
+			ExecutionRequest: req,
+			prog:             p,
 		}
 	}
 }
 
-func (tool *FuzzerTool) poll(needCandidates bool, stats map[string]uint64) bool {
-	fuzzer := tool.fuzzer
-	a := &rpctype.PollArgs{
-		Name:           tool.name,
-		NeedCandidates: needCandidates,
-		MaxSignal:      fuzzer.Cover.GrabNewSignal().Serialize(),
-		Stats:          stats,
-	}
-	r := &rpctype.PollRes{}
-	if err := tool.manager.Call("Manager.Poll", a, r); err != nil {
-		log.SyzFatalf("Manager.Poll call failed: %v", err)
-	}
-	maxSignal := r.MaxSignal.Deserialize()
-	log.Logf(1, "poll: candidates=%v inputs=%v signal=%v",
-		len(r.Candidates), len(r.NewInputs), maxSignal.Len())
-	fuzzer.Cover.AddMaxSignal(maxSignal)
-	for _, inp := range r.NewInputs {
-		tool.inputFromOtherFuzzer(inp)
-	}
-	tool.addCandidates(r.Candidates)
-	if needCandidates && len(r.Candidates) == 0 && atomic.LoadUint32(&tool.triagedCandidates) == 0 {
-		atomic.StoreUint32(&tool.triagedCandidates, 1)
-	}
-	return len(r.NewInputs) != 0 || len(r.Candidates) != 0 || maxSignal.Len() != 0
-}
-
-func (tool *FuzzerTool) sendInputsWorker(ch <-chan corpus.NewInput) {
-	for update := range ch {
-		a := &rpctype.NewInputArgs{
-			Name:  tool.name,
-			Input: update.RPCInput(),
+func (tool *FuzzerTool) exchangeDataWorker() {
+	for result := range tool.results {
+		results := []executionResult{
+			result,
 		}
-		if err := tool.manager.Call("Manager.NewInput", a, nil); err != nil {
-			log.SyzFatalf("Manager.NewInput call failed: %v", err)
+		// Grab other finished calls, just in case there are any.
+	loop:
+		for {
+			select {
+			case res := <-tool.results:
+				results = append(results, res)
+			default:
+				break loop
+			}
 		}
+		// Replenish exactly the finished requests.
+		tool.exchangeDataCall(len(results), results)
 	}
 }
 
-func (tool *FuzzerTool) grabStats() map[string]uint64 {
-	stats := tool.fuzzer.GrabStats()
-	for _, proc := range tool.procs {
-		stats["exec total"] += atomic.SwapUint64(&proc.env.StatExecs, 0)
-		stats["executor restarts"] += atomic.SwapUint64(&proc.env.StatRestarts, 0)
+func (tool *FuzzerTool) convertExecutionResult(res executionResult) rpctype.ExecutionResult {
+	if res.NeedSignal == rpctype.NewSignal {
+		tool.diffMaxSignal(res.info)
 	}
-	stats["buffer too small"] = atomic.SwapUint64(&tool.bufferTooSmall, 0)
-	return stats
-}
-
-func (tool *FuzzerTool) addCandidates(candidates []rpctype.Candidate) {
-	var inputs []fuzzer.Candidate
-	for _, candidate := range candidates {
-		p := tool.deserializeInput(candidate.Prog)
-		if p == nil {
-			continue
-		}
-		inputs = append(inputs, fuzzer.Candidate{
-			Prog:      p,
-			Smashed:   candidate.Smashed,
-			Minimized: candidate.Minimized,
-		})
+	if res.SignalFilter != nil {
+		// TODO: we can filter without maps if req.SignalFilter is sorted.
+		filterProgInfo(res.info, res.SignalFilter)
 	}
-	if len(inputs) > 0 {
-		tool.fuzzer.AddCandidates(inputs)
+	return rpctype.ExecutionResult{
+		ID:   res.ID,
+		Info: *res.info,
 	}
 }
 
-func (tool *FuzzerTool) inputFromOtherFuzzer(inp rpctype.Input) {
-	p := tool.deserializeInput(inp.Prog)
-	if p == nil {
-		return
+func (tool *FuzzerTool) grabStats() map[string]uint64 {
+	stats := map[string]uint64{}
+	for _, proc := range tool.procs {
+		stats["exec total"] += atomic.SwapUint64(&proc.env.StatExecs, 0)
+		stats["executor restarts"] += atomic.SwapUint64(&proc.env.StatRestarts, 0)
 	}
-	tool.fuzzer.Config.Corpus.Save(corpus.NewInput{
-		Prog:   p,
-		Call:   inp.Call,
-		Signal: inp.Signal.Deserialize(),
-		Cover:  inp.Cover,
-	})
+	stats["buffer too small"] = tool.bufferTooSmall.Swap(0)
+	stats["no exec requests"] = tool.noExecRequests.Swap(0)
+	return stats
 }
 
 func (tool *FuzzerTool) deserializeInput(inp []byte) *prog.Prog {
@@ -467,46 +411,41 @@ func (tool *FuzzerTool) deserializeInput(inp []byte) *prog.Prog {
 	if err != nil {
 		log.SyzFatalf("failed to deserialize prog: %v\n%s", err, inp)
 	}
-	tool.checkDisabledCalls(p)
 	if len(p.Calls) > prog.MaxCalls {
 		return nil
 	}
 	return p
 }
 
-func (tool *FuzzerTool) checkDisabledCalls(p *prog.Prog) {
-	ct := tool.fuzzer.ChoiceTable()
-	for _, call := range p.Calls {
-		if !ct.Enabled(call.Meta.ID) {
-			fmt.Printf("executing disabled syscall %v [%v]\n", call.Meta.Name, call.Meta.ID)
-			sandbox := ipc.FlagsToSandbox(tool.config.Flags)
-			fmt.Printf("check result for sandbox=%v:\n", sandbox)
-			for _, id := range tool.checkResult.EnabledCalls[sandbox] {
-				meta := tool.target.Syscalls[id]
-				fmt.Printf("  %v [%v]\n", meta.Name, meta.ID)
-			}
-			fmt.Printf("choice table:\n")
-			for i, meta := range tool.target.Syscalls {
-				fmt.Printf("  #%v: %v [%v]: enabled=%v\n", i, meta.Name, meta.ID, ct.Enabled(meta.ID))
-			}
-			panic("disabled syscall")
-		}
+// The linter is too aggressive.
+// nolint: dupl
+func filterProgInfo(info *ipc.ProgInfo, mask signal.Signal) {
+	info.Extra.Signal = mask.FilterRaw(info.Extra.Signal)
+	for i := 0; i < len(info.Calls); i++ {
+		info.Calls[i].Signal = mask.FilterRaw(info.Calls[i].Signal)
 	}
 }
 
-// nolint: unused
-// It's only needed for debugging.
-func (tool *FuzzerTool) Logf(level int, msg string, args ...interface{}) {
-	go func() {
-		a := &rpctype.LogMessageReq{
-			Level:   level,
-			Name:    tool.name,
-			Message: fmt.Sprintf(msg, args...),
-		}
-		if err := tool.manager.Call("Manager.LogMessage", a, nil); err != nil {
-			log.SyzFatalf("Manager.LogMessage call failed: %v", err)
-		}
-	}()
+// The linter is too aggressive.
+// nolint: dupl
+func diffProgInfo(info *ipc.ProgInfo, base signal.Signal) {
+	info.Extra.Signal = base.DiffFromRaw(info.Extra.Signal)
+	for i := 0; i < len(info.Calls); i++ {
+		info.Calls[i].Signal = base.DiffFromRaw(info.Calls[i].Signal)
+	}
+}
+
+func (tool *FuzzerTool) diffMaxSignal(info *ipc.ProgInfo) {
+	tool.signalMu.RLock()
+	defer tool.signalMu.RUnlock()
+
+	diffProgInfo(info, tool.maxSignal)
+}
+
+func (tool *FuzzerTool) addMaxSignal(diff []uint32) {
+	tool.signalMu.Lock()
+	defer tool.signalMu.Unlock()
+	tool.maxSignal.Merge(signal.FromRaw(diff, 0))
 }
 
 func setupPprofHandler(port int) {
diff --git a/syz-fuzzer/fuzzer_test.go b/syz-fuzzer/fuzzer_test.go
new file mode 100644
index 000000000..9ce5514b6
--- /dev/null
+++ b/syz-fuzzer/fuzzer_test.go
@@ -0,0 +1,88 @@
+// Copyright 2024 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 (
+	"testing"
+
+	"github.com/google/syzkaller/pkg/ipc"
+	"github.com/google/syzkaller/pkg/signal"
+	"github.com/stretchr/testify/assert"
+)
+
+// nolint: dupl
+func TestFilterProgInfo(t *testing.T) {
+	mask := signal.FromRaw([]uint32{2, 4, 6, 8}, 0)
+	info := ipc.ProgInfo{
+		Calls: []ipc.CallInfo{
+			{
+				Signal: []uint32{1, 2, 3},
+				Cover:  []uint32{1, 2, 3},
+			},
+			{
+				Signal: []uint32{2, 3, 4},
+				Cover:  []uint32{2, 3, 4},
+			},
+		},
+		Extra: ipc.CallInfo{
+			Signal: []uint32{3, 4, 5},
+			Cover:  []uint32{3, 4, 5},
+		},
+	}
+	filterProgInfo(&info, mask)
+	assert.Equal(t, ipc.ProgInfo{
+		Calls: []ipc.CallInfo{
+			{
+				Signal: []uint32{2},
+				Cover:  []uint32{1, 2, 3},
+			},
+			{
+				Signal: []uint32{2, 4},
+				Cover:  []uint32{2, 3, 4},
+			},
+		},
+		Extra: ipc.CallInfo{
+			Signal: []uint32{4},
+			Cover:  []uint32{3, 4, 5},
+		},
+	}, info)
+}
+
+// nolint: dupl
+func TestDiffProgInfo(t *testing.T) {
+	base := signal.FromRaw([]uint32{0, 1, 2}, 0)
+	info := ipc.ProgInfo{
+		Calls: []ipc.CallInfo{
+			{
+				Signal: []uint32{0, 1, 2},
+				Cover:  []uint32{0, 1, 2},
+			},
+			{
+				Signal: []uint32{1, 2, 3},
+				Cover:  []uint32{1, 2, 3},
+			},
+		},
+		Extra: ipc.CallInfo{
+			Signal: []uint32{2, 3, 4},
+			Cover:  []uint32{2, 3, 4},
+		},
+	}
+	diffProgInfo(&info, base)
+	assert.Equal(t, ipc.ProgInfo{
+		Calls: []ipc.CallInfo{
+			{
+				Signal: nil,
+				Cover:  []uint32{0, 1, 2},
+			},
+			{
+				Signal: []uint32{3},
+				Cover:  []uint32{1, 2, 3},
+			},
+		},
+		Extra: ipc.CallInfo{
+			Signal: []uint32{3, 4},
+			Cover:  []uint32{2, 3, 4},
+		},
+	}, info)
+}
diff --git a/syz-fuzzer/proc.go b/syz-fuzzer/proc.go
index 369ec5735..02e6aa5a5 100644
--- a/syz-fuzzer/proc.go
+++ b/syz-fuzzer/proc.go
@@ -9,13 +9,12 @@ import (
 	"math/rand"
 	"os"
 	"runtime/debug"
-	"sync/atomic"
 	"syscall"
 	"time"
 
-	"github.com/google/syzkaller/pkg/fuzzer"
 	"github.com/google/syzkaller/pkg/ipc"
 	"github.com/google/syzkaller/pkg/log"
+	"github.com/google/syzkaller/pkg/rpctype"
 	"github.com/google/syzkaller/prog"
 )
 
@@ -44,9 +43,9 @@ func newProc(tool *FuzzerTool, execOpts *ipc.ExecOpts, pid int) (*Proc, error) {
 func (proc *Proc) loop() {
 	rnd := rand.New(rand.NewSource(time.Now().UnixNano() + int64(proc.pid)))
 	for {
-		req := proc.tool.fuzzer.NextInput()
+		req := proc.nextRequest()
 		opts := *proc.execOpts
-		if !req.NeedSignal {
+		if req.NeedSignal == rpctype.NoSignal {
 			opts.Flags &= ^ipc.FlagCollectSignal
 		}
 		if req.NeedCover {
@@ -62,18 +61,32 @@ func (proc *Proc) loop() {
 		const restartIn = 600
 		restart := rnd.Intn(restartIn) == 0
 		if (restart || proc.tool.resetAccState) &&
-			(req.NeedCover || req.NeedSignal || req.NeedHints) {
+			(req.NeedCover || req.NeedSignal != rpctype.NoSignal || req.NeedHints) {
 			proc.env.ForceRestart()
 		}
-		info := proc.executeRaw(&opts, req.Prog)
-		proc.tool.fuzzer.Done(req, &fuzzer.Result{
-			Info: info,
-		})
+		info := proc.executeRaw(&opts, req.prog)
+		// Let's perform signal filtering in a separate thread to get the most
+		// exec/sec out of a syz-executor instance.
+		proc.tool.results <- executionResult{
+			ExecutionRequest: req.ExecutionRequest,
+			info:             info,
+		}
+	}
+}
+
+func (proc *Proc) nextRequest() executionRequest {
+	select {
+	case req := <-proc.tool.inputs:
+		return req
+	default:
 	}
+	// Not having enough inputs to execute is a sign of RPC communication problems.
+	// Let's count and report such situations.
+	proc.tool.noExecRequests.Add(1)
+	return <-proc.tool.inputs
 }
 
 func (proc *Proc) executeRaw(opts *ipc.ExecOpts, p *prog.Prog) *ipc.ProgInfo {
-	proc.tool.checkDisabledCalls(p)
 	for try := 0; ; try++ {
 		var output []byte
 		var info *ipc.ProgInfo
@@ -93,7 +106,7 @@ func (proc *Proc) executeRaw(opts *ipc.ExecOpts, p *prog.Prog) *ipc.ProgInfo {
 				// It's bad if we systematically fail to serialize programs,
 				// but so far we don't have a better handling than counting this.
 				// This error is observed a lot on the seeded syz_mount_image calls.
-				atomic.AddUint64(&proc.tool.bufferTooSmall, 1)
+				proc.tool.bufferTooSmall.Add(1)
 				return nil
 			}
 			if try > 10 {
diff --git a/syz-manager/http.go b/syz-manager/http.go
index f38a7713b..bdac900f7 100644
--- a/syz-manager/http.go
+++ b/syz-manager/http.go
@@ -131,9 +131,10 @@ func (mgr *Manager) collectStats() []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(mgr.corpus.Stats().Progs), Link: "/corpus"},
-		{Name: "triage queue", Value: fmt.Sprint(len(mgr.candidates))},
+		{Name: "triage queue", Value: fmt.Sprint(mgr.stats.triageQueueLen.get())},
 		{Name: "signal", Value: fmt.Sprint(rawStats["signal"])},
 		{Name: "coverage", Value: fmt.Sprint(rawStats["coverage"]), Link: "/cover"},
+		{Name: "fuzzer jobs", Value: fmt.Sprint(mgr.stats.fuzzerJobs.get())},
 	}
 	if mgr.coverFilter != nil {
 		stats = append(stats, UIStat{
@@ -147,6 +148,7 @@ func (mgr *Manager) collectStats() []UIStat {
 	delete(rawStats, "signal")
 	delete(rawStats, "coverage")
 	delete(rawStats, "filtered coverage")
+	delete(rawStats, "fuzzer jobs")
 	if mgr.checkResult != nil {
 		stats = append(stats, UIStat{
 			Name:  "syscalls",
diff --git a/syz-manager/hub.go b/syz-manager/hub.go
index d06a0cd0e..7a38310d5 100644
--- a/syz-manager/hub.go
+++ b/syz-manager/hub.go
@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/google/syzkaller/pkg/auth"
+	"github.com/google/syzkaller/pkg/fuzzer"
 	"github.com/google/syzkaller/pkg/hash"
 	"github.com/google/syzkaller/pkg/host"
 	"github.com/google/syzkaller/pkg/log"
@@ -73,7 +74,7 @@ type HubConnector struct {
 // HubManagerView restricts interface between HubConnector and Manager.
 type HubManagerView interface {
 	getMinimizedCorpus() (corpus, repros [][]byte)
-	addNewCandidates(candidates []rpctype.Candidate)
+	addNewCandidates(candidates []fuzzer.Candidate)
 	hubIsUnreachable()
 }
 
@@ -213,9 +214,9 @@ func (hc *HubConnector) sync(hub *rpctype.RPCClient, corpus [][]byte) error {
 }
 
 func (hc *HubConnector) processProgs(inputs []rpctype.HubInput) (minimized, smashed, dropped int) {
-	candidates := make([]rpctype.Candidate, 0, len(inputs))
+	candidates := make([]fuzzer.Candidate, 0, len(inputs))
 	for _, inp := range inputs {
-		_, disabled, bad := parseProgram(hc.target, hc.enabledCalls, inp.Prog)
+		p, disabled, bad := parseProgram(hc.target, hc.enabledCalls, inp.Prog)
 		if bad != nil || disabled {
 			log.Logf(0, "rejecting program from hub (bad=%v, disabled=%v):\n%s",
 				bad, disabled, inp)
@@ -229,8 +230,8 @@ func (hc *HubConnector) processProgs(inputs []rpctype.HubInput) (minimized, smas
 		if smash {
 			smashed++
 		}
-		candidates = append(candidates, rpctype.Candidate{
-			Prog:      inp.Prog,
+		candidates = append(candidates, fuzzer.Candidate{
+			Prog:      p,
 			Minimized: min,
 			Smashed:   smash,
 		})
@@ -283,7 +284,6 @@ func (hc *HubConnector) processRepros(repros [][]byte) int {
 			typ = crash.MemoryLeak
 		}
 		hc.hubReproQueue <- &Crash{
-			vmIndex: -1,
 			fromHub: true,
 			Report: &report.Report{
 				Title:  "external repro",
diff --git a/syz-manager/manager.go b/syz-manager/manager.go
index 114e455d4..673deadc9 100644
--- a/syz-manager/manager.go
+++ b/syz-manager/manager.go
@@ -24,6 +24,7 @@ import (
 	"github.com/google/syzkaller/pkg/corpus"
 	"github.com/google/syzkaller/pkg/csource"
 	"github.com/google/syzkaller/pkg/db"
+	"github.com/google/syzkaller/pkg/fuzzer"
 	"github.com/google/syzkaller/pkg/gce"
 	"github.com/google/syzkaller/pkg/hash"
 	"github.com/google/syzkaller/pkg/host"
@@ -67,14 +68,15 @@ type Manager struct {
 	fresh          bool
 	numFuzzing     uint32
 	numReproducing uint32
+	nextInstanceID atomic.Uint64
 
 	dash *dashapi.Dashboard
 
 	mu                    sync.Mutex
+	fuzzer                *fuzzer.Fuzzer
 	phase                 int
 	targetEnabledSyscalls map[*prog.Syscall]bool
 
-	candidates       []rpctype.Candidate // untriaged inputs from corpus and hub
 	disabledHashes   map[string]struct{}
 	seeds            [][]byte
 	newRepros        [][]byte
@@ -118,7 +120,7 @@ const (
 const currentDBVersion = 4
 
 type Crash struct {
-	vmIndex       int
+	instanceName  string
 	fromHub       bool // this crash was created based on a repro from syz-hub
 	fromDashboard bool // .. or from dashboard
 	*report.Report
@@ -191,7 +193,7 @@ func RunManager(cfg *mgrconfig.Config) {
 	mgr.initStats() // Initializes prometheus variables.
 	mgr.initHTTP()  // Creates HTTP server.
 	mgr.collectUsedFiles()
-	go mgr.saveCorpus(corpusUpdates)
+	go mgr.corpusInputHandler(corpusUpdates)
 
 	// Create RPC server for fuzzers.
 	mgr.serv, err = startRPCServer(mgr)
@@ -225,13 +227,13 @@ func RunManager(cfg *mgrconfig.Config) {
 				continue
 			}
 			mgr.fuzzingTime += diff * time.Duration(atomic.LoadUint32(&mgr.numFuzzing))
+			mgr.mu.Unlock()
 			executed := mgr.stats.execTotal.get()
 			crashes := mgr.stats.crashes.get()
 			corpusCover := mgr.stats.corpusCover.get()
 			corpusSignal := mgr.stats.corpusSignal.get()
 			maxSignal := mgr.stats.maxSignal.get()
-			triageQLen := len(mgr.candidates)
-			mgr.mu.Unlock()
+			triageQLen := mgr.stats.triageQueueLen.get()
 			numReproducing := atomic.LoadUint32(&mgr.numReproducing)
 			numFuzzing := atomic.LoadUint32(&mgr.numFuzzing)
 
@@ -279,7 +281,7 @@ func (mgr *Manager) initBench() {
 			vals["corpus"] = uint64(stat.Progs)
 			vals["uptime"] = uint64(time.Since(mgr.firstConnect)) / 1e9
 			vals["fuzzing"] = uint64(mgr.fuzzingTime) / 1e9
-			vals["candidates"] = uint64(len(mgr.candidates))
+			vals["candidates"] = uint64(mgr.fuzzer.Stats().Candidates)
 			mgr.mu.Unlock()
 
 			data, err := json.MarshalIndent(vals, "", "  ")
@@ -633,21 +635,29 @@ func (mgr *Manager) loadCorpus() {
 		fallthrough
 	case currentDBVersion:
 	}
+	var candidates []fuzzer.Candidate
 	broken := 0
 	for key, rec := range mgr.corpusDB.Records {
-		if !mgr.loadProg(rec.Val, minimized, smashed) {
+		drop, item := mgr.loadProg(rec.Val, minimized, smashed)
+		if drop {
 			mgr.corpusDB.Delete(key)
 			broken++
 		}
+		if item != nil {
+			candidates = append(candidates, *item)
+		}
 	}
 	mgr.fresh = len(mgr.corpusDB.Records) == 0
-	corpusSize := len(mgr.candidates)
+	corpusSize := len(candidates)
 	log.Logf(0, "%-24v: %v (deleted %v broken)", "corpus", corpusSize, broken)
 
 	for _, seed := range mgr.seeds {
-		mgr.loadProg(seed, true, false)
+		_, item := mgr.loadProg(seed, true, false)
+		if item != nil {
+			candidates = append(candidates, *item)
+		}
 	}
-	log.Logf(0, "%-24v: %v/%v", "seeds", len(mgr.candidates)-corpusSize, len(mgr.seeds))
+	log.Logf(0, "%-24v: %v/%v", "seeds", len(candidates)-corpusSize, len(candidates))
 	mgr.seeds = nil
 
 	// We duplicate all inputs in the corpus and shuffle the second part.
@@ -655,8 +665,8 @@ func (mgr *Manager) loadCorpus() {
 	// in such case it will also lost all cached candidates. Or, the input can be somewhat flaky
 	// and doesn't give the coverage on first try. So we give each input the second chance.
 	// Shuffling should alleviate deterministically losing the same inputs on fuzzer crashing.
-	mgr.candidates = append(mgr.candidates, mgr.candidates...)
-	shuffle := mgr.candidates[len(mgr.candidates)/2:]
+	candidates = append(candidates, candidates...)
+	shuffle := candidates[len(candidates)/2:]
 	rand.Shuffle(len(shuffle), func(i, j int) {
 		shuffle[i], shuffle[j] = shuffle[j], shuffle[i]
 	})
@@ -664,12 +674,14 @@ func (mgr *Manager) loadCorpus() {
 		panic(fmt.Sprintf("loadCorpus: bad phase %v", mgr.phase))
 	}
 	mgr.phase = phaseLoadedCorpus
+	mgr.fuzzer.AddCandidates(candidates)
 }
 
-func (mgr *Manager) loadProg(data []byte, minimized, smashed bool) bool {
-	_, disabled, bad := parseProgram(mgr.target, mgr.targetEnabledSyscalls, data)
+// Returns (delete item from the corpus, a fuzzer.Candidate object).
+func (mgr *Manager) loadProg(data []byte, minimized, smashed bool) (drop bool, candidate *fuzzer.Candidate) {
+	p, disabled, bad := parseProgram(mgr.target, mgr.targetEnabledSyscalls, data)
 	if bad != nil {
-		return false
+		return true, nil
 	}
 	if disabled {
 		if mgr.cfg.PreserveCorpus {
@@ -682,25 +694,24 @@ func (mgr *Manager) loadProg(data []byte, minimized, smashed bool) bool {
 			// minimize what remains from the prog. The original prog will be
 			// deleted from the corpus.
 			leftover := programLeftover(mgr.target, mgr.targetEnabledSyscalls, data)
-			if len(leftover) > 0 {
-				mgr.candidates = append(mgr.candidates, rpctype.Candidate{
+			if leftover != nil {
+				candidate = &fuzzer.Candidate{
 					Prog:      leftover,
 					Minimized: false,
 					Smashed:   smashed,
-				})
+				}
 			}
 		}
-		return true
+		return false, candidate
 	}
-	mgr.candidates = append(mgr.candidates, rpctype.Candidate{
-		Prog:      data,
+	return false, &fuzzer.Candidate{
+		Prog:      p,
 		Minimized: minimized,
 		Smashed:   smashed,
-	})
-	return true
+	}
 }
 
-func programLeftover(target *prog.Target, enabled map[*prog.Syscall]bool, data []byte) []byte {
+func programLeftover(target *prog.Target, enabled map[*prog.Syscall]bool, data []byte) *prog.Prog {
 	p, err := target.Deserialize(data, prog.NonStrict)
 	if err != nil {
 		panic(fmt.Sprintf("subsequent deserialization failed: %s", data))
@@ -713,7 +724,7 @@ func programLeftover(target *prog.Target, enabled map[*prog.Syscall]bool, data [
 		}
 		i++
 	}
-	return p.Serialize()
+	return p
 }
 
 func parseProgram(target *prog.Target, enabled map[*prog.Syscall]bool, data []byte) (
@@ -741,7 +752,8 @@ func parseProgram(target *prog.Target, enabled map[*prog.Syscall]bool, data []by
 
 func (mgr *Manager) runInstance(index int) (*Crash, error) {
 	mgr.checkUsedFiles()
-	instanceName := fmt.Sprintf("vm-%d", index)
+	// Use unique instance names to keep name collisions in case of untimely RPC messages.
+	instanceName := fmt.Sprintf("vm-%d", mgr.nextInstanceID.Add(1))
 
 	rep, vmInfo, err := mgr.runInstanceInner(index, instanceName)
 
@@ -759,9 +771,9 @@ func (mgr *Manager) runInstance(index int) (*Crash, error) {
 		return nil, nil
 	}
 	crash := &Crash{
-		vmIndex:     index,
-		Report:      rep,
-		machineInfo: machineInfo,
+		instanceName: instanceName,
+		Report:       rep,
+		machineInfo:  machineInfo,
 	}
 	return crash, nil
 }
@@ -884,7 +896,7 @@ func (mgr *Manager) saveCrash(crash *Crash) bool {
 	if crash.Suppressed {
 		flags += " [suppressed]"
 	}
-	log.Logf(0, "vm-%v: crash: %v%v", crash.vmIndex, crash.Title, flags)
+	log.Logf(0, "%s: crash: %v%v", crash.instanceName, crash.Title, flags)
 
 	if crash.Suppressed {
 		// Collect all of them into a single bucket so that it's possible to control and assess them,
@@ -1202,8 +1214,11 @@ func fullReproLog(stats *repro.Stats) []byte {
 		stats.SimplifyProgTime, stats.ExtractCTime, stats.SimplifyCTime, stats.Log))
 }
 
-func (mgr *Manager) saveCorpus(updates <-chan corpus.NewItemEvent) {
+func (mgr *Manager) corpusInputHandler(updates <-chan corpus.NewItemEvent) {
 	for update := range updates {
+		mgr.stats.newInputs.inc()
+		mgr.serv.updateFilteredCover(update.NewCover)
+
 		if update.Exists {
 			// We only save new progs into the corpus.db file.
 			continue
@@ -1231,16 +1246,16 @@ func (mgr *Manager) getMinimizedCorpus() (corpus, repros [][]byte) {
 	return
 }
 
-func (mgr *Manager) addNewCandidates(candidates []rpctype.Candidate) {
+func (mgr *Manager) addNewCandidates(candidates []fuzzer.Candidate) {
 	mgr.mu.Lock()
 	defer mgr.mu.Unlock()
+
 	if mgr.cfg.Experimental.ResetAccState {
 		// Don't accept new candidates -- the execution is already very slow,
 		// syz-hub will just overwhelm us.
 		return
 	}
-
-	mgr.candidates = append(mgr.candidates, candidates...)
+	mgr.fuzzer.AddCandidates(candidates)
 	if mgr.phase == phaseTriagedCorpus {
 		mgr.phase = phaseQueriedHub
 	}
@@ -1332,16 +1347,11 @@ func (mgr *Manager) collectSyscallInfo() map[string]*corpus.CallCov {
 }
 
 func (mgr *Manager) fuzzerConnect(modules []host.KernelModule) (
-	[]rpctype.Input, BugFrames, map[uint32]uint32, map[uint32]uint32, error) {
+	BugFrames, map[uint32]uint32, map[uint32]uint32, error) {
 	mgr.mu.Lock()
 	defer mgr.mu.Unlock()
 
 	mgr.minimizeCorpusUnlocked()
-	items := mgr.corpus.Items()
-	corpus := make([]rpctype.Input, 0, len(items))
-	for _, inp := range items {
-		corpus = append(corpus, inp.RPCInputShort())
-	}
 	frames := BugFrames{
 		memoryLeaks: make([]string, 0, len(mgr.memoryLeakFrames)),
 		dataRaces:   make([]string, 0, len(mgr.dataRaceFrames)),
@@ -1361,54 +1371,94 @@ func (mgr *Manager) fuzzerConnect(modules []host.KernelModule) (
 		}
 		mgr.modulesInitialized = true
 	}
-	return corpus, frames, mgr.coverFilter, mgr.execCoverFilter, nil
+	return frames, mgr.coverFilter, mgr.execCoverFilter, nil
 }
 
 func (mgr *Manager) machineChecked(a *rpctype.CheckArgs, enabledSyscalls map[*prog.Syscall]bool) {
 	mgr.mu.Lock()
 	defer mgr.mu.Unlock()
+	if mgr.checkResult != nil {
+		panic("machineChecked() called twice")
+	}
+
 	mgr.checkResult = a
 	mgr.targetEnabledSyscalls = enabledSyscalls
 	mgr.target.UpdateGlobs(a.GlobFiles)
-	mgr.loadCorpus()
 	mgr.firstConnect = time.Now()
+
+	rnd := rand.New(rand.NewSource(time.Now().UnixNano()))
+	calls := make(map[*prog.Syscall]bool)
+	for _, id := range a.EnabledCalls[mgr.cfg.Sandbox] {
+		calls[mgr.target.Syscalls[id]] = true
+	}
+	mgr.fuzzer = fuzzer.NewFuzzer(context.Background(), &fuzzer.Config{
+		Corpus:         mgr.corpus,
+		Coverage:       mgr.cfg.Cover,
+		FaultInjection: a.Features[host.FeatureFault].Enabled,
+		Comparisons:    a.Features[host.FeatureComparisons].Enabled,
+		Collide:        true,
+		EnabledCalls:   calls,
+		NoMutateCalls:  mgr.cfg.NoMutateCalls,
+		FetchRawCover:  mgr.cfg.RawCover,
+		Logf: func(level int, msg string, args ...interface{}) {
+			if level != 0 {
+				return
+			}
+			log.Logf(level, msg, args...)
+		},
+		NewInputFilter: func(input *corpus.NewInput) bool {
+			mgr.mu.Lock()
+			defer mgr.mu.Unlock()
+			return !mgr.saturatedCalls[input.StringCall()]
+		},
+	}, rnd, mgr.target)
+
+	mgr.loadCorpus()
+	go mgr.fuzzerLoop()
 }
 
-func (mgr *Manager) newInput(inp corpus.NewInput) bool {
+func (mgr *Manager) getFuzzer() *fuzzer.Fuzzer {
 	mgr.mu.Lock()
 	defer mgr.mu.Unlock()
-	if mgr.saturatedCalls[inp.StringCall()] {
-		// TODO: move this logic to pkg/corpus or pkg/fuzzer?
-		return false
-	}
-	mgr.corpus.Save(inp)
-	return true
+	return mgr.fuzzer
 }
 
-func (mgr *Manager) candidateBatch(size int) []rpctype.Candidate {
-	mgr.mu.Lock()
-	defer mgr.mu.Unlock()
-	var res []rpctype.Candidate
-	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.Candidate{}
-		mgr.candidates = mgr.candidates[:last]
-	}
-	if len(mgr.candidates) == 0 {
-		mgr.candidates = nil
-		if mgr.phase == phaseLoadedCorpus {
-			if mgr.cfg.HubClient != "" {
-				mgr.phase = phaseTriagedCorpus
-				go mgr.hubSyncLoop(pickGetter(mgr.cfg.HubKey))
-			} else {
+func (mgr *Manager) fuzzerLoop() {
+	for {
+		time.Sleep(time.Second / 2)
+
+		// Distribute new max signal over all instances.
+		newSignal := mgr.fuzzer.Cover.GrabNewSignal()
+		log.Logf(2, "distributing %d new signal", len(newSignal))
+		mgr.serv.distributeMaxSignal(newSignal)
+
+		// Collect statistics.
+		fuzzerStats := mgr.fuzzer.Stats()
+		mgr.stats.setNamed(fuzzerStats.Named)
+		mgr.stats.corpusCover.set(fuzzerStats.Cover)
+		mgr.stats.corpusSignal.set(fuzzerStats.Signal)
+		mgr.stats.maxSignal.set(fuzzerStats.MaxSignal)
+		mgr.stats.triageQueueLen.set(fuzzerStats.Candidates)
+		mgr.stats.fuzzerJobs.set(fuzzerStats.RunningJobs)
+		mgr.stats.rpcTraffic.add(int(mgr.serv.server.TotalBytes.Swap(0)))
+
+		// Update the state machine.
+		if fuzzerStats.Candidates == 0 {
+			mgr.mu.Lock()
+			if mgr.phase == phaseLoadedCorpus {
+				mgr.fuzzer.EnableOutOfQueue()
+				if mgr.cfg.HubClient != "" {
+					mgr.phase = phaseTriagedCorpus
+					go mgr.hubSyncLoop(pickGetter(mgr.cfg.HubKey))
+				} else {
+					mgr.phase = phaseTriagedHub
+				}
+			} else if mgr.phase == phaseQueriedHub {
 				mgr.phase = phaseTriagedHub
 			}
-		} else if mgr.phase == phaseQueriedHub {
-			mgr.phase = phaseTriagedHub
+			mgr.mu.Unlock()
 		}
 	}
-	return res
 }
 
 func (mgr *Manager) hubIsUnreachable() {
@@ -1535,7 +1585,6 @@ func (mgr *Manager) dashboardReproTasks() {
 		}
 		if len(resp.CrashLog) > 0 {
 			mgr.externalReproQueue <- &Crash{
-				vmIndex:       -1,
 				fromDashboard: true,
 				Report: &report.Report{
 					Title:  resp.Title,
diff --git a/syz-manager/rpc.go b/syz-manager/rpc.go
index 700bc6da6..472c406d2 100644
--- a/syz-manager/rpc.go
+++ b/syz-manager/rpc.go
@@ -5,13 +5,12 @@ package main
 
 import (
 	"fmt"
-	"math/rand"
 	"net"
 	"sync"
-	"time"
+	"sync/atomic"
 
-	"github.com/google/syzkaller/pkg/corpus"
 	"github.com/google/syzkaller/pkg/cover"
+	"github.com/google/syzkaller/pkg/fuzzer"
 	"github.com/google/syzkaller/pkg/host"
 	"github.com/google/syzkaller/pkg/log"
 	"github.com/google/syzkaller/pkg/mgrconfig"
@@ -23,34 +22,32 @@ import (
 type RPCServer struct {
 	mgr                   RPCManagerView
 	cfg                   *mgrconfig.Config
+	server                *rpctype.RPCServer
 	modules               []host.KernelModule
 	port                  int
 	targetEnabledSyscalls map[*prog.Syscall]bool
 	coverFilter           map[uint32]uint32
 	stats                 *Stats
-	batchSize             int
 	canonicalModules      *cover.Canonicalizer
 
 	mu          sync.Mutex
-	fuzzers     map[string]*Fuzzer
+	runners     sync.Map // Instead of map[string]*Runner.
 	checkResult *rpctype.CheckArgs
 
-	// TODO: we don't really need these anymore, but there's not much sense
-	// in rewriting the code that uses them -- most of that code will be dropped
-	// once we move pkg/fuzzer to the host.
-	maxSignal     signal.Signal
-	corpusSignal  signal.Signal
-	corpusCover   cover.Cover
-	rnd           *rand.Rand
 	checkFailures int
 }
 
-type Fuzzer struct {
-	name         string
-	inputs       []rpctype.Input
-	newMaxSignal signal.Signal
-	machineInfo  []byte
-	instModules  *cover.CanonicalizerInstance
+type Runner struct {
+	name string
+
+	machineInfo []byte
+	instModules *cover.CanonicalizerInstance
+
+	// The mutex protects newMaxSignal and requests.
+	mu            sync.Mutex
+	newMaxSignal  signal.Signal
+	nextRequestID atomic.Int64
+	requests      map[int64]*fuzzer.Request
 }
 
 type BugFrames struct {
@@ -60,24 +57,16 @@ type BugFrames struct {
 
 // RPCManagerView restricts interface between RPCServer and Manager.
 type RPCManagerView interface {
-	fuzzerConnect([]host.KernelModule) (
-		[]rpctype.Input, BugFrames, map[uint32]uint32, map[uint32]uint32, error)
+	fuzzerConnect([]host.KernelModule) (BugFrames, map[uint32]uint32, map[uint32]uint32, error)
 	machineChecked(result *rpctype.CheckArgs, enabledSyscalls map[*prog.Syscall]bool)
-	newInput(inp corpus.NewInput) bool
-	candidateBatch(size int) []rpctype.Candidate
+	getFuzzer() *fuzzer.Fuzzer
 }
 
 func startRPCServer(mgr *Manager) (*RPCServer, error) {
 	serv := &RPCServer{
-		mgr:     mgr,
-		cfg:     mgr.cfg,
-		stats:   mgr.stats,
-		fuzzers: make(map[string]*Fuzzer),
-		rnd:     rand.New(rand.NewSource(time.Now().UnixNano())),
-	}
-	serv.batchSize = 5
-	if serv.batchSize < mgr.cfg.Procs {
-		serv.batchSize = mgr.cfg.Procs
+		mgr:   mgr,
+		cfg:   mgr.cfg,
+		stats: mgr.stats,
 	}
 	s, err := rpctype.NewRPCServer(mgr.cfg.RPC, "Manager", serv)
 	if err != nil {
@@ -85,13 +74,8 @@ func startRPCServer(mgr *Manager) (*RPCServer, error) {
 	}
 	log.Logf(0, "serving rpc on tcp://%v", s.Addr())
 	serv.port = s.Addr().(*net.TCPAddr).Port
+	serv.server = s
 	go s.Serve()
-	go func() {
-		for {
-			time.Sleep(time.Second)
-			mgr.stats.rpcTraffic.add(int(s.TotalBytes.Swap(0)))
-		}
-	}()
 	return serv, nil
 }
 
@@ -99,37 +83,47 @@ func (serv *RPCServer) Connect(a *rpctype.ConnectArgs, r *rpctype.ConnectRes) er
 	log.Logf(1, "fuzzer %v connected", a.Name)
 	serv.stats.vmRestarts.inc()
 
+	serv.mu.Lock()
 	if serv.canonicalModules == nil {
 		serv.canonicalModules = cover.NewCanonicalizer(a.Modules, serv.cfg.Cover)
 		serv.modules = a.Modules
 	}
-	corpus, bugFrames, coverFilter, execCoverFilter, err := serv.mgr.fuzzerConnect(serv.modules)
+	serv.mu.Unlock()
+
+	bugFrames, coverFilter, execCoverFilter, err := serv.mgr.fuzzerConnect(serv.modules)
 	if err != nil {
 		return err
 	}
-	serv.coverFilter = coverFilter
 
 	serv.mu.Lock()
 	defer serv.mu.Unlock()
 
-	f := &Fuzzer{
+	serv.coverFilter = coverFilter
+
+	runner := &Runner{
 		name:        a.Name,
 		machineInfo: a.MachineInfo,
 		instModules: serv.canonicalModules.NewInstance(a.Modules),
+		requests:    make(map[int64]*fuzzer.Request),
+	}
+	if _, loaded := serv.runners.LoadOrStore(a.Name, runner); loaded {
+		return fmt.Errorf("duplicate connection from %s", a.Name)
 	}
-	serv.fuzzers[a.Name] = f
 	r.MemoryLeakFrames = bugFrames.memoryLeaks
 	r.DataRaceFrames = bugFrames.dataRaces
 
-	instCoverFilter := f.instModules.DecanonicalizeFilter(execCoverFilter)
+	instCoverFilter := runner.instModules.DecanonicalizeFilter(execCoverFilter)
 	r.CoverFilterBitmap = createCoverageBitmap(serv.cfg.SysTarget, instCoverFilter)
 	r.EnabledCalls = serv.cfg.Syscalls
-	r.NoMutateCalls = serv.cfg.NoMutateCalls
 	r.GitRevision = prog.GitRevision
 	r.TargetRevision = serv.cfg.Target.Revision
 	r.CheckResult = serv.checkResult
-	f.inputs = corpus
-	f.newMaxSignal = serv.maxSignal.Copy()
+
+	if fuzzer := serv.mgr.getFuzzer(); fuzzer != nil {
+		// A Fuzzer object is created after the first Check() call.
+		// If there was none, there would be no collected max signal either.
+		runner.newMaxSignal = fuzzer.Cover.CopyMaxSignal()
+	}
 	return nil
 }
 
@@ -177,140 +171,147 @@ func (serv *RPCServer) Check(a *rpctype.CheckArgs, r *int) error {
 	return nil
 }
 
-func (serv *RPCServer) NewInput(a *rpctype.NewInputArgs, r *int) error {
-	p, disabled, bad := parseProgram(serv.cfg.Target, serv.targetEnabledSyscalls, a.Input.Prog)
-	if bad != nil || disabled {
-		log.Errorf("rejecting program from fuzzer (bad=%v, disabled=%v):\n%s", bad, disabled, a.Input.Prog)
+func (serv *RPCServer) ExchangeInfo(a *rpctype.ExchangeInfoRequest, r *rpctype.ExchangeInfoReply) error {
+	var runner *Runner
+	if val, _ := serv.runners.Load(a.Name); val != nil {
+		runner = val.(*Runner)
+	} else {
+		// There might be a parallel shutdownInstance().
+		// Ignore the request then.
 		return nil
 	}
-	serv.mu.Lock()
-	defer serv.mu.Unlock()
 
-	f := serv.fuzzers[a.Name]
-	// Note: f may be nil if we called shutdownInstance,
-	// but this request is already in-flight.
-	if f != nil {
-		a.Cover, a.Signal = f.instModules.Canonicalize(a.Cover, a.Signal)
+	fuzzer := serv.mgr.getFuzzer()
+	if fuzzer == nil {
+		// ExchangeInfo calls follow MachineCheck, so the fuzzer must have been initialized.
+		panic("exchange info call with nil fuzzer")
 	}
-	inputSignal := a.Signal.Deserialize()
 
-	inp := corpus.NewInput{
-		Prog:   p,
-		Call:   a.Call,
-		Signal: inputSignal,
-		Cover:  a.Cover,
+	// First query new inputs and only then post results.
+	// It should foster a more even distribution of executions
+	// across all VMs.
+	for i := 0; i < a.NeedProgs; i++ {
+		inp := fuzzer.NextInput()
+		r.Requests = append(r.Requests, runner.newRequest(inp))
 	}
 
-	log.Logf(4, "new input from %v for syscall %v (signal=%v, cover=%v)",
-		a.Name, inp.StringCall(), inputSignal.Len(), len(a.Cover))
-	if serv.corpusSignal.Diff(inputSignal).Empty() {
-		return nil
-	}
-	if !serv.mgr.newInput(inp) {
-		return nil
+	for _, result := range a.Results {
+		runner.doneRequest(result, fuzzer)
 	}
 
-	diff := serv.corpusCover.MergeDiff(a.Cover)
-	serv.stats.corpusCover.set(len(serv.corpusCover))
-	if len(diff) != 0 && serv.coverFilter != nil {
-		// Note: ReportGenerator is already initialized if coverFilter is enabled.
-		rg, err := getReportGenerator(serv.cfg, serv.modules)
-		if err != nil {
-			return err
-		}
-		filtered := 0
-		for _, pc := range diff {
-			if serv.coverFilter[uint32(rg.RestorePC(pc))] != 0 {
-				filtered++
-			}
-		}
-		serv.stats.corpusCoverFiltered.add(filtered)
-	}
-	serv.stats.newInputs.inc()
+	serv.stats.mergeNamed(a.StatsDelta)
 
-	serv.corpusSignal.Merge(inputSignal)
-	serv.stats.corpusSignal.set(serv.corpusSignal.Len())
+	runner.mu.Lock()
+	// Let's transfer new max signal in portions.
+	const transferMaxSignal = 500000
+	maxSignalDiff := runner.newMaxSignal.Split(transferMaxSignal)
+	runner.mu.Unlock()
 
-	a.Input.Cover = nil // Don't send coverage back to all fuzzers.
-	a.Input.RawCover = nil
-	for _, other := range serv.fuzzers {
-		if other == f {
-			continue
-		}
-		other.inputs = append(other.inputs, a.Input)
-	}
-	return nil
-}
+	r.NewMaxSignal = runner.instModules.Decanonicalize(maxSignalDiff.ToRaw())
 
-func (serv *RPCServer) Poll(a *rpctype.PollArgs, r *rpctype.PollRes) error {
-	serv.stats.mergeNamed(a.Stats)
+	log.Logf(2, "exchange with %s: %d done, %d new requests, %d new max signal",
+		a.Name, len(a.Results), len(r.Requests), len(r.NewMaxSignal))
 
-	serv.mu.Lock()
-	defer serv.mu.Unlock()
+	return nil
+}
 
-	f := serv.fuzzers[a.Name]
-	if f == nil {
-		// This is possible if we called shutdownInstance,
-		// but already have a pending request from this instance in-flight.
-		log.Logf(1, "poll: fuzzer %v is not connected", a.Name)
+func (serv *RPCServer) updateFilteredCover(pcs []uint32) error {
+	if len(pcs) == 0 || serv.coverFilter == nil {
 		return nil
 	}
-	newMaxSignal := serv.maxSignal.Diff(a.MaxSignal.Deserialize())
-	if !newMaxSignal.Empty() {
-		serv.maxSignal.Merge(newMaxSignal)
-		serv.stats.maxSignal.set(len(serv.maxSignal))
-		for _, f1 := range serv.fuzzers {
-			if f1 == f {
-				continue
-			}
-			f1.newMaxSignal.Merge(newMaxSignal)
-		}
-	}
-	r.MaxSignal = f.newMaxSignal.Split(2000).Serialize()
-	if a.NeedCandidates {
-		r.Candidates = serv.mgr.candidateBatch(serv.batchSize)
+	// Note: ReportGenerator is already initialized if coverFilter is enabled.
+	rg, err := getReportGenerator(serv.cfg, serv.modules)
+	if err != nil {
+		return err
 	}
-	if len(r.Candidates) == 0 {
-		batchSize := serv.batchSize
-		// When the fuzzer starts, it pumps the whole corpus.
-		// If we do it using the final batchSize, it can be very slow
-		// (batch of size 6 can take more than 10 mins for 50K corpus and slow kernel).
-		// So use a larger batch initially (we use no stats as approximation of initial pump).
-		const initialBatch = 50
-		if len(a.Stats) == 0 && batchSize < initialBatch {
-			batchSize = initialBatch
-		}
-		for i := 0; i < batchSize && len(f.inputs) > 0; i++ {
-			last := len(f.inputs) - 1
-			r.NewInputs = append(r.NewInputs, f.inputs[last])
-			f.inputs[last] = rpctype.Input{}
-			f.inputs = f.inputs[:last]
+	filtered := 0
+	for _, pc := range pcs {
+		if serv.coverFilter[uint32(rg.RestorePC(pc))] != 0 {
+			filtered++
 		}
-		if len(f.inputs) == 0 {
-			f.inputs = nil
-		}
-	}
-	for _, inp := range r.NewInputs {
-		inp.Cover, inp.Signal = f.instModules.Decanonicalize(inp.Cover, inp.Signal)
 	}
-	log.Logf(4, "poll from %v: candidates=%v inputs=%v maxsignal=%v",
-		a.Name, len(r.Candidates), len(r.NewInputs), len(r.MaxSignal.Elems))
+	serv.stats.corpusCoverFiltered.add(filtered)
 	return nil
 }
 
 func (serv *RPCServer) shutdownInstance(name string) []byte {
-	serv.mu.Lock()
-	defer serv.mu.Unlock()
-
-	fuzzer := serv.fuzzers[name]
-	if fuzzer == nil {
+	var runner *Runner
+	if val, _ := serv.runners.LoadAndDelete(name); val != nil {
+		runner = val.(*Runner)
+	} else {
 		return nil
 	}
-	delete(serv.fuzzers, name)
-	return fuzzer.machineInfo
+
+	runner.mu.Lock()
+	if runner.requests == nil {
+		// We are supposed to invoke this code only once.
+		panic("Runner.requests is already nil")
+	}
+	oldRequests := runner.requests
+	runner.requests = nil
+	runner.mu.Unlock()
+
+	// If the object does not exist, there would be no oldRequests either.
+	fuzzerObj := serv.mgr.getFuzzer()
+	for _, req := range oldRequests {
+		// The VM likely crashed, so let's tell pkg/fuzzer to abort the affected jobs.
+		// TODO: distinguish between real VM crashes and regular VM restarts?
+		fuzzerObj.Done(req, &fuzzer.Result{Stop: true})
+	}
+	return runner.machineInfo
 }
 
-func (serv *RPCServer) LogMessage(m *rpctype.LogMessageReq, r *int) error {
-	log.Logf(m.Level, "%s: %s", m.Name, m.Message)
-	return nil
+func (serv *RPCServer) distributeMaxSignal(delta signal.Signal) {
+	serv.runners.Range(func(key, value any) bool {
+		runner := value.(*Runner)
+		runner.mu.Lock()
+		defer runner.mu.Unlock()
+		runner.newMaxSignal.Merge(delta)
+		return true
+	})
+}
+
+func (runner *Runner) doneRequest(resp rpctype.ExecutionResult, fuzzerObj *fuzzer.Fuzzer) {
+	runner.mu.Lock()
+	req, ok := runner.requests[resp.ID]
+	if ok {
+		delete(runner.requests, resp.ID)
+	}
+	runner.mu.Unlock()
+	if !ok {
+		// There may be a concurrent shutdownInstance() call.
+		return
+	}
+	info := &resp.Info
+	for i := 0; i < len(info.Calls); i++ {
+		call := &info.Calls[i]
+		call.Cover = runner.instModules.Canonicalize(call.Cover)
+		call.Signal = runner.instModules.Canonicalize(call.Signal)
+	}
+	info.Extra.Cover = runner.instModules.Canonicalize(info.Extra.Cover)
+	info.Extra.Signal = runner.instModules.Canonicalize(info.Extra.Signal)
+	fuzzerObj.Done(req, &fuzzer.Result{Info: info})
+}
+
+func (runner *Runner) newRequest(req *fuzzer.Request) rpctype.ExecutionRequest {
+	var signalFilter signal.Signal
+	if req.SignalFilter != nil {
+		newRawSignal := runner.instModules.Decanonicalize(req.SignalFilter.ToRaw())
+		// We don't care about specific priorities here.
+		signalFilter = signal.FromRaw(newRawSignal, 0)
+	}
+	id := runner.nextRequestID.Add(1)
+	runner.mu.Lock()
+	if runner.requests != nil {
+		runner.requests[id] = req
+	}
+	runner.mu.Unlock()
+	return rpctype.ExecutionRequest{
+		ID:           id,
+		ProgData:     req.Prog.Serialize(),
+		NeedCover:    req.NeedCover,
+		NeedSignal:   req.NeedSignal,
+		SignalFilter: signalFilter,
+		NeedHints:    req.NeedHints,
+	}
 }
diff --git a/syz-manager/stats.go b/syz-manager/stats.go
index 35770c527..d85a1a78b 100644
--- a/syz-manager/stats.go
+++ b/syz-manager/stats.go
@@ -32,6 +32,8 @@ type Stats struct {
 	corpusCoverFiltered Stat
 	corpusSignal        Stat
 	maxSignal           Stat
+	triageQueueLen      Stat
+	fuzzerJobs          Stat
 
 	mu         sync.Mutex
 	namedStats map[string]uint64
@@ -73,6 +75,7 @@ func (stats *Stats) all() map[string]uint64 {
 		"signal":            stats.corpusSignal.get(),
 		"max signal":        stats.maxSignal.get(),
 		"rpc traffic (MB)":  stats.rpcTraffic.get() / 1e6,
+		"fuzzer jobs":       stats.fuzzerJobs.get(),
 	}
 	if stats.haveHub {
 		m["hub: send prog add"] = stats.hubSendProgAdd.get()
@@ -107,6 +110,17 @@ func (stats *Stats) mergeNamed(named map[string]uint64) {
 	}
 }
 
+func (stats *Stats) setNamed(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 {
+		stats.namedStats[k] = v
+	}
+}
+
 func (s *Stat) get() uint64 {
 	return atomic.LoadUint64((*uint64)(s))
 }