testing: prepare for the introduction of Run methods

The biggest change is that each test is now responsible for managing
the starting and stopping of its parallel subtests.

The "Main" test could be run as a tRunner as well. This shows that
the introduction of subtests is merely a generalization of and
consistent with the current semantics.

Change-Id: Ibf8388c08f85d4b2c0df69c069326762ed36a72e
Reviewed-on: https://go-review.googlesource.com/18893
Reviewed-by: Russ Cox <rsc@golang.org>

src/testing/testing.go

@@ -147,6 +147,7 @@ import (
 	"bytes"
 	"flag"
 	"fmt"
+	"io"
 	"os"
 	"runtime"
 	"runtime/debug"
@@ -197,14 +198,18 @@ var (
 type common struct {
 	mu       sync.RWMutex // guards output and failed
 	output   []byte       // Output generated by test or benchmark.
+	w        io.Writer    // For flushToParent.
 	failed   bool         // Test or benchmark has failed.
 	skipped  bool         // Test of benchmark has been skipped.
 	finished bool
 
+	parent   *common
+	name     string    // Name of test or benchmark.
 	start    time.Time // Time test or benchmark started
 	duration time.Duration
-	self     interface{}      // To be sent on signal channel when done.
-	signal   chan interface{} // Output for serial tests.
+	barrier  chan bool // To signal parallel subtests they may start.
+	signal   chan bool // To signal a test is done.
+	sub      []*T      // Queue of subtests to be run in parallel.
 }
 
 // Short reports whether the -test.short flag is set.
@@ -251,6 +256,22 @@ func decorate(s string) string {
 	return buf.String()
 }
 
+// flushToParent writes c.output to the parent after first writing the header
+// with the given format and arguments.
+func (c *common) flushToParent(format string, args ...interface{}) {
+	p := c.parent
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	fmt.Fprintf(p.w, format, args...)
+	fmt.Fprintln(p.w)
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	io.Copy(p.w, bytes.NewReader(c.output))
+	c.output = c.output[:0]
+}
+
 // fmtDuration returns a string representing d in the form "87.00s".
 func fmtDuration(d time.Duration) string {
 	return fmt.Sprintf("%.2fs", d.Seconds())
@@ -293,15 +314,17 @@ var _ TB = (*B)(nil)
 // may be called simultaneously from multiple goroutines.
 type T struct {
 	common
-	name          string // Name of test.
-	isParallel    bool
-	startParallel chan bool // Parallel tests will wait on this.
+	isParallel bool
+	context    *testContext // For running tests and subtests.
 }
 
 func (c *common) private() {}
 
 // Fail marks the function as having failed but continues execution.
 func (c *common) Fail() {
+	if c.parent != nil {
+		c.parent.Fail()
+	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.failed = true
@@ -437,8 +460,13 @@ func (t *T) Parallel() {
 	// in the test duration. Record the elapsed time thus far and reset the
 	// timer afterwards.
 	t.duration += time.Since(t.start)
-	t.signal <- (*T)(nil) // Release main testing loop
-	<-t.startParallel     // Wait for serial tests to finish
+
+	// Add to the list of tests to be released by the parent.
+	t.parent.sub = append(t.parent.sub, t)
+
+	t.signal <- true   // Release calling test.
+	<-t.parent.barrier // Wait for the parent test to complete.
+	t.context.waitParallel()
 	t.start = time.Now()
 }
 
@@ -449,8 +477,8 @@ type InternalTest struct {
 	F    func(*T)
 }
 
-func tRunner(t *T, test *InternalTest) {
-	// When this goroutine is done, either because test.F(t)
+func tRunner(t *T, fn func(t *T)) {
+	// When this goroutine is done, either because fn(t)
 	// returned normally or because a test failure triggered
 	// a call to runtime.Goexit, record the duration and send
 	// a signal saying that the test is done.
@@ -466,14 +494,87 @@ func tRunner(t *T, test *InternalTest) {
 			t.report()
 			panic(err)
 		}
-		t.signal <- t
+
+		if len(t.sub) > 0 {
+			// Run parallel subtests.
+			// Decrease the running count for this test.
+			t.context.release()
+			// Release the parallel subtests.
+			close(t.barrier)
+			// Wait for subtests to complete.
+			for _, sub := range t.sub {
+				<-sub.signal
+			}
+			if !t.isParallel {
+				// Reacquire the count for sequential tests. See comment in Run.
+				t.context.waitParallel()
+			}
+		} else if t.isParallel {
+			// Only release the count for this test if it was run as a parallel
+			// test. See comment in Run method.
+			t.context.release()
+		}
+		t.report() // Report after all subtests have finished.
+
+		t.signal <- true
 	}()
 
 	t.start = time.Now()
-	test.F(t)
+	fn(t)
 	t.finished = true
 }
 
+// testContext holds all fields that are common to all tests. This includes
+// synchronization primitives to run at most *parallel tests.
+type testContext struct {
+	mu sync.Mutex
+
+	// Channel used to signal tests that are ready to be run in parallel.
+	startParallel chan bool
+
+	// running is the number of tests currently running in parallel.
+	// This does not include tests that are waiting for subtests to complete.
+	running int
+
+	// numWaiting is the number tests waiting to be run in parallel.
+	numWaiting int
+
+	// maxParallel is a copy of the parallel flag.
+	maxParallel int
+}
+
+func newTestContext(maxParallel int) *testContext {
+	return &testContext{
+		startParallel: make(chan bool),
+		maxParallel:   *parallel,
+		running:       1, // Set the count to 1 for the main (sequential) test.
+	}
+}
+
+func (c *testContext) waitParallel() {
+	c.mu.Lock()
+	if c.running < c.maxParallel {
+		c.running++
+		c.mu.Unlock()
+		return
+	}
+	c.numWaiting++
+	c.mu.Unlock()
+	<-c.startParallel
+}
+
+func (c *testContext) release() {
+	c.mu.Lock()
+	if c.numWaiting == 0 {
+		c.running--
+		c.mu.Unlock()
+		return
+	}
+	c.numWaiting--
+	c.mu.Unlock()
+	c.startParallel <- true // Pick a waiting test to be run.
+}
+
 // An internal function but exported because it is cross-package; part of the implementation
 // of the "go test" command.
 func Main(matchString func(pat, str string) (bool, error), tests []InternalTest, benchmarks []InternalBenchmark, examples []InternalExample) {
@@ -526,15 +627,18 @@ func (m *M) Run() int {
 }
 
 func (t *T) report() {
+	if t.parent == nil {
+		return
+	}
 	dstr := fmtDuration(t.duration)
-	format := "--- %s: %s (%s)\n%s"
+	format := "--- %s: %s (%s)"
 	if t.Failed() {
-		fmt.Printf(format, "FAIL", t.name, dstr, t.output)
+		t.flushToParent(format, "FAIL", t.name, dstr)
 	} else if *chatty {
 		if t.Skipped() {
-			fmt.Printf(format, "SKIP", t.name, dstr, t.output)
+			t.flushToParent(format, "SKIP", t.name, dstr)
 		} else {
-			fmt.Printf(format, "PASS", t.name, dstr, t.output)
+			t.flushToParent(format, "PASS", t.name, dstr)
 		}
 	}
 }
@@ -547,63 +651,55 @@ func RunTests(matchString func(pat, str string) (bool, error), tests []InternalT
 	}
 	for _, procs := range cpuList {
 		runtime.GOMAXPROCS(procs)
-		// We build a new channel tree for each run of the loop.
-		// collector merges in one channel all the upstream signals from parallel tests.
-		// If all tests pump to the same channel, a bug can occur where a test
-		// kicks off a goroutine that Fails, yet the test still delivers a completion signal,
-		// which skews the counting.
-		var collector = make(chan interface{})
-
-		numParallel := 0
-		startParallel := make(chan bool)
-
-		for i := 0; i < len(tests); i++ {
-			matched, err := matchString(*match, tests[i].Name)
-			if err != nil {
-				fmt.Fprintf(os.Stderr, "testing: invalid regexp for -test.run: %s\n", err)
-				os.Exit(1)
-			}
-			if !matched {
-				continue
-			}
-			testName := tests[i].Name
-			t := &T{
-				common: common{
-					signal: make(chan interface{}),
-				},
-				name:          testName,
-				startParallel: startParallel,
-			}
-			t.self = t
-			if *chatty {
-				fmt.Printf("=== RUN   %s\n", t.name)
-			}
-			go tRunner(t, &tests[i])
-			out := (<-t.signal).(*T)
-			if out == nil { // Parallel run.
-				go func() {
-					collector <- <-t.signal
-				}()
-				numParallel++
-				continue
-			}
-			t.report()
-			ok = ok && !out.Failed()
+		ctx := newTestContext(*parallel)
+		t := &T{
+			common: common{
+				signal:  make(chan bool),
+				barrier: make(chan bool),
+				w:       os.Stdout,
+			},
+			context: ctx,
 		}
 
-		running := 0
-		for numParallel+running > 0 {
-			if running < *parallel && numParallel > 0 {
-				startParallel <- true
-				running++
-				numParallel--
-				continue
+		tRunner(t, func(t *T) {
+			for i := 0; i < len(tests); i++ {
+				// TODO: a version of this will be the Run method.
+				matched, err := matchString(*match, tests[i].Name)
+				if err != nil {
+					fmt.Fprintf(os.Stderr, "testing: invalid regexp for -test.run: %s\n", err)
+					os.Exit(1)
+				}
+				if !matched {
+					continue
+				}
+				testName := tests[i].Name
+				t := &T{
+					common: common{
+						barrier: make(chan bool),
+						signal:  make(chan bool),
+						name:    testName,
+						parent:  &t.common,
+					},
+					context: t.context,
+				}
+
+				if *chatty {
+					fmt.Printf("=== RUN   %s\n", t.name)
+				}
+				// Instead of reducing the running count of this test before calling the
+				// tRunner and increasing it afterwards, we rely on tRunner keeping the
+				// count correct. This ensures that a sequence of sequential tests runs
+				// without being preempted, even when their parent is a parallel test. This
+				// may especially reduce surprises if *parallel == 1.
+				go tRunner(t, tests[i].F)
+				<-t.signal
 			}
-			t := (<-collector).(*T)
-			t.report()
-			ok = ok && !t.Failed()
-			running--
-		}
+			// Run catching the signal rather than the tRunner as a separate
+			// goroutine to avoid adding a goroutine during the sequential
+			// phase as this pollutes the stacktrace output when aborting.
+			go func() { <-t.signal }()
+		})
+		ok = ok && !t.Failed()
 	}
 	return
 }