beacon/light/request: add Scheduler unit test

beacon/light/request/scheduler.go

@@ -20,7 +20,6 @@ import (
 	"math"
 	"sync"
 
-	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/log"
 )
 
@@ -68,7 +67,6 @@ type Module interface {
 // allow new operations.
 type Scheduler struct {
 	lock    sync.Mutex
-	clock   mclock.Clock
 	modules []Module // first has highest priority
 	names   map[Module]string
 	servers map[server]struct{}
@@ -83,6 +81,9 @@ type Scheduler struct {
 	stopCh    chan chan struct{}
 
 	triggerCh chan struct{} // restarts waiting sync loop
+	// if trigger has already been fired then send to testWaitCh blocks until
+	// the triggered processing round is finished
+	testWaitCh chan struct{}
 }
 
 type (
@@ -118,9 +119,8 @@ type pendingRequest struct {
 }
 
 // NewScheduler creates a new Scheduler.
-func NewScheduler(clock mclock.Clock) *Scheduler {
+func NewScheduler() *Scheduler {
 	s := &Scheduler{
-		clock:   clock,
 		servers: make(map[server]struct{}),
 		names:   make(map[Module]string),
 		pending: make(map[ServerAndID]pendingRequest),
@@ -129,8 +129,8 @@ func NewScheduler(clock mclock.Clock) *Scheduler {
 		// 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{}),
+		triggerCh:  make(chan struct{}, 1),
+		testWaitCh: make(chan struct{}),
 	}
 	return s
 }
@@ -157,11 +157,10 @@ func (s *Scheduler) RegisterModule(m Module, name string) {
 }
 
 // RegisterServer registers a new server.
-func (s *Scheduler) RegisterServer(rs requestServer) {
+func (s *Scheduler) RegisterServer(server server) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
-	server := newServer(rs, s.clock)
 	s.addEvent(Event{Type: EvRegistered, Server: server})
 	server.subscribe(func(event Event) {
 		event.Server = server
@@ -170,17 +169,12 @@ func (s *Scheduler) RegisterServer(rs requestServer) {
 }
 
 // UnregisterServer removes a registered server.
-func (s *Scheduler) UnregisterServer(rs requestServer) {
+func (s *Scheduler) UnregisterServer(server server) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
-	for server := range s.servers {
-		if sl, ok := server.(*serverWithLimits); ok && sl.parent == rs {
-			server.unsubscribe()
-			s.addEvent(Event{Type: EvUnregistered, Server: server})
-			return
-		}
-	}
+	server.unsubscribe()
+	s.addEvent(Event{Type: EvUnregistered, Server: server})
 }
 
 // Start starts the scheduler. It should be called after registering all modules
@@ -191,15 +185,15 @@ func (s *Scheduler) Start() {
 
 // Stop stops the scheduler.
 func (s *Scheduler) Stop() {
+	stop := make(chan struct{})
+	s.stopCh <- stop
+	<-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 is the main event loop responsible for event/data processing and
@@ -219,7 +213,7 @@ func (s *Scheduler) syncLoop() {
 				return
 			case <-s.triggerCh:
 				break loop
-				//case <-s.testWaitCh:
+			case <-s.testWaitCh:
 			}
 		}
 	}
@@ -244,11 +238,11 @@ func (s *Scheduler) targetChanged() (changed bool) {
 // requests are generated and sent if necessary and possible.
 func (s *Scheduler) processRound() {
 	for {
-		serverEvents, requestEvents := s.filterEvents()
-		log.Debug("Processing modules", "server events", len(serverEvents))
+		filteredEvents := s.filterEvents()
+		log.Debug("Processing modules")
 		for _, module := range s.modules {
-			log.Debug("Processing module", "name", s.names[module], "request events", len(requestEvents[module]))
-			module.Process(append(serverEvents, requestEvents[module]...))
+			log.Debug("Processing module", "name", s.names[module], "events", len(filteredEvents[module]))
+			module.Process(filteredEvents[module])
 		}
 		if !s.targetChanged() {
 			break
@@ -358,13 +352,13 @@ func (s *Scheduler) addEvent(event Event) {
 // to the given server by adding a failed request event (EvFail), ensuring that
 // all requests get finalized and thereby allowing the module logic to be safe
 // and simple.
-func (s *Scheduler) filterEvents() (serverEvents []Event, requestEvents map[Module][]Event) {
+func (s *Scheduler) filterEvents() map[Module][]Event {
 	s.eventLock.Lock()
 	events := s.events
 	s.events = nil
 	s.eventLock.Unlock()
 
-	requestEvents = make(map[Module][]Event)
+	filteredEvents := make(map[Module][]Event)
 	for _, event := range events {
 		server, ok := event.Server.(server)
 		if !ok {
@@ -384,27 +378,29 @@ func (s *Scheduler) filterEvents() (serverEvents []Event, requestEvents map[Modu
 			if event.Type == EvResponse || event.Type == EvFail {
 				delete(s.pending, sid) // final event, close pending request
 			}
-			requestEvents[pending.module] = append(requestEvents[pending.module], event)
+			filteredEvents[pending.module] = append(filteredEvents[pending.module], event)
 		} else {
 			switch event.Type {
 			case EvRegistered:
 				s.servers[server] = struct{}{}
 			case EvUnregistered:
-				s.closePending(event.Server, requestEvents)
+				s.closePending(event.Server, filteredEvents)
 				delete(s.servers, server)
 			}
-			serverEvents = append(serverEvents, event)
+			for _, module := range s.modules {
+				filteredEvents[module] = append(filteredEvents[module], event)
+			}
 		}
 	}
-	return
+	return filteredEvents
 }
 
 // closePending closes all pending requests to the given server and adds an EvFail
 // event to properly finalize them
-func (s *Scheduler) closePending(server Server, requestEvents map[Module][]Event) {
+func (s *Scheduler) closePending(server Server, filteredEvents map[Module][]Event) {
 	for sid, pending := range s.pending {
 		if sid.Server == server {
-			requestEvents[pending.module] = append(requestEvents[pending.module], Event{
+			filteredEvents[pending.module] = append(filteredEvents[pending.module], Event{
 				Type:   EvFail,
 				Server: server,
 				Data: RequestResponse{

beacon/light/request/scheduler_test.go

@@ -0,0 +1,128 @@
+package request
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestEventFilter(t *testing.T) {
+	s := NewScheduler()
+	module1 := &testModule{name: "module1"}
+	module2 := &testModule{name: "module2"}
+	s.RegisterModule(module1, "module1")
+	s.RegisterModule(module2, "module2")
+	s.Start()
+	// startup process round without events
+	s.testWaitCh <- struct{}{}
+	module1.expProcess(t, nil)
+	module2.expProcess(t, nil)
+	srv := &testServer{}
+	// register server; both modules should receive server event
+	s.RegisterServer(srv)
+	s.testWaitCh <- struct{}{}
+	module1.expProcess(t, []Event{
+		Event{Type: EvRegistered, Server: srv},
+	})
+	module2.expProcess(t, []Event{
+		Event{Type: EvRegistered, Server: srv},
+	})
+	// let module1 send a request
+	srv.canRequest = 1
+	module1.reqc = testRequest
+	s.Trigger()
+	// first triggered round sends the request, no events yet
+	s.testWaitCh <- struct{}{}
+	module1.expProcess(t, nil)
+	module2.expProcess(t, nil)
+	// next round triggered by EvRequest; only module1 should receive it
+	s.testWaitCh <- struct{}{}
+	module1.expProcess(t, []Event{
+		Event{Type: EvRequest, Server: srv, Data: RequestResponse{ID: 1, Request: testRequest}},
+	})
+	module2.expProcess(t, nil)
+	// server emits EvTimeout; only module1 should receive it
+	srv.eventCb(Event{Type: EvTimeout, Data: RequestResponse{ID: 1, Request: testRequest}})
+	s.testWaitCh <- struct{}{}
+	module1.expProcess(t, []Event{
+		Event{Type: EvTimeout, Server: srv, Data: RequestResponse{ID: 1, Request: testRequest}},
+	})
+	module2.expProcess(t, nil)
+	// unregister server; both modules should receive server event
+	s.UnregisterServer(srv)
+	s.testWaitCh <- struct{}{}
+	module1.expProcess(t, []Event{
+		// module1 should also receive EvFail on its pending request
+		Event{Type: EvFail, Server: srv, Data: RequestResponse{ID: 1, Request: testRequest}},
+		Event{Type: EvUnregistered, Server: srv},
+	})
+	module2.expProcess(t, []Event{
+		Event{Type: EvUnregistered, Server: srv},
+	})
+	// response after server unregistered; should be discarded
+	srv.eventCb(Event{Type: EvResponse, Data: RequestResponse{ID: 1, Request: testRequest, Response: testResponse}})
+	s.testWaitCh <- struct{}{}
+	module1.expProcess(t, nil)
+	module2.expProcess(t, nil)
+	// no more process rounds expected; shut down
+	s.testWaitCh <- struct{}{}
+	module1.expNoMoreProcess(t)
+	module2.expNoMoreProcess(t)
+	s.Stop()
+}
+
+type testServer struct {
+	eventCb    func(Event)
+	lastID     ID
+	canRequest int
+}
+
+func (s *testServer) subscribe(eventCb func(Event)) {
+	s.eventCb = eventCb
+}
+
+func (s *testServer) canRequestNow() (bool, float32) {
+	return s.canRequest > 0, 0
+}
+
+func (s *testServer) sendRequest(req Request) ID {
+	s.canRequest--
+	s.lastID++
+	s.eventCb(Event{Type: EvRequest, Data: RequestResponse{ID: s.lastID, Request: req}})
+	return s.lastID
+}
+
+func (s *testServer) Fail(string)  {}
+func (s *testServer) unsubscribe() {}
+
+type testModule struct {
+	name      string
+	processed [][]Event
+	reqc      Request // request candidate
+}
+
+func (m *testModule) Process(events []Event) {
+	m.processed = append(m.processed, events)
+}
+
+func (m *testModule) MakeRequest(Server) (Request, float32) {
+	return m.reqc, 0
+}
+
+func (m *testModule) expProcess(t *testing.T, expEvents []Event) {
+	if len(m.processed) == 0 {
+		t.Errorf("Missing call to %s.Process", m.name)
+		return
+	}
+	events := m.processed[0]
+	m.processed = m.processed[1:]
+	if !reflect.DeepEqual(events, expEvents) {
+		t.Errorf("Call to %s.Process with wrong events (expected %v, got %v)", m.name, expEvents, events)
+	}
+}
+
+func (m *testModule) expNoMoreProcess(t *testing.T) {
+	for len(m.processed) > 0 {
+		t.Errorf("Unexpected call to %s.Process with events %v", m.name, m.processed[0])
+		m.processed = m.processed[1:]
+	}
+}

beacon/light/request/server.go

@@ -76,8 +76,8 @@ type server interface {
 	unsubscribe()
 }
 
-// newServer wraps a requestServer and returns a server
-func newServer(rs requestServer, clock mclock.Clock) server {
+// NewServer wraps a requestServer and returns a server
+func NewServer(rs requestServer, clock mclock.Clock) server {
 	s := &serverWithLimits{}
 	s.parent = rs
 	s.serverWithTimeout.init(clock)

beacon/light/request/server_test.go

@@ -16,7 +16,7 @@ var testEventType = &EventType{Name: "testEvent"}
 func TestServerEvents(t *testing.T) {
 	rs := &testRequestServer{}
 	clock := &mclock.Simulated{}
-	srv := newServer(rs, clock)
+	srv := NewServer(rs, clock)
 	var lastEventType *EventType
 	srv.subscribe(func(event Event) { lastEventType = event.Type })
 	evTypeName := func(evType *EventType) string {
@@ -57,7 +57,7 @@ func TestServerEvents(t *testing.T) {
 
 func TestServerParallel(t *testing.T) {
 	rs := &testRequestServer{}
-	srv := newServer(rs, &mclock.Simulated{})
+	srv := NewServer(rs, &mclock.Simulated{})
 	srv.subscribe(func(event Event) {})
 
 	expSend := func(expSent int) {
@@ -91,7 +91,7 @@ func TestServerParallel(t *testing.T) {
 func TestServerFail(t *testing.T) {
 	rs := &testRequestServer{}
 	clock := &mclock.Simulated{}
-	srv := newServer(rs, clock)
+	srv := NewServer(rs, clock)
 	srv.subscribe(func(event Event) {})
 	expCanRequest := func(expCanRequest bool) {
 		if canRequest, _ := srv.canRequestNow(); canRequest != expCanRequest {

cmd/blsync/main.go

@@ -115,8 +115,7 @@ func blsync(ctx *cli.Context) error {
 	headSync := sync.NewHeadSync(headTracker, committeeChain)
 
 	// set up scheduler and sync modules
-	scheduler := request.NewScheduler(&mclock.System{})
-
+	scheduler := request.NewScheduler()
 	checkpointInit := sync.NewCheckpointInit(committeeChain, chainConfig.Checkpoint)
 	forwardSync := sync.NewForwardUpdateSync(committeeChain)
 	beaconBlockSync := newBeaconBlockSync(headTracker)
@@ -132,7 +131,7 @@ func blsync(ctx *cli.Context) error {
 	// register server(s)
 	for _, url := range ctx.StringSlice(utils.BeaconApiFlag.Name) {
 		beaconApi := api.NewBeaconLightApi(url, customHeader)
-		scheduler.RegisterServer(api.NewApiServer(beaconApi))
+		scheduler.RegisterServer(request.NewServer(api.NewApiServer(beaconApi), &mclock.System{}))
 	}
 	// run until stopped
 	<-ctx.Done()