core: split (*MemCachedStore) Seek and SeekAsync methods

Use SeekAsync for System.Storage.Find and Seek for the rest of cases.
This commit is contained in:
Anna Shaleva 2021-10-19 18:03:47 +03:00
parent dcda7bec63
commit 3450371910
2 changed files with 84 additions and 85 deletions

View file

@ -338,10 +338,13 @@ func (dao *Simple) GetStorageItemsWithPrefix(id int32, prefix []byte) ([]state.S
// Seek executes f for all items with a given prefix. // Seek executes f for all items with a given prefix.
// If key is to be used outside of f, they may not be copied. // If key is to be used outside of f, they may not be copied.
func (dao *Simple) Seek(id int32, prefix []byte, f func(k, v []byte)) { func (dao *Simple) Seek(id int32, prefix []byte, f func(k, v []byte)) {
res := dao.SeekAsync(context.Background(), id, prefix) lookupKey := makeStorageItemKey(id, nil)
for r := range res { if prefix != nil {
f(r.Key, r.Value) lookupKey = append(lookupKey, prefix...)
} }
dao.Store.Seek(lookupKey, func(k, v []byte) {
f(k[len(lookupKey):], v)
})
} }
// SeekAsync sends all storage items matching given prefix to a channel and returns // SeekAsync sends all storage items matching given prefix to a channel and returns

View file

@ -91,16 +91,28 @@ func (s *MemCachedStore) GetBatch() *MemBatch {
// Seek implements the Store interface. // Seek implements the Store interface.
func (s *MemCachedStore) Seek(key []byte, f func(k, v []byte)) { func (s *MemCachedStore) Seek(key []byte, f func(k, v []byte)) {
seekres := s.SeekAsync(context.Background(), key, false) s.seek(context.Background(), key, false, f)
for kv := range seekres {
f(kv.Key, kv.Value)
}
} }
// SeekAsync returns non-buffered channel with matching KeyValue pairs. Key and // SeekAsync returns non-buffered channel with matching KeyValue pairs. Key and
// value slices may not be copied and may be modified. SeekAsync can guarantee // value slices may not be copied and may be modified. SeekAsync can guarantee
// that key-value items are sorted by key in ascending way. // that key-value items are sorted by key in ascending way.
func (s *MemCachedStore) SeekAsync(ctx context.Context, key []byte, cutPrefix bool) chan KeyValue { func (s *MemCachedStore) SeekAsync(ctx context.Context, key []byte, cutPrefix bool) chan KeyValue {
res := make(chan KeyValue)
go func() {
s.seek(ctx, key, cutPrefix, func(k, v []byte) {
res <- KeyValue{
Key: k,
Value: v,
}
})
close(res)
}()
return res
}
func (s *MemCachedStore) seek(ctx context.Context, key []byte, cutPrefix bool, f func(k, v []byte)) {
// Create memory store `mem` and `del` snapshot not to hold the lock. // Create memory store `mem` and `del` snapshot not to hold the lock.
var memRes []KeyValueExists var memRes []KeyValueExists
sk := string(key) sk := string(key)
@ -132,10 +144,6 @@ func (s *MemCachedStore) SeekAsync(ctx context.Context, key []byte, cutPrefix bo
return bytes.Compare(memRes[i].Key, memRes[j].Key) < 0 return bytes.Compare(memRes[i].Key, memRes[j].Key) < 0
}) })
var seekres = make(chan KeyValue)
// Seek over persistent store.
go func() {
var ( var (
done bool done bool
iMem int iMem int
@ -147,7 +155,6 @@ func (s *MemCachedStore) SeekAsync(ctx context.Context, key []byte, cutPrefix bo
haveMem = true haveMem = true
iMem++ iMem++
} }
// Merge results of seek operations in ascending order. // Merge results of seek operations in ascending order.
ps.Seek(key, func(k, v []byte) { ps.Seek(key, func(k, v []byte) {
if done { if done {
@ -170,10 +177,7 @@ func (s *MemCachedStore) SeekAsync(ctx context.Context, key []byte, cutPrefix bo
if cutPrefix { if cutPrefix {
kvMem.Key = kvMem.Key[len(key):] kvMem.Key = kvMem.Key[len(key):]
} }
seekres <- KeyValue{ f(kvMem.Key, kvMem.Value)
Key: kvMem.Key,
Value: kvMem.Value,
}
} }
if iMem < len(memRes) { if iMem < len(memRes) {
kvMem = memRes[iMem] kvMem = memRes[iMem]
@ -187,13 +191,12 @@ func (s *MemCachedStore) SeekAsync(ctx context.Context, key []byte, cutPrefix bo
if cutPrefix { if cutPrefix {
kvPs.Key = kvPs.Key[len(key):] kvPs.Key = kvPs.Key[len(key):]
} }
seekres <- kvPs f(kvPs.Key, kvPs.Value)
} }
break loop break loop
} }
} }
} }
}) })
if !done && haveMem { if !done && haveMem {
loop: loop:
@ -207,18 +210,11 @@ func (s *MemCachedStore) SeekAsync(ctx context.Context, key []byte, cutPrefix bo
if cutPrefix { if cutPrefix {
kvMem.Key = kvMem.Key[len(key):] kvMem.Key = kvMem.Key[len(key):]
} }
seekres <- KeyValue{ f(kvMem.Key, kvMem.Value)
Key: kvMem.Key,
Value: kvMem.Value,
} }
} }
} }
} }
}
close(seekres)
}()
return seekres
} }
// Persist flushes all the MemoryStore contents into the (supposedly) persistent // Persist flushes all the MemoryStore contents into the (supposedly) persistent