self review

executor/executor.go

@@ -13,11 +13,11 @@ import (
 )
 
 // Executor sequences the concurrent execution of
-// tasks with arbitrary keys on-the-fly.
+// tasks with arbitrary conflicts on-the-fly.
 //
 // Executor ensures that conflicting tasks
 // are executed in the order they were queued.
-// Tasks with no keys are executed immediately.
+// Tasks with no conflicts are executed immediately.
 type Executor struct {
 	metrics Metrics
 
@@ -93,9 +93,11 @@ func (e *Executor) runTask(t *task) {
 
 	t.l.Lock()
 	for _, bt := range t.blocking {
+		// Reads would have zero dependencies
 		if bt.dependencies.Load() == 0 || bt.dependencies.Add(-1) > 0 {
 			continue
 		}
+		// This might happen in Read-after-Read, so don't enqueue again
 		if !bt.executed {
 			e.executable <- bt
 		}
@@ -112,7 +114,7 @@ func (e *Executor) runTask(t *task) {
 func (e *Executor) Run(keys state.Keys, f func() error) {
 	e.outstanding.Add(1)
 
-	// Generate task
+	// Add task to map
 	id := len(e.tasks)
 	t := &task{
 		id:       id,
@@ -163,6 +165,7 @@ func (e *Executor) Run(keys state.Keys, f func() error) {
 		e.update(id, k, v)
 	}
 
+	// Start execution if there are no blocking dependencies
 	if t.dependencies.Load() > 0 {
 		if e.metrics != nil {
 			e.metrics.RecordBlocked()

executor/executor_test.go

@@ -5,7 +5,6 @@ package executor
 
 import (
 	"errors"
-	"fmt"
 	"sync"
 	"testing"
 	"time"
@@ -169,7 +168,7 @@ func TestEarlyExit(t *testing.T) {
 		})
 	}
 	require.ErrorIs(e.Wait(), terr) // no task running
-	require.True(len(completed) < 500)
+	require.Less(len(completed), 500)
 }
 
 func TestStop(t *testing.T) {
@@ -295,7 +294,6 @@ func TestWriteThenRead(t *testing.T) {
 		})
 	}
 	require.NoError(e.Wait())
-	fmt.Printf("completed %v\n", completed)
 	require.Equal(0, completed[0]) // Write first to execute
 	// 1..99 are ran in parallel, so non-deterministic
 	require.Len(completed, 100)
@@ -330,7 +328,6 @@ func TestReadThenWrite(t *testing.T) {
 		})
 	}
 	require.NoError(e.Wait())
-	fmt.Printf("completed %v\n", completed)
 	// 0..9 are ran in parallel, so non-deterministic
 	require.Equal(10, completed[10]) // First write to execute
 	// 11..99 are ran in parallel, so non-deterministic
@@ -369,7 +366,6 @@ func TestWriteThenReadRepeated(t *testing.T) {
 		})
 	}
 	require.NoError(e.Wait())
-	fmt.Printf("completed %v\n", completed)
 	require.Equal(0, completed[0]) // First write to execute
 	// 1..48 are ran in parallel, so non-deterministic
 	require.Equal(49, completed[49]) // Second write to execute
@@ -409,7 +405,6 @@ func TestReadThenWriteRepeated(t *testing.T) {
 		})
 	}
 	require.NoError(e.Wait())
-	fmt.Printf("completed %v\n", completed)
 	// 0..9 are ran in parallel, so non-deterministic
 	require.Equal(10, completed[10])
 	require.Equal(11, completed[11])