diff --git a/cmd/bltest/main.go b/cmd/bltest/main.go index 3a76809eeb..152abdc2de 100644 --- a/cmd/bltest/main.go +++ b/cmd/bltest/main.go @@ -128,7 +128,7 @@ loop: } else { log.Error("BlockByHash", "hash", head.ParentHash, "error", err) } - num := big.NewInt(2) + num := big.NewInt(10) num.Sub(head.Number, num) if block, err := client.BlockByNumber(ctx, num); err == nil { log.Info("BlockByNumber", "number", num, "block.Hash", block.Hash(), "block.Number", block.Number(), "len(block.Transactions)", len(block.Transactions())) diff --git a/ethclient/lightclient/chain.go b/ethclient/lightclient/chain.go index 690b5f1141..a0bceb39a2 100644 --- a/ethclient/lightclient/chain.go +++ b/ethclient/lightclient/chain.go @@ -22,6 +22,7 @@ import ( "errors" "math/big" "sync" + "time" "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/beacon/light" @@ -43,20 +44,24 @@ type canonicalChain struct { newHeadCb func(common.Hash) head, finality *btypes.ExecutionHeader - recent map[uint64]common.Hash // nil until initialized - recentTail uint64 // if recent != nil then recent hashes are available from recentTail to head + recent map[uint64]common.Hash // nil until initialized + recentTail uint64 // if recent != nil then recent hashes are available from recentTail to head + tailFetchCh chan struct{} finalized *lru.Cache[uint64, common.Hash] // finalized but not recent hashes requests *requestMap[uint64, common.Hash] // requested; neither recent nor finalized } func newCanonicalChain(headTracker *light.HeadTracker, blocksAndHeaders *blocksAndHeaders, newHeadCb func(common.Hash)) *canonicalChain { - return &canonicalChain{ + c := &canonicalChain{ headTracker: headTracker, blocksAndHeaders: blocksAndHeaders, newHeadCb: newHeadCb, finalized: lru.NewCache[uint64, common.Hash](10000), requests: newRequestMap[uint64, common.Hash](nil), + tailFetchCh: make(chan struct{}), } + go c.tailFetcher() + return c } // Process implements request.Module in order to get notified about new heads. @@ -75,6 +80,36 @@ func (c *canonicalChain) Process(requester request.Requester, events []request.E } } +func (c *canonicalChain) tailFetcher() { //TODO stop + for { + c.lock.Lock() + var ( + tailNum uint64 + tailHash common.Hash + ) + if c.recent != nil { + tailNum, tailHash = c.recentTail, c.recent[c.recentTail] + } + needTail := tailNum + for _, reqNum := range c.requests.allKeys() { + if reqNum < needTail { + needTail = reqNum + } + } + c.lock.Unlock() + if needTail < tailNum { //TODO check recentCanonicalLength + log.Debug("Fetching tail headers", "have", tailNum, "need", needTail) + ctx, _ := context.WithTimeout(context.Background(), time.Second*10) + //TODO parallel fetch by number + if header, err := c.blocksAndHeaders.getHeader(ctx, tailHash); err == nil { + c.addRecentTail(header) + } + } else { + <-c.tailFetchCh + } + } +} + func (c *canonicalChain) getHash(ctx context.Context, number uint64) (common.Hash, error) { c.lock.Lock() if hash, ok := c.recent[number]; ok { @@ -87,6 +122,11 @@ func (c *canonicalChain) getHash(ctx context.Context, number uint64) (common.Has } req := c.requests.request(number) c.lock.Unlock() + select { + case c.tailFetchCh <- struct{}{}: + default: + } + defer req.release() return req.getResult(ctx) } @@ -98,7 +138,7 @@ func (c *canonicalChain) setHead(head *btypes.ExecutionHeader) bool { if c.head != nil && c.head.BlockHash() == headHash { return false } - if c.recent == nil || c.head == nil || c.head.BlockNumber()+1 != headNum || headHash != head.ParentHash() { + if c.recent == nil || c.head == nil || c.head.BlockNumber()+1 != headNum || c.head.BlockHash() != head.ParentHash() { c.recent = make(map[uint64]common.Hash) if headNum > 0 { c.recent[headNum-1] = head.ParentHash() @@ -117,6 +157,7 @@ func (c *canonicalChain) setHead(head *btypes.ExecutionHeader) bool { c.recentTail++ } c.requests.tryDeliver(headNum, headHash) + log.Debug("SetHead", "recentTail", c.recentTail, "head", headNum) return true } @@ -186,12 +227,12 @@ func (c *canonicalChain) resolveBlockNumber(number *big.Int) (uint64, *btypes.Ex } func (c *canonicalChain) blockNumberToHash(ctx context.Context, number *big.Int) (common.Hash, error) { - num, header, err := c.resolveBlockNumber(number) + num, pheader, err := c.resolveBlockNumber(number) if err != nil { return common.Hash{}, err } - if header != nil { - return header.BlockHash(), nil + if pheader != nil { + return pheader.BlockHash(), nil } return c.getHash(ctx, num) } diff --git a/ethclient/lightclient/request_map.go b/ethclient/lightclient/request_map.go index fa8ea58d9d..75335e1816 100644 --- a/ethclient/lightclient/request_map.go +++ b/ethclient/lightclient/request_map.go @@ -52,6 +52,7 @@ func (rm *requestMap[K, V]) request(key K) *mappedRequest[K, V] { deliveredCh: make(chan struct{}), cancelFn: cancelFn, } + rm.requests[key] = r if rm.requestFn != nil { go func() { result, err := rm.requestFn(ctx, key) @@ -69,6 +70,17 @@ func (rm *requestMap[K, V]) has(key K) bool { return ok } +func (rm *requestMap[K, V]) allKeys() []K { + rm.lock.Lock() + defer rm.lock.Unlock() + + keys := make([]K, 0, len(rm.requests)) + for key := range rm.requests { + keys = append(keys, key) + } + return keys +} + // should only be called with validated results of successful requests func (rm *requestMap[K, V]) tryDeliver(key K, result V) { rm.lock.Lock()