From c163cbdfc2ecf06482b79ad37a0db44ecf3242f9 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Tue, 16 Jan 2024 05:42:32 +0100 Subject: [PATCH] beacon/light: simplified Module interface --- beacon/light/request/scheduler.go | 115 +++++++++++++++--------------- beacon/light/sync/head_sync.go | 24 +++---- beacon/light/sync/update_sync.go | 84 +++++++++++----------- cmd/blsync/block_sync.go | 38 +++++----- 4 files changed, 130 insertions(+), 131 deletions(-) diff --git a/beacon/light/request/scheduler.go b/beacon/light/request/scheduler.go index 6771f39584..d5b8c02275 100644 --- a/beacon/light/request/scheduler.go +++ b/beacon/light/request/scheduler.go @@ -47,8 +47,7 @@ type Module interface { // Note: Process functions of different modules are never called concurrently; // they are called by Scheduler in the same order of priority as they were // registered in. - HandleEvent(Event) - Process() + Process([]Event) MakeRequest(Server) (Request, float32) } @@ -65,10 +64,11 @@ type Scheduler struct { servers map[server]struct{} targets map[targetData]uint64 - pending map[ServerAndID]pendingRequest - eventLock sync.Mutex - events []Event - stopCh chan chan struct{} + pending map[ServerAndID]pendingRequest + eventLock sync.Mutex + serverEvents []Event + requestEvents map[Module][]Event + stopCh chan chan struct{} triggerCh chan struct{} // restarts waiting sync loop // testWaitCh chan struct{} // accepts sends when sync loop is waiting @@ -150,12 +150,16 @@ func (s *Scheduler) RegisterServer(rs requestServer) { defer s.lock.Unlock() server := newServer(rs, s.clock) + s.servers[server] = struct{}{} + s.eventLock.Lock() s.addEvent(Event{Type: EvRegistered, Server: server}) + s.eventLock.Unlock() server.subscribe(func(event Event) { event.Server = server + s.eventLock.Lock() s.addEvent(event) + s.eventLock.Unlock() }) - s.servers[server] = struct{}{} } // UnregisterServer removes a registered server. @@ -166,8 +170,10 @@ func (s *Scheduler) UnregisterServer(rs requestServer) { for server := range s.servers { if sl, ok := server.(*serverWithLimits); ok && sl.parent == rs { server.unsubscribe() - delete(s.servers, server) + s.eventLock.Lock() s.addEvent(Event{Type: EvUnregistered, Server: server}) + s.eventLock.Unlock() + delete(s.servers, server) return } } @@ -198,7 +204,6 @@ func (s *Scheduler) Stop() { func (s *Scheduler) syncLoop() { for { s.lock.Lock() - s.handleEvents() for { s.processModules() if !s.targetChanged() { @@ -234,8 +239,15 @@ func (s *Scheduler) targetChanged() (changed bool) { // processModules runs an entire processing round, calling the Process functions // of all modules, passing all relevant events. func (s *Scheduler) processModules() { + s.eventLock.Lock() + serverEvents, requestEvents := s.serverEvents, s.requestEvents + s.serverEvents, s.requestEvents = nil, nil + s.eventLock.Unlock() + + log.Debug("Processing modules", "server events", len(s.serverEvents)) for _, module := range s.modules { - module.Process() + log.Debug("Processing module", "name", s.names[module], "request events", len(s.requestEvents[module])) + module.Process(append(serverEvents, requestEvents[module]...)) } } @@ -246,26 +258,25 @@ func (s *Scheduler) sendRequests() { servers[server] = struct{}{} } } - log.Debug("Processing modules", "servers", len(servers)) + log.Debug("Generating request candidates", "servers", len(servers)) for _, module := range s.modules { - for { - if len(servers) == 0 { - return - } - if req, sent := s.tryRequest(module, servers); sent { - module.HandleEvent(Event{ - Type: EvRequest, - Server: req.Server, - Data: RequestResponse{ - ID: req.ID, - Request: req.Request, - }, - }) - } else { - break - } + if len(servers) == 0 { + return } + s.eventLock.Lock() // ensure that EvRequest is added first + if req, sent := s.tryRequest(module, servers); sent { + log.Debug("Sent request", "module", s.names[module]) + s.addEvent(Event{ + Type: EvRequest, + Server: req.Server, + Data: RequestResponse{ + ID: req.ID, + Request: req.Request, + }, + }) + } + s.eventLock.Unlock() } } @@ -314,38 +325,21 @@ func (s *Scheduler) Trigger() { } } +// addEvent adds an Event 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 adding a failed request event (EvFail), +// ensuring that all requests get finalized and thereby allowing the module logic +// to be safe and simple. func (s *Scheduler) addEvent(event Event) { - s.eventLock.Lock() - s.events = append(s.events, event) - s.Trigger() - s.eventLock.Unlock() -} - -func (s *Scheduler) handleEvents() { - s.eventLock.Lock() - events := s.events - s.events = nil - s.eventLock.Unlock() - for _, event := range events { - if _, ok := s.servers[event.Server.(server)]; ok { - s.handleEvent(event) - } else { - log.Error("Event received from unsubscribed server") - } + if _, ok := s.servers[event.Server.(server)]; !ok { + return } -} - -// 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 (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() { sid, _, _ := event.RequestInfo() if pr, ok := s.pending[sid]; ok { - pr.module.HandleEvent(event) - if event.Type != EvTimeout { + s.addRequestEvent(pr.module, event) + if event.Type == EvResponse || event.Type == EvFail { delete(s.pending, sid) } } @@ -356,7 +350,7 @@ func (s *Scheduler) handleEvent(event Event) { if id.Server != event.Server { continue } - pending.module.HandleEvent(Event{ + s.addRequestEvent(pending.module, Event{ Type: EvFail, Server: event.Server, Data: RequestResponse{ @@ -366,7 +360,12 @@ func (s *Scheduler) handleEvent(event Event) { }) } } - for _, module := range s.modules { - module.HandleEvent(event) - } + s.serverEvents = append(s.serverEvents, event) +} + +func (s *Scheduler) addRequestEvent(module Module, event Event) { + if s.requestEvents == nil { + s.requestEvents = make(map[Module][]Event) + } + s.requestEvents[module] = append(s.requestEvents[module], event) } diff --git a/beacon/light/sync/head_sync.go b/beacon/light/sync/head_sync.go index 61a4291e86..a6a7fbf378 100644 --- a/beacon/light/sync/head_sync.go +++ b/beacon/light/sync/head_sync.go @@ -67,20 +67,20 @@ func NewHeadSync(headTracker headTracker, chain committeeChain) *HeadSync { return s } -func (s *HeadSync) HandleEvent(event request.Event) { - switch event.Type { - case EvNewHead: - s.setServerHead(event.Server, event.Data.(types.HeadInfo)) - case EvNewSignedHead: - s.newSignedHead(event.Server, event.Data.(types.SignedHeader)) - case request.EvUnregistered: - s.setServerHead(event.Server, types.HeadInfo{}) - delete(s.serverHeads, event.Server) - delete(s.unvalidatedHeads, event.Server) +func (s *HeadSync) Process(events []request.Event) { + for _, event := range events { + switch event.Type { + case EvNewHead: + s.setServerHead(event.Server, event.Data.(types.HeadInfo)) + case EvNewSignedHead: + s.newSignedHead(event.Server, event.Data.(types.SignedHeader)) + case request.EvUnregistered: + s.setServerHead(event.Server, types.HeadInfo{}) + delete(s.serverHeads, event.Server) + delete(s.unvalidatedHeads, event.Server) + } } -} -func (s *HeadSync) Process() { nextPeriod, chainInit := s.chain.NextSyncPeriod() if nextPeriod != s.nextSyncPeriod || chainInit != s.chainInit { s.nextSyncPeriod, s.chainInit = nextPeriod, chainInit diff --git a/beacon/light/sync/update_sync.go b/beacon/light/sync/update_sync.go index 724178a1cb..efcb2efeb2 100644 --- a/beacon/light/sync/update_sync.go +++ b/beacon/light/sync/update_sync.go @@ -52,30 +52,30 @@ func NewCheckpointInit(chain committeeChain, checkpointHash common.Hash) *Checkp } } -func (s *CheckpointInit) HandleEvent(event request.Event) { - if !event.IsRequestEvent() { - return - } - sid, req, resp := event.RequestInfo() - if event.Type == request.EvRequest { - s.locked = sid - return - } - if s.locked == sid { - s.locked = request.ServerAndID{} - } - if resp != nil { - if checkpoint, ok := resp.(*types.BootstrapData); ok && checkpoint.Header.Hash() == common.Hash(req.(ReqCheckpointData)) { - s.chain.CheckpointInit(*checkpoint) - s.initialized = true +func (s *CheckpointInit) Process(events []request.Event) { + for _, event := range events { + if !event.IsRequestEvent() { return } - event.Server.Fail("invalid checkpoint data") + sid, req, resp := event.RequestInfo() + if event.Type == request.EvRequest { + s.locked = sid + return + } + if s.locked == sid { + s.locked = request.ServerAndID{} + } + if resp != nil { + if checkpoint, ok := resp.(*types.BootstrapData); ok && checkpoint.Header.Hash() == common.Hash(req.(ReqCheckpointData)) { + s.chain.CheckpointInit(*checkpoint) + s.initialized = true + return + } + event.Server.Fail("invalid checkpoint data") + } } } -func (s *CheckpointInit) Process() {} - func (s *CheckpointInit) MakeRequest(server request.Server) (request.Request, float32) { if s.initialized || s.locked != (request.ServerAndID{}) { return nil, 0 @@ -226,33 +226,33 @@ func (u updateResponseList) Less(i, j int) bool { u[j].Data.(request.RequestResponse).Request.(ReqUpdates).FirstPeriod } -func (s *ForwardUpdateSync) HandleEvent(event request.Event) { - switch event.Type { - case request.EvRequest: - sid, req, _ := event.RequestInfo() - s.lockRange(sid, req) - case request.EvResponse, request.EvFail, request.EvTimeout: - sid, req, resp := event.RequestInfo() - if event.Type == request.EvResponse && !s.verifyRange(req, resp) { - event.Server.Fail("invalid update range") - resp = nil - } - if resp != nil { - // there is a response with a valid format; put it in the process queue - s.processQueue = append(s.processQueue, event) +func (s *ForwardUpdateSync) Process(events []request.Event) { + for _, event := range events { + switch event.Type { + case request.EvRequest: + sid, req, _ := event.RequestInfo() s.lockRange(sid, req) - } else { - s.unlockRange(sid, req) + case request.EvResponse, request.EvFail, request.EvTimeout: + sid, req, resp := event.RequestInfo() + if event.Type == request.EvResponse && !s.verifyRange(req, resp) { + event.Server.Fail("invalid update range") + resp = nil + } + if resp != nil { + // there is a response with a valid format; put it in the process queue + s.processQueue = append(s.processQueue, event) + s.lockRange(sid, req) + } else { + s.unlockRange(sid, req) + } + case EvNewSignedHead: + signedHead := event.Data.(types.SignedHeader) + s.nextSyncPeriod[event.Server] = types.SyncPeriod(signedHead.SignatureSlot + 256) + case request.EvUnregistered: + delete(s.nextSyncPeriod, event.Server) } - case EvNewSignedHead: - signedHead := event.Data.(types.SignedHeader) - s.nextSyncPeriod[event.Server] = types.SyncPeriod(signedHead.SignatureSlot + 256) - case request.EvUnregistered: - delete(s.nextSyncPeriod, event.Server) } -} -func (s *ForwardUpdateSync) Process() { // try processing ordered list of available responses sort.Sort(updateResponseList(s.processQueue)) //TODO for s.processQueue != nil { diff --git a/cmd/blsync/block_sync.go b/cmd/blsync/block_sync.go index 935ce3aac0..b70649715e 100755 --- a/cmd/blsync/block_sync.go +++ b/cmd/blsync/block_sync.go @@ -53,28 +53,28 @@ func newBeaconBlockSync(headTracker headTracker) *beaconBlockSync { } } -func (s *beaconBlockSync) HandleEvent(event request.Event) { - switch event.Type { - case request.EvRequest: - _, req, _ := event.RequestInfo() - blockRoot := common.Hash(req.(sync.ReqBeaconBlock)) - s.locked[blockRoot] = struct{}{} - case request.EvResponse, request.EvFail, request.EvTimeout: - _, req, resp := event.RequestInfo() - blockRoot := common.Hash(req.(sync.ReqBeaconBlock)) - if resp != nil { - block := resp.(*capella.BeaconBlock) - s.recentBlocks.Add(blockRoot, block) +func (s *beaconBlockSync) Process(events []request.Event) { + for _, event := range events { + switch event.Type { + case request.EvRequest: + _, req, _ := event.RequestInfo() + blockRoot := common.Hash(req.(sync.ReqBeaconBlock)) + s.locked[blockRoot] = struct{}{} + case request.EvResponse, request.EvFail, request.EvTimeout: + _, req, resp := event.RequestInfo() + blockRoot := common.Hash(req.(sync.ReqBeaconBlock)) + if resp != nil { + block := resp.(*capella.BeaconBlock) + s.recentBlocks.Add(blockRoot, block) + } + delete(s.locked, blockRoot) + case sync.EvNewHead: + s.serverHeads[event.Server] = event.Data.(types.HeadInfo).BlockRoot + case request.EvUnregistered: + delete(s.serverHeads, event.Server) } - delete(s.locked, blockRoot) - case sync.EvNewHead: - s.serverHeads[event.Server] = event.Data.(types.HeadInfo).BlockRoot - case request.EvUnregistered: - delete(s.serverHeads, event.Server) } -} -func (s *beaconBlockSync) Process() { // send validated head block if vh := s.headTracker.ValidatedHead(); vh != (types.SignedHeader{}) { validatedHead := vh.Header.Hash()