roysc/plugeth
1
0
Fork 0
forked from cerc-io/plugeth
Code Pull Requests Activity

fix deadlock issue in AddBlock

Browse Source 
- add peer switch channel arg to activateChain - no peer locking within
- proper locking in AddBlock - fixes deadlock issue
- comment out TD check and skip incorrect TD test again for hotfix
This commit is contained in:
zelig 2015-04-08 20:33:54 +01:00
parent 262714fc6c
commit e55747a074
4 changed files with 46 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
blockpool/blockpool.go
View File

@ -377,7 +377,7 @@ func (self *BlockPool) AddBlockHashes(next func() (common.Hash, bool), peerId st
var nodes []*node var nodes []*node
hash, ok = next() hash, ok = next()
bestpeer.lock.Lock() bestpeer.lock.RLock()
plog.Debugf("AddBlockHashes: peer <%s> starting from [%s] (peer head: %s)", peerId, hex(bestpeer.parentHash), hex(bestpeer.currentBlockHash)) plog.Debugf("AddBlockHashes: peer <%s> starting from [%s] (peer head: %s)", peerId, hex(bestpeer.parentHash), hex(bestpeer.currentBlockHash))
@ -423,7 +423,7 @@ func (self *BlockPool) AddBlockHashes(next func() (common.Hash, bool), peerId st
} }
// the switch channel signals peerswitch event // the switch channel signals peerswitch event
switchC := bestpeer.switchC switchC := bestpeer.switchC
bestpeer.lock.Unlock() bestpeer.lock.RUnlock()
// iterate over hashes coming from peer (first round we have hash set above) // iterate over hashes coming from peer (first round we have hash set above)
LOOP: LOOP:
@ -549,8 +549,10 @@ LOOP:
In this case no activation should happen In this case no activation should happen
*/ */
if parent != nil && !peerswitch { if parent != nil && !peerswitch {
self.activateChain(parent, bestpeer, nil) bestpeer.lock.RLock()
self.activateChain(parent, bestpeer, bestpeer.switchC, nil)
plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): parent section [%s]", peerId, hex(bestpeer.currentBlockHash), sectionhex(parent)) plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): parent section [%s]", peerId, hex(bestpeer.currentBlockHash), sectionhex(parent))
bestpeer.lock.RUnlock()
} }
/* /*
@ -625,33 +627,40 @@ func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
entry := self.get(hash) entry := self.get(hash)
blockIsCurrentHead := false blockIsCurrentHead := false
sender.lock.Lock() sender.lock.RLock()
currentBlockHash := sender.currentBlockHash
currentBlock := sender.currentBlock
currentBlockC := sender.currentBlockC
switchC := sender.switchC
sender.lock.RUnlock()
// a peer's current head block is appearing the first time // a peer's current head block is appearing the first time
if hash == sender.currentBlockHash { if hash == currentBlockHash {
// this happens when block came in a newblock message but // this happens when block came in a newblock message but
// also if sent in a blockmsg (for instance, if we requested, only if we // also if sent in a blockmsg (for instance, if we requested, only if we
// dont apply on blockrequests the restriction of flood control) // dont apply on blockrequests the restriction of flood control)
blockIsCurrentHead = true blockIsCurrentHead = true
if sender.currentBlock == nil { if currentBlock == nil {
plog.Debugf("AddBlock: add head block %s for peer <%s> (head: %s)", hex(hash), peerId, hex(sender.currentBlockHash)) sender.lock.Lock()
sender.setChainInfoFromBlock(block) sender.setChainInfoFromBlock(block)
sender.lock.Unlock()
self.status.lock.Lock() self.status.lock.Lock()
self.status.values.BlockHashes++ self.status.values.BlockHashes++
self.status.values.Blocks++ self.status.values.Blocks++
self.status.values.BlocksInPool++ self.status.values.BlocksInPool++
self.status.lock.Unlock() self.status.lock.Unlock()
// signal to head section process
select { select {
case sender.currentBlockC <- block: case currentBlockC <- block:
case <-sender.switchC: case <-switchC:
} }
} else { } else {
plog.DebugDetailf("AddBlock: head block %s for peer <%s> (head: %s) already known", hex(hash), peerId, hex(sender.currentBlockHash)) plog.DebugDetailf("AddBlock: head block %s for peer <%s> (head: %s) already known", hex(hash), peerId, hex(currentBlockHash))
// signal to head section process
} }
} else { } else {
plog.DebugDetailf("AddBlock: block %s received from peer <%s> (head: %s)", hex(hash), peerId, hex(sender.currentBlockHash)) plog.DebugDetailf("AddBlock: block %s received from peer <%s> (head: %s)", hex(hash), peerId, hex(currentBlockHash))
/* @zelig !!! /* @zelig !!!
requested 5 hashes from both A & B. A responds sooner then B, process blocks. Close section. requested 5 hashes from both A & B. A responds sooner then B, process blocks. Close section.
@ -667,7 +676,6 @@ func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
} }
*/ */
} }
sender.lock.Unlock()
if entry == nil { if entry == nil {
// FIXME: here check the cache find or create node - // FIXME: here check the cache find or create node -
@ -721,7 +729,7 @@ func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
*/ */
node.block = block node.block = block
// node.blockBy = peerId node.blockBy = peerId
self.status.lock.Lock() self.status.lock.Lock()
self.status.values.Blocks++ self.status.values.Blocks++
@ -735,11 +743,7 @@ func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
It activates the section process on incomplete sections with peer. It activates the section process on incomplete sections with peer.
It relinks orphaned sections with their parent if root block (and its parent hash) is known. It relinks orphaned sections with their parent if root block (and its parent hash) is known.
*/ */
func (self *BlockPool) activateChain(sec *section, p *peer, connected map[common.Hash]*section) { func (self *BlockPool) activateChain(sec *section, p *peer, switchC chan bool, connected map[common.Hash]*section) {
p.lock.RLock()
switchC := p.switchC
p.lock.RUnlock()
var i int var i int
@ -786,10 +790,10 @@ func (self *BlockPool) checkTD(nodes ...*node) {
if n.td != nil && !n.block.Queued() { if n.td != nil && !n.block.Queued() {
plog.DebugDetailf("peer td %v =?= block td %v", n.td, n.block.Td) plog.DebugDetailf("peer td %v =?= block td %v", n.td, n.block.Td)
if n.td.Cmp(n.block.Td) != 0 { if n.td.Cmp(n.block.Td) != 0 {
self.peers.peerError(n.blockBy, ErrIncorrectTD, "on block %x", n.hash) // self.peers.peerError(n.blockBy, ErrIncorrectTD, "on block %x", n.hash)
self.status.lock.Lock() // self.status.lock.Lock()
self.status.badPeers[n.blockBy]++ // self.status.badPeers[n.blockBy]++
self.status.lock.Unlock() // self.status.lock.Unlock()
} }
} }
} }

2
blockpool/errors_test.go
View File

@ -128,7 +128,7 @@ func TestErrInsufficientChainInfo(t *testing.T) {
} }
func TestIncorrectTD(t *testing.T) { func TestIncorrectTD(t *testing.T) {
// t.Skip() // @zelig this one requires fixing for the TD t.Skip() // @zelig this one requires fixing for the TD
test.LogInit() test.LogInit()
_, blockPool, blockPoolTester := newTestBlockPool(t) _, blockPool, blockPoolTester := newTestBlockPool(t)

31
blockpool/peers.go
View File

@ -114,10 +114,8 @@ func (self *peers) addToBlacklist(id string) {
self.blacklist[id] = time.Now() self.blacklist[id] = time.Now()
} }
// suspended checks if peer is still suspended // suspended checks if peer is still suspended, caller should hold peers.lock
func (self *peers) suspended(id string) (s bool) { func (self *peers) suspended(id string) (s bool) {
self.lock.Lock()
defer self.lock.Unlock()
if suspendedAt, ok := self.blacklist[id]; ok { if suspendedAt, ok := self.blacklist[id]; ok {
if s = suspendedAt.Add(self.bp.Config.PeerSuspensionInterval).After(time.Now()); !s { if s = suspendedAt.Add(self.bp.Config.PeerSuspensionInterval).After(time.Now()); !s {
// no longer suspended, delete entry // no longer suspended, delete entry
@ -205,13 +203,14 @@ func (self *peers) addPeer(
peerError func(*errs.Error), peerError func(*errs.Error),
) (best bool, suspended bool) { ) (best bool, suspended bool) {
self.lock.Lock()
defer self.lock.Unlock()
var previousBlockHash common.Hash var previousBlockHash common.Hash
if self.suspended(id) { if self.suspended(id) {
suspended = true suspended = true
return return
} }
self.lock.Lock()
defer self.lock.Unlock()
p, found := self.peers[id] p, found := self.peers[id]
if found { if found {
// when called on an already connected peer, it means a newBlockMsg is received // when called on an already connected peer, it means a newBlockMsg is received
@ -255,7 +254,7 @@ func (self *peers) addPeer(
p.headSectionC <- nil p.headSectionC <- nil
if entry := self.bp.get(previousBlockHash); entry != nil { if entry := self.bp.get(previousBlockHash); entry != nil {
plog.DebugDetailf("addPeer: <%s> previous head : %v found in pool, activate", id, hex(previousBlockHash)) plog.DebugDetailf("addPeer: <%s> previous head : %v found in pool, activate", id, hex(previousBlockHash))
self.bp.activateChain(entry.section, p, nil) self.bp.activateChain(entry.section, p, p.switchC, nil)
p.sections = append(p.sections, previousBlockHash) p.sections = append(p.sections, previousBlockHash)
} }
} }
@ -265,8 +264,8 @@ func (self *peers) addPeer(
currentTD := self.bp.getTD() currentTD := self.bp.getTD()
bestpeer := self.best bestpeer := self.best
if bestpeer != nil { if bestpeer != nil {
bestpeer.lock.Lock() bestpeer.lock.RLock()
defer bestpeer.lock.Unlock() defer bestpeer.lock.RUnlock()
currentTD = self.best.td currentTD = self.best.td
} }
if td.Cmp(currentTD) > 0 { if td.Cmp(currentTD) > 0 {
@ -362,14 +361,14 @@ func (self *BlockPool) switchPeer(oldp, newp *peer) {
if connected[hash] == nil { if connected[hash] == nil {
// if not deleted, then reread from pool (it can be orphaned top half of a split section) // if not deleted, then reread from pool (it can be orphaned top half of a split section)
if entry := self.get(hash); entry != nil { if entry := self.get(hash); entry != nil {
self.activateChain(entry.section, newp, connected) self.activateChain(entry.section, newp, newp.switchC, connected)
connected[hash] = entry.section connected[hash] = entry.section
sections = append(sections, hash) sections = append(sections, hash)
} }
} }
} }
plog.DebugDetailf("<%s> section processes (%v non-contiguous sequences, was %v before)", newp.id, len(sections), len(newp.sections)) plog.DebugDetailf("<%s> section processes (%v non-contiguous sequences, was %v before)", newp.id, len(sections), len(newp.sections))
// need to lock now that newp is exposed to section processes // need to lock now that newp is exposed to section processesr
newp.lock.Lock() newp.lock.Lock()
newp.sections = sections newp.sections = sections
newp.lock.Unlock() newp.lock.Unlock()
@ -457,6 +456,8 @@ func (self *peer) getCurrentBlock(currentBlock *types.Block) {
} }
func (self *peer) getBlockHashes() bool { func (self *peer) getBlockHashes() bool {
self.lock.Lock()
defer self.lock.Unlock()
//if connecting parent is found //if connecting parent is found
if self.bp.hasBlock(self.parentHash) { if self.bp.hasBlock(self.parentHash) {
plog.DebugDetailf("HeadSection: <%s> parent block %s found in blockchain", self.id, hex(self.parentHash)) plog.DebugDetailf("HeadSection: <%s> parent block %s found in blockchain", self.id, hex(self.parentHash))
@ -470,10 +471,10 @@ func (self *peer) getBlockHashes() bool {
self.bp.status.badPeers[self.id]++ self.bp.status.badPeers[self.id]++
} else { } else {
// XXX added currentBlock check (?) // XXX added currentBlock check (?)
if self.currentBlock != nil && self.currentBlock.Td != nil { if self.currentBlock != nil && self.currentBlock.Td != nil && !self.currentBlock.Queued() {
if self.td.Cmp(self.currentBlock.Td) != 0 { if self.td.Cmp(self.currentBlock.Td) != 0 {
//self.addError(ErrIncorrectTD, "on block %x", self.currentBlockHash) // self.addError(ErrIncorrectTD, "on block %x", self.currentBlockHash)
//self.bp.status.badPeers[self.id]++ // self.bp.status.badPeers[self.id]++
} }
} }
headKey := self.parentHash headKey := self.parentHash
@ -499,7 +500,7 @@ func (self *peer) getBlockHashes() bool {
self.bp.newSection([]*node{n}).activate(self) self.bp.newSection([]*node{n}).activate(self)
} else { } else {
plog.DebugDetailf("HeadSection: <%s> connecting parent %s found in pool...head section [%s] exists...not requesting hashes", self.id, hex(self.parentHash), sectionhex(parent.section)) plog.DebugDetailf("HeadSection: <%s> connecting parent %s found in pool...head section [%s] exists...not requesting hashes", self.id, hex(self.parentHash), sectionhex(parent.section))
self.bp.activateChain(parent.section, self, nil) self.bp.activateChain(parent.section, self, self.switchC, nil)
} }
} else { } else {
plog.DebugDetailf("HeadSection: <%s> section [%s] requestBlockHashes", self.id, sectionhex(self.headSection)) plog.DebugDetailf("HeadSection: <%s> section [%s] requestBlockHashes", self.id, sectionhex(self.headSection))
@ -523,6 +524,7 @@ func (self *peer) run() {
self.lock.RLock() self.lock.RLock()
switchC := self.switchC switchC := self.switchC
plog.Debugf("HeadSection: <%s> section process for head %s started", self.id, hex(self.currentBlockHash))
self.lock.RUnlock() self.lock.RUnlock()
self.blockHashesRequestTimer = nil self.blockHashesRequestTimer = nil
@ -589,6 +591,7 @@ LOOP:
plog.Debugf("HeadSection: <%s> (headsection [%s]) quit channel closed : timed out without providing new blocks...quitting", self.id, sectionhex(self.headSection)) plog.Debugf("HeadSection: <%s> (headsection [%s]) quit channel closed : timed out without providing new blocks...quitting", self.id, sectionhex(self.headSection))
} }
} }
if !self.idle { if !self.idle {
self.idle = true self.idle = true
self.bp.wg.Done() self.bp.wg.Done()

2
blockpool/section.go
View File

@ -489,7 +489,7 @@ func (self *section) blockHashesRequest() {
// activate parent section with this peer // activate parent section with this peer
// but only if not during switch mode // but only if not during switch mode
plog.DebugDetailf("[%s] parent section [%s] activated\n", sectionhex(self), sectionhex(parentSection)) plog.DebugDetailf("[%s] parent section [%s] activated\n", sectionhex(self), sectionhex(parentSection))
self.bp.activateChain(parentSection, self.peer, nil) self.bp.activateChain(parentSection, self.peer, self.peer.switchC, nil)
// if not root of chain, switch off // if not root of chain, switch off
plog.DebugDetailf("[%s] parent found, hash requests deactivated (after %v total attempts)\n", sectionhex(self), self.blockHashesRequests) plog.DebugDetailf("[%s] parent found, hash requests deactivated (after %v total attempts)\n", sectionhex(self), self.blockHashesRequests)
self.blockHashesRequestTimer = nil self.blockHashesRequestTimer = nil