[WIP] Implement a UTXO cache

blockchain/blockindex.go

@@ -217,6 +217,47 @@ func (node *blockNode) CalcPastMedianTime() time.Time {
 	return time.Unix(medianTimestamp, 0)
 }
 
+// findHighestCommonAncestor searches for the highest common node between the
+// ancestors of the two given nodes.
+func findHighestCommonAncestor(node1, node2 *blockNode) *blockNode {
+	// Since the common ancestor cannot be higher than the lowest node, put
+	// the higher one to it's ancestor at the lower one's height.
+	if node1.height > node2.height {
+		node1 = node1.Ancestor(node2.height)
+	} else {
+		node2 = node2.Ancestor(node1.height)
+	}
+
+	// The search strategy used is to exponentially look back until the ancestor
+	// matches (or they are nil, in which case we reached the genesis block.
+	// Then, we go back from the last height that was not common to with linear
+	// steps until we find the first ancestor.
+
+	distance := int32(1)
+	for {
+		new1 := node1.RelativeAncestor(distance)
+		new2 := node2.RelativeAncestor(distance)
+		distance *= 2
+
+		if new1 == nil || new2 == nil || new1 == new2 {
+			break
+		}
+
+		node1, node2 = new1, new2
+	}
+
+	for node1 != node2 {
+		node1 = node1.parent
+		node2 = node2.parent
+
+		if node1 == nil || node2 == nil {
+			return nil
+		}
+	}
+
+	return node1
+}
+
 // blockIndex provides facilities for keeping track of an in-memory index of the
 // block chain.  Although the name block chain suggests a single chain of
 // blocks, it is actually a tree-shaped structure where any node can have

blockchain/chain.go

@@ -85,6 +85,26 @@ func newBestState(node *blockNode, blockSize, blockWeight, numTxns,
 	}
 }
 
+// FlushMode is used to indicate the different urgency types for a flush.
+type FlushMode uint8
+
+const (
+	// FlushRequired is the flush mode that means a flush must be performed
+	// regardless of the cache state.  For example right before shutting down.
+	FlushRequired FlushMode = iota
+
+	// FlushPeriodic is the flush mode that means a flush can be performed
+	// when it would be almost needed.  This is used to periodically signal when
+	// no I/O heavy operations are expected soon, so there is time to flush.
+	FlushPeriodic
+
+	// FlushIfNeeded is the flush mode that means a flush must be performed only
+	// if the cache is exceeding a safety threshold very close to its maximum
+	// size.  This is used mostly internally in between operations that can
+	// increase the cache size.
+	FlushIfNeeded
+)
+
 // BlockChain provides functions for working with the bitcoin block chain.
 // It includes functionality such as rejecting duplicate blocks, ensuring blocks
 // follow all rules, orphan handling, checkpoint handling, and best chain
@@ -126,6 +146,12 @@ type BlockChain struct {
 	index     *blockIndex
 	bestChain *chainView
 
+	// The UTXO state holds a cached view of the UTXO state of the chain.
+	//
+	// It has its own lock, however it is often also protected by the chain lock
+	// to help prevent logic races when blocks are being processed.
+	utxoCache *utxoCache
+
 	// These fields are related to handling of orphan blocks.  They are
 	// protected by a combination of the chain lock and the orphan lock.
 	orphanLock   sync.RWMutex
@@ -622,14 +648,6 @@ func (b *BlockChain) connectBlock(node *blockNode, block *btcutil.Block,
 			return err
 		}
 
-		// Update the utxo set using the state of the utxo view.  This
-		// entails removing all of the utxos spent and adding the new
-		// ones created by the block.
-		err = dbPutUtxoView(dbTx, view)
-		if err != nil {
-			return err
-		}
-
 		// Update the transaction spend journal by adding a record for
 		// the block that contains all txos spent by it.
 		err = dbPutSpendJournalEntry(dbTx, block.Hash(), stxos)
@@ -653,9 +671,13 @@ func (b *BlockChain) connectBlock(node *blockNode, block *btcutil.Block,
 		return err
 	}
 
-	// Prune fully spent entries and mark all entries in the view unmodified
-	// now that the modifications have been committed to the database.
-	view.commit()
+	// Commit all modifications made to the view into the utxo state.  This also
+	// prunes these changes from the view.
+	b.stateLock.Lock()
+	if err := b.utxoCache.Commit(view); err != nil {
+		log.Errorf("error committing block %s(%d) to utxo cache: %s", block.Hash(), block.Height(), err.Error())
+	}
+	b.stateLock.Unlock()
 
 	// This node is now the end of the best chain.
 	b.bestChain.SetTip(node)
@@ -676,7 +698,11 @@ func (b *BlockChain) connectBlock(node *blockNode, block *btcutil.Block,
 	b.sendNotification(NTBlockConnected, block)
 	b.chainLock.Lock()
 
-	return nil
+	// Since we just changed the UTXO cache, we make sure it didn't exceed its
+	// maximum size.
+	b.stateLock.Lock()
+	defer b.stateLock.Unlock()
+	return b.utxoCache.Flush(FlushIfNeeded, state)
 }
 
 // disconnectBlock handles disconnecting the passed node/block from the end of
@@ -734,14 +760,6 @@ func (b *BlockChain) disconnectBlock(node *blockNode, block *btcutil.Block, view
 			return err
 		}
 
-		// Update the utxo set using the state of the utxo view.  This
-		// entails restoring all of the utxos spent and removing the new
-		// ones created by the block.
-		err = dbPutUtxoView(dbTx, view)
-		if err != nil {
-			return err
-		}
-
 		// Before we delete the spend journal entry for this back,
 		// we'll fetch it as is so the indexers can utilize if needed.
 		stxos, err := dbFetchSpendJournalEntry(dbTx, block)
@@ -772,9 +790,11 @@ func (b *BlockChain) disconnectBlock(node *blockNode, block *btcutil.Block, view
 		return err
 	}
 
-	// Prune fully spent entries and mark all entries in the view unmodified
-	// now that the modifications have been committed to the database.
-	view.commit()
+	// Commit all modifications made to the view into the utxo state.  This also
+	// prunes these changes from the view.
+	b.stateLock.Lock()
+	b.utxoCache.Commit(view)
+	b.stateLock.Unlock()
 
 	// This node's parent is now the end of the best chain.
 	b.bestChain.SetTip(node.parent)
@@ -795,7 +815,11 @@ func (b *BlockChain) disconnectBlock(node *blockNode, block *btcutil.Block, view
 	b.sendNotification(NTBlockDisconnected, block)
 	b.chainLock.Lock()
 
-	return nil
+	// Since we just changed the UTXO cache, we make sure it didn't exceed its
+	// maximum size.
+	b.stateLock.Lock()
+	defer b.stateLock.Unlock()
+	return b.utxoCache.Flush(FlushIfNeeded, state)
 }
 
 // countSpentOutputs returns the number of utxos the passed block spends.
@@ -825,6 +849,10 @@ func (b *BlockChain) reorganizeChain(detachNodes, attachNodes *list.List) error
 		return nil
 	}
 
+	// The rest of the reorg depends on all STXOs already being in the database
+	// so we flush before reorg
+	b.utxoCache.Flush(FlushRequired, b.BestSnapshot())
+
 	// Ensure the provided nodes match the current best chain.
 	tip := b.bestChain.Tip()
 	if detachNodes.Len() != 0 {
@@ -866,7 +894,6 @@ func (b *BlockChain) reorganizeChain(detachNodes, attachNodes *list.List) error
 	// database and using that information to unspend all of the spent txos
 	// and remove the utxos created by the blocks.
 	view := NewUtxoViewpoint()
-	view.SetBestHash(&oldBest.hash)
 	for e := detachNodes.Front(); e != nil; e = e.Next() {
 		n := e.Value.(*blockNode)
 		var block *btcutil.Block
@@ -886,7 +913,7 @@ func (b *BlockChain) reorganizeChain(detachNodes, attachNodes *list.List) error
 
 		// Load all of the utxos referenced by the block that aren't
 		// already in the view.
-		err = view.fetchInputUtxos(b.db, block)
+		err = view.addInputUtxos(b.utxoCache, block)
 		if err != nil {
 			return err
 		}
@@ -906,7 +933,7 @@ func (b *BlockChain) reorganizeChain(detachNodes, attachNodes *list.List) error
 		detachBlocks = append(detachBlocks, block)
 		detachSpentTxOuts = append(detachSpentTxOuts, stxos)
 
-		err = view.disconnectTransactions(b.db, block, stxos)
+		err = disconnectTransactions(view, block, stxos, b.utxoCache)
 		if err != nil {
 			return err
 		}
@@ -953,11 +980,11 @@ func (b *BlockChain) reorganizeChain(detachNodes, attachNodes *list.List) error
 		// checkConnectBlock gets skipped, we still need to update the UTXO
 		// view.
 		if b.index.NodeStatus(n).KnownValid() {
-			err = view.fetchInputUtxos(b.db, block)
+			err = view.addInputUtxos(b.utxoCache, block)
 			if err != nil {
 				return err
 			}
-			err = view.connectTransactions(block, nil)
+			err = connectTransactions(view, block, nil, true)
 			if err != nil {
 				return err
 			}
@@ -996,7 +1023,6 @@ func (b *BlockChain) reorganizeChain(detachNodes, attachNodes *list.List) error
 	// view to be valid from the viewpoint of each block being connected or
 	// disconnected.
 	view = NewUtxoViewpoint()
-	view.SetBestHash(&b.bestChain.Tip().hash)
 
 	// Disconnect blocks from the main chain.
 	for i, e := 0, detachNodes.Front(); e != nil; i, e = i+1, e.Next() {
@@ -1005,15 +1031,15 @@ func (b *BlockChain) reorganizeChain(detachNodes, attachNodes *list.List) error
 
 		// Load all of the utxos referenced by the block that aren't
 		// already in the view.
-		err := view.fetchInputUtxos(b.db, block)
+		err := view.addInputUtxos(b.utxoCache, block)
 		if err != nil {
 			return err
 		}
 
 		// Update the view to unspend all of the spent txos and remove
 		// the utxos created by the block.
-		err = view.disconnectTransactions(b.db, block,
-			detachSpentTxOuts[i])
+		err = disconnectTransactions(view, block, detachSpentTxOuts[i],
+			b.utxoCache)
 		if err != nil {
 			return err
 		}
@@ -1032,7 +1058,7 @@ func (b *BlockChain) reorganizeChain(detachNodes, attachNodes *list.List) error
 
 		// Load all of the utxos referenced by the block that aren't
 		// already in the view.
-		err := view.fetchInputUtxos(b.db, block)
+		err := view.addInputUtxos(b.utxoCache, block)
 		if err != nil {
 			return err
 		}
@@ -1042,7 +1068,7 @@ func (b *BlockChain) reorganizeChain(detachNodes, attachNodes *list.List) error
 		// to it.  Also, provide an stxo slice so the spent txout
 		// details are generated.
 		stxos := make([]SpentTxOut, 0, countSpentOutputs(block))
-		err = view.connectTransactions(block, &stxos)
+		err = connectTransactions(view, block, &stxos, false)
 		if err != nil {
 			return err
 		}
@@ -1107,7 +1133,6 @@ func (b *BlockChain) connectBestChain(node *blockNode, block *btcutil.Block, fla
 		// to the main chain without violating any rules and without
 		// actually connecting the block.
 		view := NewUtxoViewpoint()
-		view.SetBestHash(parentHash)
 		stxos := make([]SpentTxOut, 0, countSpentOutputs(block))
 		if !fastAdd {
 			err := b.checkConnectBlock(node, block, view, &stxos)
@@ -1131,11 +1156,11 @@ func (b *BlockChain) connectBestChain(node *blockNode, block *btcutil.Block, fla
 		// utxos, spend them, and add the new utxos being created by
 		// this block.
 		if fastAdd {
-			err := view.fetchInputUtxos(b.db, block)
+			err := view.addInputUtxos(b.utxoCache, block)
 			if err != nil {
 				return false, err
 			}
-			err = view.connectTransactions(block, &stxos)
+			err = connectTransactions(view, block, &stxos, false)
 			if err != nil {
 				return false, err
 			}
@@ -1657,6 +1682,11 @@ type Config struct {
 	// This field is required.
 	DB database.DB
 
+	// The maximum size in bytes of the UTXO cache.
+	//
+	// This field is required.
+	UtxoCacheMaxSize uint64
+
 	// Interrupt specifies a channel the caller can close to signal that
 	// long running operations, such as catching up indexes or performing
 	// database migrations, should be interrupted.
@@ -1760,6 +1790,7 @@ func New(config *Config) (*BlockChain, error) {
 		maxRetargetTimespan: targetTimespan * adjustmentFactor,
 		blocksPerRetarget:   int32(targetTimespan / targetTimePerBlock),
 		index:               newBlockIndex(config.DB, params),
+		utxoCache:           newUtxoCache(config.DB, config.UtxoCacheMaxSize),
 		hashCache:           config.HashCache,
 		bestChain:           newChainView(nil),
 		orphans:             make(map[chainhash.Hash]*orphanBlock),
@@ -1780,6 +1811,13 @@ func New(config *Config) (*BlockChain, error) {
 		return nil, err
 	}
 
+	// Make sure the utxo state is catched up if it was left in an inconsistent
+	// state.
+	bestNode := b.bestChain.Tip()
+	if err := b.utxoCache.InitConsistentState(bestNode, config.Interrupt); err != nil {
+		return nil, err
+	}
+
 	// Initialize and catch up all of the currently active optional indexes
 	// as needed.
 	if config.IndexManager != nil {
@@ -1794,10 +1832,26 @@ func New(config *Config) (*BlockChain, error) {
 		return nil, err
 	}
 
-	bestNode := b.bestChain.Tip()
+	bestNode = b.bestChain.Tip()
 	log.Infof("Chain state (height %d, hash %v, totaltx %d, work %v)",
 		bestNode.height, bestNode.hash, b.stateSnapshot.TotalTxns,
 		bestNode.workSum)
 
 	return &b, nil
 }
+
+// CachedStateSize returns the total size of the cached state of the blockchain
+// in bytes.
+func (b *BlockChain) CachedStateSize() uint64 {
+	return b.utxoCache.TotalMemoryUsage()
+}
+
+// FlushCachedState flushes all the cached state of the blockchain to the
+// database.
+//
+// This method is safe for concurrent access.
+func (b *BlockChain) FlushCachedState(mode FlushMode) error {
+	b.chainLock.Lock()
+	defer b.chainLock.Unlock()
+	return b.utxoCache.Flush(mode, b.stateSnapshot)
+}

blockchain/chain_test.go

@@ -147,7 +147,6 @@ func TestCalcSequenceLock(t *testing.T) {
 	})
 	utxoView := NewUtxoViewpoint()
 	utxoView.AddTxOuts(targetTx, int32(numBlocksToActivate)-4)
-	utxoView.SetBestHash(&node.hash)
 
 	// Create a utxo that spends the fake utxo created above for use in the
 	// transactions created in the tests.  It has an age of 4 blocks.  Note