discovery: sync blocks in a dedicated goroutine
This commit moves syncing blocks into a dedicated goroutine to avoid the race condition where several go channels are ready and the block height update is pushed after a network message is processed.
This commit is contained in:
1 file changed
+37
-20
+37
-20
@@ -514,12 +514,48 @@ func (d *AuthenticatedGossiper) start() error {
|
|||||||
|
|
||||||
d.syncMgr.Start()
|
d.syncMgr.Start()
|
||||||
|
|
||||||
d.wg.Add(1)
|
// Start receiving blocks in its dedicated goroutine.
|
||||||
|
d.wg.Add(2)
|
||||||
|
go d.syncBlockHeight()
|
||||||
go d.networkHandler()
|
go d.networkHandler()
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// syncBlockHeight syncs the best block height for the gossiper by reading
|
||||||
|
// blockEpochs.
|
||||||
|
//
|
||||||
|
// NOTE: must be run as a goroutine.
|
||||||
|
func (d *AuthenticatedGossiper) syncBlockHeight() {
|
||||||
|
defer d.wg.Done()
|
||||||
|
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
// A new block has arrived, so we can re-process the previously
|
||||||
|
// premature announcements.
|
||||||
|
case newBlock, ok := <-d.blockEpochs.Epochs:
|
||||||
|
// If the channel has been closed, then this indicates
|
||||||
|
// the daemon is shutting down, so we exit ourselves.
|
||||||
|
if !ok {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Once a new block arrives, we update our running
|
||||||
|
// track of the height of the chain tip.
|
||||||
|
d.Lock()
|
||||||
|
blockHeight := uint32(newBlock.Height)
|
||||||
|
d.bestHeight = blockHeight
|
||||||
|
d.Unlock()
|
||||||
|
|
||||||
|
log.Debugf("New block: height=%d, hash=%s", blockHeight,
|
||||||
|
newBlock.Hash)
|
||||||
|
|
||||||
|
case <-d.quit:
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Stop signals any active goroutines for a graceful closure.
|
// Stop signals any active goroutines for a graceful closure.
|
||||||
func (d *AuthenticatedGossiper) Stop() error {
|
func (d *AuthenticatedGossiper) Stop() error {
|
||||||
d.stopped.Do(func() {
|
d.stopped.Do(func() {
|
||||||
@@ -1166,25 +1202,6 @@ func (d *AuthenticatedGossiper) networkHandler() {
|
|||||||
|
|
||||||
}()
|
}()
|
||||||
|
|
||||||
// A new block has arrived, so we can re-process the previously
|
|
||||||
// premature announcements.
|
|
||||||
case newBlock, ok := <-d.blockEpochs.Epochs:
|
|
||||||
// If the channel has been closed, then this indicates
|
|
||||||
// the daemon is shutting down, so we exit ourselves.
|
|
||||||
if !ok {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// Once a new block arrives, we update our running
|
|
||||||
// track of the height of the chain tip.
|
|
||||||
d.Lock()
|
|
||||||
blockHeight := uint32(newBlock.Height)
|
|
||||||
d.bestHeight = blockHeight
|
|
||||||
d.Unlock()
|
|
||||||
|
|
||||||
log.Debugf("New block: height=%d, hash=%s", blockHeight,
|
|
||||||
newBlock.Hash)
|
|
||||||
|
|
||||||
// The trickle timer has ticked, which indicates we should
|
// The trickle timer has ticked, which indicates we should
|
||||||
// flush to the network the pending batch of new announcements
|
// flush to the network the pending batch of new announcements
|
||||||
// we've received since the last trickle tick.
|
// we've received since the last trickle tick.
|
||||||
|
|||||||
Reference in new issue
Block a user