From 287fb1e583bb487e520b5744a24bc849c69cffb7 Mon Sep 17 00:00:00 2001 From: zsfelfoldi Date: Thu, 7 Dec 2023 14:17:45 +0100 Subject: [PATCH] beacon/light/request: general request framework --- beacon/light/request/scheduler.go | 183 +++++++++++++++++++ beacon/light/request/server.go | 294 ++++++++++++++++++++++++++++++ 2 files changed, 477 insertions(+) create mode 100644 beacon/light/request/scheduler.go create mode 100644 beacon/light/request/server.go diff --git a/beacon/light/request/scheduler.go b/beacon/light/request/scheduler.go new file mode 100644 index 0000000000..1be6d8e918 --- /dev/null +++ b/beacon/light/request/scheduler.go @@ -0,0 +1,183 @@ +// Copyright 2023 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package request + +import ( + "sync" + + "github.com/ethereum/go-ethereum/common/mclock" +) + +// Module represents an update mechanism which is typically responsible for a +// passive data structure or a certain aspect of it. When registered to a Scheduler, +// it can be triggered either by server events, other modules or itself. +type Module interface { + // Process is a non-blocking function that is called whenever the module is + // triggered. It can start network requests through the received Environment + // and/or do other data processing tasks. If triggers are set up correctly, + // Process is eventually called whenever it might have something new to do + // either because the data structures have been changed or because new servers + // became available or new requests became available at existing ones. + // + // 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(ServerSet, []ServerEvent) bool +} + +// Scheduler is a modular network data retrieval framework that coordinates multiple +// servers and retrieval mechanisms (modules). It implements a trigger mechanism +// that calls the Process function of registered modules whenever either the state +// of existing data structures or connected servers could allow new operations. +type Scheduler struct { + lock sync.Mutex + clock mclock.Clock + modules []Module // first has highest priority + servers map[Server]struct{} + stopCh chan chan struct{} + + triggerCh chan struct{} // restarts waiting sync loop + // testWaitCh chan struct{} // accepts sends when sync loop is waiting + // testTimerResults []bool // true is appended when simulated timer is processed; false when stopped + events []ServerEvent +} + +type ServerEvent struct { + Server Server + Event Event +} + +// NewScheduler creates a new Scheduler. +func NewScheduler(clock mclock.Clock) *Scheduler { + s := &Scheduler{ + clock: clock, + servers: make(map[Server]struct{}), + stopCh: make(chan chan struct{}), + // Note: testWaitCh should not have capacity in order to ensure + // that after a trigger happens testWaitCh will block until the resulting + // processing round has been finished + triggerCh: make(chan struct{}, 1), + //testWaitCh: make(chan struct{}), + } + return s +} + +// RegisterModule registers a module. Should be called before starting the scheduler. +// In each processing round the order of module processing depends on the order of +// registration. +func (s *Scheduler) RegisterModule(m Module) { + s.lock.Lock() + defer s.lock.Unlock() + + s.modules = append(s.modules, m) +} + +// RegisterServer registers a new server. +func (s *Scheduler) RegisterServer(server Server) { + s.lock.Lock() + defer s.lock.Unlock() + + server.Subscribe(func(event Event) { + s.handleEvent(server, event) + }) + s.servers[server] = struct{}{} + s.Trigger() +} + +// UnregisterServer removes a registered server. +func (s *Scheduler) UnregisterServer(server Server) { + s.lock.Lock() + defer s.lock.Unlock() + + server.Unsubscribe() + delete(s.servers, server) +} + +// Start starts the scheduler. It should be called after registering all modules +// and before registering any servers. +func (s *Scheduler) Start() { + go s.syncLoop() +} + +// Stop stops the scheduler. +func (s *Scheduler) Stop() { + s.lock.Lock() + for server, _ := range s.servers { + server.Unsubscribe() + } + s.servers = nil + s.lock.Unlock() + stop := make(chan struct{}) + s.stopCh <- stop + <-stop +} + +// syncLoop calls all processable modules in the order of their registration. +// A round of processing starts whenever there is at least one processable module. +// Triggers triggered during a processing round do not affect the current round +// but ensure that there is going to be a next round. +func (s *Scheduler) syncLoop() { + for { + s.lock.Lock() + s.processModules() + s.lock.Unlock() + + loop: + for { + select { + case stop := <-s.stopCh: + close(stop) + return + case <-s.triggerCh: + break loop + //case <-s.testWaitCh: + } + } + } +} + +// processModules runs an entire processing round, calling processable modules +// with the appropriate Environment. +func (s *Scheduler) processModules() { + servers := make(ServerSet) + for server, _ := range s.servers { + if ok, _ := server.CanRequestNow(); ok { + servers[server] = struct{}{} + } + } + for _, module := range s.modules { + if module.Process(servers, s.events) { + s.Trigger() + } + } + s.events = nil +} + +func (s *Scheduler) Trigger() { + select { + case s.triggerCh <- struct{}{}: + default: + } +} + +func (s *Scheduler) handleEvent(server Server, event Event) { + s.lock.Lock() + defer s.lock.Unlock() + + s.events = append(s.events, ServerEvent{Server: server, Event: event}) + s.Trigger() +} diff --git a/beacon/light/request/server.go b/beacon/light/request/server.go new file mode 100644 index 0000000000..f72feddfc1 --- /dev/null +++ b/beacon/light/request/server.go @@ -0,0 +1,294 @@ +// Copyright 2023 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package request + +import ( + "math/rand" + "sync" + "time" + + "github.com/ethereum/go-ethereum/common/mclock" +) + +type RequestServer interface { + Subscribe(eventCallback func(event Event)) + CanSendRequest(request interface{}) (bool, float32) + SendRequest(request interface{}) (reqId interface{}) + Unsubscribe() +} + +type Server interface { + RequestServer + CanRequestNow() (bool, float32) +} + +type Event struct { + Type int + ReqId, Data interface{} +} + +const ( + EvValidResponse = iota // data: response struct + EvInvalidResponse // data: nil + EvSoftTimeout // data: nil + EvHardTimeout // data: nil + EvCanRequestAgain // reqId: nil data: nil +) + +const ( + softRequestTimeout = time.Second + hardRequestTimeout = time.Second * 10 +) + +type serverWithTimeout struct { + RequestServer + lock sync.Mutex + clock mclock.Clock + childEventCb func(event Event) + timeouts map[interface{}]mclock.Timer +} + +func (s *serverWithTimeout) Subscribe(eventCallback func(event Event)) { + s.lock.Lock() + defer s.lock.Unlock() + + s.childEventCb = eventCallback + s.RequestServer.Subscribe(s.eventCallback) +} + +func (s *serverWithTimeout) eventCallback(event Event) { + s.lock.Lock() + defer s.lock.Unlock() + + switch event.Type { + case EvValidResponse, EvInvalidResponse: + if timer, ok := s.timeouts[event.ReqId]; ok { + // Note: if stopping the timer is unsuccessful then the resulting AfterFunc + // call will just do nothing + s.stopTimer(timer) + delete(s.timeouts, event.ReqId) + s.childEventCb(event) + } + default: + s.childEventCb(event) + } +} + +func (s *serverWithTimeout) SendRequest(request interface{}) (reqId interface{}) { + s.lock.Lock() + defer s.lock.Unlock() + + reqId = s.RequestServer.SendRequest(request) + s.timeouts[reqId] = s.clock.AfterFunc(softRequestTimeout, func() { + /*if s.testTimerResults != nil { + s.testTimerResults = append(s.testTimerResults, true) // simulated timer finished + }*/ + s.lock.Lock() + defer s.lock.Unlock() + + if _, ok := s.timeouts[reqId]; !ok { + return + } + s.timeouts[reqId] = s.clock.AfterFunc(hardRequestTimeout-softRequestTimeout, func() { + /*if s.testTimerResults != nil { + s.testTimerResults = append(s.testTimerResults, true) // simulated timer finished + }*/ + s.lock.Lock() + defer s.lock.Unlock() + + if _, ok := s.timeouts[reqId]; !ok { + return + } + delete(s.timeouts, reqId) + s.childEventCb(Event{Type: EvHardTimeout, ReqId: reqId}) + }) + s.childEventCb(Event{Type: EvSoftTimeout, ReqId: reqId}) + }) + return reqId +} + +// stop stops all goroutines associated with the server. +func (s *serverWithTimeout) Unsubscribe() { + s.lock.Lock() + defer s.lock.Unlock() + + for _, timer := range s.timeouts { + if timer != nil { + s.stopTimer(timer) + } + } + s.childEventCb = nil + s.RequestServer.Unsubscribe() +} + +func (s *serverWithTimeout) stopTimer(timer mclock.Timer) { + timer.Stop() + /*if timer.Stop() && s.scheduler.testTimerResults != nil { + s.scheduler.testTimerResults = append(s.scheduler.testTimerResults, false) // simulated timer stopped + }*/ +} + +type serverWithDelay struct { + serverWithTimeout + lock sync.Mutex + childEventCb func(event Event) + softTimeouts map[interface{}]struct{} + pendingCount, timeoutCount int + parallelLimit float32 + sendEvent bool + delayTimer mclock.Timer + delayCounter int + + parallelAdjustUp, parallelAdjustDown, minParallelLimit float32 +} + +func (s *serverWithDelay) Subscribe(eventCallback func(event Event)) { + s.lock.Lock() + defer s.lock.Unlock() + + s.childEventCb = eventCallback + s.serverWithTimeout.Subscribe(s.eventCallback) +} + +func (s *serverWithDelay) eventCallback(event Event) { + s.lock.Lock() + defer s.lock.Unlock() + + switch event.Type { + case EvSoftTimeout: + s.softTimeouts[event.ReqId] = struct{}{} + s.timeoutCount++ + s.parallelLimit -= s.parallelAdjustDown + if s.parallelLimit < s.minParallelLimit { + s.parallelLimit = s.minParallelLimit + } + case EvValidResponse, EvInvalidResponse, EvHardTimeout: + if _, ok := s.softTimeouts[event.ReqId]; ok { + delete(s.softTimeouts, event.ReqId) + s.timeoutCount-- + } + if event.Type == EvValidResponse && s.pendingCount >= int(s.parallelLimit) { + s.parallelLimit -= s.parallelAdjustUp + } + s.pendingCount-- + s.canRequestNow() // send event if needed + } + s.childEventCb(event) +} + +func (s *serverWithDelay) SendRequest(request interface{}) (reqId interface{}) { + s.lock.Lock() + defer s.lock.Unlock() + + s.pendingCount++ + return s.serverWithTimeout.SendRequest(request) +} + +// stop stops all goroutines associated with the server. +func (s *serverWithDelay) Unsubscribe() { + s.lock.Lock() + defer s.lock.Unlock() + + if s.delayTimer != nil { + s.stopTimer(s.delayTimer) + s.delayTimer = nil + } + s.childEventCb = nil + s.serverWithTimeout.Unsubscribe() +} + +func (s *serverWithDelay) canRequestNow() (bool, float32) { + if s.delayTimer != nil || s.pendingCount >= int(s.parallelLimit) { + return false, 0 + } + if s.sendEvent { + s.childEventCb(Event{Type: EvCanRequestAgain}) + s.sendEvent = false + } + if s.parallelLimit < s.minParallelLimit { + s.parallelLimit = s.minParallelLimit + } + return true, -(float32(s.pendingCount) + rand.Float32()) / s.parallelLimit +} + +// EvCanRequestAgain guaranteed if it returns false +func (s *serverWithDelay) CanRequestNow() (bool, float32) { + s.lock.Lock() + defer s.lock.Unlock() + + canSend, priority := s.canRequestNow() + if !canSend { + s.sendEvent = true + } + return canSend, priority +} + +func (s *serverWithDelay) Delay(delay time.Duration) { + s.lock.Lock() + defer s.lock.Unlock() + + if s.delayTimer != nil { + // Note: if stopping the timer is unsuccessful then the resulting AfterFunc + // call will just do nothing + s.stopTimer(s.delayTimer) + s.delayTimer = nil + } + + s.delayCounter++ + delayCounter := s.delayCounter + s.delayTimer = s.clock.AfterFunc(delay, func() { + /*if s.scheduler.testTimerResults != nil { + s.scheduler.testTimerResults = append(s.scheduler.testTimerResults, true) // simulated timer finished + }*/ + s.lock.Lock() + if s.delayTimer != nil && s.delayCounter == delayCounter { // do nothing if there is a new timer now + s.delayTimer = nil + s.canRequestNow() // send event if necessary + } + s.lock.Unlock() + }) +} + +//func (s *serverWithDelay) Fail() + +type ServerSet map[Server]struct{} + +// TryRequest tries to send the given request and returns true in case of success. +func (s *ServerSet) TryRequest(request interface{}) (Server, interface{}) { + var ( + maxServerPriority, maxRequestPriority float32 + bestServer Server + ) + for server, _ := range *s { + canRequest, serverPriority := server.CanRequestNow() + if !canRequest { + delete(*s, server) + continue + } + canSend, requestPriority := server.CanSendRequest(request) + if !canSend || requestPriority < maxRequestPriority || + (requestPriority == maxRequestPriority && serverPriority <= maxServerPriority) { + continue + } + maxServerPriority, maxRequestPriority = serverPriority, requestPriority + bestServer = server + } + if bestServer != nil { + return bestServer, bestServer.SendRequest(request) + } + return nil, nil +}