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
+}