Merge pull request #115 from ethpandaops/pk910/fix-fork-detection-per…

…formance

improve fork detection performance

db/forks.go

@@ -124,3 +124,16 @@ func UpdateFinalizedForkParents(finalizedRoots [][]byte, tx *sqlx.Tx) error {
 
 	return nil
 }
+
+func UpdateForkParent(parentRoot []byte, parentForkId uint64, tx *sqlx.Tx) error {
+	_, err := tx.Exec(`
+		UPDATE forks 
+		SET parent_fork = $1 
+		WHERE base_root = $2
+	`, parentForkId, parentRoot)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

indexer/beacon/blockcache.go

@@ -18,15 +18,17 @@ type blockCache struct {
 	lowestSlot  int64
 	slotMap     map[phase0.Slot][]*Block
 	rootMap     map[phase0.Root]*Block
+	parentMap   map[phase0.Root][]*Block
 	latestBlock *Block // latest added block (might not be the head block, just a marker for cache changes)
 }
 
 // newBlockCache creates a new instance of blockCache.
 func newBlockCache(indexer *Indexer) *blockCache {
 	return &blockCache{
-		indexer: indexer,
-		slotMap: map[phase0.Slot][]*Block{},
-		rootMap: map[phase0.Root]*Block{},
+		indexer:   indexer,
+		slotMap:   map[phase0.Slot][]*Block{},
+		rootMap:   map[phase0.Root]*Block{},
+		parentMap: map[phase0.Root][]*Block{},
 	}
 }
 
@@ -60,6 +62,25 @@ func (cache *blockCache) createOrGetBlock(root phase0.Root, slot phase0.Slot) (*
 	return cacheBlock, true
 }
 
+// addBlockToParentMap adds the given block to the parent map.
+func (cache *blockCache) addBlockToParentMap(block *Block) {
+	cache.cacheMutex.Lock()
+	defer cache.cacheMutex.Unlock()
+
+	parentRoot := block.GetParentRoot()
+	if parentRoot == nil {
+		return
+	}
+
+	for _, parentBlock := range cache.parentMap[*parentRoot] {
+		if parentBlock == block {
+			return
+		}
+	}
+
+	cache.parentMap[*parentRoot] = append(cache.parentMap[*parentRoot], block)
+}
+
 // getBlockByRoot returns the cached block with the given root.
 func (cache *blockCache) getBlockByRoot(root phase0.Root) *Block {
 	cache.cacheMutex.RLock()
@@ -86,27 +107,13 @@ func (cache *blockCache) getBlocksByParentRoot(parentRoot phase0.Root) []*Block
 	cache.cacheMutex.RLock()
 	defer cache.cacheMutex.RUnlock()
 
-	parentBlock := cache.rootMap[parentRoot]
-
-	resBlocks := []*Block{}
-	for slot, blocks := range cache.slotMap {
-		if parentBlock != nil && slot <= parentBlock.Slot {
-			continue
-		}
-
-		for _, block := range blocks {
-			blockParentRoot := block.GetParentRoot()
-			if blockParentRoot == nil {
-				continue
-			}
-
-			if bytes.Equal((*blockParentRoot)[:], parentRoot[:]) {
-				resBlocks = append(resBlocks, block)
-			}
-		}
+	cachedBlocks := cache.parentMap[parentRoot]
+	blocks := make([]*Block, len(cachedBlocks))
+	if len(blocks) > 0 {
+		copy(blocks, cachedBlocks)
 	}
 
-	return resBlocks
+	return blocks
 }
 
 // getBlockByStateRoot returns the block with the given state root.
@@ -265,8 +272,10 @@ func (cache *blockCache) removeBlock(block *Block) {
 	cache.cacheMutex.Lock()
 	defer cache.cacheMutex.Unlock()
 
+	// remove the block from the root map.
 	delete(cache.rootMap, block.Root)
 
+	// remove the block from the slot map.
 	slotBlocks := cache.slotMap[block.Slot]
 	if len(slotBlocks) == 1 && slotBlocks[0] == block {
 		delete(cache.slotMap, block.Slot)
@@ -278,6 +287,22 @@ func (cache *blockCache) removeBlock(block *Block) {
 			}
 		}
 	}
+
+	// remove the block from the parent map.
+	if parentRoot := block.GetParentRoot(); parentRoot != nil {
+		parentBlocks := cache.parentMap[*parentRoot]
+		if len(parentBlocks) == 1 && parentBlocks[0] == block {
+			delete(cache.parentMap, *parentRoot)
+		} else if len(parentBlocks) > 1 {
+			for i, parentBlock := range parentBlocks {
+				if parentBlock == block {
+					cache.parentMap[*parentRoot] = append(parentBlocks[:i], parentBlocks[i+1:]...)
+					break
+				}
+			}
+		}
+	}
+
 }
 
 // getEpochBlocks returns the blocks that belong to the specified epoch.

indexer/beacon/client.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"runtime/debug"
+	"strings"
 	"time"
 
 	v1 "github.com/attestantio/go-eth2-client/api/v1"
@@ -105,6 +106,25 @@ func (c *Client) startClientLoop() {
 	}
 }
 
+func (c *Client) formatProcessingTimes(processingTimes []time.Duration) string {
+	if len(processingTimes) == 0 {
+		return ""
+	}
+
+	str := strings.Builder{}
+	str.WriteString(" (")
+	for i, pt := range processingTimes {
+		if i > 0 {
+			str.WriteString(", ")
+		}
+
+		str.WriteString(fmt.Sprintf("%v ms", pt.Milliseconds()))
+	}
+	str.WriteString(")")
+
+	return str.String()
+}
+
 // runClientLoop runs the client event processing subroutine.
 func (c *Client) runClientLoop() error {
 	// 1 - load & process head block
@@ -116,7 +136,7 @@ func (c *Client) runClientLoop() error {
 
 	c.headRoot = headRoot
 
-	headBlock, isNew, err := c.processBlock(headSlot, headRoot, nil)
+	headBlock, isNew, processingTimes, err := c.processBlock(headSlot, headRoot, nil)
 	if err != nil {
 		return fmt.Errorf("failed processing head block: %v", err)
 	}
@@ -126,9 +146,9 @@ func (c *Client) runClientLoop() error {
 	}
 
 	if isNew {
-		c.logger.Infof("received block %v:%v [0x%x] head", c.client.GetPool().GetChainState().EpochOfSlot(headSlot), headSlot, headRoot)
+		c.logger.Infof("received block %v:%v [0x%x] head %v", c.client.GetPool().GetChainState().EpochOfSlot(headSlot), headSlot, headRoot, c.formatProcessingTimes(processingTimes))
 	} else {
-		c.logger.Debugf("received known block %v:%v [0x%x] head", c.client.GetPool().GetChainState().EpochOfSlot(headSlot), headSlot, headRoot)
+		c.logger.Debugf("received known block %v:%v [0x%x] head %v", c.client.GetPool().GetChainState().EpochOfSlot(headSlot), headSlot, headRoot, c.formatProcessingTimes(processingTimes))
 	}
 
 	// 2 - backfill old blocks up to the finalization checkpoint or known in cache
@@ -263,17 +283,17 @@ func (c *Client) processHeadEvent(headEvent *v1.HeadEvent) error {
 
 // processStreamBlock processes a block received from the stream (either via block or head events).
 func (c *Client) processStreamBlock(slot phase0.Slot, root phase0.Root) (*Block, error) {
-	block, isNew, err := c.processBlock(slot, root, nil)
+	block, isNew, processingTimes, err := c.processBlock(slot, root, nil)
 	if err != nil {
 		return nil, err
 	}
 
 	chainState := c.client.GetPool().GetChainState()
 
 	if isNew {
-		c.logger.Infof("received block %v:%v [0x%x] stream", chainState.EpochOfSlot(block.Slot), block.Slot, block.Root[:])
+		c.logger.Infof("received block %v:%v [0x%x] stream %v", chainState.EpochOfSlot(block.Slot), block.Slot, block.Root[:], c.formatProcessingTimes(processingTimes))
 	} else {
-		c.logger.Debugf("received known block %v:%v [0x%x] stream", chainState.EpochOfSlot(block.Slot), block.Slot, block.Root[:])
+		c.logger.Debugf("received known block %v:%v [0x%x] stream %v", chainState.EpochOfSlot(block.Slot), block.Slot, block.Root[:], c.formatProcessingTimes(processingTimes))
 	}
 
 	return block, nil
@@ -323,9 +343,10 @@ func (c *Client) processReorg(oldHead *Block, newHead *Block) error {
 }
 
 // processBlock processes a block (from stream & polling).
-func (c *Client) processBlock(slot phase0.Slot, root phase0.Root, header *phase0.SignedBeaconBlockHeader) (block *Block, isNew bool, err error) {
+func (c *Client) processBlock(slot phase0.Slot, root phase0.Root, header *phase0.SignedBeaconBlockHeader) (block *Block, isNew bool, processingTimes []time.Duration, err error) {
 	chainState := c.client.GetPool().GetChainState()
 	finalizedSlot := chainState.GetFinalizedSlot()
+	processingTimes = make([]time.Duration, 3)
 
 	if slot < finalizedSlot {
 		// block is in finalized epoch
@@ -349,20 +370,34 @@ func (c *Client) processBlock(slot phase0.Slot, root phase0.Root, header *phase0
 			return header, nil
 		}
 
+		t1 := time.Now()
+		defer func() {
+			processingTimes[0] += time.Since(t1)
+		}()
+
 		return LoadBeaconHeader(c.getContext(), c, root)
 	})
 	if err != nil {
 		return
 	}
 
 	isNew, err = block.EnsureBlock(func() (*spec.VersionedSignedBeaconBlock, error) {
+
+		t1 := time.Now()
+		defer func() {
+			processingTimes[0] += time.Since(t1)
+		}()
+
 		return LoadBeaconBlock(c.getContext(), c, root)
 	})
 	if err != nil {
 		return
 	}
 
 	if slot >= finalizedSlot && isNew {
+		c.indexer.blockCache.addBlockToParentMap(block)
+		t1 := time.Now()
+
 		// fork detection
 		err2 := c.indexer.forkCache.processBlock(block)
 		if err2 != nil {
@@ -376,6 +411,9 @@ func (c *Client) processBlock(slot phase0.Slot, root phase0.Root, header *phase0
 			return
 		}
 
+		processingTimes[1] = time.Since(t1)
+		t1 = time.Now()
+
 		// write to db
 		err = db.RunDBTransaction(func(tx *sqlx.Tx) error {
 			err := db.InsertUnfinalizedBlock(dbBlock, tx)
@@ -389,6 +427,8 @@ func (c *Client) processBlock(slot phase0.Slot, root phase0.Root, header *phase0
 			return
 		}
 
+		processingTimes[2] = time.Since(t1)
+
 		block.isInUnfinalizedDb = true
 		c.indexer.blockCache.latestBlock = block
 	}
@@ -444,19 +484,20 @@ func (c *Client) backfillParentBlocks(headBlock *Block) error {
 			break
 		}
 
+		var processingTimes []time.Duration
 		if parentBlock == nil {
 			var err error
 
-			parentBlock, isNewBlock, err = c.processBlock(parentSlot, parentRoot, parentHead)
+			parentBlock, isNewBlock, processingTimes, err = c.processBlock(parentSlot, parentRoot, parentHead)
 			if err != nil {
 				return fmt.Errorf("could not process block [0x%x]: %v", parentRoot, err)
 			}
 		}
 
 		if isNewBlock {
-			c.logger.Infof("received block %v:%v [0x%x] backfill", chainState.EpochOfSlot(parentSlot), parentSlot, parentRoot)
+			c.logger.Infof("received block %v:%v [0x%x] backfill %v", chainState.EpochOfSlot(parentSlot), parentSlot, parentRoot, c.formatProcessingTimes(processingTimes))
 		} else {
-			c.logger.Debugf("received known block %v:%v [0x%x] backfill", chainState.EpochOfSlot(parentSlot), parentSlot, parentRoot)
+			c.logger.Debugf("received known block %v:%v [0x%x] backfill %v", chainState.EpochOfSlot(parentSlot), parentSlot, parentRoot, c.formatProcessingTimes(processingTimes))
 		}
 
 		if parentSlot == 0 {

indexer/beacon/fork.go

@@ -21,11 +21,11 @@ type Fork struct {
 }
 
 // newFork creates a new Fork instance.
-func newFork(forkId ForkKey, baseBlock *Block, leafBlock *Block, parentFork ForkKey) *Fork {
+func newFork(forkId ForkKey, baseSlot phase0.Slot, baseRoot phase0.Root, leafBlock *Block, parentFork ForkKey) *Fork {
 	fork := &Fork{
 		forkId:     forkId,
-		baseSlot:   baseBlock.Slot,
-		baseRoot:   baseBlock.Root,
+		baseSlot:   baseSlot,
+		baseRoot:   baseRoot,
 		leafSlot:   leafBlock.Slot,
 		leafRoot:   leafBlock.Root,
 		parentFork: parentFork,