pkg/vminfo: run programs interactively

Use the same interfaces as the fuzzer.
Now syz-manager no longer needs to treat machine check executions
differently.

5 files changed, 135 insertions, 203 deletions

diff --git a/pkg/vminfo/features.go b/pkg/vminfo/features.go
index 2ba76a255..ab71e1168 100644
--- a/pkg/vminfo/features.go
+++ b/pkg/vminfo/features.go
@@ -8,8 +8,8 @@ import (
 	"strings"
 
 	"github.com/google/syzkaller/pkg/flatrpc"
+	"github.com/google/syzkaller/pkg/fuzzer/queue"
 	"github.com/google/syzkaller/pkg/ipc"
-	"github.com/google/syzkaller/pkg/rpctype"
 	"github.com/google/syzkaller/prog"
 )
 
@@ -46,14 +46,24 @@ type featureResult struct {
 	reason string
 }
 
-func (ctx *checkContext) checkFeatures() {
+func (ctx *checkContext) startFeaturesCheck() {
 	testProg := ctx.target.DataMmapProg()
 	for feat := range flatrpc.EnumNamesFeature {
 		feat := feat
-		ctx.pendingRequests++
 		go func() {
 			envFlags, execFlags := ctx.featureToFlags(feat)
-			res := ctx.execProg(testProg, envFlags, execFlags)
+			req := &queue.Request{
+				Prog:         testProg,
+				ReturnOutput: true,
+				ReturnError:  true,
+				ExecOpts: &ipc.ExecOpts{
+					EnvFlags:   envFlags,
+					ExecFlags:  execFlags,
+					SandboxArg: ctx.cfg.SandboxArg,
+				},
+			}
+			ctx.executor.Submit(req)
+			res := req.Wait(ctx.ctx)
 			reason := ctx.featureSucceeded(feat, testProg, res)
 			ctx.features <- featureResult{feat, reason}
 		}()
@@ -161,8 +171,8 @@ func (ctx *checkContext) featureToFlags(feat flatrpc.Feature) (ipc.EnvFlags, ipc
 // This generally checks that just all syscalls were executed and succeed,
 // for coverage features we also check that we got actual coverage.
 func (ctx *checkContext) featureSucceeded(feat flatrpc.Feature, testProg *prog.Prog,
-	res rpctype.ExecutionResult) string {
-	if res.Error != "" {
+	res *queue.Result) string {
+	if res.Status != queue.Success {
 		if len(res.Output) != 0 {
 			return string(res.Output)
 		}
diff --git a/pkg/vminfo/linux_test.go b/pkg/vminfo/linux_test.go
index e9be133c1..2a09a66cf 100644
--- a/pkg/vminfo/linux_test.go
+++ b/pkg/vminfo/linux_test.go
@@ -21,14 +21,6 @@ import (
 func TestLinuxSyscalls(t *testing.T) {
 	cfg := testConfig(t, targets.Linux, targets.AMD64)
 	checker := New(cfg)
-	_, checkProgs := checker.StartCheck()
-	t.Logf("got %v test programs", len(checkProgs))
-	if len(checkProgs) > 1000 {
-		// This is just a sanity check that we don't do something stupid accidentally.
-		// If it grows above the limit intentionally, the limit can be increased.
-		// Currently we have 641 (when we failed to properly dedup syscall tests, it was 4349).
-		t.Fatal("too many test programs")
-	}
 	filesystems := []string{
 		// Without sysfs, the checks would also disable mount().
 		"", "sysfs", "ext4", "binder", "",
@@ -45,8 +37,10 @@ func TestLinuxSyscalls(t *testing.T) {
 			Data:   []byte(strings.Join(filesystems, "\nnodev\t")),
 		},
 	}
-	results, featureInfos := createSuccessfulResults(t, cfg.Target, checkProgs)
-	enabled, disabled, features, err := checker.FinishCheck(files, results, featureInfos)
+	stop := make(chan struct{})
+	go createSuccessfulResults(checker, stop)
+	enabled, disabled, features, err := checker.Run(files, allFeatures())
+	close(stop)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -70,6 +64,17 @@ func TestLinuxSyscalls(t *testing.T) {
 		}
 		t.Errorf("disabled call %v: %v", call.Name, reason)
 	}
+	for _, id := range cfg.Syscalls {
+		call := cfg.Target.Syscalls[id]
+		if enabled[call] && disabled[call] != "" {
+			t.Fatalf("%s is both enabled and disabled", call.Name)
+		}
+		expected := !expectDisabled[call.Name]
+		got := enabled[call]
+		if expected != got {
+			t.Errorf("%s: expected %t, got %t", call.Name, expected, got)
+		}
+	}
 	expectEnabled := len(cfg.Syscalls) - len(expectDisabled)
 	if len(enabled) != expectEnabled {
 		t.Errorf("enabled only %v calls out of %v", len(enabled), expectEnabled)
diff --git a/pkg/vminfo/syscalls.go b/pkg/vminfo/syscalls.go
index 78cc973c8..863c995bd 100644
--- a/pkg/vminfo/syscalls.go
+++ b/pkg/vminfo/syscalls.go
@@ -4,18 +4,15 @@
 package vminfo
 
 import (
-	"bytes"
-	"encoding/gob"
+	"context"
 	"fmt"
-	"slices"
 	"strings"
 	"syscall"
 
 	"github.com/google/syzkaller/pkg/flatrpc"
-	"github.com/google/syzkaller/pkg/hash"
+	"github.com/google/syzkaller/pkg/fuzzer/queue"
 	"github.com/google/syzkaller/pkg/ipc"
 	"github.com/google/syzkaller/pkg/mgrconfig"
-	"github.com/google/syzkaller/pkg/rpctype"
 	"github.com/google/syzkaller/prog"
 	"github.com/google/syzkaller/sys/targets"
 )
@@ -26,44 +23,19 @@ import (
 // and provides primitives for reading target VM files, checking if a file can be opened,
 // executing test programs on the target VM, etc.
 //
-// To make use of this type simpler, we collect all test programs that need
-// to be executed on the target into a batch, send them to the target VM once,
-// then get results and finish the check. This means that impl.syscallCheck
-// cannot e.g. issue one test program, look at results, and then issue another one.
-// This is achieved by starting each impl.syscallCheck in a separate goroutine
-// and then waiting when it will call ctx.execRaw to submit syscalls that
-// need to be executed on the target. Once all impl.syscallCheck submit
-// their test syscalls, we know that we collected all of them.
-// impl.syscallCheck may also decide to read a file on the target VM instead
-// of executing a test program, this also counts as submitting an empty test program.
-// This means that impl.syscallCheck cannot execute a test program after reading a file,
-// but can do these things in opposite order (since all files are known ahead of time).
-// These rules are bit brittle, but all of the checkers are unit-tested
-// and misuse (trying to execute 2 programs, etc) will either crash or hang in tests.
-// Theoretically we can execute more than 1 program per checker, but it will
-// require some special arrangements, e.g. see handling of PseudoSyscallDeps.
-//
 // The external interface of this type contains only 2 methods:
 // startCheck - starts impl.syscallCheck goroutines and collects all test programs in progs,
 // finishCheck - accepts results of program execution, unblocks impl.syscallCheck goroutines,
 //
 //	waits and returns results of checking.
 type checkContext struct {
-	impl    checker
-	cfg     *mgrconfig.Config
-	target  *prog.Target
-	sandbox ipc.EnvFlags
-	// Checkers use requests channel to submit their test programs,
-	// main goroutine will wait for exactly pendingRequests message on this channel
-	// (similar to sync.WaitGroup, pendingRequests is incremented before starting
-	// a goroutine that will send on requests).
-	requests        chan []*rpctype.ExecutionRequest
-	pendingRequests int
-	// Ready channel is closed after we've recevied results of execution of test
-	// programs and file contents. After this results maps and fs are populated.
-	ready   chan bool
-	results map[int64]rpctype.ExecutionResult
-	fs      filesystem
+	ctx      context.Context
+	impl     checker
+	cfg      *mgrconfig.Config
+	target   *prog.Target
+	sandbox  ipc.EnvFlags
+	executor queue.Executor
+	fs       filesystem
 	// Once checking of a syscall is finished, the result is sent to syscalls.
 	// The main goroutine will wait for exactly pendingSyscalls messages.
 	syscalls        chan syscallResult
@@ -76,25 +48,26 @@ type syscallResult struct {
 	reason string
 }
 
-func newCheckContext(cfg *mgrconfig.Config, impl checker) *checkContext {
+func newCheckContext(ctx context.Context, cfg *mgrconfig.Config, impl checker,
+	executor queue.Executor) *checkContext {
 	sandbox, err := ipc.SandboxToFlags(cfg.Sandbox)
 	if err != nil {
 		panic(fmt.Sprintf("failed to parse sandbox: %v", err))
 	}
 	return &checkContext{
+		ctx:      ctx,
 		impl:     impl,
 		cfg:      cfg,
 		target:   cfg.Target,
 		sandbox:  sandbox,
-		requests: make(chan []*rpctype.ExecutionRequest),
-		results:  make(map[int64]rpctype.ExecutionResult),
+		executor: executor,
 		syscalls: make(chan syscallResult),
 		features: make(chan featureResult, 100),
-		ready:    make(chan bool),
 	}
 }
 
-func (ctx *checkContext) startCheck() []rpctype.ExecutionRequest {
+func (ctx *checkContext) start(fileInfos []flatrpc.FileInfo) {
+	ctx.fs = createVirtualFilesystem(fileInfos)
 	for _, id := range ctx.cfg.Syscalls {
 		call := ctx.target.Syscalls[id]
 		if call.Attrs.Disabled {
@@ -112,52 +85,24 @@ func (ctx *checkContext) startCheck() []rpctype.ExecutionRequest {
 		if ctx.cfg.SysTarget.HostFuzzer || ctx.target.OS == targets.TestOS {
 			syscallCheck = alwaysSupported
 		}
-
-		var depsReason chan string
-		deps := ctx.cfg.SysTarget.PseudoSyscallDeps[call.CallName]
-		if len(deps) != 0 {
-			ctx.pendingRequests++
-			depsReason = make(chan string, 1)
-			go func() {
-				depsReason <- ctx.supportedSyscalls(deps)
-			}()
-		}
-		ctx.pendingRequests++
 		go func() {
-			reason := syscallCheck(ctx, call)
-			ctx.waitForResults()
-			if reason == "" && depsReason != nil {
-				reason = <-depsReason
+			var reason string
+			deps := ctx.cfg.SysTarget.PseudoSyscallDeps[call.CallName]
+			if len(deps) != 0 {
+				reason = ctx.supportedSyscalls(deps)
+			}
+			// Only check the call if all its dependencies are satisfied.
+			if reason == "" {
+				reason = syscallCheck(ctx, call)
 			}
 			ctx.syscalls <- syscallResult{call, reason}
 		}()
 	}
-	ctx.checkFeatures()
-	var progs []rpctype.ExecutionRequest
-	dedup := make(map[hash.Sig]int64)
-	for i := 0; i < ctx.pendingRequests; i++ {
-		for _, req := range <-ctx.requests {
-			sig := hashReq(req)
-			req.ID = dedup[sig]
-			if req.ID != 0 {
-				continue
-			}
-			req.ID = int64(len(dedup) + 1)
-			dedup[sig] = req.ID
-			progs = append(progs, *req)
-		}
-	}
-	ctx.requests = nil
-	return progs
+	ctx.startFeaturesCheck()
 }
 
-func (ctx *checkContext) finishCheck(fileInfos []flatrpc.FileInfo, progs []rpctype.ExecutionResult,
-	featureInfos []flatrpc.FeatureInfo) (map[*prog.Syscall]bool, map[*prog.Syscall]string, Features, error) {
-	ctx.fs = createVirtualFilesystem(fileInfos)
-	for _, res := range progs {
-		ctx.results[res.ID] = res
-	}
-	close(ctx.ready)
+func (ctx *checkContext) wait(featureInfos []flatrpc.FeatureInfo) (
+	map[*prog.Syscall]bool, map[*prog.Syscall]string, Features, error) {
 	enabled := make(map[*prog.Syscall]bool)
 	disabled := make(map[*prog.Syscall]string)
 	for i := 0; i < ctx.pendingSyscalls; i++ {
@@ -292,16 +237,12 @@ func (ctx *checkContext) val(name string) uint64 {
 }
 
 func (ctx *checkContext) execRaw(calls []string, mode prog.DeserializeMode, root bool) *ipc.ProgInfo {
-	if ctx.requests == nil {
-		panic("only one test execution per checker is supported")
-	}
 	sandbox := ctx.sandbox
 	if root {
 		sandbox = 0
 	}
-	remain := calls
-	var requests []*rpctype.ExecutionRequest
-	for len(remain) != 0 {
+	info := &ipc.ProgInfo{}
+	for remain := calls; len(remain) != 0; {
 		// Don't put too many syscalls into a single program,
 		// it will have higher chances to time out.
 		ncalls := min(len(remain), prog.MaxCalls/2)
@@ -311,93 +252,39 @@ func (ctx *checkContext) execRaw(calls []string, mode prog.DeserializeMode, root
 		if err != nil {
 			panic(fmt.Sprintf("failed to deserialize: %v\n%v", err, progStr))
 		}
-		data, err := p.SerializeForExec()
-		if err != nil {
-			panic(fmt.Sprintf("failed to serialize test program: %v\n%s", err, progStr))
-		}
-		requests = append(requests, &rpctype.ExecutionRequest{
-			ProgData: slices.Clone(data), // clone to reduce memory usage
-			ExecOpts: ipc.ExecOpts{
+		// TODO: request that the program must be re-executed on the first failure.
+		req := &queue.Request{
+			Prog: p,
+			ExecOpts: &ipc.ExecOpts{
 				EnvFlags:   sandbox,
 				ExecFlags:  0,
 				SandboxArg: ctx.cfg.SandboxArg,
 			},
-		})
-	}
-	ctx.requests <- requests
-	<-ctx.ready
-	info := &ipc.ProgInfo{}
-	for _, req := range requests {
-		res, ok := ctx.results[req.ID]
-		if !ok {
-			panic(fmt.Sprintf("no result for request %v", req.ID))
 		}
-		if len(res.Info.Calls) == 0 {
-			panic(fmt.Sprintf("result for request %v has no calls", req.ID))
+		ctx.executor.Submit(req)
+		res := req.Wait(ctx.ctx)
+		if res.Status == queue.Success {
+			info.Calls = append(info.Calls, res.Info.Calls...)
+		} else if res.Status == queue.Crashed {
+			// Pretend these calls were not executed.
+			info.Calls = append(info.Calls, ipc.EmptyProgInfo(ncalls).Calls...)
+		} else {
+			// The program must have been either executed or not due to a crash.
+			panic(fmt.Sprintf("got unexpected execution status (%d) for the prog %s",
+				res.Status, progStr))
 		}
-		info.Calls = append(info.Calls, res.Info.Calls...)
 	}
 	if len(info.Calls) != len(calls) {
-		panic(fmt.Sprintf("got only %v results for program %v with %v calls:\n%s",
-			len(info.Calls), requests[0].ID, len(calls), strings.Join(calls, "\n")))
+		panic(fmt.Sprintf("got %v != %v results for program:\n%s",
+			len(info.Calls), len(calls), strings.Join(calls, "\n")))
 	}
 	return info
 }
 
-func (ctx *checkContext) execProg(p *prog.Prog, envFlags ipc.EnvFlags,
-	execFlags ipc.ExecFlags) rpctype.ExecutionResult {
-	if ctx.requests == nil {
-		panic("only one test execution per checker is supported")
-	}
-	data, err := p.SerializeForExec()
-	if err != nil {
-		panic(fmt.Sprintf("failed to serialize test program: %v\n%s", err, p.Serialize()))
-	}
-	req := &rpctype.ExecutionRequest{
-		ProgData:     data,
-		ReturnOutput: true,
-		ReturnError:  true,
-		ExecOpts: ipc.ExecOpts{
-			EnvFlags:   envFlags,
-			ExecFlags:  execFlags,
-			SandboxArg: ctx.cfg.SandboxArg,
-		},
-	}
-	ctx.requests <- []*rpctype.ExecutionRequest{req}
-	<-ctx.ready
-	res, ok := ctx.results[req.ID]
-	if !ok {
-		panic(fmt.Sprintf("no result for request %v", req.ID))
-	}
-	return res
-}
-
 func (ctx *checkContext) readFile(name string) ([]byte, error) {
-	ctx.waitForResults()
 	return ctx.fs.ReadFile(name)
 }
 
-func (ctx *checkContext) waitForResults() {
-	// If syscallCheck has already executed a program, then it's also waited for ctx.ready.
-	// If it hasn't, then we need to unblock the loop in startCheck by sending a nil request.
-	if ctx.requests == nil {
-		return
-	}
-	ctx.requests <- nil
-	<-ctx.ready
-	if ctx.fs == nil {
-		panic("filesystem should be initialized by now")
-	}
-}
-
-func hashReq(req *rpctype.ExecutionRequest) hash.Sig {
-	buf := new(bytes.Buffer)
-	if err := gob.NewEncoder(buf).Encode(req.ExecOpts); err != nil {
-		panic(err)
-	}
-	return hash.Hash(req.ProgData, buf.Bytes())
-}
-
 func alwaysSupported(ctx *checkContext, call *prog.Syscall) string {
 	return ""
 }
diff --git a/pkg/vminfo/vminfo.go b/pkg/vminfo/vminfo.go
index 8c5af606e..4f6ab7425 100644
--- a/pkg/vminfo/vminfo.go
+++ b/pkg/vminfo/vminfo.go
@@ -14,6 +14,7 @@ package vminfo
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -23,14 +24,15 @@ import (
 
 	"github.com/google/syzkaller/pkg/cover"
 	"github.com/google/syzkaller/pkg/flatrpc"
+	"github.com/google/syzkaller/pkg/fuzzer/queue"
 	"github.com/google/syzkaller/pkg/mgrconfig"
-	"github.com/google/syzkaller/pkg/rpctype"
 	"github.com/google/syzkaller/prog"
 	"github.com/google/syzkaller/sys/targets"
 )
 
 type Checker struct {
 	checker
+	source       queue.Source
 	checkContext *checkContext
 }
 
@@ -46,9 +48,12 @@ func New(cfg *mgrconfig.Config) *Checker {
 	default:
 		impl = new(stub)
 	}
+	ctx := context.Background()
+	executor := queue.Plain()
 	return &Checker{
 		checker:      impl,
-		checkContext: newCheckContext(cfg, impl),
+		source:       queue.Deduplicate(ctx, executor),
+		checkContext: newCheckContext(ctx, cfg, impl, executor),
 	}
 }
 
@@ -77,17 +82,25 @@ func (checker *Checker) MachineInfo(fileInfos []flatrpc.FileInfo) ([]cover.Kerne
 	return modules, info.Bytes(), nil
 }
 
-func (checker *Checker) StartCheck() ([]string, []rpctype.ExecutionRequest) {
-	return checker.checkFiles(), checker.checkContext.startCheck()
+func (checker *Checker) CheckFiles() []string {
+	return checker.checkFiles()
 }
 
-func (checker *Checker) FinishCheck(files []flatrpc.FileInfo, progs []rpctype.ExecutionResult,
-	featureInfos []flatrpc.FeatureInfo) (map[*prog.Syscall]bool, map[*prog.Syscall]string, Features, error) {
+func (checker *Checker) Run(files []flatrpc.FileInfo, featureInfos []flatrpc.FeatureInfo) (
+	map[*prog.Syscall]bool, map[*prog.Syscall]string, Features, error) {
 	ctx := checker.checkContext
 	checker.checkContext = nil
-	return ctx.finishCheck(files, progs, featureInfos)
+	ctx.start(files)
+	return ctx.wait(featureInfos)
 }
 
+// Implementation of the queue.Source interface.
+func (checker *Checker) Next() *queue.Request {
+	return checker.source.Next()
+}
+
+var _ queue.Source = &Checker{}
+
 type machineInfoFunc func(files filesystem, w io.Writer) (string, error)
 
 type checker interface {
diff --git a/pkg/vminfo/vminfo_test.go b/pkg/vminfo/vminfo_test.go
index 2f72131a5..535782a64 100644
--- a/pkg/vminfo/vminfo_test.go
+++ b/pkg/vminfo/vminfo_test.go
@@ -7,12 +7,13 @@ import (
 	"runtime"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/google/syzkaller/pkg/flatrpc"
+	"github.com/google/syzkaller/pkg/fuzzer/queue"
 	"github.com/google/syzkaller/pkg/host"
 	"github.com/google/syzkaller/pkg/ipc"
 	"github.com/google/syzkaller/pkg/mgrconfig"
-	"github.com/google/syzkaller/pkg/rpctype"
 	"github.com/google/syzkaller/prog"
 	"github.com/google/syzkaller/sys/targets"
 )
@@ -55,9 +56,10 @@ func TestSyscalls(t *testing.T) {
 				t.Parallel()
 				cfg := testConfig(t, target.OS, target.Arch)
 				checker := New(cfg)
-				_, checkProgs := checker.StartCheck()
-				results, featureInfos := createSuccessfulResults(t, cfg.Target, checkProgs)
-				enabled, disabled, _, err := checker.FinishCheck(nil, results, featureInfos)
+				stop := make(chan struct{})
+				go createSuccessfulResults(checker, stop)
+				enabled, disabled, _, err := checker.Run(nil, allFeatures())
+				close(stop)
 				if err != nil {
 					t.Fatal(err)
 				}
@@ -72,19 +74,38 @@ func TestSyscalls(t *testing.T) {
 	}
 }
 
-func createSuccessfulResults(t *testing.T, target *prog.Target,
-	progs []rpctype.ExecutionRequest) ([]rpctype.ExecutionResult, []flatrpc.FeatureInfo) {
-	var results []rpctype.ExecutionResult
-	for _, req := range progs {
-		p, err := target.DeserializeExec(req.ProgData, nil)
-		if err != nil {
-			t.Fatal(err)
+func allFeatures() []flatrpc.FeatureInfo {
+	var features []flatrpc.FeatureInfo
+	for feat := range flatrpc.EnumNamesFeature {
+		features = append(features, flatrpc.FeatureInfo{
+			Id: feat,
+		})
+	}
+	return features
+}
+
+func createSuccessfulResults(source queue.Source, stop chan struct{}) {
+	var count int
+	for {
+		select {
+		case <-stop:
+			return
+		case <-time.After(time.Millisecond):
+		}
+		req := source.Next()
+		if req == nil {
+			continue
 		}
-		res := rpctype.ExecutionResult{
-			ID: req.ID,
+		count++
+		if count > 1000 {
+			// This is just a sanity check that we don't do something stupid accidentally.
+			// If it grows above the limit intentionally, the limit can be increased.
+			// Currently we have 641 (when we failed to properly dedup syscall tests, it was 4349).
+			panic("too many test programs")
 		}
-		for range p.Calls {
-			res.Info.Calls = append(res.Info.Calls, ipc.CallInfo{
+		info := &ipc.ProgInfo{}
+		for range req.Prog.Calls {
+			info.Calls = append(info.Calls, ipc.CallInfo{
 				Cover:  []uint32{1},
 				Signal: []uint32{1},
 				Comps: map[uint64]map[uint64]bool{
@@ -92,15 +113,11 @@ func createSuccessfulResults(t *testing.T, target *prog.Target,
 				},
 			})
 		}
-		results = append(results, res)
-	}
-	var features []flatrpc.FeatureInfo
-	for feat := range flatrpc.EnumNamesFeature {
-		features = append(features, flatrpc.FeatureInfo{
-			Id: feat,
+		req.Done(&queue.Result{
+			Status: queue.Success,
+			Info:   info,
 		})
 	}
-	return results, features
 }
 
 func hostChecker(t *testing.T) (*Checker, []flatrpc.FileInfo) {