forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

beacon/light: added more comments and func/struct descriptions

Browse source
This commit is contained in:
Zsolt Felfoldi 2024-01-10 07:03:33 +01:00 committed by Felix Lange
parent 26813d2a6c
commit 7dd8190f63
10 changed files with 173 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
beacon/light/api/api_server.go
View file

@ -26,6 +26,7 @@ import (
"github.com/ethereum/go-ethereum/log"
)
// ApiServer is a wrapper around BeaconLightApi that implements request.requestServer.
type ApiServer struct {
api *BeaconLightApi
eventCallback func(event request.Event)
@ -33,10 +34,12 @@ type ApiServer struct {
lastId uint64
}
// NewApiServer creates a new ApiServer.
func NewApiServer(api *BeaconLightApi) *ApiServer {
return &ApiServer{api: api}
}
// Subscribe implements request.requestServer.
func (s *ApiServer) Subscribe(eventCallback func(event request.Event)) {
s.eventCallback = eventCallback
s.unsubscribe = s.api.StartHeadListener(func(slot uint64, blockRoot common.Hash) {
@ -50,6 +53,7 @@ func (s *ApiServer) Subscribe(eventCallback func(event request.Event)) {
})
}
// SendRequest implements request.requestServer.
func (s *ApiServer) SendRequest(req request.Request) request.ID {
id := request.ID(atomic.AddUint64(&s.lastId, 1))
go func() {
@ -86,7 +90,8 @@ func (s *ApiServer) SendRequest(req request.Request) request.ID {
return id
}
// Note: UnsubscribeHeads should not be called concurrently with SubscribeHeads
// Unsubscribe implements request.requestServer.
// Note: Unsubscribe should not be called concurrently with Subscribe.
func (s *ApiServer) Unsubscribe() {
if s.unsubscribe != nil {
s.unsubscribe()

17
beacon/light/head_tracker.go
View file

@ -25,6 +25,9 @@ import (
"github.com/ethereum/go-ethereum/log"
)
// HeadTracker keeps track of the latest validated head and the "prefetch" head
// which is the (not necessarily validated) head announced by the majority of
// servers.
type HeadTracker struct {
lock sync.RWMutex
committeeChain *CommitteeChain
@ -34,6 +37,7 @@ type HeadTracker struct {
prefetchHead types.HeadInfo
}
// NewHeadTracker creates a new HeadTracker.
func NewHeadTracker(committeeChain *CommitteeChain, minSignerCount int) *HeadTracker {
return &HeadTracker{
committeeChain: committeeChain,
@ -41,6 +45,7 @@ func NewHeadTracker(committeeChain *CommitteeChain, minSignerCount int) *HeadTra
}
}
// ValidatedHead returns the latest validated head.
func (h *HeadTracker) ValidatedHead() types.SignedHeader {
h.lock.RLock()
defer h.lock.RUnlock()
@ -48,6 +53,10 @@ func (h *HeadTracker) ValidatedHead() types.SignedHeader {
return h.signedHead
}
// Validate validates the given signed head. If the head is successfully validated
// and it is better than the old validated head (higher slot or same slot and more
// signers) then ValidatedHead is updated. The boolean return flag signals if
// ValidatedHead has been changed.
func (h *HeadTracker) Validate(head types.SignedHeader) (bool, error) {
h.lock.Lock()
defer h.lock.Unlock()
@ -76,6 +85,11 @@ func (h *HeadTracker) Validate(head types.SignedHeader) (bool, error) {
return true, nil
}
// PrefetchHead returns the latest known prefetch head's head info.
// This head can be used to start fetching related data hoping that it will be
// validated soon.
// Note that the prefetch head cannot be validated cryptographically so it should
// only be used as a performance optimization hint.
func (h *HeadTracker) PrefetchHead() types.HeadInfo {
h.lock.RLock()
defer h.lock.RUnlock()
@ -83,6 +97,9 @@ func (h *HeadTracker) PrefetchHead() types.HeadInfo {
return h.prefetchHead
}
// SetPrefetchHead sets the prefetch head info.
// Note that HeadTracker does not verify the prefetch head, just acts as a thread
// safe bulletin board.
func (h *HeadTracker) SetPrefetchHead(head types.HeadInfo) {
h.lock.Lock()
defer h.lock.Unlock()

34
beacon/light/request/scheduler.go
View file

@ -31,19 +31,18 @@ import (
// with servers but may keep track of certain parameters of registered servers,
// based on the received server events. These server parameters may affect the
// possible range of requests to be sent to a given server.
// Modules are called by Scheduler whenever a global trigger is fired. All request
// and server events fire the trigger. Modules themselves can also self-trigger,
// ensuring an immediate next processing round after the target data structure has
// been changed in a way that could make further actions possible either by the
// same or another Module.
// Modules are called by Scheduler whenever a global trigger is fired. All events
// fire the trigger. Modules themselves can also self-trigger, ensuring an
// immediate next processing round after the target data structure has been
// changed in a way that could make further actions possible either by the same
// or another Module.
type Module interface {
// Process is a non-blocking function that is called on each Module whenever
// a processing round is triggered. It can start new requests through the
// received Tracker, process events related to servers and previosly sent
// requests and/or do other data processing tasks. Note that request events
// are only passed to the module that made the given request while server
// events are passed to every module. Process can also trigger a next
// processing round by returning true.
// received Tracker, process events and/or do other data processing tasks.
// Note that request events are only passed to the module that made the given
// request while server events are passed to every module. Process can also
// trigger a next processing round by returning true.
//
// Note: Process functions of different modules are never called concurrently;
// they are called by Scheduler in the same order of priority as they were
@ -54,7 +53,8 @@ type Module interface {
// Scheduler is a modular network data retrieval framework that coordinates multiple
// servers and retrieval mechanisms (modules). It implements a trigger mechanism
// that calls the Process function of registered modules whenever either the state
// of existing data structures or connected servers could allow new operations.
// of existing data structures or events coming from registered servers could
// allow new operations.
type Scheduler struct {
lock sync.Mutex
clock mclock.Clock
@ -71,8 +71,8 @@ type Scheduler struct {
// testTimerResults []bool // true is appended when simulated timer is processed; false when stopped
}
// pendingRequest keeps track of sent and not finalized requests and their sender
// modules and whether a soft timeout has already happened.
// pendingRequest keeps track of sent and not yet finalized requests and their
// sender modules.
type pendingRequest struct {
request Request
module Module
@ -185,7 +185,7 @@ func (s *Scheduler) syncLoop() {
}
}
// processModules runs an entire processing round, calling the process functions
// processModules runs an entire processing round, calling the Process functions
// of all modules, passing all relevant events.
func (s *Scheduler) processModules() {
s.lock.Lock()
@ -241,7 +241,7 @@ func (s *Scheduler) Trigger() {
}
}
// addRequestEvent adds a request event to the sender module's Tracker, ensuring
// addRequestEvent adds a request event to the sender module's tracker, ensuring
// that the module receives it in the next processing round.
func (s *Scheduler) addRequestEvent(event Event) {
sid, _, _ := event.RequestInfo()
@ -263,8 +263,8 @@ func (s *Scheduler) addServerEvent(event Event) {
// handleEvent processes an Event and adds it either as a request event or a
// server event, depending on its type. In case of an EvUnregistered server event
// it also closes all pending requests to the given server by emitting a failed
// request event (Finalized without Response), ensuring that all requests get
// finalized and thereby allowing the module logic to be safe and simple.
// request event (EvFail), ensuring that all requests get finalized and thereby
// allowing the module logic to be safe and simple.
func (s *Scheduler) handleEvent(event Event) {
s.Trigger()
if event.IsRequestEvent() {

77
beacon/light/request/server.go
View file

@ -28,9 +28,9 @@ import (
var (
// request events
EvResponse = &EventType{Name: "response", requestEvent: true} // data: RequestResponse
EvFail = &EventType{Name: "fail", requestEvent: true} // data: RequestResponse
EvTimeout = &EventType{Name: "timeout", requestEvent: true} // data: RequestResponse
EvResponse = &EventType{Name: "response", requestEvent: true} // data: RequestResponse; sent by requestServer
EvFail = &EventType{Name: "fail", requestEvent: true} // data: RequestResponse; sent by requestServer
EvTimeout = &EventType{Name: "timeout", requestEvent: true} // data: RequestResponse; sent by serverWithTimeout
// server events
EvRegistered = &EventType{Name: "registered"} // data: nil; sent by Scheduler
EvUnregistered = &EventType{Name: "unregistered"} // data: nil; sent by Scheduler
@ -43,26 +43,30 @@ const (
)
const (
parallelAdjustUp = 0.1
parallelAdjustDown = 1
minParallelLimit = 1
defaultParallelLimit = 3
minFailureDelay = time.Millisecond * 100
maxFailureDelay = time.Minute
// serverWithLimits parameters
parallelAdjustUp = 0.1 // adjust parallelLimit up in case of success under full load
parallelAdjustDown = 1 // adjust parallelLimit down in case of timeout/failure
minParallelLimit = 1 // parallelLimit lower bound
defaultParallelLimit = 3 // parallelLimit initial value
minFailureDelay = time.Millisecond * 100 // minimum disable time in case of request failure
maxFailureDelay = time.Minute // maximum disable time in case of request failure
)
// requestServer can send a set of requests pre-defined by the application and
// signal events through the event callback. After each request, it should send
// back either EvResponse or EvFail. Additionally, it may also send application-
// defined events that the Modules can interpret.
//
//TODO ?separate request and server events here?
// requestServer can send requests in a non-blocking way and feed back events
// through the event callback. After each request, it should send back either
// EvResponse or EvFail. Additionally, it may also send application-defined
// events that the Modules can interpret.
type requestServer interface {
Subscribe(eventCallback func(event Event))
SendRequest(request Request) ID
Unsubscribe()
}
// server is implemented by a requestServer wrapped into serverWithTimeout and
// serverWithLimits and is used by Scheduler.
// In addition to requestServer functionality, server can also handle timeouts,
// limit the number of parallel in-flight requests and temporarily disable
// new requests based on timeouts and response failures.
type server interface {
subscribe(eventCallback func(event Event))
canRequestNow() (bool, float32)
@ -71,6 +75,7 @@ type server interface {
unsubscribe()
}
// newServer wraps a requestServer and returns a server
func newServer(rs requestServer, clock mclock.Clock) server {
s := &serverWithLimits{}
s.parent = rs
@ -81,26 +86,36 @@ func newServer(rs requestServer, clock mclock.Clock) server {
type serverSet map[server]struct{}
// EventType identifies an event type, either related to a request or the server
// in general. Server events can also be externally defined.
type EventType struct {
Name string
requestEvent bool
requestEvent bool // all request events are pre-defined in request package
}
// Event describes an event where the type of Data depends on Type.
// Server field is not required when sent through the event callback; it is filled
// out when processed by the Scheduler. Note that the Scheduler can also create
// and send events (EvRegistered, EvUnregistered) directly.
type Event struct {
Type *EventType
Server Server // filled by Scheduler
Data any
}
// IsRequestEvent returns true if the event is a request event
func (e *Event) IsRequestEvent() bool {
return e.Type.requestEvent
}
// RequestInfo assumes that the event is a request event and returns its contents
// in a convenient form.
func (e *Event) RequestInfo() (ServerAndID, Request, Response) {
data := e.Data.(RequestResponse)
return ServerAndID{Server: e.Server, ID: data.ID}, data.Request, data.Response
}
// RequestResponse is the Data type of request events.
type RequestResponse struct {
ID ID
Request Request
@ -120,11 +135,14 @@ type serverWithTimeout struct {
timeouts map[ID]mclock.Timer
}
// init initializes serverWithTimeout
func (s *serverWithTimeout) init(clock mclock.Clock) {
s.clock = clock
s.timeouts = make(map[ID]mclock.Timer)
}
// subscribe subscribes to events which include parent (requestServer) events
// plus EvTimeout.
func (s *serverWithTimeout) subscribe(eventCallback func(event Event)) {
s.lock.Lock()
defer s.lock.Unlock()
@ -133,6 +151,7 @@ func (s *serverWithTimeout) subscribe(eventCallback func(event Event)) {
s.parent.Subscribe(s.eventCallback)
}
// eventCallback is called by parent (requestServer) event subscription.
func (s *serverWithTimeout) eventCallback(event Event) {
s.lock.Lock()
defer s.lock.Unlock()
@ -152,6 +171,8 @@ func (s *serverWithTimeout) eventCallback(event Event) {
}
}
// sendRequest sends a request through the parent (requestServer) and starts a
// timer for request timeout.
func (s *serverWithTimeout) sendRequest(request Request) (reqId ID) {
s.lock.Lock()
defer s.lock.Unlock()
@ -201,6 +222,7 @@ func (s *serverWithTimeout) unsubscribe() {
s.parent.Unsubscribe()
}
// stopTimer stops the given timer
func (s *serverWithTimeout) stopTimer(timer mclock.Timer) {
timer.Stop()
/*if timer.Stop() && s.scheduler.testTimerResults != nil {
@ -231,11 +253,14 @@ type serverWithLimits struct {
failureDelay float64
}
// init initializes serverWithLimits
func (s *serverWithLimits) init() {
s.softTimeouts = make(map[ID]struct{})
s.parallelLimit = defaultParallelLimit
}
// subscribe subscribes to events which include parent (serverWithTimeout) events
// plus EvCanRequstAgain.
func (s *serverWithLimits) subscribe(eventCallback func(event Event)) {
s.lock.Lock()
defer s.lock.Unlock()
@ -244,6 +269,7 @@ func (s *serverWithLimits) subscribe(eventCallback func(event Event)) {
s.serverWithTimeout.subscribe(s.eventCallback)
}
// eventCallback is called by parent (serverWithTimeout) event subscription.
func (s *serverWithLimits) eventCallback(event Event) {
s.lock.Lock()
var sendCanRequestAgain bool
@ -284,13 +310,13 @@ func (s *serverWithLimits) eventCallback(event Event) {
}
}
// sendRequest sends a request through the parent (serverWithTimeout).
func (s *serverWithLimits) sendRequest(request Request) (reqId ID) {
s.lock.Lock()
defer s.lock.Unlock()
s.pendingCount++
id := s.serverWithTimeout.sendRequest(request)
return id
return s.serverWithTimeout.sendRequest(request)
}
// stop stops all goroutines associated with the server.
@ -306,6 +332,7 @@ func (s *serverWithLimits) unsubscribe() {
s.serverWithTimeout.unsubscribe()
}
// canRequest checks whether a new request can be started.
func (s *serverWithLimits) canRequest() (bool, float32) {
if s.delayTimer != nil || s.pendingCount >= int(s.parallelLimit) {
return false, 0
@ -316,7 +343,14 @@ func (s *serverWithLimits) canRequest() (bool, float32) {
return true, -(float32(s.pendingCount) + rand.Float32()) / s.parallelLimit
}
// EvCanRequestAgain guaranteed if it returns false
// canRequestNow checks whether a new request can be started, according to the
// current in-flight request count and parallelLimit, and also the failure delay
// timer.
// If a new request is allowed then it also returns a priority value that can be
// used to select the least overloaded server from an otherwise equally suitable
// set of servers.
// If it returns false then it is guaranteed that an EvCanRequestAgain will be
// sent whenever the server becomes available for requesting again.
func (s *serverWithLimits) canRequestNow() (bool, float32) {
var sendCanRequestAgain bool
s.lock.Lock()
@ -333,6 +367,8 @@ func (s *serverWithLimits) canRequestNow() (bool, float32) {
return canRequest, priority
}
// delay sets the delay timer to the given duration, disabling new requests for
// the given period.
func (s *serverWithLimits) delay(delay time.Duration) {
if s.delayTimer != nil {
// Note: if stopping the timer is unsuccessful then the resulting AfterFunc
@ -366,6 +402,8 @@ func (s *serverWithLimits) delay(delay time.Duration) {
})
}
// fail reports that a response from the server was found invalid by the processing
// Module, disabling new requests for a dynamically adjused time period.
func (s *serverWithLimits) fail(desc string) {
s.lock.Lock()
defer s.lock.Unlock()
@ -373,6 +411,7 @@ func (s *serverWithLimits) fail(desc string) {
s.failLocked(desc)
}
// failLocked calculates the dynamic failure delay and applies it.
func (s *serverWithLimits) failLocked(desc string) {
log.Debug("Server error", "description", desc)
s.failureDelay *= 2

16
beacon/light/request/tracker.go
View file

@ -23,6 +23,11 @@ import (
)
type (
// Server identifies a server without allowing any direct interaction.
// Note: server interface is used by Scheduler and Tracker but not used by
// the modules that do not interact with them directly.
// In order to make module testing easier, Server interface is used in
// events and modules.
Server any
Request any
Response any
@ -60,14 +65,20 @@ type Tracker interface {
InvalidResponse(id ServerAndID, desc string)
}
// one per sync process
// tracker implements Tracker. A separate instance is created for each Module.
type tracker struct {
servers serverSet // one per trigger
// servers is a set of currently available servers; it is recreated at every
// processModule round.
servers serverSet
scheduler *Scheduler
module Module
// requestEvents is a list of events related to requests sent by the given
// module. It is reset before module processing and the previous contents are
// passed to Module.Process along with the globally collected server events.
requestEvents []Event
}
// TryRequest implements Tracker.
func (p *tracker) TryRequest(requestFn func(server Server) (Request, float32)) (RequestWithID, bool) {
var (
maxServerPriority, maxRequestPriority float32
@ -104,6 +115,7 @@ func (p *tracker) TryRequest(requestFn func(server Server) (Request, float32)) (
return RequestWithID{ServerAndID: id, Request: bestRequest}, true
}
// InvalidResponse implements Tracker.
func (p *tracker) InvalidResponse(id ServerAndID, desc string) {
id.Server.(server).fail(desc)
}

14
beacon/light/sync/head_sync.go
View file

@ -28,6 +28,11 @@ type headTracker interface {
SetPrefetchHead(head types.HeadInfo)
}
// HeadSync implements request.Module; it updates the validated and prefetch
// heads of HeadTracker based on the EvHead and EvSignedHead events coming from
// registered servers.
// It can also postpone the validation of the latest announced signed head
// until the committee chain is synced up to at least the required period.
type HeadSync struct {
headTracker headTracker
chain committeeChain
@ -40,11 +45,16 @@ type HeadSync struct {
prefetchHead types.HeadInfo
}
// headServerCount is associated with most recently seen head infos; it counts
// the number of servers currently having the given head info as their announced
// head and a counter signaling how recent that head is.
// This data is used for selecting the prefetch head.
type headServerCount struct {
serverCount int
headCounter uint64
}
// NewHeadSync creates a new HeadSync.
func NewHeadSync(headTracker headTracker, chain committeeChain) *HeadSync {
s := &HeadSync{
headTracker: headTracker,
@ -85,6 +95,8 @@ func (s *HeadSync) Process(tracker request.Tracker, events []request.Event) (tri
return
}
// newSignedHead handles received signed head; either validates it if the chain
// is properly synced or stores it for further validation.
func (s *HeadSync) newSignedHead(server request.Server, signedHead types.SignedHeader) (trigger bool) {
if !s.chainInit || types.SyncPeriod(signedHead.SignatureSlot) > s.nextSyncPeriod {
s.unvalidatedHeads[server] = signedHead
@ -94,6 +106,8 @@ func (s *HeadSync) newSignedHead(server request.Server, signedHead types.SignedH
return updated
}
// processUnvalidatedHeads iterates the list of unvalidated heads and validates
// those which can be validated.
func (s *HeadSync) processUnvalidatedHeads() (trigger bool) {
if !s.chainInit {
return false

29
beacon/light/sync/update_sync.go
View file

@ -26,7 +26,7 @@ import (
"github.com/ethereum/go-ethereum/log"
)
const maxUpdateRequest = 8
const maxUpdateRequest = 8 // maximum number of updates requested in a single request
type committeeChain interface {
CheckpointInit(bootstrap types.BootstrapData) error
@ -34,6 +34,9 @@ type committeeChain interface {
NextSyncPeriod() (uint64, bool)
}
// CheckpointInit implements request.Module; it fetches the light client bootstrap
// data belonging to the given checkpoint hash and initializes the committee chain
// if successful.
type CheckpointInit struct {
chain committeeChain
checkpointHash common.Hash
@ -41,6 +44,7 @@ type CheckpointInit struct {
initialized bool
}
// NewCheckpointInit creates a new CheckpointInit.
func NewCheckpointInit(chain committeeChain, checkpointHash common.Hash) *CheckpointInit {
return &CheckpointInit{
chain: chain,
@ -78,6 +82,12 @@ func (s *CheckpointInit) Process(tracker request.Tracker, events []request.Event
return false
}
// ForwardUpdateSync implements request.Module; it fetches updates between the
// committee chain head and each server's announced head. Updates are fetched
// in batches and multiple batches can also be requested in parallel.
// Out of order responses are also handled; if a batch of updates cannot be added
// to the chain immediately because of a gap then the future updates are
// remembered until they can be processed.
type ForwardUpdateSync struct {
chain committeeChain
rangeLock rangeLock
@ -86,6 +96,7 @@ type ForwardUpdateSync struct {
nextSyncPeriod map[request.Server]uint64
}
// NewForwardUpdateSync creates a new ForwardUpdateSync.
func NewForwardUpdateSync(chain committeeChain) *ForwardUpdateSync {
return &ForwardUpdateSync{
chain: chain,
@ -95,8 +106,12 @@ func NewForwardUpdateSync(chain committeeChain) *ForwardUpdateSync {
}
}
// rangeLock allows locking sections of an integer space, preventing the syncing
// mechanism from making requests again for sections where a not timed out request
// is already pending or where already fetched and unprocessed data is available.
type rangeLock map[uint64]int
// lock locks or unlocks the given section, depending on the sign of the add parameter.
func (r rangeLock) lock(first, count uint64, add int) {
for i := first; i < first+count; i++ {
if v := r[i] + add; v > 0 {
@ -107,6 +122,8 @@ func (r rangeLock) lock(first, count uint64, add int) {
}
}
// firstUnlocked returns the first unlocked section starting at or after start
// and not longer than maxCount.
func (r rangeLock) firstUnlocked(start, maxCount uint64) (first, count uint64) {
first = start
for {
@ -127,6 +144,8 @@ func (r rangeLock) firstUnlocked(start, maxCount uint64) (first, count uint64) {
return
}
// lockRange locks the range belonging to the given update request, unless the
// same request has already been locked
func (s *ForwardUpdateSync) lockRange(sid request.ServerAndID, req request.Request) {
if _, ok := s.lockedIDs[sid]; ok {
return
@ -136,6 +155,8 @@ func (s *ForwardUpdateSync) lockRange(sid request.ServerAndID, req request.Reque
s.rangeLock.lock(r.FirstPeriod, r.Count, 1)
}
// unlockRange unlocks the range belonging to the given update request, unless
// same request has already been unlocked
func (s *ForwardUpdateSync) unlockRange(sid request.ServerAndID, req request.Request) {
if _, ok := s.lockedIDs[sid]; !ok {
return
@ -145,6 +166,8 @@ func (s *ForwardUpdateSync) unlockRange(sid request.ServerAndID, req request.Req
s.rangeLock.lock(r.FirstPeriod, r.Count, -1)
}
// verifyRange returns true if the number of updates and the individual update
// periods in the response match the requested section.
func (s *ForwardUpdateSync) verifyRange(req request.Request, resp request.Response) bool {
request, ok := req.(ReqUpdates)
if !ok {
@ -165,7 +188,8 @@ func (s *ForwardUpdateSync) verifyRange(req request.Request, resp request.Respon
return true
}
// returns true for partial success
// processResponse adds the fetched updates and committees to the committee chain.
// Returns true in case of full or partial success.
func (s *ForwardUpdateSync) processResponse(tracker request.Tracker, event request.Event) (success bool) {
sid, _, resp := event.RequestInfo()
response, ok := resp.(RespUpdates)
@ -191,6 +215,7 @@ func (s *ForwardUpdateSync) processResponse(tracker request.Tracker, event reque
return
}
// updateResponseList implements sort.Sort and sorts update request/response events by FirstPeriod.
type updateResponseList []request.Event
func (u updateResponseList) Len() int { return len(u) }

18
cmd/blsync/block_sync.go
View file

@ -37,6 +37,8 @@ import (
"github.com/protolambda/ztyp/tree"
)
// beaconBlockSync implements request.Module; it fetches the beacon blocks belonging
// to the validated and prefetch heads.
type beaconBlockSync struct {
recentBlocks *lru.Cache[common.Hash, *capella.BeaconBlock]
validatedHead common.Hash
@ -50,7 +52,8 @@ type headTracker interface {
ValidatedHead() types.SignedHeader
}
func newBeaconBlockSyncer(headTracker headTracker) *beaconBlockSync {
// newBeaconBlockSync returns a new beaconBlockSync.
func newBeaconBlockSync(headTracker headTracker) *beaconBlockSync {
return &beaconBlockSync{
headTracker: headTracker,
recentBlocks: lru.NewCache[common.Hash, *capella.BeaconBlock](10),
@ -96,12 +99,18 @@ func (s *beaconBlockSync) Process(tracker request.Tracker, events []request.Even
return
}
// belongs to validatedHead (or nil)
// getHeadBlock returns the beacon block belonging to ValidatedHead or nil if not available.
func (s *beaconBlockSync) getHeadBlock() *capella.BeaconBlock {
block, _ := s.recentBlocks.Get(s.validatedHead)
return block
}
// tryRequestBlock tries to send a block request for the given root if the block
// is not available and the root is not locked by another pending request.
// If prefetch is true then the request is only sent to a server whose latest
// announced head has the same block root. If prefetch is false then a validated
// block is requested which is expected to be available at every properly synced
// server, therefore no such restriction is applied.
func (s *beaconBlockSync) tryRequestBlock(tracker request.Tracker, blockRoot common.Hash, prefetch bool) {
if _, ok := s.recentBlocks.Get(blockRoot); ok {
return
@ -121,6 +130,7 @@ func (s *beaconBlockSync) tryRequestBlock(tracker request.Tracker, blockRoot com
}
}
// getExecBlock extracts the execution block from the beacon block's payload.
func getExecBlock(beaconBlock *capella.BeaconBlock) (*ctypes.Block, error) {
payload := &beaconBlock.Body.ExecutionPayload
txs := make([]*ctypes.Transaction, len(payload.Transactions))
@ -167,10 +177,14 @@ func getExecBlock(beaconBlock *capella.BeaconBlock) (*ctypes.Block, error) {
return execBlock, nil
}
// beaconBlockHash calculates the hash of a beacon block.
func beaconBlockHash(beaconBlock *capella.BeaconBlock) common.Hash {
return common.Hash(beaconBlock.HashTreeRoot(configs.Mainnet, tree.GetHashFn()))
}
// engineApiUpdater implements request.Module. This module does not start requests,
// it is only implemented as a module in order to easily trigger it by successful
// head block retrieval.
type engineApiUpdater struct {
client *rpc.Client
trigger func()

2
cmd/blsync/block_sync_test.go
View file

@ -41,7 +41,7 @@ func TestBlockSync(t *testing.T) {
tracker.AddServer(testServer1, 1)
tracker.AddServer(testServer2, 1)
ht := &testHeadTracker{}
blockSync := newBeaconBlockSyncer(ht)
blockSync := newBeaconBlockSync(ht)
expHeadBlock := func(tci int, expHead *capella.BeaconBlock) {
expInfo := blockHeadInfo(expHead)

2
cmd/blsync/main.go
View file

@ -125,7 +125,7 @@ func blsync(ctx *cli.Context) error {
checkpointInit := sync.NewCheckpointInit(committeeChain, chainConfig.Checkpoint)
forwardSync := sync.NewForwardUpdateSync(committeeChain)
beaconBlockSync := newBeaconBlockSyncer(headTracker)
beaconBlockSync := newBeaconBlockSync(headTracker)
engineApiUpdater := &engineApiUpdater{ //TODO constructor
client: makeRPCClient(ctx),
blockSync: beaconBlockSync,