eth/downloader: fix possible data race by inconsistent field protection

This commit is contained in:
BurtonQin 2020-02-12 10:28:11 +08:00
parent 34bb132b10
commit 22f6d73203
2 changed files with 14 additions and 6 deletions

View file

@ -226,6 +226,8 @@ func (p *peerConnection) FetchNodeData(hashes []common.Hash) error {
// requests. Its estimated header retrieval throughput is updated with that measured // requests. Its estimated header retrieval throughput is updated with that measured
// just now. // just now.
func (p *peerConnection) SetHeadersIdle(delivered int) { func (p *peerConnection) SetHeadersIdle(delivered int) {
p.lock.Lock()
defer p.lock.Unlock()
p.setIdle(p.headerStarted, delivered, &p.headerThroughput, &p.headerIdle) 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 // requests. Its estimated body retrieval throughput is updated with that measured
// just now. // just now.
func (p *peerConnection) SetBodiesIdle(delivered int) { func (p *peerConnection) SetBodiesIdle(delivered int) {
p.lock.Lock()
defer p.lock.Unlock()
p.setIdle(p.blockStarted, delivered, &p.blockThroughput, &p.blockIdle) 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 // retrieval requests. Its estimated receipt retrieval throughput is updated
// with that measured just now. // with that measured just now.
func (p *peerConnection) SetReceiptsIdle(delivered int) { func (p *peerConnection) SetReceiptsIdle(delivered int) {
p.lock.Lock()
defer p.lock.Unlock()
p.setIdle(p.receiptStarted, delivered, &p.receiptThroughput, &p.receiptIdle) 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 // data retrieval requests. Its estimated state retrieval throughput is updated
// with that measured just now. // with that measured just now.
func (p *peerConnection) SetNodeDataIdle(delivered int) { func (p *peerConnection) SetNodeDataIdle(delivered int) {
p.lock.Lock()
defer p.lock.Unlock()
p.setIdle(p.stateStarted, delivered, &p.stateThroughput, &p.stateIdle) 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 // Irrelevant of the scaling, make sure the peer ends up idle
defer atomic.StoreInt32(idle, 0) 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 nothing was delivered (hard timeout / unavailable data), reduce throughput to minimum
if delivered == 0 { if delivered == 0 {
*throughput = 0 *throughput = 0

View file

@ -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. // CancelHeaders aborts a fetch request, returning all pending skeleton indexes to the queue.
func (q *queue) CancelHeaders(request *fetchRequest) { func (q *queue) CancelHeaders(request *fetchRequest) {
q.lock.Lock()
defer q.lock.Unlock()
q.cancel(request, q.headerTaskQueue, q.headerPendPool) q.cancel(request, q.headerTaskQueue, q.headerPendPool)
} }
// CancelBodies aborts a body fetch request, returning all pending headers to the // CancelBodies aborts a body fetch request, returning all pending headers to the
// task queue. // task queue.
func (q *queue) CancelBodies(request *fetchRequest) { func (q *queue) CancelBodies(request *fetchRequest) {
q.lock.Lock()
defer q.lock.Unlock()
q.cancel(request, q.blockTaskQueue, q.blockPendPool) q.cancel(request, q.blockTaskQueue, q.blockPendPool)
} }
// CancelReceipts aborts a body fetch request, returning all pending headers to // CancelReceipts aborts a body fetch request, returning all pending headers to
// the task queue. // the task queue.
func (q *queue) CancelReceipts(request *fetchRequest) { func (q *queue) CancelReceipts(request *fetchRequest) {
q.lock.Lock()
defer q.lock.Unlock()
q.cancel(request, q.receiptTaskQueue, q.receiptPendPool) q.cancel(request, q.receiptTaskQueue, q.receiptPendPool)
} }
// Cancel aborts a fetch request, returning all pending hashes to the task queue. // 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) { func (q *queue) cancel(request *fetchRequest, taskQueue *prque.Prque, pendPool map[string]*fetchRequest) {
q.lock.Lock()
defer q.lock.Unlock()
if request.From > 0 { if request.From > 0 {
taskQueue.Push(request.From, -int64(request.From)) taskQueue.Push(request.From, -int64(request.From))
} }