Merge branch 'master' into mempool

Cargo.lock

@@ -63,6 +63,7 @@ name = "chain"
 version = "0.1.0"
 dependencies = [
  "bitcrypto 0.1.0",
+ "heapsize 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
  "primitives 0.1.0",
  "rustc-serialize 0.3.19 (registry+https://github.com/rust-lang/crates.io-index)",
  "serialization 0.1.0",
@@ -161,6 +162,14 @@ dependencies = [
  "rayon 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "heapsize"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "kernel32-sys 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "kernel32-sys"
 version = "0.2.2"
@@ -219,6 +228,7 @@ name = "miner"
 version = "0.1.0"
 dependencies = [
  "chain 0.1.0",
+ "heapsize 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
  "primitives 0.1.0",
  "serialization 0.1.0",
 ]
@@ -320,7 +330,6 @@ dependencies = [
  "parking_lot 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
  "primitives 0.1.0",
  "rand 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)",
- "serialization 0.1.0",
  "time 0.1.35 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-core 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -350,6 +359,7 @@ dependencies = [
 name = "primitives"
 version = "0.1.0"
 dependencies = [
+ "heapsize 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustc-serialize 0.3.19 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
@@ -550,6 +560,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 "checksum futures 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "0bd34f72c0fffc9d2f6c570fd392bf99b9c5cd1481d79809e1cc2320befc0af0"
 "checksum futures-cpupool 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "bb982bb25cd8fa5da6a8eb3a460354c984ff1113da82bcb4f0b0862b5795db82"
 "checksum gcc 0.3.35 (registry+https://github.com/rust-lang/crates.io-index)" = "91ecd03771effb0c968fd6950b37e89476a578aaf1c70297d8e92b6516ec3312"
+"checksum heapsize 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)" = "8c80e194758495a9109566134dc06e42ea0423987d6ceca016edaa90381b3549"
 "checksum kernel32-sys 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7507624b29483431c0ba2d82aece8ca6cdba9382bff4ddd0f7490560c056098d"
 "checksum lazy_static 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "49247ec2a285bb3dcb23cbd9c35193c025e7251bfce77c1d5da97e6362dffe7f"
 "checksum lazycell 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)" = "ce12306c4739d86ee97c23139f3a34ddf0387bbf181bc7929d287025a8c3ef6b"

chain/Cargo.toml

@@ -5,6 +5,7 @@ authors = ["debris <marek.kotewicz@gmail.com>"]
 
 [dependencies]
 rustc-serialize = "0.3"
+heapsize = "0.3"
 bitcrypto = { path = "../crypto" }
 primitives = { path = "../primitives" }
 serialization = { path = "../serialization" }

chain/src/lib.rs

@@ -1,4 +1,5 @@
 extern crate rustc_serialize;
+extern crate heapsize;
 extern crate primitives;
 extern crate bitcrypto as crypto;
 extern crate serialization as ser;

chain/src/transaction.rs

@@ -2,6 +2,7 @@
 //! Bitcoin trainsaction.
 //! https://en.bitcoin.it/wiki/Protocol_documentation#tx
 
+use heapsize::HeapSizeOf;
 use hex::FromHex;
 use bytes::Bytes;
 use ser::{
@@ -105,6 +106,12 @@ impl Deserializable for TransactionInput {
 	}
 }
 
+impl HeapSizeOf for TransactionInput {
+	fn heap_size_of_children(&self) -> usize {
+		self.script_sig.heap_size_of_children()
+	}
+}
+
 impl TransactionInput {
 	pub fn previous_output(&self) -> &OutPoint {
 		&self.previous_output
@@ -159,6 +166,12 @@ impl Default for TransactionOutput {
 	}
 }
 
+impl HeapSizeOf for TransactionOutput {
+	fn heap_size_of_children(&self) -> usize {
+		self.script_pubkey.heap_size_of_children()
+	}
+}
+
 impl TransactionOutput {
 	pub fn value(&self) -> u64 {
 		self.value
@@ -214,6 +227,12 @@ impl From<&'static str> for Transaction {
 	}
 }
 
+impl HeapSizeOf for Transaction {
+	fn heap_size_of_children(&self) -> usize {
+		self.inputs.heap_size_of_children() + self.outputs.heap_size_of_children()
+	}
+}
+
 impl Transaction {
 	pub fn hash(&self) -> H256 {
 		dhash256(&serialize(self))

message/src/common/address.rs

@@ -3,11 +3,11 @@ use ser::{
 	Stream, Serializable,
 	Reader, Deserializable, Error as ReaderError, deserialize,
 };
-use common::{Port, IpAddress, ServiceFlags};
+use common::{Port, IpAddress, Services};
 
 #[derive(Debug, PartialEq)]
 pub struct NetAddress {
-	pub services: ServiceFlags,
+	pub services: Services,
 	pub address: IpAddress,
 	pub port: Port,
 }
@@ -42,7 +42,7 @@ impl From<&'static str> for NetAddress {
 #[cfg(test)]
 mod tests {
 	use ser::{serialize, deserialize};
-	use common::ServiceFlags;
+	use common::Services;
 	use super::NetAddress;
 
 	#[test]
@@ -54,7 +54,7 @@ mod tests {
 		].into();
 
 		let address = NetAddress {
-			services: ServiceFlags::default().with_network(true),
+			services: Services::default().with_network(true),
 			address: "::ffff:a00:1".into(),
 			port: 8333.into(),
 		};
@@ -71,7 +71,7 @@ mod tests {
 		];
 
 		let expected = NetAddress {
-			services: ServiceFlags::default().with_network(true),
+			services: Services::default().with_network(true),
 			address: "::ffff:a00:1".into(),
 			port: 8333.into(),
 		};
@@ -82,7 +82,7 @@ mod tests {
 	#[test]
 	fn test_net_address_from_static_str() {
 		let expected = NetAddress {
-			services: ServiceFlags::default().with_network(true),
+			services: Services::default().with_network(true),
 			address: "::ffff:a00:1".into(),
 			port: 8333.into(),
 

message/src/common/mod.rs

@@ -20,4 +20,4 @@ pub use self::ip::IpAddress;
 pub use self::magic::Magic;
 pub use self::port::Port;
 pub use self::prefilled_transaction::PrefilledTransaction;
-pub use self::service::ServiceFlags;
+pub use self::service::Services;

message/src/common/service.rs

@@ -3,23 +3,22 @@ use ser::{
 	Deserializable, Reader, Error as ReaderError
 };
 
-#[derive(Debug, Default, PartialEq, Clone, Copy)]
-pub struct ServiceFlags(u64);
+#[derive(Debug, Default, PartialEq, Eq, Clone, Copy)]
+pub struct Services(u64);
 
-impl From<ServiceFlags> for u64 {
-	fn from(s: ServiceFlags) -> Self {
+impl From<Services> for u64 {
+	fn from(s: Services) -> Self {
 		s.0
 	}
 }
 
-impl From<u64> for ServiceFlags {
+impl From<u64> for Services {
 	fn from(v: u64) -> Self {
-		ServiceFlags(v)
+		Services(v)
 	}
 }
 
-
-impl ServiceFlags {
+impl Services {
 	pub fn network(&self) -> bool {
 		self.bit_at(0)
 	}
@@ -63,7 +62,11 @@ impl ServiceFlags {
 	pub fn with_xthin(mut self, v: bool) -> Self {
 		self.set_bit(4, v);
 		self
-		}
+	}
+
+	pub fn includes(&self, other: &Self) -> bool {
+		self.0 & other.0 == other.0
+	}
 
 	fn set_bit(&mut self, bit: usize, bit_value: bool) {
 		if bit_value {
@@ -78,14 +81,35 @@ impl ServiceFlags {
 	}
 }
 
-impl Serializable for ServiceFlags {
+impl Serializable for Services {
 	fn serialize(&self, stream: &mut Stream) {
 		stream.append(&self.0);
 	}
 }
 
-impl Deserializable for ServiceFlags {
+impl Deserializable for Services {
 	fn deserialize(reader: &mut Reader) -> Result<Self, ReaderError> where Self: Sized {
-		reader.read().map(ServiceFlags)
+		reader.read().map(Services)
+	}
+}
+
+#[cfg(test)]
+mod test {
+	use super::Services;
+
+	#[test]
+	fn test_serivces_includes() {
+		let s1 = Services::default()
+			.with_witness(true)
+			.with_xthin(true);
+		let s2 = Services::default()
+			.with_witness(true);
+
+		assert!(s1.witness());
+		assert!(s1.xthin());
+		assert!(s2.witness());
+		assert!(!s2.xthin());
+		assert!(s1.includes(&s2));
+		assert!(!s2.includes(&s1));
 	}
 }

message/src/serialization/stream.rs

@@ -29,10 +29,6 @@ impl PayloadStream {
 		t.serialize_payload(&mut self.stream, self.version)
 	}
 
-	pub fn raw_stream(&mut self) -> &mut Stream {
-		&mut self.stream
-	}
-
 	pub fn out(self) -> Bytes {
 		self.stream.out()
 	}

message/src/types/version.rs

@@ -3,7 +3,7 @@ use ser::{
 	Serializable, Stream,
 	Deserializable, Reader, Error as ReaderError,
 };
-use common::{NetAddress, ServiceFlags};
+use common::{NetAddress, Services};
 use {Payload, MessageResult};
 use serialization::deserialize_payload;
 
@@ -69,12 +69,20 @@ impl Version {
 			Version::V70001(ref s, _, _) => s.version,
 		}
 	}
+
+	pub fn services(&self) -> Services {
+		match *self {
+			Version::V0(ref s) => s.services,
+			Version::V106(ref s, _) => s.services,
+			Version::V70001(ref s, _, _) => s.services,
+		}
+	}
 }
 
 #[derive(Debug, PartialEq)]
 pub struct V0 {
 	pub version: u32,
-	pub services: ServiceFlags,
+	pub services: Services,
 	pub timestamp: i64,
 	pub receiver: NetAddress,
 }

miner/Cargo.toml

@@ -4,6 +4,7 @@ version = "0.1.0"
 authors = ["Ethcore <admin@ethcore.io>"]
 
 [dependencies]
+heapsize = "0.3"
 chain = { path = "../chain" }
 primitives = { path = "../primitives" }
 serialization = { path = "../serialization" }

miner/src/lib.rs

@@ -1,4 +1,5 @@
 extern crate chain;
+extern crate heapsize;
 extern crate primitives;
 extern crate serialization as ser;
 

miner/src/memory_pool.rs

@@ -12,6 +12,7 @@ use std::collections::HashMap;
 use std::collections::HashSet;
 use std::collections::BTreeSet;
 use ser::Serializable;
+use heapsize::HeapSizeOf;
 
 /// Transactions ordering strategy
 #[derive(Debug, Clone, Copy)]
@@ -27,9 +28,9 @@ pub enum OrderingStrategy {
 /// Information on current `MemoryPool` state
 #[derive(Debug)]
 pub struct Information {
-	/// The number of transactions currently in the `MemoryPool`
+	/// Number of transactions currently in the `MemoryPool`
 	pub transactions_count: usize,
-	/// The total number of bytes in the transactions in the `MemoryPool`
+	/// Total number of bytes occupied by transactions from the `MemoryPool`
 	pub transactions_size_in_bytes: usize,
 }
 
@@ -220,6 +221,12 @@ impl Ord for ByPackageScoreOrderedEntry {
 	}
 }
 
+impl HeapSizeOf for Entry {
+	fn heap_size_of_children(&self) -> usize {
+		self.transaction.heap_size_of_children() + self.ancestors.heap_size_of_children() 
+	}
+}
+
 impl Storage {
 	pub fn new() -> Self {
 		Storage {
@@ -507,6 +514,31 @@ impl OrderedReferenceStorage {
 	}
 }
 
+impl HeapSizeOf for Storage {
+	fn heap_size_of_children(&self) -> usize {
+		self.by_hash.heap_size_of_children() + self.references.heap_size_of_children()
+	}
+}
+
+impl HeapSizeOf for ReferenceStorage {
+	fn heap_size_of_children(&self) -> usize {
+		self.by_input.heap_size_of_children()
+			+ self.pending.heap_size_of_children()
+			+ self.ordered.heap_size_of_children()
+	}
+}
+
+impl HeapSizeOf for OrderedReferenceStorage {
+	fn heap_size_of_children(&self) -> usize {
+		// HeapSizeOf is not implemented for BTreeSet => rough estimation here
+		use std::mem::size_of;
+		let len = self.by_storage_index.len();
+		len * (size_of::<ByTimestampOrderedEntry>()
+			+ size_of::<ByTransactionScoreOrderedEntry>()
+			+ size_of::<ByPackageScoreOrderedEntry>())
+	}
+}
+
 impl MemoryPool {
 	/// Creates new memory pool
 	pub fn new() -> Self {
@@ -542,6 +574,7 @@ impl MemoryPool {
 
 	/// Reads hashes of up to n transactions from the `MemoryPool`, using selected strategy.
 	/// Ancestors are always returned before descendant transactions.
+	/// Use this function with care, only if really needed (heavy memory usage)
 	pub fn read_n_with_strategy(&mut self, n: usize, strategy: OrderingStrategy) -> Vec<H256> {
 		self.storage.read_n_with_strategy(n, strategy)
 	}
@@ -578,7 +611,7 @@ impl MemoryPool {
 	pub fn information(&self) -> Information {
 		Information {
 			transactions_count: self.storage.by_hash.len(),
-			transactions_size_in_bytes: self.storage.transactions_size_in_bytes
+			transactions_size_in_bytes: self.storage.transactions_size_in_bytes,
 		}
 	}
 
@@ -644,11 +677,18 @@ impl MemoryPool {
 	}
 }
 
+impl HeapSizeOf for MemoryPool {
+	fn heap_size_of_children(&self) -> usize {
+		self.storage.heap_size_of_children()
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use std::cmp::Ordering;
 	use hash::H256;
 	use chain::Transaction;
+	use heapsize::HeapSizeOf;
 	use super::{MemoryPool, OrderingStrategy};
 
 	// output_value = 898126612, size = 225, miner_score ~ 3991673.83
@@ -688,6 +728,21 @@ mod tests {
 		pool
 	}
 
+	#[test]
+	fn test_memory_pool_heap_size() {
+		let mut pool = MemoryPool::new();
+
+		let size1 = pool.heap_size_of_children();
+
+		pool.insert_verified(RAW_TRANSACTION1.into());
+		let size2 = pool.heap_size_of_children();
+		assert!(size2 > size1);
+
+		pool.insert_verified(RAW_TRANSACTION2.into());
+		let size3 = pool.heap_size_of_children();
+		assert!(size3 > size2);
+	}
+
 	#[test]
 	fn test_memory_pool_insert_same_transaction() {
 		let mut pool = MemoryPool::new();

p2p/Cargo.toml

@@ -13,5 +13,4 @@ rand = "0.3"
 
 primitives = { path = "../primitives" }
 bitcrypto = { path = "../crypto" }
-serialization = { path = "../serialization" }
 message = { path = "../message" }

p2p/src/config.rs

@@ -3,10 +3,20 @@ use net::Config as NetConfig;
 
 #[derive(Debug)]
 pub struct Config {
+	/// Number of threads used by p2p thread pool.
+	pub threads: usize,
+	/// Lowest supported protocol version.
+	pub protocol_minimum: u32,
+	/// Highest supported protocol version.
+	pub protocol_maximum: u32,
+	/// Number of inbound connections.
+	pub inbound_connections: usize,
+	/// Number of outbound connections.
+	pub outbound_connections: usize,
 	/// Configuration for every connection.
 	pub connection: NetConfig,
-	/// Connect to these nodes to retrieve peer addresses, and disconnect.
-	pub seednodes: Vec<IpAddr>,
 	/// Connect only ot these nodes.
-	pub limited_connect: Option<Vec<IpAddr>>,
+	pub peers: Vec<IpAddr>,
+	/// Connect to these nodes to retrieve peer addresses, and disconnect.
+	pub seeds: Vec<IpAddr>,
 }

p2p/src/io/sharedtcpstream.rs

@@ -1,4 +1,5 @@
 use std::sync::Arc;
+use std::net::Shutdown;
 use std::io::{Read, Write, Error};
 use tokio_core::net::TcpStream;
 
@@ -12,6 +13,11 @@ impl SharedTcpStream {
 			io: a,
 		}
 	}
+
+	pub fn shutdown(&self) {
+		// error is irrelevant here, the connection is dropped anyway
+		let _ = self.io.shutdown(Shutdown::Both);
+	}
 }
 
 impl From<TcpStream> for SharedTcpStream {

p2p/src/lib.rs

@@ -9,10 +9,11 @@ extern crate parking_lot;
 extern crate bitcrypto as crypto;
 extern crate message;
 extern crate primitives;
-extern crate serialization as ser;
 
 pub mod io;
 pub mod net;
+pub mod protocol;
+pub mod session;
 pub mod util;
 mod config;
 mod event_loop;
@@ -26,6 +27,5 @@ pub use primitives::{hash, bytes};
 pub use config::Config;
 pub use event_loop::{event_loop, forever};
 pub use p2p::P2P;
-
-pub type PeerId = usize;
+pub use util::{PeerId, PeerInfo};
 

p2p/src/net/channel.rs

@@ -3,33 +3,53 @@ use futures::Poll;
 use futures::stream::Stream;
 use parking_lot::Mutex;
 use bytes::Bytes;
-use message::{MessageResult, Payload, Command};
+use message::{MessageResult, Payload, Command, Magic, Message};
 use net::Connection;
-use io::{read_message_stream, ReadMessageStream, SharedTcpStream, WriteMessage};
+use io::{read_message_stream, ReadMessageStream, SharedTcpStream, WriteMessage, write_message};
+use {PeerId, PeerInfo};
 
 pub struct Channel {
-	connection: Connection,
-	message_stream: Mutex<ReadMessageStream<SharedTcpStream>>,
+	version: u32,
+	magic: Magic,
+	peer_info: PeerInfo,
+	write_stream: SharedTcpStream,
+	read_stream: Mutex<ReadMessageStream<SharedTcpStream>>,
 }
 
 impl Channel {
-	pub fn new(connection: Connection) -> Self {
+	pub fn new(connection: Connection, peer_id: PeerId) -> Self {
 		let stream = read_message_stream(connection.stream.clone(), connection.magic);
 		Channel {
-			connection: connection,
-			message_stream: Mutex::new(stream),
+			version: connection.version,
+			magic: connection.magic,
+			peer_info: PeerInfo {
+				address: connection.address,
+				id: peer_id,
+			},
+			write_stream: connection.stream,
+			read_stream: Mutex::new(stream),
 		}
 	}
 
 	pub fn write_message<T>(&self, payload: &T) -> WriteMessage<T, SharedTcpStream> where T: Payload {
-		self.connection.write_message(payload)
+		// TODO: some tracing here
+		let message = Message::new(self.magic, self.version, payload).expect("failed to create outgoing message");
+		write_message(self.write_stream.clone(), message)
 	}
 
 	pub fn poll_message(&self) -> Poll<Option<(MessageResult<(Command, Bytes)>)>, io::Error> {
-		self.message_stream.lock().poll()
+		self.read_stream.lock().poll()
+	}
+
+	pub fn shutdown(&self) {
+		self.write_stream.shutdown();
 	}
 
 	pub fn version(&self) -> u32 {
-		self.connection.version
+		self.version
+	}
+
+	pub fn peer_info(&self) -> PeerInfo {
+		self.peer_info
 	}
 }

p2p/src/net/config.rs

@@ -1,5 +1,5 @@
 use std::net::SocketAddr;
-use message::common::{Magic, ServiceFlags, NetAddress};
+use message::common::{Magic, Services, NetAddress};
 use message::types::version::{Version, V0, V106, V70001};
 use util::time::{Time, RealTime};
 use util::nonce::{NonceGenerator, RandomNonce};
@@ -9,7 +9,7 @@ use VERSION;
 pub struct Config {
 	pub magic: Magic,
 	pub local_address: SocketAddr,
-	pub services: ServiceFlags,
+	pub services: Services,
 	pub user_agent: String,
 	pub start_height: i32,
 	pub relay: bool,

p2p/src/net/connect.rs

@@ -57,6 +57,7 @@ impl Future for Connect {
 					stream: stream.into(),
 					version: result.negotiated_version,
 					magic: self.magic,
+					services: result.version.services(),
 					address: self.address,
 				};
 				(ConnectState::Connected, Async::Ready(Ok(connection)))

p2p/src/net/connection.rs

@@ -1,23 +1,12 @@
 use std::net;
-use message::{Message, Payload, Magic};
-use io::{write_message, WriteMessage, SharedTcpStream};
+use message::Magic;
+use message::common::Services;
+use io::SharedTcpStream;
 
 pub struct Connection {
 	pub stream: SharedTcpStream,
 	pub version: u32,
 	pub magic: Magic,
+	pub services: Services,
 	pub address: net::SocketAddr,
 }
-
-impl Connection {
-	pub fn write_message<T>(&self, payload: &T) -> WriteMessage<T, SharedTcpStream> where T: Payload {
-		let message = match Message::new(self.magic, self.version, payload) {
-			Ok(message) => message,
-			Err(_err) => {
-				// trace here! outgoing messages should always be written properly
-				panic!();
-			}
-		};
-		write_message(self.stream.clone(), message)
-	}
-}

p2p/src/net/connections.rs

@@ -2,15 +2,13 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::collections::HashMap;
 use parking_lot::RwLock;
-use futures::{finished, Future};
-use futures_cpupool::CpuPool;
-use tokio_core::reactor::Handle;
-use message::Payload;
 use net::{Connection, Channel};
 use PeerId;
 
 pub struct Connections {
+	/// Incremental peer counter.
 	peer_counter: AtomicUsize,
+	/// All open connections.
 	channels: RwLock<HashMap<PeerId, Arc<Channel>>>,
 }
 
@@ -22,28 +20,6 @@ impl Connections {
 		}
 	}
 
-	/// Broadcast messages to the network.
-	/// Returned future completes of first confirmed receive.
-	pub fn broadcast<T>(connections: &Arc<Connections>, handle: &Handle, pool: &CpuPool, payload: T) where T: Payload {
-		let channels = connections.channels();
-		for (id, channel) in channels.into_iter() {
-			let write = channel.write_message(&payload);
-			let cs = connections.clone();
-			let pool_work = pool.spawn(write).then(move |x| {
-				match x {
-					Ok(_) => {
-						// successfully sent message
-					},
-					Err(_) => {
-						cs.remove(id);
-					}
-				}
-				finished(())
-			});
-			handle.spawn(pool_work);
-		}
-	}
-
 	/// Returns safe (nonblocking) copy of channels.
 	pub fn channels(&self) -> HashMap<PeerId, Arc<Channel>> {
 		self.channels.read().clone()
@@ -57,11 +33,13 @@ impl Connections {
 	/// Stores new channel.
 	pub fn store(&self, connection: Connection) {
 		let id = self.peer_counter.fetch_add(1, Ordering::AcqRel);
-		self.channels.write().insert(id, Arc::new(Channel::new(connection)));
+		self.channels.write().insert(id, Arc::new(Channel::new(connection, id)));
 	}
 
 	/// Removes channel with given id.
 	pub fn remove(&self, id: PeerId) {
-		self.channels.write().remove(&id);
+		if let Some(channel) = self.channels.write().remove(&id) {
+			channel.shutdown();
+		}
 	}
 }

p2p/src/net/listen.rs

@@ -52,6 +52,7 @@ impl Future for AcceptConnection {
 			stream: stream.into(),
 			version: result.negotiated_version,
 			magic: self.magic,
+			services: result.version.services(),
 			address: self.address,
 		};
 		Ok(Ok(connection).into())

p2p/src/net/messages.rs

@@ -5,15 +5,28 @@ use futures::{Poll, Async};
 use futures::stream::Stream;
 use message::common::Command;
 use net::Connections;
-use PeerId;
+use PeerInfo;
 
-pub struct MessagesHandler {
+pub enum MessagePoll {
+	Ready {
+		command: Command,
+		payload: Bytes,
+		version: u32,
+		peer_info: PeerInfo,
+		errored_peers: Vec<PeerInfo>,
+	},
+	OnlyErrors {
+		errored_peers: Vec<PeerInfo>,
+	}
+}
+
+pub struct MessagePoller {
 	last_polled: usize,
 	connections: Weak<Connections>,
 }
 
 fn next_to_poll(channels: usize, last_polled: usize) -> usize {
-	// it's irrelevant if we sometimes poll the same peer
+	// it's irrelevant if we sometimes poll the same peer twice in a row
 	if channels > last_polled + 1 {
 		// let's poll the next peer
 		last_polled + 1
@@ -23,17 +36,17 @@ fn next_to_poll(channels: usize, last_polled: usize) -> usize {
 	}
 }
 
-impl MessagesHandler {
+impl MessagePoller {
 	pub fn new(connections: Weak<Connections>) -> Self {
-		MessagesHandler {
+		MessagePoller {
 			last_polled: usize::max_value(),
 			connections: connections,
 		}
 	}
 }
 
-impl Stream for MessagesHandler {
-	type Item = (Command, Bytes, u32, PeerId);
+impl Stream for MessagePoller {
+	type Item = MessagePoll;
 	type Error = io::Error;
 
 	fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
@@ -50,14 +63,15 @@ impl Stream for MessagesHandler {
 
 		let mut to_poll = next_to_poll(channels.len(), self.last_polled);
 		let mut result = None;
+		let mut errored_peers = Vec::new();
 
 		while result.is_none() && to_poll != self.last_polled {
-			let (id, channel) = channels.iter().nth(to_poll).expect("to_poll < channels.len()");
+			let (_, channel) = channels.iter().nth(to_poll).expect("to_poll < channels.len()");
 			let status = channel.poll_message();
 
 			match status {
-				Ok(Async::Ready(Some(Ok((command, message))))) => {
-					result = Some((command, message, channel.version(), *id));
+				Ok(Async::Ready(Some(Ok((command, payload))))) => {
+					result = Some((command, payload, channel.version(), channel.peer_info()));
 				},
 				Ok(Async::NotReady) => {
 					// no messages yet, try next channel
@@ -65,16 +79,32 @@ impl Stream for MessagesHandler {
 				},
 				_ => {
 					// channel has been closed or there was error
-					connections.remove(*id);
+					errored_peers.push(channel.peer_info());
 					to_poll = next_to_poll(channels.len(), to_poll);
 				},
 			}
 		}
 
 		self.last_polled = to_poll;
-		match result.is_some() {
-			true => Ok(Async::Ready(result)),
-			false => Ok(Async::NotReady),
+		match result {
+			Some((command, payload, version, info)) => {
+				let message_poll = MessagePoll::Ready {
+					command: command,
+					payload: payload,
+					version: version,
+					peer_info: info,
+					errored_peers: errored_peers,
+				};
+
+				Ok(Async::Ready(Some(message_poll)))
+			},
+			None if errored_peers.is_empty() => Ok(Async::NotReady),
+			_ => {
+				let message_poll = MessagePoll::OnlyErrors {
+					errored_peers: errored_peers,
+				};
+				Ok(Async::Ready(Some(message_poll)))
+			}
 		}
 	}
 }

p2p/src/net/mod.rs

@@ -5,13 +5,11 @@ mod connection;
 mod connections;
 mod messages;
 mod listen;
-mod subscriber;
 
 pub use self::channel::Channel;
 pub use self::config::Config;
 pub use self::connect::{Connect, connect};
 pub use self::connection::Connection;
 pub use self::connections::Connections;
-pub use self::messages::MessagesHandler;
+pub use self::messages::{MessagePoller, MessagePoll};
 pub use self::listen::{Listen, listen};
-pub use self::subscriber::Subscriber;

p2p/src/net/subscriber.rs

@@ -1,78 +0,0 @@
-use std::sync::mpsc::{Sender, Receiver, channel};
-use std::mem;
-use parking_lot::Mutex;
-use message::{Error, Payload, Command, deserialize_payload};
-use message::types::{Addr, GetAddr};
-use PeerId;
-
-struct Handler<S> {
-	sender: Mutex<Option<Sender<(S, PeerId)>>>,
-}
-
-impl<S> Default for Handler<S> {
-	fn default() -> Self {
-		Handler {
-			sender: Mutex::default(),
-		}
-	}
-}
-
-impl<S> Handler<S> where S: Payload {
-	fn command(&self) -> Command {
-		S::command().into()
-	}
-
-	fn handle(&self, payload: &[u8], version: u32, peerid: PeerId) -> Result<(), Error> {
-		let payload: S = try!(deserialize_payload(payload, version));
-		if let Some(sender) = self.sender() {
-			if let Err(_err) = sender.send((payload, peerid)) {
-				// TODO: unsubscribe channel?
-				// TODO: trace
-			}
-		}
-		Ok(())
-	}
-
-	fn sender(&self) -> Option<Sender<(S, PeerId)>> {
-		self.sender.lock().clone()
-	}
-
-	fn store(&self, sender: Sender<(S, PeerId)>) {
-		mem::replace(&mut *self.sender.lock(), Some(sender));
-	}
-}
-
-#[derive(Default)]
-pub struct Subscriber {
-	addr: Handler<Addr>,
-	getaddr: Handler<GetAddr>,
-}
-
-macro_rules! define_subscribe {
-	($name: ident, $result: ident, $sub: ident) => {
-		pub fn $name(&self) -> Receiver<($result, PeerId)> {
-			let (sender, receiver) = channel();
-			self.$sub.store(sender);
-			receiver
-		}
-	}
-}
-
-macro_rules! maybe_handle {
-	($command: expr, $sub: expr, $payload: expr, $version: expr, $peerid: expr) => {
-		if $command == $sub.command() {
-			return $sub.handle($payload, $version, $peerid);
-		}
-	}
-}
-
-impl Subscriber {
-	define_subscribe!(subscribe_addr, Addr, addr);
-	define_subscribe!(subscribe_getaddr, GetAddr, getaddr);
-
-	pub fn try_handle(&self, payload: &[u8], version: u32, command: Command, peerid: PeerId) -> Result<(), Error> {
-		maybe_handle!(command, self.addr, payload, version, peerid);
-		maybe_handle!(command, self.getaddr, payload, version, peerid);
-		Err(Error::InvalidCommand)
-	}
-}

p2p/src/p2p.rs

@@ -1,12 +1,14 @@
 use std::{io, net};
 use std::sync::Arc;
+use parking_lot::RwLock;
 use futures::{Future, finished};
 use futures::stream::Stream;
 use futures_cpupool::CpuPool;
 use tokio_core::reactor::Handle;
 use message::Payload;
-use net::{connect, listen, Connections, Subscriber, MessagesHandler};
-use Config;
+use net::{connect, listen, Connections, MessagePoller, MessagePoll, Channel};
+use util::NodeTable;
+use {Config, PeerId};
 
 pub struct P2P {
 	/// Global event loop handle.
@@ -17,40 +19,44 @@ pub struct P2P {
 	config: Config,
 	/// Connections.
 	connections: Arc<Connections>,
-	/// Message subscriber.
-	subscriber: Arc<Subscriber>,
+	/// Node Table.
+	node_table: Arc<RwLock<NodeTable>>,
 }
 
 impl P2P {
 	pub fn new(config: Config, handle: Handle) -> Self {
-		let pool = CpuPool::new(4);
+		let pool = CpuPool::new(config.threads);
 
 		P2P {
 			event_loop_handle: handle.clone(),
 			pool: pool.clone(),
 			config: config,
 			connections: Arc::new(Connections::new()),
-			subscriber: Arc::new(Subscriber::default()),
+			node_table: Arc::default(),
 		}
 	}
 
 	pub fn run(&self) -> Result<(), io::Error> {
-		for seednode in self.config.seednodes.iter() {
+		for seednode in self.config.peers.iter() {
 			self.connect(*seednode)
 		}
 
 		try!(self.listen());
-		self.handle_messages();
+		self.attach_protocols();
 		Ok(())
 	}
 
 	pub fn connect(&self, ip: net::IpAddr) {
 		let socket = net::SocketAddr::new(ip, self.config.connection.magic.port());
 		let connections = self.connections.clone();
+		let node_table  = self.node_table.clone();
 		let connection = connect(&socket, &self.event_loop_handle, &self.config.connection);
-		let pool_work = self.pool.spawn(connection).then(move |x| {
-			if let Ok(Ok(con)) = x {
+		let pool_work = self.pool.spawn(connection).then(move |result| {
+			if let Ok(Ok(con)) = result {
+				node_table.write().insert(con.address, con.services);
 				connections.store(con);
+			} else {
+				node_table.write().note_failure(&socket);
 			}
 			finished(())
 		});
@@ -60,8 +66,10 @@ impl P2P {
 	fn listen(&self) -> Result<(), io::Error> {
 		let listen = try!(listen(&self.event_loop_handle, self.config.connection.clone()));
 		let connections = self.connections.clone();
-		let server = listen.for_each(move |x| {
-			if let Ok(con) = x {
+		let node_table  = self.node_table.clone();
+		let server = listen.for_each(move |result| {
+			if let Ok(con) = result {
+				node_table.write().insert(con.address, con.services);
 				connections.store(con);
 			}
 			Ok(())
@@ -73,24 +81,70 @@ impl P2P {
 		Ok(())
 	}
 
-	fn handle_messages(&self) {
-		let incoming = MessagesHandler::new(Arc::downgrade(&self.connections));
-		let subscriber = self.subscriber.clone();
+	fn attach_protocols(&self) {
+		// TODO: here all network protocols will be attached
+
+		let poller = MessagePoller::new(Arc::downgrade(&self.connections));
 		let connections = self.connections.clone();
-		let incoming_future = incoming.for_each(move |result| {
-			let (command, payload, version, peerid) = result;
-			if let Err(_err) = subscriber.try_handle(&payload, version, command, peerid) {
-				connections.remove(peerid);
+		let node_table  = self.node_table.clone();
+		let polling = poller.for_each(move |result| {
+			match result {
+				MessagePoll::Ready { errored_peers, .. } => {
+					// TODO: handle new messasges here!
+
+					let mut node_table = node_table.write();
+					for peer in errored_peers.into_iter() {
+						node_table.note_failure(&peer.address);
+						connections.remove(peer.id);
+					}
+				},
+				MessagePoll::OnlyErrors { errored_peers } => {
+					let mut node_table = node_table.write();
+					for peer in errored_peers.into_iter() {
+						node_table.note_failure(&peer.address);
+						connections.remove(peer.id);
+					}
+				}
 			}
 			Ok(())
 		}).then(|_| {
 			finished(())
 		});
-		let pool_work = self.pool.spawn(incoming_future);
+		let pool_work = self.pool.spawn(polling);
 		self.event_loop_handle.spawn(pool_work);
 	}
 
 	pub fn broadcast<T>(&self, payload: T) where T: Payload {
-		Connections::broadcast(&self.connections, &self.event_loop_handle, &self.pool, payload)
+		let channels = self.connections.channels();
+		for (_id, channel) in channels.into_iter() {
+			self.send_to_channel(&payload, &channel);
+		}
+	}
+
+	pub fn send<T>(&self, payload: T, peer: PeerId) where T: Payload {
+		let channels = self.connections.channels();
+		if let Some(channel) = channels.get(&peer) {
+			self.send_to_channel(&payload, channel);
+		}
+	}
+
+	fn send_to_channel<T>(&self, payload: &T, channel: &Arc<Channel>) where T: Payload {
+		let connections = self.connections.clone();
+		let node_table  = self.node_table.clone();
+		let peer_info = channel.peer_info();
+		let write = channel.write_message(payload);
+		let pool_work = self.pool.spawn(write).then(move |result| {
+			match result {
+				Ok(_) => {
+					node_table.write().note_used(&peer_info.address);
+				},
+				Err(_err) => {
+					node_table.write().note_failure(&peer_info.address);
+					connections.remove(peer_info.id);
+				}
+			}
+			finished(())
+		});
+		self.event_loop_handle.spawn(pool_work);
 	}
 }

p2p/src/protocol/mod.rs

@@ -0,0 +1 @@
+mod ping;

p2p/src/session/mod.rs

@@ -0,0 +1,3 @@
+mod manual;
+mod normal;
+mod seednode;

p2p/src/util/mod.rs

@@ -1,2 +1,7 @@
 pub mod nonce;
 pub mod time;
+mod node_table;
+mod peer;
+
+pub use self::node_table::{NodeTable, Node};
+pub use self::peer::{PeerId, PeerInfo};

p2p/src/util/node_table.rs

@@ -0,0 +1,161 @@
+use std::collections::{HashMap, BTreeSet};
+use std::net::SocketAddr;
+use std::cmp::{PartialOrd, Ord, Ordering};
+use message::common::Services;
+use util::time::{Time, RealTime};
+
+#[derive(PartialEq, Eq, Clone)]
+pub struct Node {
+	/// Node address.
+	addr: SocketAddr,
+	/// Timestamp of last interaction with a node.
+	time: i64,
+	/// Services supported by the node.
+	services: Services,
+	/// Node failures counter.
+	failures: u32,
+}
+
+impl PartialOrd for Node {
+	fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+		if self.failures == other.failures {
+			self.time.partial_cmp(&other.time)
+		} else {
+			other.failures.partial_cmp(&self.failures)
+		}
+	}
+}
+
+impl Ord for Node {
+	fn cmp(&self, other: &Self) -> Ordering {
+		if self.failures == other.failures {
+			self.time.cmp(&other.time)
+		} else {
+			other.failures.cmp(&self.failures)
+		}
+	}
+}
+
+#[derive(Default)]
+pub struct NodeTable<T = RealTime> where T: Time {
+	/// Time source.
+	time: T,
+	/// Nodes by socket address.
+	by_addr: HashMap<SocketAddr, Node>,
+	/// Nodes sorted by score.
+	by_score: BTreeSet<Node>,
+}
+
+impl<T> NodeTable<T> where T: Time {
+	/// Inserts new address and services pair into NodeTable.
+	pub fn insert(&mut self, addr: SocketAddr, services: Services) {
+		let failures = self.by_addr.get(&addr).map_or(0, |ref node| node.failures);
+
+		let node = Node {
+			addr: addr,
+			time: self.time.get().sec,
+			services: services,
+			failures: failures,
+		};
+
+		self.by_addr.insert(addr, node.clone());
+		self.by_score.insert(node);
+	}
+
+	/// Returnes most reliable nodes with desired services.
+	pub fn nodes_with_services(&self, services: &Services, limit: usize) -> Vec<Node> {
+		self.by_score.iter()
+			.rev()
+			.filter(|s| s.services.includes(services))
+			.map(Clone::clone)
+			.take(limit)
+			.collect()
+	}
+
+	/// Marks address as recently used.
+	pub fn note_used(&mut self, addr: &SocketAddr) {
+		if let Some(ref mut node) = self.by_addr.get_mut(addr) {
+			assert!(self.by_score.remove(node));
+			node.time = self.time.get().sec;
+			self.by_score.insert(node.clone());
+		}
+	}
+
+	/// Notes failure.
+	pub fn note_failure(&mut self, addr: &SocketAddr) {
+		if let Some(ref mut node) = self.by_addr.get_mut(addr) {
+			assert!(self.by_score.remove(node));
+			node.failures += 1;
+			self.by_score.insert(node.clone());
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use std::net::SocketAddr;
+	use message::common::Services;
+	use util::time::IncrementalTime;
+	use super::NodeTable;
+
+	#[test]
+	fn test_node_table_insert() {
+		let s0: SocketAddr = "127.0.0.1:8000".parse().unwrap();
+		let s1: SocketAddr = "127.0.0.1:8001".parse().unwrap();
+		let s2: SocketAddr = "127.0.0.1:8002".parse().unwrap();
+		let mut table = NodeTable::<IncrementalTime>::default();
+		table.insert(s0, Services::default());
+		table.insert(s1, Services::default());
+		table.insert(s2, Services::default());
+		let nodes = table.nodes_with_services(&Services::default(), 2);
+		assert_eq!(nodes.len(), 2);
+		assert_eq!(nodes[0].addr, s2);
+		assert_eq!(nodes[0].time, 2);
+		assert_eq!(nodes[0].failures, 0);
+		assert_eq!(nodes[1].addr, s1);
+		assert_eq!(nodes[1].time, 1);
+		assert_eq!(nodes[1].failures, 0);
+	}
+
+	#[test]
+	fn test_node_table_note() {
+		let s0: SocketAddr = "127.0.0.1:8000".parse().unwrap();
+		let s1: SocketAddr = "127.0.0.1:8001".parse().unwrap();
+		let s2: SocketAddr = "127.0.0.1:8002".parse().unwrap();
+		let s3: SocketAddr = "127.0.0.1:8003".parse().unwrap();
+		let s4: SocketAddr = "127.0.0.1:8004".parse().unwrap();
+		let mut table = NodeTable::<IncrementalTime>::default();
+		table.insert(s0, Services::default());
+		table.insert(s1, Services::default());
+		table.insert(s2, Services::default());
+		table.insert(s3, Services::default());
+		table.insert(s4, Services::default());
+		table.note_used(&s2);
+		table.note_used(&s4);
+		table.note_used(&s1);
+		table.note_failure(&s2);
+		table.note_failure(&s3);
+		let nodes = table.nodes_with_services(&Services::default(), 10);
+		assert_eq!(nodes.len(), 5);
+
+		assert_eq!(nodes[0].addr, s1);
+		assert_eq!(nodes[0].time, 7);
+		assert_eq!(nodes[0].failures, 0);
+
+		assert_eq!(nodes[1].addr, s4);
+		assert_eq!(nodes[1].time, 6);
+		assert_eq!(nodes[1].failures, 0);
+
+		assert_eq!(nodes[2].addr, s0);
+		assert_eq!(nodes[2].time, 0);
+		assert_eq!(nodes[2].failures, 0);
+
+		assert_eq!(nodes[3].addr, s2);
+		assert_eq!(nodes[3].time, 5);
+		assert_eq!(nodes[3].failures, 1);
+
+		assert_eq!(nodes[4].addr, s3);
+		assert_eq!(nodes[4].time, 3);
+		assert_eq!(nodes[4].failures, 1);
+	}
+}

p2p/src/util/peer.rs

@@ -0,0 +1,10 @@
+use std::net::SocketAddr;
+
+pub type PeerId = usize;
+
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
+pub struct PeerInfo {
+	pub id: PeerId,
+	pub address: SocketAddr,
+}
+

p2p/src/util/time.rs

@@ -1,3 +1,4 @@
+use std::cell::Cell;
 use time;
 
 pub trait Time {
@@ -26,3 +27,17 @@ impl Time for StaticTime {
 		self.0
 	}
 }
+
+#[derive(Default)]
+pub struct IncrementalTime {
+	counter: Cell<i64>,
+}
+
+impl Time for IncrementalTime {
+	fn get(&self) -> time::Timespec {
+		let c = self.counter.get();
+		let result = time::Timespec::new(c, 0);
+		self.counter.set(c + 1);
+		result
+	}
+}

pbtc/main.rs

@@ -28,6 +28,11 @@ fn run() -> Result<(), String> {
 	let mut el = event_loop();
 
 	let p2p_cfg = p2p::Config {
+		threads: 4,
+		protocol_minimum: 70001,
+		protocol_maximum: 70017,
+		inbound_connections: 10,
+		outbound_connections: 10,
 		connection: net::Config {
 			magic: cfg.magic,
 			local_address: SocketAddr::new("127.0.0.1".parse().unwrap(), cfg.port),
@@ -36,8 +41,8 @@ fn run() -> Result<(), String> {
 			start_height: 0,
 			relay: false,
 		},
-		seednodes: cfg.seednode.map_or_else(|| vec![], |x| vec![x]),
-		limited_connect: cfg.connect.map_or(None, |x| Some(vec![x])),
+		peers: cfg.connect.map_or_else(|| vec![], |x| vec![x]),
+		seeds: cfg.seednode.map_or_else(|| vec![], |x| vec![x]),
 	};
 
 	let p2p = P2P::new(p2p_cfg, el.handle());

primitives/Cargo.toml

@@ -4,4 +4,5 @@ version = "0.1.0"
 authors = ["debris <marek.kotewicz@gmail.com>"]
 
 [dependencies]
+heapsize = "0.3"
 rustc-serialize = "0.3"

primitives/src/hash.rs

@@ -137,6 +137,8 @@ impl_hash!(H264, 33);
 impl_hash!(H512, 64);
 impl_hash!(H520, 65);
 
+known_heap_size!(0, H32, H48, H96, H160, H256, H264, H512, H520);
+
 impl H256 {
 	#[inline]
 	pub fn from_reversed_str(s: &'static str) -> Self {

primitives/src/lib.rs

@@ -1,4 +1,5 @@
 extern crate rustc_serialize;
+#[macro_use] extern crate heapsize;
 
 pub mod bytes;
 pub mod hash;

tools/graph.dot

@@ -107,7 +107,6 @@ digraph dependencies {
 	N6 -> N4[label="",style=dashed];
 	N6 -> N13[label="",style=dashed];
 	N6 -> N14[label="",style=dashed];
-	N6 -> N22[label="",style=dashed];
 	N6 -> N33[label="",style=dashed];
 	N6 -> N36[label="",style=dashed];
 	N6 -> N39[label="",style=dashed];