diff --git a/.changeset/change_chain_manager_mutex_to_rwmutex_to_enable_parallel_reads.md b/.changeset/change_chain_manager_mutex_to_rwmutex_to_enable_parallel_reads.md new file mode 100644 index 00000000..f047e0bb --- /dev/null +++ b/.changeset/change_chain_manager_mutex_to_rwmutex_to_enable_parallel_reads.md @@ -0,0 +1,5 @@ +--- +default: patch +--- + +# Change Chain Manager Mutex to RWMutex to enable parallel reads diff --git a/chain/concurrency_test.go b/chain/concurrency_test.go new file mode 100644 index 00000000..f3b87790 --- /dev/null +++ b/chain/concurrency_test.go @@ -0,0 +1,223 @@ +package chain_test + +import ( + "fmt" + "path/filepath" + "sync" + "testing" + "time" + + "go.sia.tech/core/types" + "go.sia.tech/coreutils" + "go.sia.tech/coreutils/chain" + "go.sia.tech/coreutils/testutil" +) + +// TestManagerConcurrentReadsWrites hammers the Manager's read methods from many +// goroutines while a writer continuously appends blocks and periodically forces +// reorgs. It exercises the RWMutex + per-reader snapshot design and, in +// particular, the flush-before-unlock invariant: once a tip is observable, the +// block and state behind it must be committed, so any later snapshot can read +// them. If a writer ever released the lock without flushing (m.mu.Unlock +// instead of m.writeUnlock), a reader could observe an advanced tip whose block +// is missing from the snapshot it reads, and the assertions below would fail. +// +// It runs against both store backends. The bbolt backend is the one that +// actually exercises the invariant: its snapshots are isolated read +// transactions that do not observe uncommitted writes, whereas MemDB has no +// MVCC and reads pending writes regardless of flushing. Run with -race to also +// catch any cross-goroutine sharing of store state. +func TestManagerConcurrentReadsWrites(t *testing.T) { + for _, tc := range []struct { + name string + makeDB func(testing.TB) chain.DB + }{ + {"MemDB", func(testing.TB) chain.DB { return chain.NewMemDB() }}, + {"BoltDB", func(tb testing.TB) chain.DB { + db, err := coreutils.OpenBoltChainDB(filepath.Join(tb.TempDir(), "consensus.db")) + if err != nil { + tb.Fatal(err) + } + tb.Cleanup(func() { db.Close() }) + return db + }}, + } { + t.Run(tc.name, func(t *testing.T) { + runConcurrentReadsWrites(t, tc.makeDB(t)) + }) + } +} + +func runConcurrentReadsWrites(t *testing.T, db chain.DB) { + n, genesis := testutil.V2Network() + store, tipState, err := chain.NewDBStore(db, n, genesis, nil) + if err != nil { + t.Fatal(err) + } + cm := chain.NewManager(store, tipState) + + // give readers some history to traverse + for i := 0; i < 20; i++ { + b, ok := coreutils.MineBlock(cm, types.VoidAddress, time.Second) + if !ok { + t.Fatal("PoW failed") + } else if err := cm.AddBlocks([]types.Block{b}); err != nil { + t.Fatal(err) + } + } + + const ( + totalBlocks = 200 + readers = 8 + reorgInterval = 20 // force a reorg roughly every N writer iterations + reorgDepth = 3 // blocks reverted per forced reorg + ) + + done := make(chan struct{}) + var doneOnce sync.Once + closeDone := func() { doneOnce.Do(func() { close(done) }) } + + var failOnce sync.Once + var failure error + fail := func(e error) { + failOnce.Do(func() { failure = e }) + closeDone() + } + + // forkReorg builds a competing chain reorgDepth+2 blocks long from an + // ancestor reorgDepth back and submits it, forcing the Manager to revert + // and reapply. The fork is mined on an independent store, so building it + // never touches cm's store; only the final AddBlocks does. + forkReorg := func() bool { + tip := cm.Tip() + if tip.Height <= reorgDepth { + return false + } + aidx, ok := cm.BestIndex(tip.Height - reorgDepth) + if !ok { + return false + } + ab, ok := cm.Block(aidx.ID) + if !ok { + return false + } + parentState, ok := cm.State(ab.ParentID) + if !ok { + return false + } + fdb, ftip, err := chain.NewDBStoreAtCheckpoint(chain.NewMemDB(), parentState, ab, nil) + if err != nil { + fail(fmt.Errorf("fork: checkpoint init: %w", err)) + return false + } + fcm := chain.NewManager(fdb, ftip) + fork := make([]types.Block, 0, reorgDepth+2) + for j := 0; j < reorgDepth+2; j++ { + b, ok := coreutils.MineBlock(fcm, types.VoidAddress, 5*time.Second) + if !ok { + fail(fmt.Errorf("fork: PoW failed")) + return false + } else if err := fcm.AddBlocks([]types.Block{b}); err != nil { + fail(fmt.Errorf("fork: AddBlocks: %w", err)) + return false + } + fork = append(fork, b) + } + // the fork is strictly heavier (cm cannot advance while this single + // writer builds it), so this triggers a reorg. + if err := cm.AddBlocks(fork); err != nil { + fail(fmt.Errorf("writer: reorg AddBlocks: %w", err)) + return false + } + return true + } + + var wg sync.WaitGroup + + // writer: extend the chain, periodically forcing a reorg. + wg.Add(1) + go func() { + defer wg.Done() + defer closeDone() + for i := 0; i < totalBlocks; i++ { + if i > 0 && i%reorgInterval == 0 { + if forkReorg() { + continue + } + select { + case <-done: + return // forkReorg failed + default: + } + } + b, ok := coreutils.MineBlock(cm, types.VoidAddress, 5*time.Second) + if !ok { + fail(fmt.Errorf("writer: PoW failed at block %d", i)) + return + } else if err := cm.AddBlocks([]types.Block{b}); err != nil { + fail(fmt.Errorf("writer: AddBlocks failed at block %d: %w", i, err)) + return + } + } + }() + + genesisIndex := types.ChainIndex{Height: 0, ID: genesis.ID()} + for r := 0; r < readers; r++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-done: + return + default: + } + + // The core invariant, which survives reorgs: whatever tip the + // Manager reports, that tip's block and state must be committed, + // so a fresh snapshot can read them. This is exactly what + // flush-before-unlock guarantees. (We can't assert + // BestIndex(tip.Height) == tip here, because a reorg may land + // between observing the tip and the lookup.) + tip := cm.Tip() + if b, ok := cm.Block(tip.ID); !ok { + fail(fmt.Errorf("observed tip %v but its block is absent from the read snapshot", tip)) + return + } else if b.ID() != tip.ID { + fail(fmt.Errorf("Block(%v) returned block %v", tip.ID, b.ID())) + return + } + if _, ok := cm.State(tip.ID); !ok { + fail(fmt.Errorf("observed tip %v but its state is absent from the read snapshot", tip)) + return + } + + // multi-lookup reads (each runs in a single snapshot): must + // remain internally consistent and never error on a live best + // chain, even while reorgs are happening. + hist, err := cm.History() + if err != nil { + fail(fmt.Errorf("History: %w", err)) + return + } + if _, _, err := cm.BlocksForHistory(hist[:], 10); err != nil { + fail(fmt.Errorf("BlocksForHistory: %w", err)) + return + } + if _, _, err := cm.Headers(genesisIndex, 100); err != nil { + fail(fmt.Errorf("Headers: %w", err)) + return + } + if _, _, err := cm.UpdatesSince(types.ChainIndex{}, 10); err != nil { + fail(fmt.Errorf("UpdatesSince: %w", err)) + return + } + } + }() + } + + wg.Wait() + if failure != nil { + t.Fatal(failure) + } +} diff --git a/chain/db.go b/chain/db.go index a3ab8f23..79a7fa72 100644 --- a/chain/db.go +++ b/chain/db.go @@ -65,14 +65,46 @@ type DB interface { CreateBucket(name []byte) (DBBucket, error) Flush() error Cancel() + + // Snapshot returns a read-only, point-in-time view of the database. + // Implementations must guarantee that the returned ReadonlyDB observes a + // consistent snapshot and may be read concurrently with writes to the DB + // and with other snapshots. Callers must Close it when finished. + Snapshot() ReadonlyDB +} + +// A ReadonlyDB is a read-only, point-in-time view of a DB. It exposes +// no mutating methods. The underlying database cannot be modified through it. +type ReadonlyDB interface { + Bucket(name []byte) ReadonlyDBBucket + // Close releases the resources held by the snapshot. + Close() error +} + +// noSnapshotDB is a ReadonlyDB for databases that do not support true +// snapshots. Reads observe live state, so callers must exclude concurrent +// writes themselves; Close is a no-op. +type noSnapshotDB struct { + db interface { + Bucket(name []byte) DBBucket + } +} + +func (s noSnapshotDB) Bucket(name []byte) ReadonlyDBBucket { return s.db.Bucket(name) } +func (noSnapshotDB) Close() error { return nil } + +// A ReadonlyDBBucket is a read-only view of a DBBucket; it exposes no mutating +// methods, so a bucket obtained from a ReadonlyDB cannot be written to. +type ReadonlyDBBucket interface { + Get(key []byte) []byte + Iter() iter.Seq2[[]byte, []byte] } // A DBBucket is a set of key-value pairs. type DBBucket interface { - Get(key []byte) []byte + ReadonlyDBBucket Put(key, value []byte) error Delete(key []byte) error - Iter() iter.Seq2[[]byte, []byte] } // MemDB implements DB with an in-memory map. @@ -105,6 +137,14 @@ func (db *MemDB) Flush() error { return nil } +// Snapshot implements DB. MemDB has no internal locking to prevent a write +// through db racing a read from the returned snapshot, so it must only be used +// behind a Manager, which guards snapshot reads against writes with its +// RWMutex. +func (db *MemDB) Snapshot() ReadonlyDB { + return noSnapshotDB{db} +} + // Cancel implements DB. func (db *MemDB) Cancel() { for k := range db.puts { @@ -336,6 +376,14 @@ func (db *CacheDB) Cancel() { db.db.Cancel() } +// Snapshot implements DB. CacheDB has no internal locking to prevent a write +// through db racing a read from the returned snapshot, so it must only be used +// behind a Manager, which guards snapshot reads against writes with its +// RWMutex. +func (db *CacheDB) Snapshot() ReadonlyDB { + return noSnapshotDB{db} +} + // NewCacheDB returns a new CacheDB that wraps the given DB. func NewCacheDB(db DB) DB { return &CacheDB{ @@ -353,10 +401,17 @@ func check(err error) { // dbBucket is a helper type for implementing Store. type dbBucket struct { - b DBBucket + b ReadonlyDBBucket db *DBStore } +// writable returns the bucket as a DBBucket for mutation. It is only reached +// from write paths, which only run on a writable DBStore; such a store is +// always backed by a full DB, so its buckets always implement DBBucket. +func (b *dbBucket) writable() DBBucket { + return b.b.(DBBucket) +} + func (b *dbBucket) getRaw(key []byte) []byte { if b.b == nil { return nil @@ -379,7 +434,7 @@ func (b *dbBucket) get(key []byte, v types.DecoderFrom) bool { } func (b *dbBucket) putRaw(key, value []byte) { - check(b.b.Put(key, value)) + check(b.writable().Put(key, value)) b.db.unflushed += len(value) } @@ -392,7 +447,7 @@ func (b *dbBucket) put(key []byte, v types.EncoderTo) { } func (b *dbBucket) delete(key []byte) { - check(b.b.Delete(key)) + check(b.writable().Delete(key)) b.db.unflushed += len(key) } @@ -410,9 +465,26 @@ var ( keyHeight = []byte("Height") ) +// rwDB is the subset of DB that DBStore accesses directly. A read-only +// snapshot can satisfy it (see roDB), letting a snapshot reuse DBStore's +// accessors without exposing any write methods. +type rwDB interface { + Bucket(name []byte) ReadonlyDBBucket + Flush() error + Snapshot() ReadonlyDB +} + +// rwView adapts a full DB to the rwDB interface, narrowing Bucket's return type +// to ReadonlyDBBucket so that DBStore's shared read path cannot mutate through +// it. Write paths recover the full DBBucket via dbBucket.writable, which is +// sound because a writable DBStore is always backed by a full DB. +type rwView struct{ DB } + +func (v rwView) Bucket(name []byte) ReadonlyDBBucket { return v.DB.Bucket(name) } + // DBStore implements Store using a key-value database. type DBStore struct { - db DB + db rwDB n *consensus.Network // for getState enc types.Encoder @@ -948,6 +1020,38 @@ func (db *DBStore) Flush() error { return err } +// roDB adapts a ReadonlyDB to the rwDB interface so that a read-only DBStore +// can reuse DBStore's accessors. The snapshot is never written or +// re-snapshotted, so Flush and Snapshot are inert. +type roDB struct{ ReadonlyDB } + +func (roDB) Flush() error { return nil } +func (r roDB) Snapshot() ReadonlyDB { return r.ReadonlyDB } + +// readonlyStore is a read-only Store backed by a database snapshot. It embeds a +// *DBStore for its read accessors but is only ever exposed as a ReadonlyStore, +// so its inherited write methods are unreachable. +type readonlyStore struct { + *DBStore + rdb ReadonlyDB +} + +// Close releases the underlying snapshot. +func (s *readonlyStore) Close() error { + return s.rdb.Close() +} + +// Snapshot implements Store. It returns a read-only view of the store backed by +// a consistent snapshot of the underlying DB. The returned ReadonlyStore must +// be closed when finished and may be read concurrently with other snapshots. +func (db *DBStore) Snapshot() ReadonlyStore { + rdb := db.db.Snapshot() + return &readonlyStore{ + DBStore: &DBStore{db: roDB{rdb}, n: db.n}, + rdb: rdb, + } +} + // NewDBStore creates a new DBStore using the provided database. The tip state // is also returned. The DB will be automatically migrated if necessary. The // provided logger may be nil. @@ -970,7 +1074,7 @@ func NewDBStore(db DB, n *consensus.Network, genesisBlock types.Block, logger Mi } dbs := &DBStore{ - db: db, + db: rwView{db}, n: n, } @@ -1046,7 +1150,7 @@ func NewDBStoreAtCheckpoint(db DB, cs consensus.State, b types.Block, logger Mig } dbs := &DBStore{ - db: db, + db: rwView{db}, n: cs.Network, } diff --git a/chain/manager.go b/chain/manager.go index 8b053162..a13dab9d 100644 --- a/chain/manager.go +++ b/chain/manager.go @@ -66,11 +66,47 @@ type Store interface { ApplyBlock(s consensus.State, cau consensus.ApplyUpdate) RevertBlock(s consensus.State, cru consensus.RevertUpdate) Flush() error + + // Snapshot returns a read-only, point-in-time view of the Store. It may be + // read concurrently with other snapshots and must be closed when finished. + Snapshot() ReadonlyStore +} + +// A ReadonlyStore is a read-only, point-in-time view of a Store. It exposes +// only the Store's read methods, so the underlying storage cannot be modified +// through it. +type ReadonlyStore interface { + BestIndex(height uint64) (types.ChainIndex, bool) + Block(id types.BlockID) (types.Block, *consensus.V1BlockSupplement, bool) + Header(id types.BlockID) (types.BlockHeader, bool) + State(id types.BlockID) (consensus.State, bool) + AncestorTimestamp(id types.BlockID) (time.Time, bool) + // Close releases the snapshot. Callers may ignore the returned error. + Close() error +} + +// blockReader provides the reads needed by blockAndParent. Both Store and +// ReadonlyStore satisfy it. +type blockReader interface { + Block(id types.BlockID) (types.Block, *consensus.V1BlockSupplement, bool) + State(id types.BlockID) (consensus.State, bool) +} + +// storeReader is the subset of Store reads used to compute reorg paths and +// update transaction proofs. Both Store and ReadonlyStore satisfy it, so these +// (read-only) operations can run against the live store under the write lock or +// against a Snapshot under a read lock. +type storeReader interface { + BestIndex(height uint64) (types.ChainIndex, bool) + State(id types.BlockID) (consensus.State, bool) + Header(id types.BlockID) (types.BlockHeader, bool) + Block(id types.BlockID) (types.Block, *consensus.V1BlockSupplement, bool) + AncestorTimestamp(id types.BlockID) (time.Time, bool) } // blockAndParent returns the block with the specified ID, along with its parent // state. -func blockAndParent(s Store, id types.BlockID) (types.Block, *consensus.V1BlockSupplement, consensus.State, bool) { +func blockAndParent(s blockReader, id types.BlockID) (types.Block, *consensus.V1BlockSupplement, consensus.State, bool) { b, bs, ok := s.Block(id) cs, ok2 := s.State(b.ParentID) return b, bs, cs, ok && ok2 @@ -99,13 +135,32 @@ type Manager struct { lastRevertedV2 []types.V2Transaction } - mu sync.Mutex + mu sync.RWMutex +} + +// flushStore commits any pending store writes. Writer paths must not release +// m.mu without flushing first, or concurrent View readers would observe a +// snapshot older than m.tipState; use writeUnlock to release the write lock so +// the two cannot drift apart. +func (m *Manager) flushStore() { + if err := m.store.Flush(); err != nil { + panic(err) + } +} + +// writeUnlock flushes pending store writes, then releases the write lock. It +// must be used (never m.mu.Unlock directly) to release a lock taken with +// m.mu.Lock in a store-mutating path, so that flush-before-unlock cannot be +// forgotten. Pool-only writers that don't touch the store use m.mu.Unlock. +func (m *Manager) writeUnlock() { + m.flushStore() + m.mu.Unlock() } // TipState returns the consensus state for the current tip. func (m *Manager) TipState() consensus.State { - m.mu.Lock() - defer m.mu.Unlock() + m.mu.RLock() + defer m.mu.RUnlock() return m.tipState } @@ -116,36 +171,44 @@ func (m *Manager) Tip() types.ChainIndex { // Block returns the block with the specified ID. func (m *Manager) Block(id types.BlockID) (types.Block, bool) { - m.mu.Lock() - defer m.mu.Unlock() - b, _, ok := m.store.Block(id) + m.mu.RLock() + defer m.mu.RUnlock() + snap := m.store.Snapshot() + defer snap.Close() + b, _, ok := snap.Block(id) return b, ok } // State returns the state with the specified ID. func (m *Manager) State(id types.BlockID) (consensus.State, bool) { - m.mu.Lock() - defer m.mu.Unlock() - return m.store.State(id) + m.mu.RLock() + defer m.mu.RUnlock() + snap := m.store.Snapshot() + defer snap.Close() + return snap.State(id) } // BestIndex returns the index of the block at the specified height within the // best chain. func (m *Manager) BestIndex(height uint64) (types.ChainIndex, bool) { - m.mu.Lock() - defer m.mu.Unlock() - return m.store.BestIndex(height) + m.mu.RLock() + defer m.mu.RUnlock() + snap := m.store.Snapshot() + defer snap.Close() + return snap.BestIndex(height) } // MinReorgIndex returns the index on the best chain below which the manager // cannot perform a reorg. func (m *Manager) MinReorgIndex() types.ChainIndex { - m.mu.Lock() - defer m.mu.Unlock() + m.mu.RLock() + defer m.mu.RUnlock() + snap := m.store.Snapshot() + defer snap.Close() index := m.tipState.Index for index.Height > 0 { - prevIndex, ok := m.store.BestIndex(index.Height - 1) - _, _, ok2 := m.store.Block(prevIndex.ID) + prevIndex, ok := snap.BestIndex(index.Height - 1) + _, _, ok2 := snap.Block(prevIndex.ID) if !ok || !ok2 { break } @@ -158,8 +221,10 @@ func (m *Manager) MinReorgIndex() types.ChainIndex { // the 10 most-recent blocks, and subsequently spaced exponentionally farther // apart until reaching the genesis block. func (m *Manager) History() ([32]types.BlockID, error) { - m.mu.Lock() - defer m.mu.Unlock() + m.mu.RLock() + defer m.mu.RUnlock() + snap := m.store.Snapshot() + defer snap.Close() tipHeight := m.tipState.Index.Height histHeight := func(i int) uint64 { @@ -174,7 +239,7 @@ func (m *Manager) History() ([32]types.BlockID, error) { } var history [32]types.BlockID for i := range history { - index, ok := m.store.BestIndex(histHeight(i)) + index, ok := snap.BestIndex(histHeight(i)) if !ok { break } @@ -187,18 +252,20 @@ func (m *Manager) History() ([32]types.BlockID, error) { // which must be on the best chain. It also returns the number of headers // between the end of the returned slice and the current tip. func (m *Manager) Headers(index types.ChainIndex, maxHeaders uint64) ([]types.BlockHeader, uint64, error) { - m.mu.Lock() - defer m.mu.Unlock() - if bestIndex, ok := m.store.BestIndex(index.Height); !ok || bestIndex != index { + m.mu.RLock() + defer m.mu.RUnlock() + snap := m.store.Snapshot() + defer snap.Close() + if bestIndex, ok := snap.BestIndex(index.Height); !ok || bestIndex != index { return nil, 0, fmt.Errorf("index %v is not on our best chain", index) } maxHeaders = min(maxHeaders, m.tipState.Index.Height-index.Height) headers := make([]types.BlockHeader, maxHeaders) for i := range headers { - index, _ := m.store.BestIndex(index.Height + uint64(i) + 1) - bh, ok := m.store.Header(index.ID) + next, _ := snap.BestIndex(index.Height + uint64(i) + 1) + bh, ok := snap.Header(next.ID) if !ok { - return nil, 0, fmt.Errorf("missing block header %v", index) + return nil, 0, fmt.Errorf("missing block header %v", next) } headers[i] = bh } @@ -211,13 +278,15 @@ func (m *Manager) Headers(index types.ChainIndex, maxHeaders uint64) ([]types.Bl // returns the number of blocks between the end of the returned slice and the // current tip. func (m *Manager) BlocksForHistory(history []types.BlockID, maxBlocks uint64) ([]types.Block, uint64, error) { - m.mu.Lock() - defer m.mu.Unlock() + m.mu.RLock() + defer m.mu.RUnlock() + snap := m.store.Snapshot() + defer snap.Close() var attachHeight uint64 for _, id := range history { - if cs, ok := m.store.State(id); !ok { + if cs, ok := snap.State(id); !ok { continue - } else if index, ok := m.store.BestIndex(cs.Index.Height); ok && index == cs.Index { + } else if index, ok := snap.BestIndex(cs.Index.Height); ok && index == cs.Index { attachHeight = cs.Index.Height break } @@ -227,11 +296,11 @@ func (m *Manager) BlocksForHistory(history []types.BlockID, maxBlocks uint64) ([ } blocks := make([]types.Block, maxBlocks) for i := range blocks { - index, ok := m.store.BestIndex(attachHeight + uint64(i) + 1) + index, ok := snap.BestIndex(attachHeight + uint64(i) + 1) if !ok { return nil, 0, fmt.Errorf("unknown block at height %v", attachHeight+uint64(i)+1) } - b, _, ok := m.store.Block(index.ID) + b, _, ok := snap.Block(index.ID) if !ok { return nil, 0, fmt.Errorf("missing block %v", index) } @@ -244,7 +313,7 @@ func (m *Manager) BlocksForHistory(history []types.BlockID, maxBlocks uint64) ([ // belong to may become the new best chain, triggering a reorg. func (m *Manager) AddBlocks(blocks []types.Block) error { m.mu.Lock() - defer m.mu.Unlock() + defer m.writeUnlock() if len(blocks) == 0 { return nil } @@ -298,7 +367,7 @@ func (m *Manager) AddBlocks(blocks []types.Block) error { for _, fn := range m.onPool { fns = append(fns, fn) } - m.mu.Unlock() + m.writeUnlock() for _, fn := range fns { fn() } @@ -312,7 +381,7 @@ func (m *Manager) AddBlocks(blocks []types.Block) error { // sufficient work, it may become the new best chain, triggering a reorg. func (m *Manager) AddValidatedV2Blocks(blocks []types.Block, states []consensus.State) error { m.mu.Lock() - defer m.mu.Unlock() + defer m.writeUnlock() if len(blocks) == 0 { return nil } else if len(states) != len(blocks) { @@ -349,7 +418,7 @@ func (m *Manager) AddValidatedV2Blocks(blocks []types.Block, states []consensus. for _, fn := range m.onPool { fns = append(fns, fn) } - m.mu.Unlock() + m.writeUnlock() for _, fn := range fns { fn() } @@ -433,14 +502,14 @@ func (m *Manager) applyTip(index types.ChainIndex) error { return nil } -func (m *Manager) reorgPath(a, b types.ChainIndex, maxLen int) (revert, apply []types.ChainIndex, err error) { +func (m *Manager) reorgPath(s storeReader, a, b types.ChainIndex, maxLen int) (revert, apply []types.ChainIndex, err error) { // helper function for "rewinding" to the parent index rewind := func(index *types.ChainIndex) bool { if len(revert)+len(apply) > maxLen { err = fmt.Errorf("reorg path is too long (-%d +%d, max %d)", len(revert), len(apply), maxLen) return false } - bh, ok := m.store.Header(index.ID) + bh, ok := s.Header(index.ID) if !ok { err = fmt.Errorf("%w %v", ErrMissingBlock, *index) } else { @@ -465,7 +534,7 @@ func (m *Manager) reorgPath(a, b types.ChainIndex, maxLen int) (revert, apply [] // special case: if a is uninitialized, we're starting from genesis if a == (types.ChainIndex{}) { - a, _ = m.store.BestIndex(0) + a, _ = s.BestIndex(0) apply = append(apply, a) } @@ -483,7 +552,7 @@ func (m *Manager) reorgPath(a, b types.ChainIndex, maxLen int) (revert, apply [] } func (m *Manager) reorgTo(index types.ChainIndex) error { - revert, apply, err := m.reorgPath(m.tipState.Index, index, math.MaxInt) + revert, apply, err := m.reorgPath(m.store, m.tipState.Index, index, math.MaxInt) if err != nil { return err } @@ -537,7 +606,7 @@ func (m *Manager) reorgTo(index types.ChainIndex) error { // a large backlog. func (m *Manager) PruneBlocks(height uint64) { m.mu.Lock() - defer m.mu.Unlock() + defer m.writeUnlock() for h := height; h > 0; h-- { index, ok := m.store.BestIndex(h - 1) @@ -553,17 +622,19 @@ func (m *Manager) PruneBlocks(height uint64) { // UpdatesSince returns at most max updates on the path between index and the // Manager's current tip. func (m *Manager) UpdatesSince(index types.ChainIndex, maxBlocks int) (rus []RevertUpdate, aus []ApplyUpdate, err error) { - m.mu.Lock() - defer m.mu.Unlock() + m.mu.RLock() + defer m.mu.RUnlock() + snap := m.store.Snapshot() + defer snap.Close() onBestChain := func(index types.ChainIndex) bool { - bi, _ := m.store.BestIndex(index.Height) + bi, _ := snap.BestIndex(index.Height) return bi.ID == index.ID || index == types.ChainIndex{} } for index != m.tipState.Index && len(rus)+len(aus) < maxBlocks { // revert until we are on the best chain, then apply if !onBestChain(index) { - b, bs, cs, ok := blockAndParent(m.store, index.ID) + b, bs, cs, ok := blockAndParent(snap, index.ID) if !ok { return nil, nil, fmt.Errorf("%w %v", ErrMissingBlock, index) } else if bs == nil { @@ -575,17 +646,17 @@ func (m *Manager) UpdatesSince(index types.ChainIndex, maxBlocks int) (rus []Rev } else { // special case: if index is uninitialized, we're starting from genesis if index == (types.ChainIndex{}) { - index, _ = m.store.BestIndex(0) + index, _ = snap.BestIndex(0) } else { - index, _ = m.store.BestIndex(index.Height + 1) + index, _ = snap.BestIndex(index.Height + 1) } - b, bs, cs, ok := blockAndParent(m.store, index.ID) + b, bs, cs, ok := blockAndParent(snap, index.ID) if !ok { return nil, nil, fmt.Errorf("%w %v", ErrMissingBlock, index) } else if bs == nil { return nil, nil, fmt.Errorf("missing supplement for block %v", index) } - ancestorTimestamp, ok := m.store.AncestorTimestamp(b.ParentID) + ancestorTimestamp, ok := snap.AncestorTimestamp(b.ParentID) if !ok && index.Height != 0 { return nil, nil, fmt.Errorf("missing ancestor timestamp for block %v", b.ParentID) } @@ -1173,7 +1244,7 @@ func (m *Manager) V2TransactionSet(basis types.ChainIndex, txn types.V2Transacti } // update the transaction's basis to match tip - txns, err := m.updateV2TransactionProofs(append(parents, txn), basis, m.tipState.Index) + txns, err := m.updateV2TransactionProofs(m.store, append(parents, txn), basis, m.tipState.Index) if err != nil { return types.ChainIndex{}, nil, fmt.Errorf("failed to update transaction set basis: %w", err) } @@ -1205,10 +1276,10 @@ func (m *Manager) checkTxnSet(txns []types.Transaction, v2txns []types.V2Transac return allInPool, nil } -func (m *Manager) updateV2TransactionProofs(txns []types.V2Transaction, from, to types.ChainIndex) (updated []types.V2Transaction, err error) { +func (m *Manager) updateV2TransactionProofs(s storeReader, txns []types.V2Transaction, from, to types.ChainIndex) (updated []types.V2Transaction, err error) { // first validate the transaction set against its claimed basis; attempting // to update an invalid proof can cause a panic - basisState, ok := m.store.State(from.ID) + basisState, ok := s.State(from.ID) if !ok { return nil, fmt.Errorf("couldn't find state for basis %v", from) } @@ -1218,7 +1289,7 @@ func (m *Manager) updateV2TransactionProofs(txns []types.V2Transaction, from, to } } - revert, apply, err := m.reorgPath(from, to, 144) + revert, apply, err := m.reorgPath(s, from, to, 144) if err != nil { return nil, fmt.Errorf("couldn't determine reorg path from %v to %v: %w", from, to, err) } @@ -1229,7 +1300,7 @@ func (m *Manager) updateV2TransactionProofs(txns []types.V2Transaction, from, to updated = append(updated, txn.DeepCopy()) } for _, index := range revert { - b, bs, cs, ok := blockAndParent(m.store, index.ID) + b, bs, cs, ok := blockAndParent(s, index.ID) if !ok { return nil, fmt.Errorf("missing reverted block at index %v", index) } else if bs == nil { @@ -1246,7 +1317,7 @@ func (m *Manager) updateV2TransactionProofs(txns []types.V2Transaction, from, to } for _, index := range apply { - b, bs, cs, ok := blockAndParent(m.store, index.ID) + b, bs, cs, ok := blockAndParent(s, index.ID) if !ok { return nil, fmt.Errorf("missing applied block at index %v", index) } else if bs == nil { @@ -1254,7 +1325,7 @@ func (m *Manager) updateV2TransactionProofs(txns []types.V2Transaction, from, to } else if err := m.overwriteExpirations(b, bs); err != nil { return nil, fmt.Errorf("failed to overwrite expirations for block %v: %w", index, err) } - ancestorTimestamp, _ := m.store.AncestorTimestamp(b.ParentID) + ancestorTimestamp, _ := s.AncestorTimestamp(b.ParentID) cs, cau := consensus.ApplyBlock(cs, b, *bs, ancestorTimestamp) // get the transactions that were confirmed in this block @@ -1375,9 +1446,11 @@ func (m *Manager) UpdateV2TransactionSet(txns []types.V2Transaction, from, to ty if from == to { return txns, nil } - m.mu.Lock() - defer m.mu.Unlock() - return m.updateV2TransactionProofs(txns, from, to) + m.mu.RLock() + defer m.mu.RUnlock() + snap := m.store.Snapshot() + defer snap.Close() + return m.updateV2TransactionProofs(snap, txns, from, to) } // AddV2PoolTransactions validates a transaction set and adds it to the txpool. @@ -1404,7 +1477,7 @@ func (m *Manager) AddV2PoolTransactions(basis types.ChainIndex, txns []types.V2T for i := range txns { txns[i] = txns[i].DeepCopy() } - txns, err := m.updateV2TransactionProofs(txns, basis, m.tipState.Index) + txns, err := m.updateV2TransactionProofs(m.store, txns, basis, m.tipState.Index) if err != nil { return false, fmt.Errorf("failed to update set basis: %w", err) } diff --git a/chain/manager_test.go b/chain/manager_test.go index 691e4964..8654818a 100644 --- a/chain/manager_test.go +++ b/chain/manager_test.go @@ -615,7 +615,7 @@ func TestReorgPathMaxLen(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - _, _, err := cm.reorgPath(genesisIdx, tip, tc.maxLen) + _, _, err := cm.reorgPath(cm.store, genesisIdx, tip, tc.maxLen) if tc.shouldErr { if err == nil { t.Fatalf("expected error, got nil") @@ -660,7 +660,7 @@ func TestReorgPathBogusBasisBailsFast(t *testing.T) { bogus := types.ChainIndex{Height: math.MaxUint64, ID: tip.ID} const maxLen = 144 - revert, apply, err := cm.reorgPath(bogus, tip, maxLen) + revert, apply, err := cm.reorgPath(cm.store, bogus, tip, maxLen) if err == nil { t.Fatal("expected error for bogus basis, got nil") } diff --git a/db.go b/db.go index ff0eb359..518b2d10 100644 --- a/db.go +++ b/db.go @@ -78,6 +78,35 @@ func (db *BoltChainDB) Cancel() { db.tx = nil } +// Snapshot implements chain.DB. It begins a read-only bbolt transaction, which +// observes a consistent snapshot and can execute concurrently with other +// snapshots and with the open writer transaction. +func (db *BoltChainDB) Snapshot() chain.ReadonlyDB { + tx, err := db.db.Begin(false) + if err != nil { + panic(err) + } + return &boltSnapshot{tx: tx} +} + +// boltSnapshot is a read-only chain.ReadonlyDB backed by a bbolt read +// transaction. +type boltSnapshot struct { + tx *bbolt.Tx +} + +// Bucket implements chain.ReadonlyDB. +func (s *boltSnapshot) Bucket(name []byte) chain.ReadonlyDBBucket { + b := s.tx.Bucket(name) + if b == nil { + return nil + } + return boltBucket{b} +} + +// Close implements chain.ReadonlyDB, ending the read transaction. +func (s *boltSnapshot) Close() error { return s.tx.Rollback() } + // Close closes the BoltDB database. func (db *BoltChainDB) Close() error { db.Flush()