From 44147d057dd91d8b35dd6f4ed025bdb4baf225eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 9 Jun 2015 14:27:44 +0300 Subject: [PATCH 1/7] eth: fix data race accessing peer.recentHash --- eth/handler.go | 6 +++--- eth/peer.go | 31 +++++++++++++++++++++++++------ eth/sync.go | 7 ++++--- 3 files changed, 32 insertions(+), 12 deletions(-) diff --git a/eth/handler.go b/eth/handler.go index 64f89b273..847e7a0e8 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -157,7 +157,7 @@ func (pm *ProtocolManager) handle(p *peer) error { } defer pm.removePeer(p.id) - if err := pm.downloader.RegisterPeer(p.id, p.recentHash, p.requestHashes, p.requestBlocks); err != nil { + if err := pm.downloader.RegisterPeer(p.id, p.Head(), p.requestHashes, p.requestBlocks); err != nil { return err } // propagate existing transactions. new transactions appearing @@ -303,7 +303,7 @@ func (self *ProtocolManager) handleMsg(p *peer) error { // Mark the hashes as present at the remote node for _, hash := range hashes { p.blockHashes.Add(hash) - p.recentHash = hash + p.SetHead(hash) } // Schedule all the unknown hashes for retrieval unknown := make([]common.Hash, 0, len(hashes)) @@ -354,7 +354,7 @@ func (pm *ProtocolManager) importBlock(p *peer, block *types.Block, td *big.Int) // Mark the block as present at the remote node (don't duplicate already held data) p.blockHashes.Add(hash) - p.recentHash = hash + p.SetHead(hash) if td != nil { p.td = td } diff --git a/eth/peer.go b/eth/peer.go index cf2c58ec9..5f5007891 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -40,9 +40,11 @@ type peer struct { protv, netid int - recentHash common.Hash - id string - td *big.Int + id string + td *big.Int + + head common.Hash + headLock sync.RWMutex genesis, ourHash common.Hash ourTd *big.Int @@ -51,14 +53,14 @@ type peer struct { blockHashes *set.Set } -func newPeer(protv, netid int, genesis, recentHash common.Hash, td *big.Int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { +func newPeer(protv, netid int, genesis, head common.Hash, td *big.Int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { id := p.ID() return &peer{ Peer: p, rw: rw, genesis: genesis, - ourHash: recentHash, + ourHash: head, ourTd: td, protv: protv, netid: netid, @@ -68,6 +70,23 @@ func newPeer(protv, netid int, genesis, recentHash common.Hash, td *big.Int, p * } } +// Head retrieves a copy of the current head (most recent) hash of the peer. +func (p *peer) Head() (hash common.Hash) { + p.headLock.RLock() + defer p.headLock.RUnlock() + + copy(hash[:], p.head[:]) + return hash +} + +// SetHead updates the head (most recent) hash of the peer. +func (p *peer) SetHead(hash common.Hash) { + p.headLock.Lock() + defer p.headLock.Unlock() + + copy(p.head[:], hash[:]) +} + // sendTransactions sends transactions to the peer and includes the hashes // in it's tx hash set for future reference. The tx hash will allow the // manager to check whether the peer has already received this particular @@ -160,7 +179,7 @@ func (p *peer) handleStatus() error { // Set the total difficulty of the peer p.td = status.TD // set the best hash of the peer - p.recentHash = status.CurrentBlock + p.head = status.CurrentBlock return <-errc } diff --git a/eth/sync.go b/eth/sync.go index dd7414da8..b3184364f 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -214,14 +214,15 @@ func (pm *ProtocolManager) synchronise(peer *peer) { // FIXME if we have the hash in our chain and the TD of the peer is // much higher than ours, something is wrong with us or the peer. // Check if the hash is on our own chain - if pm.chainman.HasBlock(peer.recentHash) { + head := peer.Head() + if pm.chainman.HasBlock(head) { glog.V(logger.Debug).Infoln("Synchronisation canceled: head already known") return } // Get the hashes from the peer (synchronously) - glog.V(logger.Detail).Infof("Attempting synchronisation: %v, 0x%x", peer.id, peer.recentHash) + glog.V(logger.Detail).Infof("Attempting synchronisation: %v, 0x%x", peer.id, head) - err := pm.downloader.Synchronise(peer.id, peer.recentHash) + err := pm.downloader.Synchronise(peer.id, head) switch err { case nil: glog.V(logger.Detail).Infof("Synchronisation completed") From f86707713c42da02b70a3a389684e19e902d8759 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 9 Jun 2015 14:56:27 +0300 Subject: [PATCH 2/7] eth: fix data race accessing peer.td --- eth/handler.go | 4 ++-- eth/peer.go | 41 ++++++++++++++++++++++++++++++----------- eth/sync.go | 2 +- 3 files changed, 33 insertions(+), 14 deletions(-) diff --git a/eth/handler.go b/eth/handler.go index 847e7a0e8..f2027c3c6 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -356,7 +356,7 @@ func (pm *ProtocolManager) importBlock(p *peer, block *types.Block, td *big.Int) p.blockHashes.Add(hash) p.SetHead(hash) if td != nil { - p.td = td + p.SetTd(td) } // Log the block's arrival _, chainHead, _ := pm.chainman.Status() @@ -369,7 +369,7 @@ func (pm *ProtocolManager) importBlock(p *peer, block *types.Block, td *big.Int) }) // If the block's already known or its difficulty is lower than ours, drop if pm.chainman.HasBlock(hash) { - p.td = pm.chainman.GetBlock(hash).Td // update the peer's TD to the real value + p.SetTd(pm.chainman.GetBlock(hash).Td) // update the peer's TD to the real value return nil } if td != nil && pm.chainman.Td().Cmp(td) > 0 && new(big.Int).Add(block.Number(), big.NewInt(7)).Cmp(pm.chainman.CurrentBlock().Number()) < 0 { diff --git a/eth/peer.go b/eth/peer.go index 5f5007891..c7045282b 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -41,10 +41,10 @@ type peer struct { protv, netid int id string - td *big.Int - head common.Hash - headLock sync.RWMutex + head common.Hash + td *big.Int + lock sync.RWMutex genesis, ourHash common.Hash ourTd *big.Int @@ -72,8 +72,8 @@ func newPeer(protv, netid int, genesis, head common.Hash, td *big.Int, p *p2p.Pe // Head retrieves a copy of the current head (most recent) hash of the peer. func (p *peer) Head() (hash common.Hash) { - p.headLock.RLock() - defer p.headLock.RUnlock() + p.lock.RLock() + defer p.lock.RUnlock() copy(hash[:], p.head[:]) return hash @@ -81,12 +81,28 @@ func (p *peer) Head() (hash common.Hash) { // SetHead updates the head (most recent) hash of the peer. func (p *peer) SetHead(hash common.Hash) { - p.headLock.Lock() - defer p.headLock.Unlock() + p.lock.Lock() + defer p.lock.Unlock() copy(p.head[:], hash[:]) } +// Td retrieves the current total difficulty of a peer. +func (p *peer) Td() *big.Int { + p.lock.RLock() + defer p.lock.RUnlock() + + return new(big.Int).Set(p.td) +} + +// SetTd updates the current total difficulty of a peer. +func (p *peer) SetTd(td *big.Int) { + p.lock.Lock() + defer p.lock.Unlock() + + p.td.Set(td) +} + // sendTransactions sends transactions to the peer and includes the hashes // in it's tx hash set for future reference. The tx hash will allow the // manager to check whether the peer has already received this particular @@ -275,11 +291,14 @@ func (ps *peerSet) BestPeer() *peer { ps.lock.RLock() defer ps.lock.RUnlock() - var best *peer + var ( + bestPeer *peer + bestTd *big.Int + ) for _, p := range ps.peers { - if best == nil || p.td.Cmp(best.td) > 0 { - best = p + if td := p.Td(); bestPeer == nil || td.Cmp(bestTd) > 0 { + bestPeer, bestTd = p, td } } - return best + return bestPeer } diff --git a/eth/sync.go b/eth/sync.go index b3184364f..3a33fe149 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -208,7 +208,7 @@ func (pm *ProtocolManager) synchronise(peer *peer) { return } // Make sure the peer's TD is higher than our own. If not drop. - if peer.td.Cmp(pm.chainman.Td()) <= 0 { + if peer.Td().Cmp(pm.chainman.Td()) <= 0 { return } // FIXME if we have the hash in our chain and the TD of the peer is From d09ead546cbdf8e4659e65581f23715101f5b686 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 9 Jun 2015 15:09:15 +0300 Subject: [PATCH 3/7] eth: fix a data race in the hash announcement processing --- eth/sync.go | 28 ++++++++++++++++++---------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/eth/sync.go b/eth/sync.go index 3a33fe149..8e4e3cfbe 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -109,17 +109,25 @@ func (pm *ProtocolManager) fetcher() { // If any explicit fetches were replied to, import them if count := len(explicit); count > 0 { glog.V(logger.Debug).Infof("Importing %d explicitly fetched blocks", count) - go func() { - for _, block := range explicit { - hash := block.Hash() - // Make sure there's still something pending to import - if announce := pending[hash]; announce != nil { - delete(pending, hash) - if err := pm.importBlock(announce.peer, block, nil); err != nil { - glog.V(logger.Detail).Infof("Failed to import explicitly fetched block: %v", err) - return - } + // Create a closure with the retrieved blocks and origin peers + peers := make([]*peer, 0, count) + blocks := make([]*types.Block, 0, count) + for _, block := range explicit { + hash := block.Hash() + if announce := pending[hash]; announce != nil { + peers = append(peers, announce.peer) + blocks = append(blocks, block) + + delete(pending, hash) + } + } + // Run the importer on a new thread + go func() { + for i := 0; i < len(blocks); i++ { + if err := pm.importBlock(peers[i], blocks[i], nil); err != nil { + glog.V(logger.Detail).Infof("Failed to import explicitly fetched block: %v", err) + return } } }() From 07baf66200c74a97b440a199dce7321b23aea4cd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 9 Jun 2015 15:23:20 +0300 Subject: [PATCH 4/7] core: fix data race in accessing ChainManager.td --- core/chain_manager.go | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/core/chain_manager.go b/core/chain_manager.go index a0ce20006..2333368de 100644 --- a/core/chain_manager.go +++ b/core/chain_manager.go @@ -56,10 +56,7 @@ func CalcTD(block, parent *types.Block) *big.Int { if parent == nil { return block.Difficulty() } - - td := new(big.Int).Add(parent.Td, block.Header().Difficulty) - - return td + return new(big.Int).Add(parent.Td, block.Header().Difficulty) } func CalcGasLimit(parent *types.Block) *big.Int { @@ -178,7 +175,7 @@ func (self *ChainManager) Td() *big.Int { self.mu.RLock() defer self.mu.RUnlock() - return self.td + return new(big.Int).Set(self.td) } func (self *ChainManager) GasLimit() *big.Int { @@ -204,7 +201,7 @@ func (self *ChainManager) Status() (td *big.Int, currentBlock common.Hash, genes self.mu.RLock() defer self.mu.RUnlock() - return self.td, self.currentBlock.Hash(), self.genesisBlock.Hash() + return new(big.Int).Set(self.td), self.currentBlock.Hash(), self.genesisBlock.Hash() } func (self *ChainManager) SetProcessor(proc types.BlockProcessor) { @@ -488,8 +485,10 @@ func (self *ChainManager) GetAncestors(block *types.Block, length int) (blocks [ } func (bc *ChainManager) setTotalDifficulty(td *big.Int) { - //bc.blockDb.Put([]byte("LTD"), td.Bytes()) - bc.td = td + bc.mu.Lock() + defer bc.mu.Unlock() + + bc.td.Set(td) } func (self *ChainManager) CalcTotalDiff(block *types.Block) (*big.Int, error) { @@ -626,7 +625,7 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) { cblock := self.currentBlock // Compare the TD of the last known block in the canonical chain to make sure it's greater. // At this point it's possible that a different chain (fork) becomes the new canonical chain. - if block.Td.Cmp(self.td) > 0 { + if block.Td.Cmp(self.Td()) > 0 { // chain fork if block.ParentHash() != cblock.Hash() { // during split we merge two different chains and create the new canonical chain From ca8cb65b73b5bdb6a30b6a45304b3c45acc66bcc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 9 Jun 2015 15:30:46 +0300 Subject: [PATCH 5/7] core: fix data race accessing ChainManager.currentBlock --- core/chain_manager.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/core/chain_manager.go b/core/chain_manager.go index 2333368de..2ba81550e 100644 --- a/core/chain_manager.go +++ b/core/chain_manager.go @@ -379,8 +379,11 @@ func (self *ChainManager) ExportN(w io.Writer, first uint64, last uint64) error func (bc *ChainManager) insert(block *types.Block) { key := append(blockNumPre, block.Number().Bytes()...) bc.blockDb.Put(key, block.Hash().Bytes()) - bc.blockDb.Put([]byte("LastBlock"), block.Hash().Bytes()) + + bc.mu.Lock() + defer bc.mu.Unlock() + bc.currentBlock = block bc.lastBlockHash = block.Hash() } From ff84352fb73bdbb07a7b2cf24b417927bf5a5c5e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 9 Jun 2015 16:05:40 +0300 Subject: [PATCH 6/7] p2p: fix close data race --- p2p/rlpx.go | 1 + 1 file changed, 1 insertion(+) diff --git a/p2p/rlpx.go b/p2p/rlpx.go index e1cb13aae..6bbf20671 100644 --- a/p2p/rlpx.go +++ b/p2p/rlpx.go @@ -102,6 +102,7 @@ func (t *rlpx) doProtoHandshake(our *protoHandshake) (their *protoHandshake, err werr := make(chan error, 1) go func() { werr <- Send(t.rw, handshakeMsg, our) }() if their, err = readProtocolHandshake(t.rw, our); err != nil { + <-werr // make sure the write terminates too return nil, err } if err := <-werr; err != nil { From ebf2aabd254a4e765b68cdb46b18806fa7e4cb4b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 9 Jun 2015 16:26:44 +0300 Subject: [PATCH 7/7] core: fix up a deadlock caused by double locking --- core/chain_manager.go | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/core/chain_manager.go b/core/chain_manager.go index 2ba81550e..c69d3a10e 100644 --- a/core/chain_manager.go +++ b/core/chain_manager.go @@ -381,9 +381,6 @@ func (bc *ChainManager) insert(block *types.Block) { bc.blockDb.Put(key, block.Hash().Bytes()) bc.blockDb.Put([]byte("LastBlock"), block.Hash().Bytes()) - bc.mu.Lock() - defer bc.mu.Unlock() - bc.currentBlock = block bc.lastBlockHash = block.Hash() } @@ -488,10 +485,7 @@ func (self *ChainManager) GetAncestors(block *types.Block, length int) (blocks [ } func (bc *ChainManager) setTotalDifficulty(td *big.Int) { - bc.mu.Lock() - defer bc.mu.Unlock() - - bc.td.Set(td) + bc.td = new(big.Int).Set(td) } func (self *ChainManager) CalcTotalDiff(block *types.Block) (*big.Int, error) { @@ -546,6 +540,9 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) { self.wg.Add(1) defer self.wg.Done() + self.mu.Lock() + defer self.mu.Unlock() + self.chainmu.Lock() defer self.chainmu.Unlock() @@ -628,7 +625,7 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) { cblock := self.currentBlock // Compare the TD of the last known block in the canonical chain to make sure it's greater. // At this point it's possible that a different chain (fork) becomes the new canonical chain. - if block.Td.Cmp(self.Td()) > 0 { + if block.Td.Cmp(self.td) > 0 { // chain fork if block.ParentHash() != cblock.Hash() { // during split we merge two different chains and create the new canonical chain