pkg/manager, syz-verifier, syz-manager: refactor corpus management and improve verifier

Extract corpus minimization logic into shared pkg/manager/corpus.go
for reuse by both syz-manager and syz-verifier.

Improve syz-verifier fuzzing implementation:
- Update request distribution strategy for better workload management
- Change result comparison logic to more accurately detect mismatches
- Print mismatches in more readable format for easier debugging
- Enforce restarting executor between program executions to avoid state pollution
- Fix coverage-guided fuzzing to properly track code coverage
- Fix maxSignal phase management for accurate signal tracking
- Simplify crash handling logic
- Refactor corpus management to use single goroutine with one channel

Enhance HTTP server monitoring:
- Add log, coverage statistics, and VM information to HTTP server views
- Enable HTTP server logging for better observability

Fix race condition where multiple executor goroutines call MaxSignal()
simultaneously from RPC server connection loops.

Author: matankalina

CONTRIBUTORS

@@ -143,4 +143,5 @@ Jeongjun Park
 Nikita Zhandarovich
 Jiacheng Xu
 Kuzey Arda Bulut
-Daniel Bransky
+Daniel Bransky
+Matan Kalina

pkg/manager/corpus.go

@@ -0,0 +1,89 @@
+// Copyright 2015 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 manager
+
+import (
+	"github.com/google/syzkaller/pkg/corpus"
+	"github.com/google/syzkaller/pkg/db"
+	"github.com/google/syzkaller/pkg/log"
+)
+
+// CorpusMinimizer provides the interface needed to minimize a corpus.
+type CorpusMinimizer struct {
+	Corpus         *corpus.Corpus
+	CorpusDB       *db.DB
+	Cover          bool
+	LastMinCorpus  int
+	SaturatedCalls map[string]bool
+	DisabledHashes map[string]struct{}
+	PhaseCheck     func() bool // returns true if we should proceed with minimization
+}
+
+// Minimize performs corpus minimization if conditions are met.
+// It should be called with appropriate locking in place.
+func (cm *CorpusMinimizer) Minimize() int {
+	// Don't minimize corpus until we have triaged all inputs from it.
+	// During corpus triage it would happen very often since we are actively adding inputs,
+	// and presumably the persistent corpus was reasonably minimal, and we don't use it for fuzzing yet.
+	if cm.PhaseCheck != nil && !cm.PhaseCheck() {
+		return cm.LastMinCorpus
+	}
+	currSize := cm.Corpus.StatProgs.Val()
+	if currSize <= cm.LastMinCorpus*103/100 {
+		return cm.LastMinCorpus
+	}
+	cm.Corpus.Minimize(cm.Cover)
+	newSize := cm.Corpus.StatProgs.Val()
+
+	log.Logf(1, "minimized corpus: %v -> %v", currSize, newSize)
+	cm.LastMinCorpus = newSize
+
+	// From time to time we get corpus explosion due to different reason:
+	// generic bugs, per-OS bugs, problems with fallback coverage, kcov bugs, etc.
+	// This has bad effect on the instance and especially on instances
+	// connected via hub. Do some per-syscall sanity checking to prevent this.
+	for call, info := range cm.Corpus.CallCover() {
+		if cm.Cover {
+			// If we have less than 1K inputs per this call,
+			// accept all new inputs unconditionally.
+			if info.Count < 1000 {
+				continue
+			}
+			// If we have more than 3K already, don't accept any more.
+			// Between 1K and 3K look at amount of coverage we are getting from these programs.
+			// Empirically, real coverage for the most saturated syscalls is ~30-60
+			// per program (even when we have a thousand of them). For explosion
+			// case coverage tend to be much lower (~0.3-5 per program).
+			if info.Count < 3000 && len(info.Cover)/info.Count >= 10 {
+				continue
+			}
+		} else {
+			// If we don't have real coverage, signal is weak.
+			// If we have more than several hundreds, there is something wrong.
+			if info.Count < 300 {
+				continue
+			}
+		}
+		if cm.SaturatedCalls[call] {
+			continue
+		}
+		cm.SaturatedCalls[call] = true
+		log.Logf(0, "coverage for %v has saturated, not accepting more inputs", call)
+	}
+
+	// Clean up the corpus database
+	for key := range cm.CorpusDB.Records {
+		ok1 := cm.Corpus.Item(key) != nil
+		_, ok2 := cm.DisabledHashes[key]
+		if !ok1 && !ok2 {
+			cm.CorpusDB.Delete(key)
+		}
+	}
+	if err := cm.CorpusDB.Flush(); err != nil {
+		log.Fatalf("failed to save corpus database: %v", err)
+	}
+	cm.CorpusDB.BumpVersion(CurrentDBVersion)
+
+	return newSize
+}

syz-manager/manager.go

@@ -1032,68 +1032,20 @@ func (mgr *Manager) addNewCandidates(candidates []fuzzer.Candidate) {
 }
 
 func (mgr *Manager) minimizeCorpusLocked() {
-	// Don't minimize corpus until we have triaged all inputs from it.
-	// During corpus triage it would happen very often since we are actively adding inputs,
-	// and presumably the persistent corpus was reasonably minimial, and we don't use it for fuzzing yet.
-	if mgr.phase < phaseTriagedCorpus {
-		return
-	}
-	currSize := mgr.corpus.StatProgs.Val()
-	if currSize <= mgr.lastMinCorpus*103/100 {
-		return
-	}
-	mgr.corpus.Minimize(mgr.cfg.Cover)
-	newSize := mgr.corpus.StatProgs.Val()
-
-	log.Logf(1, "minimized corpus: %v -> %v", currSize, newSize)
-	mgr.lastMinCorpus = newSize
-
-	// From time to time we get corpus explosion due to different reason:
-	// generic bugs, per-OS bugs, problems with fallback coverage, kcov bugs, etc.
-	// This has bad effect on the instance and especially on instances
-	// connected via hub. Do some per-syscall sanity checking to prevent this.
-	for call, info := range mgr.corpus.CallCover() {
-		if mgr.cfg.Cover {
-			// If we have less than 1K inputs per this call,
-			// accept all new inputs unconditionally.
-			if info.Count < 1000 {
-				continue
-			}
-			// If we have more than 3K already, don't accept any more.
-			// Between 1K and 3K look at amount of coverage we are getting from these programs.
-			// Empirically, real coverage for the most saturated syscalls is ~30-60
-			// per program (even when we have a thousand of them). For explosion
-			// case coverage tend to be much lower (~0.3-5 per program).
-			if info.Count < 3000 && len(info.Cover)/info.Count >= 10 {
-				continue
-			}
-		} else {
-			// If we don't have real coverage, signal is weak.
-			// If we have more than several hundreds, there is something wrong.
-			if info.Count < 300 {
-				continue
-			}
-		}
-		if mgr.saturatedCalls[call] {
-			continue
-		}
-		mgr.saturatedCalls[call] = true
-		log.Logf(0, "coverage for %v has saturated, not accepting more inputs", call)
+	cm := &manager.CorpusMinimizer{
+		Corpus:         mgr.corpus,
+		CorpusDB:       mgr.corpusDB,
+		Cover:          mgr.cfg.Cover,
+		LastMinCorpus:  mgr.lastMinCorpus,
+		SaturatedCalls: mgr.saturatedCalls,
+		DisabledHashes: mgr.disabledHashes,
+		PhaseCheck: func() bool {
+			return mgr.phase >= phaseTriagedCorpus
+		},
 	}
-
 	mgr.corpusDBMu.Lock()
 	defer mgr.corpusDBMu.Unlock()
-	for key := range mgr.corpusDB.Records {
-		ok1 := mgr.corpus.Item(key) != nil
-		_, ok2 := mgr.disabledHashes[key]
-		if !ok1 && !ok2 {
-			mgr.corpusDB.Delete(key)
-		}
-	}
-	if err := mgr.corpusDB.Flush(); err != nil {
-		log.Fatalf("failed to save corpus database: %v", err)
-	}
-	mgr.corpusDB.BumpVersion(manager.CurrentDBVersion)
+	mgr.lastMinCorpus = cm.Minimize()
 }
 
 func setGuiltyFiles(crash *dashapi.Crash, report *report.Report) {

syz-verifier/main.go

@@ -50,6 +50,10 @@ func Setup(name string, cfg *mgrconfig.Config, debug bool) (*Kernel, error) {
 		return nil, fmt.Errorf("failed to create reporter for %q: %w", name, err)
 	}
 
+	// Enforce deterministic execution for verifier
+	cfg.Experimental.ResetAccState = true // executor process restarts between program executions to clear accumulated kernel/VM stat.
+	//cfg.Procs = 1 // No parallel processes execution. When enabled slows down execution significantly.
+
 	kernel.serv, err = rpcserver.New(&rpcserver.RemoteConfig{
 		Config:  cfg,
 		Manager: kernel,
@@ -109,6 +113,7 @@ func main() {
 	}
 	osutil.MkdirAll(workdir)
 
+	log.EnableLogCaching(1000, 1<<20)
 	log.Logf(0, "initialized %d sources", len(sources))
 
 	vrf := &Verifier{