syz-verifier: add exectask.go tests, fix races

4 files changed, 124 insertions, 25 deletions

diff --git a/syz-verifier/exectask.go b/syz-verifier/exectask.go
index 052469ed7..e7e366d49 100644
--- a/syz-verifier/exectask.go
+++ b/syz-verifier/exectask.go
@@ -43,40 +43,53 @@ func (t *ExecTask) ToRPC() *rpctype.ExecTask {
 	}
 }
 
-var (
-	ChanMapMutex           = sync.Mutex{}
-	TaskIDToExecResultChan = map[int64]ExecResultChan{}
-	TaskCounter            = int64(-1)
-)
+type ExecTaskFactory struct {
+	chanMapMutex           sync.Mutex
+	taskIDToExecResultChan map[int64]ExecResultChan
+	taskCounter            int64
+}
+
+func MakeExecTaskFactory() *ExecTaskFactory {
+	return &ExecTaskFactory{
+		taskIDToExecResultChan: make(map[int64]ExecResultChan),
+		taskCounter:            -1,
+	}
+}
 
 type ExecResultChan chan *ExecResult
 
-func MakeExecTask(prog *prog.Prog) *ExecTask {
+func (factory *ExecTaskFactory) MakeExecTask(prog *prog.Prog) *ExecTask {
 	task := &ExecTask{
 		CreationTime:   time.Now(),
 		Program:        prog,
 		ExecResultChan: make(ExecResultChan),
-		ID:             atomic.AddInt64(&TaskCounter, 1),
+		ID:             atomic.AddInt64(&factory.taskCounter, 1),
 	}
 
-	ChanMapMutex.Lock()
-	defer ChanMapMutex.Unlock()
-	TaskIDToExecResultChan[task.ID] = task.ExecResultChan
+	factory.chanMapMutex.Lock()
+	defer factory.chanMapMutex.Unlock()
+	factory.taskIDToExecResultChan[task.ID] = task.ExecResultChan
 
 	return task
 }
 
-func DeleteExecTask(task *ExecTask) {
-	ChanMapMutex.Lock()
-	defer ChanMapMutex.Unlock()
-	delete(TaskIDToExecResultChan, task.ID)
+func (factory *ExecTaskFactory) ExecTasksQueued() int {
+	factory.chanMapMutex.Lock()
+	defer factory.chanMapMutex.Unlock()
+	return len(factory.taskIDToExecResultChan)
+}
+
+func (factory *ExecTaskFactory) DeleteExecTask(task *ExecTask) {
+	factory.chanMapMutex.Lock()
+	defer factory.chanMapMutex.Unlock()
+	delete(factory.taskIDToExecResultChan, task.ID)
 }
 
-func GetExecResultChan(taskID int64) ExecResultChan {
-	ChanMapMutex.Lock()
-	defer ChanMapMutex.Unlock()
+func (factory *ExecTaskFactory) GetExecResultChan(taskID int64) ExecResultChan {
+	factory.chanMapMutex.Lock()
+	defer factory.chanMapMutex.Unlock()
 
-	return TaskIDToExecResultChan[taskID]
+	return factory.taskIDToExecResultChan[taskID]
 }
 
 func MakeExecTaskQueue() *ExecTaskQueue {
@@ -88,22 +101,28 @@ func MakeExecTaskQueue() *ExecTaskQueue {
 // ExecTaskQueue respects the pq.priority. Internally it is a thread-safe PQ.
 type ExecTaskQueue struct {
 	pq ExecTaskPriorityQueue
+	mu sync.Mutex
 }
 
 // PopTask return false if no tasks are available.
 func (q *ExecTaskQueue) PopTask() (*ExecTask, bool) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
 	if q.pq.Len() == 0 {
 		return nil, false
 	}
-
 	return heap.Pop(&q.pq).(*ExecTask), true
 }
 
 func (q *ExecTaskQueue) PushTask(task *ExecTask) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
 	heap.Push(&q.pq, task)
 }
 
 func (q *ExecTaskQueue) Len() int {
+	q.mu.Lock()
+	defer q.mu.Unlock()
 	return q.pq.Len()
 }
 
diff --git a/syz-verifier/exectask_test.go b/syz-verifier/exectask_test.go
new file mode 100644
index 000000000..3ffc8a94f
--- /dev/null
+++ b/syz-verifier/exectask_test.go
@@ -0,0 +1,77 @@
+// Copyright 2022 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"
+)
+
+func TestExecTask_MakeDelete(t *testing.T) {
+	program := getTestProgram(t)
+	taskFactory := MakeExecTaskFactory()
+	if l := taskFactory.ExecTasksQueued(); l != 0 {
+		t.Errorf("expected to see empty map, current size is %v", l)
+	}
+	task := taskFactory.MakeExecTask(program)
+	if l := taskFactory.ExecTasksQueued(); l != 1 {
+		t.Errorf("expected map len is 0, current size is %v", l)
+	}
+	taskFactory.DeleteExecTask(task)
+	if l := taskFactory.ExecTasksQueued(); l != 0 {
+		t.Errorf("expected map len is 0, current size is %v", l)
+	}
+}
+
+func TestExecTask_ToRPC(t *testing.T) {
+	program := getTestProgram(t)
+	taskFactory := MakeExecTaskFactory()
+	task := taskFactory.MakeExecTask(program)
+	if task.ToRPC() == nil {
+		t.Errorf("rpcView generation failed")
+	}
+}
+
+func TestGetExecResultChan(t *testing.T) {
+	taskFactory := MakeExecTaskFactory()
+	if l := taskFactory.ExecTasksQueued(); l != 0 {
+		t.Errorf("expected to see empty map, current size is %v", l)
+	}
+	ch := taskFactory.GetExecResultChan(100)
+	if l := taskFactory.ExecTasksQueued(); l != 0 {
+		t.Errorf("expected to see empty map, current size is %v", l)
+	}
+	if ch != nil {
+		t.Errorf("expected to see nil channel")
+	}
+}
+
+func TestExecTaskQueue_PushTask(t *testing.T) {
+	q := MakeExecTaskQueue()
+	if l := q.Len(); l != 0 {
+		t.Errorf("expected to see zero len, current is %v", l)
+	}
+
+	taskFactory := MakeExecTaskFactory()
+	q.PushTask(taskFactory.MakeExecTask(getTestProgram(t)))
+	if l := q.Len(); l != 1 {
+		t.Errorf("expected to see single element, current size is %v", l)
+	}
+}
+
+func TestExecTaskQueue_PopTask(t *testing.T) {
+	q := MakeExecTaskQueue()
+	task, gotResult := q.PopTask()
+	if task != nil || gotResult != false {
+		t.Errorf("empty queue operation error")
+	}
+	program := getTestProgram(t)
+	taskFactory := MakeExecTaskFactory()
+	q.PushTask(taskFactory.MakeExecTask(program))
+	q.PushTask(taskFactory.MakeExecTask(program))
+	q.PushTask(taskFactory.MakeExecTask(program))
+	task, gotResult = q.PopTask()
+	if task == nil || gotResult == false {
+		t.Errorf("non-empty task or error was expected")
+	}
+}
diff --git a/syz-verifier/rpcserver.go b/syz-verifier/rpcserver.go
index 3726799c1..f80fb7a46 100644
--- a/syz-verifier/rpcserver.go
+++ b/syz-verifier/rpcserver.go
@@ -91,7 +91,7 @@ func (srv *RPCServer) UpdateUnsupported(a *rpctype.UpdateUnsupportedArgs, r *int
 func (srv *RPCServer) NextExchange(a *rpctype.NextExchangeArgs, r *rpctype.NextExchangeRes) error {
 	if a.Info.Calls != nil {
 		srv.stopWaitResult(a.Pool, a.VM, a.ExecTaskID)
-		PutExecResult(&ExecResult{
+		srv.vrf.PutExecResult(&ExecResult{
 			Pool:       a.Pool,
 			Hanged:     a.Hanged,
 			Info:       a.Info,
@@ -140,7 +140,7 @@ func (srv *RPCServer) cleanup(poolID, vmID int) {
 
 	// Signal error for every VM related task and let upper level logic to process it.
 	for taskID := range srv.vmTasksInProgress[vmTasksKey(poolID, vmID)] {
-		PutExecResult(&ExecResult{
+		srv.vrf.PutExecResult(&ExecResult{
 			Pool:       poolID,
 			ExecTaskID: taskID,
 			Crashed:    true,
diff --git a/syz-verifier/verifier.go b/syz-verifier/verifier.go
index 33dabebe0..4fcb73735 100644
--- a/syz-verifier/verifier.go
+++ b/syz-verifier/verifier.go
@@ -57,6 +57,7 @@ type Verifier struct {
 	tasksMutex     sync.Mutex
 	onTaskAdded    *sync.Cond
 	kernelEnvTasks [][]*ExecTaskQueue
+	taskFactory    *ExecTaskFactory
 }
 
 func (vrf *Verifier) Init() {
@@ -77,6 +78,8 @@ func (vrf *Verifier) Init() {
 		log.Fatalf("failed to initialise RPC server: %v", err)
 	}
 	vrf.srv = srv
+
+	vrf.taskFactory = MakeExecTaskFactory()
 }
 
 func (vrf *Verifier) StartProgramsAnalysis() {
@@ -126,8 +129,8 @@ func (vrf *Verifier) GetRunnerTask(kernel int, existing EnvDescr) *rpctype.ExecT
 	}
 }
 
-func PutExecResult(result *ExecResult) {
-	c := GetExecResultChan(result.ExecTaskID)
+func (vrf *Verifier) PutExecResult(result *ExecResult) {
+	c := vrf.taskFactory.GetExecResultChan(result.ExecTaskID)
 	c <- result
 }
 
@@ -176,8 +179,8 @@ func (vrf *Verifier) Run(prog *prog.Prog, env EnvDescr) (result []*ExecResult, e
 
 		go func() {
 			defer wg.Done()
-			task := MakeExecTask(prog)
-			defer DeleteExecTask(task)
+			task := vrf.taskFactory.MakeExecTask(prog)
+			defer vrf.taskFactory.DeleteExecTask(task)
 
 			vrf.tasksMutex.Lock()
 			q.PushTask(task)