beacon/light/request: server unit tests added, minor changes and bugfixes in server logic

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 {
+func (s *ApiServer) SendRequest(id request.ID, req request.Request) {
-	id := request.ID(atomic.AddUint64(&s.lastId, 1))
-	s.eventCallback(request.Event{Type: request.EvRequest, Data: request.RequestResponse{ID: id, Request: req}})
 	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.

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 (

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
+// through the event callback. After each request it should send back either
-// send back an EvRequest event before returning from SendRequest. When finished,
+// EvResponse or EvFail. Additionally, it may also send application-defined
-// it should send back either EvResponse or EvFail. Additionally, it may also
+// events that the Modules can interpret.
-// send application-defined events that the Modules can interpret.
 type requestServer interface {
-	Subscribe(eventCallback func(event Event))
+	Subscribe(eventCallback func(Event))
-	SendRequest(request Request) ID
+	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
+	sendRequest(Request) ID
-	Fail(desc string)
+	Fail(string)
 	unsubscribe()
 }
 
@@ -122,17 +121,22 @@ type RequestResponse struct {
 	Response Response
 }
 
-// serverWithTimeout wraps a requestServer and implements timeouts. After
+// serverWithTimeout wraps a requestServer and introduces two new request event
-// softRequestTimeout it sends an EvTimeout after which and EvResponse or an
+// types: EvRequest and EvTimeout. Whenever a request is successfully sent, an
-// EvFail will still follow (EvTimeout cannot follow the latter two).
+// EvRequest event is emitted first. The request's lifecycle is concluded if
-// After hardRequestTimeout it sends an EvFail and blocks any further events
+// EvResponse or EvFail emitted by the parent requestServer. If this does not
-// related to the given request coming from the parent requestServer.
+// 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

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()                  {}