syncer: reset db connections during resume (pingcap#254)

Author: amyangfei

dm/master/server.go

@@ -1870,7 +1870,7 @@ func (s *Server) waitOperationOk(ctx context.Context, cli workerrpc.Client, name
 		resp, err := cli.SendRequest(ctx, req, s.cfg.RPCTimeout)
 		var queryResp *pb.QueryTaskOperationResponse
 		if err != nil {
-			log.L().Error("fail to query task operation", log.ShortError(err))
+			log.L().Error("fail to query task operation", zap.String("task", name), zap.Int64("operation log ID", opLogID), log.ShortError(err))
 		} else {
 			queryResp = resp.QueryTaskOperation
 			respLog := queryResp.Log

syncer/checkpoint.go

@@ -127,6 +127,9 @@ type CheckPoint interface {
 	// Close closes the CheckPoint
 	Close()
 
+	// ResetConn resets database connections owned by the Checkpoint
+	ResetConn() error
+
 	// Clear clears all checkpoints
 	Clear() error
 
@@ -242,6 +245,11 @@ func (cp *RemoteCheckPoint) Close() {
 	closeConns(cp.tctx, cp.db)
 }
 
+// ResetConn implements CheckPoint.ResetConn
+func (cp *RemoteCheckPoint) ResetConn() error {
+	return cp.db.ResetConn(cp.tctx)
+}
+
 // Clear implements CheckPoint.Clear
 func (cp *RemoteCheckPoint) Clear() error {
 	cp.Lock()

syncer/syncer.go

@@ -503,6 +503,34 @@ func (s *Syncer) reset() {
 	}
 }
 
+func (s *Syncer) resetDBs() error {
+	var err error
+
+	// toDBs share the same `*sql.DB` in underlying `*baseconn.BaseConn`, currently the `BaseConn.ResetConn`
+	// can only reset the `*sql.DB` and point to the new `*sql.DB`, it is hard to reset all the `*sql.DB` by
+	// calling `BaseConn.ResetConn` once. On the other side if we simply change the underlying value of a
+	// `*sql.DB` by `*conn.DB = *db`, there exists some data race and invalid memory address in db driver.
+	// So we use the close and recreate way here.
+	closeConns(s.tctx, s.toDBs...)
+	s.toDBs, err = createConns(s.cfg, s.cfg.To, s.cfg.WorkerCount, maxDMLConnectionTimeout)
+	if err != nil {
+		return terror.WithScope(err, terror.ScopeDownstream)
+	}
+	s.tctx.L().Info("createDBs", zap.String("toDBs baseConn", fmt.Sprintf("%p", s.toDBs[0].baseConn.DB)))
+
+	err = s.ddlDB.ResetConn(s.tctx)
+	if err != nil {
+		return terror.WithScope(err, terror.ScopeDownstream)
+	}
+
+	err = s.checkpoint.ResetConn()
+	if err != nil {
+		return terror.WithScope(err, terror.ScopeDownstream)
+	}
+
+	return nil
+}
+
 // Process implements the dm.Unit interface.
 func (s *Syncer) Process(ctx context.Context, pr chan pb.ProcessResult) {
 	syncerExitWithErrorCounter.WithLabelValues(s.cfg.Name).Add(0)
@@ -1392,7 +1420,7 @@ func (s *Syncer) handleRowsEvent(ev *replication.RowsEvent, ec eventContext) err
 
 	table, columns, err := s.getTable(schemaName, tableName)
 	if err != nil {
-		return err
+		return terror.WithScope(err, terror.ScopeDownstream)
 	}
 	rows, err := s.mappingDML(originSchema, originTable, columns, ev.Rows)
 	if err != nil {
@@ -2205,6 +2233,17 @@ func (s *Syncer) Resume(ctx context.Context, pr chan pb.ProcessResult) {
 
 	// continue the processing
 	s.reset()
+	// reset database conns
+	err := s.resetDBs()
+	if err != nil {
+		pr <- pb.ProcessResult{
+			IsCanceled: false,
+			Errors: []*pb.ProcessError{
+				unit.NewProcessError(pb.ErrorType_UnknownError, errors.ErrorStack(err)),
+			},
+		}
+		return
+	}
 	s.Process(ctx, pr)
 }
 

syncer/syncer_test.go

@@ -1513,7 +1513,9 @@ func (s *testSyncerSuite) TestRun(c *C) {
 
 	ctx, cancel = context.WithCancel(context.Background())
 	resultCh = make(chan pb.ProcessResult)
-	go syncer.Resume(ctx, resultCh)
+	// simulate `syncer.Resume` here, but doesn't reset database conns
+	syncer.reset()
+	go syncer.Process(ctx, resultCh)
 
 	expectJobs2 := []*expectJob{
 		{