diff --git a/beacon/light/api/api_server.go b/beacon/light/api/api_server.go index 9ba2343545..b2ae934717 100755 --- a/beacon/light/api/api_server.go +++ b/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 diff --git a/beacon/light/request/scheduler.go b/beacon/light/request/scheduler.go index 173fce9f1a..7f16d77d1e 100644 --- a/beacon/light/request/scheduler.go +++ b/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) } diff --git a/beacon/light/request/server.go b/beacon/light/request/server.go index 16bfc2ceea..92ce4f62a5 100644 --- a/beacon/light/request/server.go +++ b/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 { diff --git a/beacon/light/sync/head_sync.go b/beacon/light/sync/head_sync.go index 1474347063..d35fa6aeea 100644 --- a/beacon/light/sync/head_sync.go +++ b/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 { diff --git a/beacon/light/sync/update_sync.go b/beacon/light/sync/update_sync.go index 5513f9becd..533e470fb0 100644 --- a/beacon/light/sync/update_sync.go +++ b/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 { diff --git a/beacon/light/sync/update_sync_test.go b/beacon/light/sync/update_sync_test.go index 07498b8366..1c4b3d6d76 100644 --- a/beacon/light/sync/update_sync_test.go +++ b/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) } diff --git a/cmd/blsync/block_sync.go b/cmd/blsync/block_sync.go index 135fd4dfd6..b07fe13bc3 100755 --- a/cmd/blsync/block_sync.go +++ b/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 {