diff options
author | ale <ale@incal.net> | 2018-08-31 08:28:51 +0100 |
---|---|---|
committer | ale <ale@incal.net> | 2018-08-31 08:28:51 +0100 |
commit | b3d419486a87c9193c2fd6c16168f600876e0f73 (patch) | |
tree | 849d1333d4fac8e5654e65a0fdf4b2cb00355ab9 /vendor/github.com/syndtr/goleveldb/leveldb/db_compaction.go | |
parent | 86a0bd2d15a07662fdae4e24589b08706c2e80b9 (diff) | |
download | crawl-b3d419486a87c9193c2fd6c16168f600876e0f73.tar.gz crawl-b3d419486a87c9193c2fd6c16168f600876e0f73.zip |
Update dependencies
Diffstat (limited to 'vendor/github.com/syndtr/goleveldb/leveldb/db_compaction.go')
-rw-r--r-- | vendor/github.com/syndtr/goleveldb/leveldb/db_compaction.go | 40 |
1 files changed, 37 insertions, 3 deletions
diff --git a/vendor/github.com/syndtr/goleveldb/leveldb/db_compaction.go b/vendor/github.com/syndtr/goleveldb/leveldb/db_compaction.go index b6563e8..28e5090 100644 --- a/vendor/github.com/syndtr/goleveldb/leveldb/db_compaction.go +++ b/vendor/github.com/syndtr/goleveldb/leveldb/db_compaction.go @@ -640,6 +640,16 @@ func (db *DB) tableNeedCompaction() bool { return v.needCompaction() } +// resumeWrite returns an indicator whether we should resume write operation if enough level0 files are compacted. +func (db *DB) resumeWrite() bool { + v := db.s.version() + defer v.release() + if v.tLen(0) < db.s.o.GetWriteL0PauseTrigger() { + return true + } + return false +} + func (db *DB) pauseCompaction(ch chan<- struct{}) { select { case ch <- struct{}{}: @@ -653,6 +663,7 @@ type cCmd interface { } type cAuto struct { + // Note for table compaction, an empty ackC represents it's a compaction waiting command. ackC chan<- error } @@ -765,8 +776,10 @@ func (db *DB) mCompaction() { } func (db *DB) tCompaction() { - var x cCmd - var ackQ []cCmd + var ( + x cCmd + ackQ, waitQ []cCmd + ) defer func() { if x := recover(); x != nil { @@ -778,6 +791,10 @@ func (db *DB) tCompaction() { ackQ[i].ack(ErrClosed) ackQ[i] = nil } + for i := range waitQ { + waitQ[i].ack(ErrClosed) + waitQ[i] = nil + } if x != nil { x.ack(ErrClosed) } @@ -795,12 +812,25 @@ func (db *DB) tCompaction() { return default: } + // Resume write operation as soon as possible. + if len(waitQ) > 0 && db.resumeWrite() { + for i := range waitQ { + waitQ[i].ack(nil) + waitQ[i] = nil + } + waitQ = waitQ[:0] + } } else { for i := range ackQ { ackQ[i].ack(nil) ackQ[i] = nil } ackQ = ackQ[:0] + for i := range waitQ { + waitQ[i].ack(nil) + waitQ[i] = nil + } + waitQ = waitQ[:0] select { case x = <-db.tcompCmdC: case ch := <-db.tcompPauseC: @@ -813,7 +843,11 @@ func (db *DB) tCompaction() { if x != nil { switch cmd := x.(type) { case cAuto: - ackQ = append(ackQ, x) + if cmd.ackC != nil { + waitQ = append(waitQ, x) + } else { + ackQ = append(ackQ, x) + } case cRange: x.ack(db.tableRangeCompaction(cmd.level, cmd.min, cmd.max)) default: |