Shane Xu's Home

commit c2370b437b9840363ed3d12638fe0ca0ea5ed296
Author: Jason Wilder <mail@jasonwilder.com>
Date:   Fri Jul 15 23:26:25 2016 -0600

    Limit in-flight wal writes/encodings
    
    A slower disk can can cause excessive allocations to occur when
    writing to the WAL because the slower encoding and compression occurs
    before taking the write lock.  The encoding/compression grabs a large
    byte slice from a pool and ultimately waits until it can acquire the
    write lock.
    
    This adds a throttle to limit how many inflight WAL writes can be queued
    up to prevent OOMing the processess with slower disks and heavy writes.

  1: diff --git a/pkg/throttle/throttle.go b/pkg/throttle/throttle.go
  2: new file mode 100644
  3: index 0000000..f29c4a2
  4: --- /dev/null
  5: +++ b/pkg/throttle/throttle.go
  6: @@ -0,0 +1,18 @@
  7: +package throttle
  8: +
  9: +// Fixed is a simple channel based concurrency limiter.  It uses a fixed
 10: +// size channel to limit callers from proceeding until there is a value avalable
 11: +// in the channel.  If all are in-use, the caller blocks until one is freed.
 12: +type Fixed chan struct{}
 13: +
 14: +func New(limit int) Fixed {
 15: +   return make(Fixed, limit)
 16: +}
 17: +
 18: +func (t Fixed) Take() {
 19: +   t <- struct{}{}
 20: +}
 21: +
 22: +func (t Fixed) Release() {
 23: +   <-t
 24: +}
 25: diff --git a/tsdb/engine/tsm1/wal.go b/tsdb/engine/tsm1/wal.go
 26: index 3926f93..d7880cc 100644
 27: --- a/tsdb/engine/tsm1/wal.go
 28: +++ b/tsdb/engine/tsm1/wal.go
 29: @@ -18,6 +18,7 @@ import (
 30:  
 31:     "github.com/golang/snappy"
 32:     "github.com/influxdata/influxdb/models"
 33: +   "github.com/influxdata/influxdb/pkg/throttle"
 34:  )
 35:  
 36:  const (
 37: @@ -90,7 +91,8 @@ type WAL struct {
 38:     LoggingEnabled bool
 39:  
 40:     // statistics for the WAL
 41: -   stats *WALStatistics
 42: +   stats    *WALStatistics
 43: +   throttle throttle.Fixed
 44:  }
 45:  
 46:  func NewWAL(path string) *WAL {
 47: @@ -103,6 +105,7 @@ func NewWAL(path string) *WAL {
 48:         logger:      log.New(os.Stderr, "[tsm1wal] ", log.LstdFlags),
 49:         closing:     make(chan struct{}),
 50:         stats:       &WALStatistics{},
 51: +       throttle:    throttle.New(10),
 52:     }
 53:  }
 54:  
 55: @@ -277,6 +280,12 @@ func (l *WAL) LastWriteTime() time.Time {
 56:  }
 57:  
 58:  func (l *WAL) writeToLog(entry WALEntry) (int, error) {
 59: +   // limit how many concurrent encodings can be in flight.  Since we can only
 60: +   // write one at a time to disk, a slow disk can cause the allocations below
 61: +   // to increase quickly.  If we're backed up, wait until others have completed.
 62: +   l.throttle.Take()
 63: +   defer l.throttle.Release()
 64: +
 65:     // encode and compress the entry while we're not locked
 66:     bytes := getBuf(walEncodeBufSize)
 67:     defer putBuf(bytes)
 68: diff --git a/tsdb/store.go b/tsdb/store.go
 69: index 5889577..3076d01 100644
 70: --- a/tsdb/store.go
 71: +++ b/tsdb/store.go
 72: @@ -16,6 +16,7 @@ import (
 73:  
 74:     "github.com/influxdata/influxdb/influxql"
 75:     "github.com/influxdata/influxdb/models"
 76: +   "github.com/influxdata/influxdb/pkg/throttle"
 77:  )
 78:  
 79:  var (
 80: @@ -145,7 +146,7 @@ func (s *Store) loadShards() error {
 81:         err error
 82:     }
 83:  
 84: -   throttle := newthrottle(runtime.GOMAXPROCS(0))
 85: +   t := throttle.New(runtime.GOMAXPROCS(0))
 86:  
 87:     resC := make(chan *res)
 88:     var n int
 89: @@ -171,8 +172,8 @@ func (s *Store) loadShards() error {
 90:             for _, sh := range shards {
 91:                 n++
 92:                 go func(index *DatabaseIndex, db, rp, sh string) {
 93: -                   throttle.take()
 94: -                   defer throttle.release()
 95: +                   t.Take()
 96: +                   defer t.Release()
 97:  
 98:                     start := time.Now()
 99:                     path := filepath.Join(s.path, db, rp, sh)
100: @@ -514,7 +515,7 @@ func (s *Store) walkShards(shards []*Shard, fn func(sh *Shard) error) error {
101:         err error
102:     }
103:  
104: -   throttle := newthrottle(runtime.GOMAXPROCS(0))
105: +   t := throttle.New(runtime.GOMAXPROCS(0))
106:  
107:     resC := make(chan res)
108:     var n int
109: @@ -523,8 +524,8 @@ func (s *Store) walkShards(shards []*Shard, fn func(sh *Shard) error) error {
110:         n++
111:  
112:         go func(sh *Shard) {
113: -           throttle.take()
114: -           defer throttle.release()
115: +           t.Take()
116: +           defer t.Release()
117:  
118:             if err := fn(sh); err != nil {
119:                 resC <- res{err: fmt.Errorf("shard %d: %s", sh.id, err)}
120: @@ -914,20 +915,3 @@ func measurementsFromSourcesOrDB(db *DatabaseIndex, sources ...influxql.Source)
121:  
122:     return measurements, nil
123:  }
124: -
125: -// throttle is a simple channel based concurrency limiter.  It uses a fixed
126: -// size channel to limit callers from proceeding until there is a value avalable
127: -// in the channel.  If all are in-use, the caller blocks until one is freed.
128: -type throttle chan struct{}
129: -
130: -func newthrottle(limit int) throttle {
131: -   return make(throttle, limit)
132: -}
133: -
134: -func (t throttle) take() {
135: -   t <- struct{}{}
136: -}
137: -
138: -func (t throttle) release() {
139: -   <-t
140: -}
141: 

Requests      [total, rate]            1500000, 5000.00
Duration      [total, attack, wait]    5m0.000129097s, 4m59.999799858s, 329.239µs
Latencies     [mean, 50, 95, 99, max]  1.227473ms, 230.474µs, 1.237975ms, 33.100274ms, 251.524774ms
Bytes In      [total, mean]            0, 0.00
Bytes Out     [total, mean]            59994600, 40.00
Success       [ratio]                  99.99%
Status Codes  [code:count]             204:1499865  0:135

Requests      [total, rate]            1500000, 5000.00
Duration      [total, attack, wait]    5m0.000380718s, 4m59.999799905s, 580.813µs
Latencies     [mean, 50, 95, 99, max]  1.477462ms, 239.262µs, 2.878066ms, 36.939105ms, 328.367048ms
Bytes In      [total, mean]            0, 0.00
Bytes Out     [total, mean]            60000000, 40.00
Success       [ratio]                  100.00%
Status Codes  [code:count]             204:1500000

Requests      [total, rate]            1500000, 5000.00
Duration      [total, attack, wait]    5m0.000769812s, 4m59.999799915s, 969.897µs
Latencies     [mean, 50, 95, 99, max]  1.957877ms, 242.364µs, 3.483565ms, 41.994757ms, 638.448146ms
Bytes In      [total, mean]            0, 0.00
Bytes Out     [total, mean]            59999880, 40.00
Success       [ratio]                  100.00%
Status Codes  [code:count]             204:1499997  0:3