cerc-io/lotus
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

refactor: eliminate distinction between markset and markset visitors

Browse Source
This commit is contained in:
vyzo 2022-01-25 16:31:45 +02:00
parent a2468656e4
commit 25284b5325
8 changed files with 24 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
blockstore/splitstore/markset.go
View File

@ -10,20 +10,12 @@ import (
var errMarkSetClosed = errors.New("markset closed") var errMarkSetClosed = errors.New("markset closed")
// MarkSet is a utility to keep track of seen CID, and later query for them. // MarkSet is an interface for tracking CIDs during chain and object walks
//
// * If the expected dataset is large, it can be backed by a datastore (e.g. bbolt).
// * If a probabilistic result is acceptable, it can be backed by a bloom filter
type MarkSet interface { type MarkSet interface {
ObjectVisitor
Mark(cid.Cid) error Mark(cid.Cid) error
Has(cid.Cid) (bool, error) Has(cid.Cid) (bool, error)
Close() error Close() error
SetConcurrent()
}
type MarkSetVisitor interface {
MarkSet
ObjectVisitor
} }
type MarkSetEnv interface { type MarkSetEnv interface {
@ -31,11 +23,7 @@ type MarkSetEnv interface {
// name is a unique name for this markset, mapped to the filesystem in disk-backed environments // name is a unique name for this markset, mapped to the filesystem in disk-backed environments
// sizeHint is a hint about the expected size of the markset // sizeHint is a hint about the expected size of the markset
Create(name string, sizeHint int64) (MarkSet, error) Create(name string, sizeHint int64) (MarkSet, error)
// CreateVisitor is like Create, but returns a wider interface that supports atomic visits. // Close closes the markset
// It may not be supported by some markset types (e.g. bloom).
CreateVisitor(name string, sizeHint int64) (MarkSetVisitor, error)
// SupportsVisitor returns true if the marksets created by this environment support the visitor interface.
SupportsVisitor() bool
Close() error Close() error
} }

13
blockstore/splitstore/markset_badger.go
View File

@ -34,7 +34,6 @@ type BadgerMarkSet struct {
} }
var _ MarkSet = (*BadgerMarkSet)(nil) var _ MarkSet = (*BadgerMarkSet)(nil)
var _ MarkSetVisitor = (*BadgerMarkSet)(nil)
var badgerMarkSetBatchSize = 16384 var badgerMarkSetBatchSize = 16384
@ -48,7 +47,7 @@ func NewBadgerMarkSetEnv(path string) (MarkSetEnv, error) {
return &BadgerMarkSetEnv{path: msPath}, nil return &BadgerMarkSetEnv{path: msPath}, nil
} }
func (e *BadgerMarkSetEnv) create(name string, sizeHint int64) (*BadgerMarkSet, error) { func (e *BadgerMarkSetEnv) Create(name string, sizeHint int64) (MarkSet, error) {
name += ".tmp" name += ".tmp"
path := filepath.Join(e.path, name) path := filepath.Join(e.path, name)
@ -68,16 +67,6 @@ func (e *BadgerMarkSetEnv) create(name string, sizeHint int64) (*BadgerMarkSet,
return ms, nil return ms, nil
} }
func (e *BadgerMarkSetEnv) Create(name string, sizeHint int64) (MarkSet, error) {
return e.create(name, sizeHint)
}
func (e *BadgerMarkSetEnv) CreateVisitor(name string, sizeHint int64) (MarkSetVisitor, error) {
return e.create(name, sizeHint)
}
func (e *BadgerMarkSetEnv) SupportsVisitor() bool { return true }
func (e *BadgerMarkSetEnv) Close() error { func (e *BadgerMarkSetEnv) Close() error {
return os.RemoveAll(e.path) return os.RemoveAll(e.path)
} }

44
blockstore/splitstore/markset_map.go
View File

@ -13,42 +13,27 @@ var _ MarkSetEnv = (*MapMarkSetEnv)(nil)
type MapMarkSet struct { type MapMarkSet struct {
mx sync.RWMutex mx sync.RWMutex
set map[string]struct{} set map[string]struct{}
ts bool
} }
var _ MarkSet = (*MapMarkSet)(nil) var _ MarkSet = (*MapMarkSet)(nil)
var _ MarkSetVisitor = (*MapMarkSet)(nil)
func NewMapMarkSetEnv() (*MapMarkSetEnv, error) { func NewMapMarkSetEnv() (*MapMarkSetEnv, error) {
return &MapMarkSetEnv{}, nil return &MapMarkSetEnv{}, nil
} }
func (e *MapMarkSetEnv) create(name string, sizeHint int64) (*MapMarkSet, error) { func (e *MapMarkSetEnv) Create(name string, sizeHint int64) (MarkSet, error) {
return &MapMarkSet{ return &MapMarkSet{
set: make(map[string]struct{}, sizeHint), set: make(map[string]struct{}, sizeHint),
}, nil }, nil
} }
func (e *MapMarkSetEnv) Create(name string, sizeHint int64) (MarkSet, error) {
return e.create(name, sizeHint)
}
func (e *MapMarkSetEnv) CreateVisitor(name string, sizeHint int64) (MarkSetVisitor, error) {
return e.create(name, sizeHint)
}
func (e *MapMarkSetEnv) SupportsVisitor() bool { return true }
func (e *MapMarkSetEnv) Close() error { func (e *MapMarkSetEnv) Close() error {
return nil return nil
} }
func (s *MapMarkSet) Mark(cid cid.Cid) error { func (s *MapMarkSet) Mark(cid cid.Cid) error {
if s.ts { s.mx.Lock()
s.mx.Lock() defer s.mx.Unlock()
defer s.mx.Unlock()
}
if s.set == nil { if s.set == nil {
return errMarkSetClosed return errMarkSetClosed
@ -59,10 +44,8 @@ func (s *MapMarkSet) Mark(cid cid.Cid) error {
} }
func (s *MapMarkSet) Has(cid cid.Cid) (bool, error) { func (s *MapMarkSet) Has(cid cid.Cid) (bool, error) {
if s.ts { s.mx.RLock()
s.mx.RLock() defer s.mx.RUnlock()
defer s.mx.RUnlock()
}
if s.set == nil { if s.set == nil {
return false, errMarkSetClosed return false, errMarkSetClosed
@ -73,10 +56,8 @@ func (s *MapMarkSet) Has(cid cid.Cid) (bool, error) {
} }
func (s *MapMarkSet) Visit(c cid.Cid) (bool, error) { func (s *MapMarkSet) Visit(c cid.Cid) (bool, error) {
if s.ts { s.mx.Lock()
s.mx.Lock() defer s.mx.Unlock()
defer s.mx.Unlock()
}
if s.set == nil { if s.set == nil {
return false, errMarkSetClosed return false, errMarkSetClosed
@ -92,14 +73,9 @@ func (s *MapMarkSet) Visit(c cid.Cid) (bool, error) {
} }
func (s *MapMarkSet) Close() error { func (s *MapMarkSet) Close() error {
if s.ts { s.mx.Lock()
s.mx.Lock() defer s.mx.Unlock()
defer s.mx.Unlock()
}
s.set = nil s.set = nil
return nil return nil
} }
func (s *MapMarkSet) SetConcurrent() {
s.ts = true
}

2
blockstore/splitstore/markset_test.go
View File

@ -167,7 +167,7 @@ func testMarkSetVisitor(t *testing.T, lsType string) {
} }
defer env.Close() //nolint:errcheck defer env.Close() //nolint:errcheck
visitor, err := env.CreateVisitor("test", 0) visitor, err := env.Create("test", 0)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }

4
blockstore/splitstore/splitstore.go
View File

@ -186,10 +186,6 @@ func Open(path string, ds dstore.Datastore, hot, cold bstore.Blockstore, cfg *Co
return nil, err return nil, err
} }
if !markSetEnv.SupportsVisitor() {
return nil, xerrors.Errorf("markset type does not support atomic visitors")
}
// and now we can make a SplitStore // and now we can make a SplitStore
ss := &SplitStore{ ss := &SplitStore{
cfg: cfg, cfg: cfg,

2
blockstore/splitstore/splitstore_check.go
View File

@ -84,7 +84,7 @@ func (s *SplitStore) doCheck(curTs *types.TipSet) error {
var coldCnt, missingCnt int64 var coldCnt, missingCnt int64
visitor, err := s.markSetEnv.CreateVisitor("check", 0) visitor, err := s.markSetEnv.Create("check", 0)
if err != nil { if err != nil {
return xerrors.Errorf("error creating visitor: %w", err) return xerrors.Errorf("error creating visitor: %w", err)
} }

14
blockstore/splitstore/splitstore_compact.go
View File

@ -227,7 +227,7 @@ func (s *SplitStore) trackTxnRefMany(cids []cid.Cid) {
} }
// protect all pending transactional references // protect all pending transactional references
func (s *SplitStore) protectTxnRefs(markSet MarkSetVisitor) error { func (s *SplitStore) protectTxnRefs(markSet MarkSet) error {
for { for {
var txnRefs map[cid.Cid]struct{} var txnRefs map[cid.Cid]struct{}
@ -299,7 +299,7 @@ func (s *SplitStore) protectTxnRefs(markSet MarkSetVisitor) error {
// transactionally protect a reference by walking the object and marking. // transactionally protect a reference by walking the object and marking.
// concurrent markings are short circuited by checking the markset. // concurrent markings are short circuited by checking the markset.
func (s *SplitStore) doTxnProtect(root cid.Cid, markSet MarkSetVisitor) error { func (s *SplitStore) doTxnProtect(root cid.Cid, markSet MarkSet) error {
if err := s.checkClosing(); err != nil { if err := s.checkClosing(); err != nil {
return err return err
} }
@ -397,7 +397,7 @@ func (s *SplitStore) doCompact(curTs *types.TipSet) error {
log.Infow("running compaction", "currentEpoch", currentEpoch, "baseEpoch", s.baseEpoch, "boundaryEpoch", boundaryEpoch, "inclMsgsEpoch", inclMsgsEpoch, "compactionIndex", s.compactionIndex) log.Infow("running compaction", "currentEpoch", currentEpoch, "baseEpoch", s.baseEpoch, "boundaryEpoch", boundaryEpoch, "inclMsgsEpoch", inclMsgsEpoch, "compactionIndex", s.compactionIndex)
markSet, err := s.markSetEnv.CreateVisitor("live", s.markSetSize) markSet, err := s.markSetEnv.Create("live", s.markSetSize)
if err != nil { if err != nil {
return xerrors.Errorf("error creating mark set: %w", err) return xerrors.Errorf("error creating mark set: %w", err)
} }
@ -602,8 +602,8 @@ func (s *SplitStore) beginTxnProtect() {
s.txnMissing = make(map[cid.Cid]struct{}) s.txnMissing = make(map[cid.Cid]struct{})
} }
func (s *SplitStore) beginTxnMarking(markSet MarkSetVisitor) { func (s *SplitStore) beginTxnMarking(markSet MarkSet) {
markSet.SetConcurrent() log.Info("beginning transactional marking")
} }
func (s *SplitStore) endTxnProtect() { func (s *SplitStore) endTxnProtect() {
@ -1011,7 +1011,7 @@ func (s *SplitStore) purgeBatch(cids []cid.Cid, deleteBatch func([]cid.Cid) erro
return nil return nil
} }
func (s *SplitStore) purge(cids []cid.Cid, markSet MarkSetVisitor) error { func (s *SplitStore) purge(cids []cid.Cid, markSet MarkSet) error {
deadCids := make([]cid.Cid, 0, batchSize) deadCids := make([]cid.Cid, 0, batchSize)
var purgeCnt, liveCnt int var purgeCnt, liveCnt int
defer func() { defer func() {
@ -1077,7 +1077,7 @@ func (s *SplitStore) purge(cids []cid.Cid, markSet MarkSetVisitor) error {
// have this gem[TM]. // have this gem[TM].
// My best guess is that they are parent message receipts or yet to be computed state roots; magik // My best guess is that they are parent message receipts or yet to be computed state roots; magik
// thinks the cause may be block validation. // thinks the cause may be block validation.
func (s *SplitStore) waitForMissingRefs(markSet MarkSetVisitor) { func (s *SplitStore) waitForMissingRefs(markSet MarkSet) {
s.txnLk.Lock() s.txnLk.Lock()
missing := s.txnMissing missing := s.txnMissing
s.txnMissing = nil s.txnMissing = nil

2
blockstore/splitstore/splitstore_warmup.go
View File

@ -60,7 +60,7 @@ func (s *SplitStore) doWarmup(curTs *types.TipSet) error {
xcount := int64(0) xcount := int64(0)
missing := int64(0) missing := int64(0)
visitor, err := s.markSetEnv.CreateVisitor("warmup", 0) visitor, err := s.markSetEnv.Create("warmup", 0)
if err != nil { if err != nil {
return xerrors.Errorf("error creating visitor: %w", err) return xerrors.Errorf("error creating visitor: %w", err)
} }