From 245d166d45efe51317769ad594393ca7dcbb5b9a Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Wed, 24 Jan 2024 04:35:00 +0100 Subject: [PATCH] beacon/light/request: add Scheduler unit test --- beacon/light/request/scheduler.go | 60 ++++++------ beacon/light/request/scheduler_test.go | 128 +++++++++++++++++++++++++ beacon/light/request/server.go | 4 +- beacon/light/request/server_test.go | 6 +- cmd/blsync/main.go | 5 +- 5 files changed, 163 insertions(+), 40 deletions(-) create mode 100644 beacon/light/request/scheduler_test.go diff --git a/beacon/light/request/scheduler.go b/beacon/light/request/scheduler.go index 43574690e3..b78c5fe5a7 100644 --- a/beacon/light/request/scheduler.go +++ b/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{ diff --git a/beacon/light/request/scheduler_test.go b/beacon/light/request/scheduler_test.go new file mode 100644 index 0000000000..68784140a3 --- /dev/null +++ b/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:] + } +} diff --git a/beacon/light/request/server.go b/beacon/light/request/server.go index 164928eef7..3fc2c5fc32 100644 --- a/beacon/light/request/server.go +++ b/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) diff --git a/beacon/light/request/server_test.go b/beacon/light/request/server_test.go index b524313f8d..3567780e9a 100644 --- a/beacon/light/request/server_test.go +++ b/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 { diff --git a/cmd/blsync/main.go b/cmd/blsync/main.go index 8d43ad2281..93be22ecb6 100644 --- a/cmd/blsync/main.go +++ b/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()