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: new Module and Requester interfaces

Browse source
This commit is contained in:
Zsolt Felfoldi 2024-01-29 02:46:24 +01:00 committed by Felix Lange
parent 084f23cfa0
commit c249d129b6
9 changed files with 316 additions and 346 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -17,7 +17,6 @@
package request
import (
"math"
"sync"
"github.com/ethereum/go-ethereum/log"
@ -45,19 +44,13 @@ 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.
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)
Process(Requester, []Event)
}
type Requester interface {
CanSendTo() []Server
Send(Server, Request) ID
Fail(Server, string)
}
// Scheduler is a modular network data retrieval framework that coordinates multiple
@ -72,7 +65,10 @@ type Scheduler struct {
servers map[server]struct{}
targets map[targetData]uint64
pending map[ServerAndID]pendingRequest
requesterLock sync.RWMutex
serverOrder []server
pending map[ServerAndID]pendingRequest
// eventLock guards access to the events list. Note that eventLock can be
// locked either while lock is locked or unlocked but lock cannot be locked
// while eventLock is locked.
@ -87,15 +83,12 @@ type Scheduler struct {
}
type (
// Server identifies a server without allowing any direct interaction except
// for reporting a server side failure.
// 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 interface {
Fail(desc string)
}
Server any
Request any
Response any
ID uint64
@ -238,91 +231,16 @@ func (s *Scheduler) targetChanged() (changed bool) {
// requests are generated and sent if necessary and possible.
func (s *Scheduler) processRound() {
for {
filteredEvents := s.filterEvents()
log.Debug("Processing modules")
filteredEvents := s.filterEvents()
for _, module := range s.modules {
log.Debug("Processing module", "name", s.names[module], "events", len(filteredEvents[module]))
module.Process(filteredEvents[module])
module.Process(requester{s, module}, filteredEvents[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 {
if ok, _ := server.canRequestNow(); ok {
servers[server] = struct{}{}
}
}
log.Debug("Generating request candidates", "servers", len(servers))
for _, module := range s.modules {
if len(servers) == 0 {
return
}
if s.tryRequest(module, servers) {
log.Debug("Sent request", "module", s.names[module])
}
}
}
// 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 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
// typically have the same priority while multiple item requests may have a
// priority based on the number of items requested.
// If there are multiple candidates with identical request priority then they are
// 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
bestServer server
bestRequest Request
)
maxServerPriority, maxRequestPriority = -math.MaxFloat32, -math.MaxFloat32
serverCount := len(servers)
var removed, candidates int
for server := range servers {
canRequest, serverPriority := server.canRequestNow()
if !canRequest {
delete(servers, server)
removed++
continue
}
request, requestPriority := module.MakeRequest(server)
if request != nil {
candidates++
}
if request == nil || requestPriority < maxRequestPriority ||
(requestPriority == maxRequestPriority && serverPriority <= maxServerPriority) {
continue
}
maxServerPriority, maxRequestPriority = serverPriority, requestPriority
bestServer, bestRequest = server, request
}
log.Debug("Request attempt", "serverCount", serverCount, "removedServers", removed, "requestCandidates", candidates)
if bestServer == nil {
return false
}
sid := ServerAndID{Server: bestServer, ID: bestServer.sendRequest(bestRequest)}
s.pending[sid] = pendingRequest{request: bestRequest, module: module}
return true
}
// Trigger starts a new processing round. If fired during processing, it ensures
@ -358,6 +276,9 @@ func (s *Scheduler) filterEvents() map[Module][]Event {
s.events = nil
s.eventLock.Unlock()
s.requesterLock.Lock()
defer s.requesterLock.Unlock()
filteredEvents := make(map[Module][]Event)
for _, event := range events {
server := event.Server.(server)
@ -379,9 +300,19 @@ func (s *Scheduler) filterEvents() map[Module][]Event {
switch event.Type {
case EvRegistered:
s.servers[server] = struct{}{}
s.serverOrder = append(s.serverOrder, nil)
copy(s.serverOrder[1:], s.serverOrder[:len(s.serverOrder)-1])
s.serverOrder[0] = server
case EvUnregistered:
s.closePending(event.Server, filteredEvents)
delete(s.servers, server)
for i, srv := range s.serverOrder {
if srv == server {
copy(s.serverOrder[i:len(s.serverOrder)-1], s.serverOrder[i+1:])
s.serverOrder = s.serverOrder[:len(s.serverOrder)-1]
break
}
}
}
for _, module := range s.modules {
filteredEvents[module] = append(filteredEvents[module], event)
@ -408,3 +339,44 @@ func (s *Scheduler) closePending(server Server, filteredEvents map[Module][]Even
}
}
}
type requester struct {
*Scheduler
module Module
}
func (s requester) CanSendTo() []Server {
s.requesterLock.RLock()
defer s.requesterLock.RUnlock()
list := make([]Server, 0, len(s.serverOrder))
for _, server := range s.serverOrder {
if server.canRequestNow() {
list = append(list, server)
}
}
return list
}
func (s requester) Send(srv Server, req Request) ID {
s.requesterLock.Lock()
defer s.requesterLock.Unlock()
server := srv.(server)
id := server.sendRequest(req)
sid := ServerAndID{Server: srv, ID: id}
s.pending[sid] = pendingRequest{request: req, module: s.module}
for i, ss := range s.serverOrder {
if ss == server {
copy(s.serverOrder[i:len(s.serverOrder)-1], s.serverOrder[i+1:])
s.serverOrder[len(s.serverOrder)-1] = server
return id
}
}
log.Error("Target server not found in ordered list of registered servers")
return id
}
func (s requester) Fail(srv Server, desc string) {
srv.(server).fail(desc)
}

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

@ -18,7 +18,6 @@ package request
import (
"math"
"math/rand"
"sync"
"time"
@ -28,7 +27,6 @@ import (
var (
// request events
EvRequest = &EventType{Name: "request", requestEvent: true} // data: RequestResponse; sent by Scheduler
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
@ -70,9 +68,9 @@ type requestServer interface {
// new requests based on timeouts and response failures.
type server interface {
subscribe(eventCallback func(Event))
canRequestNow() (bool, float32)
canRequestNow() bool
sendRequest(Request) ID
Fail(string)
fail(string)
unsubscribe()
}
@ -121,7 +119,7 @@ type RequestResponse struct {
Response Response
}
// serverWithTimeout wraps a requestServer and introduces two new request event
//TODO serverWithTimeout wraps a requestServer and introduces two new request event
// types: EvRequest and EvTimeout. Whenever a request is successfully sent, an
// EvRequest event is emitted first. The request's lifecycle is concluded if
// EvResponse or EvFail emitted by the parent requestServer. If this does not
@ -161,9 +159,7 @@ func (s *serverWithTimeout) sendRequest(request Request) (reqId ID) {
s.lock.Lock()
s.lastID++
id := s.lastID
reqData := RequestResponse{ID: id, Request: request}
s.childEventCb(Event{Type: EvRequest, Data: reqData})
s.startTimeout(reqData)
s.startTimeout(RequestResponse{ID: id, Request: request})
s.lock.Unlock()
s.parent.SendRequest(id, request)
return id
@ -293,12 +289,12 @@ func (s *serverWithLimits) eventCallback(event Event) {
s.parallelLimit += parallelAdjustUp
}
s.pendingCount--
if canRequest, _ := s.canRequest(); canRequest {
if s.canRequest() {
sendCanRequestAgain = s.sendEvent
s.sendEvent = false
}
if event.Type == EvFail {
s.fail("failed request")
s.failLocked("failed request")
}
}
childEventCb := s.childEventCb
@ -331,14 +327,14 @@ func (s *serverWithLimits) unsubscribe() {
}
// canRequest checks whether a new request can be started.
func (s *serverWithLimits) canRequest() (bool, float32) {
func (s *serverWithLimits) canRequest() bool {
if s.delayTimer != nil || s.pendingCount >= int(s.parallelLimit) || s.timeoutCount > 0 {
return false, 0
return false
}
if s.parallelLimit < minParallelLimit {
s.parallelLimit = minParallelLimit
}
return true, -(float32(s.pendingCount) + rand.Float32()) / s.parallelLimit
return true
}
// canRequestNow checks whether a new request can be started, according to the
@ -349,10 +345,10 @@ func (s *serverWithLimits) canRequest() (bool, float32) {
// 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) {
func (s *serverWithLimits) canRequestNow() bool {
var sendCanRequestAgain bool
s.lock.Lock()
canRequest, priority := s.canRequest()
canRequest := s.canRequest()
if canRequest {
sendCanRequestAgain = s.sendEvent
s.sendEvent = false
@ -362,7 +358,7 @@ func (s *serverWithLimits) canRequestNow() (bool, float32) {
if sendCanRequestAgain {
childEventCb(Event{Type: EvCanRequestAgain})
}
return canRequest, priority
return canRequest
}
// delay sets the delay timer to the given duration, disabling new requests for
@ -384,7 +380,7 @@ func (s *serverWithLimits) delay(delay time.Duration) {
s.lock.Lock()
if s.delayTimer != nil && s.delayCounter == delayCounter { // do nothing if there is a new timer now
s.delayTimer = nil
if canRequest, _ := s.canRequest(); canRequest {
if s.canRequest() {
sendCanRequestAgain = s.sendEvent
s.sendEvent = false
}
@ -399,15 +395,15 @@ 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) {
func (s *serverWithLimits) fail(desc string) {
s.lock.Lock()
defer s.lock.Unlock()
s.fail(desc)
s.failLocked(desc)
}
// fail calculates the dynamic failure delay and applies it.
func (s *serverWithLimits) fail(desc string) {
// failLocked calculates the dynamic failure delay and applies it.
func (s *serverWithLimits) failLocked(desc string) {
log.Debug("Server error", "description", desc)
s.failureDelay *= 2
now := s.clock.Now()

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

@ -67,7 +67,7 @@ func NewHeadSync(headTracker headTracker, chain committeeChain) *HeadSync {
return s
}
func (s *HeadSync) Process(events []request.Event) {
func (s *HeadSync) Process(requester request.Requester, events []request.Event) {
for _, event := range events {
switch event.Type {
case EvNewHead:
@ -90,10 +90,6 @@ func (s *HeadSync) Process(events []request.Event) {
}
}
func (s *HeadSync) MakeRequest(server request.Server) (request.Request, float32) {
return nil, 0
}
// 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) {

44
beacon/light/sync/head_sync_test.go
View file

@ -24,10 +24,10 @@ import (
)
var (
testServer1 = &TestServer{ID: 1}
testServer2 = &TestServer{ID: 2}
testServer3 = &TestServer{ID: 3}
testServer4 = &TestServer{ID: 4}
testServer1 = "testServer1"
testServer2 = "testServer2"
testServer3 = "testServer3"
testServer4 = "testServer4"
testHead0 = types.HeadInfo{}
testHead1 = types.HeadInfo{Slot: 123, BlockRoot: common.Hash{1}}
@ -52,12 +52,12 @@ func TestValidatedHead(t *testing.T) {
ts.AddServer(testServer1, 1)
ts.ServerEvent(EvNewSignedHead, testServer1, testSHead1)
ts.Run(1, nil, nil)
ts.Run(1)
// announced head should be queued because of uninitialized chain
ht.ExpValidated(t, 1, nil)
chain.SetNextSyncPeriod(0) // initialize chain
ts.Run(2, nil, nil)
ts.Run(2)
// expect previously queued head to be validated
ht.ExpValidated(t, 2, []types.SignedHeader{testSHead1})
@ -65,34 +65,34 @@ func TestValidatedHead(t *testing.T) {
ts.ServerEvent(EvNewSignedHead, testServer1, testSHead2)
ts.AddServer(testServer2, 1)
ts.ServerEvent(EvNewSignedHead, testServer2, testSHead2)
ts.Run(3, nil, nil)
ts.Run(3)
// expect both head announcements to be validated instantly
ht.ExpValidated(t, 3, []types.SignedHeader{testSHead2, testSHead2})
ts.ServerEvent(EvNewSignedHead, testServer1, testSHead3)
ts.AddServer(testServer3, 1)
ts.ServerEvent(EvNewSignedHead, testServer3, testSHead4)
ts.Run(4, nil, nil)
ts.Run(4)
// future period annonced heads should be queued
ht.ExpValidated(t, 4, nil)
chain.SetNextSyncPeriod(2)
ts.Run(5, nil, nil)
ts.Run(5)
// testSHead3 can be validated now but not testSHead4
ht.ExpValidated(t, 5, []types.SignedHeader{testSHead3})
// server 3 disconnected without proving period 3, its announced head should be dropped
ts.RemoveServer(testServer3)
ts.Run(6, nil, nil)
ts.Run(6)
ht.ExpValidated(t, 6, nil)
chain.SetNextSyncPeriod(3)
ts.Run(7, nil, nil)
ts.Run(7)
// testSHead4 could be validated now but it's not queued by any registered server
ht.ExpValidated(t, 7, nil)
ts.ServerEvent(EvNewSignedHead, testServer2, testSHead4)
ts.Run(8, nil, nil)
ts.Run(8)
// now testSHead4 should be validated
ht.ExpValidated(t, 8, []types.SignedHeader{testSHead4})
}
@ -107,45 +107,45 @@ func TestPrefetchHead(t *testing.T) {
ts.AddServer(testServer1, 1)
ts.ServerEvent(EvNewHead, testServer1, testHead1)
ts.Run(1, nil, nil)
ts.Run(1)
ht.ExpPrefetch(t, 1, testHead1) // s1: h1
ts.AddServer(testServer2, 1)
ts.ServerEvent(EvNewHead, testServer2, testHead2)
ts.Run(2, nil, nil)
ts.Run(2)
ht.ExpPrefetch(t, 2, testHead2) // s1: h1, s2: h2
ts.ServerEvent(EvNewHead, testServer1, testHead2)
ts.Run(3, nil, nil)
ts.Run(3)
ht.ExpPrefetch(t, 3, testHead2) // s1: h2, s2: h2
ts.AddServer(testServer3, 1)
ts.ServerEvent(EvNewHead, testServer3, testHead3)
ts.Run(4, nil, nil)
ts.Run(4)
ht.ExpPrefetch(t, 4, testHead2) // s1: h2, s2: h2, s3: h3
ts.AddServer(testServer4, 1)
ts.ServerEvent(EvNewHead, testServer4, testHead4)
ts.Run(5, nil, nil)
ts.Run(5)
ht.ExpPrefetch(t, 5, testHead2) // s1: h2, s2: h2, s3: h3, s4: h4
ts.ServerEvent(EvNewHead, testServer2, testHead3)
ts.Run(6, nil, nil)
ts.Run(6)
ht.ExpPrefetch(t, 6, testHead3) // s1: h2, s2: h3, s3: h3, s4: h4
ts.RemoveServer(testServer3)
ts.Run(7, nil, nil)
ts.Run(7)
ht.ExpPrefetch(t, 7, testHead4) // s1: h2, s2: h3, s4: h4
ts.RemoveServer(testServer1)
ts.Run(8, nil, nil)
ts.Run(8)
ht.ExpPrefetch(t, 8, testHead4) // s2: h3, s4: h4
ts.RemoveServer(testServer4)
ts.Run(9, nil, nil)
ts.Run(9)
ht.ExpPrefetch(t, 9, testHead3) // s2: h3
ts.RemoveServer(testServer2)
ts.Run(10, nil, nil)
ts.Run(10)
ht.ExpPrefetch(t, 10, testHead0) // no servers registered
}

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

@ -17,6 +17,7 @@
package sync
import (
"reflect"
"testing"
"github.com/ethereum/go-ethereum/beacon/light"
@ -24,15 +25,6 @@ import (
"github.com/ethereum/go-ethereum/beacon/types"
)
type TestServer struct {
ts *TestScheduler
ID int
}
func (s *TestServer) Fail(desc string) {
s.ts.serverFail(s)
}
type requestWithID struct {
sid request.ServerAndID
request request.Request
@ -44,7 +36,7 @@ type TestScheduler struct {
events []request.Event
servers []request.Server
allowance map[request.Server]int
sent map[int]requestWithID
sent map[int][]requestWithID
testIndex int
expFail map[request.Server]int // expected Server.Fail calls during next Run
lastId request.ID
@ -56,13 +48,26 @@ 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]requestWithID),
sent: make(map[int][]requestWithID),
}
}
func (ts *TestScheduler) Run(testIndex int, expServer request.Server, expReq request.Request) {
func (ts *TestScheduler) Run(testIndex int, exp ...any) {
expReqs := make([]requestWithID, len(exp)/2)
id := ts.lastId
for i := range expReqs {
id++
expReqs[i] = requestWithID{
sid: request.ServerAndID{Server: exp[i*2].(request.Server), ID: id},
request: exp[i*2+1].(request.Request),
}
}
if len(expReqs) == 0 {
expReqs = nil
}
ts.testIndex = testIndex
ts.module.Process(ts.events)
ts.module.Process(ts, ts.events)
ts.events = nil
for server, count := range ts.expFail {
@ -70,31 +75,47 @@ func (ts *TestScheduler) Run(testIndex int, expServer request.Server, expReq req
if count == 0 {
continue
}
ts.t.Errorf("Missing %d Server.Fail(s) from server %d in test case #%d", count, server.(*TestServer).ID, testIndex)
ts.t.Errorf("Missing %d Server.Fail(s) from server %s in test case #%d", count, server.(string), testIndex)
}
expReqWithID := requestWithID{
sid: request.ServerAndID{Server: expServer, ID: ts.lastId + 1},
request: expReq,
}
req, ok := ts.tryRequest(testIndex, ts.module.MakeRequest)
if expReq == nil {
if ok {
ts.t.Errorf("Unexpected request in test case #%d (expected none, got %v)", testIndex, req)
}
return
}
if !ok {
ts.t.Errorf("Missing request in test case #%d (expected %v, got none)", testIndex, expReqWithID)
return
}
if req != expReqWithID {
ts.t.Errorf("Wrong request in test case #%d (expected %v, got %v)", testIndex, expReqWithID, req)
if !reflect.DeepEqual(ts.sent[testIndex], expReqs) {
ts.t.Errorf("Wrong sent requests in test case #%d (expected %v, got %v)", testIndex, expReqs, ts.sent[testIndex])
}
}
func (ts *TestScheduler) Request(testIndex int) request.Request {
return ts.sent[testIndex].request
func (ts *TestScheduler) CanSendTo() (cs []request.Server) {
for _, server := range ts.servers {
if ts.allowance[server] > 0 {
cs = append(cs, server)
}
}
return
}
func (ts *TestScheduler) Send(server request.Server, req request.Request) request.ID {
ts.lastId++
ts.sent[ts.testIndex] = append(ts.sent[ts.testIndex], requestWithID{
sid: request.ServerAndID{Server: server, ID: ts.lastId},
request: req,
})
ts.allowance[server]--
return ts.lastId
}
func (ts *TestScheduler) Fail(server request.Server, desc string) {
if ts.expFail[server] == 0 {
ts.t.Errorf("Unexpected Fail from server %s in test case #%d: %s", server.(string), ts.testIndex, desc)
return
}
ts.expFail[server]--
}
func (ts *TestScheduler) Request(testIndex, reqIndex int) requestWithID {
if len(ts.sent[testIndex]) < reqIndex {
ts.t.Errorf("Missing request from test case %d index %d", testIndex, reqIndex)
return requestWithID{}
}
return ts.sent[testIndex][reqIndex-1]
}
func (ts *TestScheduler) ServerEvent(evType *request.EventType, server request.Server, data any) {
@ -105,10 +126,8 @@ func (ts *TestScheduler) ServerEvent(evType *request.EventType, server request.S
})
}
func (ts *TestScheduler) RequestEvent(evType *request.EventType, testIndex int, resp request.Response) {
req, ok := ts.sent[testIndex]
if !ok {
ts.t.Errorf("Missing request from test case %v", testIndex)
func (ts *TestScheduler) RequestEvent(evType *request.EventType, req requestWithID, resp request.Response) {
if req.request == nil {
return
}
ts.events = append(ts.events, request.Event{
@ -123,7 +142,6 @@ func (ts *TestScheduler) RequestEvent(evType *request.EventType, testIndex int,
}
func (ts *TestScheduler) AddServer(server request.Server, allowance int) {
server.(*TestServer).ts = ts
ts.servers = append(ts.servers, server)
ts.allowance[server] = allowance
ts.ServerEvent(request.EvRegistered, server, nil)
@ -150,43 +168,6 @@ func (ts *TestScheduler) ExpFail(server request.Server) {
ts.expFail[server]++
}
func (ts *TestScheduler) serverFail(server request.Server) {
if ts.expFail[server] == 0 {
ts.t.Errorf("Unexpected Server.Fail from server %d in test case #%d", server.(*TestServer).ID, ts.testIndex)
return
}
ts.expFail[server]--
}
func (ts *TestScheduler) tryRequest(testIndex int, requestFn func(server request.Server) (request.Request, float32)) (requestWithID, bool) {
var (
bestServer request.Server
bestReq request.Request
bestPri float32
)
for _, server := range ts.servers {
if ts.allowance[server] == 0 {
continue
}
req, pri := requestFn(server)
if req != nil && (bestReq == nil || pri > bestPri) {
bestServer, bestReq, bestPri = server, req, pri
}
}
if bestServer == nil {
return requestWithID{}, false
}
ts.allowance[bestServer]--
ts.lastId++
req := requestWithID{
sid: request.ServerAndID{Server: bestServer, ID: ts.lastId},
request: bestReq,
}
ts.sent[testIndex] = req
ts.RequestEvent(request.EvRequest, testIndex, nil)
return req, true
}
type TestCommitteeChain struct {
fsp, nsp uint64
init bool

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

@ -52,16 +52,12 @@ func NewCheckpointInit(chain committeeChain, checkpointHash common.Hash) *Checkp
}
}
func (s *CheckpointInit) Process(events []request.Event) {
func (s *CheckpointInit) Process(requester request.Requester, events []request.Event) {
for _, event := range events {
if !event.IsRequestEvent() {
continue
}
sid, req, resp := event.RequestInfo()
if event.Type == request.EvRequest {
s.locked = sid
continue
}
if s.locked == sid {
s.locked = request.ServerAndID{}
}
@ -71,16 +67,21 @@ func (s *CheckpointInit) Process(events []request.Event) {
s.initialized = true
return
}
event.Server.Fail("invalid checkpoint data")
requester.Fail(event.Server, "invalid checkpoint data")
}
}
}
func (s *CheckpointInit) MakeRequest(server request.Server) (request.Request, float32) {
// start a request if possible
if s.initialized || s.locked != (request.ServerAndID{}) {
return nil, 0
return
}
return ReqCheckpointData(s.checkpointHash), 0
cs := requester.CanSendTo()
if len(cs) == 0 {
return
}
server := cs[0]
id := requester.Send(server, ReqCheckpointData(s.checkpointHash))
s.locked = request.ServerAndID{Server: server, ID: id}
}
// ForwardUpdateSync implements request.Module; it fetches updates between the
@ -194,12 +195,9 @@ func (u updateResponseList) Less(i, j int) bool {
return u[i].request.FirstPeriod < u[j].request.FirstPeriod
}
func (s *ForwardUpdateSync) Process(events []request.Event) {
func (s *ForwardUpdateSync) Process(requester request.Requester, events []request.Event) {
for _, event := range events {
switch event.Type {
case request.EvRequest:
sid, req, _ := event.RequestInfo()
s.lockRange(sid, req.(ReqUpdates))
case request.EvResponse, request.EvFail, request.EvTimeout:
sid, rq, rs := event.RequestInfo()
req := rq.(ReqUpdates)
@ -212,7 +210,7 @@ func (s *ForwardUpdateSync) Process(events []request.Event) {
s.lockRange(sid, req)
queued = true
} else {
event.Server.Fail("invalid update range")
requester.Fail(event.Server, "invalid update range")
}
}
if !queued {
@ -230,8 +228,8 @@ func (s *ForwardUpdateSync) Process(events []request.Event) {
sort.Sort(updateResponseList(s.processQueue))
for s.processQueue != nil {
u := s.processQueue[0]
if !s.processResponse(u) {
return
if !s.processResponse(requester, u) {
break
}
s.unlockRange(u.sid, u.request)
s.processQueue = s.processQueue[1:]
@ -239,11 +237,43 @@ func (s *ForwardUpdateSync) Process(events []request.Event) {
s.processQueue = nil
}
}
// start new requests if possible
startPeriod, chainInit := s.chain.NextSyncPeriod()
if !chainInit {
return
}
for {
firstPeriod, maxCount := s.rangeLock.firstUnlocked(startPeriod, maxUpdateRequest)
var (
sendTo request.Server
bestCount uint64
)
for _, server := range requester.CanSendTo() {
nextPeriod := s.nextSyncPeriod[server]
if nextPeriod <= firstPeriod {
continue
}
count := maxCount
if nextPeriod < firstPeriod+maxCount {
count = nextPeriod - firstPeriod
}
if count > bestCount {
sendTo, bestCount = server, count
}
}
if sendTo == nil {
return
}
req := ReqUpdates{FirstPeriod: firstPeriod, Count: bestCount}
id := requester.Send(sendTo, req)
s.lockRange(request.ServerAndID{Server: sendTo, ID: id}, req)
}
}
// processResponse adds the fetched updates and committees to the committee chain.
// Returns true in case of full or partial success.
func (s *ForwardUpdateSync) processResponse(u updateResponse) (success bool) {
func (s *ForwardUpdateSync) processResponse(requester request.Requester, u updateResponse) (success bool) {
for i, update := range u.response.Updates {
if err := s.chain.InsertUpdate(update, u.response.Committees[i]); err != nil {
if err == light.ErrInvalidPeriod {
@ -252,7 +282,7 @@ func (s *ForwardUpdateSync) processResponse(u updateResponse) (success bool) {
return
}
if err == light.ErrInvalidUpdate || err == light.ErrWrongCommitteeRoot || err == light.ErrCannotReorg {
u.sid.Server.Fail("invalid update received")
requester.Fail(u.sid.Server, "invalid update received")
} else {
log.Error("Unexpected InsertUpdate error", "error", err)
}
@ -262,20 +292,3 @@ func (s *ForwardUpdateSync) processResponse(u updateResponse) (success bool) {
}
return
}
func (s *ForwardUpdateSync) MakeRequest(server request.Server) (request.Request, float32) {
startPeriod, chainInit := s.chain.NextSyncPeriod()
if !chainInit {
return nil, 0
}
firstPeriod, maxCount := s.rangeLock.firstUnlocked(startPeriod, maxUpdateRequest)
nextPeriod := s.nextSyncPeriod[server]
if nextPeriod <= firstPeriod {
return nil, 0
}
count := maxCount
if nextPeriod < firstPeriod+maxCount {
count = nextPeriod - firstPeriod
}
return ReqUpdates{FirstPeriod: firstPeriod, Count: count}, float32(count)
}

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

@ -37,18 +37,18 @@ func TestCheckpointInit(t *testing.T) {
ts.Run(1, testServer1, ReqCheckpointData(checkpointHash))
// server 1 times out; expect request to server 2
ts.RequestEvent(request.EvTimeout, 1, nil)
ts.RequestEvent(request.EvTimeout, ts.Request(1, 1), nil)
ts.Run(2, testServer2, ReqCheckpointData(checkpointHash))
// invalid response from server 2; expect init state to still be false
ts.RequestEvent(request.EvResponse, 2, &types.BootstrapData{Header: types.Header{Slot: 123456}})
ts.RequestEvent(request.EvResponse, ts.Request(2, 1), &types.BootstrapData{Header: types.Header{Slot: 123456}})
ts.ExpFail(testServer2)
ts.Run(3, nil, nil)
ts.Run(3)
chain.ExpInit(t, false)
// server 1 fails (hard timeout)
ts.RequestEvent(request.EvFail, 1, nil)
ts.Run(4, nil, nil)
ts.RequestEvent(request.EvFail, ts.Request(1, 1), nil)
ts.Run(4)
chain.ExpInit(t, false)
// server 3 is registered; expect bootstrap request to server 3
@ -56,8 +56,8 @@ func TestCheckpointInit(t *testing.T) {
ts.Run(5, testServer3, ReqCheckpointData(checkpointHash))
// valid response from server 3; expect chain to be initialized
ts.RequestEvent(request.EvResponse, 5, checkpoint)
ts.Run(6, nil, nil)
ts.RequestEvent(request.EvResponse, ts.Request(5, 1), checkpoint)
ts.Run(6)
chain.ExpInit(t, true)
}
@ -73,68 +73,72 @@ func TestUpdateSyncParallel(t *testing.T) {
ts.ServerEvent(EvNewSignedHead, testServer2, types.SignedHeader{SignatureSlot: 0x2000*100 + 0x1000})
// expect 6 requests to be sent
ts.Run(1, testServer1, ReqUpdates{FirstPeriod: 0, Count: 8})
ts.Run(2, testServer1, ReqUpdates{FirstPeriod: 8, Count: 8})
ts.Run(3, testServer1, ReqUpdates{FirstPeriod: 16, Count: 8})
ts.Run(4, testServer2, ReqUpdates{FirstPeriod: 24, Count: 8})
ts.Run(5, testServer2, ReqUpdates{FirstPeriod: 32, Count: 8})
ts.Run(6, testServer2, ReqUpdates{FirstPeriod: 40, Count: 8})
ts.Run(1,
testServer1, ReqUpdates{FirstPeriod: 0, Count: 8},
testServer1, ReqUpdates{FirstPeriod: 8, Count: 8},
testServer1, ReqUpdates{FirstPeriod: 16, Count: 8},
testServer2, ReqUpdates{FirstPeriod: 24, Count: 8},
testServer2, ReqUpdates{FirstPeriod: 32, Count: 8},
testServer2, ReqUpdates{FirstPeriod: 40, Count: 8})
// valid response to request 1; expect 8 periods synced and a new request started
ts.RequestEvent(request.EvResponse, 1, testRespUpdate(ts.Request(1)))
ts.RequestEvent(request.EvResponse, ts.Request(1, 1), testRespUpdate(ts.Request(1, 1)))
ts.AddAllowance(testServer1, 1)
ts.Run(7, testServer1, ReqUpdates{FirstPeriod: 48, Count: 8})
chain.ExpNextSyncPeriod(t, 8)
// valid response to requests 4 and 5
ts.RequestEvent(request.EvResponse, 4, testRespUpdate(ts.Request(4)))
ts.RequestEvent(request.EvResponse, 5, testRespUpdate(ts.Request(5)))
ts.RequestEvent(request.EvResponse, ts.Request(1, 4), testRespUpdate(ts.Request(1, 4)))
ts.RequestEvent(request.EvResponse, ts.Request(1, 5), testRespUpdate(ts.Request(1, 5)))
ts.AddAllowance(testServer2, 2)
// expect 2 more requests but no sync progress (responses 4 and 5 cannot be added before 2 and 3)
ts.Run(8, testServer2, ReqUpdates{FirstPeriod: 56, Count: 8})
ts.Run(9, testServer2, ReqUpdates{FirstPeriod: 64, Count: 8})
ts.Run(8,
testServer2, ReqUpdates{FirstPeriod: 56, Count: 8},
testServer2, ReqUpdates{FirstPeriod: 64, Count: 8})
chain.ExpNextSyncPeriod(t, 8)
// soft timeout for requests 2 and 3 (server 1 is overloaded)
ts.RequestEvent(request.EvTimeout, 2, nil)
ts.RequestEvent(request.EvTimeout, 3, nil)
ts.RequestEvent(request.EvTimeout, ts.Request(1, 2), nil)
ts.RequestEvent(request.EvTimeout, ts.Request(1, 3), nil)
// no allowance, no more requests
ts.Run(10, nil, nil)
ts.Run(10)
// valid response to requests 6 and 8 and 9
ts.RequestEvent(request.EvResponse, 6, testRespUpdate(ts.Request(6)))
ts.RequestEvent(request.EvResponse, 8, testRespUpdate(ts.Request(8)))
ts.RequestEvent(request.EvResponse, 9, testRespUpdate(ts.Request(9)))
ts.RequestEvent(request.EvResponse, ts.Request(1, 6), testRespUpdate(ts.Request(1, 6)))
ts.RequestEvent(request.EvResponse, ts.Request(8, 1), testRespUpdate(ts.Request(8, 1)))
ts.RequestEvent(request.EvResponse, ts.Request(8, 2), testRespUpdate(ts.Request(8, 2)))
ts.AddAllowance(testServer2, 3)
// server 2 can now resend requests 2 and 3 (timed out by server 1) and also send a new one
ts.Run(11, testServer2, ReqUpdates{FirstPeriod: 8, Count: 8})
ts.Run(12, testServer2, ReqUpdates{FirstPeriod: 16, Count: 8})
ts.Run(13, testServer2, ReqUpdates{FirstPeriod: 72, Count: 8})
ts.Run(11,
testServer2, ReqUpdates{FirstPeriod: 8, Count: 8},
testServer2, ReqUpdates{FirstPeriod: 16, Count: 8},
testServer2, ReqUpdates{FirstPeriod: 72, Count: 8})
// server 1 finally answers timed out request 2
ts.RequestEvent(request.EvResponse, 2, testRespUpdate(ts.Request(2)))
ts.RequestEvent(request.EvResponse, ts.Request(1, 2), testRespUpdate(ts.Request(1, 2)))
ts.AddAllowance(testServer1, 1)
// expect sync progress and one new request
ts.Run(14, testServer1, ReqUpdates{FirstPeriod: 80, Count: 8})
chain.ExpNextSyncPeriod(t, 16)
// server 2 answers requests 11 and 12 (resends of requests 2 and 3)
ts.RequestEvent(request.EvResponse, 11, testRespUpdate(ts.Request(11)))
ts.RequestEvent(request.EvResponse, 12, testRespUpdate(ts.Request(12)))
ts.RequestEvent(request.EvResponse, ts.Request(11, 1), testRespUpdate(ts.Request(11, 1)))
ts.RequestEvent(request.EvResponse, ts.Request(11, 2), testRespUpdate(ts.Request(11, 2)))
ts.AddAllowance(testServer2, 2)
ts.Run(15, testServer2, ReqUpdates{FirstPeriod: 88, Count: 8})
ts.Run(16, testServer2, ReqUpdates{FirstPeriod: 96, Count: 4})
ts.Run(15,
testServer2, ReqUpdates{FirstPeriod: 88, Count: 8},
testServer2, ReqUpdates{FirstPeriod: 96, Count: 4})
// finally the gap is filled, update can process responses up to req6
chain.ExpNextSyncPeriod(t, 48)
// all remaining requests are answered
ts.RequestEvent(request.EvResponse, 3, testRespUpdate(ts.Request(3)))
ts.RequestEvent(request.EvResponse, 7, testRespUpdate(ts.Request(7)))
ts.RequestEvent(request.EvResponse, 13, testRespUpdate(ts.Request(13)))
ts.RequestEvent(request.EvResponse, 14, testRespUpdate(ts.Request(14)))
ts.RequestEvent(request.EvResponse, 15, testRespUpdate(ts.Request(15)))
ts.RequestEvent(request.EvResponse, 16, testRespUpdate(ts.Request(16)))
ts.Run(17, nil, nil)
ts.RequestEvent(request.EvResponse, ts.Request(1, 3), testRespUpdate(ts.Request(1, 3)))
ts.RequestEvent(request.EvResponse, ts.Request(7, 1), testRespUpdate(ts.Request(7, 1)))
ts.RequestEvent(request.EvResponse, ts.Request(11, 3), testRespUpdate(ts.Request(11, 3)))
ts.RequestEvent(request.EvResponse, ts.Request(14, 1), testRespUpdate(ts.Request(14, 1)))
ts.RequestEvent(request.EvResponse, ts.Request(15, 1), testRespUpdate(ts.Request(15, 1)))
ts.RequestEvent(request.EvResponse, ts.Request(15, 2), testRespUpdate(ts.Request(15, 2)))
ts.Run(17)
// expect chain to be fully synced
chain.ExpNextSyncPeriod(t, 100)
}
@ -156,30 +160,32 @@ func TestUpdateSyncDifferentHeads(t *testing.T) {
ts.Run(1, testServer3, ReqUpdates{FirstPeriod: 10, Count: 7})
// request times out, expect request to the next best head
ts.RequestEvent(request.EvTimeout, 1, nil)
ts.RequestEvent(request.EvTimeout, ts.Request(1, 1), nil)
ts.Run(2, testServer2, ReqUpdates{FirstPeriod: 10, Count: 6})
// request times out, expect request to the last available server
ts.RequestEvent(request.EvTimeout, 2, nil)
ts.RequestEvent(request.EvTimeout, ts.Request(2, 1), nil)
ts.Run(3, testServer1, ReqUpdates{FirstPeriod: 10, Count: 5})
// valid response to request 3, expect chain synced to period 15
ts.RequestEvent(request.EvResponse, 3, testRespUpdate(ts.Request(3)))
ts.RequestEvent(request.EvResponse, ts.Request(3, 1), testRespUpdate(ts.Request(3, 1)))
ts.AddAllowance(testServer1, 1)
ts.Run(4, nil, nil)
ts.Run(4)
chain.ExpNextSyncPeriod(t, 15)
// invalid response to request 1, server can only deliver updates up to period 15 despite announced head
ts.RequestEvent(request.EvResponse, 1, testRespUpdate(ReqUpdates{FirstPeriod: 10, Count: 5}))
truncated := ts.Request(1, 1)
truncated.request = ReqUpdates{FirstPeriod: 10, Count: 5}
ts.RequestEvent(request.EvResponse, ts.Request(1, 1), testRespUpdate(truncated))
ts.ExpFail(testServer3)
ts.Run(5, nil, nil)
ts.Run(5)
// expect no progress of chain head
chain.ExpNextSyncPeriod(t, 15)
// valid response to request 2, expect chain synced to period 16
ts.RequestEvent(request.EvResponse, 2, testRespUpdate(ts.Request(2)))
ts.RequestEvent(request.EvResponse, ts.Request(2, 1), testRespUpdate(ts.Request(2, 1)))
ts.AddAllowance(testServer2, 1)
ts.Run(6, nil, nil)
ts.Run(6)
chain.ExpNextSyncPeriod(t, 16)
// a new server is registered with announced head period 17
@ -189,15 +195,18 @@ func TestUpdateSyncDifferentHeads(t *testing.T) {
ts.Run(7, testServer4, ReqUpdates{FirstPeriod: 16, Count: 1})
// valid response, expect chain synced to period 17
ts.RequestEvent(request.EvResponse, 7, testRespUpdate(ts.Request(7)))
ts.RequestEvent(request.EvResponse, ts.Request(7, 1), testRespUpdate(ts.Request(7, 1)))
ts.AddAllowance(testServer4, 1)
ts.Run(8, nil, nil)
ts.Run(8)
chain.ExpNextSyncPeriod(t, 17)
}
func testRespUpdate(request request.Request) request.Response {
func testRespUpdate(request requestWithID) request.Response {
var resp RespUpdates
req := request.(ReqUpdates)
if request.request == nil {
return resp
}
req := request.request.(ReqUpdates)
resp.Updates = make([]*types.LightClientUpdate, int(req.Count))
resp.Committees = make([]*types.SerializedSyncCommittee, int(req.Count))
period := req.FirstPeriod

59
cmd/blsync/block_sync.go
View file

@ -29,7 +29,7 @@ import (
// to the validated and prefetch heads.
type beaconBlockSync struct {
recentBlocks *lru.Cache[common.Hash, *capella.BeaconBlock]
locked map[common.Hash]struct{}
locked map[common.Hash]request.ServerAndID
serverHeads map[request.Server]common.Hash
headTracker headTracker
@ -53,34 +53,42 @@ func newBeaconBlockSync(headTracker headTracker) *beaconBlockSync {
return &beaconBlockSync{
headTracker: headTracker,
recentBlocks: lru.NewCache[common.Hash, *capella.BeaconBlock](10),
locked: make(map[common.Hash]struct{}),
locked: make(map[common.Hash]request.ServerAndID),
serverHeads: make(map[request.Server]common.Hash),
headCh: make(chan headData, 1),
}
}
func (s *beaconBlockSync) Process(events []request.Event) {
func (s *beaconBlockSync) Process(requester request.Requester, 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()
sid, req, resp := event.RequestInfo()
blockRoot := common.Hash(req.(sync.ReqBeaconBlock))
if resp != nil {
s.recentBlocks.Add(blockRoot, resp.(*capella.BeaconBlock))
}
delete(s.locked, blockRoot)
if s.locked[blockRoot] == sid {
delete(s.locked, blockRoot)
}
case sync.EvNewHead:
s.serverHeads[event.Server] = event.Data.(types.HeadInfo).BlockRoot
case request.EvUnregistered:
delete(s.serverHeads, event.Server)
}
}
s.updateValidatedHead()
// request validated head block if unavailable and not yet requested
if vh, ok := s.headTracker.ValidatedHead(); ok {
s.tryRequestBlock(requester, vh.Header.Hash(), false)
}
// request prefetch head if the given server has announced it
if prefetchHead := s.headTracker.PrefetchHead().BlockRoot; prefetchHead != (common.Hash{}) {
s.tryRequestBlock(requester, prefetchHead, true)
}
}
// send validated head block
func (s *beaconBlockSync) updateValidatedHead() {
head, ok := s.headTracker.ValidatedHead()
if !ok {
return
@ -105,27 +113,22 @@ func (s *beaconBlockSync) Process(events []request.Event) {
}
}
func (s *beaconBlockSync) MakeRequest(server request.Server) (request.Request, float32) {
// request validated head block if unavailable and not yet requested
if vh, ok := s.headTracker.ValidatedHead(); ok {
validatedHead := vh.Header.Hash()
if _, ok := s.recentBlocks.Get(validatedHead); !ok {
if _, ok := s.locked[validatedHead]; !ok {
return sync.ReqBeaconBlock(validatedHead), 1
}
}
func (s *beaconBlockSync) tryRequestBlock(requester request.Requester, blockRoot common.Hash, needSameHead bool) {
if _, ok := s.recentBlocks.Get(blockRoot); ok {
return
}
// request prefetch head if the given server has announced it
if prefetchHead := s.headTracker.PrefetchHead().BlockRoot; prefetchHead != (common.Hash{}) && prefetchHead == s.serverHeads[server] {
if _, ok := s.recentBlocks.Get(prefetchHead); !ok {
if _, ok := s.locked[prefetchHead]; !ok {
return sync.ReqBeaconBlock(prefetchHead), 0
}
}
if _, ok := s.locked[blockRoot]; ok {
return
}
for _, server := range requester.CanSendTo() {
if needSameHead && (s.serverHeads[server] != blockRoot) {
continue
}
id := requester.Send(server, sync.ReqBeaconBlock(blockRoot))
s.locked[blockRoot] = request.ServerAndID{Server: server, ID: id}
return
}
return nil, 0
}
func blockHeadInfo(block *capella.BeaconBlock) types.HeadInfo {
if block == nil {
return types.HeadInfo{}

20
cmd/blsync/block_sync_test.go
View file

@ -29,8 +29,8 @@ import (
)
var (
testServer1 = &sync.TestServer{ID: 1}
testServer2 = &sync.TestServer{ID: 2}
testServer1 = "testServer1"
testServer2 = "testServer2"
testBlock1 = &capella.BeaconBlock{Slot: 123}
testBlock2 = &capella.BeaconBlock{Slot: 124}
@ -57,7 +57,7 @@ func TestBlockSync(t *testing.T) {
}
// no block requests expected until head tracker knows about a head
ts.Run(1, nil, nil)
ts.Run(1)
expHeadBlock(1, nil)
// set block 1 as prefetch head, announced by server 2
@ -68,15 +68,15 @@ func TestBlockSync(t *testing.T) {
ts.Run(2, testServer2, sync.ReqBeaconBlock(head1.BlockRoot))
// valid response
ts.RequestEvent(request.EvResponse, 2, testBlock1)
ts.RequestEvent(request.EvResponse, ts.Request(2, 1), testBlock1)
ts.AddAllowance(testServer2, 1)
ts.Run(3, nil, nil)
ts.Run(3)
// head block still not expected as the fetched block is not the validated head yet
expHeadBlock(3, nil)
// set as validated head, expect no further requests but block 1 set as head block
ht.validated.Header = blockHeader(testBlock1)
ts.Run(4, nil, nil)
ts.Run(4)
expHeadBlock(4, testBlock1)
// set block 2 as prefetch head, announced by server 1
@ -87,8 +87,8 @@ func TestBlockSync(t *testing.T) {
ts.Run(5, testServer1, sync.ReqBeaconBlock(head2.BlockRoot))
// req2 fails, no further requests expected because server 2 has not announced it
ts.RequestEvent(request.EvFail, 5, nil)
ts.Run(6, nil, nil)
ts.RequestEvent(request.EvFail, ts.Request(5, 1), nil)
ts.Run(6)
// set as validated head before retrieving block; now it's assumed to be available from server 2 too
ht.validated.Header = blockHeader(testBlock2)
@ -98,8 +98,8 @@ func TestBlockSync(t *testing.T) {
expHeadBlock(4, nil)
// valid response, now head block should be block 2 immediately as it is already validated
ts.RequestEvent(request.EvResponse, 7, testBlock2)
ts.Run(8, nil, nil)
ts.RequestEvent(request.EvResponse, ts.Request(7, 1), testBlock2)
ts.Run(8)
expHeadBlock(5, testBlock2)
}