eliminate handle.ctx

Author: ffffwh

driver/driver.go

@@ -560,7 +560,7 @@ func (d *Driver) StartTask(cfg *drivers.TaskConfig) (*drivers.TaskHandle, *drive
 		return nil, nil, errors.Wrap(err, "SetDriverState")
 	}
 
-	h := newDtleTaskHandle(d.ctx, d.logger, cfg, drivers.TaskStateRunning, time.Now().Round(time.Millisecond))
+	h := newDtleTaskHandle(d.logger, cfg, drivers.TaskStateRunning, time.Now().Round(time.Millisecond))
 	h.driverConfig = &common.MySQLDriverConfig{DtleTaskConfig: dtleTaskConfig}
 	d.tasks.Set(cfg.ID, h)
 	AllocIdTaskNameToTaskHandler.Set(cfg.AllocID, cfg.Name, cfg.ID, h)
@@ -655,14 +655,6 @@ func (d *Driver) handleWait(ctx context.Context, handle *taskHandle, ch chan *dr
 		return
 	case <-d.ctx.Done():
 		return
-	case <-handle.ctx.Done():
-		result := &drivers.ExitResult{
-			ExitCode:  0,
-			Signal:    0,
-			OOMKilled: false,
-			Err:       nil,
-		}
-		ch <- result
 	case result := <-handle.waitCh: // Do not refer to handle.runner.waitCh. It might be nil.
 		handle.stateLock.Lock()
 		handle.procState = drivers.TaskStateExited

driver/handle.go

@@ -1,7 +1,6 @@
 package mysql
 
 import (
-	"context"
 	"fmt"
 	"sync"
 	"time"
@@ -33,15 +32,14 @@ type taskHandle struct {
 
 	runner DriverHandle
 
-	ctx        context.Context
-	cancelFunc context.CancelFunc
-	waitCh     chan *drivers.ExitResult
-	stats      *common.TaskStatistics
+	waitCh chan *drivers.ExitResult
+	doneCh chan struct{}
+	stats  *common.TaskStatistics
 
 	driverConfig *common.MySQLDriverConfig
 }
 
-func newDtleTaskHandle(ctx context.Context, logger g.LoggerType, cfg *drivers.TaskConfig, state drivers.TaskState, started time.Time) *taskHandle {
+func newDtleTaskHandle(logger g.LoggerType, cfg *drivers.TaskConfig, state drivers.TaskState, started time.Time) *taskHandle {
 	h := &taskHandle{
 		logger:      logger,
 		stateLock:   sync.RWMutex{},
@@ -50,9 +48,9 @@ func newDtleTaskHandle(ctx context.Context, logger g.LoggerType, cfg *drivers.Ta
 		startedAt:   started,
 		completedAt: time.Time{},
 		exitResult:  nil,
-		waitCh:      make(chan *drivers.ExitResult, 1),
+		waitCh:      make(chan *drivers.ExitResult),
+		doneCh:      make(chan struct{}),
 	}
-	h.ctx, h.cancelFunc = context.WithCancel(ctx)
 	return h
 }
 
@@ -137,7 +135,8 @@ func (h *taskHandle) run(d *Driver) {
 		t := time.NewTimer(0)
 		for {
 			select {
-			case <-h.ctx.Done():
+			case <-h.doneCh:
+				t.Stop()
 				return
 			case <-t.C:
 				if h.runner != nil {
@@ -168,12 +167,12 @@ func (h *taskHandle) NewRunner(d *Driver) (runner DriverHandle, err error) {
 	case common.TaskTypeSrc:
 		if h.driverConfig.OracleConfig != nil {
 			h.logger.Debug("found oracle src", "OracleConfig", h.driverConfig.OracleConfig)
-			runner, err = extractor.NewExtractorOracle(ctx, h.driverConfig, h.logger, d.storeManager, h.waitCh, h.ctx)
+			runner, err = extractor.NewExtractorOracle(ctx, h.driverConfig, h.logger, d.storeManager, h.waitCh, d.ctx)
 			if err != nil {
 				return nil, errors.Wrap(err, "NewExtractor")
 			}
 		} else {
-			runner, err = mysql.NewExtractor(ctx, h.driverConfig, h.logger, d.storeManager, h.waitCh, h.ctx)
+			runner, err = mysql.NewExtractor(ctx, h.driverConfig, h.logger, d.storeManager, h.waitCh, d.ctx)
 			if err != nil {
 				return nil, errors.Wrap(err, "NewOracleExtractor")
 			}
@@ -183,13 +182,13 @@ func (h *taskHandle) NewRunner(d *Driver) (runner DriverHandle, err error) {
 		if h.driverConfig.KafkaConfig != nil {
 			h.logger.Debug("found kafka", "KafkaConfig", h.driverConfig.KafkaConfig)
 			runner, err = kafka.NewKafkaRunner(ctx, h.driverConfig.KafkaConfig, h.logger,
-				d.storeManager, d.config.NatsAdvertise, h.waitCh, h.ctx)
+				d.storeManager, d.config.NatsAdvertise, h.waitCh, d.ctx)
 			if err != nil {
 				return nil, errors.Wrap(err, "NewKafkaRunner")
 			}
 		} else {
 			runner, err = mysql.NewApplier(ctx, h.driverConfig, h.logger, d.storeManager,
-				d.config.NatsAdvertise, h.waitCh, d.eventer, h.taskConfig, h.ctx)
+				d.config.NatsAdvertise, h.waitCh, d.eventer, h.taskConfig, d.ctx)
 			if err != nil {
 				return nil, errors.Wrap(err, "NewApplier")
 			}
@@ -267,14 +266,17 @@ func (h *taskHandle) emitStats(ru *common.TaskStatistics) {
 
 func (h *taskHandle) Destroy() bool {
 	h.stateLock.RLock()
-	//driver.des
-	h.cancelFunc()
+	defer h.stateLock.RUnlock()
+
+	close(h.doneCh)
+
 	if h.runner != nil {
 		err := h.runner.Shutdown()
 		if err != nil {
 			h.logger.Error("error in h.runner.Shutdown", "err", err)
 		}
 	}
+
 	return h.procState == drivers.TaskStateExited
 }