syncer: fix in-memory tps count (#286)

syncer/syncer.go

@@ -687,6 +687,7 @@ func (s *Syncer) getTable(schema string, table string) (*table, []string, error)
 func (s *Syncer) addCount(isFinished bool, queueBucket string, tp opType, n int64) {
 	m := addedJobsTotal
 	if isFinished {
+		s.count.Add(n)
 		m = finishedJobsTotal
 	}
 
@@ -708,8 +709,6 @@ func (s *Syncer) addCount(isFinished bool, queueBucket string, tp opType, n int6
 	default:
 		s.tctx.L().Warn("unknown job operation type", zap.Stringer("type", tp))
 	}
-
-	s.count.Add(n)
 }
 
 func (s *Syncer) checkWait(job *job) bool {

syncer/syncer_test.go

@@ -1472,6 +1472,8 @@ func (s *testSyncerSuite) TestRun(c *C) {
 	}
 
 	executeSQLAndWait(len(expectJobs1))
+	c.Assert(syncer.Status().(*pb.SyncStatus).TotalEvents, Equals, int64(0))
+	syncer.mockFinishJob(expectJobs1)
 
 	testJobs.Lock()
 	checkJobs(c, testJobs.jobs, expectJobs1)
@@ -1534,14 +1536,14 @@ func (s *testSyncerSuite) TestRun(c *C) {
 	}
 
 	executeSQLAndWait(len(expectJobs2))
+	c.Assert(syncer.Status().(*pb.SyncStatus).TotalEvents, Equals, int64(len(expectJobs1)))
+	syncer.mockFinishJob(expectJobs2)
+	c.Assert(syncer.Status().(*pb.SyncStatus).TotalEvents, Equals, int64(len(expectJobs1)+len(expectJobs2)))
 
 	testJobs.RLock()
 	checkJobs(c, testJobs.jobs, expectJobs2)
 	testJobs.RUnlock()
 
-	status := syncer.Status().(*pb.SyncStatus)
-	c.Assert(status.TotalEvents, Equals, int64(len(expectJobs1)+len(expectJobs2)))
-
 	cancel()
 	syncer.Close()
 	c.Assert(syncer.isClosed(), IsTrue)
@@ -1595,6 +1597,15 @@ var testJobs struct {
 	jobs []*job
 }
 
+func (s *Syncer) mockFinishJob(jobs []*expectJob) {
+	for _, job := range jobs {
+		switch job.tp {
+		case ddl, insert, update, del, flush:
+			s.addCount(true, "test", job.tp, 1)
+		}
+	}
+}
+
 func (s *Syncer) addJobToMemory(job *job) error {
 	log.L().Info("add job to memory", zap.Stringer("job", job))