Commit e6c79ba9 authored by Svyatoslav Nikolsky's avatar Svyatoslav Nikolsky Committed by Gav Wood
Browse files

Relaying tx/blocks by light clients (#190)

* do not import external transactions into light tx pool

* do not announce blocks on light clients

* blocks_are_not_announced_by_light_nodes
parent 48651dc4
......@@ -18,7 +18,7 @@
use std::time::{Duration, Instant};
use futures::stream::Stream;
use service::Service;
use service::{Service, Components};
use tokio_core::reactor;
use network::{SyncState, SyncProvider};
use polkadot_primitives::Block;
......@@ -28,11 +28,10 @@ use client::{self, BlockchainEvents};
const TIMER_INTERVAL_MS: u64 = 5000;
/// Spawn informant on the event loop
pub fn start<B, E>(service: &Service<B, E>, handle: reactor::Handle)
pub fn start<C>(service: &Service<C>, handle: reactor::Handle)
where
B: client::backend::Backend<Block> + Send + Sync + 'static,
E: client::CallExecutor<Block> + Send + Sync + 'static,
client::error::Error: From<<<B as client::backend::Backend<Block>>::State as state_machine::backend::Backend>::Error>
C: Components,
client::error::Error: From<<<<C as Components>::Backend as client::backend::Backend<Block>>::State as state_machine::Backend>::Error>,
{
let interval = reactor::Interval::new_at(Instant::now(), Duration::from_millis(TIMER_INTERVAL_MS), &handle)
.expect("Error creating informant timer");
......
......@@ -213,11 +213,10 @@ pub fn run<I, T>(args: I) -> error::Result<()> where
}
}
fn run_until_exit<B, E>(mut core: reactor::Core, service: service::Service<B, E>, matches: &clap::ArgMatches, config: service::Configuration) -> error::Result<()>
fn run_until_exit<C>(mut core: reactor::Core, service: service::Service<C>, matches: &clap::ArgMatches, config: service::Configuration) -> error::Result<()>
where
B: client::backend::Backend<Block> + Send + Sync + 'static,
E: client::CallExecutor<Block> + Send + Sync + 'static,
client::error::Error: From<<<B as client::backend::Backend<Block>>::State as state_machine::backend::Backend>::Error>
C: service::Components,
client::error::Error: From<<<<C as service::Components>::Backend as client::backend::Backend<Block>>::State as state_machine::Backend>::Error>,
{
let exit = {
// can't use signal directly here because CtrlC takes only `Fn`.
......
// 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 <http://www.gnu.org/licenses/>.?
//! Polkadot service components.
use std::collections::HashMap;
use std::sync::Arc;
use client::{self, genesis, Client};
use client_db;
use codec::{self, Slicable};
use consensus;
use keystore::Store as Keystore;
use network;
use polkadot_api;
use polkadot_executor::Executor as LocalDispatch;
use polkadot_runtime::{GenesisConfig, BuildExternalities};
use polkadot_primitives::{Block, BlockId, Hash, Header};
use state_machine;
use substrate_executor::NativeExecutor;
use transaction_pool::{self, TransactionPool};
use error;
/// Code executor.
pub type CodeExecutor = NativeExecutor<LocalDispatch>;
/// Polkadot service components.
pub trait Components {
/// Client backend type.
type Backend: 'static + client::backend::Backend<Block>;
/// Polkadot API type.
type Api: 'static + polkadot_api::PolkadotApi + Send + Sync;
/// Code executor type.
type Executor: 'static + client::CallExecutor<Block> + Send + Sync;
/// Create client.
fn build_client(&self, settings: client_db::DatabaseSettings, executor: CodeExecutor, genesis: GenesisBuilder)
-> Result<(Arc<Client<Self::Backend, Self::Executor, Block>>, Option<Arc<network::OnDemand<Block, network::Service<Block>>>>), error::Error>;
/// Create api.
fn build_api(&self, client: Arc<Client<Self::Backend, Self::Executor, Block>>) -> Arc<Self::Api>;
/// Create network transaction pool adapter.
fn build_network_tx_pool(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, api: Arc<Self::Api>, tx_pool: Arc<TransactionPool>)
-> Arc<network::TransactionPool<Block>>;
/// Create consensus service.
fn build_consensus(&self, client: Arc<Client<Self::Backend, Self::Executor, Block>>, network: Arc<network::Service<Block>>, tx_pool: Arc<TransactionPool>, keystore: &Keystore)
-> Result<Option<consensus::Service>, error::Error>;
}
/// Genesis block builder.
pub struct GenesisBuilder {
pub config: GenesisConfig,
}
impl client::GenesisBuilder<Block> for GenesisBuilder {
fn build(self) -> (Header, Vec<(Vec<u8>, Vec<u8>)>) {
let storage = self.config.build_externalities();
let block = genesis::construct_genesis_block::<Block>(&storage);
(block.header, storage.into_iter().collect())
}
}
/// Components for full Polkadot service.
pub struct FullComponents {
/// Is this a validator node?
pub is_validator: bool,
}
impl Components for FullComponents {
type Backend = client_db::Backend<Block>;
type Api = Client<Self::Backend, Self::Executor, Block>;
type Executor = client::LocalCallExecutor<client_db::Backend<Block>, NativeExecutor<LocalDispatch>>;
fn build_client(&self, db_settings: client_db::DatabaseSettings, executor: CodeExecutor, genesis: GenesisBuilder)
-> Result<(Arc<client::Client<Self::Backend, Self::Executor, Block>>, Option<Arc<network::OnDemand<Block, network::Service<Block>>>>), error::Error> {
Ok((Arc::new(client_db::new_client(db_settings, executor, genesis)?), None))
}
fn build_api(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>) -> Arc<Self::Api> {
client
}
fn build_network_tx_pool(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, api: Arc<Self::Api>, pool: Arc<TransactionPool>)
-> Arc<network::TransactionPool<Block>> {
Arc::new(TransactionPoolAdapter {
imports_external_transactions: true,
pool,
client,
api,
})
}
fn build_consensus(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, network: Arc<network::Service<Block>>, tx_pool: Arc<TransactionPool>, keystore: &Keystore)
-> Result<Option<consensus::Service>, error::Error> {
if !self.is_validator {
return Ok(None);
}
// Load the first available key
let key = keystore.load(&keystore.contents()?[0], "")?;
info!("Using authority key {:?}", key.public());
Ok(Some(consensus::Service::new(
client.clone(),
client.clone(),
network.clone(),
tx_pool.clone(),
::std::time::Duration::from_millis(4000), // TODO: dynamic
key,
)))
}
}
/// Components for light Polkadot service.
pub struct LightComponents;
impl Components for LightComponents {
type Backend = client::light::Backend<Block>;
type Api = polkadot_api::light::RemotePolkadotApiWrapper<Self::Backend, Self::Executor>;
type Executor = client::RemoteCallExecutor<client::light::Backend<Block>, network::OnDemand<Block, network::Service<Block>>>;
fn build_client(&self, _settings: client_db::DatabaseSettings, executor: CodeExecutor, genesis: GenesisBuilder)
-> Result<(Arc<client::Client<Self::Backend, Self::Executor, Block>>, Option<Arc<network::OnDemand<Block, network::Service<Block>>>>), error::Error> {
let client_backend = client::light::new_light_backend();
let fetch_checker = Arc::new(client::light::new_fetch_checker(client_backend.clone(), executor));
let fetcher = Arc::new(network::OnDemand::new(fetch_checker));
let client = client::light::new_light(client_backend, fetcher.clone(), genesis)?;
Ok((Arc::new(client), Some(fetcher)))
}
fn build_api(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>) -> Arc<Self::Api> {
Arc::new(polkadot_api::light::RemotePolkadotApiWrapper(client.clone()))
}
fn build_network_tx_pool(&self, client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, api: Arc<Self::Api>, pool: Arc<TransactionPool>)
-> Arc<network::TransactionPool<Block>> {
Arc::new(TransactionPoolAdapter {
imports_external_transactions: false,
pool,
client,
api,
})
}
fn build_consensus(&self, _client: Arc<client::Client<Self::Backend, Self::Executor, Block>>, _network: Arc<network::Service<Block>>, _tx_pool: Arc<TransactionPool>, _keystore: &Keystore)
-> Result<Option<consensus::Service>, error::Error> {
Ok(None)
}
}
/// Transaction pool adapter.
pub struct TransactionPoolAdapter<B, E, A> where A: Send + Sync, E: Send + Sync {
imports_external_transactions: bool,
pool: Arc<TransactionPool>,
client: Arc<Client<B, E, Block>>,
api: Arc<A>,
}
impl<B, E, A> network::TransactionPool<Block> for TransactionPoolAdapter<B, E, A>
where
B: client::backend::Backend<Block> + Send + Sync,
E: client::CallExecutor<Block> + Send + Sync,
client::error::Error: From<<<B as client::backend::Backend<Block>>::State as state_machine::backend::Backend>::Error>,
A: polkadot_api::PolkadotApi + Send + Sync,
{
fn transactions(&self) -> Vec<(Hash, Vec<u8>)> {
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 = match self.api.check_id(BlockId::hash(best_block)) {
Ok(id) => id,
Err(_) => return Vec::new(),
};
let ready = transaction_pool::Ready::create(id, &*self.api);
self.pool.cull_and_get_pending(ready, |pending| pending
.map(|t| {
let hash = t.hash().clone();
(hash, t.primitive_extrinsic())
})
.collect()
)
}
fn import(&self, transaction: &Vec<u8>) -> Option<Hash> {
if !self.imports_external_transactions {
return None;
}
let encoded = transaction.encode();
if let Some(uxt) = codec::Slicable::decode(&mut &encoded[..]) {
match self.pool.import_unchecked_extrinsic(uxt) {
Ok(xt) => Some(*xt.hash()),
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
}
}
fn on_broadcasted(&self, propagations: HashMap<Hash, Vec<String>>) {
self.pool.on_broadcasted(propagations)
}
}
......@@ -51,106 +51,39 @@ extern crate log;
#[macro_use]
extern crate hex_literal;
mod components;
mod error;
mod config;
use std::collections::HashMap;
use std::sync::Arc;
use std::thread;
use futures::prelude::*;
use tokio_core::reactor::Core;
use codec::Slicable;
use primitives::AuthorityId;
use transaction_pool::TransactionPool;
use substrate_executor::NativeExecutor;
use polkadot_executor::Executor as LocalDispatch;
use keystore::Store as Keystore;
use polkadot_api::PolkadotApi;
use polkadot_primitives::{Block, BlockId, Hash, Header};
use polkadot_primitives::{Block, BlockId, Hash};
use polkadot_runtime::{GenesisConfig, ConsensusConfig, CouncilConfig, DemocracyConfig,
SessionConfig, StakingConfig, BuildExternalities};
use client::backend::Backend;
use client::{genesis, Client, BlockchainEvents, CallExecutor};
SessionConfig, StakingConfig};
use client::{Client, BlockchainEvents};
use network::ManageNetwork;
use exit_future::Signal;
pub use self::error::{ErrorKind, Error};
pub use self::components::{Components, FullComponents, LightComponents};
pub use config::{Configuration, Role, OptionChainSpec, ChainSpec};
type CodeExecutor = NativeExecutor<LocalDispatch>;
/// Polkadot service.
pub struct Service<B, E> {
pub struct Service<Components: components::Components> {
thread: Option<thread::JoinHandle<()>>,
client: Arc<Client<B, E, Block>>,
client: Arc<Client<Components::Backend, Components::Executor, Block>>,
network: Arc<network::Service<Block>>,
transaction_pool: Arc<TransactionPool>,
signal: Option<Signal>,
_consensus: Option<consensus::Service>,
}
struct TransactionPoolAdapter<B, E, A> where A: Send + Sync, E: Send + Sync {
pool: Arc<TransactionPool>,
client: Arc<Client<B, E, Block>>,
api: Arc<A>,
}
impl<B, E, A> network::TransactionPool<Block> for TransactionPoolAdapter<B, E, A>
where
B: Backend<Block> + Send + Sync,
E: client::CallExecutor<Block> + Send + Sync,
client::error::Error: From<<<B as Backend<Block>>::State as state_machine::backend::Backend>::Error>,
A: PolkadotApi + Send + Sync,
{
fn transactions(&self) -> Vec<(Hash, Vec<u8>)> {
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 = match self.api.check_id(BlockId::hash(best_block)) {
Ok(id) => id,
Err(_) => return Vec::new(),
};
let ready = transaction_pool::Ready::create(id, &*self.api);
self.pool.cull_and_get_pending(ready, |pending| pending
.map(|t| {
let hash = t.hash().clone();
(hash, t.primitive_extrinsic())
})
.collect()
)
}
fn import(&self, transaction: &Vec<u8>) -> Option<Hash> {
let encoded = transaction.encode();
if let Some(uxt) = codec::Slicable::decode(&mut &encoded[..]) {
match self.pool.import_unchecked_extrinsic(uxt) {
Ok(xt) => Some(*xt.hash()),
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
}
}
fn on_broadcasted(&self, propagations: HashMap<Hash, Vec<String>>) {
self.pool.on_broadcasted(propagations)
}
}
pub struct ChainConfig {
genesis_config: GenesisConfig,
boot_nodes: Vec<String>,
......@@ -294,90 +227,24 @@ fn local_testnet_config() -> ChainConfig {
])
}
struct GenesisBuilder {
config: GenesisConfig,
}
impl client::GenesisBuilder<Block> for GenesisBuilder {
fn build(self) -> (Header, Vec<(Vec<u8>, Vec<u8>)>) {
let storage = self.config.build_externalities();
let block = genesis::construct_genesis_block::<Block>(&storage);
(block.header, storage.into_iter().collect())
}
}
/// Creates light client and register protocol with the network service
pub fn new_light(config: Configuration)
-> Result<
Service<
client::light::Backend<Block>,
client::RemoteCallExecutor<client::light::Backend<Block>, network::OnDemand<Block, network::Service<Block>>>
>,
error::Error,
> {
Service::new(move |_, executor, genesis_builder: GenesisBuilder| {
let client_backend = client::light::new_light_backend();
let fetch_checker = Arc::new(client::light::new_fetch_checker(client_backend.clone(), executor));
let fetcher = Arc::new(network::OnDemand::new(fetch_checker));
let client = client::light::new_light(client_backend, fetcher.clone(), genesis_builder)?;
Ok((Arc::new(client), Some(fetcher)))
},
|client| Arc::new(polkadot_api::light::RemotePolkadotApiWrapper(client.clone())),
|_client, _network, _tx_pool, _keystore| Ok(None),
config)
pub fn new_light(config: Configuration) -> Result<Service<components::LightComponents>, error::Error> {
Service::new(components::LightComponents, config)
}
/// Creates full client and register protocol with the network service
pub fn new_full(config: Configuration) -> Result<Service<client_db::Backend<Block>, client::LocalCallExecutor<client_db::Backend<Block>, CodeExecutor>>, error::Error> {
pub fn new_full(config: Configuration) -> Result<Service<components::FullComponents>, error::Error> {
let is_validator = (config.roles & Role::VALIDATOR) == Role::VALIDATOR;
Service::new(|db_settings, executor, genesis_builder: GenesisBuilder|
Ok((Arc::new(client_db::new_client(db_settings, executor, genesis_builder)?), None)),
|client| client,
|client, network, tx_pool, keystore| {
if !is_validator {
return Ok(None);
}
// Load the first available key. Code above makes sure it exisis.
let key = keystore.load(&keystore.contents()?[0], "")?;
info!("Using authority key {:?}", key.public());
Ok(Some(consensus::Service::new(
client.clone(),
client.clone(),
network.clone(),
tx_pool.clone(),
::std::time::Duration::from_millis(4000), // TODO: dynamic
key,
)))
},
config)
Service::new(components::FullComponents { is_validator }, config)
}
impl<B, E> Service<B, E>
impl<Components> Service<Components>
where
B: Backend<Block> + Send + Sync + 'static,
E: CallExecutor<Block> + Send + Sync + 'static,
client::error::Error: From<<<B as Backend<Block>>::State as state_machine::backend::Backend>::Error>
Components: components::Components,
client::error::Error: From<<<<Components as components::Components>::Backend as client::backend::Backend<Block>>::State as state_machine::Backend>::Error>,
{
/// Creates and register protocol with the network service
fn new<F, G, C, A>(client_creator: F, api_creator: G, consensus_creator: C, mut config: Configuration) -> Result<Self, error::Error>
where
F: FnOnce(
client_db::DatabaseSettings,
CodeExecutor,
GenesisBuilder,
) -> Result<(Arc<Client<B, E, Block>>, Option<Arc<network::OnDemand<Block, network::Service<Block>>>>), error::Error>,
G: Fn(
Arc<Client<B, E, Block>>,
) -> Arc<A>,
C: Fn(
Arc<Client<B, E, Block>>,
Arc<network::Service<Block>>,
Arc<TransactionPool>,
&Keystore
) -> Result<Option<consensus::Service>, error::Error>,
A: PolkadotApi + Send + Sync + 'static,
{
fn new(components: Components, mut config: Configuration) -> Result<Self, error::Error> {
use std::sync::Barrier;
let (signal, exit) = ::exit_future::signal();
......@@ -402,7 +269,7 @@ impl<B, E> Service<B, E>
};
config.network.boot_nodes.extend(boot_nodes);
let genesis_builder = GenesisBuilder {
let genesis_builder = components::GenesisBuilder {
config: genesis_config,
};
......@@ -411,17 +278,15 @@ impl<B, E> Service<B, E>
path: config.database_path.into(),
};
let (client, on_demand) = client_creator(db_settings, executor, genesis_builder)?;
let api = api_creator(client.clone());
let (client, on_demand) = components.build_client(db_settings, executor, genesis_builder)?;
let api = components.build_api(client.clone());
let best_header = client.best_block_header()?;
info!("Best block is #{}", best_header.number);
telemetry!("node.start"; "height" => best_header.number, "best" => ?best_header.hash());
let transaction_pool = Arc::new(TransactionPool::new(config.transaction_pool));
let transaction_pool_adapter = Arc::new(TransactionPoolAdapter {
pool: transaction_pool.clone(),
client: client.clone(),
api: api.clone(),
});
let transaction_pool_adapter = components.build_network_tx_pool(client.clone(), api.clone(), transaction_pool.clone());
let network_params = network::Params {
config: network::ProtocolConfig {
roles: config.roles,
......@@ -481,7 +346,7 @@ impl<B, E> Service<B, E>
barrier.wait();
// Spin consensus service if configured
let consensus_service = consensus_creator(client.clone(), network.clone(), transaction_pool.clone(), &keystore)?;
let consensus_service = components.build_consensus(client.clone(), network.clone(), transaction_pool.clone(), &keystore)?;
Ok(Service {
thread: Some(thread),
......@@ -494,7 +359,7 @@ impl<B, E> Service<B, E>
}
/// Get shared client instance.
pub fn client(&self) -> Arc<Client<B, E, Block>> {
pub fn client(&self) -> Arc<Client<Components::Backend, Components::Executor, Block>> {
self.client.clone()
}
......@@ -523,7 +388,7 @@ pub fn prune_imported<A>(api: &A, pool: &TransactionPool, hash: Hash)
}
}
impl<B, E> Drop for Service<B, E> {
impl<Components> Drop for Service<Components> where Components: components::Components {
fn drop(&mut self) {
self.network.stop_network();
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment