luxfi/zapdb
1
0
Fork 0
mirror of https://github.com/luxfi/zapdb.git synced 2026-07-27 06:54:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix a few race conditions

Browse Source
This commit is contained in:
Ganesh Acharya
2017-04-26 22:11:01 +08:00
parent 47c2c67447
commit 0f63b21469
6 changed files with 66 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
badger/iterator.go
+2 -1
View File
@@ -174,7 +174,8 @@ func (s *KV) NewIterator(
return nil
}
tables := s.getMemTables()
tables, decr := s.getMemTables()
defer decr()
var iters []y.Iterator
for i := 0; i < len(tables); i++ {
iters = append(iters, tables[i].NewUniIterator(reversed))
badger/kv.go
+34 -15
View File
@@ -74,17 +74,18 @@ var DefaultOptions = Options{
type KV struct {
sync.RWMutex // Guards imm, mem.
closer *y.Closer
elog trace.EventLog
mt *skl.Skiplist
imm []*skl.Skiplist // Add here only AFTER pushing to flushChan.
opt Options
lc *levelsController
vlog valueLog
vptr valuePointer
arenaPool *skl.ArenaPool
writeCh chan *block
flushChan chan flushTask // For flushing memtables.
closer *y.Closer
elog trace.EventLog
mt *skl.Skiplist
imm []*skl.Skiplist // Add here only AFTER pushing to flushChan.
opt Options
lc *levelsController
vlog valueLog
vptr valuePointer
arenaPool *skl.ArenaPool
writeCh chan *block
flushChan chan flushTask // For flushing memtables.
blockWriterWg sync.WaitGroup
}
// NewKV returns a new KV object. Compact levels are created as well.
@@ -132,6 +133,7 @@ func NewKV(opt *Options) *KV {
out.vlog.Replay(vptr, fn)
out.closer.Register()
out.blockWriterWg.Add(1)
go out.doWrites()
return out
}
@@ -146,7 +148,13 @@ func (s *KV) Close() {
s.elog.Printf("Closing database")
s.closer.Signal()
// TODO(ganesh): Block writes to mem.
// Make sure that block writer is done pushing stuff into memtable!
// Otherwise, you will have a race condition: we are trying to flush memtables
// and remove them completely, while the block / memtable writer is still
// trying to push stuff into the memtable.
// TODO: Closer framework doesn't care about order. We need to improve it and clean this up.
s.blockWriterWg.Wait()
if s.mt.Size() > 0 {
if s.opt.Verbose {
y.Printf("Flushing memtable\n")
@@ -169,16 +177,23 @@ func (s *KV) Close() {
s.elog.Finish()
}
func (s *KV) getMemTables() []*skl.Skiplist {
// getMemtables returns the current memtables and get references.
func (s *KV) getMemTables() ([]*skl.Skiplist, func()) {
s.RLock()
defer s.RUnlock()
tables := make([]*skl.Skiplist, len(s.imm)+1)
tables[0] = s.mt
tables[0].IncrRef()
last := len(s.imm) - 1
for i := range s.imm {
tables[i+1] = s.imm[last-i]
tables[i+1].IncrRef()
}
return tables, func() {
for _, tbl := range tables {
tbl.DecrRef()
}
}
return tables
}
func (s *KV) decodeValue(ctx context.Context, val []byte, meta byte) []byte {
@@ -206,7 +221,8 @@ func (s *KV) decodeValue(ctx context.Context, val []byte, meta byte) []byte {
// Note that value will include meta byte.
func (s *KV) get(ctx context.Context, key []byte) ([]byte, byte) {
y.Trace(ctx, "Retrieving key from memtables")
tables := s.getMemTables() // Lock should be released.
tables, decr := s.getMemTables() // Lock should be released.
defer decr()
for i := 0; i < len(tables); i++ {
v, meta := tables[i].Get(key)
if meta != 0 || v != nil {
@@ -280,6 +296,7 @@ func (s *KV) writeBlocks(blocks []*block) {
func (s *KV) doWrites() {
defer s.closer.Done()
defer s.blockWriterWg.Done()
blocks := make([]*block, 0, 10)
for {
@@ -402,7 +419,9 @@ func (s *KV) flushMemtable() {
fmt.Printf("Storing offset: %+v\n", ft.vptr)
}
offset := make([]byte, 16)
s.Lock() // For vptr.
s.vptr.Encode(offset)
s.Unlock()
ft.mt.Put(Head, offset, 0)
}
fileID, _ := s.lc.reserveFileIDs(1)
badger/kv_test.go
-1
View File
@@ -94,7 +94,6 @@ func TestConcurrentWrite(t *testing.T) {
var i, j int
for kv := range it.Ch() {
k := kv.Key()
// fmt.Printf("Key=%s\n", k)
if k == nil {
break // end of iteration.
}
badger/levels.go
+12 -1
View File
@@ -294,7 +294,7 @@ func (s *levelsController) pickCompactLevel() int {
if s.beingCompacted[i] || s.beingCompacted[i+1] {
continue
}
if (i == 0 && len(s.levels[0].tables) > s.kv.opt.NumLevelZeroTables) ||
if (i == 0 && s.levels[0].numTables() > s.kv.opt.NumLevelZeroTables) ||
(i > 0 && s.levels[i].getTotalSize() > s.levels[i].maxTotalSize) {
s.beingCompacted[i], s.beingCompacted[i+1] = true, true
return i
@@ -417,10 +417,15 @@ func (s *levelsController) doCompact(l int) error {
// remain unchanged.
var cds []compactDef
if l == 0 {
thisLevel.RLock() // For accessing tables. We may be adding new table into level 0.
top := make([]*table.Table, len(thisLevel.tables))
y.AssertTrue(len(thisLevel.tables) == copy(top, thisLevel.tables))
thisLevel.RUnlock()
smallest, biggest := keyRange(top)
// nextLevel.RLock() // Shouldn't be necessary.
left, right := overlappingTables(smallest, biggest, nextLevel.tables)
// nextLevel.RUnlock()
bot := make([]*table.Table, right-left)
copy(bot, nextLevel.tables[left:right])
cds = append(cds, compactDef{
@@ -569,6 +574,12 @@ func (s *levelHandler) tryAddLevel0Table(t *table.Table, verbose bool) bool {
return true
}
func (s *levelHandler) numTables() int {
s.RLock()
defer s.RUnlock()
return len(s.tables)
}
func (s *levelsController) close() {
if s.kv.opt.Verbose {
y.Printf("Sending close signal to compact workers\n")
skl/skl.go
+14 -4
View File
@@ -109,7 +109,10 @@ func (s *Skiplist) DecrRef() {
x = next
}
s.arenaPool.Put(s.arena)
s.arena = nil // Indicate we are closed. Good for testing.
// Indicate we are closed. Good for testing.
// Race condition here would suggest we are accessing skiplist when we are
// supposed to have no reference!
s.arena = nil
}
func (s *Skiplist) Valid() bool { return s.arena != nil }
@@ -156,7 +159,10 @@ func (s *node) setValue(arena *Arena, val []byte, meta byte) {
s.valueSize = uint16(len(val))
}
func (s *node) getNext(h int) *node { return (*node)(s.next[h]) }
func (s *node) getNext(h int) *node {
// return (*node)(s.next[h])
return (*node)(atomic.LoadPointer(&s.next[h]))
}
func (s *node) casNext(h int, old, val *node) bool {
return atomic.CompareAndSwapPointer(&s.next[h], unsafe.Pointer(old), unsafe.Pointer(val))
@@ -273,12 +279,16 @@ func (s *Skiplist) findSpliceForLevel(key []byte, before *node, level int) (*nod
}
}
func (s *Skiplist) Height() int32 {
return atomic.LoadInt32(&s.height)
}
// Put inserts the key-value pair.
func (s *Skiplist) Put(key []byte, val []byte, meta byte) {
// Since we allow overwrite, we may not need to create a new node. We might not even need to
// increase the height. Let's defer these actions.
listHeight := s.height
listHeight := s.Height()
var prev [kMaxHeight + 1]*node
var next [kMaxHeight + 1]*node
prev[listHeight] = s.head
@@ -297,7 +307,7 @@ func (s *Skiplist) Put(key []byte, val []byte, meta byte) {
x := newNode(s.arena, key, val, meta, height)
// Try to increase s.height via CAS.
listHeight = s.height
listHeight = s.Height()
for height > int(listHeight) {
if atomic.CompareAndSwapInt32(&s.height, listHeight, int32(height)) {
// Successfully increased skiplist.height.
skl/skl_test.go
+4 -3
View File
@@ -22,6 +22,7 @@ import (
"math/rand"
"strconv"
"sync"
"sync/atomic"
"testing"
"github.com/stretchr/testify/require"
@@ -161,7 +162,7 @@ func TestOneKey(t *testing.T) {
}(i)
}
// We expect that at least some write made it such that some read returns a value.
var sawValue bool
var sawValue int32
for i := 0; i < n; i++ {
wg.Add(1)
go func() {
@@ -170,14 +171,14 @@ func TestOneKey(t *testing.T) {
if p == nil {
return
}
sawValue = true
atomic.StoreInt32(&sawValue, 1)
v, err := strconv.Atoi(string(p))
require.NoError(t, err)
require.True(t, 0 <= v && v < n)
}()
}
wg.Wait()
require.True(t, sawValue)
require.True(t, sawValue > 0)
require.EqualValues(t, 1, length(l))
}