From 57953512364013713a09fb856c5d07d9833cd0c7 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Tue, 23 Jan 2024 02:38:14 +0100 Subject: [PATCH] beacon/light/request: server unit tests added, minor changes and bugfixes in server logic --- beacon/light/api/api_server.go | 7 +- beacon/light/request/scheduler.go | 2 - beacon/light/request/server.go | 75 +++++++--------- beacon/light/request/server_test.go | 134 ++++++++++++++++++++++++++++ 4 files changed, 169 insertions(+), 49 deletions(-) create mode 100644 beacon/light/request/server_test.go diff --git a/beacon/light/api/api_server.go b/beacon/light/api/api_server.go index 8f20455365..9ba2343545 100755 --- a/beacon/light/api/api_server.go +++ b/beacon/light/api/api_server.go @@ -17,8 +17,6 @@ package api import ( - "sync/atomic" - "github.com/ethereum/go-ethereum/beacon/light/request" "github.com/ethereum/go-ethereum/beacon/light/sync" "github.com/ethereum/go-ethereum/beacon/types" @@ -57,9 +55,7 @@ func (s *ApiServer) Subscribe(eventCallback func(event request.Event)) { } // SendRequest implements request.requestServer. -func (s *ApiServer) SendRequest(req request.Request) request.ID { - id := request.ID(atomic.AddUint64(&s.lastId, 1)) - s.eventCallback(request.Event{Type: request.EvRequest, Data: request.RequestResponse{ID: id, Request: req}}) +func (s *ApiServer) SendRequest(id request.ID, req request.Request) { go func() { var resp request.Response switch data := req.(type) { @@ -91,7 +87,6 @@ func (s *ApiServer) SendRequest(req request.Request) request.ID { s.eventCallback(request.Event{Type: request.EvFail, Data: request.RequestResponse{ID: id, Request: req}}) } }() - return id } // Unsubscribe implements request.requestServer. diff --git a/beacon/light/request/scheduler.go b/beacon/light/request/scheduler.go index b62cce19d4..43574690e3 100644 --- a/beacon/light/request/scheduler.go +++ b/beacon/light/request/scheduler.go @@ -83,8 +83,6 @@ type Scheduler 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 } type ( diff --git a/beacon/light/request/server.go b/beacon/light/request/server.go index c10f744d33..164928eef7 100644 --- a/beacon/light/request/server.go +++ b/beacon/light/request/server.go @@ -54,13 +54,12 @@ const ( ) // requestServer can send requests in a non-blocking way and feed back events -// through the event callback. When successfully sending a request it should -// send back an EvRequest event before returning from SendRequest. When finished, -// it should send back either EvResponse or EvFail. Additionally, it may also -// send application-defined events that the Modules can interpret. +// through the event callback. After each request it should send back either +// EvResponse or EvFail. Additionally, it may also send application-defined +// events that the Modules can interpret. type requestServer interface { - Subscribe(eventCallback func(event Event)) - SendRequest(request Request) ID + Subscribe(eventCallback func(Event)) + SendRequest(ID, Request) Unsubscribe() } @@ -70,10 +69,10 @@ type requestServer interface { // limit the number of parallel in-flight requests and temporarily disable // new requests based on timeouts and response failures. type server interface { - subscribe(eventCallback func(event Event)) + subscribe(eventCallback func(Event)) canRequestNow() (bool, float32) - sendRequest(request Request) ID - Fail(desc string) + sendRequest(Request) ID + Fail(string) unsubscribe() } @@ -122,17 +121,22 @@ type RequestResponse struct { Response Response } -// serverWithTimeout wraps a requestServer and implements timeouts. After -// softRequestTimeout it sends an EvTimeout after which and EvResponse or an -// EvFail will still follow (EvTimeout cannot follow the latter two). -// After hardRequestTimeout it sends an EvFail and blocks any further events -// related to the given request coming from the parent requestServer. +// serverWithTimeout wraps a requestServer and introduces two new request event +// types: EvRequest and EvTimeout. Whenever a request is successfully sent, an +// EvRequest event is emitted first. The request's lifecycle is concluded if +// EvResponse or EvFail emitted by the parent requestServer. If this does not +// happen until softRequestTimeout then EvTimeout is emitted, after which the +// final EvResponse or EvFail is still guaranteed to follow. +// If the parent fails to send this final event for hardRequestTimeout then +// serverWithTimeout emits EvFail and discards any further events from the +// parent related to the given request. type serverWithTimeout struct { parent requestServer lock sync.Mutex clock mclock.Clock childEventCb func(event Event) timeouts map[ID]mclock.Timer + lastID ID } // init initializes serverWithTimeout @@ -151,9 +155,18 @@ func (s *serverWithTimeout) subscribe(eventCallback func(event Event)) { s.parent.Subscribe(s.eventCallback) } -// sendRequest sends a request through the parent (requestServer). +// sendRequest generated a new request ID, emits EvRequest, sets up the timeout +// timer, then sends the request through the parent (requestServer). func (s *serverWithTimeout) sendRequest(request Request) (reqId ID) { - return s.parent.SendRequest(request) + s.lock.Lock() + s.lastID++ + id := s.lastID + reqData := RequestResponse{ID: id, Request: request} + s.childEventCb(Event{Type: EvRequest, Data: reqData}) + s.startTimeout(reqData) + s.lock.Unlock() + s.parent.SendRequest(id, request) + return id } // eventCallback is called by parent (requestServer) event subscription. @@ -162,15 +175,12 @@ func (s *serverWithTimeout) eventCallback(event Event) { defer s.lock.Unlock() switch event.Type { - case EvRequest: - s.startTimeout(event.Data.(RequestResponse)) - s.childEventCb(event) case EvResponse, EvFail: id := event.Data.(RequestResponse).ID if timer, ok := s.timeouts[id]; ok { // Note: if stopping the timer is unsuccessful then the resulting AfterFunc // call will just do nothing - s.stopTimer(timer) + timer.Stop() delete(s.timeouts, id) s.childEventCb(event) } @@ -183,18 +193,12 @@ func (s *serverWithTimeout) eventCallback(event Event) { func (s *serverWithTimeout) startTimeout(reqData RequestResponse) { id := reqData.ID s.timeouts[id] = s.clock.AfterFunc(softRequestTimeout, func() { - /*if s.testTimerResults != nil { - s.testTimerResults = append(s.testTimerResults, true) // simulated timer finished - }*/ s.lock.Lock() if _, ok := s.timeouts[id]; !ok { s.lock.Unlock() return } s.timeouts[id] = s.clock.AfterFunc(hardRequestTimeout-softRequestTimeout, func() { - /*if s.testTimerResults != nil { - s.testTimerResults = append(s.testTimerResults, true) // simulated timer finished - }*/ s.lock.Lock() if _, ok := s.timeouts[id]; !ok { s.lock.Unlock() @@ -218,21 +222,13 @@ func (s *serverWithTimeout) unsubscribe() { for _, timer := range s.timeouts { if timer != nil { - s.stopTimer(timer) + timer.Stop() } } s.childEventCb = nil s.parent.Unsubscribe() } -// stopTimer stops the given timer -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 - }*/ -} - // 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 @@ -327,7 +323,7 @@ func (s *serverWithLimits) unsubscribe() { defer s.lock.Unlock() if s.delayTimer != nil { - s.stopTimer(s.delayTimer) + s.delayTimer.Stop() s.delayTimer = nil } s.childEventCb = nil @@ -336,7 +332,7 @@ func (s *serverWithLimits) unsubscribe() { // canRequest checks whether a new request can be started. func (s *serverWithLimits) canRequest() (bool, float32) { - if s.delayTimer != nil || s.pendingCount >= int(s.parallelLimit) { + if s.delayTimer != nil || s.pendingCount >= int(s.parallelLimit) || s.timeoutCount > 0 { return false, 0 } if s.parallelLimit < minParallelLimit { @@ -375,7 +371,7 @@ func (s *serverWithLimits) delay(delay time.Duration) { 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.Stop() s.delayTimer = nil } @@ -384,9 +380,6 @@ func (s *serverWithLimits) delay(delay time.Duration) { log.Debug("Server delay started", "length", delay) s.delayTimer = s.clock.AfterFunc(delay, func() { log.Debug("Server delay ended", "length", delay) - /*if s.scheduler.testTimerResults != nil { - s.scheduler.testTimerResults = append(s.scheduler.testTimerResults, true) // simulated timer finished - }*/ var sendCanRequestAgain bool s.lock.Lock() if s.delayTimer != nil && s.delayCounter == delayCounter { // do nothing if there is a new timer now diff --git a/beacon/light/request/server_test.go b/beacon/light/request/server_test.go new file mode 100644 index 0000000000..b524313f8d --- /dev/null +++ b/beacon/light/request/server_test.go @@ -0,0 +1,134 @@ +package request + +import ( + "testing" + + "github.com/ethereum/go-ethereum/common/mclock" +) + +const ( + testRequest = "Life, the Universe, and Everything" + testResponse = 42 +) + +var testEventType = &EventType{Name: "testEvent"} + +func TestServerEvents(t *testing.T) { + rs := &testRequestServer{} + clock := &mclock.Simulated{} + srv := newServer(rs, clock) + var lastEventType *EventType + srv.subscribe(func(event Event) { lastEventType = event.Type }) + evTypeName := func(evType *EventType) string { + if evType == nil { + return "none" + } + return evType.Name + } + expEvent := func(expType *EventType) { + if lastEventType != expType { + t.Errorf("Wrong event type (expected %s, got %s)", evTypeName(expType), evTypeName(lastEventType)) + } + lastEventType = nil + } + // user events should simply be passed through + rs.eventCb(Event{Type: testEventType}) + expEvent(testEventType) + // send request, soft timeout, then valid response + srv.sendRequest(testRequest) + expEvent(EvRequest) + clock.WaitForTimers(1) + clock.Run(softRequestTimeout) + expEvent(EvTimeout) + rs.eventCb(Event{Type: EvResponse, Data: RequestResponse{ID: 1, Request: testRequest, Response: testResponse}}) + 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) + clock.WaitForTimers(1) + clock.Run(hardRequestTimeout) + expEvent(EvFail) + rs.eventCb(Event{Type: EvResponse, Data: RequestResponse{ID: 1, Request: testRequest, Response: testResponse}}) + expEvent(nil) +} + +func TestServerParallel(t *testing.T) { + rs := &testRequestServer{} + srv := newServer(rs, &mclock.Simulated{}) + srv.subscribe(func(event Event) {}) + + expSend := func(expSent int) { + var sent int + for sent <= expSent { + if ok, _ := srv.canRequestNow(); !ok { + break + } + sent++ + srv.sendRequest(testRequest) + } + if sent != expSent { + t.Errorf("Wrong number of parallel requests accepted (expected %d, got %d)", expSent, sent) + } + } + // max out parallel allowance + expSend(defaultParallelLimit) + // 1 answered, should accept 1 more + rs.eventCb(Event{Type: EvResponse, Data: RequestResponse{ID: 1, Request: testRequest, Response: testResponse}}) + expSend(1) + // 2 answered, should accept 2 more + rs.eventCb(Event{Type: EvResponse, Data: RequestResponse{ID: 2, Request: testRequest, Response: testResponse}}) + rs.eventCb(Event{Type: EvResponse, Data: RequestResponse{ID: 3, Request: testRequest, Response: testResponse}}) + expSend(2) + // failed request, should decrease allowance and not accept more + rs.eventCb(Event{Type: EvFail, Data: RequestResponse{ID: 4, Request: testRequest}}) + expSend(0) + srv.unsubscribe() +} + +func TestServerFail(t *testing.T) { + rs := &testRequestServer{} + clock := &mclock.Simulated{} + srv := newServer(rs, clock) + srv.subscribe(func(event Event) {}) + expCanRequest := func(expCanRequest bool) { + if canRequest, _ := srv.canRequestNow(); canRequest != expCanRequest { + t.Errorf("Wrong result for canRequestNow (expected %v, got %v)", expCanRequest, canRequest) + } + } + // timed out request + expCanRequest(true) + srv.sendRequest(testRequest) + clock.WaitForTimers(1) + expCanRequest(true) + clock.Run(softRequestTimeout) + expCanRequest(false) // cannot request when there is a timed out request + rs.eventCb(Event{Type: EvResponse, Data: RequestResponse{ID: 1, Request: testRequest, Response: testResponse}}) + expCanRequest(true) + // explicit server.Fail + srv.Fail("") + clock.WaitForTimers(1) + expCanRequest(false) // cannot request for a while after a failure + clock.Run(minFailureDelay) + expCanRequest(true) + // request returned with EvFail + srv.sendRequest(testRequest) + rs.eventCb(Event{Type: EvFail, Data: RequestResponse{ID: 2, Request: testRequest}}) + clock.WaitForTimers(1) + expCanRequest(false) // EvFail should also start failure delay + clock.Run(minFailureDelay) + expCanRequest(false) // second failure delay is longer, should still be disabled + clock.Run(minFailureDelay) + expCanRequest(true) + srv.unsubscribe() +} + +type testRequestServer struct { + eventCb func(Event) +} + +func (rs *testRequestServer) Subscribe(eventCb func(Event)) { rs.eventCb = eventCb } +func (rs *testRequestServer) SendRequest(ID, Request) {} +func (rs *testRequestServer) Unsubscribe() {}