node.hpp

// anynet
// Copyright (C) 2009  Steven Siloti
//
// This program 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 2
// of the License, or (at your option) any later version.
//
// This program 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 this program; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
//
// In addition, as a special exception, the copyright holders give
// permission to link the code of portions of this program with the
// OpenSSL library under certain conditions as described in each
// individual source file, and distribute linked combinations
// including the two.
//
// You must obey the GNU General Public License in all respects
// for all of the code used other than OpenSSL.  If you modify
// file(s) with this exception, you may extend this exception to your
// version of the file(s), but you are not obligated to do so.  If you
// do not wish to do so, delete this exception statement from your
// version.  If you delete this exception statement from all source
// files in the program, then also delete it here.
//
// Contact:  Steven Siloti <ssiloti@gmail.com>

#ifndef NODE_HPP
#define NODE_HPP

#include <glog/logging.h>

#include "known_peers.hpp"
#include "connection.hpp"
#include <protocol.hpp>
#include "hunk.hpp"
#include "authority.hpp"
#include "config.hpp"
#include <boost/asio/deadline_timer.hpp>
#include <boost/date_time/posix_time/posix_time_types.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/function.hpp>
#include <vector>
#include <map>
#include <set>

class bad_content : public std::exception {};

class local_node : public boost::enable_shared_from_this<local_node>
{
#ifdef SIMULATION
	friend class traffic_generator;
#endif

	static const int replication_factor = 20;

public:
	local_node(boost::asio::io_service& io_service, client_config& config);
	~local_node();

	void start();
	void stop();

	// getters
	boost::asio::io_service& io_service()                        { return acceptor_.get_io_service(); }
	const network_key& id() const                                { return identity_; }
	const ip::tcp::endpoint& public_endpoint() const             { return public_endpoint_; }
	bool is_v4() const                                           { return public_endpoint_.address().is_v4(); }
	bool is_v6() const                                           { return public_endpoint_.address().is_v6(); }
	std::vector<connection::ptr_t>::size_type connection_count() { return ib_peers_.size(); }
	const client_config& config() const                          { return config_; }
	boost::posix_time::time_duration age() const                 { return boost::posix_time::second_clock::universal_time() - created_; }
	std::size_t average_oob_threshold() const                    { return avg_oob_threshold_; }

	void update_threshold_stats();

	std::vector<protocol_id> supported_protocols() const;

	boost::posix_time::time_duration base_hunk_lifetime();

	// connections
	void make_connection(ip::tcp::endpoint peer);
	void connection_in_progress(connection::ptr_t con) { connecting_peers_.push_back(con); }
	void register_connection(connection::ptr_t con);

	// protocols
	template <typename P>
	P& protocol()
	{
		return *boost::static_pointer_cast<P>( protocol_handlers_.find(P::protocol_id)->second );
	}
	network_protocol& get_protocol(packet::const_ptr_t pkt) { return get_protocol(pkt->protocol()); }
	network_protocol& get_protocol(protocol_id pid) { return *protocol_handlers_.find(pid)->second; }

	bool register_protocol_handler(protocol_id id, network_protocol::ptr_t proto)
	{
		supported_protocols_.push_back(id);
		return protocol_handlers_.insert(std::make_pair(id, proto)).second;
	}

	// successors
	ip::tcp::endpoint successor_endpoint(const network_key& key);
	ip::tcp::endpoint predecessor_endpoint(const network_key& key);
	network_key self_predecessor() const;
	network_key self_sucessor() const;
	int closer_peers(const network_key& key) const;

	// requests
	connection::ptr_t local_request(packet::ptr_t pkt);
	connection::ptr_t local_request(packet::ptr_t pkt, const network_key& inner_id);
	connection::ptr_t dispatch(packet::ptr_t pkt);
	connection::ptr_t dispatch(packet::ptr_t pkt, const network_key& inner_id, bool local_request = false);

	// content carrying message handling
	// incoming_* is to notify the local node that a message header has just been received
	// the node is expected to return a buffer into which the payload data will be written
	void incoming_packet(connection::ptr_t con, packet::ptr_t pkt, std::size_t payload_size);
	void packet_received(connection::ptr_t con, packet::ptr_t pkt);

	// failures
	void send_failure(connection::ptr_t con);
	void receive_failure(connection::ptr_t con) { disconnect_peer(con); }

	// cache policy
	hunk_descriptor_t cache_local_request(protocol_id pid, content_identifier id, std::size_t size);
	hunk_descriptor_t cache_store(protocol_id pid, content_identifier id, std::size_t size);
	hunk_descriptor_t cache_remote_request(protocol_id pid, content_identifier id, std::size_t size, boost::posix_time::time_duration request_delta);
	hunk_descriptor_t load_existing_hunk(protocol_id pid, content_identifier id, std::size_t size);
	hunk_descriptor_t not_a_hunk()          { return stored_hunks_.end(); }
	void erase_hunk(hunk_descriptor_t desc) { stored_hunks_.erase(desc); }

	// credit accounting
	void sent_content(const network_key& id, std::size_t bytes) { traffic_stats_.sent_content(id, bytes); }
	const known_peers& get_known_peers() const                  { return traffic_stats_; }

	rolling_stats sources_per_hunk;

	boost::asio::ssl::context context;

private:
	void bootstrap();
	void recompute_identity();
	void snoop(packet::ptr_t pkt);
	void disconnect_peer(connection::ptr_t con);

	void peer_set_changed();
	void remove_peer(std::vector<connection::ptr_t>::iterator peer);

	network_protocol::ptr_t validate_protocol(protocol_id protocol);

	bool try_prune_cache(std::size_t size, int closer_peers, boost::posix_time::time_duration age);

	// successors
	std::vector<connection::ptr_t>::iterator
	sucessor(const network_key& key,
	         const network_key& inner_id,
	         content_size_t content_size = 0);

	template <network_key dist_fn(const network_key& src, const network_key& dest)>
	std::vector<connection::ptr_t>::iterator
	best_peer(const network_key& outer_id,
	          const network_key& inner_id,
	          const network_key& key,
	          content_size_t content_size = 0);

	// doesn't take ancillary factors into account, it will always return the closest node in terms of key distance
	template <network_key dist_fn(const network_key& src, const network_key& dest)>
	std::vector<connection::ptr_t>::iterator
	closest_peer(const network_key& outer_id,
	             const network_key& inner_id,
	             const network_key& key,
	             content_size_t content_size = 0);

	template <network_key dist_fn(const network_key& src, const network_key& dest)>
	std::vector<connection::ptr_t>::const_iterator
	closest_peer(const network_key& outer_id,
	             const network_key& inner_id,
	             const network_key& key,
	             content_size_t content_size = 0) const;

	template <network_key dist_fn(const network_key& src, const network_key& dest)>
	int closer_peers(const network_key& src, const network_key& dest) const;

	bool running_;
	client_config&           config_;
	ip::tcp::acceptor        acceptor_;
	boost::posix_time::ptime created_;

	std::multiset<ip::address> reported_addresses_;
	ip::tcp::endpoint          public_endpoint_;
	network_key                identity_;

	std::vector<connection::ptr_t> ib_peers_;
	std::vector<connection::ptr_t> connecting_peers_;
	std::vector<connection::ptr_t> disconnecting_peers_;

	typedef std::map<protocol_id, network_protocol::ptr_t> protocol_handlers_t;

	protocol_handlers_t protocol_handlers_;
	std::vector<protocol_id> supported_protocols_;

	std::size_t min_oob_threshold_, max_oob_threshold_, avg_oob_threshold_;

	network_key replication_min_dist_;

	// cache policy
	stored_hunks_t stored_hunks_;
	boost::uint64_t stored_size_;

	author client_key_;

	known_peers traffic_stats_;
};

#endif