Browse Source

Adding StructuredLogger class and using it in various places

- Structured logger is injected as a dependency where needed.
- Still a work in progress
cl-refactor
Lefteris Karapetsas 10 years ago
parent
commit
d0c1c510bd
  1. 26
      eth/main.cpp
  2. 1
      libdevcore/CMakeLists.txt
  3. 166
      libdevcore/StructuredLogger.cpp
  4. 59
      libdevcore/StructuredLogger.h
  5. 25
      libethereum/BlockChain.cpp
  6. 7
      libethereum/BlockChain.h
  7. 14
      libethereum/Client.cpp
  8. 6
      libethereum/Client.h
  9. 5
      libethereum/Miner.cpp
  10. 5
      libethereum/Miner.h
  11. 12
      libethereum/State.cpp
  12. 4
      libethereum/State.h
  13. 9
      libp2p/Host.cpp
  14. 66
      libp2p/Host.h
  15. 8
      libwebthree/WebThree.cpp
  16. 2
      libwebthree/WebThree.h

26
eth/main.cpp

@ -29,6 +29,7 @@
#include <boost/algorithm/string/trim_all.hpp>
#include <libdevcrypto/FileSystem.h>
#include <libevmcore/Instruction.h>
#include <libdevcore/StructuredLogger.h>
#include <libevm/VM.h>
#include <libevm/VMFactory.h>
#include <libethereum/All.h>
@ -354,22 +355,25 @@ int main(int argc, char** argv)
cout << credits();
unique_ptr<StructuredLogger> structuredLogger(new StructuredLogger(structuredLogging));
VMFactory::setKind(jit ? VMKind::JIT : VMKind::Interpreter);
NetworkPreferences netPrefs(listenPort, publicIP, upnp, useLocal);
auto nodesState = contents((dbPath.size() ? dbPath : getDataDir()) + "/network.rlp");
std::string clientImplString = "Ethereum(++)/" + clientName + "v" + dev::Version + "/" DEV_QUOTED(ETH_BUILD_TYPE) "/" DEV_QUOTED(ETH_BUILD_PLATFORM) + (jit ? "/JIT" : "");
dev::WebThreeDirect web3(
"Ethereum(++)/" + clientName + "v" + dev::Version + "/" DEV_QUOTED(ETH_BUILD_TYPE) "/" DEV_QUOTED(ETH_BUILD_PLATFORM) + (jit ? "/JIT" : ""),
clientImplString,
dbPath,
false,
mode == NodeMode::Full ? set<string>{"eth", "shh"} : set<string>(),
netPrefs,
&nodesState,
miners,
structuredLogging
structuredLogger.get()
);
web3.setIdealPeerCount(peers);
eth::Client* c = mode == NodeMode::Full ? web3.ethereum() : nullptr;
if (structuredLogging)
structuredLogger->logStarting(clientImplString, dev::Version);
if (c)
{
c->setForceMining(forceMining);
@ -530,7 +534,7 @@ int main(int argc, char** argv)
string sdata;
iss >> hexAddr >> amount >> gasPrice >> gas >> sechex >> sdata;
cnote << "Data:";
cnote << sdata;
bytes data = dev::eth::parseData(sdata);
@ -613,7 +617,7 @@ int main(int argc, char** argv)
if (size > 0)
cwarn << "Invalid address length:" << size;
}
else
else
{
auto const& bc =c->blockChain();
auto h = bc.currentHash();
@ -635,7 +639,7 @@ int main(int argc, char** argv)
cwarn << "transaction rejected";
}
}
}
}
else
cwarn << "Require parameters: send ADDRESS AMOUNT";
}
@ -693,7 +697,7 @@ int main(int argc, char** argv)
cwarn << "Minimum gas amount is" << minGas;
else
c->transact(us.secret(), endowment, init, gas, gasPrice);
}
}
else
cwarn << "Require parameters: contract ENDOWMENT GASPRICE GAS CODEHEX";
}
@ -811,7 +815,7 @@ int main(int argc, char** argv)
string hexSec;
iss >> hexSec;
us = KeyPair(h256(fromHex(hexSec)));
}
}
else
cwarn << "Require parameter: setSecret HEXSECRETKEY";
}
@ -852,7 +856,7 @@ int main(int argc, char** argv)
RLPStream config(2);
config << us.secret() << coinbase;
writeFile(path, config.out());
}
}
else
cwarn << "Require parameter: exportConfig PATH";
}
@ -868,10 +872,10 @@ int main(int argc, char** argv)
RLP config(b);
us = KeyPair(config[0].toHash<Secret>());
coinbase = config[1].toHash<Address>();
}
}
else
cwarn << path << "has no content!";
}
}
else
cwarn << "Require parameter: importConfig PATH";
}

1
libdevcore/CMakeLists.txt

@ -27,6 +27,7 @@ endif()
target_link_libraries(${EXECUTABLE} ${Boost_THREAD_LIBRARIES})
target_link_libraries(${EXECUTABLE} ${Boost_SYSTEM_LIBRARIES})
target_link_libraries(${EXECUTABLE} ${JSONCPP_LIBRARIES})
# transitive dependencies for windows executables
if ("${CMAKE_CXX_COMPILER_ID}" STREQUAL "MSVC")

166
libdevcore/StructuredLogger.cpp

@ -0,0 +1,166 @@
/*
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 StructuredLogger.h
* @author Lefteris Karapetsas <lefteris@ethdev.com>
* @date 2015
*
* A simple helper class for the structured logging
*/
#include "StructuredLogger.h"
#include <json/json.h>
using namespace std;
namespace dev
{
char const* StructuredLogger::timePointToString(chrono::system_clock::time_point const& _ts) const
{
// not using C++11 std::put_time due to gcc bug
// http://stackoverflow.com/questions/14136833/stdput-time-implementation-status-in-gcc
// TODO: Format it according to Log event Requirements
time_t time = chrono::system_clock::to_time_t(_ts);
return ctime(&time);
}
void StructuredLogger::outputJson(Json::Value const* _value, std::string const& _name) const
{
Json::Value event;
event[_name] = _value;
cout << event;
}
void StructuredLogger::logStarting(string const& _clientImpl, const char* _ethVersion)
{
if (m_enabled)
{
Json::Value event;
event["comment"] = "one of the first log events, before any operation is started";
event["client_implt"] = _clientImpl;
event["eth_version"] = std::string(_ethVersion);
event["ts"] = string(timePointToString(std::chrono::system_clock::now()));
outputJson(&event, "starting");
}
}
void StructuredLogger::logP2PConnected(string const& _id, bi::tcp::endpoint const& _addr,
chrono::system_clock::time_point const& _ts, unsigned int _numConnections) const
{
if (m_enabled)
{
std::stringstream addrStream;
addrStream << _addr;
Json::Value event;
event["remote_version_string"] = ""; //TODO
event["comment"] = "as soon as a successful connection to another node is established";
event["remote_addr"] = addrStream.str();
event["remote_id"] = _id;
event["num_connections"] = Json::Value(_numConnections);
event["ts"] = string(timePointToString(_ts));
outputJson(&event, "p2p.connected");
}
}
void StructuredLogger::logP2PDisconnected(string const& _id, unsigned int _numConnections, bi::tcp::endpoint const& _addr) const
{
if (m_enabled)
{
std::stringstream addrStream;
addrStream << _addr;
Json::Value event;
event["comment"] = "as soon as a disconnection from another node happened";
event["remote_addr"] = addrStream.str();
event["remote_id"] = _id;
event["num_connections"] = Json::Value(_numConnections);
event["ts"] = string(timePointToString(chrono::system_clock::now()));
outputJson(&event, "p2p.disconnected");
}
}
void StructuredLogger::logMinedNewBlock(string const& _hash, string const& _blockNumber,
string const& _chainHeadHash, string const& _prevHash) const
{
if (m_enabled)
{
Json::Value event;
event["comment"] = "as soon as the block was mined, before adding as new head";
event["block_hash"] = _hash;
event["block_number"] = _blockNumber;
event["chain_head_hash"] = _chainHeadHash;
event["ts"] = string(timePointToString(std::chrono::system_clock::now()));
event["block_prev_hash"] = _prevHash;
outputJson(&event, "eth.miner.new_block");
}
}
void StructuredLogger::logChainReceivedNewBlock(string const& _hash, string const& _blockNumber,
string const& _chainHeadHash, string const& _remoteID, string const& _prevHash) const
{
if (m_enabled)
{
Json::Value event;
event["comment"] = "whenever a _new_ block is received, before adding";
event["block_hash"] = _hash;
event["block_number"] = _blockNumber;
event["chain_head_hash"] = _chainHeadHash;
event["remote_id"] = _remoteID;
event["ts"] = string(timePointToString(chrono::system_clock::now()));
event["block_prev_hash"] = _prevHash;
outputJson(&event, "eth.chain.received.new_block");
}
}
void StructuredLogger::logChainNewHead(string const& _hash, string const& _blockNumber,
string const& _chainHeadHash, string const& _prevHash) const
{
if (m_enabled)
{
Json::Value event;
event["comment"] = "whenever head changes";
event["block_hash"] = _hash;
event["block_number"] = _blockNumber;
event["chain_head_hash"] = _chainHeadHash;
event["ts"] = string(timePointToString(chrono::system_clock::now()));
event["block_prev_hash"] = _prevHash;
outputJson(&event, "eth.miner.new_block");
}
}
void StructuredLogger::logTransactionReceived(string const& _hash, string const& _remoteId) const
{
if (m_enabled)
{
Json::Value event;
event["tx_hash"] = _hash;
event["remote_id"] = _remoteId;
event["ts"] = string(timePointToString(chrono::system_clock::now()));
outputJson(&event, "eth.tx.received");
}
}
}

59
libdevcore/StructuredLogger.h

@ -0,0 +1,59 @@
/*
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 StructuredLogger.h
* @author Lefteris Karapetsas <lefteris@ethdev.com>
* @date 2015
*
* A simple helper class for the structured logging
*/
#pragma once
#include <string>
#include <chrono>
#include <libp2p/Network.h>
namespace Json { class Value; }
namespace dev
{
class StructuredLogger
{
public:
StructuredLogger(bool _enabled): m_enabled(_enabled) {}
void logStarting(std::string const& _clientImpl, const char* _ethVersion);
void logP2PConnected(std::string const& _id, bi::tcp::endpoint const& _addr,
std::chrono::system_clock::time_point const& _ts, unsigned int _numConnections) const;
void logP2PDisconnected(std::string const& _id, unsigned int _numConnections, bi::tcp::endpoint const& _addr) const;
void logMinedNewBlock(std::string const& _hash, std::string const& _blockNumber,
std::string const& _chainHeadHash, std::string const& _prevHash) const;
void logChainReceivedNewBlock(std::string const& _hash, std::string const& _blockNumber, std::string const& _chainHeadHash,
std::string const& _remoteID, std::string const& _prevHash) const;
void logChainNewHead(std::string const& _hash, std::string const& _blockNumber, std::string const& _chainHeadHash,
std::string const& _prevHash) const;
void logTransactionReceived(std::string const& _hash, std::string const& _remoteId) const;
private:
/// @returns a string representation of a timepoint
char const* timePointToString(std::chrono::system_clock::time_point const& _ts) const;
void outputJson(Json::Value const* _value, std::string const& _name) const;
bool m_enabled;
};
}

25
libethereum/BlockChain.cpp

@ -25,6 +25,7 @@
#include <test/JsonSpiritHeaders.h>
#include <libdevcore/Common.h>
#include <libdevcore/RLP.h>
#include <libdevcore/StructuredLogger.h>
#include <libdevcrypto/FileSystem.h>
#include <libethcore/Exceptions.h>
#include <libethcore/ProofOfWork.h>
@ -152,7 +153,7 @@ inline string toString(h256s const& _bs)
return out.str();
}
h256s BlockChain::sync(BlockQueue& _bq, OverlayDB const& _stateDB, unsigned _max)
h256s BlockChain::sync(BlockQueue& _bq, OverlayDB const& _stateDB, unsigned _max, StructuredLogger const* _logger)
{
_bq.tick(*this);
@ -164,7 +165,7 @@ h256s BlockChain::sync(BlockQueue& _bq, OverlayDB const& _stateDB, unsigned _max
{
try
{
for (auto h: import(block, _stateDB))
for (auto h: import(block, _stateDB, _logger))
if (!_max--)
break;
else
@ -187,11 +188,11 @@ h256s BlockChain::sync(BlockQueue& _bq, OverlayDB const& _stateDB, unsigned _max
return ret;
}
h256s BlockChain::attemptImport(bytes const& _block, OverlayDB const& _stateDB) noexcept
h256s BlockChain::attemptImport(bytes const& _block, OverlayDB const& _stateDB, StructuredLogger const* _logger) noexcept
{
try
{
return import(_block, _stateDB);
return import(_block, _stateDB, _logger);
}
catch (...)
{
@ -200,7 +201,7 @@ h256s BlockChain::attemptImport(bytes const& _block, OverlayDB const& _stateDB)
}
}
h256s BlockChain::import(bytes const& _block, OverlayDB const& _db)
h256s BlockChain::import(bytes const& _block, OverlayDB const& _db, StructuredLogger const* _logger)
{
// VERIFY: populates from the block and checks the block is internally coherent.
BlockInfo bi;
@ -317,6 +318,13 @@ h256s BlockChain::import(bytes const& _block, OverlayDB const& _db)
}
#endif
if (_logger)
_logger->logChainReceivedNewBlock(
bi.headerHash(WithoutNonce).abridged(),
bi.nonce.abridged(),
currentHash().abridged(),
"", // TODO: remote id ??
bi.parentHash.abridged());
// cnote << "Parent " << bi.parentHash << " has " << details(bi.parentHash).children.size() << " children.";
h256s ret;
@ -331,6 +339,13 @@ h256s BlockChain::import(bytes const& _block, OverlayDB const& _db)
}
m_extrasDB->Put(m_writeOptions, ldb::Slice("best"), ldb::Slice((char const*)&newHash, 32));
clog(BlockChainNote) << " Imported and best" << td << ". Has" << (details(bi.parentHash).children.size() - 1) << "siblings. Route:" << toString(ret);
if (_logger)
_logger->logChainNewHead(
bi.headerHash(WithoutNonce).abridged(),
bi.nonce.abridged(),
currentHash().abridged(),
bi.parentHash.abridged()
);
}
else
{

7
libethereum/BlockChain.h

@ -40,6 +40,7 @@ namespace ldb = leveldb;
namespace dev
{
class StructuredLogger;
class OverlayDB;
namespace eth
@ -79,15 +80,15 @@ public:
void process();
/// Sync the chain with any incoming blocks. All blocks should, if processed in order
h256s sync(BlockQueue& _bq, OverlayDB const& _stateDB, unsigned _max);
h256s sync(BlockQueue& _bq, OverlayDB const& _stateDB, unsigned _max, StructuredLogger const* _logger);
/// Attempt to import the given block directly into the CanonBlockChain and sync with the state DB.
/// @returns the block hashes of any blocks that came into/went out of the canonical block chain.
h256s attemptImport(bytes const& _block, OverlayDB const& _stateDB) noexcept;
h256s attemptImport(bytes const& _block, OverlayDB const& _stateDB, StructuredLogger const* _logger = nullptr) noexcept;
/// Import block into disk-backed DB
/// @returns the block hashes of any blocks that came into/went out of the canonical block chain.
h256s import(bytes const& _block, OverlayDB const& _stateDB);
h256s import(bytes const& _block, OverlayDB const& _stateDB, StructuredLogger const* _logger = nullptr);
/// Returns true if the given block is known (though not necessarily a part of the canon chain).
bool isKnown(h256 _hash) const;

14
libethereum/Client.cpp

@ -25,6 +25,7 @@
#include <thread>
#include <boost/filesystem.hpp>
#include <libdevcore/Log.h>
#include <libdevcore/StructuredLogger.h>
#include <libp2p/Host.h>
#include "Defaults.h"
#include "Executive.h"
@ -60,15 +61,16 @@ void VersionChecker::setOk()
}
Client::Client(p2p::Host* _extNet, std::string const& _dbPath, bool _forceClean,
u256 _networkId, int _miners, bool _structuredLogging):
u256 _networkId, int _miners, StructuredLogger const* _structuredLogger):
Worker("eth"),
m_vc(_dbPath),
m_bc(_dbPath, !m_vc.ok() || _forceClean),
m_stateDB(State::openDB(_dbPath, !m_vc.ok() || _forceClean)),
m_preMine(Address(), m_stateDB),
m_postMine(Address(), m_stateDB)
m_postMine(Address(), m_stateDB),
m_structuredLogger(_structuredLogger)
{
m_host = _extNet->registerCapability(new EthereumHost(m_bc, m_tq, m_bq, _networkId, _structuredLogging));
m_host = _extNet->registerCapability(new EthereumHost(m_bc, m_tq, m_bq, _networkId, _structuredLogger));
if (_miners > -1)
setMiningThreads(_miners);
@ -416,6 +418,8 @@ void Client::transact(Secret _secret, u256 _value, Address _dest, bytes const& _
}
Transaction t(_value, _gasPrice, _gas, _dest, _data, n, _secret);
// cdebug << "Nonce at " << toAddress(_secret) << " pre:" << m_preMine.transactionsFrom(toAddress(_secret)) << " post:" << m_postMine.transactionsFrom(toAddress(_secret));
if (m_structuredLogger)
m_structuredLogger->logTransactionReceived(t.sha3().abridged(), t.sender().abridged());
cnote << "New transaction " << t;
m_tq.attemptImport(t.rlp());
}
@ -521,7 +525,7 @@ void Client::doWork()
h256s hs;
{
WriteGuard l(x_stateDB);
hs = m_bc.attemptImport(m.blockData(), m_stateDB);
hs = m_bc.attemptImport(m.blockData(), m_stateDB, m_structuredLogger);
}
if (hs.size())
{
@ -556,7 +560,7 @@ void Client::doWork()
cwork << "BQ ==> CHAIN ==> STATE";
OverlayDB db = m_stateDB;
x_stateDB.unlock();
h256s newBlocks = m_bc.sync(m_bq, db, 100); // TODO: remove transactions from m_tq nicely rather than relying on out of date nonce later on.
h256s newBlocks = m_bc.sync(m_bq, db, 100, m_structuredLogger); // TODO: remove transactions from m_tq nicely rather than relying on out of date nonce later on.
if (newBlocks.size())
{
for (auto i: newBlocks)

6
libethereum/Client.h

@ -44,6 +44,9 @@
namespace dev
{
class StructuredLogger;
namespace eth
{
@ -167,7 +170,7 @@ class Client: public MinerHost, public Interface, Worker
public:
/// New-style Constructor.
explicit Client(p2p::Host* _host, std::string const& _dbPath = std::string(), bool _forceClean = false,
u256 _networkId = 0, int _miners = -1, bool _structuredLogging = false);
u256 _networkId = 0, int _miners = -1, StructuredLogger const* _structuredLogger = nullptr);
/// Destructor.
virtual ~Client();
@ -365,6 +368,7 @@ private:
std::map<h256, InstalledFilter> m_filters;
std::map<unsigned, ClientWatch> m_watches;
StructuredLogger const* m_structuredLogger;
mutable std::chrono::system_clock::time_point m_lastGarbageCollection;
};

5
libethereum/Miner.cpp

@ -36,9 +36,10 @@ LocalMiner::LocalMiner(MinerHost* _host, unsigned _id):
{
}
void LocalMiner::setup(MinerHost* _host, unsigned _id)
void LocalMiner::setup(MinerHost* _host, unsigned _id, StructuredLogger const* _structuredLogger)
{
m_host = _host;
m_structuredLogger = _structuredLogger;
setName("miner-" + toString(_id));
}
@ -79,7 +80,7 @@ void LocalMiner::doWork()
}
if (mineInfo.completed)
{
m_mineState.completeMine();
m_mineState.completeMine(m_structuredLogger);
m_host->onComplete();
m_miningStatus = Mined;
}

5
libethereum/Miner.h

@ -32,6 +32,8 @@
namespace dev
{
class StructuredLogger;
namespace eth
{
@ -104,7 +106,7 @@ public:
~LocalMiner() { stop(); }
/// Setup its basics.
void setup(MinerHost* _host, unsigned _id = 0);
void setup(MinerHost* _host, unsigned _id = 0, StructuredLogger const* _structuredLogger = nullptr);
/// Start mining.
void start() { startWorking(); }
@ -143,6 +145,7 @@ private:
mutable std::mutex x_mineInfo; ///< Lock for the mining progress & history.
MineProgress m_mineProgress; ///< What's our progress?
std::list<MineInfo> m_mineHistory; ///< What the history of our mining?
StructuredLogger const* m_structuredLogger;
};
}

12
libethereum/State.cpp

@ -27,6 +27,7 @@
#include <boost/timer.hpp>
#include <secp256k1/secp256k1.h>
#include <libdevcore/CommonIO.h>
#include <libdevcore/StructuredLogger.h>
#include <libevmcore/Instruction.h>
#include <libethcore/Exceptions.h>
#include <libevm/VMFactory.h>
@ -801,12 +802,12 @@ bool State::completeMine(h256 const& _nonce)
m_currentBlock.nonce = _nonce;
cnote << "Completed" << m_currentBlock.headerHash(WithoutNonce).abridged() << m_currentBlock.nonce.abridged() << m_currentBlock.difficulty << ProofOfWork::verify(m_currentBlock.headerHash(WithoutNonce), m_currentBlock.nonce, m_currentBlock.difficulty);
completeMine();
completeMine(nullptr);
return true;
}
void State::completeMine()
void State::completeMine(StructuredLogger const* _structuredLogger)
{
cdebug << "Completing mine!";
// Got it!
@ -820,6 +821,13 @@ void State::completeMine()
ret.swapOut(m_currentBytes);
m_currentBlock.hash = sha3(RLP(m_currentBytes)[0].data());
cnote << "Mined " << m_currentBlock.hash.abridged() << "(parent: " << m_currentBlock.parentHash.abridged() << ")";
if (_structuredLogger)
_structuredLogger->logMinedNewBlock(
m_currentBlock.hash.abridged(),
m_currentBlock.nonce.abridged(),
"", //TODO: chain head hash here ??
m_currentBlock.parentHash.abridged()
);
// Quickly reset the transactions.
// TODO: Leave this in a better state than this limbo, or at least record that it's in limbo.

4
libethereum/State.h

@ -41,6 +41,8 @@
namespace dev
{
class StructuredLogger;
namespace test { class ImportTest; }
namespace eth
@ -139,7 +141,7 @@ public:
* // unlock
* @endcode
*/
void completeMine();
void completeMine(StructuredLogger const* _structuredLogger = nullptr);
/// Get the complete current block, including valid nonce.
/// Only valid after mine() returns true.

9
libp2p/Host.cpp

@ -27,6 +27,7 @@
#include <boost/algorithm/string.hpp>
#include <libdevcore/Common.h>
#include <libdevcore/CommonIO.h>
#include <libdevcore/StructuredLogger.h>
#include <libethcore/Exceptions.h>
#include <libdevcrypto/FileSystem.h>
#include "Session.h"
@ -45,7 +46,7 @@ void HostNodeTableHandler::processEvent(NodeId const& _n, NodeTableEventType con
m_host.onNodeTableEvent(_n, _e);
}
Host::Host(std::string const& _clientVersion, NetworkPreferences const& _n, bytesConstRef _restoreNetwork, bool _structuredLogging):
Host::Host(std::string const& _clientVersion, NetworkPreferences const& _n, bytesConstRef _restoreNetwork, StructuredLogger const* _structuredLogger):
Worker("p2p", 0),
m_restoreNetwork(_restoreNetwork.toBytes()),
m_clientVersion(_clientVersion),
@ -55,7 +56,7 @@ Host::Host(std::string const& _clientVersion, NetworkPreferences const& _n, byte
m_tcp4Acceptor(m_ioService),
m_alias(networkAlias(_restoreNetwork)),
m_lastPing(chrono::steady_clock::time_point::min()),
m_structuredLogging(_structuredLogging)
m_structuredLogger(_structuredLogger)
{
for (auto address: m_ifAddresses)
if (address.is_v4())
@ -476,6 +477,10 @@ void Host::connect(std::shared_ptr<Peer> const& _p)
_p->m_lastConnected = std::chrono::system_clock::now();
_p->m_failedAttempts = 0;
if (m_structuredLogger)
m_structuredLogger->logP2PConnected(_p->id.abridged(), _p->peerEndpoint(), _p->m_lastConnected,
0);// TODO: num_connections
auto ps = make_shared<Session>(this, std::move(*s), _p);
ps->start();

66
libp2p/Host.h

@ -45,6 +45,8 @@ namespace bi = ba::ip;
namespace dev
{
class StructuredLogger;
namespace p2p
{
@ -56,7 +58,7 @@ public:
HostNodeTableHandler(Host& _host);
Host const& host() const { return m_host; }
private:
virtual void processEvent(NodeId const& _n, NodeTableEventType const& _e);
@ -82,23 +84,24 @@ class Host: public Worker
friend class HostNodeTableHandler;
friend class Session;
friend class HostCapabilityFace;
public:
/// Start server, listening for connections on the given port.
Host(std::string const& _clientVersion, NetworkPreferences const& _n = NetworkPreferences(), bytesConstRef _restoreNetwork = bytesConstRef(), bool _structuredLogging);
Host(std::string const& _clientVersion, NetworkPreferences const& _n = NetworkPreferences(),
bytesConstRef _restoreNetwork = bytesConstRef(), StructuredLogger const* _structuredLogger = nullptr);
/// Will block on network process events.
virtual ~Host();
/// Interval at which Host::run will call keepAlivePeers to ping peers.
std::chrono::seconds const c_keepAliveInterval = std::chrono::seconds(30);
/// Disconnect timeout after failure to respond to keepAlivePeers ping.
std::chrono::milliseconds const c_keepAliveTimeOut = std::chrono::milliseconds(1000);
/// Default host for current version of client.
static std::string pocHost();
/// Basic peer network protocol version.
unsigned protocolVersion() const;
@ -108,23 +111,23 @@ public:
bool haveCapability(CapDesc const& _name) const { return m_capabilities.count(_name) != 0; }
CapDescs caps() const { CapDescs ret; for (auto const& i: m_capabilities) ret.push_back(i.first); return ret; }
template <class T> std::shared_ptr<T> cap() const { try { return std::static_pointer_cast<T>(m_capabilities.at(std::make_pair(T::staticName(), T::staticVersion()))); } catch (...) { return nullptr; } }
bool havePeerSession(NodeId _id) { RecursiveGuard l(x_sessions); return m_sessions.count(_id) ? !!m_sessions[_id].lock() : false; }
void addNode(NodeId const& _node, std::string const& _addr, unsigned short _tcpPort, unsigned short _udpPort);
/// Set ideal number of peers.
void setIdealPeerCount(unsigned _n) { m_idealPeerCount = _n; }
/// Get peer information.
PeerSessionInfos peerSessionInfo() const;
/// Get number of peers connected.
size_t peerCount() const;
/// Get the address we're listening on currently.
std::string listenAddress() const { return m_tcpPublic.address().to_string(); }
/// Get the port we're listening on currently.
unsigned short listenPort() const { return m_tcpPublic.port(); }
@ -138,11 +141,11 @@ public:
/// Start network. @threadsafe
void start();
/// Stop network. @threadsafe
/// Resets acceptor, socket, and IO service. Called by deallocator.
void stop();
/// @returns if network is running.
bool isStarted() const { return m_run; }
@ -155,25 +158,25 @@ protected:
/// Deserialise the data and populate the set of known peers.
void restoreNetwork(bytesConstRef _b);
private:
/// Populate m_peerAddresses with available public addresses.
void determinePublic(std::string const& _publicAddress, bool _upnp);
void connect(std::shared_ptr<Peer> const& _p);
/// Ping the peers to update the latency information and disconnect peers which have timed out.
void keepAlivePeers();
/// Disconnect peers which didn't respond to keepAlivePeers ping prior to c_keepAliveTimeOut.
void disconnectLatePeers();
/// Called only from startedWorking().
void runAcceptor();
/// Handler for verifying handshake siganture before creating session. _nodeId is passed for outbound connections. If successful, socket is moved to Session via std::move.
void doHandshake(bi::tcp::socket* _socket, NodeId _nodeId = NodeId());
void seal(bytes& _b);
/// Called by Worker. Not thread-safe; to be called only by worker.
@ -183,7 +186,7 @@ private:
/// Run network. Not thread-safe; to be called only by worker.
virtual void doWork();
/// Shutdown network. Not thread-safe; to be called only by worker.
virtual void doneWorking();
@ -191,14 +194,14 @@ private:
static KeyPair networkAlias(bytesConstRef _b);
bytes m_restoreNetwork; ///< Set by constructor and used to set Host key and restore network peers & nodes.
bool m_run = false; ///< Whether network is running.
std::mutex x_runTimer; ///< Start/stop mutex.
std::string m_clientVersion; ///< Our version string.
NetworkPreferences m_netPrefs; ///< Network settings.
/// Interface addresses (private, public)
std::vector<bi::address> m_ifAddresses; ///< Interface addresses.
@ -206,10 +209,10 @@ private:
ba::io_service m_ioService; ///< IOService for network stuff.
bi::tcp::acceptor m_tcp4Acceptor; ///< Listening acceptor.
std::unique_ptr<boost::asio::deadline_timer> m_timer; ///< Timer which, when network is running, calls scheduler() every c_timerInterval ms.
static const unsigned c_timerInterval = 100; ///< Interval which m_timer is run when network is connected.
std::set<Peer*> m_pendingPeerConns; /// Used only by connect(Peer&) to limit concurrently connecting to same node. See connect(shared_ptr<Peer>const&).
Mutex x_pendingNodeConns;
@ -219,23 +222,22 @@ private:
/// Shared storage of Peer objects. Peers are created or destroyed on demand by the Host. Active sessions maintain a shared_ptr to a Peer;
std::map<NodeId, std::shared_ptr<Peer>> m_peers;
/// The nodes to which we are currently connected. Used by host to service peer requests and keepAlivePeers and for shutdown. (see run())
/// Mutable because we flush zombie entries (null-weakptrs) as regular maintenance from a const method.
mutable std::map<NodeId, std::weak_ptr<Session>> m_sessions;
mutable RecursiveMutex x_sessions;
unsigned m_idealPeerCount = 5; ///< Ideal number of peers to be connected to.
std::set<bi::address> m_peerAddresses; ///< Public addresses that peers (can) know us by.
std::map<CapDesc, std::shared_ptr<HostCapabilityFace>> m_capabilities; ///< Each of the capabilities we support.
std::chrono::steady_clock::time_point m_lastPing; ///< Time we sent the last ping to all peers.
StructuredLogger const* m_structuredLogger;
bool m_accepting = false;
bool m_structuredLogging;
};
}
}

8
libwebthree/WebThree.cpp

@ -37,14 +37,14 @@ using namespace dev::shh;
WebThreeDirect::WebThreeDirect(std::string const& _clientVersion, std::string const& _dbPath, bool _forceClean,
std::set<std::string> const& _interfaces, NetworkPreferences const& _n,
bytesConstRef _network, int _miners, bool _structuredLogging):
m_clientVersion(_clientVersion),
m_net(_clientVersion, _n, _network)
bytesConstRef _network, int _miners, StructuredLogger const* _structuredLogger):
m_clientVersion(_clientVersion),
m_net(_clientVersion, _n, _network, _structuredLogger)
{
if (_dbPath.size())
Defaults::setDBPath(_dbPath);
if (_interfaces.count("eth"))
m_ethereum.reset(new eth::Client(&m_net, _dbPath, _forceClean, 0, _miners, _structuredLogging));
m_ethereum.reset(new eth::Client(&m_net, _dbPath, _forceClean, 0, _miners, _structuredLogger));
if (_interfaces.count("shh"))
m_whisper = m_net.registerCapability<WhisperHost>(new WhisperHost);

2
libwebthree/WebThree.h

@ -106,7 +106,7 @@ public:
WebThreeDirect(std::string const& _clientVersion, std::string const& _dbPath, bool _forceClean = false,
std::set<std::string> const& _interfaces = {"eth", "shh"},
p2p::NetworkPreferences const& _n = p2p::NetworkPreferences(),
bytesConstRef _network = bytesConstRef(), int _miners = -1, bool _structuredLogging = false);
bytesConstRef _network = bytesConstRef(), int _miners = -1, StructuredLogger const* _structuredLogger = nullptr);
/// Destructor.
~WebThreeDirect();

Loading…
Cancel
Save