forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

eth/downloader: hashing avoidance

Browse source
This commit is contained in:
Martin Holst Swende 2019-11-13 13:58:24 +01:00
parent 554269ebe6
commit 757ce48d87
No known key found for this signature in database
GPG key ID: 683B438C05A5DDF0
3 changed files with 31 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
eth/downloader/peer.go
View file

@ -175,12 +175,14 @@ func (p *peerConnection) FetchBodies(request *fetchRequest) error {
}
p.blockStarted = time.Now()
go func() {
// Convert the header set to a retrievable slice
hashes := make([]common.Hash, 0, len(request.Headers))
for _, header := range request.Headers {
hashes = append(hashes, header.Hash())
}
go p.peer.RequestBodies(hashes)
p.peer.RequestBodies(hashes)
}()
return nil
}
@ -197,12 +199,14 @@ func (p *peerConnection) FetchReceipts(request *fetchRequest) error {
}
p.receiptStarted = time.Now()
go func() {
// Convert the header set to a retrievable slice
hashes := make([]common.Hash, 0, len(request.Headers))
for _, header := range request.Headers {
hashes = append(hashes, header.Hash())
}
go p.peer.RequestReceipts(hashes)
p.peer.RequestReceipts(hashes)
}()
return nil
}

16
eth/downloader/queue.go
View file

@ -61,7 +61,6 @@ type fetchRequest struct {
// all outstanding pieces complete and the result as a whole can be processed.
type fetchResult struct {
pending int32 // Flag telling what deliveries are outstanding
Hash common.Hash // Hash of the header to prevent recalculating
Header *types.Header
Uncles []*types.Header
@ -71,7 +70,6 @@ type fetchResult struct {
func newFetchResult(header *types.Header, fastSync bool) *fetchResult {
item := &fetchResult{
Hash: header.Hash(),
Header: header,
}
if !header.EmptyBody() {
@ -511,13 +509,12 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common
// we can ask the resultcache if this header is within the
// "prioritized" segment of blocks. If it is not, we need to throttle
hash := header.Hash()
stale, throttle, item, err := q.resultCache.AddFetch(header, q.mode == FastSync)
if stale {
// Don't put back in the task queue, this item has already been
// delivered upstream
progress = true
delete(taskPool, hash)
delete(taskPool, header.Hash())
proc = proc - 1
continue
}
@ -539,13 +536,13 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common
}
if item.Done(typ) {
// If it's a noop, we can skip this task
delete(taskPool, hash)
delete(taskPool, header.Hash())
proc = proc - 1
progress = true
continue
}
// Otherwise unless the peer is known not to have the data, add to the retrieve list
if p.Lacks(hash) {
if p.Lacks(header.Hash()) {
skip = append(skip, header)
} else {
send = append(send, header)
@ -854,24 +851,25 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
var (
failure error
i int
hashes []common.Hash
)
for _, header := range request.Headers {
// Short circuit assembly if no more fetch results are found
if i >= results {
break
}
header.Hash()
// Validate the fields
if err := validate(i, header); err != nil {
failure = err
break
}
hashes = append(hashes, header.Hash())
i++
}
q.lock.Lock()
var acceptCount = 0
for _, header := range request.Headers[:i] {
if res, stale, err := q.resultCache.GetDeliverySlot(header); err == nil {
if res, stale, err := q.resultCache.GetDeliverySlot(header.Number.Uint64()); err == nil {
reconstruct(acceptCount, res)
} else {
// else: betweeen here and above, some other peer filled this result,
@ -880,8 +878,8 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header,
log.Info("delivery stale?", "err", err, "stale", stale)
failure = errStaleDelivery
}
delete(taskPool, header.Hash())
// Clean up a successful fetch
delete(taskPool, hashes[acceptCount])
request.Headers[acceptCount] = nil
acceptCount++
}

15
eth/downloader/resultcache.go
View file

@ -74,10 +74,9 @@ func (r *resultStore) SetThrottleThreshold(threshold uint64) {
// fetchResult -- the result to store data into
// err -- any error that occurred
func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (bool, bool, *fetchResult, error) {
header.Hash()
r.lock.RLock()
var index int
item, index, stale, throttled, err := r.getFetchResult(header)
item, index, stale, throttled, err := r.getFetchResult(header.Number.Uint64())
if err != nil || stale || throttled {
r.lock.RUnlock()
// Index is above the current threshold of 'prioritized' blocks,
@ -99,7 +98,7 @@ func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (bool, bool,
r.lock.Lock()
defer r.lock.Unlock()
// Same checks as above, now with wlock
item, index, stale, throttled, err = r.getFetchResult(header)
item, index, stale, throttled, err = r.getFetchResult(header.Number.Uint64())
if err != nil || stale || throttled {
return stale, throttled, item, err
}
@ -114,25 +113,25 @@ func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (bool, bool,
// is true, that means the header has already been delivered 'upstream'.
// This method does not bubble up the 'throttle' flag, since it's moot at the
// point in time when the item is downloaded and ready for delivery
func (r *resultStore) GetDeliverySlot(header *types.Header) (*fetchResult, bool, error) {
func (r *resultStore) GetDeliverySlot(headerNumber uint64) (*fetchResult, bool, error) {
r.lock.RLock()
defer r.lock.RUnlock()
res, _, stale, _, err := r.getFetchResult(header)
res, _, stale, _, err := r.getFetchResult(headerNumber)
return res, stale, err
}
// getFetchResult returns the fetchResult corresponding to the given item, and the index where
// the result is stored.
func (r *resultStore) getFetchResult(header *types.Header) (item *fetchResult, index int, stale, throttle bool, err error) {
func (r *resultStore) getFetchResult(headerNumber uint64) (item *fetchResult, index int, stale, throttle bool, err error) {
index = int(header.Number.Int64() - int64(r.resultOffset))
index = int(int64(headerNumber) - int64(r.resultOffset))
throttle = index >= int(r.throttleThreshold)
stale = index < 0
if index >= len(r.items) {
err = fmt.Errorf("index allocation went beyond available resultStore space "+
"(index [%d] = header [%d] - resultOffset [%d], len(resultStore) = %d",
index, header.Number.Int64(), r.resultOffset, len(r.items))
index, headerNumber, r.resultOffset, len(r.items))
return
}
if stale {