pkg/fuzzer: introduce a request restarter layer

Make Result statuses more elaborate.
Instead of retrying inputs directly in rpc.go, extract this logic to a
separate entity in pkg/fuzzer/queue.

4 files changed, 83 insertions, 20 deletions

diff --git a/pkg/fuzzer/fuzzer_test.go b/pkg/fuzzer/fuzzer_test.go
index ab38fa783..5a903dedb 100644
--- a/pkg/fuzzer/fuzzer_test.go
+++ b/pkg/fuzzer/fuzzer_test.go
@@ -242,7 +242,7 @@ func (f *testFuzzer) registerExecutor(proc *executorProc) {
 				return err
 			}
 			if crash != "" {
-				res = &queue.Result{Stop: true}
+				res = &queue.Result{Status: queue.Crashed}
 				if !f.expectedCrashes[crash] {
 					return fmt.Errorf("unexpected crash: %q", crash)
 				}
diff --git a/pkg/fuzzer/job.go b/pkg/fuzzer/job.go
index 8f81ef9fa..9fd294ca0 100644
--- a/pkg/fuzzer/job.go
+++ b/pkg/fuzzer/job.go
@@ -164,7 +164,7 @@ func (job *triageJob) deflake(exec func(*queue.Request, ...execOpt) *queue.Resul
 			NeedCover:  true,
 			Stat:       stat,
 		}, &dontTriage{})
-		if result.Stop {
+		if result.Stop() {
 			stop = true
 			return
 		}
@@ -204,7 +204,7 @@ func (job *triageJob) minimize(newSignal signal.Signal) (stop bool) {
 					SignalFilterCall: call1,
 					Stat:             job.fuzzer.statExecMinimize,
 				})
-				if result.Stop {
+				if result.Stop() {
 					stop = true
 					return false
 				}
@@ -273,7 +273,7 @@ func (job *smashJob) run(fuzzer *Fuzzer) {
 			NeedSignal: queue.NewSignal,
 			Stat:       fuzzer.statExecSmash,
 		})
-		if result.Stop {
+		if result.Stop() {
 			return
 		}
 		if fuzzer.Config.Collide {
@@ -281,7 +281,7 @@ func (job *smashJob) run(fuzzer *Fuzzer) {
 				Prog: randomCollide(p, rnd),
 				Stat: fuzzer.statExecCollide,
 			})
-			if result.Stop {
+			if result.Stop() {
 				return
 			}
 		}
@@ -323,7 +323,7 @@ func (job *smashJob) faultInjection(fuzzer *Fuzzer) {
 			Prog: newProg,
 			Stat: fuzzer.statExecSmash,
 		})
-		if result.Stop {
+		if result.Stop() {
 			return
 		}
 		info := result.Info
@@ -343,6 +343,7 @@ func (job *hintsJob) run(fuzzer *Fuzzer) {
 	// First execute the original program twice to get comparisons from KCOV.
 	// The second execution lets us filter out flaky values, which seem to constitute ~30-40%.
 	p := job.p
+
 	var comps prog.CompMap
 	for i := 0; i < 2; i++ {
 		result := fuzzer.execute(fuzzer.smashQueue, &queue.Request{
@@ -350,7 +351,7 @@ func (job *hintsJob) run(fuzzer *Fuzzer) {
 			NeedHints: true,
 			Stat:      fuzzer.statExecSeed,
 		})
-		if result.Stop || result.Info == nil {
+		if result.Stop() || result.Info == nil {
 			return
 		}
 		if i == 0 {
@@ -373,6 +374,6 @@ func (job *hintsJob) run(fuzzer *Fuzzer) {
 				NeedSignal: queue.NewSignal,
 				Stat:       fuzzer.statExecHint,
 			})
-			return !result.Stop
+			return !result.Stop()
 		})
 }
diff --git a/pkg/fuzzer/queue/queue.go b/pkg/fuzzer/queue/queue.go
index 00e83a69e..cb5aa134b 100644
--- a/pkg/fuzzer/queue/queue.go
+++ b/pkg/fuzzer/queue/queue.go
@@ -72,7 +72,7 @@ func (r *Request) Wait(ctx context.Context) *Result {
 	r.initChannel()
 	select {
 	case <-ctx.Done():
-		return &Result{Stop: true}
+		return &Result{Status: ExecFailure}
 	case <-r.done:
 		return r.result
 	}
@@ -95,10 +95,23 @@ const (
 )
 
 type Result struct {
-	Info *ipc.ProgInfo
-	Stop bool
+	Info   *ipc.ProgInfo
+	Status Status
 }
 
+func (r *Result) Stop() bool {
+	return r.Status == ExecFailure || r.Status == Crashed
+}
+
+type Status int
+
+const (
+	Success     Status = iota
+	ExecFailure        // For e.g. serialization errors.
+	Crashed            // The VM crashed holding the request.
+	Restarted          // The VM was restarted holding the request.
+)
+
 // Executor describes the interface wanted by the producers of requests.
 // After a Request is submitted, it's expected that the consumer will eventually
 // take it and report the execution result via Done().
@@ -157,16 +170,28 @@ func (pq *PlainQueue) Submit(req *Request) {
 func (pq *PlainQueue) Next() *Request {
 	pq.mu.Lock()
 	defer pq.mu.Unlock()
-	if pq.pos < len(pq.queue) {
-		ret := pq.queue[pq.pos]
-		pq.queue[pq.pos] = nil
-		pq.pos++
-		if pq.stat != nil {
-			pq.stat.Add(-1)
-		}
-		return ret
+	return pq.nextLocked()
+}
+
+func (pq *PlainQueue) tryNext() *Request {
+	if !pq.mu.TryLock() {
+		return nil
 	}
-	return nil
+	defer pq.mu.Unlock()
+	return pq.nextLocked()
+}
+
+func (pq *PlainQueue) nextLocked() *Request {
+	if pq.pos == len(pq.queue) {
+		return nil
+	}
+	ret := pq.queue[pq.pos]
+	pq.queue[pq.pos] = nil
+	pq.pos++
+	if pq.stat != nil {
+		pq.stat.Add(-1)
+	}
+	return ret
 }
 
 // Order combines several different sources in a particular order.
diff --git a/pkg/fuzzer/queue/retry.go b/pkg/fuzzer/queue/retry.go
new file mode 100644
index 000000000..0b2e02ba5
--- /dev/null
+++ b/pkg/fuzzer/queue/retry.go
@@ -0,0 +1,37 @@
+// 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 queue
+
+type retryer struct {
+	pq   *PlainQueue
+	base Source
+}
+
+// Retry adds a layer that resends results with Status=Restarted.
+func Retry(base Source) Source {
+	return &retryer{
+		base: base,
+		pq:   Plain(),
+	}
+}
+
+func (r *retryer) Next() *Request {
+	req := r.pq.tryNext()
+	if req == nil {
+		req = r.base.Next()
+	}
+	if req != nil {
+		req.OnDone(r.done)
+	}
+	return req
+}
+
+func (r *retryer) done(req *Request, res *Result) bool {
+	// The input was on a restarted VM.
+	if res.Status == Restarted {
+		r.pq.Submit(req)
+		return false
+	}
+	return true
+}