From 002cee3868a3a25e0ac4efb409daf08ba31e5568 Mon Sep 17 00:00:00 2001 From: lash Date: Fri, 1 Feb 2019 11:42:36 +0100 Subject: [PATCH] swarm/network, swarm/storage: Make context fetcher-wide --- swarm/network/fetcher.go | 43 +++++++----- swarm/network/fetcher_test.go | 115 ++++++++++++++++----------------- swarm/storage/feed/testutil.go | 4 +- swarm/storage/netstore.go | 8 +-- swarm/storage/netstore_test.go | 4 +- 5 files changed, 89 insertions(+), 85 deletions(-) diff --git a/swarm/network/fetcher.go b/swarm/network/fetcher.go index 3043f64758..d42e95e2cc 100644 --- a/swarm/network/fetcher.go +++ b/swarm/network/fetcher.go @@ -23,6 +23,7 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/enode" + "github.com/ethereum/go-ethereum/swarm/spancontext" "github.com/ethereum/go-ethereum/swarm/storage" ) @@ -52,6 +53,8 @@ type Fetcher struct { requestC chan uint8 // channel for incoming requests (with the hopCount value in it) searchTimeout time.Duration skipCheck bool + ctx context.Context + quitFunc func() } type Request struct { @@ -109,14 +112,15 @@ func NewFetcherFactory(request RequestFunc, skipCheck bool) *FetcherFactory { // contain the peers which are actively requesting this chunk, to make sure we // don't request back the chunks from them. // The created Fetcher is started and returned. -func (f *FetcherFactory) New(ctx context.Context, source storage.Address, peersToSkip *sync.Map) storage.NetFetcher { - fetcher := NewFetcher(source, f.request, f.skipCheck) - go fetcher.run(ctx, peersToSkip) +func (f *FetcherFactory) New(ctx context.Context, source storage.Address, peers *sync.Map) storage.NetFetcher { + fetcher := NewFetcher(ctx, source, f.request, f.skipCheck) + go fetcher.run(peers) return fetcher } // NewFetcher creates a new Fetcher for the given chunk address using the given request function. -func NewFetcher(addr storage.Address, rf RequestFunc, skipCheck bool) *Fetcher { +func NewFetcher(ctx context.Context, addr storage.Address, rf RequestFunc, skipCheck bool) *Fetcher { + sctx, sp := spancontext.StartSpan(ctx, "fetcher") return &Fetcher{ addr: addr, protoRequestFunc: rf, @@ -124,14 +128,18 @@ func NewFetcher(addr storage.Address, rf RequestFunc, skipCheck bool) *Fetcher { requestC: make(chan uint8), searchTimeout: defaultSearchTimeout, skipCheck: skipCheck, + ctx: sctx, + quitFunc: func() { + sp.Finish() + }, } } // Offer is called when an upstream peer offers the chunk via syncing as part of `OfferedHashesMsg` and the node does not have the chunk locally. -func (f *Fetcher) Offer(ctx context.Context, source *enode.ID) { +func (f *Fetcher) Offer(source *enode.ID) { // First we need to have this select to make sure that we return if context is done select { - case <-ctx.Done(): + case <-f.ctx.Done(): return default: } @@ -140,15 +148,15 @@ func (f *Fetcher) Offer(ctx context.Context, source *enode.ID) { // push to offerC instead if offerC is available (see number 2 in https://golang.org/ref/spec#Select_statements) select { case f.offerC <- source: - case <-ctx.Done(): + case <-f.ctx.Done(): } } // Request is called when an upstream peer request the chunk as part of `RetrieveRequestMsg`, or from a local request through FileStore, and the node does not have the chunk locally. -func (f *Fetcher) Request(ctx context.Context, hopCount uint8) { +func (f *Fetcher) Request(hopCount uint8) { // First we need to have this select to make sure that we return if context is done select { - case <-ctx.Done(): + case <-f.ctx.Done(): return default: } @@ -162,13 +170,13 @@ func (f *Fetcher) Request(ctx context.Context, hopCount uint8) { // push to offerC instead if offerC is available (see number 2 in https://golang.org/ref/spec#Select_statements) select { case f.requestC <- hopCount + 1: - case <-ctx.Done(): + case <-f.ctx.Done(): } } // start prepares the Fetcher // it keeps the Fetcher alive within the lifecycle of the passed context -func (f *Fetcher) run(ctx context.Context, peers *sync.Map) { +func (f *Fetcher) run(peers *sync.Map) { var ( doRequest bool // determines if retrieval is initiated in the current iteration wait *time.Timer // timer for search timeout @@ -219,16 +227,17 @@ func (f *Fetcher) run(ctx context.Context, peers *sync.Map) { doRequest = requested // all Fetcher context closed, can quit - case <-ctx.Done(): + case <-f.ctx.Done(): log.Trace("terminate fetcher", "request addr", f.addr) // TODO: send cancellations to all peers left over in peers map (i.e., those we requested from) + f.quitFunc() return } // need to issue a new request if doRequest { var err error - sources, err = f.doRequest(ctx, gone, peers, sources, hopCount) + sources, err = f.doRequest(gone, peers, sources, hopCount) if err != nil { log.Info("unable to request", "request addr", f.addr, "err", err) } @@ -266,7 +275,7 @@ func (f *Fetcher) run(ctx context.Context, peers *sync.Map) { // * the peer's address is added to the set of peers to skip // * the peer's address is removed from prospective sources, and // * a go routine is started that reports on the gone channel if the peer is disconnected (or terminated their streamer) -func (f *Fetcher) doRequest(ctx context.Context, gone chan *enode.ID, peersToSkip *sync.Map, sources []*enode.ID, hopCount uint8) ([]*enode.ID, error) { +func (f *Fetcher) doRequest(gone chan *enode.ID, peersToSkip *sync.Map, sources []*enode.ID, hopCount uint8) ([]*enode.ID, error) { var i int var sourceID *enode.ID var quit chan struct{} @@ -283,7 +292,7 @@ func (f *Fetcher) doRequest(ctx context.Context, gone chan *enode.ID, peersToSki for i = 0; i < len(sources); i++ { req.Source = sources[i] var err error - sourceID, quit, err = f.protoRequestFunc(ctx, req) + sourceID, quit, err = f.protoRequestFunc(f.ctx, req) if err == nil { // remove the peer from known sources // Note: we can modify the source although we are looping on it, because we break from the loop immediately @@ -297,7 +306,7 @@ func (f *Fetcher) doRequest(ctx context.Context, gone chan *enode.ID, peersToSki if !foundSource { req.Source = nil var err error - sourceID, quit, err = f.protoRequestFunc(ctx, req) + sourceID, quit, err = f.protoRequestFunc(f.ctx, req) if err != nil { // if no peers found to request from return sources, err @@ -314,7 +323,7 @@ func (f *Fetcher) doRequest(ctx context.Context, gone chan *enode.ID, peersToSki select { case <-quit: gone <- sourceID - case <-ctx.Done(): + case <-f.ctx.Done(): } }() return sources, nil diff --git a/swarm/network/fetcher_test.go b/swarm/network/fetcher_test.go index 563c6cece7..4e464f10f3 100644 --- a/swarm/network/fetcher_test.go +++ b/swarm/network/fetcher_test.go @@ -69,7 +69,11 @@ func (m *mockRequester) doRequest(ctx context.Context, request *Request) (*enode func TestFetcherSingleRequest(t *testing.T) { requester := newMockRequester() addr := make([]byte, 32) - fetcher := NewFetcher(addr, requester.doRequest, true) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + fetcher := NewFetcher(ctx, addr, requester.doRequest, true) peers := []string{"a", "b", "c", "d"} peersToSkip := &sync.Map{} @@ -77,13 +81,9 @@ func TestFetcherSingleRequest(t *testing.T) { peersToSkip.Store(p, time.Now()) } - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() + go fetcher.run(peersToSkip) - go fetcher.run(ctx, peersToSkip) - - rctx := context.Background() - fetcher.Request(rctx, 0) + fetcher.Request(0) select { case request := <-requester.requestC: @@ -115,20 +115,19 @@ func TestFetcherSingleRequest(t *testing.T) { func TestFetcherCancelStopsFetcher(t *testing.T) { requester := newMockRequester() addr := make([]byte, 32) - fetcher := NewFetcher(addr, requester.doRequest, true) - - peersToSkip := &sync.Map{} ctx, cancel := context.WithCancel(context.Background()) + fetcher := NewFetcher(ctx, addr, requester.doRequest, true) + + peersToSkip := &sync.Map{} + // we start the fetcher, and then we immediately cancel the context - go fetcher.run(ctx, peersToSkip) + go fetcher.run(peersToSkip) cancel() - rctx, rcancel := context.WithTimeout(ctx, 100*time.Millisecond) - defer rcancel() // we call Request with an active context - fetcher.Request(rctx, 0) + fetcher.Request(0) // fetcher should not initiate request, we can only check by waiting a bit and making sure no request is happening select { @@ -140,23 +139,23 @@ func TestFetcherCancelStopsFetcher(t *testing.T) { // TestFetchCancelStopsRequest tests that calling a Request function with a cancelled context does not initiate a request func TestFetcherCancelStopsRequest(t *testing.T) { + t.Skip("since context is now per fetcher, this test is likely redundant") + requester := newMockRequester(100 * time.Millisecond) addr := make([]byte, 32) - fetcher := NewFetcher(addr, requester.doRequest, true) - - peersToSkip := &sync.Map{} ctx, cancel := context.WithCancel(context.Background()) defer cancel() - // we start the fetcher with an active context - go fetcher.run(ctx, peersToSkip) + fetcher := NewFetcher(ctx, addr, requester.doRequest, true) - rctx, rcancel := context.WithCancel(context.Background()) - rcancel() + peersToSkip := &sync.Map{} + + // we start the fetcher with an active context + go fetcher.run(peersToSkip) // we call Request with a cancelled context - fetcher.Request(rctx, 0) + fetcher.Request(0) // fetcher should not initiate request, we can only check by waiting a bit and making sure no request is happening select { @@ -166,8 +165,7 @@ func TestFetcherCancelStopsRequest(t *testing.T) { } // if there is another Request with active context, there should be a request, because the fetcher itself is not cancelled - rctx = context.Background() - fetcher.Request(rctx, 0) + fetcher.Request(0) select { case <-requester.requestC: @@ -182,19 +180,19 @@ func TestFetcherCancelStopsRequest(t *testing.T) { func TestFetcherOfferUsesSource(t *testing.T) { requester := newMockRequester(100 * time.Millisecond) addr := make([]byte, 32) - fetcher := NewFetcher(addr, requester.doRequest, true) - - peersToSkip := &sync.Map{} ctx, cancel := context.WithCancel(context.Background()) defer cancel() - // start the fetcher - go fetcher.run(ctx, peersToSkip) + fetcher := NewFetcher(ctx, addr, requester.doRequest, true) + + peersToSkip := &sync.Map{} + + // start the fetcher + go fetcher.run(peersToSkip) - rctx := context.Background() // call the Offer function with the source peer - fetcher.Offer(rctx, &sourcePeerID) + fetcher.Offer(&sourcePeerID) // fetcher should not initiate request select { @@ -204,8 +202,7 @@ func TestFetcherOfferUsesSource(t *testing.T) { } // call Request after the Offer - rctx = context.Background() - fetcher.Request(rctx, 0) + fetcher.Request(0) // there should be exactly 1 request coming from fetcher var request *Request @@ -234,19 +231,19 @@ func TestFetcherOfferUsesSource(t *testing.T) { func TestFetcherOfferAfterRequestUsesSourceFromContext(t *testing.T) { requester := newMockRequester(100 * time.Millisecond) addr := make([]byte, 32) - fetcher := NewFetcher(addr, requester.doRequest, true) - - peersToSkip := &sync.Map{} ctx, cancel := context.WithCancel(context.Background()) defer cancel() + fetcher := NewFetcher(ctx, addr, requester.doRequest, true) + + peersToSkip := &sync.Map{} + // start the fetcher - go fetcher.run(ctx, peersToSkip) + go fetcher.run(peersToSkip) // call Request first - rctx := context.Background() - fetcher.Request(rctx, 0) + fetcher.Request(0) // there should be a request coming from fetcher var request *Request @@ -260,7 +257,7 @@ func TestFetcherOfferAfterRequestUsesSourceFromContext(t *testing.T) { } // after the Request call Offer - fetcher.Offer(context.Background(), &sourcePeerID) + fetcher.Offer(&sourcePeerID) // there should be a request coming from fetcher select { @@ -283,21 +280,21 @@ func TestFetcherOfferAfterRequestUsesSourceFromContext(t *testing.T) { func TestFetcherRetryOnTimeout(t *testing.T) { requester := newMockRequester() addr := make([]byte, 32) - fetcher := NewFetcher(addr, requester.doRequest, true) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + fetcher := NewFetcher(ctx, addr, requester.doRequest, true) // set searchTimeOut to low value so the test is quicker fetcher.searchTimeout = 250 * time.Millisecond peersToSkip := &sync.Map{} - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - // start the fetcher - go fetcher.run(ctx, peersToSkip) + go fetcher.run(peersToSkip) // call the fetch function with an active context - rctx := context.Background() - fetcher.Request(rctx, 0) + fetcher.Request(0) // after 100ms the first request should be initiated time.Sleep(100 * time.Millisecond) @@ -339,7 +336,7 @@ func TestFetcherFactory(t *testing.T) { fetcher := fetcherFactory.New(context.Background(), addr, peersToSkip) - fetcher.Request(context.Background(), 0) + fetcher.Request(0) // check if the created fetchFunction really starts a fetcher and initiates a request select { @@ -353,7 +350,11 @@ func TestFetcherFactory(t *testing.T) { func TestFetcherRequestQuitRetriesRequest(t *testing.T) { requester := newMockRequester() addr := make([]byte, 32) - fetcher := NewFetcher(addr, requester.doRequest, true) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + fetcher := NewFetcher(ctx, addr, requester.doRequest, true) // make sure the searchTimeout is long so it is sure the request is not // retried because of timeout @@ -361,13 +362,9 @@ func TestFetcherRequestQuitRetriesRequest(t *testing.T) { peersToSkip := &sync.Map{} - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() + go fetcher.run(peersToSkip) - go fetcher.run(ctx, peersToSkip) - - rctx := context.Background() - fetcher.Request(rctx, 0) + fetcher.Request(0) select { case <-requester.requestC: @@ -460,17 +457,15 @@ func TestRequestSkipPeerPermanent(t *testing.T) { func TestFetcherMaxHopCount(t *testing.T) { requester := newMockRequester() addr := make([]byte, 32) - fetcher := NewFetcher(addr, requester.doRequest, true) ctx, cancel := context.WithCancel(context.Background()) defer cancel() + fetcher := NewFetcher(ctx, addr, requester.doRequest, true) + peersToSkip := &sync.Map{} - go fetcher.run(ctx, peersToSkip) - - rctx := context.Background() - fetcher.Request(rctx, maxHopCount) + go fetcher.run(peersToSkip) // if hopCount is already at max no request should be initiated select { diff --git a/swarm/storage/feed/testutil.go b/swarm/storage/feed/testutil.go index b513fa1f2f..caa39d9ffc 100644 --- a/swarm/storage/feed/testutil.go +++ b/swarm/storage/feed/testutil.go @@ -40,9 +40,9 @@ func (t *TestHandler) Close() { type mockNetFetcher struct{} -func (m *mockNetFetcher) Request(ctx context.Context, hopCount uint8) { +func (m *mockNetFetcher) Request(hopCount uint8) { } -func (m *mockNetFetcher) Offer(ctx context.Context, source *enode.ID) { +func (m *mockNetFetcher) Offer(source *enode.ID) { } func newFakeNetFetcher(context.Context, storage.Address, *sync.Map) storage.NetFetcher { diff --git a/swarm/storage/netstore.go b/swarm/storage/netstore.go index 16bc48a9a2..502e4ee2ea 100644 --- a/swarm/storage/netstore.go +++ b/swarm/storage/netstore.go @@ -34,8 +34,8 @@ type ( ) type NetFetcher interface { - Request(ctx context.Context, hopCount uint8) - Offer(ctx context.Context, source *enode.ID) + Request(hopCount uint8) + Offer(source *enode.ID) } // NetStore is an extension of local storage @@ -271,9 +271,9 @@ func (f *fetcher) Fetch(rctx context.Context) (Chunk, error) { if err := source.UnmarshalText([]byte(sourceIF.(string))); err != nil { return nil, err } - f.netFetcher.Offer(rctx, &source) + f.netFetcher.Offer(&source) } else { - f.netFetcher.Request(rctx, hopCount) + f.netFetcher.Request(hopCount) } // wait until either the chunk is delivered or the context is done diff --git a/swarm/storage/netstore_test.go b/swarm/storage/netstore_test.go index a6a9f551ab..88ec6c28f4 100644 --- a/swarm/storage/netstore_test.go +++ b/swarm/storage/netstore_test.go @@ -46,12 +46,12 @@ type mockNetFetcher struct { mu sync.Mutex } -func (m *mockNetFetcher) Offer(ctx context.Context, source *enode.ID) { +func (m *mockNetFetcher) Offer(source *enode.ID) { m.offerCalled = true m.sources = append(m.sources, source) } -func (m *mockNetFetcher) Request(ctx context.Context, hopCount uint8) { +func (m *mockNetFetcher) Request(hopCount uint8) { m.mu.Lock() defer m.mu.Unlock()