Gav Wood
9 years ago
13 changed files with 1148 additions and 636 deletions
@ -0,0 +1,781 @@ |
|||
/*
|
|||
This file is part of cpp-ethereum. |
|||
|
|||
cpp-ethereum is free software: you can redistribute it and/or modify |
|||
it under the terms of the GNU General Public License as published by |
|||
the Free Software Foundation, either version 3 of the License, or |
|||
(at your option) any later version. |
|||
|
|||
cpp-ethereum is distributed in the hope that it will be useful, |
|||
but WITHOUT ANY WARRANTY; without even the implied warranty of |
|||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|||
GNU General Public License for more details. |
|||
|
|||
You should have received a copy of the GNU General Public License |
|||
along with cpp-ethereum. If not, see <http://www.gnu.org/licenses/>.
|
|||
*/ |
|||
/** @file BlockChainSync.cpp
|
|||
* @author Gav Wood <i@gavwood.com> |
|||
* @date 2014 |
|||
*/ |
|||
|
|||
#include "BlockChainSync.h" |
|||
|
|||
#include <chrono> |
|||
#include <libdevcore/Common.h> |
|||
#include <libp2p/Host.h> |
|||
#include <libp2p/Session.h> |
|||
#include <libethcore/Exceptions.h> |
|||
#include <libethcore/Params.h> |
|||
#include "BlockChain.h" |
|||
#include "BlockQueue.h" |
|||
#include "EthereumPeer.h" |
|||
#include "EthereumHost.h" |
|||
#include "DownloadMan.h" |
|||
|
|||
using namespace std; |
|||
using namespace dev; |
|||
using namespace dev::eth; |
|||
using namespace p2p; |
|||
|
|||
unsigned const c_chainReorgSize = 30000; |
|||
|
|||
BlockChainSync::BlockChainSync(EthereumHost& _host): |
|||
m_host(_host) |
|||
{ |
|||
m_bqRoomAvailable = host().bq().onRoomAvailable([this]() |
|||
{ |
|||
RecursiveGuard l(x_sync); |
|||
continueSync(); |
|||
}); |
|||
} |
|||
|
|||
BlockChainSync::~BlockChainSync() |
|||
{ |
|||
abortSync(); |
|||
} |
|||
|
|||
DownloadMan const& BlockChainSync::downloadMan() const |
|||
{ |
|||
return host().downloadMan(); |
|||
} |
|||
|
|||
DownloadMan& BlockChainSync::downloadMan() |
|||
{ |
|||
return host().downloadMan(); |
|||
} |
|||
|
|||
void BlockChainSync::abortSync() |
|||
{ |
|||
host().foreachPeer([this](EthereumPeer* _p) { onPeerAborting(_p); return true; }); |
|||
downloadMan().resetToChain(h256s()); |
|||
} |
|||
|
|||
void BlockChainSync::onPeerStatus(EthereumPeer* _peer) |
|||
{ |
|||
RecursiveGuard l(x_sync); |
|||
DEV_INVARIANT_CHECK; |
|||
if (_peer->m_genesisHash != host().chain().genesisHash()) |
|||
_peer->disable("Invalid genesis hash"); |
|||
else if (_peer->m_protocolVersion != host().protocolVersion() && _peer->m_protocolVersion != EthereumHost::c_oldProtocolVersion) |
|||
_peer->disable("Invalid protocol version."); |
|||
else if (_peer->m_networkId != host().networkId()) |
|||
_peer->disable("Invalid network identifier."); |
|||
else if (_peer->session()->info().clientVersion.find("/v0.7.0/") != string::npos) |
|||
_peer->disable("Blacklisted client version."); |
|||
else if (host().isBanned(_peer->session()->id())) |
|||
_peer->disable("Peer banned for previous bad behaviour."); |
|||
else |
|||
{ |
|||
unsigned estimatedHashes = estimateHashes(); |
|||
_peer->m_expectedHashes = estimatedHashes; |
|||
onNewPeer(_peer); |
|||
} |
|||
DEV_INVARIANT_CHECK; |
|||
} |
|||
|
|||
unsigned BlockChainSync::estimateHashes() |
|||
{ |
|||
BlockInfo block = host().chain().info(); |
|||
time_t lastBlockTime = (block.hash() == host().chain().genesisHash()) ? 1428192000 : (time_t)block.timestamp; |
|||
time_t now = time(0); |
|||
unsigned blockCount = c_chainReorgSize; |
|||
if (lastBlockTime > now) |
|||
clog(NetWarn) << "Clock skew? Latest block is in the future"; |
|||
else |
|||
blockCount += (now - lastBlockTime) / (unsigned)c_durationLimit; |
|||
clog(NetAllDetail) << "Estimated hashes: " << blockCount; |
|||
return blockCount; |
|||
} |
|||
|
|||
void BlockChainSync::requestBlocks(EthereumPeer* _peer) |
|||
{ |
|||
if (host().bq().knownFull()) |
|||
{ |
|||
clog(NetAllDetail) << "Waiting for block queue before downloading blocks"; |
|||
m_lastActiveState = m_state; |
|||
pauseSync(); |
|||
_peer->setIdle(); |
|||
return; |
|||
} |
|||
_peer->requestBlocks(); |
|||
if (_peer->m_asking != Asking::Blocks) //nothing to download
|
|||
{ |
|||
peerDoneBlocks(_peer); |
|||
if (downloadMan().isComplete()) |
|||
completeSync(); |
|||
return; |
|||
} |
|||
} |
|||
|
|||
void BlockChainSync::onPeerBlocks(EthereumPeer* _peer, RLP const& _r) |
|||
{ |
|||
RecursiveGuard l(x_sync); |
|||
DEV_INVARIANT_CHECK; |
|||
unsigned itemCount = _r.itemCount(); |
|||
clog(NetMessageSummary) << "Blocks (" << dec << itemCount << "entries)" << (itemCount ? "" : ": NoMoreBlocks"); |
|||
|
|||
_peer->setIdle(); |
|||
if (m_state != SyncState::Blocks && m_state != SyncState::NewBlocks) |
|||
clog(NetWarn) << "Unexpected Blocks received!"; |
|||
if (m_state == SyncState::Waiting) |
|||
{ |
|||
clog(NetAllDetail) << "Ignored blocks while waiting"; |
|||
return; |
|||
} |
|||
|
|||
if (itemCount == 0) |
|||
{ |
|||
// Got to this peer's latest block - just give up.
|
|||
peerDoneBlocks(_peer); |
|||
if (downloadMan().isComplete()) |
|||
completeSync(); |
|||
return; |
|||
} |
|||
|
|||
unsigned success = 0; |
|||
unsigned future = 0; |
|||
unsigned unknown = 0; |
|||
unsigned got = 0; |
|||
unsigned repeated = 0; |
|||
u256 maxUnknownNumber = 0; |
|||
h256 maxUnknown; |
|||
|
|||
for (unsigned i = 0; i < itemCount; ++i) |
|||
{ |
|||
auto h = BlockInfo::headerHash(_r[i].data()); |
|||
if (_peer->m_sub.noteBlock(h)) |
|||
{ |
|||
_peer->addRating(10); |
|||
switch (host().bq().import(_r[i].data(), host().chain())) |
|||
{ |
|||
case ImportResult::Success: |
|||
success++; |
|||
break; |
|||
|
|||
case ImportResult::Malformed: |
|||
case ImportResult::BadChain: |
|||
_peer->disable("Malformed block received."); |
|||
return; |
|||
|
|||
case ImportResult::FutureTimeKnown: |
|||
future++; |
|||
break; |
|||
case ImportResult::AlreadyInChain: |
|||
case ImportResult::AlreadyKnown: |
|||
got++; |
|||
break; |
|||
|
|||
case ImportResult::FutureTimeUnknown: |
|||
future++; //Fall through
|
|||
|
|||
case ImportResult::UnknownParent: |
|||
{ |
|||
unknown++; |
|||
if (m_state == SyncState::NewBlocks) |
|||
{ |
|||
BlockInfo bi; |
|||
bi.populateFromHeader(_r[i][0]); |
|||
if (bi.number > maxUnknownNumber) |
|||
{ |
|||
maxUnknownNumber = bi.number; |
|||
maxUnknown = h; |
|||
} |
|||
} |
|||
break; |
|||
} |
|||
|
|||
default:; |
|||
} |
|||
} |
|||
else |
|||
{ |
|||
_peer->addRating(0); // -1?
|
|||
repeated++; |
|||
} |
|||
} |
|||
|
|||
clog(NetMessageSummary) << dec << success << "imported OK," << unknown << "with unknown parents," << future << "with future timestamps," << got << " already known," << repeated << " repeats received."; |
|||
|
|||
if (host().bq().unknownFull()) |
|||
{ |
|||
clog(NetWarn) << "Too many unknown blocks, restarting sync"; |
|||
restartSync(); |
|||
return; |
|||
} |
|||
|
|||
if (m_state == SyncState::NewBlocks && unknown > 0) |
|||
{ |
|||
completeSync(); |
|||
resetSyncFor(_peer, maxUnknown, std::numeric_limits<u256>::max()); //TODO: proper total difficuty
|
|||
} |
|||
if (m_state == SyncState::Blocks || m_state == SyncState::NewBlocks) |
|||
{ |
|||
if (downloadMan().isComplete()) |
|||
completeSync(); |
|||
else if (!got) |
|||
requestBlocks(_peer); |
|||
else |
|||
peerDoneBlocks(_peer); |
|||
} |
|||
DEV_INVARIANT_CHECK; |
|||
} |
|||
|
|||
void BlockChainSync::onPeerNewBlock(EthereumPeer* _peer, RLP const& _r) |
|||
{ |
|||
DEV_INVARIANT_CHECK; |
|||
RecursiveGuard l(x_sync); |
|||
auto h = BlockInfo::headerHash(_r[0].data()); |
|||
clog(NetMessageSummary) << "NewBlock: " << h; |
|||
|
|||
if (_r.itemCount() != 2) |
|||
_peer->disable("NewBlock without 2 data fields."); |
|||
else |
|||
{ |
|||
switch (host().bq().import(_r[0].data(), host().chain())) |
|||
{ |
|||
case ImportResult::Success: |
|||
_peer->addRating(100); |
|||
break; |
|||
case ImportResult::FutureTimeKnown: |
|||
//TODO: Rating dependent on how far in future it is.
|
|||
break; |
|||
|
|||
case ImportResult::Malformed: |
|||
case ImportResult::BadChain: |
|||
_peer->disable("Malformed block received."); |
|||
return; |
|||
|
|||
case ImportResult::AlreadyInChain: |
|||
case ImportResult::AlreadyKnown: |
|||
break; |
|||
|
|||
case ImportResult::FutureTimeUnknown: |
|||
case ImportResult::UnknownParent: |
|||
clog(NetMessageSummary) << "Received block with no known parent. Resyncing..."; |
|||
resetSyncFor(_peer, h, _r[1].toInt<u256>()); |
|||
break; |
|||
default:; |
|||
} |
|||
|
|||
DEV_GUARDED(_peer->x_knownBlocks) |
|||
_peer->m_knownBlocks.insert(h); |
|||
} |
|||
DEV_INVARIANT_CHECK; |
|||
} |
|||
|
|||
PV60Sync::PV60Sync(EthereumHost& _host): |
|||
BlockChainSync(_host) |
|||
{ |
|||
resetSync(); |
|||
} |
|||
|
|||
SyncStatus PV60Sync::status() const |
|||
{ |
|||
RecursiveGuard l(x_sync); |
|||
SyncStatus res; |
|||
res.state = m_state; |
|||
if (m_state == SyncState::Hashes) |
|||
{ |
|||
res.hashesTotal = m_estimatedHashes; |
|||
res.hashesReceived = static_cast<unsigned>(m_syncingNeededBlocks.size()); |
|||
res.hashesEstimated = true; |
|||
} |
|||
else if (m_state == SyncState::Blocks || m_state == SyncState::NewBlocks || m_state == SyncState::Waiting) |
|||
{ |
|||
res.blocksTotal = downloadMan().chainSize(); |
|||
res.blocksReceived = downloadMan().blocksGot().size(); |
|||
} |
|||
return res; |
|||
} |
|||
|
|||
void PV60Sync::setState(EthereumPeer* _peer, SyncState _s, bool _isSyncing, bool _needHelp) |
|||
{ |
|||
bool changedState = (m_state != _s); |
|||
m_state = _s; |
|||
|
|||
if (_isSyncing != (m_syncer == _peer) || (_isSyncing && changedState)) |
|||
changeSyncer(_isSyncing ? _peer : nullptr, _needHelp); |
|||
else if (_s == SyncState::Idle) |
|||
changeSyncer(nullptr, _needHelp); |
|||
|
|||
assert(!!m_syncer || _s == SyncState::Idle); |
|||
} |
|||
|
|||
void PV60Sync::resetSync() |
|||
{ |
|||
m_syncingLatestHash = h256(); |
|||
m_syncingLastReceivedHash = h256(); |
|||
m_syncingTotalDifficulty = 0; |
|||
m_syncingNeededBlocks.clear(); |
|||
} |
|||
|
|||
void PV60Sync::restartSync() |
|||
{ |
|||
resetSync(); |
|||
host().bq().clear(); |
|||
if (isSyncing()) |
|||
transition(m_syncer, SyncState::Idle); |
|||
} |
|||
|
|||
void PV60Sync::completeSync() |
|||
{ |
|||
if (isSyncing()) |
|||
transition(m_syncer, SyncState::Idle); |
|||
} |
|||
|
|||
void PV60Sync::pauseSync() |
|||
{ |
|||
if (isSyncing()) |
|||
setState(m_syncer, SyncState::Waiting, true); |
|||
} |
|||
|
|||
void PV60Sync::continueSync() |
|||
{ |
|||
transition(m_syncer, SyncState::Blocks); |
|||
} |
|||
|
|||
void PV60Sync::onNewPeer(EthereumPeer* _peer) |
|||
{ |
|||
setNeedsSyncing(_peer, _peer->m_latestHash, _peer->m_totalDifficulty); |
|||
} |
|||
|
|||
void PV60Sync::transition(EthereumPeer* _peer, SyncState _s, bool _force, bool _needHelp) |
|||
{ |
|||
clog(NetMessageSummary) << "Transition!" << EthereumHost::stateName(_s) << "from" << EthereumHost::stateName(m_state) << ", " << (isSyncing(_peer) ? "syncing" : "holding") << (needsSyncing(_peer) ? "& needed" : ""); |
|||
|
|||
if (m_state == SyncState::Idle && _s != SyncState::Idle) |
|||
_peer->m_requireTransactions = true; |
|||
|
|||
RLPStream s; |
|||
if (_s == SyncState::Hashes) |
|||
{ |
|||
if (m_state == SyncState::Idle) |
|||
{ |
|||
if (isSyncing(_peer)) |
|||
clog(NetWarn) << "Bad state: not asking for Hashes, yet syncing!"; |
|||
|
|||
m_syncingLatestHash = _peer->m_latestHash; |
|||
m_syncingTotalDifficulty = _peer->m_totalDifficulty; |
|||
setState(_peer, _s, true); |
|||
_peer->requestHashes(m_syncingLastReceivedHash ? m_syncingLastReceivedHash : m_syncingLatestHash); |
|||
DEV_INVARIANT_CHECK; |
|||
return; |
|||
} |
|||
else if (m_state == SyncState::Hashes) |
|||
{ |
|||
if (!isSyncing(_peer)) |
|||
clog(NetWarn) << "Bad state: asking for Hashes yet not syncing!"; |
|||
|
|||
setState(_peer, _s, true); |
|||
_peer->requestHashes(m_syncingLastReceivedHash); |
|||
DEV_INVARIANT_CHECK; |
|||
return; |
|||
} |
|||
} |
|||
else if (_s == SyncState::Blocks) |
|||
{ |
|||
if (m_state == SyncState::Hashes) |
|||
{ |
|||
if (!isSyncing(_peer)) |
|||
{ |
|||
clog(NetWarn) << "Bad state: asking for Hashes yet not syncing!"; |
|||
return; |
|||
} |
|||
if (shouldGrabBlocks(_peer)) |
|||
{ |
|||
clog(NetNote) << "Difficulty of hashchain HIGHER. Grabbing" << m_syncingNeededBlocks.size() << "blocks [latest now" << m_syncingLatestHash << ", was" << host().latestBlockSent() << "]"; |
|||
downloadMan().resetToChain(m_syncingNeededBlocks); |
|||
resetSync(); |
|||
} |
|||
else |
|||
{ |
|||
clog(NetNote) << "Difficulty of hashchain not HIGHER. Ignoring."; |
|||
resetSync(); |
|||
setState(_peer, SyncState::Idle, false); |
|||
return; |
|||
} |
|||
assert (isSyncing(_peer)); |
|||
} |
|||
// run through into...
|
|||
if (m_state == SyncState::Idle || m_state == SyncState::Hashes || m_state == SyncState::Blocks || m_state == SyncState::Waiting) |
|||
{ |
|||
// Looks like it's the best yet for total difficulty. Set to download.
|
|||
setState(_peer, SyncState::Blocks, isSyncing(_peer), _needHelp); // will kick off other peers to help if available.
|
|||
requestBlocks(_peer); |
|||
DEV_INVARIANT_CHECK; |
|||
return; |
|||
} |
|||
} |
|||
else if (_s == SyncState::NewBlocks) |
|||
{ |
|||
if (m_state != SyncState::Idle && m_state != SyncState::NewBlocks && m_state != SyncState::Waiting) |
|||
clog(NetWarn) << "Bad state: Asking new blocks while syncing!"; |
|||
else |
|||
{ |
|||
setState(_peer, SyncState::NewBlocks, true, _needHelp); |
|||
requestBlocks(_peer); |
|||
DEV_INVARIANT_CHECK; |
|||
return; |
|||
} |
|||
} |
|||
else if (_s == SyncState::Waiting) |
|||
{ |
|||
if (m_state != SyncState::Blocks && m_state != SyncState::NewBlocks && m_state != SyncState::Hashes && m_state != SyncState::Waiting) |
|||
clog(NetWarn) << "Bad state: Entering waiting state while not downloading blocks!"; |
|||
else |
|||
{ |
|||
setState(_peer, SyncState::Waiting, isSyncing(_peer), _needHelp); |
|||
return; |
|||
} |
|||
} |
|||
else if (_s == SyncState::Idle) |
|||
{ |
|||
host().foreachPeer([this](EthereumPeer* _p) { _p->setIdle(); return true; }); |
|||
if (m_state == SyncState::Blocks || m_state == SyncState::NewBlocks) |
|||
{ |
|||
clog(NetNote) << "Finishing blocks fetch..."; |
|||
|
|||
// a bit overkill given that the other nodes may yet have the needed blocks, but better to be safe than sorry.
|
|||
if (isSyncing(_peer)) |
|||
noteDoneBlocks(_peer, _force); |
|||
|
|||
// NOTE: need to notify of giving up on chain-hashes, too, altering state as necessary.
|
|||
_peer->m_sub.doneFetch(); |
|||
_peer->setIdle(); |
|||
setState(_peer, SyncState::Idle, false); |
|||
} |
|||
else if (m_state == SyncState::Hashes) |
|||
{ |
|||
clog(NetNote) << "Finishing hashes fetch..."; |
|||
setState(_peer, SyncState::Idle, false); |
|||
} |
|||
// Otherwise it's fine. We don't care if it's Nothing->Nothing.
|
|||
DEV_INVARIANT_CHECK; |
|||
return; |
|||
} |
|||
|
|||
clog(NetWarn) << "Invalid state transition:" << EthereumHost::stateName(_s) << "from" << EthereumHost::stateName(m_state) << ", " << (isSyncing(_peer) ? "syncing" : "holding") << (needsSyncing(_peer) ? "& needed" : ""); |
|||
} |
|||
|
|||
void PV60Sync::resetSyncFor(EthereumPeer* _peer, h256 _latestHash, u256 _td) |
|||
{ |
|||
setNeedsSyncing(_peer, _latestHash, _td); |
|||
} |
|||
|
|||
void PV60Sync::setNeedsSyncing(EthereumPeer* _peer, h256 _latestHash, u256 _td) |
|||
{ |
|||
_peer->m_latestHash = _latestHash; |
|||
_peer->m_totalDifficulty = _td; |
|||
|
|||
if (_peer->m_latestHash) |
|||
noteNeedsSyncing(_peer); |
|||
|
|||
_peer->session()->addNote("sync", string(isSyncing(_peer) ? "ongoing" : "holding") + (needsSyncing(_peer) ? " & needed" : "")); |
|||
} |
|||
|
|||
bool PV60Sync::needsSyncing(EthereumPeer* _peer) const |
|||
{ |
|||
return !!_peer->m_latestHash; |
|||
} |
|||
|
|||
bool PV60Sync::isSyncing(EthereumPeer* _peer) const |
|||
{ |
|||
return m_syncer == _peer; |
|||
} |
|||
|
|||
bool PV60Sync::shouldGrabBlocks(EthereumPeer* _peer) const |
|||
{ |
|||
auto td = _peer->m_totalDifficulty; |
|||
auto lh = _peer->m_latestHash; |
|||
auto ctd = host().chain().details().totalDifficulty; |
|||
|
|||
if (m_syncingNeededBlocks.empty()) |
|||
return false; |
|||
|
|||
clog(NetNote) << "Should grab blocks? " << td << "vs" << ctd << ";" << m_syncingNeededBlocks.size() << " blocks, ends" << m_syncingNeededBlocks.back(); |
|||
|
|||
if (td < ctd || (td == ctd && host().chain().currentHash() == lh)) |
|||
return false; |
|||
|
|||
return true; |
|||
} |
|||
|
|||
void PV60Sync::attemptSync(EthereumPeer* _peer) |
|||
{ |
|||
if (m_state != SyncState::Idle) |
|||
{ |
|||
clog(NetAllDetail) << "Can't sync with this peer - outstanding asks."; |
|||
return; |
|||
} |
|||
|
|||
// if already done this, then ignore.
|
|||
if (!needsSyncing(_peer)) |
|||
{ |
|||
clog(NetAllDetail) << "Already synced with this peer."; |
|||
return; |
|||
} |
|||
|
|||
unsigned n = host().chain().number(); |
|||
u256 td = host().chain().details().totalDifficulty; |
|||
if (host().bq().isActive()) |
|||
td += host().bq().difficulty(); |
|||
|
|||
clog(NetAllDetail) << "Attempt chain-grab? Latest:" << (m_syncingLastReceivedHash ? m_syncingLastReceivedHash : m_syncingLatestHash) << ", number:" << n << ", TD:" << td << " versus " << _peer->m_totalDifficulty; |
|||
if (td >= _peer->m_totalDifficulty) |
|||
{ |
|||
clog(NetAllDetail) << "No. Our chain is better."; |
|||
resetNeedsSyncing(_peer); |
|||
transition(_peer, SyncState::Idle); |
|||
} |
|||
else |
|||
{ |
|||
clog(NetAllDetail) << "Yes. Their chain is better."; |
|||
m_estimatedHashes = _peer->m_expectedHashes - c_chainReorgSize; |
|||
transition(_peer, SyncState::Hashes); |
|||
} |
|||
} |
|||
|
|||
void PV60Sync::noteNeedsSyncing(EthereumPeer* _peer) |
|||
{ |
|||
// if already downloading hash-chain, ignore.
|
|||
if (isSyncing()) |
|||
{ |
|||
clog(NetAllDetail) << "Sync in progress: Just set to help out."; |
|||
if (m_state == SyncState::Blocks) |
|||
requestBlocks(_peer); |
|||
} |
|||
else |
|||
// otherwise check to see if we should be downloading...
|
|||
attemptSync(_peer); |
|||
} |
|||
|
|||
void PV60Sync::changeSyncer(EthereumPeer* _syncer, bool _needHelp) |
|||
{ |
|||
if (_syncer) |
|||
clog(NetAllDetail) << "Changing syncer to" << _syncer->session()->socketId(); |
|||
else |
|||
clog(NetAllDetail) << "Clearing syncer."; |
|||
|
|||
m_syncer = _syncer; |
|||
if (isSyncing()) |
|||
{ |
|||
if (_needHelp && (m_state == SyncState::Blocks || m_state == SyncState::NewBlocks)) |
|||
host().foreachPeer([&](EthereumPeer* _p) |
|||
{ |
|||
clog(NetNote) << "Getting help with downloading blocks"; |
|||
if (_p != _syncer && _p->m_asking == Asking::Nothing) |
|||
transition(_p, m_state); |
|||
return true; |
|||
}); |
|||
} |
|||
else |
|||
{ |
|||
// start grabbing next hash chain if there is one.
|
|||
host().foreachPeer([this](EthereumPeer* _p) |
|||
{ |
|||
attemptSync(_p); |
|||
return !isSyncing(); |
|||
}); |
|||
if (!isSyncing()) |
|||
{ |
|||
if (m_state != SyncState::Idle) |
|||
setState(_syncer, SyncState::Idle); |
|||
clog(NetNote) << "No more peers to sync with."; |
|||
} |
|||
} |
|||
assert(!!m_syncer || m_state == SyncState::Idle); |
|||
} |
|||
|
|||
void PV60Sync::peerDoneBlocks(EthereumPeer* _peer) |
|||
{ |
|||
noteDoneBlocks(_peer, false); |
|||
} |
|||
|
|||
void PV60Sync::noteDoneBlocks(EthereumPeer* _peer, bool _clemency) |
|||
{ |
|||
resetNeedsSyncing(_peer); |
|||
if (downloadMan().isComplete()) |
|||
{ |
|||
// Done our chain-get.
|
|||
clog(NetNote) << "Chain download complete."; |
|||
// 1/100th for each useful block hash.
|
|||
_peer->addRating(downloadMan().chainSize() / 100); |
|||
downloadMan().reset(); |
|||
} |
|||
else if (isSyncing(_peer)) |
|||
{ |
|||
if (_clemency) |
|||
clog(NetNote) << "Chain download failed. Aborted while incomplete."; |
|||
else |
|||
{ |
|||
// Done our chain-get.
|
|||
clog(NetWarn) << "Chain download failed. Peer with blocks didn't have them all. This peer is bad and should be punished."; |
|||
clog(NetWarn) << downloadMan().remaining(); |
|||
clog(NetWarn) << "WOULD BAN."; |
|||
// m_banned.insert(_peer->session()->id()); // We know who you are!
|
|||
// _peer->disable("Peer sent hashes but was unable to provide the blocks.");
|
|||
} |
|||
downloadMan().reset(); |
|||
} |
|||
_peer->m_sub.doneFetch(); |
|||
} |
|||
|
|||
void PV60Sync::onPeerHashes(EthereumPeer* _peer, h256s const& _hashes) |
|||
{ |
|||
RecursiveGuard l(x_sync); |
|||
DEV_INVARIANT_CHECK; |
|||
_peer->setIdle(); |
|||
if (!isSyncing(_peer)) |
|||
{ |
|||
clog(NetMessageSummary) << "Ignoring hashes synce not syncing"; |
|||
return; |
|||
} |
|||
if (_hashes.size() == 0) |
|||
{ |
|||
transition(_peer, SyncState::Blocks); |
|||
return; |
|||
} |
|||
unsigned knowns = 0; |
|||
unsigned unknowns = 0; |
|||
for (unsigned i = 0; i < _hashes.size(); ++i) |
|||
{ |
|||
auto h = _hashes[i]; |
|||
auto status = host().bq().blockStatus(h); |
|||
if (status == QueueStatus::Importing || status == QueueStatus::Ready || host().chain().isKnown(h)) |
|||
{ |
|||
clog(NetMessageSummary) << "block hash ready:" << h << ". Start blocks download..."; |
|||
assert (isSyncing(_peer)); |
|||
transition(_peer, SyncState::Blocks); |
|||
return; |
|||
} |
|||
else if (status == QueueStatus::Bad) |
|||
{ |
|||
cwarn << "block hash bad!" << h << ". Bailing..."; |
|||
transition(_peer, SyncState::Idle); |
|||
return; |
|||
} |
|||
else if (status == QueueStatus::Unknown) |
|||
{ |
|||
unknowns++; |
|||
m_syncingNeededBlocks.push_back(h); |
|||
} |
|||
else |
|||
knowns++; |
|||
m_syncingLastReceivedHash = h; |
|||
} |
|||
clog(NetMessageSummary) << knowns << "knowns," << unknowns << "unknowns; now at" << m_syncingLastReceivedHash; |
|||
if (m_syncingNeededBlocks.size() > _peer->m_expectedHashes) |
|||
{ |
|||
_peer->disable("Too many hashes"); |
|||
restartSync(); |
|||
return; |
|||
} |
|||
// run through - ask for more.
|
|||
transition(_peer, SyncState::Hashes); |
|||
DEV_INVARIANT_CHECK; |
|||
} |
|||
|
|||
void PV60Sync::onPeerNewHashes(EthereumPeer* _peer, h256s const& _hashes) |
|||
{ |
|||
RecursiveGuard l(x_sync); |
|||
DEV_INVARIANT_CHECK; |
|||
if (isSyncing()) |
|||
{ |
|||
clog(NetMessageSummary) << "Ignoring since we're already downloading."; |
|||
return; |
|||
} |
|||
unsigned knowns = 0; |
|||
unsigned unknowns = 0; |
|||
for (auto h: _hashes) |
|||
{ |
|||
_peer->addRating(1); |
|||
DEV_GUARDED(_peer->x_knownBlocks) |
|||
_peer->m_knownBlocks.insert(h); |
|||
auto status = host().bq().blockStatus(h); |
|||
if (status == QueueStatus::Importing || status == QueueStatus::Ready || host().chain().isKnown(h)) |
|||
knowns++; |
|||
else if (status == QueueStatus::Bad) |
|||
{ |
|||
cwarn << "block hash bad!" << h << ". Bailing..."; |
|||
return; |
|||
} |
|||
else if (status == QueueStatus::Unknown) |
|||
{ |
|||
unknowns++; |
|||
m_syncingNeededBlocks.push_back(h); |
|||
} |
|||
else |
|||
knowns++; |
|||
} |
|||
clog(NetMessageSummary) << knowns << "knowns," << unknowns << "unknowns"; |
|||
if (unknowns > 0) |
|||
{ |
|||
clog(NetNote) << "Not syncing and new block hash discovered: syncing without help."; |
|||
downloadMan().resetToChain(m_syncingNeededBlocks); |
|||
resetSync(); |
|||
transition(_peer, SyncState::NewBlocks, false, false); |
|||
} |
|||
DEV_INVARIANT_CHECK; |
|||
} |
|||
|
|||
void PV60Sync::abortSync(EthereumPeer* _peer) |
|||
{ |
|||
if (isSyncing(_peer)) |
|||
{ |
|||
host().foreachPeer([this](EthereumPeer* _p) { _p->setIdle(); return true; }); |
|||
transition(_peer, SyncState::Idle, true); |
|||
} |
|||
DEV_INVARIANT_CHECK; |
|||
} |
|||
|
|||
void PV60Sync::onPeerAborting(EthereumPeer* _peer) |
|||
{ |
|||
abortSync(_peer); |
|||
DEV_INVARIANT_CHECK; |
|||
} |
|||
|
|||
bool PV60Sync::invariants() const |
|||
{ |
|||
if (m_state == SyncState::Idle && !!m_syncer) |
|||
return false; |
|||
if (m_state != SyncState::Idle && !m_syncer) |
|||
return false; |
|||
if (m_state == SyncState::Hashes) |
|||
{ |
|||
bool hashes = false; |
|||
host().foreachPeer([&](EthereumPeer* _p) { if (_p->m_asking == Asking::Hashes) hashes = true; return !hashes; }); |
|||
if (!hashes) |
|||
return false; |
|||
} |
|||
if (m_state == SyncState::Blocks || m_state == SyncState::NewBlocks) |
|||
{ |
|||
bool blocks = false; |
|||
host().foreachPeer([&](EthereumPeer* _p) { if (_p->m_asking == Asking::Blocks) blocks = true; return !blocks; }); |
|||
if (!blocks) |
|||
return false; |
|||
if (downloadMan().isComplete()) |
|||
return false; |
|||
} |
|||
return true; |
|||
} |
|||
@ -0,0 +1,213 @@ |
|||
/*
|
|||
This file is part of cpp-ethereum. |
|||
|
|||
cpp-ethereum is free software: you can redistribute it and/or modify |
|||
it under the terms of the GNU General Public License as published by |
|||
the Free Software Foundation, either version 3 of the License, or |
|||
(at your option) any later version. |
|||
|
|||
cpp-ethereum is distributed in the hope that it will be useful, |
|||
but WITHOUT ANY WARRANTY; without even the implied warranty of |
|||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|||
GNU General Public License for more details. |
|||
|
|||
You should have received a copy of the GNU General Public License |
|||
along with cpp-ethereum. If not, see <http://www.gnu.org/licenses/>.
|
|||
*/ |
|||
/** @file BlockChainSync.h
|
|||
* @author Gav Wood <i@gavwood.com> |
|||
* @date 2014 |
|||
*/ |
|||
|
|||
#pragma once |
|||
|
|||
#include <mutex> |
|||
|
|||
#include <libdevcore/Guards.h> |
|||
#include <libdevcore/RangeMask.h> |
|||
#include <libethcore/Common.h> |
|||
#include <libp2p/Common.h> |
|||
#include "CommonNet.h" |
|||
#include "DownloadMan.h" |
|||
|
|||
namespace dev |
|||
{ |
|||
|
|||
class RLPStream; |
|||
|
|||
namespace eth |
|||
{ |
|||
|
|||
class EthereumHost; |
|||
class BlockQueue; |
|||
class EthereumPeer; |
|||
|
|||
/**
|
|||
* @brief Base BlockChain synchronization strategy class. |
|||
* Syncs to peers and keeps up to date. Base class handles blocks downloading but does not contain any details on state transfer logic. |
|||
*/ |
|||
class BlockChainSync: public HasInvariants |
|||
{ |
|||
public: |
|||
BlockChainSync(EthereumHost& _host); |
|||
virtual ~BlockChainSync(); |
|||
void abortSync(); ///< Abort all sync activity
|
|||
|
|||
DownloadMan const& downloadMan() const; |
|||
DownloadMan& downloadMan(); |
|||
|
|||
/// @returns true is Sync is in progress
|
|||
virtual bool isSyncing() const = 0; |
|||
|
|||
/// Called by peer to report status
|
|||
virtual void onPeerStatus(EthereumPeer* _peer); |
|||
|
|||
/// Called by peer once it has new blocks during syn
|
|||
virtual void onPeerBlocks(EthereumPeer* _peer, RLP const& _r); |
|||
|
|||
/// Called by peer once it has new blocks
|
|||
virtual void onPeerNewBlock(EthereumPeer* _peer, RLP const& _r); |
|||
|
|||
/// Called by peer once it has new hashes
|
|||
virtual void onPeerNewHashes(EthereumPeer* _peer, h256s const& _hashes) = 0; |
|||
|
|||
/// Called by peer once it has another sequential block of hashes during sync
|
|||
virtual void onPeerHashes(EthereumPeer* _peer, h256s const& _hashes) = 0; |
|||
|
|||
/// Called by peer when it is disconnecting
|
|||
virtual void onPeerAborting(EthereumPeer* _peer) = 0; |
|||
|
|||
/// @returns Synchonization status
|
|||
virtual SyncStatus status() const = 0; |
|||
|
|||
static char const* stateName(SyncState _s) { return s_stateNames[static_cast<int>(_s)]; } |
|||
|
|||
protected: |
|||
//To be implemented in derived classes:
|
|||
/// New valid peer appears
|
|||
virtual void onNewPeer(EthereumPeer* _peer) = 0; |
|||
|
|||
/// Peer done downloading blocks
|
|||
virtual void peerDoneBlocks(EthereumPeer* _peer) = 0; |
|||
|
|||
/// Resume downloading after witing state
|
|||
virtual void continueSync() = 0; |
|||
|
|||
/// Restart sync
|
|||
virtual void restartSync() = 0; |
|||
|
|||
/// Called after all blocks have been donloaded
|
|||
virtual void completeSync() = 0; |
|||
|
|||
/// Enter waiting state
|
|||
virtual void pauseSync() = 0; |
|||
|
|||
/// Restart sync for given peer
|
|||
virtual void resetSyncFor(EthereumPeer* _peer, h256 _latestHash, u256 _td) = 0; |
|||
|
|||
EthereumHost& host() { return m_host; } |
|||
EthereumHost const& host() const { return m_host; } |
|||
|
|||
/// Estimates max number of hashes peers can give us.
|
|||
unsigned estimateHashes(); |
|||
|
|||
/// Request blocks from peer if needed
|
|||
void requestBlocks(EthereumPeer* _peer); |
|||
|
|||
private: |
|||
static char const* const s_stateNames[static_cast<int>(SyncState::Size)]; |
|||
bool invariants() const override = 0; |
|||
EthereumHost& m_host; |
|||
HashDownloadMan m_hashMan; |
|||
|
|||
protected: |
|||
Handler m_bqRoomAvailable; |
|||
mutable RecursiveMutex x_sync; |
|||
SyncState m_state = SyncState::Idle; ///< Current sync state
|
|||
SyncState m_lastActiveState = SyncState::Idle; ///< Saved state before entering waiting queue mode
|
|||
unsigned m_estimatedHashes = 0; ///< Number of estimated hashes for the last peer over PV60. Used for status reporting only.
|
|||
}; |
|||
|
|||
|
|||
/**
|
|||
* @brief Syncrhonization over PV60. Selects a single peer and tries to downloading hashes from it. After hash downaload is complete |
|||
* Syncs to peers and keeps up to date |
|||
*/ |
|||
class PV60Sync: public BlockChainSync |
|||
{ |
|||
public: |
|||
PV60Sync(EthereumHost& _host); |
|||
|
|||
/// @returns true is Sync is in progress
|
|||
bool isSyncing() const override { return !!m_syncer; } |
|||
|
|||
/// Called by peer once it has new hashes
|
|||
void onPeerNewHashes(EthereumPeer* _peer, h256s const& _hashes) override; |
|||
|
|||
/// Called by peer once it has another sequential block of hashes during sync
|
|||
void onPeerHashes(EthereumPeer* _peer, h256s const& _hashes) override; |
|||
|
|||
/// Called by peer when it is disconnecting
|
|||
void onPeerAborting(EthereumPeer* _peer) override; |
|||
|
|||
/// @returns Sync status
|
|||
SyncStatus status() const override; |
|||
|
|||
void onNewPeer(EthereumPeer* _peer) override; |
|||
void continueSync() override; |
|||
void peerDoneBlocks(EthereumPeer* _peer) override; |
|||
void restartSync() override; |
|||
void completeSync() override; |
|||
void pauseSync() override; |
|||
void resetSyncFor(EthereumPeer* _peer, h256 _latestHash, u256 _td) override; |
|||
|
|||
private: |
|||
/// Transition sync state in a particular direction. @param _peer Peer that is responsible for state tranfer
|
|||
void transition(EthereumPeer* _peer, SyncState _s, bool _force = false, bool _needHelp = true); |
|||
|
|||
/// Reset peer syncing requirements state.
|
|||
void resetNeedsSyncing(EthereumPeer* _peer) { setNeedsSyncing(_peer, h256(), 0); } |
|||
|
|||
/// Update peer syncing requirements state.
|
|||
void setNeedsSyncing(EthereumPeer* _peer, h256 _latestHash, u256 _td); |
|||
|
|||
/// Do we presently need syncing with this peer?
|
|||
bool needsSyncing(EthereumPeer* _peer) const; |
|||
|
|||
/// Check whether the session should bother grabbing blocks from a peer.
|
|||
bool shouldGrabBlocks(EthereumPeer* _peer) const; |
|||
|
|||
/// Attempt to begin syncing with the peer; first check the peer has a more difficlult chain to download, then start asking for hashes, then move to blocks
|
|||
void attemptSync(EthereumPeer* _peer); |
|||
|
|||
/// Update our syncing state
|
|||
void setState(EthereumPeer* _peer, SyncState _s, bool _isSyncing = false, bool _needHelp = false); |
|||
|
|||
/// Check if peer is main syncer
|
|||
bool isSyncing(EthereumPeer* _peer) const; |
|||
|
|||
/// Check if we need (re-)syncing with the peer.
|
|||
void noteNeedsSyncing(EthereumPeer* _who); |
|||
|
|||
/// Set main syncing peer
|
|||
void changeSyncer(EthereumPeer* _syncer, bool _needHelp); |
|||
|
|||
/// Called when peer done downloading blocks
|
|||
void noteDoneBlocks(EthereumPeer* _who, bool _clemency); |
|||
|
|||
/// Abort syncing for peer
|
|||
void abortSync(EthereumPeer* _peer); |
|||
|
|||
/// Reset hash chain syncing
|
|||
void resetSync(); |
|||
|
|||
bool invariants() const override; |
|||
|
|||
h256s m_syncingNeededBlocks; ///< The blocks that we should download from this peer.
|
|||
h256 m_syncingLastReceivedHash; ///< Hash most recently received from peer.
|
|||
h256 m_syncingLatestHash; ///< Peer's latest block's hash, as of the current sync.
|
|||
u256 m_syncingTotalDifficulty; ///< Peer's latest block's total difficulty, as of the current sync.
|
|||
EthereumPeer* m_syncer = nullptr; // TODO: switch to weak_ptr
|
|||
}; |
|||
} |
|||
} |
|||
Loading…
Reference in new issue