mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-20 02:42:27 +00:00
swarm/network: fix data race in fetcher_test.go
Problem: Let's say TestA() and TestB() was started right after each other. TestA() called `go Fetcher.run(ctx)` (read: searchTimeout), on test completion the context was cancelled, but the test did not wait for goroutine termination. Test(B) started by modifying searchTimeout (write). Solution: Let's just store searchTimeout on the Fetcher struct and change the package var (default) to const (thread-safe). Alternative (almost) solution: The above could have been solved by (little ugly, more complex) waitGroups. However that would have not work for TestFetcherFactory. As fetcherFactory.New()starts a goroutine inside it's body without providing a way for clear termination. (We did not want to modify the interface for this one, as we think the unclean termination does not cause a problem in this case in production.) fixes ethersphere/go-ethereum#1109
This commit is contained in:
parent
bad8c1e64c
commit
13d7b59726
2 changed files with 16 additions and 17 deletions
|
|
@ -26,14 +26,17 @@ import (
|
||||||
"github.com/ethereum/go-ethereum/swarm/storage"
|
"github.com/ethereum/go-ethereum/swarm/storage"
|
||||||
)
|
)
|
||||||
|
|
||||||
var searchTimeout = 1 * time.Second
|
const (
|
||||||
|
defaultSearchTimeout = 1 * time.Second
|
||||||
|
// maximum number of forwarded requests (hops), to make sure requests are not
|
||||||
|
// forwarded forever in peer loops
|
||||||
|
maxHopCount uint8 = 20
|
||||||
|
)
|
||||||
|
|
||||||
// Time to consider peer to be skipped.
|
// Time to consider peer to be skipped.
|
||||||
// Also used in stream delivery.
|
// Also used in stream delivery.
|
||||||
var RequestTimeout = 10 * time.Second
|
var RequestTimeout = 10 * time.Second
|
||||||
|
|
||||||
var maxHopCount uint8 = 20 // maximum number of forwarded requests (hops), to make sure requests are not forwarded forever in peer loops
|
|
||||||
|
|
||||||
type RequestFunc func(context.Context, *Request) (*enode.ID, chan struct{}, error)
|
type RequestFunc func(context.Context, *Request) (*enode.ID, chan struct{}, error)
|
||||||
|
|
||||||
// Fetcher is created when a chunk is not found locally. It starts a request handler loop once and
|
// Fetcher is created when a chunk is not found locally. It starts a request handler loop once and
|
||||||
|
|
@ -47,6 +50,7 @@ type Fetcher struct {
|
||||||
addr storage.Address // the address of the chunk to be fetched
|
addr storage.Address // the address of the chunk to be fetched
|
||||||
offerC chan *enode.ID // channel of sources (peer node id strings)
|
offerC chan *enode.ID // channel of sources (peer node id strings)
|
||||||
requestC chan uint8 // channel for incoming requests (with the hopCount value in it)
|
requestC chan uint8 // channel for incoming requests (with the hopCount value in it)
|
||||||
|
searchTimeout time.Duration
|
||||||
skipCheck bool
|
skipCheck bool
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -118,6 +122,7 @@ func NewFetcher(addr storage.Address, rf RequestFunc, skipCheck bool) *Fetcher {
|
||||||
protoRequestFunc: rf,
|
protoRequestFunc: rf,
|
||||||
offerC: make(chan *enode.ID),
|
offerC: make(chan *enode.ID),
|
||||||
requestC: make(chan uint8),
|
requestC: make(chan uint8),
|
||||||
|
searchTimeout: defaultSearchTimeout,
|
||||||
skipCheck: skipCheck,
|
skipCheck: skipCheck,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -232,7 +237,7 @@ func (f *Fetcher) run(ctx context.Context, peers *sync.Map) {
|
||||||
// if wait channel is not set, set it to a timer
|
// if wait channel is not set, set it to a timer
|
||||||
if requested {
|
if requested {
|
||||||
if wait == nil {
|
if wait == nil {
|
||||||
wait = time.NewTimer(searchTimeout)
|
wait = time.NewTimer(f.searchTimeout)
|
||||||
defer wait.Stop()
|
defer wait.Stop()
|
||||||
waitC = wait.C
|
waitC = wait.C
|
||||||
} else {
|
} else {
|
||||||
|
|
@ -243,8 +248,8 @@ func (f *Fetcher) run(ctx context.Context, peers *sync.Map) {
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// reset the timer to go off after searchTimeout
|
// reset the timer to go off after defaultSearchTimeout
|
||||||
wait.Reset(searchTimeout)
|
wait.Reset(f.searchTimeout)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
doRequest = false
|
doRequest = false
|
||||||
|
|
|
||||||
|
|
@ -284,15 +284,11 @@ func TestFetcherRetryOnTimeout(t *testing.T) {
|
||||||
requester := newMockRequester()
|
requester := newMockRequester()
|
||||||
addr := make([]byte, 32)
|
addr := make([]byte, 32)
|
||||||
fetcher := NewFetcher(addr, requester.doRequest, true)
|
fetcher := NewFetcher(addr, requester.doRequest, true)
|
||||||
|
// set searchTimeOut to low value so the test is quicker
|
||||||
|
fetcher.searchTimeout = 250 * time.Millisecond
|
||||||
|
|
||||||
peersToSkip := &sync.Map{}
|
peersToSkip := &sync.Map{}
|
||||||
|
|
||||||
// set searchTimeOut to low value so the test is quicker
|
|
||||||
defer func(t time.Duration) {
|
|
||||||
searchTimeout = t
|
|
||||||
}(searchTimeout)
|
|
||||||
searchTimeout = 250 * time.Millisecond
|
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
|
|
@ -359,11 +355,9 @@ func TestFetcherRequestQuitRetriesRequest(t *testing.T) {
|
||||||
addr := make([]byte, 32)
|
addr := make([]byte, 32)
|
||||||
fetcher := NewFetcher(addr, requester.doRequest, true)
|
fetcher := NewFetcher(addr, requester.doRequest, true)
|
||||||
|
|
||||||
// make sure searchTimeout is long so it is sure the request is not retried because of timeout
|
// make sure the searchTimeout is long so it is sure the request is not
|
||||||
defer func(t time.Duration) {
|
// retried because of timeout
|
||||||
searchTimeout = t
|
fetcher.searchTimeout = 10 * time.Second
|
||||||
}(searchTimeout)
|
|
||||||
searchTimeout = 10 * time.Second
|
|
||||||
|
|
||||||
peersToSkip := &sync.Map{}
|
peersToSkip := &sync.Map{}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue