fixed starved download, removed obsolete invariant

libethereum/BlockChainSync.cpp

@@ -868,6 +868,7 @@ void PV61Sync::syncHashes(std::shared_ptr<EthereumPeer> _peer)
 		m_syncingBlockNumber = 0;
 		m_chainSyncPeers.clear();
 		m_knownHashes.clear();
+		m_hashScanComplete = false;
 		PV60Sync::syncHashes(_peer);
 		return;
 	}
@@ -913,7 +914,7 @@ void PV61Sync::requestSubchain(std::shared_ptr<EthereumPeer> _peer)
 			m_chainSyncPeers[_peer] = m_readyChainMap.begin()->first;
 			m_readyChainMap.erase(m_readyChainMap.begin());
 		}
-		else if (!m_downloadingChainMap.empty() && !m_completeChainMap.empty())
+		else if (!m_downloadingChainMap.empty() && m_hashScanComplete)
 		{
 			// Lead syncer is done, just grab whatever we can
 			h256s& d = m_downloadingChainMap.begin()->second;
@@ -936,6 +937,13 @@ void PV61Sync::completeSubchain(std::shared_ptr<EthereumPeer> _peer, unsigned _n
 {
 	m_completeChainMap[_n] = move(m_downloadingChainMap.at(_n));
 	m_downloadingChainMap.erase(_n);
+	for (auto s = m_chainSyncPeers.begin(); s != m_chainSyncPeers.end(); ++s)
+		if (s->second == _n) //TODO: optimize this
+		{
+			m_chainSyncPeers.erase(s);
+			break;
+		}
+
 	_peer->m_syncHashNumber = 0;
 
 	auto syncer = m_syncer.lock();
@@ -945,7 +953,7 @@ void PV61Sync::completeSubchain(std::shared_ptr<EthereumPeer> _peer, unsigned _n
 		return;
 	}
 
-	if (m_readyChainMap.empty() && m_downloadingChainMap.empty() && syncer->m_asking == Asking::Nothing)
+	if (m_readyChainMap.empty() && m_downloadingChainMap.empty() && m_hashScanComplete)
 	{
 		//Done chain-get
 		m_syncingNeededBlocks.clear();
@@ -968,6 +976,7 @@ void PV61Sync::restartSync()
 	m_chainSyncPeers.clear();
 	m_syncingBlockNumber = 0;
 	m_knownHashes.clear();
+	m_hashScanComplete = false;
 	PV60Sync::restartSync();
 }
 
@@ -986,6 +995,7 @@ void PV61Sync::onPeerHashes(std::shared_ptr<EthereumPeer> _peer, h256s const& _h
 		{
 			// End of hash chain, add last chunk to download
 			m_readyChainMap.insert(make_pair(m_syncingBlockNumber, h256s { _peer->m_latestHash }));
+			m_hashScanComplete = true;
 			_peer->m_syncHashNumber = 0;
 			requestSubchain(_peer);
 		}
@@ -1043,8 +1053,17 @@ void PV61Sync::onPeerHashes(std::shared_ptr<EthereumPeer> _peer, h256s const& _h
 			clog(NetAllDetail) << "Hashes response from unexpected/expired peer";
 			return;
 		}
-		h256s& hashes = m_downloadingChainMap.at(number);
 
+		auto downloadingPeer = m_downloadingChainMap.find(number);
+		if (downloadingPeer == m_downloadingChainMap.end() || downloadingPeer->second.back() != _peer->m_syncHash)
+		{
+			// Too late, other peer has already downloaded our hashes
+			m_chainSyncPeers.erase(_peer);
+			requestSubchain(_peer);
+			return;
+		}
+
+		h256s& hashes = downloadingPeer->second;
 		unsigned knowns = 0;
 		unsigned unknowns = 0;
 		for (unsigned i = 0; i < _hashes.size(); ++i)
@@ -1156,8 +1175,6 @@ bool PV61Sync::isPV61Syncing() const
 
 bool PV61Sync::invariants() const
 {
-	if (m_downloadingChainMap.size() != m_chainSyncPeers.size())
-		BOOST_THROW_EXCEPTION(FailedInvariant() << errinfo_comment("m_downloadingChainMap and m_chainSyncPeers out of sync"));
 	if (m_state == SyncState::Hashes)
 	{
 		if (isPV61Syncing() && !m_syncingBlockNumber)

libethereum/BlockChainSync.h

@@ -315,6 +315,7 @@ private:
 	std::map<std::weak_ptr<EthereumPeer>, unsigned, std::owner_less<std::weak_ptr<EthereumPeer>>> m_chainSyncPeers; ///< Peers to m_downloadingSubchain number map
 	h256Hash m_knownHashes;								///< Subchain start markers. Used to track suchain completion
 	unsigned m_syncingBlockNumber = 0;					///< Current subchain marker
+	bool m_hashScanComplete = false;						///< True if leading peer completed hashchain scan and we have a list of subchains ready
 };
 }
 }

libp2p/RLPXSocket.h

@@ -45,7 +45,7 @@ public:
 	
 	bool isConnected() const { return m_socket.is_open(); }
 	void close() { try { boost::system::error_code ec; m_socket.shutdown(bi::tcp::socket::shutdown_both, ec); if (m_socket.is_open()) m_socket.close(); } catch (...){} }
-	bi::tcp::endpoint remoteEndpoint() { try { return m_socket.remote_endpoint(); } catch (...){ return bi::tcp::endpoint(); } }
+	bi::tcp::endpoint remoteEndpoint() { boost::system::error_code ec; return m_socket.remote_endpoint(ec); }
 	bi::tcp::socket& ref() { return m_socket; }
 	
 protected:
@@ -53,4 +53,4 @@ protected:
 };
 
 }
-}
+}