You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 

336 lines
9.1 KiB

/*
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 EthereumPeer.cpp
* @author Gav Wood <i@gavwood.com>
* @date 2014
*/
#include "EthereumPeer.h"
#include <chrono>
#include <libdevcore/Common.h>
#include <libethcore/Exceptions.h>
#include <libp2p/Session.h>
#include "BlockChain.h"
#include "EthereumHost.h"
using namespace std;
using namespace dev;
using namespace dev::eth;
using namespace p2p;
#define clogS(X) dev::LogOutputStream<X, true>(false) << "| " << std::setw(2) << session()->socketId() << "] "
EthereumPeer::EthereumPeer(Session* _s, HostCapabilityFace* _h):
Capability(_s, _h),
m_sub(host()->m_man)
{
setGrabbing(Grabbing::State);
sendStatus();
}
EthereumPeer::~EthereumPeer()
{
giveUpOnFetch();
}
EthereumHost* EthereumPeer::host() const
{
return static_cast<EthereumHost*>(Capability::hostCapability());
}
void EthereumPeer::sendStatus()
{
RLPStream s;
prep(s);
s.appendList(6) << StatusPacket
<< host()->protocolVersion()
<< host()->networkId()
<< host()->m_chain.details().totalDifficulty
<< host()->m_chain.currentHash()
<< host()->m_chain.genesisHash();
sealAndSend(s);
}
void EthereumPeer::startInitialSync()
{
// Grab transactions off them.
{
RLPStream s;
prep(s).appendList(1);
s << GetTransactionsPacket;
sealAndSend(s);
}
host()->noteHavePeerState(this);
}
void EthereumPeer::tryGrabbingHashChain()
{
// if already done this, then ignore.
if (m_grabbing != Grabbing::State)
{
clogS(NetAllDetail) << "Already synced with this peer.";
return;
}
h256 c = host()->m_chain.currentHash();
unsigned n = host()->m_chain.number();
u256 td = host()->m_chain.details().totalDifficulty;
clogS(NetAllDetail) << "Attempt chain-grab? Latest:" << c.abridged() << ", number:" << n << ", TD:" << td << " versus " << m_totalDifficulty;
if (td >= m_totalDifficulty)
{
clogS(NetAllDetail) << "No. Our chain is better.";
setGrabbing(Grabbing::Nothing);
return; // All good - we have the better chain.
}
// Our chain isn't better - grab theirs.
{
clogS(NetAllDetail) << "Yes. Their chain is better.";
host()->updateGrabbing(Grabbing::Hashes);
setGrabbing(Grabbing::Hashes);
RLPStream s;
prep(s).appendList(3);
s << GetBlockHashesPacket << m_latestHash << c_maxHashesAsk;
m_neededBlocks = h256s(1, m_latestHash);
sealAndSend(s);
}
}
void EthereumPeer::giveUpOnFetch()
{
clogS(NetNote) << "GIVE UP FETCH";
// a bit overkill given that the other nodes may yet have the needed blocks, but better to be safe than sorry.
if (m_grabbing == Grabbing::Chain || m_grabbing == Grabbing::ChainHelper)
{
host()->noteDoneBlocks(this);
setGrabbing(Grabbing::Nothing);
}
// NOTE: need to notify of giving up on chain-hashes, too, altering state as necessary.
m_sub.doneFetch();
}
bool EthereumPeer::interpret(RLP const& _r)
{
switch (_r[0].toInt<unsigned>())
{
case StatusPacket:
{
m_protocolVersion = _r[1].toInt<unsigned>();
m_networkId = _r[2].toInt<u256>();
m_totalDifficulty = _r[3].toInt<u256>();
m_latestHash = _r[4].toHash<h256>();
auto genesisHash = _r[5].toHash<h256>();
clogS(NetMessageSummary) << "Status:" << m_protocolVersion << "/" << m_networkId << "/" << genesisHash.abridged() << ", TD:" << m_totalDifficulty << "=" << m_latestHash.abridged();
if (genesisHash != host()->m_chain.genesisHash())
disable("Invalid genesis hash");
else if (m_protocolVersion != host()->protocolVersion())
disable("Invalid protocol version.");
else if (m_networkId != host()->networkId())
disable("Invalid network identifier.");
else if (session()->info().clientVersion.find("/v0.6.9/") != string::npos)
disable("Blacklisted client version.");
else
startInitialSync();
break;
}
case GetTransactionsPacket:
{
m_requireTransactions = true;
break;
}
case TransactionsPacket:
{
clogS(NetMessageSummary) << "Transactions (" << dec << (_r.itemCount() - 1) << "entries)";
addRating(_r.itemCount() - 1);
lock_guard<recursive_mutex> l(host()->m_incomingLock);
for (unsigned i = 1; i < _r.itemCount(); ++i)
{
host()->addIncomingTransaction(_r[i].data().toBytes());
lock_guard<mutex> l(x_knownTransactions);
m_knownTransactions.insert(sha3(_r[i].data()));
}
break;
}
case GetBlockHashesPacket:
{
h256 later = _r[1].toHash<h256>();
unsigned limit = _r[2].toInt<unsigned>();
clogS(NetMessageSummary) << "GetBlockHashes (" << limit << "entries," << later.abridged() << ")";
unsigned c = min<unsigned>(host()->m_chain.number(later), limit);
RLPStream s;
prep(s).appendList(1 + c).append(BlockHashesPacket);
h256 p = host()->m_chain.details(later).parent;
for (unsigned i = 0; i < c && p; ++i, p = host()->m_chain.details(p).parent)
s << p;
sealAndSend(s);
break;
}
case BlockHashesPacket:
{
clogS(NetMessageSummary) << "BlockHashes (" << dec << (_r.itemCount() - 1) << "entries)" << (_r.itemCount() - 1 ? "" : ": NoMoreHashes");
if (m_grabbing != Grabbing::Hashes)
{
cwarn << "Peer giving us hashes when we didn't ask for them.";
break;
}
if (_r.itemCount() == 1)
{
host()->noteHaveChain(this);
return true;
}
for (unsigned i = 1; i < _r.itemCount(); ++i)
{
auto h = _r[i].toHash<h256>();
if (host()->m_chain.isKnown(h))
{
host()->noteHaveChain(this);
return true;
}
else
m_neededBlocks.push_back(h);
}
// run through - ask for more.
RLPStream s;
prep(s).appendList(3);
s << GetBlockHashesPacket << m_neededBlocks.back() << c_maxHashesAsk;
sealAndSend(s);
break;
}
case GetBlocksPacket:
{
clogS(NetMessageSummary) << "GetBlocks (" << dec << (_r.itemCount() - 1) << "entries)";
// return the requested blocks.
bytes rlp;
unsigned n = 0;
for (unsigned i = 1; i < _r.itemCount() && i <= c_maxBlocks; ++i)
{
auto b = host()->m_chain.block(_r[i].toHash<h256>());
if (b.size())
{
rlp += b;
++n;
}
}
RLPStream s;
sealAndSend(prep(s).appendList(n + 1).append(BlocksPacket).appendRaw(rlp, n));
break;
}
case BlocksPacket:
{
clogS(NetMessageSummary) << "Blocks (" << dec << (_r.itemCount() - 1) << "entries)" << (_r.itemCount() - 1 ? "" : ": NoMoreBlocks");
if (_r.itemCount() == 1)
{
// Couldn't get any from last batch - probably got to this peer's latest block - just give up.
if (m_grabbing == Grabbing::Chain || m_grabbing == Grabbing::ChainHelper)
giveUpOnFetch();
break;
}
unsigned used = 0;
for (unsigned i = 1; i < _r.itemCount(); ++i)
{
auto h = BlockInfo::headerHash(_r[i].data());
m_sub.noteBlock(h);
if (host()->noteBlock(h, _r[i].data()))
used++;
Guard l(x_knownBlocks);
m_knownBlocks.insert(h);
}
addRating(used);
unsigned knownParents = 0;
unsigned unknownParents = 0;
if (g_logVerbosity >= NetMessageSummary::verbosity)
{
unsigned ic = _r.itemCount();
for (unsigned i = 1; i < ic; ++i)
{
auto h = BlockInfo::headerHash(_r[i].data());
BlockInfo bi(_r[i].data());
Guard l(x_knownBlocks);
if (!host()->m_chain.details(bi.parentHash) && !m_knownBlocks.count(bi.parentHash))
{
unknownParents++;
clogS(NetAllDetail) << "Unknown parent" << bi.parentHash.abridged() << "of block" << h.abridged();
}
else
{
knownParents++;
clogS(NetAllDetail) << "Known parent" << bi.parentHash.abridged() << "of block" << h.abridged();
}
}
}
clogS(NetMessageSummary) << dec << knownParents << "known parents," << unknownParents << "unknown," << used << "used.";
if (m_grabbing == Grabbing::Chain || m_grabbing == Grabbing::ChainHelper)
continueGettingChain();
break;
}
default:
return false;
}
return true;
}
void EthereumPeer::ensureGettingChain()
{
if (m_grabbing == Grabbing::ChainHelper)
return; // Already asked & waiting for some.
// Switch to ChainHelper otherwise, unless we're already the Chain grabber.
if (m_grabbing != Grabbing::Chain)
setGrabbing(Grabbing::ChainHelper);
continueGettingChain();
}
void EthereumPeer::continueGettingChain()
{
// If we're getting the hashes already, then we shouldn't be asking for the chain.
if (m_grabbing == Grabbing::Hashes)
return;
auto blocks = m_sub.nextFetch(c_maxBlocksAsk);
if (blocks.size())
{
RLPStream s;
prep(s);
s.appendList(blocks.size() + 1) << GetBlocksPacket;
for (auto const& i: blocks)
s << i;
sealAndSend(s);
}
else
giveUpOnFetch();
}
void EthereumPeer::setGrabbing(Grabbing _g)
{
m_grabbing = _g;
session()->addNote("grab", _g == Grabbing::Nothing ? "nothing" : _g == Grabbing::State ? "state" : _g == Grabbing::Hashes ? "hashes" : _g == Grabbing::Chain ? "chain" : _g == Grabbing::ChainHelper ? "chainhelper" : "?");
}