diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index 60f86d0e14..5897752682 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -226,6 +226,8 @@ func (p *peerConnection) FetchNodeData(hashes []common.Hash) error { // requests. Its estimated header retrieval throughput is updated with that measured // just now. func (p *peerConnection) SetHeadersIdle(delivered int) { + p.lock.Lock() + defer p.lock.Unlock() p.setIdle(p.headerStarted, delivered, &p.headerThroughput, &p.headerIdle) } @@ -233,6 +235,8 @@ func (p *peerConnection) SetHeadersIdle(delivered int) { // requests. Its estimated body retrieval throughput is updated with that measured // just now. func (p *peerConnection) SetBodiesIdle(delivered int) { + p.lock.Lock() + defer p.lock.Unlock() p.setIdle(p.blockStarted, delivered, &p.blockThroughput, &p.blockIdle) } @@ -240,6 +244,8 @@ func (p *peerConnection) SetBodiesIdle(delivered int) { // retrieval requests. Its estimated receipt retrieval throughput is updated // with that measured just now. func (p *peerConnection) SetReceiptsIdle(delivered int) { + p.lock.Lock() + defer p.lock.Unlock() p.setIdle(p.receiptStarted, delivered, &p.receiptThroughput, &p.receiptIdle) } @@ -247,6 +253,8 @@ func (p *peerConnection) SetReceiptsIdle(delivered int) { // data retrieval requests. Its estimated state retrieval throughput is updated // with that measured just now. func (p *peerConnection) SetNodeDataIdle(delivered int) { + p.lock.Lock() + defer p.lock.Unlock() p.setIdle(p.stateStarted, delivered, &p.stateThroughput, &p.stateIdle) } @@ -256,9 +264,6 @@ func (p *peerConnection) setIdle(started time.Time, delivered int, throughput *f // Irrelevant of the scaling, make sure the peer ends up idle defer atomic.StoreInt32(idle, 0) - p.lock.Lock() - defer p.lock.Unlock() - // If nothing was delivered (hard timeout / unavailable data), reduce throughput to minimum if delivered == 0 { *throughput = 0 diff --git a/eth/downloader/queue.go b/eth/downloader/queue.go index 7c33953811..1ec432faff 100644 --- a/eth/downloader/queue.go +++ b/eth/downloader/queue.go @@ -564,26 +564,29 @@ func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common // CancelHeaders aborts a fetch request, returning all pending skeleton indexes to the queue. func (q *queue) CancelHeaders(request *fetchRequest) { + q.lock.Lock() + defer q.lock.Unlock() q.cancel(request, q.headerTaskQueue, q.headerPendPool) } // CancelBodies aborts a body fetch request, returning all pending headers to the // task queue. func (q *queue) CancelBodies(request *fetchRequest) { + q.lock.Lock() + defer q.lock.Unlock() q.cancel(request, q.blockTaskQueue, q.blockPendPool) } // CancelReceipts aborts a body fetch request, returning all pending headers to // the task queue. func (q *queue) CancelReceipts(request *fetchRequest) { + q.lock.Lock() + defer q.lock.Unlock() q.cancel(request, q.receiptTaskQueue, q.receiptPendPool) } // Cancel aborts a fetch request, returning all pending hashes to the task queue. func (q *queue) cancel(request *fetchRequest, taskQueue *prque.Prque, pendPool map[string]*fetchRequest) { - q.lock.Lock() - defer q.lock.Unlock() - if request.From > 0 { taskQueue.Push(request.From, -int64(request.From)) }