diff --git a/polkadot/cli/Cargo.toml b/polkadot/cli/Cargo.toml index 37d920c3c0..232c105f6f 100644 --- a/polkadot/cli/Cargo.toml +++ b/polkadot/cli/Cargo.toml @@ -23,4 +23,4 @@ substrate-rpc-servers = { path = "../../substrate/rpc-servers" } polkadot-primitives = { path = "../primitives" } polkadot-executor = { path = "../executor" } polkadot-runtime = { path = "../runtime" } -polkadot-keystore = { path = "../keystore" } +polkadot-service = { path = "../service" } diff --git a/polkadot/cli/src/error.rs b/polkadot/cli/src/error.rs index 6c9e22cd55..d7f6afca49 100644 --- a/polkadot/cli/src/error.rs +++ b/polkadot/cli/src/error.rs @@ -22,15 +22,11 @@ error_chain! { foreign_links { Io(::std::io::Error) #[doc="IO error"]; Cli(::clap::Error) #[doc="CLI error"]; + Service(::service::Error) #[doc="Polkadot service error"]; } links { Client(client::error::Error, client::error::ErrorKind) #[doc="Client error"]; } errors { - /// Key store errors - Keystore(e: ::keystore::Error) { - description("Keystore error"), - display("Keystore error: {:?}", e), - } } } diff --git a/polkadot/cli/src/lib.rs b/polkadot/cli/src/lib.rs index c5e34af10a..5c0492af01 100644 --- a/polkadot/cli/src/lib.rs +++ b/polkadot/cli/src/lib.rs @@ -30,10 +30,8 @@ extern crate substrate_rpc_servers as rpc; extern crate polkadot_primitives; extern crate polkadot_executor; extern crate polkadot_runtime; -extern crate polkadot_keystore as keystore; +extern crate polkadot_service as service; -#[macro_use] -extern crate hex_literal; #[macro_use] extern crate clap; #[macro_use] @@ -45,11 +43,6 @@ pub mod error; use std::path::{Path, PathBuf}; -use codec::Slicable; -use polkadot_runtime::genesismap::{additional_storage_with_genesis, GenesisConfig}; -use client::genesis; -use keystore::Store as Keystore; - /// Parse command line arguments and start the node. /// /// IANA unassigned port ranges that we could use: @@ -69,52 +62,33 @@ pub fn run(args: I) -> error::Result<()> where let log_pattern = matches.value_of("log").unwrap_or(""); init_logger(log_pattern); - // Create client - let executor = polkadot_executor::Executor::new(); - let mut storage = Default::default(); - let god_key = hex!["3d866ec8a9190c8343c2fc593d21d8a6d0c5c4763aaab2349de3a6111d64d124"]; + let mut config = service::Configuration::default(); - let genesis_config = GenesisConfig { - validators: vec![god_key.clone()], - authorities: vec![god_key.clone()], - balances: vec![(god_key.clone(), 1u64 << 63)].into_iter().collect(), - block_time: 5, // 5 second block time. - session_length: 720, // that's 1 hour per session. - sessions_per_era: 24, // 24 hours per era. - bonding_duration: 90, // 90 days per bond. - approval_ratio: 667, // 66.7% approvals required for legislation. - }; - - let prepare_genesis = || { - storage = genesis_config.genesis_map(); - let block = genesis::construct_genesis_block(&storage); - storage.extend(additional_storage_with_genesis(&block)); - (primitives::block::Header::decode(&mut block.header.encode().as_ref()).expect("to_vec() always gives a valid serialisation; qed"), storage.into_iter().collect()) - }; - - let keystore_path = matches.value_of("keystore") + config.keystore_path = matches.value_of("keystore") .map(|x| Path::new(x).to_owned()) - .unwrap_or_else(default_keystore_path); - - let _keystore = Keystore::open(keystore_path).map_err(::error::ErrorKind::Keystore)?; - let client = client::new_in_mem(executor, prepare_genesis)?; - - let address = "127.0.0.1:9933".parse().unwrap(); - let handler = rpc::rpc_handler(client); - let server = rpc::start_http(&address, handler)?; + .unwrap_or_else(default_keystore_path) + .to_string_lossy() + .into(); + let mut role = service::Role::FULL; if let Some(_) = matches.subcommand_matches("collator") { info!("Starting collator."); - server.wait(); - return Ok(()); + role = service::Role::COLLATOR; } - - if let Some(_) = matches.subcommand_matches("validator") { + else if let Some(_) = matches.subcommand_matches("validator") { info!("Starting validator."); - server.wait(); - return Ok(()); + role = service::Role::VALIDATOR; } + config.roles = role; + + let service = service::Service::new(config)?; + + let address = "127.0.0.1:9933".parse().unwrap(); + let handler = rpc::rpc_handler(service.client()); + let server = rpc::start_http(&address, handler)?; + + server.wait(); println!("No command given.\n"); let _ = clap::App::from_yaml(yaml).print_long_help(); diff --git a/polkadot/consensus/Cargo.toml b/polkadot/consensus/Cargo.toml index b7ccbc6534..df49b2a1cf 100644 --- a/polkadot/consensus/Cargo.toml +++ b/polkadot/consensus/Cargo.toml @@ -18,3 +18,8 @@ polkadot-transaction-pool = { path = "../transaction-pool" } substrate-bft = { path = "../../substrate/bft" } substrate-codec = { path = "../../substrate/codec" } substrate-primitives = { path = "../../substrate/primitives" } +substrate-network = { path = "../../substrate/network" } + +tokio-core = "0.1.12" +substrate-keyring = { path = "../../substrate/keyring" } +substrate-client = { path = "../../substrate/client" } diff --git a/polkadot/consensus/src/error.rs b/polkadot/consensus/src/error.rs index 4d1387d17e..38ba4ab607 100644 --- a/polkadot/consensus/src/error.rs +++ b/polkadot/consensus/src/error.rs @@ -48,6 +48,10 @@ error_chain! { ::MAX_TRANSACTIONS_SIZE, ::MAX_TRANSACTIONS_SIZE.saturating_sub(*size) ), } + Executor(e: ::futures::future::ExecuteErrorKind) { + description("Unable to dispatch agreement future"), + display("Unable to dispatch agreement future: {:?}", e), + } } } diff --git a/polkadot/consensus/src/lib.rs b/polkadot/consensus/src/lib.rs index a5c34e30ee..8ee93bafa8 100644 --- a/polkadot/consensus/src/lib.rs +++ b/polkadot/consensus/src/lib.rs @@ -41,10 +41,14 @@ extern crate polkadot_transaction_pool as transaction_pool; extern crate substrate_bft as bft; extern crate substrate_codec as codec; extern crate substrate_primitives as primitives; +extern crate substrate_network; + +extern crate tokio_core; +extern crate substrate_keyring; +extern crate substrate_client as client; #[macro_use] extern crate error_chain; - #[macro_use] extern crate log; @@ -67,8 +71,10 @@ use futures::future; use parking_lot::Mutex; pub use self::error::{ErrorKind, Error}; +pub use service::Service; mod error; +mod service; // block size limit. const MAX_TRANSACTIONS_SIZE: usize = 4 * 1024 * 1024; @@ -83,7 +89,7 @@ pub trait TableRouter { type FetchExtrinsic: IntoFuture; /// Note local candidate data. - fn local_candidate_data(&self, block_data: BlockData, extrinsic: Extrinsic); + fn local_candidate_data(&self, hash: Hash, block_data: BlockData, extrinsic: Extrinsic); /// Fetch block data for a specific candidate. fn fetch_block_data(&self, candidate: &CandidateReceipt) -> Self::FetchCandidate; diff --git a/polkadot/consensus/src/service.rs b/polkadot/consensus/src/service.rs new file mode 100644 index 0000000000..288f7a7d83 --- /dev/null +++ b/polkadot/consensus/src/service.rs @@ -0,0 +1,233 @@ +// Copyright 2017 Parity Technologies (UK) Ltd. +// This file is part of Polkadot. + +// Polkadot 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. + +// Polkadot 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 Polkadot. If not, see . + +//! Consensus service. + +/// Consensus service. A long runnung service that manages BFT agreement and parachain +/// candidate agreement over the network. + +use std::thread; +use std::sync::Arc; +use futures::{future, Future, Stream, Sink, Async, Canceled}; +use parking_lot::Mutex; +use substrate_network as net; +use tokio_core::reactor; +use client::BlockchainEvents; +use substrate_keyring::Keyring; +use primitives::{Hash, AuthorityId}; +use primitives::block::{Id as BlockId, HeaderHash, Header}; +use polkadot_primitives::parachain::{BlockData, Extrinsic, CandidateReceipt}; +use polkadot_api::PolkadotApi; +use bft::{self, BftService}; +use transaction_pool::TransactionPool; +use ed25519; +use super::{TableRouter, SharedTable, ProposerFactory}; +use error::Error; + +struct BftSink { + network: Arc, + _e: ::std::marker::PhantomData, +} + +fn process_message(msg: net::BftMessage, authorities: &[AuthorityId], parent_hash: HeaderHash) -> Result { + Ok(match msg { + net::BftMessage::Consensus(c) => bft::generic::Communication::Consensus(match c { + net::SignedConsensusMessage::Propose(proposal) => bft::generic::LocalizedMessage::Propose({ + let proposal = bft::generic::LocalizedProposal { + round_number: proposal.round_number as usize, + proposal: proposal.proposal, + digest: proposal.digest, + sender: proposal.sender, + digest_signature: ed25519::LocalizedSignature { + signature: proposal.digest_signature, + signer: ed25519::Public(proposal.sender), + }, + full_signature: ed25519::LocalizedSignature { + signature: proposal.full_signature, + signer: ed25519::Public(proposal.sender), + } + }; + bft::check_proposal(authorities, &parent_hash, &proposal)?; + proposal + }), + net::SignedConsensusMessage::Vote(vote) => bft::generic::LocalizedMessage::Vote({ + let vote = bft::generic::LocalizedVote { + sender: vote.sender, + signature: ed25519::LocalizedSignature { + signature: vote.signature, + signer: ed25519::Public(vote.sender), + }, + vote: match vote.vote { + net::ConsensusVote::Prepare(r, h) => bft::generic::Vote::Prepare(r as usize, h), + net::ConsensusVote::Commit(r, h) => bft::generic::Vote::Commit(r as usize, h), + net::ConsensusVote::AdvanceRound(r) => bft::generic::Vote::AdvanceRound(r as usize), + } + }; + bft::check_vote(authorities, &parent_hash, &vote)?; + vote + }), + }), + net::BftMessage::Auxiliary(a) => { + let justification = bft::UncheckedJustification::from(a); + // TODO: get proper error + let justification: Result<_, bft::Error> = bft::check_prepare_justification(authorities, parent_hash, justification) + .map_err(|_| bft::ErrorKind::InvalidJustification.into()); + bft::generic::Communication::Auxiliary(justification?) + }, + }) +} + +impl Sink for BftSink { + type SinkItem = bft::Communication; + // TODO: replace this with the ! type when that's stabilized + type SinkError = E; + + fn start_send(&mut self, message: bft::Communication) -> ::futures::StartSend { + let network_message = match message { + bft::generic::Communication::Consensus(c) => net::BftMessage::Consensus(match c { + bft::generic::LocalizedMessage::Propose(proposal) => net::SignedConsensusMessage::Propose(net::SignedConsensusProposal { + round_number: proposal.round_number as u32, + proposal: proposal.proposal, + digest: proposal.digest, + sender: proposal.sender, + digest_signature: proposal.digest_signature.signature, + full_signature: proposal.full_signature.signature, + }), + bft::generic::LocalizedMessage::Vote(vote) => net::SignedConsensusMessage::Vote(net::SignedConsensusVote { + sender: vote.sender, + signature: vote.signature.signature, + vote: match vote.vote { + bft::generic::Vote::Prepare(r, h) => net::ConsensusVote::Prepare(r as u32, h), + bft::generic::Vote::Commit(r, h) => net::ConsensusVote::Commit(r as u32, h), + bft::generic::Vote::AdvanceRound(r) => net::ConsensusVote::AdvanceRound(r as u32), + } + }), + }), + bft::generic::Communication::Auxiliary(justification) => net::BftMessage::Auxiliary(justification.uncheck().into()), + }; + self.network.send_bft_message(network_message); + Ok(::futures::AsyncSink::Ready) + } + + fn poll_complete(&mut self) -> ::futures::Poll<(), E> { + Ok(Async::Ready(())) + } +} + +/// Consensus service. Starts working when created. +pub struct Service { + thread: Option>, +} + +struct Network(Arc); + +impl Service { + /// Create and start a new instance. + pub fn new(client: Arc, network: Arc, transaction_pool: Arc>, best_header: &Header) -> Service + where C: BlockchainEvents + bft::BlockImport + bft::Authorities + PolkadotApi + Send + Sync + 'static + { + + let best_header = best_header.clone(); + let thread = thread::spawn(move || { + let mut core = reactor::Core::new().expect("tokio::Core could not be created"); + let key = Arc::new(Keyring::One.into()); + let factory = ProposerFactory { + client: client.clone(), + transaction_pool: transaction_pool.clone(), + network: Network(network.clone()), + }; + let bft_service = BftService::new(client.clone(), key, factory); + let build_bft = |header: &Header| -> Result<_, Error> { + let hash = header.hash(); + let authorities = client.authorities(&BlockId::Hash(hash))?; + let input = network.bft_messages() + .filter_map(move |message| { + process_message(message, &authorities, hash.clone()) + .map_err(|e| debug!("Message validation failed: {:?}", e)) + .ok() + }) + .map_err(|_| bft::InputStreamConcluded.into()); + let output = BftSink { network: network.clone(), _e: Default::default() }; + Ok(bft_service.build_upon(&header, input, output)?) + }; + // Kickstart BFT agreement on start. + if let Err(e) = build_bft(&best_header) + .map_err(|e| debug!("Error creating initial BFT agreement: {:?}", e)) + .and_then(|bft| core.run(bft)) + { + debug!("Error starting initial BFT agreement: {:?}", e); + } + let bft = client.import_notification_stream().and_then(|notification| { + build_bft(¬ification.header).map_err(|e| debug!("BFT agreement error: {:?}", e)) + }).for_each(|f| f); + if let Err(e) = core.run(bft) { + debug!("BFT event loop error {:?}", e); + } + }); + Service { + thread: Some(thread) + } + } +} + +impl Drop for Service { + fn drop(&mut self) { + if let Some(thread) = self.thread.take() { + thread.join().expect("The service thread has panicked"); + } + } +} + +impl super::Network for Network { + type TableRouter = Router; + fn table_router(&self, _table: Arc) -> Self::TableRouter { + Router { + network: self.0.clone() + } + } +} + +type FetchCandidateAdapter = future::Map) -> BlockData>; + +struct Router { + network: Arc, +} + +impl Router { + fn fetch_candidate_adapter(data: Vec) -> BlockData { + BlockData(data) + } +} + +impl TableRouter for Router { + type Error = Canceled; + type FetchCandidate = FetchCandidateAdapter; + type FetchExtrinsic = future::FutureResult; + + fn local_candidate_data(&self, hash: Hash, block_data: BlockData, _extrinsic: Extrinsic) { + let data = block_data.0; + self.network.set_local_candidate(Some((hash, data))) + } + + fn fetch_block_data(&self, candidate: &CandidateReceipt) -> Self::FetchCandidate { + let hash = candidate.hash(); + self.network.fetch_candidate(&hash).map(Self::fetch_candidate_adapter) + } + + fn fetch_extrinsic_data(&self, _candidate: &CandidateReceipt) -> Self::FetchExtrinsic { + future::ok(Extrinsic) + } +} diff --git a/polkadot/service/Cargo.toml b/polkadot/service/Cargo.toml new file mode 100644 index 0000000000..a5361e59e2 --- /dev/null +++ b/polkadot/service/Cargo.toml @@ -0,0 +1,26 @@ +[package] +name = "polkadot-service" +version = "0.1.0" +authors = ["Parity Technologies "] + +[dependencies] +futures = "0.1.17" +parking_lot = "0.4" +tokio-timer = "0.1.2" +ed25519 = { path = "../../substrate/ed25519" } +error-chain = "0.11" +log = "0.4" +tokio-core = "0.1.12" +polkadot-primitives = { path = "../primitives" } +polkadot-runtime = { path = "../runtime" } +polkadot-consensus = { path = "../consensus" } +polkadot-executor = { path = "../executor" } +polkadot-api = { path = "../api" } +polkadot-transaction-pool = { path = "../transaction-pool" } +polkadot-keystore = { path = "../keystore" } +substrate-primitives = { path = "../../substrate/primitives" } +substrate-network = { path = "../../substrate/network" } +substrate-client = { path = "../../substrate/client" } +substrate-keyring = { path = "../../substrate/keyring" } +substrate-codec = { path = "../../substrate/codec" } +substrate-executor = { path = "../../substrate/executor" } diff --git a/polkadot/service/src/config.rs b/polkadot/service/src/config.rs new file mode 100644 index 0000000000..6b87d0e49b --- /dev/null +++ b/polkadot/service/src/config.rs @@ -0,0 +1,46 @@ +// Copyright 2017 Parity Technologies (UK) Ltd. +// This file is part of Polkadot. + +// Polkadot 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. + +// Polkadot 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 Polkadot. If not, see .? + +//! Service configuration. + +use transaction_pool; +pub use network::Role; +pub use network::NetworkConfiguration; + +/// Service configuration. +pub struct Configuration { + /// Node roles. + pub roles: Role, + /// Transaction pool configuration. + pub transaction_pool: transaction_pool::Options, + /// Network configuration. + pub network: NetworkConfiguration, + /// Path to key files. + pub keystore_path: String, + // TODO: add more network, client, tx pool configuration options +} + +impl Default for Configuration { + fn default() -> Configuration { + Configuration { + roles: Role::FULL, + transaction_pool: Default::default(), + network: Default::default(), + keystore_path: Default::default(), + } + } +} + diff --git a/polkadot/service/src/error.rs b/polkadot/service/src/error.rs new file mode 100644 index 0000000000..58bd8d633b --- /dev/null +++ b/polkadot/service/src/error.rs @@ -0,0 +1,35 @@ +// Copyright 2017 Parity Technologies (UK) Ltd. +// This file is part of Polkadot. + +// Polkadot 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. + +// Polkadot 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 Polkadot. If not, see . + +//! Errors that can occur during the service operation. + +use client; +use network; + +error_chain! { + links { + Client(client::error::Error, client::error::ErrorKind) #[doc="Client error"]; + Network(network::error::Error, network::error::ErrorKind) #[doc="Network error"]; + } + + errors { + /// Key store errors + Keystore(e: ::keystore::Error) { + description("Keystore error"), + display("Keystore error: {:?}", e), + } + } +} diff --git a/polkadot/service/src/lib.rs b/polkadot/service/src/lib.rs new file mode 100644 index 0000000000..6bf972878d --- /dev/null +++ b/polkadot/service/src/lib.rs @@ -0,0 +1,216 @@ +// Copyright 2017 Parity Technologies (UK) Ltd. +// This file is part of Polkadot. + +// Polkadot 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. + +// Polkadot 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 Polkadot. If not, see . + +//! Polkadot service. Starts a thread that spins the network, the client and the transaction pool. +//! Manages communication between them. + +extern crate futures; +extern crate ed25519; +extern crate parking_lot; +extern crate tokio_timer; +extern crate polkadot_primitives; +extern crate polkadot_runtime; +extern crate polkadot_executor; +extern crate polkadot_api; +extern crate polkadot_consensus as consensus; +extern crate polkadot_transaction_pool as transaction_pool; +extern crate polkadot_keystore as keystore; +extern crate substrate_primitives as primitives; +extern crate substrate_network as network; +extern crate substrate_codec as codec; +extern crate substrate_executor; + +extern crate tokio_core; +extern crate substrate_keyring; +extern crate substrate_client as client; + +#[macro_use] +extern crate error_chain; +#[macro_use] +extern crate log; + +mod error; +mod config; + +use std::sync::Arc; +use std::thread; +use futures::prelude::*; +use parking_lot::Mutex; +use tokio_core::reactor::Core; +use codec::Slicable; +use primitives::block::{Id as BlockId, TransactionHash}; +use transaction_pool::TransactionPool; +use substrate_keyring::Keyring; +use substrate_executor::NativeExecutor; +use polkadot_executor::Executor as LocalDispatch; +use polkadot_primitives::AccountId; +use keystore::Store as Keystore; +use polkadot_api::PolkadotApi; +use polkadot_runtime::genesismap::{additional_storage_with_genesis, GenesisConfig}; +use client::{genesis, BlockchainEvents}; +use client::in_mem::Backend as InMemory; +use network::ManageNetwork; + +pub use self::error::{ErrorKind, Error}; +pub use config::{Configuration, Role}; + +type Client = client::Client>; + + +/// Polkadot service. +pub struct Service { + thread: Option>, + client: Arc, + network: Arc, + _consensus: Option, +} + +struct TransactionPoolAdapter { + pool: Arc>, + client: Arc, +} + +impl network::TransactionPool for TransactionPoolAdapter { + fn transactions(&self) -> Vec<(TransactionHash, Vec)> { + let best_block = match self.client.info() { + Ok(info) => info.chain.best_hash, + Err(e) => { + debug!("Error getting best block: {:?}", e); + return Vec::new(); + } + }; + let id = self.client.check_id(BlockId::Hash(best_block)).expect("Best block is always valid; qed."); + let ready = transaction_pool::Ready::create(id, &*self.client); + self.pool.lock().pending(ready).map(|t| { + let hash = ::primitives::Hash::from(&t.hash()[..]); + let tx = codec::Slicable::encode(t.as_transaction()); + (hash, tx) + }).collect() + } + + fn import(&self, transaction: &[u8]) -> Option { + if let Some(tx) = codec::Slicable::decode(&mut &transaction[..]) { + match self.pool.lock().import(tx) { + Ok(t) => Some(t.hash()[..].into()), + Err(e) => match *e.kind() { + transaction_pool::ErrorKind::AlreadyImported(hash) => Some(hash[..].into()), + _ => { + debug!("Error adding transaction to the pool: {:?}", e); + None + }, + } + } + } else { + debug!("Error decoding transaction"); + None + } + } +} + +impl Service { + /// Creates and register protocol with the network service + pub fn new(config: Configuration) -> Result { + // Create client + let executor = polkadot_executor::Executor::new(); + let mut storage = Default::default(); + let key: AccountId = Keyring::One.into(); + + let genesis_config = GenesisConfig { + validators: vec![key.clone()], + authorities: vec![key.clone()], + balances: vec![(Keyring::One.into(), 1u64 << 63), (Keyring::Two.into(), 1u64 << 63)].into_iter().collect(), + block_time: 5, // 5 second block time. + session_length: 720, // that's 1 hour per session. + sessions_per_era: 24, // 24 hours per era. + bonding_duration: 90, // 90 days per bond. + approval_ratio: 667, // 66.7% approvals required for legislation. + }; + let prepare_genesis = || { + storage = genesis_config.genesis_map(); + let block = genesis::construct_genesis_block(&storage); + storage.extend(additional_storage_with_genesis(&block)); + (primitives::block::Header::decode(&mut block.header.encode().as_ref()).expect("to_vec() always gives a valid serialisation; qed"), storage.into_iter().collect()) + }; + + let _keystore = Keystore::open(config.keystore_path.into()).map_err(::error::ErrorKind::Keystore)?; + let client = Arc::new(client::new_in_mem(executor, prepare_genesis)?); + let best_header = client.header(&BlockId::Hash(client.info()?.chain.best_hash))?.expect("Best header always exists; qed"); + info!("Starting Polkadot. Best block is #{}", best_header.number); + let transaction_pool = Arc::new(Mutex::new(TransactionPool::new(config.transaction_pool))); + let transaction_pool_adapter = Arc::new(TransactionPoolAdapter { + pool: transaction_pool.clone(), + client: client.clone(), + }); + let network_params = network::Params { + config: network::ProtocolConfig { + roles: config.roles, + }, + network_config: config.network, + chain: client.clone(), + transaction_pool: transaction_pool_adapter, + }; + let network = network::Service::new(network_params)?; + + // Spin consensus service if configured + let consensus_service = if config.roles & Role::VALIDATOR == Role::VALIDATOR { + Some(consensus::Service::new(client.clone(), network.clone(), transaction_pool, &best_header)) + } else { + None + }; + + let thread_client = client.clone(); + let thread_network = network.clone(); + let thread = thread::spawn(move || { + thread_network.start_network(); + let mut core = Core::new().expect("tokio::Core could not be created"); + let events = thread_client.import_notification_stream().for_each(|notification| { + thread_network.on_block_imported(¬ification.header); + Ok(()) + }); + if let Err(e) = core.run(events) { + debug!("Polkadot service event loop shutdown with {:?}", e); + } + debug!("Polkadot service shutdown"); + }); + Ok(Service { + thread: Some(thread), + client: client.clone(), + network: network.clone(), + _consensus: consensus_service, + }) + } + + /// Get shared client instance. + pub fn client(&self) -> Arc { + self.client.clone() + } + + /// Get shared network instance. + pub fn network(&self) -> Arc { + self.network.clone() + } +} + +impl Drop for Service { + fn drop(&mut self) { + self.client.stop_notifications(); + self.network.stop_network(); + if let Some(thread) = self.thread.take() { + thread.join().expect("The service thread has panicked"); + } + } +} + diff --git a/polkadot/transaction-pool/src/lib.rs b/polkadot/transaction-pool/src/lib.rs index 4ab327bd18..3864852f8c 100644 --- a/polkadot/transaction-pool/src/lib.rs +++ b/polkadot/transaction-pool/src/lib.rs @@ -60,6 +60,11 @@ error_chain! { description("Transaction had bad signature."), display("Transaction had bad signature."), } + /// Attempted to queue a transaction that is already in the pool. + AlreadyImported(hash: TransactionHash) { + description("Transaction is already in the pool."), + display("Transaction {:?} is already in the pool.", hash), + } /// Import error. Import(err: Box<::std::error::Error + Send>) { description("Error importing transaction"), @@ -257,8 +262,15 @@ impl TransactionPool { let verified = VerifiedTransaction::create(tx, insertion_index)?; // TODO: just use a foreign link when the error type is made public. + let hash = verified.hash.clone(); self.inner.import(verified) - .map_err(|e| ErrorKind::Import(Box::new(e))) + .map_err(|e| + match e { + // TODO: make error types public in transaction_pool. For now just treat all errors as AlradyImported + _ => ErrorKind::AlreadyImported(hash), + // transaction_pool::error::AlreadyImported(h) => ErrorKind::AlreadyImported(h), + // e => ErrorKind::Import(Box::new(e)), + }) .map_err(Into::into) }