beacon/light: added and updated descriptions

beacon/light/api/api_server.go

@@ -63,10 +63,6 @@ func (s *ApiServer) SendRequest(id request.ID, req request.Request) {
 			if updates, committees, err := s.api.GetBestUpdatesAndCommittees(data.FirstPeriod, data.Count); err == nil {
 				resp = sync.RespUpdates{Updates: updates, Committees: committees}
 			}
-		/*case sync.ReqOptimisticHead:
-		if signedHead, err := s.api.GetOptimisticHeadUpdate(); err == nil {
-			resp = signedHead
-		}*/ //TODO ???
 		case sync.ReqHeader:
 			if header, err := s.api.GetHeader(common.Hash(data)); err == nil {
 				resp = header

beacon/light/request/scheduler.go

@@ -24,20 +24,21 @@ import (
 
 // Module represents a mechanism which is typically responsible for downloading
 // and updating a passive data structure. It does not directly interact with the
-// servers (except for reporting server side failures). It receives and processes
-// events, maintains its internal state and generates request candidates. It is
-// the Scheduler's responsibility to feed events to the modules, call Process as
-// long as there might be something to process and then generate request
+// servers. It can start requests using the Requester interface, maintain its
+// internal state by receiving and processing Events and update its target data
+// structure based on the obtained data.
+// It is the Scheduler's responsibility to feed events to the modules, call
+// Process as long as there might be something to process and then generate request
 // candidates using MakeRequest and start the best possible requests.
 // Modules are called by Scheduler whenever a global trigger is fired. All events
 // fire the trigger. Changing a target data structure also triggers a next
 // processing round as it could make further actions possible either by the same
 // or another Module.
 type Module interface {
-	// Process is a non-blocking function responsible for maintaining the target
-	// data structures(s) and the internal state of the module. This state
-	// typically consists of information about pending requests and registered
-	// servers and it is updated based on the received events.
+	// Process is a non-blocking function responsible for starting requests,
+	// processing events and updating the target data structures(s) and the
+	// internal state of the module. Module state typically consists of information
+	// about pending requests and registered servers.
 	// Process is always called after an event is received or after a target data
 	// structure has been changed.
 	//
@@ -47,6 +48,10 @@ type Module interface {
 	Process(Requester, []Event)
 }
 
+// Requester allows Modules to obtain the list of momentarily available servers,
+// start new requests and report server failure when a response has been proven
+// to be invalid in the processing phase.
+// Note that all Requester functions should be safe to call from Module.Process.
 type Requester interface {
 	CanSendTo() []Server
 	Send(Server, Request) ID
@@ -340,11 +345,18 @@ func (s *Scheduler) closePending(server Server, filteredEvents map[Module][]Even
 	}
 }
 
+// requester implements Requester. Note that while requester basically wraps
+// Scheduler (with the added information of the currently processed Module), all
+// functions are safe to call from Module.Process which is running while
+// the Scheduler.lock mutex is held.
 type requester struct {
 	*Scheduler
 	module Module
 }
 
+// CanSendTo returns the list of currently available servers. It also returns
+// them in an order of least to most recently used, ensuring a round-robin usage
+// of suitable servers if the module always chooses the first suitable one.
 func (s requester) CanSendTo() []Server {
 	s.requesterLock.RLock()
 	defer s.requesterLock.RUnlock()
@@ -358,6 +370,8 @@ func (s requester) CanSendTo() []Server {
 	return list
 }
 
+// Send sends a request and adds an entry to Scheduler.pending map, ensuring that
+// related request events will be delivered to the sender Module.
 func (s requester) Send(srv Server, req Request) ID {
 	s.requesterLock.Lock()
 	defer s.requesterLock.Unlock()
@@ -377,6 +391,11 @@ func (s requester) Send(srv Server, req Request) ID {
 	return id
 }
 
+// Fail should be called when a server delivers invalid or useless information.
+// Calling Fail disables the given server for a period that is initially short
+// but is exponentially growing if it happens frequently. This results in a
+// somewhat fault tolerant operation that avoids hammering servers with requests
+// that they cannot serve but still gives them a chance periodically.
 func (s requester) Fail(srv Server, desc string) {
 	srv.(server).fail(desc)
 }

beacon/light/request/server.go

@@ -119,12 +119,11 @@ type RequestResponse struct {
 	Response Response
 }
 
-//TODO 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.
+// serverWithTimeout wraps a requestServer and introduces timeouts.
+// 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.
@@ -340,9 +339,6 @@ func (s *serverWithLimits) canRequest() bool {
 // canRequestNow checks whether a new request can be started, according to the
 // current in-flight request count and parallelLimit, and also the failure delay
 // timer.
-// If a new request is allowed then it also returns a priority value that can be
-// used to select the least overloaded server from an otherwise equally suitable
-// set of servers.
 // If it returns false then it is guaranteed that an EvCanRequestAgain will be
 // sent whenever the server becomes available for requesting again.
 func (s *serverWithLimits) canRequestNow() bool {

beacon/light/sync/head_sync.go

@@ -67,6 +67,7 @@ func NewHeadSync(headTracker headTracker, chain committeeChain) *HeadSync {
 	return s
 }
 
+// Process implements request.Module.
 func (s *HeadSync) Process(requester request.Requester, events []request.Event) {
 	for _, event := range events {
 		switch event.Type {

beacon/light/sync/update_sync.go

@@ -52,6 +52,7 @@ func NewCheckpointInit(chain committeeChain, checkpointHash common.Hash) *Checkp
 	}
 }
 
+// Process implements request.Module.
 func (s *CheckpointInit) Process(requester request.Requester, events []request.Event) {
 	for _, event := range events {
 		if !event.IsRequestEvent() {
@@ -180,6 +181,9 @@ func (s *ForwardUpdateSync) verifyRange(request ReqUpdates, response RespUpdates
 	return true
 }
 
+// updateResponse is a response that has passed initial verification and has been
+// queued for processing. Note that an update response cannot be processed until
+// the previous updates have also been added to the chain.
 type updateResponse struct {
 	sid      request.ServerAndID
 	request  ReqUpdates
@@ -195,6 +199,7 @@ func (u updateResponseList) Less(i, j int) bool {
 	return u[i].request.FirstPeriod < u[j].request.FirstPeriod
 }
 
+// Process implements request.Module.
 func (s *ForwardUpdateSync) Process(requester request.Requester, events []request.Event) {
 	for _, event := range events {
 		switch event.Type {

beacon/light/sync/update_sync_test.go

@@ -84,7 +84,7 @@ func TestUpdateSyncParallel(t *testing.T) {
 	// valid response to request 1; expect 8 periods synced and a new request started
 	ts.RequestEvent(request.EvResponse, ts.Request(1, 1), testRespUpdate(ts.Request(1, 1)))
 	ts.AddAllowance(testServer1, 1)
-	ts.Run(7, testServer1, ReqUpdates{FirstPeriod: 48, Count: 8})
+	ts.Run(2, testServer1, ReqUpdates{FirstPeriod: 48, Count: 8})
 	chain.ExpNextSyncPeriod(t, 8)
 
 	// valid response to requests 4 and 5
@@ -92,7 +92,7 @@ func TestUpdateSyncParallel(t *testing.T) {
 	ts.RequestEvent(request.EvResponse, ts.Request(1, 5), testRespUpdate(ts.Request(1, 5)))
 	ts.AddAllowance(testServer2, 2)
 	// expect 2 more requests but no sync progress (responses 4 and 5 cannot be added before 2 and 3)
-	ts.Run(8,
+	ts.Run(3,
 		testServer2, ReqUpdates{FirstPeriod: 56, Count: 8},
 		testServer2, ReqUpdates{FirstPeriod: 64, Count: 8})
 	chain.ExpNextSyncPeriod(t, 8)
@@ -101,15 +101,15 @@ func TestUpdateSyncParallel(t *testing.T) {
 	ts.RequestEvent(request.EvTimeout, ts.Request(1, 2), nil)
 	ts.RequestEvent(request.EvTimeout, ts.Request(1, 3), nil)
 	// no allowance, no more requests
-	ts.Run(10)
+	ts.Run(4)
 
 	// valid response to requests 6 and 8 and 9
 	ts.RequestEvent(request.EvResponse, ts.Request(1, 6), testRespUpdate(ts.Request(1, 6)))
-	ts.RequestEvent(request.EvResponse, ts.Request(8, 1), testRespUpdate(ts.Request(8, 1)))
-	ts.RequestEvent(request.EvResponse, ts.Request(8, 2), testRespUpdate(ts.Request(8, 2)))
+	ts.RequestEvent(request.EvResponse, ts.Request(3, 1), testRespUpdate(ts.Request(3, 1)))
+	ts.RequestEvent(request.EvResponse, ts.Request(3, 2), testRespUpdate(ts.Request(3, 2)))
 	ts.AddAllowance(testServer2, 3)
 	// server 2 can now resend requests 2 and 3 (timed out by server 1) and also send a new one
-	ts.Run(11,
+	ts.Run(5,
 		testServer2, ReqUpdates{FirstPeriod: 8, Count: 8},
 		testServer2, ReqUpdates{FirstPeriod: 16, Count: 8},
 		testServer2, ReqUpdates{FirstPeriod: 72, Count: 8})
@@ -118,14 +118,14 @@ func TestUpdateSyncParallel(t *testing.T) {
 	ts.RequestEvent(request.EvResponse, ts.Request(1, 2), testRespUpdate(ts.Request(1, 2)))
 	ts.AddAllowance(testServer1, 1)
 	// expect sync progress and one new request
-	ts.Run(14, testServer1, ReqUpdates{FirstPeriod: 80, Count: 8})
+	ts.Run(6, testServer1, ReqUpdates{FirstPeriod: 80, Count: 8})
 	chain.ExpNextSyncPeriod(t, 16)
 
 	// server 2 answers requests 11 and 12 (resends of requests 2 and 3)
-	ts.RequestEvent(request.EvResponse, ts.Request(11, 1), testRespUpdate(ts.Request(11, 1)))
-	ts.RequestEvent(request.EvResponse, ts.Request(11, 2), testRespUpdate(ts.Request(11, 2)))
+	ts.RequestEvent(request.EvResponse, ts.Request(5, 1), testRespUpdate(ts.Request(5, 1)))
+	ts.RequestEvent(request.EvResponse, ts.Request(5, 2), testRespUpdate(ts.Request(5, 2)))
 	ts.AddAllowance(testServer2, 2)
-	ts.Run(15,
+	ts.Run(7,
 		testServer2, ReqUpdates{FirstPeriod: 88, Count: 8},
 		testServer2, ReqUpdates{FirstPeriod: 96, Count: 4})
 	// finally the gap is filled, update can process responses up to req6
@@ -133,12 +133,12 @@ func TestUpdateSyncParallel(t *testing.T) {
 
 	// all remaining requests are answered
 	ts.RequestEvent(request.EvResponse, ts.Request(1, 3), testRespUpdate(ts.Request(1, 3)))
+	ts.RequestEvent(request.EvResponse, ts.Request(2, 1), testRespUpdate(ts.Request(2, 1)))
+	ts.RequestEvent(request.EvResponse, ts.Request(5, 3), testRespUpdate(ts.Request(5, 3)))
+	ts.RequestEvent(request.EvResponse, ts.Request(6, 1), testRespUpdate(ts.Request(6, 1)))
 	ts.RequestEvent(request.EvResponse, ts.Request(7, 1), testRespUpdate(ts.Request(7, 1)))
-	ts.RequestEvent(request.EvResponse, ts.Request(11, 3), testRespUpdate(ts.Request(11, 3)))
-	ts.RequestEvent(request.EvResponse, ts.Request(14, 1), testRespUpdate(ts.Request(14, 1)))
-	ts.RequestEvent(request.EvResponse, ts.Request(15, 1), testRespUpdate(ts.Request(15, 1)))
-	ts.RequestEvent(request.EvResponse, ts.Request(15, 2), testRespUpdate(ts.Request(15, 2)))
-	ts.Run(17)
+	ts.RequestEvent(request.EvResponse, ts.Request(7, 2), testRespUpdate(ts.Request(7, 2)))
+	ts.Run(8)
 	// expect chain to be fully synced
 	chain.ExpNextSyncPeriod(t, 100)
 }

cmd/blsync/block_sync.go

@@ -59,6 +59,7 @@ func newBeaconBlockSync(headTracker headTracker) *beaconBlockSync {
 	}
 }
 
+// Process implements request.Module.
 func (s *beaconBlockSync) Process(requester request.Requester, events []request.Event) {
 	for _, event := range events {
 		switch event.Type {