beacon/light/request: general request framework

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 <http://www.gnu.org/licenses/>.
+
+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()
+}

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 <http://www.gnu.org/licenses/>.
+
+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
+}