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: use explicit request/response types where possible

Browse source
This commit is contained in:
Zsolt Felfoldi 2024-01-26 02:03:00 +01:00 committed by Felix Lange
parent 245d166d45
commit 084f23cfa0
3 changed files with 55 additions and 66 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -360,11 +360,7 @@ func (s *Scheduler) filterEvents() map[Module][]Event {
filteredEvents := make(map[Module][]Event)
for _, event := range events {
server, ok := event.Server.(server)
if !ok {
log.Error("Server interface type unknown for Scheduler")
continue
}
server := event.Server.(server)
if _, ok := s.servers[server]; !ok && event.Type != EvRegistered {
continue // before EvRegister or after EvUnregister, discard
}

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

@ -66,7 +66,7 @@ func (s *CheckpointInit) Process(events []request.Event) {
s.locked = request.ServerAndID{}
}
if resp != nil {
if checkpoint, ok := resp.(*types.BootstrapData); ok && checkpoint.Header.Hash() == common.Hash(req.(ReqCheckpointData)) {
if checkpoint := resp.(*types.BootstrapData); checkpoint.Header.Hash() == common.Hash(req.(ReqCheckpointData)) {
s.chain.CheckpointInit(*checkpoint)
s.initialized = true
return
@ -93,7 +93,7 @@ type ForwardUpdateSync struct {
chain committeeChain
rangeLock rangeLock
lockedIDs map[request.ServerAndID]struct{}
processQueue []request.Event
processQueue []updateResponse
nextSyncPeriod map[request.Server]uint64
}
@ -147,37 +147,27 @@ func (r rangeLock) firstUnlocked(start, maxCount uint64) (first, count uint64) {
// 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) {
func (s *ForwardUpdateSync) lockRange(sid request.ServerAndID, req ReqUpdates) {
if _, ok := s.lockedIDs[sid]; ok {
return
}
s.lockedIDs[sid] = struct{}{}
r := req.(ReqUpdates)
s.rangeLock.lock(r.FirstPeriod, r.Count, 1)
s.rangeLock.lock(req.FirstPeriod, req.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) {
func (s *ForwardUpdateSync) unlockRange(sid request.ServerAndID, req ReqUpdates) {
if _, ok := s.lockedIDs[sid]; !ok {
return
}
delete(s.lockedIDs, sid)
r := req.(ReqUpdates)
s.rangeLock.lock(r.FirstPeriod, r.Count, -1)
s.rangeLock.lock(req.FirstPeriod, req.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 {
return false
}
response, ok := resp.(RespUpdates)
if !ok {
return false
}
func (s *ForwardUpdateSync) verifyRange(request ReqUpdates, response RespUpdates) bool {
if uint64(len(response.Updates)) != request.Count || uint64(len(response.Committees)) != request.Count {
return false
}
@ -189,41 +179,19 @@ func (s *ForwardUpdateSync) verifyRange(req request.Request, resp request.Respon
return true
}
// processResponse adds the fetched updates and committees to the committee chain.
// Returns true in case of full or partial success.
func (s *ForwardUpdateSync) processResponse(event request.Event) (success bool) {
_, _, resp := event.RequestInfo()
response, ok := resp.(RespUpdates)
if !ok {
return false
}
for i, update := range response.Updates {
if err := s.chain.InsertUpdate(update, response.Committees[i]); err != nil {
if err == light.ErrInvalidPeriod {
// there is a gap in the update periods; stop processing without
// failing and try again next time
return
}
if err == light.ErrInvalidUpdate || err == light.ErrWrongCommitteeRoot || err == light.ErrCannotReorg {
event.Server.Fail("invalid update received")
} else {
log.Error("Unexpected InsertUpdate error", "error", err)
}
return
}
success = true
}
return
type updateResponse struct {
sid request.ServerAndID
request ReqUpdates
response RespUpdates
}
// updateResponseList implements sort.Sort and sorts update request/response events by FirstPeriod.
type updateResponseList []request.Event
type updateResponseList []updateResponse
func (u updateResponseList) Len() int { return len(u) }
func (u updateResponseList) Swap(i, j int) { u[i], u[j] = u[j], u[i] }
func (u updateResponseList) Less(i, j int) bool {
return u[i].Data.(request.RequestResponse).Request.(ReqUpdates).FirstPeriod <
u[j].Data.(request.RequestResponse).Request.(ReqUpdates).FirstPeriod
return u[i].request.FirstPeriod < u[j].request.FirstPeriod
}
func (s *ForwardUpdateSync) Process(events []request.Event) {
@ -231,18 +199,23 @@ func (s *ForwardUpdateSync) Process(events []request.Event) {
switch event.Type {
case request.EvRequest:
sid, req, _ := event.RequestInfo()
s.lockRange(sid, req)
s.lockRange(sid, req.(ReqUpdates))
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
sid, rq, rs := event.RequestInfo()
req := rq.(ReqUpdates)
var queued bool
if event.Type == request.EvResponse {
resp := rs.(RespUpdates)
if s.verifyRange(req, resp) {
// there is a response with a valid format; put it in the process queue
s.processQueue = append(s.processQueue, updateResponse{sid: sid, request: req, response: resp})
s.lockRange(sid, req)
queued = true
} else {
event.Server.Fail("invalid update range")
}
}
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 {
if !queued {
s.unlockRange(sid, req)
}
case EvNewSignedHead:
@ -256,12 +229,11 @@ func (s *ForwardUpdateSync) Process(events []request.Event) {
// try processing ordered list of available responses
sort.Sort(updateResponseList(s.processQueue))
for s.processQueue != nil {
event := s.processQueue[0]
if !s.processResponse(event) {
u := s.processQueue[0]
if !s.processResponse(u) {
return
}
sid, req, _ := event.RequestInfo()
s.unlockRange(sid, req)
s.unlockRange(u.sid, u.request)
s.processQueue = s.processQueue[1:]
if len(s.processQueue) == 0 {
s.processQueue = nil
@ -269,6 +241,28 @@ func (s *ForwardUpdateSync) Process(events []request.Event) {
}
}
// 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) {
for i, update := range u.response.Updates {
if err := s.chain.InsertUpdate(update, u.response.Committees[i]); err != nil {
if err == light.ErrInvalidPeriod {
// there is a gap in the update periods; stop processing without
// failing and try again next time
return
}
if err == light.ErrInvalidUpdate || err == light.ErrWrongCommitteeRoot || err == light.ErrCannotReorg {
u.sid.Server.Fail("invalid update received")
} else {
log.Error("Unexpected InsertUpdate error", "error", err)
}
return
}
success = true
}
return
}
func (s *ForwardUpdateSync) MakeRequest(server request.Server) (request.Request, float32) {
startPeriod, chainInit := s.chain.NextSyncPeriod()
if !chainInit {

3
cmd/blsync/block_sync.go
View file

@ -70,8 +70,7 @@ func (s *beaconBlockSync) Process(events []request.Event) {
_, req, resp := event.RequestInfo()
blockRoot := common.Hash(req.(sync.ReqBeaconBlock))
if resp != nil {
block := resp.(*capella.BeaconBlock)
s.recentBlocks.Add(blockRoot, block)
s.recentBlocks.Add(blockRoot, resp.(*capella.BeaconBlock))
}
delete(s.locked, blockRoot)
case sync.EvNewHead: