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.

2 files changed, 74 insertions, 12 deletions

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
+}