diff --git a/beacon/light/request/scheduler_test.go b/beacon/light/request/scheduler_test.go index 68784140a3..310e33af77 100644 --- a/beacon/light/request/scheduler_test.go +++ b/beacon/light/request/scheduler_test.go @@ -28,18 +28,12 @@ func TestEventFilter(t *testing.T) { }) // let module1 send a request srv.canRequest = 1 - module1.reqc = testRequest + module1.sendReq = testRequest s.Trigger() - // first triggered round sends the request, no events yet + // in first triggered round module1 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{}{} @@ -80,32 +74,32 @@ func (s *testServer) subscribe(eventCb func(Event)) { s.eventCb = eventCb } -func (s *testServer) canRequestNow() (bool, float32) { - return s.canRequest > 0, 0 +func (s *testServer) canRequestNow() bool { + return s.canRequest > 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) fail(string) {} func (s *testServer) unsubscribe() {} type testModule struct { name string processed [][]Event - reqc Request // request candidate + sendReq Request } -func (m *testModule) Process(events []Event) { +func (m *testModule) Process(requester Requester, events []Event) { m.processed = append(m.processed, events) -} - -func (m *testModule) MakeRequest(Server) (Request, float32) { - return m.reqc, 0 + if m.sendReq != nil { + if cs := requester.CanSendTo(); len(cs) > 0 { + requester.Send(cs[0], m.sendReq) + } + } } func (m *testModule) expProcess(t *testing.T, expEvents []Event) { diff --git a/beacon/light/request/server.go b/beacon/light/request/server.go index 92ce4f62a5..999f64178a 100644 --- a/beacon/light/request/server.go +++ b/beacon/light/request/server.go @@ -49,6 +49,8 @@ const ( defaultParallelLimit = 3 // parallelLimit initial value minFailureDelay = time.Millisecond * 100 // minimum disable time in case of request failure maxFailureDelay = time.Minute // maximum disable time in case of request failure + maxServerEventBuffer = 5 // server event allowance buffer limit + maxServerEventRate = time.Second // server event allowance buffer recharge rate ) // requestServer can send requests in a non-blocking way and feed back events @@ -226,13 +228,11 @@ func (s *serverWithTimeout) unsubscribe() { // serverWithLimits wraps serverWithTimeout and implements server. It limits the // number of parallel in-flight requests and prevents sending new requests when a -// pending one has already timed out. It also implements a failure delay mechanism -// that adds an exponentially growing delay each time a request fails (wrong answer -// or hard timeout). This makes the syncing mechanism less brittle as temporary -// failures of the server might happen sometimes, but still avoids hammering a -// non-functional server with requests. -// -// TODO protect against excessive server events +// pending one has already timed out. Server events are also rate limited. +// It also implements a failure delay mechanism that adds an exponentially growing +// delay each time a request fails (wrong answer or hard timeout). This makes the +// syncing mechanism less brittle as temporary failures of the server might happen +// sometimes, but still avoids hammering a non-functional server with requests. type serverWithLimits struct { serverWithTimeout lock sync.Mutex @@ -245,12 +245,15 @@ type serverWithLimits struct { delayCounter int failureDelayEnd mclock.AbsTime failureDelay float64 + serverEventBuffer int + eventBufferUpdated mclock.AbsTime } // init initializes serverWithLimits func (s *serverWithLimits) init() { s.softTimeouts = make(map[ID]struct{}) s.parallelLimit = defaultParallelLimit + s.serverEventBuffer = maxServerEventBuffer } // subscribe subscribes to events which include parent (serverWithTimeout) events @@ -267,6 +270,7 @@ func (s *serverWithLimits) subscribe(eventCallback func(event Event)) { func (s *serverWithLimits) eventCallback(event Event) { s.lock.Lock() var sendCanRequestAgain bool + passEvent := true switch event.Type { case EvTimeout: id := event.Data.(RequestResponse).ID @@ -295,10 +299,31 @@ func (s *serverWithLimits) eventCallback(event Event) { if event.Type == EvFail { s.failLocked("failed request") } + default: + // server event; check rate limit + if s.serverEventBuffer < maxServerEventBuffer { + now := s.clock.Now() + sinceUpdate := time.Duration(now - s.eventBufferUpdated) + if sinceUpdate >= maxServerEventRate*time.Duration(maxServerEventBuffer-s.serverEventBuffer) { + s.serverEventBuffer = maxServerEventBuffer + s.eventBufferUpdated = now + } else { + addBuffer := int(sinceUpdate / maxServerEventRate) + s.serverEventBuffer += addBuffer + s.eventBufferUpdated += mclock.AbsTime(maxServerEventRate * time.Duration(addBuffer)) + } + } + if s.serverEventBuffer > 0 { + s.serverEventBuffer-- + } else { + passEvent = false + } } childEventCb := s.childEventCb s.lock.Unlock() - childEventCb(event) + if passEvent { + childEventCb(event) + } if sendCanRequestAgain { childEventCb(Event{Type: EvCanRequestAgain}) } diff --git a/beacon/light/request/server_test.go b/beacon/light/request/server_test.go index 3567780e9a..b6b9edf9a0 100644 --- a/beacon/light/request/server_test.go +++ b/beacon/light/request/server_test.go @@ -36,7 +36,6 @@ func TestServerEvents(t *testing.T) { expEvent(testEventType) // send request, soft timeout, then valid response srv.sendRequest(testRequest) - expEvent(EvRequest) clock.WaitForTimers(1) clock.Run(softRequestTimeout) expEvent(EvTimeout) @@ -44,7 +43,6 @@ func TestServerEvents(t *testing.T) { expEvent(EvResponse) // send request, hard timeout (response after hard timeout should be ignored) srv.sendRequest(testRequest) - expEvent(EvRequest) clock.WaitForTimers(1) clock.Run(softRequestTimeout) expEvent(EvTimeout) @@ -63,7 +61,7 @@ func TestServerParallel(t *testing.T) { expSend := func(expSent int) { var sent int for sent <= expSent { - if ok, _ := srv.canRequestNow(); !ok { + if !srv.canRequestNow() { break } sent++ @@ -94,7 +92,7 @@ func TestServerFail(t *testing.T) { srv := NewServer(rs, clock) srv.subscribe(func(event Event) {}) expCanRequest := func(expCanRequest bool) { - if canRequest, _ := srv.canRequestNow(); canRequest != expCanRequest { + if canRequest := srv.canRequestNow(); canRequest != expCanRequest { t.Errorf("Wrong result for canRequestNow (expected %v, got %v)", expCanRequest, canRequest) } } @@ -108,7 +106,7 @@ func TestServerFail(t *testing.T) { rs.eventCb(Event{Type: EvResponse, Data: RequestResponse{ID: 1, Request: testRequest, Response: testResponse}}) expCanRequest(true) // explicit server.Fail - srv.Fail("") + srv.fail("") clock.WaitForTimers(1) expCanRequest(false) // cannot request for a while after a failure clock.Run(minFailureDelay) @@ -125,6 +123,32 @@ func TestServerFail(t *testing.T) { srv.unsubscribe() } +func TestServerEventRateLimit(t *testing.T) { + rs := &testRequestServer{} + clock := &mclock.Simulated{} + srv := NewServer(rs, clock) + var eventCount int + srv.subscribe(func(event Event) { + if !event.IsRequestEvent() { + eventCount++ + } + }) + expEvents := func(send, expAllowed int) { + eventCount = 0 + for sent := 0; sent < send; sent++ { + rs.eventCb(Event{Type: testEventType}) + } + if eventCount != expAllowed { + t.Errorf("Wrong number of server events passing rate limitation (sent %d, expected %d, got %d)", send, expAllowed, eventCount) + } + } + expEvents(maxServerEventBuffer+5, maxServerEventBuffer) + clock.Run(maxServerEventRate) + expEvents(5, 1) + clock.Run(maxServerEventRate * maxServerEventBuffer * 2) + expEvents(maxServerEventBuffer+5, maxServerEventBuffer) +} + type testRequestServer struct { eventCb func(Event) }