drainer/: Reduce memory usage (#735)

drainer/config.go

@@ -52,7 +52,7 @@ const (
 
 var (
 	maxBinlogItemCount        int
-	defaultBinlogItemCount    = 512
+	defaultBinlogItemCount    = 8
 	supportedCompressors      = [...]string{"gzip"}
 	newZKFromConnectionString = zk.NewFromConnectionString
 )

drainer/merge.go

@@ -192,7 +192,7 @@ func NewMerger(ts int64, strategy string, sources ...MergeSource) *Merger {
 	m := &Merger{
 		latestTS: ts,
 		sources:  make(map[string]MergeSource),
-		output:   make(chan MergeItem, 10),
+		output:   make(chan MergeItem),
 		strategy: mergeStrategy,
 	}
 

drainer/metrics.go

@@ -109,7 +109,7 @@ var (
 			Subsystem: "drainer",
 			Name:      "read_binlog_size",
 			Help:      "Bucketed histogram of size of a binlog.",
-			Buckets:   prometheus.ExponentialBuckets(16, 2, 20),
+			Buckets:   prometheus.ExponentialBuckets(16, 2, 25),
 		}, []string{"nodeID"})
 
 	queueSizeGauge = prometheus.NewGaugeVec(

drainer/pump.go

@@ -32,7 +32,7 @@ import (
 )
 
 const (
-	binlogChanSize = 10
+	binlogChanSize = 0
 )
 
 // Pump holds the connection to a pump node, and keeps the savepoint of binlog last read

drainer/pump_test.go

@@ -80,10 +80,12 @@ func (s *pumpSuite) TestPullBinlog(c *C) {
 
 	// ascending commitTs order
 	pullBinlogCommitTSChecker(commitTsArray, ret, binlogBytesChan, c)
+	time.Sleep(10 * time.Microsecond)
 	c.Assert(p.latestTS, Equals, commitTsArray[len(commitTsArray)-1])
 
 	// should omit disorder binlog item, latestTs should be 29
 	pullBinlogCommitTSChecker(wrongCommitTsArray, ret, binlogBytesChan, c)
+	time.Sleep(10 * time.Microsecond)
 	c.Assert(p.latestTS, Equals, wrongCommitTsArray[len(wrongCommitTsArray)-2])
 }
 

drainer/sync/syncer.go

@@ -50,7 +50,7 @@ type baseSyncer struct {
 func newBaseSyncer(tableInfoGetter translator.TableInfoGetter) *baseSyncer {
 	return &baseSyncer{
 		baseError:       newBaseError(),
-		success:         make(chan *Item, 1024),
+		success:         make(chan *Item, 8),
 		tableInfoGetter: tableInfoGetter,
 	}
 }

drainer/sync/syncer_test.go

@@ -120,13 +120,6 @@ func (s *syncerSuite) TestOpenAndClose(c *check.C) {
 
 func (s *syncerSuite) TestGetFromSuccesses(c *check.C) {
 	gen := translator.BinlogGenrator{}
-	gen.SetDDL()
-	item := &Item{
-		Binlog:        gen.TiBinlog,
-		PrewriteValue: gen.PV,
-		Schema:        gen.Schema,
-		Table:         gen.Table,
-	}
 
 	// set up mysql db mock expect
 	s.mysqlMock.ExpectBegin()
@@ -144,6 +137,14 @@ func (s *syncerSuite) TestGetFromSuccesses(c *check.C) {
 
 	var successCount = make([]int64, len(s.syncers))
 	for idx, syncer := range s.syncers {
+		gen.SetDDL()
+		item := &Item{
+			Binlog:        gen.TiBinlog,
+			PrewriteValue: gen.PV,
+			Schema:        gen.Schema,
+			Table:         gen.Table,
+		}
+
 		go func(idx int) {
 			for range syncer.Successes() {
 				atomic.AddInt64(&successCount[idx], 1)

pkg/loader/load.go

@@ -159,8 +159,8 @@ func NewLoader(db *gosql.DB, opt ...Option) (Loader, error) {
 		workerCount:   opts.workerCount,
 		batchSize:     opts.batchSize,
 		metrics:       opts.metrics,
-		input:         make(chan *Txn, 1024),
-		successTxn:    make(chan *Txn, 1024),
+		input:         make(chan *Txn),
+		successTxn:    make(chan *Txn),
 		merge:         true,
 		saveAppliedTS: opts.saveAppliedTS,
 
@@ -430,16 +430,19 @@ func (s *loaderImpl) execDMLs(dmls []*DML) error {
 
 // Run will quit when meet any error, or all the txn are drained
 func (s *loaderImpl) Run() error {
+	txnManager := newTxnManager(1024)
 	defer func() {
 		log.Info("Run()... in Loader quit")
 		close(s.successTxn)
+		txnManager.Close()
 	}()
 
 	batch := fNewBatchManager(s)
+	input := txnManager.put(s.input)
 
 	for {
 		select {
-		case txn, ok := <-s.input:
+		case txn, ok := <-input:
 			if !ok {
 				log.Info("Loader closed, quit running")
 				if err := batch.execAccumulatedDMLs(); err != nil {
@@ -449,6 +452,7 @@ func (s *loaderImpl) Run() error {
 			}
 
 			s.metricsInputTxn(txn)
+			txnManager.pop(txn)
 			if err := batch.put(txn); err != nil {
 				return errors.Trace(err)
 			}
@@ -464,12 +468,13 @@ func (s *loaderImpl) Run() error {
 			}
 
 			// get first
-			txn, ok := <-s.input
+			txn, ok := <-input
 			if !ok {
 				return nil
 			}
 
 			s.metricsInputTxn(txn)
+			txnManager.pop(txn)
 			if err := batch.put(txn); err != nil {
 				return errors.Trace(err)
 			}
@@ -624,6 +629,82 @@ func (b *batchManager) put(txn *Txn) error {
 	return nil
 }
 
+type txnManager struct {
+	cacheChan    chan *Txn
+	shutdown     chan struct{}
+	cachedSize   int
+	maxCacheSize int
+	cond         *sync.Cond
+	closed       bool
+}
+
+func newTxnManager(maxCacheSize int) *txnManager {
+	return &txnManager{
+		cacheChan:    make(chan *Txn, 1024),
+		maxCacheSize: maxCacheSize,
+		cond:         sync.NewCond(new(sync.Mutex)),
+		shutdown:     make(chan struct{}),
+	}
+}
+
+func (t *txnManager) put(input chan *Txn) chan *Txn {
+	ret := t.cacheChan
+	go func() {
+		for !t.closed {
+			var txn *Txn
+			var ok bool
+			select {
+			case txn, ok = <-input:
+				if !ok {
+					close(ret)
+					return
+				}
+			case <-t.shutdown:
+				return
+			}
+			txnSize := len(txn.DMLs)
+
+			t.cond.L.Lock()
+			if txnSize < t.maxCacheSize {
+				for !t.closed && txnSize+t.cachedSize > t.maxCacheSize {
+					t.cond.Wait()
+				}
+			} else {
+				for !t.closed && t.cachedSize != 0 {
+					t.cond.Wait()
+				}
+			}
+			t.cond.L.Unlock()
+
+			select {
+			case ret <- txn:
+				t.cond.L.Lock()
+				t.cachedSize += txnSize
+				t.cond.L.Unlock()
+			case <-t.shutdown:
+				return
+			}
+		}
+	}()
+	return ret
+}
+
+func (t *txnManager) pop(txn *Txn) {
+	t.cond.L.Lock()
+	t.cachedSize -= len(txn.DMLs)
+	t.cond.Signal()
+	t.cond.L.Unlock()
+}
+
+func (t *txnManager) Close() {
+	if t.closed {
+		return
+	}
+	close(t.shutdown)
+	t.closed = true
+	t.cond.Signal()
+}
+
 func getAppliedTS(db *gosql.DB) int64 {
 	appliedTS, err := pkgsql.GetTidbPosition(db)
 	if err != nil {

tests/kafka/kafka.go

@@ -84,19 +84,20 @@ func main() {
 
 	go func() {
 		defer ld.Close()
-		for {
-			select {
-			case msg := <-breader.Messages():
-				str := msg.Binlog.String()
-				log.S().Debugf("recv: %.2000s", str)
-				txn, err := loader.SlaveBinlogToTxn(msg.Binlog)
-				if err != nil {
-					log.S().Fatal(err)
-				}
-				ld.Input() <- txn
-			case txn := <-ld.Successes():
-				log.S().Debug("succ: ", txn)
+		for msg := range breader.Messages() {
+			str := msg.Binlog.String()
+			log.S().Debugf("recv: %.2000s", str)
+			txn, err := loader.SlaveBinlogToTxn(msg.Binlog)
+			if err != nil {
+				log.S().Fatal(err)
 			}
+			ld.Input() <- txn
+		}
+	}()
+
+	go func() {
+		for txn := range ld.Successes() {
+			log.S().Debug("succ: ", txn)
 		}
 	}()
 

tests/status/run.sh

@@ -8,8 +8,17 @@ cd "$(dirname "$0")"
 OUT_DIR=/tmp/tidb_binlog_test
 STATUS_LOG="${OUT_DIR}/status.log"
 
+# use latest ts as initial-commit-ts, so we can skip binlog by previous test case
+ms=$(date +'%s')
+ts=$(($ms*1000<<18))
+args="-initial-commit-ts=$ts"
+down_run_sql "DROP DATABASE IF EXISTS tidb_binlog"
+rm -rf /tmp/tidb_binlog_test/data.drainer
+
+
 # run drainer, and drainer's status should be online
-run_drainer &
+run_drainer "$args" &
+
 sleep 2
 echo "check drainer's status, should be online"
 check_status drainers online