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: updated docs and improved code readability

Browse source
This commit is contained in:
Zsolt Felfoldi 2024-01-22 05:14:48 +01:00 committed by Felix Lange
parent f734747364
commit d42b0e9b55
4 changed files with 161 additions and 119 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -25,29 +25,39 @@ import (
)
// Module represents a mechanism which is typically responsible for downloading
// and updating a passive data structure.
// Modules can start network requests through Tracker and receive request events
// related to the sent requests that can signal a response, a failure or a timeout.
// They also receive server-related events. Note that they do not directly interact
// 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.
// and updating a passive data structure. It does not directly interact with the
// servers (except for reporting server side failures). It receives and processes
// events, maintains its internal state and generates request candidates. It is
// the Scheduler's responsibility to feed events to the modules, call Process as
// long as there might be something to process and then generate request
// candidates using MakeRequest and start the best possible requests.
// 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
// fire the trigger. Changing a target data structure also triggers a next
// processing round as it 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 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 is a non-blocking function responsible for maintaining the target
// data structures(s) and the internal state of the module. This state
// typically consists of information about pending requests and registered
// servers and it is updated based on the received events.
// Process is always called after an event is received or after a target data
// structure has been changed.
//
// 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.
Process([]Event)
// MakeRequest generates a request candidate based on the state of the target
// structure(s) and the internal state of the module. This candidate is
// typically the next obtainable item (or range of items) of the target
// structure that is assumed to be available at the given server and has not
// been requested yet (or has been requested but already timed out and should
// be resent).
// MakeRequest is always called after Process. Note that it is the Scheduler's
// job to select the best possible requests and actually send them. If a
// request has been sent, the module is notified through an EvRequest event
// which also immediately triggers a next processing round, allowing modules
// to send more requests if possible and necessary.
MakeRequest(Server) (Request, float32)
}
@ -78,7 +88,8 @@ type Scheduler struct {
}
type (
// Server identifies a server without allowing any direct interaction.
// Server identifies a server without allowing any direct interaction except
// for reporting a server side failure.
// 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
@ -93,12 +104,10 @@ type (
Server Server
ID ID
}
RequestWithID struct {
ServerAndID
Request Request
}
)
// targetData represents a registered target data structure that increases its
// ChangeCounter whenever it has been changed.
type targetData interface {
ChangeCounter() uint64
}
@ -128,6 +137,9 @@ func NewScheduler(clock mclock.Clock) *Scheduler {
return s
}
// RegisterTarget registers a target data structure, ensuring that any changes
// made to it trigger a new round of Module.Process calls, giving a chance to
// modules to react to the changes.
func (s *Scheduler) RegisterTarget(t targetData) {
s.lock.Lock()
defer s.lock.Unlock()
@ -192,19 +204,14 @@ func (s *Scheduler) Stop() {
<-stop
}
// syncLoop calls all modules in the order of their registration.
// syncLoop is the main event loop responsible for event/data processing and
// sending new requests.
// A round of processing starts whenever the global trigger is fired. Triggers
// fired during a processing round ensure that there is going to be a next round.
func (s *Scheduler) syncLoop() {
for {
s.lock.Lock()
for {
s.processModules()
if !s.targetChanged() {
break
}
}
s.sendRequests()
s.processRound()
s.lock.Unlock()
loop:
for {
@ -220,6 +227,8 @@ func (s *Scheduler) syncLoop() {
}
}
// targetChanged returns true if a registered target data structure has been
// changed since the last call to this function.
func (s *Scheduler) targetChanged() (changed bool) {
for target, counter := range s.targets {
if newCounter := target.ChangeCounter(); newCounter != counter {
@ -230,17 +239,31 @@ func (s *Scheduler) targetChanged() (changed bool) {
return
}
// processModules runs an entire processing round, calling the Process functions
// of all modules, passing all relevant events.
func (s *Scheduler) processModules() {
serverEvents, requestEvents := s.filterEvents()
log.Debug("Processing modules", "server events", len(serverEvents))
for _, module := range s.modules {
log.Debug("Processing module", "name", s.names[module], "request events", len(requestEvents[module]))
module.Process(append(serverEvents, requestEvents[module]...))
// processRound runs an entire processing round. It calls the Process functions
// of all modules, passing all relevant events and repeating Process calls as
// long as any changes have been made to the registered target data structures.
// Once all events have been processed and a stable state has been achieved,
// requests are generated and sent if necessary and possible.
func (s *Scheduler) processRound() {
for {
serverEvents, requestEvents := s.filterEvents()
log.Debug("Processing modules", "server events", len(serverEvents))
for _, module := range s.modules {
log.Debug("Processing module", "name", s.names[module], "request events", len(requestEvents[module]))
module.Process(append(serverEvents, requestEvents[module]...))
}
if !s.targetChanged() {
break
}
}
s.sendRequests()
}
// sendRequests lets each module generate a request if necessary and sends it to
// a suitable server if possible.
// Note that if a request is sent, an EvRequest event will immediately trigger a
// next processing round, thereby allowing modules to create any number of requests
// in any suitable moment as long as there is a server that can accept them.
func (s *Scheduler) sendRequests() {
servers := make(map[server]struct{})
for server := range s.servers {
@ -262,7 +285,7 @@ func (s *Scheduler) sendRequests() {
// tryRequest tries to generate request candidates for a given module and a given
// set of servers, then selects the best candidate if there is one and sends the
// request.
// request to the server it was generated for.
// The candidates are primarily ranked based on "request priority", a number that
// Module.MakeRequest has returned along with the request candidate. This ranking
// may or may not be used depending on the type of the request, identical requests
@ -272,6 +295,8 @@ func (s *Scheduler) sendRequests() {
// ranked based on "server priority" which is determined by the server. This value
// is typically higher is the server is expected to respond quicker or with a
// higher chance (typically a lower number of pending requests).
// Note that tryRequest can also remove items from the set of available servers
// if they are no longer able to accept requests in the current processing round.
func (s *Scheduler) tryRequest(module Module, servers map[server]struct{}) bool {
var (
maxServerPriority, maxRequestPriority float32
@ -303,8 +328,8 @@ func (s *Scheduler) tryRequest(module Module, servers map[server]struct{}) bool
if bestServer == nil {
return false
}
id := ServerAndID{Server: bestServer, ID: bestServer.sendRequest(bestRequest)}
s.pending[id] = pendingRequest{request: bestRequest, module: module}
sid := ServerAndID{Server: bestServer, ID: bestServer.sendRequest(bestRequest)}
s.pending[sid] = pendingRequest{request: bestRequest, module: module}
return true
}
@ -348,39 +373,48 @@ func (s *Scheduler) filterEvents() (serverEvents []Event, requestEvents map[Modu
log.Error("Server interface type unknown for Scheduler")
continue
}
if event.Type == EvRegistered {
s.servers[server] = struct{}{}
}
if _, ok := s.servers[server]; !ok {
continue
if _, ok := s.servers[server]; !ok && event.Type != EvRegistered {
continue // before EvRegister or after EvUnregister, discard
}
if event.IsRequestEvent() {
sid, _, _ := event.RequestInfo()
if pr, ok := s.pending[sid]; ok {
requestEvents[pr.module] = append(requestEvents[pr.module], event)
if event.Type == EvResponse || event.Type == EvFail {
delete(s.pending, sid)
}
pending, ok := s.pending[sid]
if !ok {
continue // request already closed, ignore further events
}
return
}
if event.Type == EvUnregistered {
delete(s.servers, server)
for id, pending := range s.pending {
if id.Server != event.Server {
continue
}
requestEvents[pending.module] = append(requestEvents[pending.module], Event{
Type: EvFail,
Server: event.Server,
Data: RequestResponse{
ID: id.ID,
Request: pending.request,
},
})
if event.Type == EvResponse || event.Type == EvFail {
delete(s.pending, sid) // final event, close pending request
}
requestEvents[pending.module] = append(requestEvents[pending.module], event)
} else {
switch event.Type {
case EvRegistered:
s.servers[server] = struct{}{}
case EvUnregistered:
s.closePending(event.Server, requestEvents)
delete(s.servers, server)
}
serverEvents = append(serverEvents, event)
}
serverEvents = append(serverEvents, event)
}
return
}
// closePending closes all pending requests to the given server and adds an EvFail
// event to properly finalize them
func (s *Scheduler) closePending(server Server, requestEvents map[Module][]Event) {
for sid, pending := range s.pending {
if sid.Server == server {
requestEvents[pending.module] = append(requestEvents[pending.module], Event{
Type: EvFail,
Server: server,
Data: RequestResponse{
ID: sid.ID,
Request: pending.request,
},
})
delete(s.pending, sid)
}
}
}

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

@ -39,8 +39,8 @@ var (
)
const (
softRequestTimeout = time.Second
hardRequestTimeout = time.Second * 10
softRequestTimeout = time.Second // allow resending request to a different server but do not cancel yet
hardRequestTimeout = time.Second * 10 // cancel request
)
const (
@ -54,7 +54,8 @@ const (
)
// 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
// through the event callback. When successfully sending a request it should
// send back an EvRequest event. When finished, it should send back either
// EvResponse or EvFail. Additionally, it may also send application-defined
// events that the Modules can interpret.
type requestServer interface {
@ -157,34 +158,7 @@ func (s *serverWithTimeout) eventCallback(event Event) {
switch event.Type {
case EvRequest:
id := event.Data.(RequestResponse).ID
s.timeouts[id] = s.clock.AfterFunc(softRequestTimeout, func() {
/*if s.testTimerResults != nil {
s.testTimerResults = append(s.testTimerResults, true) // simulated timer finished
}*/
s.lock.Lock()
if _, ok := s.timeouts[id]; !ok {
s.lock.Unlock()
return
}
s.timeouts[id] = s.clock.AfterFunc(hardRequestTimeout-softRequestTimeout, func() {
/*if s.testTimerResults != nil {
s.testTimerResults = append(s.testTimerResults, true) // simulated timer finished
}*/
s.lock.Lock()
if _, ok := s.timeouts[id]; !ok {
s.lock.Unlock()
return
}
delete(s.timeouts, id)
childEventCb := s.childEventCb
s.lock.Unlock()
childEventCb(Event{Type: EvFail, Data: event.Data})
})
childEventCb := s.childEventCb
s.lock.Unlock()
childEventCb(Event{Type: EvTimeout, Data: event.Data})
})
s.startTimeout(event.Data.(RequestResponse))
case EvResponse, EvFail:
id := event.Data.(RequestResponse).ID
if timer, ok := s.timeouts[id]; ok {
@ -199,8 +173,39 @@ func (s *serverWithTimeout) eventCallback(event Event) {
}
}
// sendRequest sends a request through the parent (requestServer) and starts a
// timer for request timeout.
// startTimeout starts a timeout timer for the given request.
func (s *serverWithTimeout) startTimeout(reqData RequestResponse) {
id := reqData.ID
s.timeouts[id] = s.clock.AfterFunc(softRequestTimeout, func() {
/*if s.testTimerResults != nil {
s.testTimerResults = append(s.testTimerResults, true) // simulated timer finished
}*/
s.lock.Lock()
if _, ok := s.timeouts[id]; !ok {
s.lock.Unlock()
return
}
s.timeouts[id] = s.clock.AfterFunc(hardRequestTimeout-softRequestTimeout, func() {
/*if s.testTimerResults != nil {
s.testTimerResults = append(s.testTimerResults, true) // simulated timer finished
}*/
s.lock.Lock()
if _, ok := s.timeouts[id]; !ok {
s.lock.Unlock()
return
}
delete(s.timeouts, id)
childEventCb := s.childEventCb
s.lock.Unlock()
childEventCb(Event{Type: EvFail, Data: reqData})
})
childEventCb := s.childEventCb
s.lock.Unlock()
childEventCb(Event{Type: EvTimeout, Data: reqData})
})
}
// sendRequest sends a request through the parent (requestServer).
func (s *serverWithTimeout) sendRequest(request Request) (reqId ID) {
return s.parent.SendRequest(request)
}

35
beacon/light/sync/test_helpers.go
View file

@ -33,13 +33,18 @@ func (s *TestServer) Fail(desc string) {
s.ts.serverFail(s)
}
type requestWithID struct {
sid request.ServerAndID
request request.Request
}
type TestScheduler struct {
t *testing.T
module request.Module
events []request.Event
servers []request.Server
allowance map[request.Server]int
sent map[int]request.RequestWithID
sent map[int]requestWithID
testIndex int
expFail map[request.Server]int // expected Server.Fail calls during next Run
lastId request.ID
@ -51,7 +56,7 @@ func NewTestScheduler(t *testing.T, module request.Module) *TestScheduler {
module: module,
allowance: make(map[request.Server]int),
expFail: make(map[request.Server]int),
sent: make(map[int]request.RequestWithID),
sent: make(map[int]requestWithID),
}
}
@ -68,9 +73,9 @@ func (ts *TestScheduler) Run(testIndex int, expServer request.Server, expReq req
ts.t.Errorf("Missing %d Server.Fail(s) from server %d in test case #%d", count, server.(*TestServer).ID, testIndex)
}
expReqWithID := request.RequestWithID{
ServerAndID: request.ServerAndID{Server: expServer, ID: ts.lastId + 1},
Request: expReq,
expReqWithID := requestWithID{
sid: request.ServerAndID{Server: expServer, ID: ts.lastId + 1},
request: expReq,
}
req, ok := ts.tryRequest(testIndex, ts.module.MakeRequest)
if expReq == nil {
@ -88,8 +93,8 @@ func (ts *TestScheduler) Run(testIndex int, expServer request.Server, expReq req
}
}
func (ts *TestScheduler) Request(testIndex int) request.RequestWithID {
return ts.sent[testIndex]
func (ts *TestScheduler) Request(testIndex int) request.Request {
return ts.sent[testIndex].request
}
func (ts *TestScheduler) ServerEvent(evType *request.EventType, server request.Server, data any) {
@ -108,10 +113,10 @@ func (ts *TestScheduler) RequestEvent(evType *request.EventType, testIndex int,
}
ts.events = append(ts.events, request.Event{
Type: evType,
Server: req.ServerAndID.Server,
Server: req.sid.Server,
Data: request.RequestResponse{
ID: req.ServerAndID.ID,
Request: req.Request,
ID: req.sid.ID,
Request: req.request,
Response: resp,
},
})
@ -153,7 +158,7 @@ func (ts *TestScheduler) serverFail(server request.Server) {
ts.expFail[server]--
}
func (ts *TestScheduler) tryRequest(testIndex int, requestFn func(server request.Server) (request.Request, float32)) (request.RequestWithID, bool) {
func (ts *TestScheduler) tryRequest(testIndex int, requestFn func(server request.Server) (request.Request, float32)) (requestWithID, bool) {
var (
bestServer request.Server
bestReq request.Request
@ -169,13 +174,13 @@ func (ts *TestScheduler) tryRequest(testIndex int, requestFn func(server request
}
}
if bestServer == nil {
return request.RequestWithID{}, false
return requestWithID{}, false
}
ts.allowance[bestServer]--
ts.lastId++
req := request.RequestWithID{
ServerAndID: request.ServerAndID{Server: bestServer, ID: ts.lastId},
Request: bestReq,
req := requestWithID{
sid: request.ServerAndID{Server: bestServer, ID: ts.lastId},
request: bestReq,
}
ts.sent[testIndex] = req
ts.RequestEvent(request.EvRequest, testIndex, nil)

8
beacon/light/sync/update_sync_test.go
View file

@ -170,9 +170,7 @@ func TestUpdateSyncDifferentHeads(t *testing.T) {
chain.ExpNextSyncPeriod(t, 15)
// invalid response to request 1, server can only deliver updates up to period 15 despite announced head
req1x := ts.Request(1)
req1x.Request = ReqUpdates{FirstPeriod: 10, Count: 5}
ts.RequestEvent(request.EvResponse, 1, testRespUpdate(req1x))
ts.RequestEvent(request.EvResponse, 1, testRespUpdate(ReqUpdates{FirstPeriod: 10, Count: 5}))
ts.ExpFail(testServer3)
ts.Run(5, nil, nil)
// expect no progress of chain head
@ -197,9 +195,9 @@ func TestUpdateSyncDifferentHeads(t *testing.T) {
chain.ExpNextSyncPeriod(t, 17)
}
func testRespUpdate(request request.RequestWithID) request.Response {
func testRespUpdate(request request.Request) request.Response {
var resp RespUpdates
req := request.Request.(ReqUpdates)
req := request.(ReqUpdates)
resp.Updates = make([]*types.LightClientUpdate, int(req.Count))
resp.Committees = make([]*types.SerializedSyncCommittee, int(req.Count))
period := req.FirstPeriod