executor: open doubleread close test and fix bug. (pingcap#3316)

executor/distsql_test.go

@@ -32,7 +32,6 @@ import (
 // TestIndexDoubleReadClose checks that when a index double read returns before reading all the rows, the goroutine doesn't
 // leak. For testing distsql with multiple regions, we need to manually split a mock TiKV.
 func (s *testSuite) TestIndexDoubleReadClose(c *C) {
-	c.Skip("new dist sql use different executor, reopen this test in the future.")
 	if _, ok := s.store.GetClient().(*tikv.CopClient); !ok {
 		// Make sure the store is tikv store.
 		return

executor/new_distsql.go

@@ -295,6 +295,22 @@ func (e *IndexLookUpExecutor) executeTask(task *lookupTableTask, goCtx goctx.Con
 	}
 }
 
+func (e *IndexLookUpExecutor) pickAndExecTask(workCh <-chan *lookupTableTask, txnCtx goctx.Context) {
+	childCtx, cancel := goctx.WithCancel(txnCtx)
+	defer cancel()
+	for {
+		select {
+		case task := <-workCh:
+			if task == nil {
+				return
+			}
+			e.executeTask(task, childCtx)
+		case <-childCtx.Done():
+			return
+		}
+	}
+}
+
 // fetchHandlesAndStartWorkers fetches a batch of handles from index data and builds the index lookup tasks.
 // We initialize some workers to execute this tasks concurrently and put the task to taskCh by order.
 func (e *IndexLookUpExecutor) fetchHandlesAndStartWorkers() {
@@ -309,18 +325,7 @@ func (e *IndexLookUpExecutor) fetchHandlesAndStartWorkers() {
 	lookupConcurrencyLimit := e.ctx.GetSessionVars().IndexLookupConcurrency
 	txnCtx := e.ctx.GoCtx()
 	for i := 0; i < lookupConcurrencyLimit; i++ {
-		go func() {
-			childCtx, cancel := goctx.WithCancel(txnCtx)
-			defer cancel()
-			select {
-			case task := <-workCh:
-				if task == nil {
-					return
-				}
-				e.executeTask(task, childCtx)
-			case <-childCtx.Done():
-			}
-		}()
+		go e.pickAndExecTask(workCh, txnCtx)
 	}
 
 	for {