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: concurrency fixes

Browse source
This commit is contained in:
Martin Holst Swende 2019-10-25 13:49:04 +02:00
parent 4d84eba61a
commit 08e6e27c8f
No known key found for this signature in database
GPG key ID: 683B438C05A5DDF0
1 changed files with 27 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
eth/downloader/queue.go
View file

@ -773,15 +773,21 @@ func (q *queue) DeliverHeaders(id string, headers []*types.Header, headerProcCh
// also wakes any threads waiting for data delivery. // also wakes any threads waiting for data delivery.
func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLists [][]*types.Header) (int, error) { func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLists [][]*types.Header) (int, error) {
reconstruct := func(header *types.Header, index int, result *fetchResult) error { validate := func(index int, txHash, uncleHash, receiptHash common.Hash) error {
if types.DeriveSha(types.Transactions(txLists[index])) != header.TxHash || types.CalcUncleHash(uncleLists[index]) != header.UncleHash { if types.DeriveSha(types.Transactions(txLists[index])) != txHash {
return errInvalidBody
}
if types.CalcUncleHash(uncleLists[index]) != uncleHash {
return errInvalidBody return errInvalidBody
} }
result.Transactions = txLists[index]
result.Uncles = uncleLists[index]
return nil return nil
} }
return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, bodyReqTimer, len(txLists), reconstruct)
reconstruct := func(index int, result *fetchResult) {
result.Transactions = txLists[index]
result.Uncles = uncleLists[index]
}
return q.deliver(id, q.blockTaskPool, q.blockTaskQueue, q.blockPendPool, q.blockDonePool, bodyReqTimer, len(txLists), validate, reconstruct)
} }
// DeliverReceipts injects a receipt retrieval response into the results queue. // DeliverReceipts injects a receipt retrieval response into the results queue.
@ -789,14 +795,16 @@ func (q *queue) DeliverBodies(id string, txLists [][]*types.Transaction, uncleLi
// and also wakes any threads waiting for data delivery. // and also wakes any threads waiting for data delivery.
func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) (int, error) { func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) (int, error) {
reconstruct := func(header *types.Header, index int, result *fetchResult) error { validate := func(index int, txHash, uncleHash, receiptHash common.Hash) error {
if types.DeriveSha(types.Receipts(receiptList[index])) != header.ReceiptHash { if types.DeriveSha(types.Receipts(receiptList[index])) != receiptHash {
return errInvalidReceipt return errInvalidReceipt
} }
result.Receipts = receiptList[index]
return nil return nil
} }
return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, receiptReqTimer, len(receiptList), reconstruct) reconstruct := func(index int, result *fetchResult) {
result.Receipts = receiptList[index]
}
return q.deliver(id, q.receiptTaskPool, q.receiptTaskQueue, q.receiptPendPool, q.receiptDonePool, receiptReqTimer, len(receiptList), validate, reconstruct)
} }
// deliver injects a data retrieval response into the results queue. // deliver injects a data retrieval response into the results queue.
@ -804,7 +812,7 @@ func (q *queue) DeliverReceipts(id string, receiptList [][]*types.Receipt) (int,
// This method obtains the lock as needed // This method obtains the lock as needed
func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque, func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque,
pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, reqTimer metrics.Timer, pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, reqTimer metrics.Timer,
results int, reconstruct func(header *types.Header, index int, result *fetchResult) error) (int, error) { results int, validate func(index int, txHash, uncleHash, receiptHash common.Hash) error, reconstruct func(index int, result *fetchResult)) (int, error) {
q.lock.Lock() q.lock.Lock()
// Short circuit if the data was never requested // Short circuit if the data was never requested
@ -847,7 +855,7 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQ
failure = errInvalidChain failure = errInvalidChain
break break
} }
if err := reconstruct(header, i, q.resultCache[index]); err != nil { if err := validate(index, header.TxHash, header.UncleHash, header.ReceiptHash); err != nil {
failure = err failure = err
break break
} }
@ -861,10 +869,15 @@ func (q *queue) deliver(id string, taskPool map[common.Hash]*types.Header, taskQ
q.lock.Lock() q.lock.Lock()
for _, item := range acceptedItems { for _, item := range acceptedItems {
donePool[item.hash] = struct{}{} donePool[item.hash] = struct{}{}
if res := q.resultCache[item.index]; res != nil { if item.index < len(q.resultCache) {
res.Pending-- if res := q.resultCache[item.index]; res != nil {
reconstruct(item.index, res)
res.Pending--
delete(taskPool, item.hash)
}
} }
delete(taskPool, item.hash) // else: betweeen here and above, some other peer filled this result
// we just ignore and move on
} }
// Return all failed or missing fetches to the queue // Return all failed or missing fetches to the queue
for _, header := range request.Headers[i:] { for _, header := range request.Headers[i:] {