*: simple code for readProjection

ddl/concurrentddltest/BUILD.bazel

@@ -2,7 +2,7 @@ load("@io_bazel_rules_go//go:def.bzl", "go_test")
 
 go_test(
     name = "concurrentddltest_test",
-    timeout = "short",
+    timeout = "moderate",
     srcs = [
         "main_test.go",
         "switch_test.go",

executor/projection.go

@@ -94,7 +94,7 @@ func (e *ProjectionExec) Open(ctx context.Context) error {
 	return e.open(ctx)
 }
 
-func (e *ProjectionExec) open(ctx context.Context) error {
+func (e *ProjectionExec) open(_ context.Context) error {
 	e.prepared = false
 	e.parentReqRows = int64(e.maxChunkSize)
 
@@ -364,14 +364,14 @@ func (f *projectionInputFetcher) run(ctx context.Context) {
 	}()
 
 	for {
-		input := readProjectionInput(f.inputCh, f.globalFinishCh)
-		if input == nil {
+		input, isNil := readProjection[*projectionInput](f.inputCh, f.globalFinishCh)
+		if isNil {
 			return
 		}
 		targetWorker := input.targetWorker
 
-		output = readProjectionOutput(f.outputCh, f.globalFinishCh)
-		if output == nil {
+		output, isNil = readProjection[*projectionOutput](f.outputCh, f.globalFinishCh)
+		if isNil {
 			f.proj.memTracker.Consume(-input.chk.MemoryUsage())
 			return
 		}
@@ -431,13 +431,13 @@ func (w *projectionWorker) run(ctx context.Context) {
 		w.proj.wg.Done()
 	}()
 	for {
-		input := readProjectionInput(w.inputCh, w.globalFinishCh)
-		if input == nil {
+		input, isNil := readProjection[*projectionInput](w.inputCh, w.globalFinishCh)
+		if isNil {
 			return
 		}
 
-		output = readProjectionOutput(w.outputCh, w.globalFinishCh)
-		if output == nil {
+		output, isNil = readProjection[*projectionOutput](w.outputCh, w.globalFinishCh)
+		if isNil {
 			return
 		}
 
@@ -462,26 +462,14 @@ func recoveryProjection(output *projectionOutput, r interface{}) {
 	logutil.BgLogger().Error("projection executor panicked", zap.String("error", fmt.Sprintf("%v", r)), zap.Stack("stack"))
 }
 
-func readProjectionInput(inputCh <-chan *projectionInput, finishCh <-chan struct{}) *projectionInput {
+func readProjection[T any](ch <-chan T, finishCh <-chan struct{}) (t T, isNil bool) {
 	select {
 	case <-finishCh:
-		return nil
-	case input, ok := <-inputCh:
-		if !ok {
-			return nil
-		}
-		return input
-	}
-}
-
-func readProjectionOutput(outputCh <-chan *projectionOutput, finishCh <-chan struct{}) *projectionOutput {
-	select {
-	case <-finishCh:
-		return nil
-	case output, ok := <-outputCh:
+		return t, true
+	case t, ok := <-ch:
 		if !ok {
-			return nil
+			return t, true
 		}
-		return output
+		return t, false
 	}
 }

executor/seqtest/main_test.go

@@ -31,7 +31,7 @@ func TestMain(m *testing.M) {
 	opts := []goleak.Option{
 		goleak.IgnoreTopFunction("github.com/golang/glog.(*loggingT).flushDaemon"),
 		goleak.IgnoreTopFunction("github.com/lestrrat-go/httprc.runFetchWorker"),
-		goleak.IgnoreTopFunction("github.com/pingcap/tidb/executor.readProjectionInput"),
+		goleak.IgnoreTopFunction("github.com/pingcap/tidb/executor.readProjection[...]"),
 		goleak.IgnoreTopFunction("go.etcd.io/etcd/client/pkg/v3/logutil.(*MergeLogger).outputLoop"),
 		goleak.IgnoreTopFunction("go.opencensus.io/stats/view.(*worker).start"),
 	}