NodeTable into own files.

libp2p/NodeTable.cpp

@@ -0,0 +1,351 @@
+/*
+ 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 NodeTable.cpp
+ * @author Alex Leverington <nessence@gmail.com>
+ * @date 2014
+ */
+
+#include "NodeTable.h"
+using namespace std;
+using namespace dev;
+using namespace dev::p2p;
+
+NodeTable::NodeTable(ba::io_service& _io):
+		m_node(NodeEntry(Address(), Public(), bi::udp::endpoint())),
+		m_socket(new nodeSocket(_io, *this, 30300)),
+		m_socketPtr(m_socket.get()),
+		m_io(_io),
+		m_bucketRefreshTimer(m_io),
+		m_evictionCheckTimer(m_io)
+	{
+		for (unsigned i = 0; i < s_bins; i++)
+			m_state[i].distance = i, m_state[i].modified = chrono::steady_clock::now() - chrono::seconds(1);
+		doRefreshBuckets(boost::system::error_code());
+	}
+	
+NodeTable::~NodeTable()
+{
+	m_evictionCheckTimer.cancel();
+	m_bucketRefreshTimer.cancel();
+	m_socketPtr->disconnect();
+}
+	
+	void NodeTable::join()
+{
+	doFindNode(m_node.id);
+}
+	
+std::list<Address> NodeTable::nodes() const
+{
+	std::list<Address> nodes;
+	Guard l(x_nodes);
+	for (auto& i: m_nodes)
+		nodes.push_back(i.second->id);
+	return std::move(nodes);
+}
+
+NodeTable::NodeEntry NodeTable::operator[](Address _id)
+{
+	Guard l(x_nodes);
+	return *m_nodes[_id];
+}
+
+void NodeTable::requestNeighbors(NodeEntry const& _node, Address _target) const
+{
+	FindNeighbors p;
+	p.target = _target;
+	
+	p.to = _node.endpoint.udp;
+	p.seal(m_secret);
+	m_socketPtr->send(p);
+}
+
+void NodeTable::doFindNode(Address _node, unsigned _round, std::shared_ptr<std::set<std::shared_ptr<NodeEntry>>> _tried)
+{
+	if (!m_socketPtr->isOpen() || _round == 7)
+		return;
+
+	auto nearest = findNearest(_node);
+	std::list<std::shared_ptr<NodeEntry>> tried;
+	for (unsigned i = 0; i < nearest.size() && tried.size() < s_alpha; i++)
+		if (!_tried->count(nearest[i]))
+		{
+			tried.push_back(nearest[i]);
+			requestNeighbors(*nearest[i], _node);
+		}
+		else
+			continue;
+	
+	while (auto n = tried.front())
+	{
+		_tried->insert(n);
+		tried.pop_front();
+	}
+	
+	auto self(shared_from_this());
+	m_evictionCheckTimer.expires_from_now(boost::posix_time::milliseconds(s_findTimout));
+	m_evictionCheckTimer.async_wait([this, self, _node, _round, _tried](boost::system::error_code const& _ec)
+	{
+		if (_ec)
+			return;
+		doFindNode(_node, _round + 1, _tried);
+	});
+}
+
+std::vector<std::shared_ptr<NodeTable::NodeEntry>> NodeTable::findNearest(Address _target)
+{
+	// send s_alpha FindNeighbors packets to nodes we know, closest to target
+	unsigned head = dist(m_node.id, _target);
+	unsigned tail = (head - 1) % (s_bits - 1);
+	
+	// todo: optimize with tree
+	std::map<unsigned, std::list<std::shared_ptr<NodeEntry>>> found;
+	unsigned count = 0;
+	
+	// if d is 0, then we roll look forward, if last, we reverse, else, spread from d
+	if (head != 0 && tail != s_bits)
+		while (head != tail && count < s_bucketSize)
+		{
+			Guard l(x_state);
+			for (auto& n: m_state[head].nodes)
+				if (auto p = n.lock())
+				{
+					if (count < s_bucketSize)
+						found[dist(_target, p->id)].push_back(p);
+					else
+						break;
+				}
+			
+			if (count < s_bucketSize && head)
+				for (auto& n: m_state[tail].nodes)
+					if (auto p = n.lock())
+					{
+						if (count < s_bucketSize)
+							found[dist(_target, p->id)].push_back(p);
+						else
+							break;
+					}
+			head++;
+			tail = (tail - 1) % (s_bits - 1);
+		}
+	else if (head == 0)
+		while (head < s_bucketSize && count < s_bucketSize)
+		{
+			Guard l(x_state);
+			for (auto& n: m_state[head].nodes)
+				if (auto p = n.lock())
+				{
+					if (count < s_bucketSize)
+						found[dist(_target, p->id)].push_back(p);
+					else
+						break;
+				}
+			head--;
+		}
+	else if (tail == s_bits - 1)
+		while (tail > 0 && count < s_bucketSize)
+		{
+			Guard l(x_state);
+			for (auto& n: m_state[tail].nodes)
+				if (auto p = n.lock())
+				{
+					if (count < s_bucketSize)
+						found[dist(_target, p->id)].push_back(p);
+					else
+						break;
+				}
+			tail--;
+		}
+	
+	std::vector<std::shared_ptr<NodeEntry>> ret;
+	for (auto& nodes: found)
+		for (auto& n: nodes.second)
+			ret.push_back(n);
+	return std::move(ret);
+}
+
+void NodeTable::ping(bi::address _address, unsigned _port) const
+{
+	PingNode p;
+	string ip = m_node.endpoint.udp.address().to_string();
+	p.ipAddress = asBytes(ip);
+	p.port = m_node.endpoint.udp.port();
+//		p.expiration;
+	p.seal(m_secret);
+	m_socketPtr->send(p);
+}
+
+void NodeTable::ping(NodeEntry* _n) const
+{
+	if (_n && _n->endpoint.udp.address().is_v4())
+		ping(_n->endpoint.udp.address(), _n->endpoint.udp.port());
+}
+
+void NodeTable::evict(std::shared_ptr<NodeEntry> _leastSeen, std::shared_ptr<NodeEntry> _new)
+{
+	if (!m_socketPtr->isOpen())
+		return;
+	
+	Guard l(x_evictions);
+	m_evictions.push_back(EvictionTimeout(make_pair(_leastSeen->id,chrono::steady_clock::now()), _new->id));
+	if (m_evictions.size() == 1)
+		doCheckEvictions(boost::system::error_code());
+	
+	m_evictions.push_back(EvictionTimeout(make_pair(_leastSeen->id,chrono::steady_clock::now()), _new->id));
+	ping(_leastSeen.get());
+}
+
+void NodeTable::noteNode(Public _pubk, bi::udp::endpoint _endpoint)
+{
+	Address id = right160(sha3(_pubk));
+	std::shared_ptr<NodeEntry> node;
+	{
+		Guard l(x_nodes);
+		auto n = m_nodes.find(id);
+		if (n == m_nodes.end())
+		{
+			m_nodes[id] = std::shared_ptr<NodeEntry>(new NodeEntry(m_node, id, _pubk, _endpoint));
+			node = m_nodes[id];
+		}
+		else
+			node = n->second;
+	}
+	
+	noteNode(node);
+}
+
+void NodeTable::noteNode(std::shared_ptr<NodeEntry> _n)
+{
+	std::shared_ptr<NodeEntry> contested;
+	{
+		NodeBucket s = bucket(_n.get());
+		Guard l(x_state);
+		s.nodes.remove_if([&_n](std::weak_ptr<NodeEntry> n)
+		{
+			auto p = n.lock();
+			if (!p || p == _n)
+				return true;
+			return false;
+		});
+
+		if (s.nodes.size() >= s_bucketSize)
+		{
+			contested = s.nodes.front().lock();
+			if (!contested)
+			{
+				s.nodes.pop_front();
+				s.nodes.push_back(_n);
+			}
+		}
+		else
+			s.nodes.push_back(_n);
+	}
+	
+	if (contested)
+		evict(contested, _n);
+}
+
+void NodeTable::dropNode(std::shared_ptr<NodeEntry> _n)
+{
+	NodeBucket s = bucket(_n.get());
+	{
+		Guard l(x_state);
+		s.nodes.remove_if([&_n](std::weak_ptr<NodeEntry> n) { return n.lock() == _n; });
+	}
+	Guard l(x_nodes);
+	m_nodes.erase(_n->id);
+}
+
+NodeTable::NodeBucket const& NodeTable::bucket(NodeEntry* _n) const
+{
+	return m_state[_n->distance];
+}
+
+void NodeTable::onReceived(UDPSocketFace*, bi::udp::endpoint const& _from, bytesConstRef _packet)
+{
+	RLP rlp(_packet);
+	
+	
+	// whenever a pong is received, first check if it's in m_evictions, if so, remove it
+	Guard l(x_evictions);
+}
+	
+void NodeTable::doCheckEvictions(boost::system::error_code const& _ec)
+{
+	if (_ec || !m_socketPtr->isOpen())
+		return;
+
+	m_evictionCheckTimer.expires_from_now(boost::posix_time::milliseconds(s_evictionCheckInterval));
+	auto self(shared_from_this());
+	m_evictionCheckTimer.async_wait([this, self](boost::system::error_code const& _ec)
+	{
+		if (_ec)
+			return;
+		
+		bool evictionsRemain = false;
+		std::list<shared_ptr<NodeEntry>> drop;
+		{
+			Guard l(x_evictions);
+			for (auto& e: m_evictions)
+				if (chrono::steady_clock::now() - e.first.second > chrono::milliseconds(s_pingTimeout))
+				{
+					Guard l(x_nodes);
+					drop.push_back(m_nodes[e.second]);
+				}
+			evictionsRemain = m_evictions.size() - drop.size() > 0;
+		}
+		
+		for (auto& n: drop)
+			dropNode(n);
+		
+		if (evictionsRemain)
+			doCheckEvictions(boost::system::error_code());
+	});
+}
+
+void NodeTable::doRefreshBuckets(boost::system::error_code const& _ec)
+{
+	cout << "refreshing buckets" << endl;
+	if (_ec)
+		return;
+	
+	bool connected = m_socketPtr->isOpen();
+	bool refreshed = false;
+	if (connected)
+	{
+		Guard l(x_state);
+		for (auto& d: m_state)
+			if (chrono::steady_clock::now() - d.modified > chrono::seconds(s_bucketRefresh))
+				while (!d.nodes.empty())
+				{
+					auto n = d.nodes.front();
+					if (auto p = n.lock())
+					{
+						refreshed = true;
+						ping(p.get());
+						break;
+					}
+					d.nodes.pop_front();
+				}
+	}
+
+	unsigned nextRefresh = connected ? (refreshed ? 200 : s_bucketRefresh*1000) : 10000;
+	auto runcb = [this](boost::system::error_code const& error) -> void { doRefreshBuckets(error); };
+	m_bucketRefreshTimer.expires_from_now(boost::posix_time::milliseconds(nextRefresh));
+	m_bucketRefreshTimer.async_wait(runcb);
+}
+

libp2p/NodeTable.h

@@ -0,0 +1,226 @@
+/*
+ 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 NodeTable.h
+ * @author Alex Leverington <nessence@gmail.com>
+ * @date 2014
+ */
+
+#pragma once
+
+#include <libdevcrypto/Common.h>
+#include <libp2p/UDP.h>
+
+namespace dev
+{
+namespace p2p
+{
+
+/**
+ * Ping packet: Check if node is alive.
+ * PingNode is cached and regenerated after expiration - t, where t is timeout.
+ *
+ * signature: Signature of message.
+ * ipAddress: Our IP address.
+ * port: Our port.
+ * expiration: Triggers regeneration of packet. May also provide control over synchronization.
+ *
+ * Ping is used to implement evict. When a new node is seen for
+ * a given bucket which is full, the least-responsive node is pinged.
+ * If the pinged node doesn't respond then it is removed and the new
+ * node is inserted.
+ */
+struct PingNode: RLPDatagram
+{
+	bytes ipAddress;
+	uint16_t port;
+	uint64_t expiration;
+
+	Signature signature;
+	
+	void streamRLP(RLPStream& _s) const { _s.appendList(3); _s << ipAddress << port << expiration; }
+};
+
+struct Pong: RLPDatagram
+{
+	// todo: weak-signed pong
+	Address from;
+	uint64_t replyTo;	/// expiration from PingNode
+	
+	void streamRLP(RLPStream& _s) const { _s.appendList(2); _s << from << replyTo; }
+};
+
+/**
+ * FindNeighbors Packet: Request k-nodes, closest to the target.
+ * FindNeighbors is cached and regenerated after expiration - t, where t is timeout.
+ *
+ * signature: Signature of message.
+ * target: Address of NodeId. The responding node will send back nodes closest to the target.
+ * expiration: Triggers regeneration of packet. May also provide control over synchronization.
+ *
+ */
+struct FindNeighbors: RLPDatagram
+{
+	h160 target;
+	uint64_t expiration;
+	
+	Signature signature;
+	
+	void streamRLP(RLPStream& _s) const { _s.appendList(2); _s << target << expiration; }
+};
+
+/**
+ * Node Packet: Multiple node packets are sent in response to FindNeighbors.
+ */
+struct Neighbors: RLPDatagram
+{
+	struct Node
+	{
+		bytes ipAddress;
+		uint16_t port;
+		NodeId node;
+		void streamRLP(RLPStream& _s) const { _s.appendList(3); _s << ipAddress << port << node; }
+	};
+	
+	std::set<Node> nodes;
+	h256 nonce;
+	
+	Signature signature;
+	
+	void streamRLP(RLPStream& _s) const { _s.appendList(2); _s.appendList(nodes.size()); for (auto& n: nodes) n.streamRLP(_s); _s << nonce; }
+};
+
+/**
+ * NodeTable using S/Kademlia system for node discovery and preference.
+ * untouched buckets are refreshed if they have not been touched within an hour
+ *
+ * Thread-safety is ensured by modifying NodeEntry details via 
+ * shared_ptr replacement instead of mutating values.
+ *
+ * @todo don't try to evict node if node isRequired. (support for makeRequired)
+ * @todo optimize (use tree for state (or set w/custom compare for cache))
+ * @todo constructor support for m_node, m_secret
+ * @todo use s_bitsPerStep for find and refresh/ping
+ * @todo exclude bucket from refresh if we have node as peer
+ * @todo restore nodes
+ */
+class NodeTable: UDPSocketEvents, public std::enable_shared_from_this<NodeTable>
+{
+	using nodeSocket = UDPSocket<NodeTable, 1024>;
+	using timePoint = std::chrono::steady_clock::time_point;
+	
+	static unsigned const s_bucketSize = 16;			// Denoted by k in [Kademlia]. Number of nodes stored in each bucket.
+//	const unsigned s_bitsPerStep = 5;					// @todo Denoted by b in [Kademlia]. Bits by which address space will be divided for find responses.
+	static unsigned const s_alpha = 3;				// Denoted by \alpha in [Kademlia]. Number of concurrent FindNeighbors requests.
+	const unsigned s_findTimout = 300;				// How long to wait between find queries.
+//	const unsigned s_siblings = 5;					// @todo Denoted by s in [S/Kademlia]. User-defined by sub-protocols.
+	const unsigned s_bucketRefresh = 3600;				// Refresh interval prevents bucket from becoming stale. [Kademlia]
+	static unsigned const s_bits = 8 * Address::size;	// Denoted by n.
+	static unsigned const s_bins = s_bits - 1;			//
+	const unsigned s_evictionCheckInterval = 75;		// Interval by which eviction timeouts are checked.
+	const unsigned s_pingTimeout = 500;
+	
+public:
+	static unsigned dist(Address const& _a, Address const& _b) { u160 d = _a ^ _b; unsigned ret; for (ret = 0; d >>= 1; ++ret) {}; return ret; }
+
+	struct NodeDefaultEndpoint
+	{
+		NodeDefaultEndpoint(bi::udp::endpoint _udp): udp(_udp) {}
+		bi::udp::endpoint udp;
+	};
+	
+	struct NodeEntry
+	{
+		NodeEntry(Address _id, Public _pubk, bi::udp::endpoint _udp): id(_id), pubk(_pubk), endpoint(NodeDefaultEndpoint(_udp)), distance(0) {}
+		NodeEntry(NodeEntry _src, Address _id, Public _pubk, bi::udp::endpoint _udp): id(_id), pubk(_pubk), endpoint(NodeDefaultEndpoint(_udp)), distance(dist(_src.id,_id)) {}
+		NodeEntry(NodeEntry _src, Address _id, Public _pubk, NodeDefaultEndpoint _gw): id(_id), pubk(_pubk), endpoint(_gw), distance(dist(_src.id,_id)) {}
+		Address id;
+		Public pubk;
+		NodeDefaultEndpoint endpoint;		///< How we've previously connected to this node. (must match node's reported endpoint)
+		const unsigned distance;
+		timePoint activePing;
+	};
+	
+	struct NodeBucket
+	{
+		unsigned distance;
+		timePoint modified;
+		std::list<std::weak_ptr<NodeEntry>> nodes;
+	};
+	
+	using EvictionTimeout = std::pair<std::pair<Address,timePoint>,Address>;
+
+	NodeTable(ba::io_service& _io);
+	~NodeTable();
+	
+	void join();
+	
+	std::list<Address> nodes() const;
+	
+	NodeEntry operator[](Address _id);
+	
+protected:
+	void requestNeighbors(NodeEntry const& _node, Address _target) const;
+	
+	/// Sends requests to other nodes requesting nodes "near" to us in order to populate node table such that connected nodes form centrality.
+	void doFindNode(Address _node, unsigned _round = 0, std::shared_ptr<std::set<std::shared_ptr<NodeEntry>>> _tried = std::shared_ptr<std::set<std::shared_ptr<NodeEntry>>>());
+
+	std::vector<std::shared_ptr<NodeEntry>> findNearest(Address _target);
+	
+	void ping(bi::address _address, unsigned _port) const;
+	
+	void ping(NodeEntry* _n) const;
+	
+	void evict(std::shared_ptr<NodeEntry> _leastSeen, std::shared_ptr<NodeEntry> _new);
+	
+	void noteNode(Public _pubk, bi::udp::endpoint _endpoint);
+	
+	void noteNode(std::shared_ptr<NodeEntry> _n);
+	
+	void dropNode(std::shared_ptr<NodeEntry> _n);
+	
+	NodeBucket const& bucket(NodeEntry* _n) const;
+	
+	void onReceived(UDPSocketFace*, bi::udp::endpoint const& _from, bytesConstRef _packet);
+	
+	void onDisconnected(UDPSocketFace*) {};
+	
+	void doCheckEvictions(boost::system::error_code const& _ec);
+	
+	void doRefreshBuckets(boost::system::error_code const& _ec);
+	
+private:
+	NodeEntry m_node;										///< This node.
+	Secret m_secret;											///< This nodes secret key.
+
+	mutable Mutex x_nodes;									///< Mutable for thread-safe copy in nodes() const.
+	std::map<Address, std::shared_ptr<NodeEntry>> m_nodes;		///< Address -> Node table (most common lookup path)
+
+	Mutex x_state;
+	std::array<NodeBucket, s_bins> m_state;					///< State table of binned nodes.
+
+	Mutex x_evictions;
+	std::deque<EvictionTimeout> m_evictions;					///< Eviction timeouts.
+	
+	std::shared_ptr<nodeSocket> m_socket;							///< Shared pointer for our UDPSocket; ASIO requires shared_ptr.
+	nodeSocket* m_socketPtr;									///< Set to m_socket.get().
+	ba::io_service& m_io;										///< Used by bucket refresh timer.
+	boost::asio::deadline_timer m_bucketRefreshTimer;			///< Timer which schedules and enacts bucket refresh.
+	boost::asio::deadline_timer m_evictionCheckTimer;			///< Timer for handling node evictions.
+};
+
+}
+}