mirror of
https://gitlab.com/pulsechaincom/erigon-pulse.git
synced 2024-12-24 20:47:16 +00:00
eth, eth/downloader: fix processing interrupt caused by temp cancel
This commit is contained in:
parent
ae36beb38f
commit
2f4cbe22f5
@ -87,6 +87,8 @@ type Downloader struct {
|
|||||||
checks map[common.Hash]*crossCheck // Pending cross checks to verify a hash chain
|
checks map[common.Hash]*crossCheck // Pending cross checks to verify a hash chain
|
||||||
banned *set.Set // Set of hashes we've received and banned
|
banned *set.Set // Set of hashes we've received and banned
|
||||||
|
|
||||||
|
interrupt int32 // Atomic boolean to signal termination
|
||||||
|
|
||||||
// Statistics
|
// Statistics
|
||||||
importStart time.Time // Instance when the last blocks were taken from the cache
|
importStart time.Time // Instance when the last blocks were taken from the cache
|
||||||
importQueue []*Block // Previously taken blocks to check import progress
|
importQueue []*Block // Previously taken blocks to check import progress
|
||||||
@ -245,12 +247,6 @@ func (d *Downloader) synchronise(id string, hash common.Hash) error {
|
|||||||
if atomic.CompareAndSwapInt32(&d.notified, 0, 1) {
|
if atomic.CompareAndSwapInt32(&d.notified, 0, 1) {
|
||||||
glog.V(logger.Info).Infoln("Block synchronisation started")
|
glog.V(logger.Info).Infoln("Block synchronisation started")
|
||||||
}
|
}
|
||||||
|
|
||||||
// Create cancel channel for aborting mid-flight
|
|
||||||
d.cancelLock.Lock()
|
|
||||||
d.cancelCh = make(chan struct{})
|
|
||||||
d.cancelLock.Unlock()
|
|
||||||
|
|
||||||
// Abort if the queue still contains some leftover data
|
// Abort if the queue still contains some leftover data
|
||||||
if _, cached := d.queue.Size(); cached > 0 && d.queue.GetHeadBlock() != nil {
|
if _, cached := d.queue.Size(); cached > 0 && d.queue.GetHeadBlock() != nil {
|
||||||
return errPendingQueue
|
return errPendingQueue
|
||||||
@ -260,12 +256,16 @@ func (d *Downloader) synchronise(id string, hash common.Hash) error {
|
|||||||
d.peers.Reset()
|
d.peers.Reset()
|
||||||
d.checks = make(map[common.Hash]*crossCheck)
|
d.checks = make(map[common.Hash]*crossCheck)
|
||||||
|
|
||||||
|
// Create cancel channel for aborting mid-flight
|
||||||
|
d.cancelLock.Lock()
|
||||||
|
d.cancelCh = make(chan struct{})
|
||||||
|
d.cancelLock.Unlock()
|
||||||
|
|
||||||
// Retrieve the origin peer and initiate the downloading process
|
// Retrieve the origin peer and initiate the downloading process
|
||||||
p := d.peers.Peer(id)
|
p := d.peers.Peer(id)
|
||||||
if p == nil {
|
if p == nil {
|
||||||
return errUnknownPeer
|
return errUnknownPeer
|
||||||
}
|
}
|
||||||
|
|
||||||
return d.syncWithPeer(p, hash)
|
return d.syncWithPeer(p, hash)
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -282,7 +282,7 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash) (err error) {
|
|||||||
defer func() {
|
defer func() {
|
||||||
// reset on error
|
// reset on error
|
||||||
if err != nil {
|
if err != nil {
|
||||||
d.Cancel()
|
d.cancel()
|
||||||
d.mux.Post(FailedEvent{err})
|
d.mux.Post(FailedEvent{err})
|
||||||
} else {
|
} else {
|
||||||
d.mux.Post(DoneEvent{})
|
d.mux.Post(DoneEvent{})
|
||||||
@ -301,9 +301,9 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash) (err error) {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Cancel cancels all of the operations and resets the queue. It returns true
|
// cancel cancels all of the operations and resets the queue. It returns true
|
||||||
// if the cancel operation was completed.
|
// if the cancel operation was completed.
|
||||||
func (d *Downloader) Cancel() {
|
func (d *Downloader) cancel() {
|
||||||
// Close the current cancel channel
|
// Close the current cancel channel
|
||||||
d.cancelLock.Lock()
|
d.cancelLock.Lock()
|
||||||
if d.cancelCh != nil {
|
if d.cancelCh != nil {
|
||||||
@ -320,6 +320,12 @@ func (d *Downloader) Cancel() {
|
|||||||
d.queue.Reset()
|
d.queue.Reset()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Terminate interrupts the downloader, canceling all pending operations.
|
||||||
|
func (d *Downloader) Terminate() {
|
||||||
|
atomic.StoreInt32(&d.interrupt, 1)
|
||||||
|
d.cancel()
|
||||||
|
}
|
||||||
|
|
||||||
// fetchHahes starts retrieving hashes backwards from a specific peer and hash,
|
// fetchHahes starts retrieving hashes backwards from a specific peer and hash,
|
||||||
// up until it finds a common ancestor. If the source peer times out, alternative
|
// up until it finds a common ancestor. If the source peer times out, alternative
|
||||||
// ones are tried for continuation.
|
// ones are tried for continuation.
|
||||||
@ -737,12 +743,6 @@ func (d *Downloader) process() (err error) {
|
|||||||
|
|
||||||
atomic.StoreInt32(&d.processing, 0)
|
atomic.StoreInt32(&d.processing, 0)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
// Fetch the current cancel channel to allow termination
|
|
||||||
d.cancelLock.RLock()
|
|
||||||
cancel := d.cancelCh
|
|
||||||
d.cancelLock.RUnlock()
|
|
||||||
|
|
||||||
// Repeat the processing as long as there are blocks to import
|
// Repeat the processing as long as there are blocks to import
|
||||||
for {
|
for {
|
||||||
// Fetch the next batch of blocks
|
// Fetch the next batch of blocks
|
||||||
@ -759,12 +759,10 @@ func (d *Downloader) process() (err error) {
|
|||||||
|
|
||||||
// Actually import the blocks
|
// Actually import the blocks
|
||||||
glog.V(logger.Debug).Infof("Inserting chain with %d blocks (#%v - #%v)\n", len(blocks), blocks[0].RawBlock.Number(), blocks[len(blocks)-1].RawBlock.Number())
|
glog.V(logger.Debug).Infof("Inserting chain with %d blocks (#%v - #%v)\n", len(blocks), blocks[0].RawBlock.Number(), blocks[len(blocks)-1].RawBlock.Number())
|
||||||
for len(blocks) != 0 { // TODO: quit
|
for len(blocks) != 0 {
|
||||||
// Check for any termination requests
|
// Check for any termination requests
|
||||||
select {
|
if atomic.LoadInt32(&d.interrupt) == 1 {
|
||||||
case <-cancel:
|
|
||||||
return errCancelChainImport
|
return errCancelChainImport
|
||||||
default:
|
|
||||||
}
|
}
|
||||||
// Retrieve the first batch of blocks to insert
|
// Retrieve the first batch of blocks to insert
|
||||||
max := int(math.Min(float64(len(blocks)), float64(maxBlockProcess)))
|
max := int(math.Min(float64(len(blocks)), float64(maxBlockProcess)))
|
||||||
@ -777,7 +775,7 @@ func (d *Downloader) process() (err error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
glog.V(logger.Debug).Infof("Block #%d import failed: %v", raw[index].NumberU64(), err)
|
glog.V(logger.Debug).Infof("Block #%d import failed: %v", raw[index].NumberU64(), err)
|
||||||
d.dropPeer(blocks[index].OriginPeer)
|
d.dropPeer(blocks[index].OriginPeer)
|
||||||
d.Cancel()
|
d.cancel()
|
||||||
return errCancelChainImport
|
return errCancelChainImport
|
||||||
}
|
}
|
||||||
blocks = blocks[max:]
|
blocks = blocks[max:]
|
||||||
|
@ -247,7 +247,7 @@ func TestCancel(t *testing.T) {
|
|||||||
tester.newPeer("peer", hashes, blocks)
|
tester.newPeer("peer", hashes, blocks)
|
||||||
|
|
||||||
// Make sure canceling works with a pristine downloader
|
// Make sure canceling works with a pristine downloader
|
||||||
tester.downloader.Cancel()
|
tester.downloader.cancel()
|
||||||
hashCount, blockCount := tester.downloader.queue.Size()
|
hashCount, blockCount := tester.downloader.queue.Size()
|
||||||
if hashCount > 0 || blockCount > 0 {
|
if hashCount > 0 || blockCount > 0 {
|
||||||
t.Errorf("block or hash count mismatch: %d hashes, %d blocks, want 0", hashCount, blockCount)
|
t.Errorf("block or hash count mismatch: %d hashes, %d blocks, want 0", hashCount, blockCount)
|
||||||
@ -256,7 +256,7 @@ func TestCancel(t *testing.T) {
|
|||||||
if err := tester.sync("peer"); err != nil {
|
if err := tester.sync("peer"); err != nil {
|
||||||
t.Fatalf("failed to synchronise blocks: %v", err)
|
t.Fatalf("failed to synchronise blocks: %v", err)
|
||||||
}
|
}
|
||||||
tester.downloader.Cancel()
|
tester.downloader.cancel()
|
||||||
hashCount, blockCount = tester.downloader.queue.Size()
|
hashCount, blockCount = tester.downloader.queue.Size()
|
||||||
if hashCount > 0 || blockCount > 0 {
|
if hashCount > 0 || blockCount > 0 {
|
||||||
t.Errorf("block or hash count mismatch: %d hashes, %d blocks, want 0", hashCount, blockCount)
|
t.Errorf("block or hash count mismatch: %d hashes, %d blocks, want 0", hashCount, blockCount)
|
||||||
|
@ -251,7 +251,7 @@ func (pm *ProtocolManager) fetcher() {
|
|||||||
// downloading hashes and blocks as well as retrieving cached ones.
|
// downloading hashes and blocks as well as retrieving cached ones.
|
||||||
func (pm *ProtocolManager) syncer() {
|
func (pm *ProtocolManager) syncer() {
|
||||||
// Abort any pending syncs if we terminate
|
// Abort any pending syncs if we terminate
|
||||||
defer pm.downloader.Cancel()
|
defer pm.downloader.Terminate()
|
||||||
|
|
||||||
forceSync := time.Tick(forceSyncCycle)
|
forceSync := time.Tick(forceSyncCycle)
|
||||||
for {
|
for {
|
||||||
|
Loading…
Reference in New Issue
Block a user