beacon/light: simplified Module interface

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([]Event)
-	Process()
 	MakeRequest(Server) (Request, float32)
 }
 
@@ -65,10 +64,11 @@ type Scheduler struct {
 	servers map[server]struct{}
 	targets map[targetData]uint64
 
-	pending   map[ServerAndID]pendingRequest
+	pending       map[ServerAndID]pendingRequest
-	eventLock sync.Mutex
+	eventLock     sync.Mutex
-	events    []Event
+	serverEvents  []Event
-	stopCh    chan chan struct{}
+	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 {
-			if len(servers) == 0 {
+			return
-				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
-			}
 		}
+		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()
+	if _, ok := s.servers[event.Server.(server)]; !ok {
-	s.events = append(s.events, event)
+		return
-	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")
-		}
 	}
-}
+	s.Trigger()
-
-// 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) {
 	if event.IsRequestEvent() {
 		sid, _, _ := event.RequestInfo()
 		if pr, ok := s.pending[sid]; ok {
-			pr.module.HandleEvent(event)
+			s.addRequestEvent(pr.module, event)
-			if event.Type != EvTimeout {
+			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 {
+	s.serverEvents = append(s.serverEvents, event)
-		module.HandleEvent(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)
 }

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) {
+func (s *HeadSync) Process(events []request.Event) {
-	switch event.Type {
+	for _, event := range events {
-	case EvNewHead:
+		switch event.Type {
-		s.setServerHead(event.Server, event.Data.(types.HeadInfo))
+		case EvNewHead:
-	case EvNewSignedHead:
+			s.setServerHead(event.Server, event.Data.(types.HeadInfo))
-		s.newSignedHead(event.Server, event.Data.(types.SignedHeader))
+		case EvNewSignedHead:
-	case request.EvUnregistered:
+			s.newSignedHead(event.Server, event.Data.(types.SignedHeader))
-		s.setServerHead(event.Server, types.HeadInfo{})
+		case request.EvUnregistered:
-		delete(s.serverHeads, event.Server)
+			s.setServerHead(event.Server, types.HeadInfo{})
-		delete(s.unvalidatedHeads, event.Server)
+			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

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) {
+func (s *CheckpointInit) Process(events []request.Event) {
-	if !event.IsRequestEvent() {
+	for _, event := range events {
-		return
+		if !event.IsRequestEvent() {
-	}
-	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")
+		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) {
+func (s *ForwardUpdateSync) Process(events []request.Event) {
-	switch event.Type {
+	for _, event := range events {
-	case request.EvRequest:
+		switch event.Type {
-		sid, req, _ := event.RequestInfo()
+		case request.EvRequest:
-		s.lockRange(sid, req)
+			sid, req, _ := event.RequestInfo()
-	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 {
+		case request.EvResponse, request.EvFail, request.EvTimeout:
-			s.unlockRange(sid, req)
+			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 {

cmd/blsync/block_sync.go

@@ -53,28 +53,28 @@ func newBeaconBlockSync(headTracker headTracker) *beaconBlockSync {
 	}
 }
 
-func (s *beaconBlockSync) HandleEvent(event request.Event) {
+func (s *beaconBlockSync) Process(events []request.Event) {
-	switch event.Type {
+	for _, event := range events {
-	case request.EvRequest:
+		switch event.Type {
-		_, req, _ := event.RequestInfo()
+		case request.EvRequest:
-		blockRoot := common.Hash(req.(sync.ReqBeaconBlock))
+			_, req, _ := event.RequestInfo()
-		s.locked[blockRoot] = struct{}{}
+			blockRoot := common.Hash(req.(sync.ReqBeaconBlock))
-	case request.EvResponse, request.EvFail, request.EvTimeout:
+			s.locked[blockRoot] = struct{}{}
-		_, req, resp := event.RequestInfo()
+		case request.EvResponse, request.EvFail, request.EvTimeout:
-		blockRoot := common.Hash(req.(sync.ReqBeaconBlock))
+			_, req, resp := event.RequestInfo()
-		if resp != nil {
+			blockRoot := common.Hash(req.(sync.ReqBeaconBlock))
-			block := resp.(*capella.BeaconBlock)
+			if resp != nil {
-			s.recentBlocks.Add(blockRoot, block)
+				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()