diff --git a/beacon/light/api/api_server.go b/beacon/light/api/api_server.go index 92fe494278..493b651f11 100755 --- a/beacon/light/api/api_server.go +++ b/beacon/light/api/api_server.go @@ -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() diff --git a/beacon/light/head_tracker.go b/beacon/light/head_tracker.go index 2b2afb1f60..56301c386b 100644 --- a/beacon/light/head_tracker.go +++ b/beacon/light/head_tracker.go @@ -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() diff --git a/beacon/light/request/scheduler.go b/beacon/light/request/scheduler.go index 1de4c8aa41..f7bca17568 100644 --- a/beacon/light/request/scheduler.go +++ b/beacon/light/request/scheduler.go @@ -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() { diff --git a/beacon/light/request/server.go b/beacon/light/request/server.go index 3718db6f83..8d0d7d97ad 100644 --- a/beacon/light/request/server.go +++ b/beacon/light/request/server.go @@ -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 diff --git a/beacon/light/request/tracker.go b/beacon/light/request/tracker.go index 12099e951d..a50ae127b5 100644 --- a/beacon/light/request/tracker.go +++ b/beacon/light/request/tracker.go @@ -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 - scheduler *Scheduler - module Module + // 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) } diff --git a/beacon/light/sync/head_sync.go b/beacon/light/sync/head_sync.go index d69d382ece..49593f2443 100644 --- a/beacon/light/sync/head_sync.go +++ b/beacon/light/sync/head_sync.go @@ -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 diff --git a/beacon/light/sync/update_sync.go b/beacon/light/sync/update_sync.go index 92b4e7f840..8d2d6e3296 100644 --- a/beacon/light/sync/update_sync.go +++ b/beacon/light/sync/update_sync.go @@ -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) } diff --git a/cmd/blsync/block_sync.go b/cmd/blsync/block_sync.go index c28094f285..a840afd077 100755 --- a/cmd/blsync/block_sync.go +++ b/cmd/blsync/block_sync.go @@ -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() diff --git a/cmd/blsync/block_sync_test.go b/cmd/blsync/block_sync_test.go index 1d5fdb0dea..eef3c77a9b 100644 --- a/cmd/blsync/block_sync_test.go +++ b/cmd/blsync/block_sync_test.go @@ -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) diff --git a/cmd/blsync/main.go b/cmd/blsync/main.go index 68b3d0a218..cd4db01f7f 100644 --- a/cmd/blsync/main.go +++ b/cmd/blsync/main.go @@ -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,