Parallel integration tests

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/tidwall/geojson v1.3.6
 	github.com/tidwall/gjson v1.14.3
 	github.com/tidwall/hashmap v1.6.1
+	github.com/tidwall/limiter v0.4.0
 	github.com/tidwall/match v1.1.1
 	github.com/tidwall/pretty v1.2.0
 	github.com/tidwall/redbench v0.1.0
@@ -31,6 +32,7 @@ require (
 	github.com/tidwall/sjson v1.2.4
 	github.com/xdg/scram v1.0.5
 	github.com/yuin/gopher-lua v0.0.0-20200816102855-ee81675732da
+	go.uber.org/atomic v1.5.0
 	go.uber.org/zap v1.13.0
 	golang.org/x/net v0.0.0-20220809184613-07c6da5e1ced
 	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211
@@ -95,7 +97,6 @@ require (
 	github.com/tidwall/tinyqueue v0.1.1 // indirect
 	github.com/xdg/stringprep v1.0.3 // indirect
 	go.opencensus.io v0.22.4 // indirect
-	go.uber.org/atomic v1.5.0 // indirect
 	go.uber.org/multierr v1.3.0 // indirect
 	go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee // indirect
 	golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa // indirect

go.sum

@@ -368,6 +368,8 @@ github.com/tidwall/grect v0.1.4 h1:dA3oIgNgWdSspFzn1kS4S/RDpZFLrIxAZOdJKjYapOg=
 github.com/tidwall/grect v0.1.4/go.mod h1:9FBsaYRaR0Tcy4UwefBX/UDcDcDy9V5jUcxHzv2jd5Q=
 github.com/tidwall/hashmap v1.6.1 h1:FIAHjKwcyOo1Y3/orsQO08floKhInbEX2VQv7CQRNuw=
 github.com/tidwall/hashmap v1.6.1/go.mod h1:hX452N3VtFD8okD3/6q/yOquJvJmYxmZ1H0nLtwkaxM=
+github.com/tidwall/limiter v0.4.0 h1:nj+7mS6aMDRzp15QTVDrgkun0def5/PfB4ogs5NlIVQ=
+github.com/tidwall/limiter v0.4.0/go.mod h1:n+qBGuSOgAvgcq1xUvo+mXWg8oBLQC8wkkheN9KZou0=
 github.com/tidwall/lotsa v1.0.2 h1:dNVBH5MErdaQ/xd9s769R31/n2dXavsQ0Yf4TMEHHw8=
 github.com/tidwall/lotsa v1.0.2/go.mod h1:X6NiU+4yHA3fE3Puvpnn1XMDrFZrE9JO2/w+UMuqgR8=
 github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=

internal/server/server.go

@@ -79,7 +79,9 @@ type Server struct {
 	started time.Time
 	config  *Config
 	epc     *endpoint.Manager
-	ln      net.Listener // server listener
+
+	lnmu sync.Mutex
+	ln   net.Listener // server listener
 
 	// env opts
 	geomParseOpts geojson.ParseOptions
@@ -296,7 +298,14 @@ func Serve(opts Options) error {
 		<-opts.Shutdown
 		s.stopServer.set(true)
 		log.Warnf("Shutting down...")
-		s.ln.Close()
+		s.lnmu.Lock()
+		ln := s.ln
+		s.ln = nil
+		s.lnmu.Unlock()
+
+		if ln != nil {
+			ln.Close()
+		}
 	}()
 
 	// Load the queue before the aof
@@ -432,6 +441,9 @@ func (s *Server) netServe() error {
 	if err != nil {
 		return err
 	}
+	s.lnmu.Lock()
+	s.ln = ln
+	s.lnmu.Unlock()
 
 	var wg sync.WaitGroup
 	defer func() {
@@ -445,7 +457,6 @@ func (s *Server) netServe() error {
 		ln.Close()
 		log.Debug("Client connection closed")
 	}()
-	s.ln = ln
 
 	log.Infof("Ready to accept connections at %s", ln.Addr())
 	var clientID int64

tests/aof_test.go

@@ -4,12 +4,11 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
-	"testing"
 )
 
-func subTestAOF(t *testing.T, mc *mockServer) {
+func subTestAOF(g *testGroup) {
-	runStep(t, mc, "loading", aof_loading_test)
+	g.regSubTest("loading", aof_loading_test)
-	// runStep(t, mc, "AOFMD5", aof_AOFMD5_test)
+	// g.regSubTest("AOFMD5", aof_AOFMD5_test)
 }
 
 func loadAOFAndClose(aof any) error {

tests/client_test.go

@@ -4,16 +4,15 @@ import (
 	"errors"
 	"fmt"
 	"strings"
-	"testing"
 
 	"github.com/gomodule/redigo/redis"
 	"github.com/tidwall/gjson"
 	"github.com/tidwall/pretty"
 )
 
-func subTestClient(t *testing.T, mc *mockServer) {
+func subTestClient(g *testGroup) {
-	runStep(t, mc, "OUTPUT", client_OUTPUT_test)
+	g.regSubTest("OUTPUT", client_OUTPUT_test)
-	runStep(t, mc, "CLIENT", client_CLIENT_test)
+	g.regSubTest("CLIENT", client_CLIENT_test)
 }
 
 func client_OUTPUT_test(mc *mockServer) error {

tests/fence_test.go

@@ -12,30 +12,29 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
-	"testing"
 	"time"
 
 	"github.com/gomodule/redigo/redis"
 	"github.com/tidwall/gjson"
 )
 
-func subTestFence(t *testing.T, mc *mockServer) {
+func subTestFence(g *testGroup) {
 
 	// Standard
-	runStep(t, mc, "basic", fence_basic_test)
+	g.regSubTest("basic", fence_basic_test)
-	runStep(t, mc, "channel message order", fence_channel_message_order_test)
+	g.regSubTest("channel message order", fence_channel_message_order_test)
-	runStep(t, mc, "detect inside,outside", fence_detect_inside_test)
+	g.regSubTest("detect inside,outside", fence_detect_inside_test)
 
 	// Roaming
-	runStep(t, mc, "roaming live", fence_roaming_live_test)
+	g.regSubTest("roaming live", fence_roaming_live_test)
-	runStep(t, mc, "roaming channel", fence_roaming_channel_test)
+	g.regSubTest("roaming channel", fence_roaming_channel_test)
-	runStep(t, mc, "roaming webhook", fence_roaming_webhook_test)
+	g.regSubTest("roaming webhook", fence_roaming_webhook_test)
 
 	// channel meta
-	runStep(t, mc, "channel meta", fence_channel_meta_test)
+	g.regSubTest("channel meta", fence_channel_meta_test)
 
 	// various
-	runStep(t, mc, "detect eecio", fence_eecio_test)
+	g.regSubTest("detect eecio", fence_eecio_test)
 }
 
 type fenceReader struct {

tests/json_test.go

@@ -1,11 +1,9 @@
 package tests
 
-import "testing"
+func subTestJSON(g *testGroup) {
-
+	g.regSubTest("basic", json_JSET_basic_test)
-func subTestJSON(t *testing.T, mc *mockServer) {
+	g.regSubTest("geojson", json_JSET_geojson_test)
-	runStep(t, mc, "basic", json_JSET_basic_test)
+	g.regSubTest("number", json_JSET_number_test)
-	runStep(t, mc, "geojson", json_JSET_geojson_test)
-	runStep(t, mc, "number", json_JSET_number_test)
 
 }
 func json_JSET_basic_test(mc *mockServer) error {

tests/keys_search_test.go

@@ -13,26 +13,26 @@ import (
 	"github.com/tidwall/gjson"
 )
 
-func subTestSearch(t *testing.T, mc *mockServer) {
+func subTestSearch(g *testGroup) {
-	runStep(t, mc, "KNN_BASIC", keys_KNN_basic_test)
+	g.regSubTest("KNN_BASIC", keys_KNN_basic_test)
-	runStep(t, mc, "KNN_RANDOM", keys_KNN_random_test)
+	g.regSubTest("KNN_RANDOM", keys_KNN_random_test)
-	runStep(t, mc, "KNN_CURSOR", keys_KNN_cursor_test)
+	g.regSubTest("KNN_CURSOR", keys_KNN_cursor_test)
-	runStep(t, mc, "NEARBY_SPARSE", keys_NEARBY_SPARSE_test)
+	g.regSubTest("NEARBY_SPARSE", keys_NEARBY_SPARSE_test)
-	runStep(t, mc, "WITHIN_CIRCLE", keys_WITHIN_CIRCLE_test)
+	g.regSubTest("WITHIN_CIRCLE", keys_WITHIN_CIRCLE_test)
-	runStep(t, mc, "WITHIN_SECTOR", keys_WITHIN_SECTOR_test)
+	g.regSubTest("WITHIN_SECTOR", keys_WITHIN_SECTOR_test)
-	runStep(t, mc, "INTERSECTS_CIRCLE", keys_INTERSECTS_CIRCLE_test)
+	g.regSubTest("INTERSECTS_CIRCLE", keys_INTERSECTS_CIRCLE_test)
-	runStep(t, mc, "INTERSECTS_SECTOR", keys_INTERSECTS_SECTOR_test)
+	g.regSubTest("INTERSECTS_SECTOR", keys_INTERSECTS_SECTOR_test)
-	runStep(t, mc, "WITHIN", keys_WITHIN_test)
+	g.regSubTest("WITHIN", keys_WITHIN_test)
-	runStep(t, mc, "WITHIN_CURSOR", keys_WITHIN_CURSOR_test)
+	g.regSubTest("WITHIN_CURSOR", keys_WITHIN_CURSOR_test)
-	runStep(t, mc, "WITHIN_CLIPBY", keys_WITHIN_CLIPBY_test)
+	g.regSubTest("WITHIN_CLIPBY", keys_WITHIN_CLIPBY_test)
-	runStep(t, mc, "INTERSECTS", keys_INTERSECTS_test)
+	g.regSubTest("INTERSECTS", keys_INTERSECTS_test)
-	runStep(t, mc, "INTERSECTS_CURSOR", keys_INTERSECTS_CURSOR_test)
+	g.regSubTest("INTERSECTS_CURSOR", keys_INTERSECTS_CURSOR_test)
-	runStep(t, mc, "INTERSECTS_CLIPBY", keys_INTERSECTS_CLIPBY_test)
+	g.regSubTest("INTERSECTS_CLIPBY", keys_INTERSECTS_CLIPBY_test)
-	runStep(t, mc, "SCAN_CURSOR", keys_SCAN_CURSOR_test)
+	g.regSubTest("SCAN_CURSOR", keys_SCAN_CURSOR_test)
-	runStep(t, mc, "SEARCH_CURSOR", keys_SEARCH_CURSOR_test)
+	g.regSubTest("SEARCH_CURSOR", keys_SEARCH_CURSOR_test)
-	runStep(t, mc, "MATCH", keys_MATCH_test)
+	g.regSubTest("MATCH", keys_MATCH_test)
-	runStep(t, mc, "FIELDS", keys_FIELDS_search_test)
+	g.regSubTest("FIELDS", keys_FIELDS_search_test)
-	runStep(t, mc, "BUFFER", keys_BUFFER_search_test)
+	g.regSubTest("BUFFER", keys_BUFFER_search_test)
 }
 
 func keys_KNN_basic_test(mc *mockServer) error {

tests/keys_test.go

@@ -5,37 +5,36 @@ import (
 	"fmt"
 	"math/rand"
 	"strings"
-	"testing"
 	"time"
 
 	"github.com/gomodule/redigo/redis"
 	"github.com/tidwall/gjson"
 )
 
-func subTestKeys(t *testing.T, mc *mockServer) {
+func subTestKeys(g *testGroup) {
-	runStep(t, mc, "BOUNDS", keys_BOUNDS_test)
+	g.regSubTest("BOUNDS", keys_BOUNDS_test)
-	runStep(t, mc, "DEL", keys_DEL_test)
+	g.regSubTest("DEL", keys_DEL_test)
-	runStep(t, mc, "DROP", keys_DROP_test)
+	g.regSubTest("DROP", keys_DROP_test)
-	runStep(t, mc, "RENAME", keys_RENAME_test)
+	g.regSubTest("RENAME", keys_RENAME_test)
-	runStep(t, mc, "RENAMENX", keys_RENAMENX_test)
+	g.regSubTest("RENAMENX", keys_RENAMENX_test)
-	runStep(t, mc, "EXPIRE", keys_EXPIRE_test)
+	g.regSubTest("EXPIRE", keys_EXPIRE_test)
-	runStep(t, mc, "FSET", keys_FSET_test)
+	g.regSubTest("FSET", keys_FSET_test)
-	runStep(t, mc, "GET", keys_GET_test)
+	g.regSubTest("GET", keys_GET_test)
-	runStep(t, mc, "KEYS", keys_KEYS_test)
+	g.regSubTest("KEYS", keys_KEYS_test)
-	runStep(t, mc, "PERSIST", keys_PERSIST_test)
+	g.regSubTest("PERSIST", keys_PERSIST_test)
-	runStep(t, mc, "SET", keys_SET_test)
+	g.regSubTest("SET", keys_SET_test)
-	runStep(t, mc, "STATS", keys_STATS_test)
+	g.regSubTest("STATS", keys_STATS_test)
-	runStep(t, mc, "TTL", keys_TTL_test)
+	g.regSubTest("TTL", keys_TTL_test)
-	runStep(t, mc, "SET EX", keys_SET_EX_test)
+	g.regSubTest("SET EX", keys_SET_EX_test)
-	runStep(t, mc, "PDEL", keys_PDEL_test)
+	g.regSubTest("PDEL", keys_PDEL_test)
-	runStep(t, mc, "FIELDS", keys_FIELDS_test)
+	g.regSubTest("FIELDS", keys_FIELDS_test)
-	runStep(t, mc, "WHEREIN", keys_WHEREIN_test)
+	g.regSubTest("WHEREIN", keys_WHEREIN_test)
-	runStep(t, mc, "WHEREEVAL", keys_WHEREEVAL_test)
+	g.regSubTest("WHEREEVAL", keys_WHEREEVAL_test)
-	runStep(t, mc, "TYPE", keys_TYPE_test)
+	g.regSubTest("TYPE", keys_TYPE_test)
-	runStep(t, mc, "FLUSHDB", keys_FLUSHDB_test)
+	g.regSubTest("FLUSHDB", keys_FLUSHDB_test)
-	runStep(t, mc, "HEALTHZ", keys_HEALTHZ_test)
+	g.regSubTest("HEALTHZ", keys_HEALTHZ_test)
-	runStep(t, mc, "SERVER", keys_SERVER_test)
+	g.regSubTest("SERVER", keys_SERVER_test)
-	runStep(t, mc, "INFO", keys_INFO_test)
+	g.regSubTest("INFO", keys_INFO_test)
 }
 
 func keys_BOUNDS_test(mc *mockServer) error {

tests/metrics_test.go

@@ -5,11 +5,10 @@ import (
 	"io"
 	"net/http"
 	"strings"
-	"testing"
 )
 
-func subTestMetrics(t *testing.T, mc *mockServer) {
+func subTestMetrics(g *testGroup) {
-	runStep(t, mc, "basic", metrics_basic_test)
+	g.regSubTest("basic", metrics_basic_test)
 }
 
 func downloadURLWithStatusCode(u string) (int, string, error) {

tests/mock_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"math/rand"
+	"net"
 	"os"
 	"path/filepath"
 	"strings"
@@ -43,9 +44,9 @@ type mockServer struct {
 	shutdown chan bool
 }
 
-func (mc *mockServer) readAOF() ([]byte, error) {
+// func (mc *mockServer) readAOF() ([]byte, error) {
-	return os.ReadFile(filepath.Join(mc.dir, "appendonly.aof"))
+// 	return os.ReadFile(filepath.Join(mc.dir, "appendonly.aof"))
-}
+// }
 
 func (mc *mockServer) metricsPort() int {
 	return mc.mport
@@ -57,6 +58,20 @@ type MockServerOptions struct {
 	Metrics bool
 }
 
+var nextPort int32 = 10000
+
+func getRandPort() int {
+	// choose a valid port between 10000-50000
+	for {
+		port := int(atomic.AddInt32(&nextPort, 1))
+		ln, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
+		if err == nil {
+			ln.Close()
+			return port
+		}
+	}
+}
+
 func mockOpenServer(opts MockServerOptions) (*mockServer, error) {
 
 	logOutput := io.Discard
@@ -67,7 +82,7 @@ func mockOpenServer(opts MockServerOptions) (*mockServer, error) {
 	log.SetOutput(logOutput)
 
 	rand.Seed(time.Now().UnixNano())
-	port := rand.Int()%20000 + 20000
+	port := getRandPort()
 	dir := fmt.Sprintf("data-mock-%d", port)
 	if !opts.Silent {
 		fmt.Printf("Starting test server at port %d\n", port)
@@ -86,7 +101,7 @@ func mockOpenServer(opts MockServerOptions) (*mockServer, error) {
 	shutdown := make(chan bool)
 	s := &mockServer{port: port, dir: dir, shutdown: shutdown}
 	if opts.Metrics {
-		s.mport = rand.Int()%20000 + 20000
+		s.mport = getRandPort()
 	}
 	var ferrt int32 // atomic flag for when ferr has been set
 	var ferr error  // ferr for when the server fails to start

tests/scripts_test.go

@@ -3,14 +3,13 @@ package tests
 import (
 	"fmt"
 	"strings"
-	"testing"
 )
 
-func subTestScripts(t *testing.T, mc *mockServer) {
+func subTestScripts(g *testGroup) {
-	runStep(t, mc, "BASIC", scripts_BASIC_test)
+	g.regSubTest("BASIC", scripts_BASIC_test)
-	runStep(t, mc, "ATOMIC", scripts_ATOMIC_test)
+	g.regSubTest("ATOMIC", scripts_ATOMIC_test)
-	runStep(t, mc, "READONLY", scripts_READONLY_test)
+	g.regSubTest("READONLY", scripts_READONLY_test)
-	runStep(t, mc, "NONATOMIC", scripts_NONATOMIC_test)
+	g.regSubTest("NONATOMIC", scripts_NONATOMIC_test)
 }
 
 func scripts_BASIC_test(mc *mockServer) error {

tests/stats_test.go

@@ -2,13 +2,12 @@ package tests
 
 import (
 	"errors"
-	"testing"
 
 	"github.com/tidwall/gjson"
 )
 
-func subTestInfo(t *testing.T, mc *mockServer) {
+func subTestInfo(g *testGroup) {
-	runStep(t, mc, "valid json", info_valid_json_test)
+	g.regSubTest("valid json", info_valid_json_test)
 }
 
 func info_valid_json_test(mc *mockServer) error {

tests/testcmd_test.go

@@ -1,15 +1,11 @@
 package tests
 
-import (
+func subTestTestCmd(g *testGroup) {
-	"testing"
+	g.regSubTest("WITHIN", testcmd_WITHIN_test)
-)
+	g.regSubTest("INTERSECTS", testcmd_INTERSECTS_test)
-
+	g.regSubTest("INTERSECTS_CLIP", testcmd_INTERSECTS_CLIP_test)
-func subTestTestCmd(t *testing.T, mc *mockServer) {
+	g.regSubTest("ExpressionErrors", testcmd_expressionErrors_test)
-	runStep(t, mc, "WITHIN", testcmd_WITHIN_test)
+	g.regSubTest("Expressions", testcmd_expression_test)
-	runStep(t, mc, "INTERSECTS", testcmd_INTERSECTS_test)
-	runStep(t, mc, "INTERSECTS_CLIP", testcmd_INTERSECTS_CLIP_test)
-	runStep(t, mc, "ExpressionErrors", testcmd_expressionErrors_test)
-	runStep(t, mc, "Expressions", testcmd_expression_test)
 }
 
 func testcmd_WITHIN_test(mc *mockServer) error {

tests/tests_test.go

@@ -5,11 +5,16 @@ import (
 	"math/rand"
 	"os"
 	"os/signal"
+	"runtime"
+	"strings"
+	"sync"
 	"syscall"
 	"testing"
 	"time"
 
 	"github.com/gomodule/redigo/redis"
+	"github.com/tidwall/limiter"
+	"go.uber.org/atomic"
 )
 
 const (
@@ -26,10 +31,10 @@ const (
 	white   = "\x1b[37m"
 )
 
-func TestAll(t *testing.T) {
+func TestIntegration(t *testing.T) {
 
-	mockCleanup(false)
+	mockCleanup(true)
-	defer mockCleanup(false)
+	defer mockCleanup(true)
 
 	ch := make(chan os.Signal, 1)
 	signal.Notify(ch, os.Interrupt, syscall.SIGTERM)
@@ -39,63 +44,180 @@ func TestAll(t *testing.T) {
 		os.Exit(1)
 	}()
 
-	runSubTest(t, "keys", subTestKeys)
+	regTestGroup("keys", subTestKeys)
-	runSubTest(t, "json", subTestJSON)
+	regTestGroup("json", subTestJSON)
-	runSubTest(t, "search", subTestSearch)
+	regTestGroup("search", subTestSearch)
-	runSubTest(t, "testcmd", subTestTestCmd)
+	regTestGroup("testcmd", subTestTestCmd)
-	runSubTest(t, "client", subTestClient)
+	regTestGroup("client", subTestClient)
-	runSubTest(t, "scripts", subTestScripts)
+	regTestGroup("scripts", subTestScripts)
-	runSubTest(t, "fence", subTestFence)
+	regTestGroup("fence", subTestFence)
-	runSubTest(t, "info", subTestInfo)
+	regTestGroup("info", subTestInfo)
-	runSubTest(t, "timeouts", subTestTimeout)
+	regTestGroup("timeouts", subTestTimeout)
-	runSubTest(t, "metrics", subTestMetrics)
+	regTestGroup("metrics", subTestMetrics)
-	runSubTest(t, "aof", subTestAOF)
+	regTestGroup("aof", subTestAOF)
+	runTestGroups(t)
 }
 
-func runSubTest(t *testing.T, name string, test func(t *testing.T, mc *mockServer)) {
+var allGroups []*testGroup
-	t.Run(name, func(t *testing.T) {
-		// t.Parallel()
-		t.Helper()
 
+func runTestGroups(t *testing.T) {
+	limit := runtime.NumCPU()
+	if limit > 16 {
+		limit = 16
+	}
+	l := limiter.New(limit)
+
+	// Initialize all stores as "skipped", but they'll be unset if the test is
+	// not actually skipped.
+	for _, g := range allGroups {
+		for _, s := range g.subs {
+			s.skipped.Store(true)
+		}
+	}
+	for _, g := range allGroups {
+		func(g *testGroup) {
+			t.Run(g.name, func(t *testing.T) {
+				for _, s := range g.subs {
+					func(s *testGroupSub) {
+						t.Run(s.name, func(t *testing.T) {
+							s.skipped.Store(false)
+							var wg sync.WaitGroup
+							wg.Add(1)
+							var err error
+							go func() {
+								l.Begin()
+								defer func() {
+									l.End()
+									wg.Done()
+								}()
+								err = s.run()
+							}()
+							if false {
+								t.Parallel()
+								t.Run("bg", func(t *testing.T) {
+									wg.Wait()
+									if err != nil {
+										t.Fatal(err)
+									}
+								})
+							}
+						})
+					}(s)
+				}
+			})
+		}(g)
+	}
+
+	done := make(chan bool)
+	go func() {
+		defer func() { done <- true }()
+		for {
+			finished := true
+			for _, g := range allGroups {
+				skipped := true
+				for _, s := range g.subs {
+					if !s.skipped.Load() {
+						skipped = false
+						break
+					}
+				}
+				if !skipped && !g.printed.Load() {
+					fmt.Printf(bright+"Testing %s\n"+clear, g.name)
+					g.printed.Store(true)
+				}
+				for _, s := range g.subs {
+					if !s.skipped.Load() && !s.printedName.Load() {
+						fmt.Printf("[..] %s (running) ", s.name)
+						s.printedName.Store(true)
+					}
+					if s.done.Load() && !s.printedResult.Load() {
+						fmt.Printf("\r")
+						msg := fmt.Sprintf("[..] %s (running) ", s.name)
+						fmt.Print(strings.Repeat(" ", len(msg)))
+						fmt.Printf("\r")
+						if s.err != nil {
+							fmt.Printf("["+red+"fail"+clear+"] %s\n", s.name)
+						} else {
+							fmt.Printf("["+green+"ok"+clear+"] %s\n", s.name)
+						}
+						s.printedResult.Store(true)
+					}
+					if !s.skipped.Load() && !s.done.Load() {
+						finished = false
+						break
+					}
+				}
+				if !finished {
+					break
+				}
+			}
+			if finished {
+				break
+			}
+			time.Sleep(time.Second / 4)
+		}
+	}()
+	<-done
+	var fail bool
+	for _, g := range allGroups {
+		for _, s := range g.subs {
+			if s.err != nil {
+				t.Errorf("%s/%s/%s\n%s", t.Name(), g.name, s.name, s.err)
+				fail = true
+			}
+		}
+	}
+	if fail {
+		t.Fail()
+	}
+
+}
+
+type testGroup struct {
+	name    string
+	subs    []*testGroupSub
+	printed atomic.Bool
+}
+
+type testGroupSub struct {
+	g             *testGroup
+	name          string
+	fn            func(mc *mockServer) error
+	err           error
+	skipped       atomic.Bool
+	done          atomic.Bool
+	printedName   atomic.Bool
+	printedResult atomic.Bool
+}
+
+func regTestGroup(name string, fn func(g *testGroup)) {
+	g := &testGroup{name: name}
+	allGroups = append(allGroups, g)
+	fn(g)
+}
+
+func (g *testGroup) regSubTest(name string, fn func(mc *mockServer) error) {
+	s := &testGroupSub{g: g, name: name, fn: fn}
+	g.subs = append(g.subs, s)
+}
+
+func (s *testGroupSub) run() (err error) {
+	// This all happens in a background routine.
+	defer func() {
+		s.err = err
+		s.done.Store(true)
+	}()
+	return func() error {
 		mc, err := mockOpenServer(MockServerOptions{
 			Silent:  true,
 			Metrics: true,
 		})
 		if err != nil {
-			t.Fatal(err)
+			return err
 		}
 		defer mc.Close()
-
+		return s.fn(mc)
-		fmt.Printf(bright+"Testing %s\n"+clear, name)
+	}()
-		test(t, mc)
-	})
-}
-
-func runStep(t *testing.T, mc *mockServer, name string, step func(mc *mockServer) error) {
-	t.Run(name, func(t *testing.T) {
-		t.Helper()
-		if err := func() error {
-			// reset the current server
-			mc.ResetConn()
-			defer mc.ResetConn()
-			// clear the database so the test is consistent
-			if err := mc.DoBatch(
-				Do("OUTPUT", "resp").OK(),
-				Do("FLUSHDB").OK(),
-			); err != nil {
-				return err
-			}
-			if err := step(mc); err != nil {
-				return err
-			}
-			return nil
-		}(); err != nil {
-			fmt.Fprintf(os.Stderr, "["+red+"fail"+clear+"]: %s\n", name)
-			t.Fatal(err)
-			// t.Fatal(err)
-		}
-		fmt.Printf("["+green+"ok"+clear+"]: %s\n", name)
-	})
 }
 
 func BenchmarkAll(b *testing.B) {

tests/timeout_test.go

@@ -4,19 +4,18 @@ import (
 	"fmt"
 	"math/rand"
 	"strings"
-	"testing"
 	"time"
 
 	"github.com/gomodule/redigo/redis"
 )
 
-func subTestTimeout(t *testing.T, mc *mockServer) {
+func subTestTimeout(g *testGroup) {
-	runStep(t, mc, "spatial", timeout_spatial_test)
+	g.regSubTest("spatial", timeout_spatial_test)
-	runStep(t, mc, "search", timeout_search_test)
+	g.regSubTest("search", timeout_search_test)
-	runStep(t, mc, "scripts", timeout_scripts_test)
+	g.regSubTest("scripts", timeout_scripts_test)
-	runStep(t, mc, "no writes", timeout_no_writes_test)
+	g.regSubTest("no writes", timeout_no_writes_test)
-	runStep(t, mc, "within scripts", timeout_within_scripts_test)
+	g.regSubTest("within scripts", timeout_within_scripts_test)
-	runStep(t, mc, "no writes within scripts", timeout_no_writes_within_scripts_test)
+	g.regSubTest("no writes within scripts", timeout_no_writes_within_scripts_test)
 }
 
 func setup(mc *mockServer, count int, points bool) (err error) {