drainer/: Reduce memory usage (#735)

drainer/config.go

@@ -52,7 +52,7 @@ const (
 
 var (
 	maxBinlogItemCount        int
-	defaultBinlogItemCount    = 512
+	defaultBinlogItemCount    = 0
 	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])
 }