mydump: set read block buffer size to BlockSize * config.BufferSizeScale

go.mod

@@ -12,11 +12,8 @@ require (
  github.com/cznic/y v0.0.0-20160420101755-9fdf92d4aac0
  github.com/go-sql-driver/mysql v1.4.0
  github.com/gogo/protobuf v1.1.1
- github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57 // indirect
- github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6 // indirect
  github.com/joho/sqltocsv v0.0.0-20180904231936-b24deec2b806
  github.com/pingcap/check v0.0.0-20171206051426-1c287c953996
- github.com/pingcap/errors v0.11.0
  github.com/pingcap/gofail v0.0.0-20181121072748-c3f835e5a7d8
  github.com/pingcap/kvproto v0.0.0-20181105061835-1b5d69cd1d26
  github.com/pingcap/parser v0.0.0-20181113072426-4a9a1b13b591

lightning/config/config.go

@@ -71,7 +71,7 @@ type Lightning struct {
  common.LogConfig
  TableConcurrency int `toml:"table-concurrency" json:"table-concurrency"`
  RegionConcurrency int `toml:"region-concurrency" json:"region-concurrency"`
- IOConcurrency int `toml:"io-concurrency" json:"region-concurrency"`
+ IOConcurrency int `toml:"io-concurrency" json:"io-concurrency"`
  ProfilePort int `toml:"pprof-port" json:"pprof-port"`
  CheckRequirements bool `toml:"check-requirements" json:"check-requirements"`
 }
@@ -130,7 +130,7 @@ func NewConfig() *Config {
  App: Lightning{
  RegionConcurrency: runtime.NumCPU(),
  TableConcurrency: 8,
- IOConcurrency: 2,
+ IOConcurrency: 5,
  CheckRequirements: true,
  },
  TiDB: DBStore{

lightning/config/const.go

@@ -9,6 +9,8 @@ const (
  ReadBlockSize int64 = 64 * _K
  MinRegionSize int64 = 256 * _M
 
+ BufferSizeScale = 5
+
  // kv import
  KVMaxBatchSize int64 = 200 * _G
 )

lightning/metric/metric.go

@@ -106,15 +106,15 @@ var (
  prometheus.HistogramOpts{
  Namespace: "lightning",
  Name: "chunk_parser_read_block_seconds",
- Help: "time needed to chunk parser read a block",
+ Help: "time needed for chunk parser read a block",
  Buckets: prometheus.ExponentialBuckets(0.001, 3.1622776601683795, 10),
  },
  )
  ChunkParserReadRowSecondsHistogram = prometheus.NewHistogram(
  prometheus.HistogramOpts{
  Namespace: "lightning",
  Name: "chunk_parser_read_row_seconds",
- Help: "time needed to chunk parser read a row",
+ Help: "time needed for chunk parser read a row",
  Buckets: prometheus.ExponentialBuckets(0.001, 3.1622776601683795, 10),
  },
  )

lightning/mydump/parser.go

@@ -7,6 +7,7 @@ import (
 
  "github.com/pkg/errors"
 
+ "github.com/pingcap/tidb-lightning/lightning/config"
  "github.com/pingcap/tidb-lightning/lightning/metric"
  "github.com/pingcap/tidb-lightning/lightning/worker"
 )
@@ -54,7 +55,7 @@ type Row struct {
 func NewChunkParser(reader io.Reader, blockBufSize int64, ioWorkers *worker.RestoreWorkerPool) *ChunkParser {
  return &ChunkParser{
  reader: reader,
- blockBuf: make([]byte, blockBufSize),
+ blockBuf: make([]byte, blockBufSize*config.BufferSizeScale),
  remainBuf: &bytes.Buffer{},
  appendBuf: &bytes.Buffer{},
  ioWorkers: ioWorkers,
@@ -92,7 +93,7 @@ func (parser *ChunkParser) readBlock() error {
  // limit IO concurrency
  w := parser.ioWorkers.Apply()
  n, err := parser.reader.Read(parser.blockBuf)
- defer parser.ioWorkers.Recycle(w)
+ parser.ioWorkers.Recycle(w)
 
  switch err {
  case io.ErrUnexpectedEOF, io.EOF:

tidb-lightning.toml

@@ -12,9 +12,14 @@ table-concurrency = 8
 # region-concurrency changes the concurrency number of data. It is set to the number of logical CPU cores by default and needs no configuration.
 # In mixed configuration, you can set it to 75% of the size of logical CPU cores.
 # region-concurrency default to runtime.NumCPU()
+
 # region-concurrency =
-# io-concurrency controls the maximum io concurrent
-# io-concurrency = 2
+# io-concurrency controls the maximum IO concurrency
+# Excessive IO concurrency causes an increase in IO latency because the disk
+# internal buffer is frequently refreshed causing a cache miss. For different
+# disk media, concurrency has different effects on IO latency, which can be
+# adjusted according to monitoring.
+# io-concurrency = 5
 
 # logging
 level = "info"