Convert block manager block handler to pure FIFO.
Rather than having a separate query channel for the block manager, use the same channel so the block handler acts as a pure FIFO queue. This prevents possible starvation of query related messages. ok @owainga
This commit is contained in:
parent
33af740985
commit
b661dd3693
1 changed files with 4 additions and 14 deletions
|
@ -70,7 +70,7 @@ type txMsg struct {
|
||||||
peer *peer
|
peer *peer
|
||||||
}
|
}
|
||||||
|
|
||||||
// getSyncPeerMsg is a message type to be sent across the query channel for
|
// getSyncPeerMsg is a message type to be sent across the message channel for
|
||||||
// retrieving the current sync peer.
|
// retrieving the current sync peer.
|
||||||
type getSyncPeerMsg struct {
|
type getSyncPeerMsg struct {
|
||||||
reply chan *peer
|
reply chan *peer
|
||||||
|
@ -99,7 +99,6 @@ type blockManager struct {
|
||||||
processingReqs bool
|
processingReqs bool
|
||||||
syncPeer *peer
|
syncPeer *peer
|
||||||
msgChan chan interface{}
|
msgChan chan interface{}
|
||||||
query chan interface{}
|
|
||||||
wg sync.WaitGroup
|
wg sync.WaitGroup
|
||||||
quit chan bool
|
quit chan bool
|
||||||
|
|
||||||
|
@ -905,20 +904,12 @@ out:
|
||||||
case *donePeerMsg:
|
case *donePeerMsg:
|
||||||
b.handleDonePeerMsg(candidatePeers, msg.peer)
|
b.handleDonePeerMsg(candidatePeers, msg.peer)
|
||||||
|
|
||||||
default:
|
|
||||||
bmgrLog.Warnf("Invalid message type in block "+
|
|
||||||
"handler: %T", msg)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Queries used for atomically retrieving internal state.
|
|
||||||
case qmsg := <-b.query:
|
|
||||||
switch msg := qmsg.(type) {
|
|
||||||
case getSyncPeerMsg:
|
case getSyncPeerMsg:
|
||||||
msg.reply <- b.syncPeer
|
msg.reply <- b.syncPeer
|
||||||
|
|
||||||
default:
|
default:
|
||||||
bmgrLog.Warnf("Invalid query type in block "+
|
bmgrLog.Warnf("Invalid message type in block "+
|
||||||
"handler query: %T", msg)
|
"handler: %T", msg)
|
||||||
}
|
}
|
||||||
|
|
||||||
case <-b.quit:
|
case <-b.quit:
|
||||||
|
@ -1123,7 +1114,7 @@ func (b *blockManager) Stop() error {
|
||||||
// SyncPeer returns the current sync peer.
|
// SyncPeer returns the current sync peer.
|
||||||
func (b *blockManager) SyncPeer() *peer {
|
func (b *blockManager) SyncPeer() *peer {
|
||||||
reply := make(chan *peer)
|
reply := make(chan *peer)
|
||||||
b.query <- getSyncPeerMsg{reply: reply}
|
b.msgChan <- getSyncPeerMsg{reply: reply}
|
||||||
return <-reply
|
return <-reply
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -1142,7 +1133,6 @@ func newBlockManager(s *server) (*blockManager, error) {
|
||||||
requestedBlocks: make(map[btcwire.ShaHash]bool),
|
requestedBlocks: make(map[btcwire.ShaHash]bool),
|
||||||
lastBlockLogTime: time.Now(),
|
lastBlockLogTime: time.Now(),
|
||||||
msgChan: make(chan interface{}, cfg.MaxPeers*3),
|
msgChan: make(chan interface{}, cfg.MaxPeers*3),
|
||||||
query: make(chan interface{}),
|
|
||||||
headerList: list.New(),
|
headerList: list.New(),
|
||||||
quit: make(chan bool),
|
quit: make(chan bool),
|
||||||
}
|
}
|
||||||
|
|
Loading…
Add table
Reference in a new issue