Merge branch 'master' into evm-instructions

Conflicts:
	evmbin/bench.sh
This commit is contained in:
Tomasz Drwięga 2016-07-12 10:32:40 +02:00
commit 2746c1c18c
44 changed files with 559 additions and 536 deletions

View File

@ -27,8 +27,8 @@ ethcore = { path = "ethcore" }
ethcore-util = { path = "util" } ethcore-util = { path = "util" }
ethsync = { path = "sync" } ethsync = { path = "sync" }
ethcore-devtools = { path = "devtools" } ethcore-devtools = { path = "devtools" }
ethcore-rpc = { path = "rpc", optional = true } ethcore-rpc = { path = "rpc" }
ethcore-signer = { path = "signer", optional = true } ethcore-signer = { path = "signer" }
ethcore-dapps = { path = "dapps", optional = true } ethcore-dapps = { path = "dapps", optional = true }
semver = "0.2" semver = "0.2"
ethcore-ipc-nano = { path = "ipc/nano" } ethcore-ipc-nano = { path = "ipc/nano" }
@ -48,8 +48,7 @@ version = "0.8"
default-features = false default-features = false
[features] [features]
default = ["rpc", "dapps", "ethcore-signer"] default = ["dapps"]
rpc = ["ethcore-rpc"]
dapps = ["ethcore-dapps"] dapps = ["ethcore-dapps"]
dev = ["clippy", "ethcore/dev", "ethcore-util/dev", "ethsync/dev", "ethcore-rpc/dev", dev = ["clippy", "ethcore/dev", "ethcore-util/dev", "ethsync/dev", "ethcore-rpc/dev",
"ethcore-dapps/dev", "ethcore-signer/dev"] "ethcore-dapps/dev", "ethcore-signer/dev"]

View File

@ -60,7 +60,7 @@ impl WriteCache {
self.entries.insert(key, WriteCacheEntry::Remove); self.entries.insert(key, WriteCacheEntry::Remove);
} }
fn get(&self, key: &Vec<u8>) -> Option<Vec<u8>> { fn get(&self, key: &[u8]) -> Option<Vec<u8>> {
self.entries.get(key).and_then( self.entries.get(key).and_then(
|vec_ref| match vec_ref { |vec_ref| match vec_ref {
&WriteCacheEntry::Write(ref val) => Some(val.clone()), &WriteCacheEntry::Write(ref val) => Some(val.clone()),

View File

@ -16,8 +16,6 @@
//! Blockchain block. //! Blockchain block.
#![cfg_attr(feature="dev", allow(ptr_arg))] // Because of &LastHashes -> &Vec<_>
use common::*; use common::*;
use engine::*; use engine::*;
use state::*; use state::*;
@ -76,11 +74,11 @@ pub struct BlockRefMut<'a> {
/// Block header. /// Block header.
pub header: &'a mut Header, pub header: &'a mut Header,
/// Block transactions. /// Block transactions.
pub transactions: &'a Vec<SignedTransaction>, pub transactions: &'a [SignedTransaction],
/// Block uncles. /// Block uncles.
pub uncles: &'a Vec<Header>, pub uncles: &'a [Header],
/// Transaction receipts. /// Transaction receipts.
pub receipts: &'a Vec<Receipt>, pub receipts: &'a [Receipt],
/// State. /// State.
pub state: &'a mut State, pub state: &'a mut State,
/// Traces. /// Traces.
@ -92,11 +90,11 @@ pub struct BlockRef<'a> {
/// Block header. /// Block header.
pub header: &'a Header, pub header: &'a Header,
/// Block transactions. /// Block transactions.
pub transactions: &'a Vec<SignedTransaction>, pub transactions: &'a [SignedTransaction],
/// Block uncles. /// Block uncles.
pub uncles: &'a Vec<Header>, pub uncles: &'a [Header],
/// Transaction receipts. /// Transaction receipts.
pub receipts: &'a Vec<Receipt>, pub receipts: &'a [Receipt],
/// State. /// State.
pub state: &'a State, pub state: &'a State,
/// Traces. /// Traces.
@ -152,16 +150,16 @@ pub trait IsBlock {
fn state(&self) -> &State { &self.block().state } fn state(&self) -> &State { &self.block().state }
/// Get all information on transactions in this block. /// Get all information on transactions in this block.
fn transactions(&self) -> &Vec<SignedTransaction> { &self.block().base.transactions } fn transactions(&self) -> &[SignedTransaction] { &self.block().base.transactions }
/// Get all information on receipts in this block. /// Get all information on receipts in this block.
fn receipts(&self) -> &Vec<Receipt> { &self.block().receipts } fn receipts(&self) -> &[Receipt] { &self.block().receipts }
/// Get all information concerning transaction tracing in this block. /// Get all information concerning transaction tracing in this block.
fn traces(&self) -> &Option<Vec<Trace>> { &self.block().traces } fn traces(&self) -> &Option<Vec<Trace>> { &self.block().traces }
/// Get all uncles in this block. /// Get all uncles in this block.
fn uncles(&self) -> &Vec<Header> { &self.block().base.uncles } fn uncles(&self) -> &[Header] { &self.block().base.uncles }
} }
/// Trait for a object that has a state database. /// Trait for a object that has a state database.

View File

@ -104,7 +104,7 @@ struct VerifyingBlock {
struct QueueSignal { struct QueueSignal {
deleting: Arc<AtomicBool>, deleting: Arc<AtomicBool>,
signalled: AtomicBool, signalled: AtomicBool,
message_channel: IoChannel<NetSyncMessage>, message_channel: IoChannel<ClientIoMessage>,
} }
impl QueueSignal { impl QueueSignal {
@ -116,7 +116,7 @@ impl QueueSignal {
} }
if self.signalled.compare_and_swap(false, true, AtomicOrdering::Relaxed) == false { if self.signalled.compare_and_swap(false, true, AtomicOrdering::Relaxed) == false {
if let Err(e) = self.message_channel.send(UserMessage(SyncMessage::BlockVerified)) { if let Err(e) = self.message_channel.send(ClientIoMessage::BlockVerified) {
debug!("Error sending BlockVerified message: {:?}", e); debug!("Error sending BlockVerified message: {:?}", e);
} }
} }
@ -137,7 +137,7 @@ struct Verification {
impl BlockQueue { impl BlockQueue {
/// Creates a new queue instance. /// Creates a new queue instance.
pub fn new(config: BlockQueueConfig, engine: Arc<Box<Engine>>, message_channel: IoChannel<NetSyncMessage>) -> BlockQueue { pub fn new(config: BlockQueueConfig, engine: Arc<Box<Engine>>, message_channel: IoChannel<ClientIoMessage>) -> BlockQueue {
let verification = Arc::new(Verification { let verification = Arc::new(Verification {
unverified: Mutex::new(VecDeque::new()), unverified: Mutex::new(VecDeque::new()),
verified: Mutex::new(VecDeque::new()), verified: Mutex::new(VecDeque::new()),

View File

@ -0,0 +1,40 @@
// Copyright 2015, 2016 Ethcore (UK) Ltd.
// This file is part of Parity.
// Parity 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.
// Parity 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 Parity. If not, see <http://www.gnu.org/licenses/>.
use util::numbers::*;
/// Represents what has to be handled by actor listening to chain events
pub trait ChainNotify : Send + Sync {
/// fires when chain has new blocks
fn new_blocks(&self,
_imported: Vec<H256>,
_invalid: Vec<H256>,
_enacted: Vec<H256>,
_retracted: Vec<H256>,
_sealed: Vec<H256>) {
// does nothing by default
}
/// fires when chain achieves active mode
fn start(&self) {
// does nothing by default
}
/// fires when chain achieves passive mode
fn stop(&self) {
// does nothing by default
}
}

View File

@ -28,7 +28,6 @@ use std::time::Instant;
// util // util
use util::numbers::*; use util::numbers::*;
use util::panics::*; use util::panics::*;
use util::network::*;
use util::io::*; use util::io::*;
use util::rlp; use util::rlp;
use util::sha3::*; use util::sha3::*;
@ -47,7 +46,7 @@ use state::State;
use spec::Spec; use spec::Spec;
use engine::Engine; use engine::Engine;
use views::HeaderView; use views::HeaderView;
use service::{NetSyncMessage, SyncMessage}; use service::ClientIoMessage;
use env_info::LastHashes; use env_info::LastHashes;
use verification; use verification;
use verification::{PreverifiedBlock, Verifier}; use verification::{PreverifiedBlock, Verifier};
@ -60,7 +59,7 @@ use block_queue::{BlockQueue, BlockQueueInfo};
use blockchain::{BlockChain, BlockProvider, TreeRoute, ImportRoute}; use blockchain::{BlockChain, BlockProvider, TreeRoute, ImportRoute};
use client::{BlockID, TransactionID, UncleID, TraceId, ClientConfig, use client::{BlockID, TransactionID, UncleID, TraceId, ClientConfig,
DatabaseCompactionProfile, BlockChainClient, MiningBlockChainClient, DatabaseCompactionProfile, BlockChainClient, MiningBlockChainClient,
TraceFilter, CallAnalytics, BlockImportError, Mode}; TraceFilter, CallAnalytics, BlockImportError, Mode, ChainNotify};
use client::Error as ClientError; use client::Error as ClientError;
use env_info::EnvInfo; use env_info::EnvInfo;
use executive::{Executive, Executed, TransactOptions, contract_address}; use executive::{Executive, Executed, TransactOptions, contract_address};
@ -141,7 +140,8 @@ pub struct Client {
miner: Arc<Miner>, miner: Arc<Miner>,
sleep_state: Mutex<SleepState>, sleep_state: Mutex<SleepState>,
liveness: AtomicBool, liveness: AtomicBool,
io_channel: IoChannel<NetSyncMessage>, io_channel: IoChannel<ClientIoMessage>,
notify: RwLock<Option<Weak<ChainNotify>>>,
queue_transactions: AtomicUsize, queue_transactions: AtomicUsize,
previous_enode: Mutex<Option<String>>, previous_enode: Mutex<Option<String>>,
} }
@ -178,7 +178,7 @@ impl Client {
spec: Spec, spec: Spec,
path: &Path, path: &Path,
miner: Arc<Miner>, miner: Arc<Miner>,
message_channel: IoChannel<NetSyncMessage> message_channel: IoChannel<ClientIoMessage>,
) -> Result<Arc<Client>, ClientError> { ) -> Result<Arc<Client>, ClientError> {
let path = get_db_path(path, config.pruning, spec.genesis_header().hash()); let path = get_db_path(path, config.pruning, spec.genesis_header().hash());
let gb = spec.genesis_block(); let gb = spec.genesis_block();
@ -228,12 +228,24 @@ impl Client {
trie_factory: TrieFactory::new(config.trie_spec), trie_factory: TrieFactory::new(config.trie_spec),
miner: miner, miner: miner,
io_channel: message_channel, io_channel: message_channel,
notify: RwLock::new(None),
queue_transactions: AtomicUsize::new(0), queue_transactions: AtomicUsize::new(0),
previous_enode: Mutex::new(None), previous_enode: Mutex::new(None),
}; };
Ok(Arc::new(client)) Ok(Arc::new(client))
} }
/// Sets the actor to be notified on certain events
pub fn set_notify(&self, target: &Arc<ChainNotify>) {
let mut write_lock = self.notify.unwrapped_write();
*write_lock = Some(Arc::downgrade(target));
}
fn notify(&self) -> Option<Arc<ChainNotify>> {
let read_lock = self.notify.unwrapped_read();
read_lock.as_ref().and_then(|weak| weak.upgrade())
}
/// Flush the block import queue. /// Flush the block import queue.
pub fn flush_queue(&self) { pub fn flush_queue(&self) {
self.block_queue.flush(); self.block_queue.flush();
@ -327,52 +339,54 @@ impl Client {
} }
/// This is triggered by a message coming from a block queue when the block is ready for insertion /// This is triggered by a message coming from a block queue when the block is ready for insertion
pub fn import_verified_blocks(&self, io: &IoChannel<NetSyncMessage>) -> usize { pub fn import_verified_blocks(&self, io: &IoChannel<ClientIoMessage>) -> usize {
let max_blocks_to_import = 64; let max_blocks_to_import = 64;
let (imported_blocks, import_results, invalid_blocks, original_best, imported) = {
let mut imported_blocks = Vec::with_capacity(max_blocks_to_import);
let mut invalid_blocks = HashSet::new();
let mut import_results = Vec::with_capacity(max_blocks_to_import);
let mut imported_blocks = Vec::with_capacity(max_blocks_to_import); let _import_lock = self.import_lock.lock();
let mut invalid_blocks = HashSet::new(); let _timer = PerfTimer::new("import_verified_blocks");
let mut import_results = Vec::with_capacity(max_blocks_to_import); let blocks = self.block_queue.drain(max_blocks_to_import);
let _import_lock = self.import_lock.lock(); let original_best = self.chain_info().best_block_hash;
let _timer = PerfTimer::new("import_verified_blocks");
let blocks = self.block_queue.drain(max_blocks_to_import);
let original_best = self.chain_info().best_block_hash; for block in blocks {
let header = &block.header;
for block in blocks { if invalid_blocks.contains(&header.parent_hash) {
let header = &block.header; invalid_blocks.insert(header.hash());
continue;
}
let closed_block = self.check_and_close_block(&block);
if let Err(_) = closed_block {
invalid_blocks.insert(header.hash());
continue;
}
let closed_block = closed_block.unwrap();
imported_blocks.push(header.hash());
if invalid_blocks.contains(&header.parent_hash) { let route = self.commit_block(closed_block, &header.hash(), &block.bytes);
invalid_blocks.insert(header.hash()); import_results.push(route);
continue;
self.report.unwrapped_write().accrue_block(&block);
trace!(target: "client", "Imported #{} ({})", header.number(), header.hash());
} }
let closed_block = self.check_and_close_block(&block);
if let Err(_) = closed_block { let imported = imported_blocks.len();
invalid_blocks.insert(header.hash()); let invalid_blocks = invalid_blocks.into_iter().collect::<Vec<H256>>();
continue;
{
if !invalid_blocks.is_empty() {
self.block_queue.mark_as_bad(&invalid_blocks);
}
if !imported_blocks.is_empty() {
self.block_queue.mark_as_good(&imported_blocks);
}
} }
let closed_block = closed_block.unwrap(); (imported_blocks, import_results, invalid_blocks, original_best, imported)
imported_blocks.push(header.hash()); };
let route = self.commit_block(closed_block, &header.hash(), &block.bytes);
import_results.push(route);
self.report.unwrapped_write().accrue_block(&block);
trace!(target: "client", "Imported #{} ({})", header.number(), header.hash());
}
let imported = imported_blocks.len();
let invalid_blocks = invalid_blocks.into_iter().collect::<Vec<H256>>();
{
if !invalid_blocks.is_empty() {
self.block_queue.mark_as_bad(&invalid_blocks);
}
if !imported_blocks.is_empty() {
self.block_queue.mark_as_good(&imported_blocks);
}
}
{ {
if !imported_blocks.is_empty() && self.block_queue.queue_info().is_empty() { if !imported_blocks.is_empty() && self.block_queue.queue_info().is_empty() {
@ -382,13 +396,15 @@ impl Client {
self.miner.chain_new_blocks(self, &imported_blocks, &invalid_blocks, &enacted, &retracted); self.miner.chain_new_blocks(self, &imported_blocks, &invalid_blocks, &enacted, &retracted);
} }
io.send(NetworkIoMessage::User(SyncMessage::NewChainBlocks { if let Some(notify) = self.notify() {
imported: imported_blocks, notify.new_blocks(
invalid: invalid_blocks, imported_blocks,
enacted: enacted, invalid_blocks,
retracted: retracted, enacted,
sealed: Vec::new(), retracted,
})).unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e)); Vec::new(),
);
}
} }
} }
@ -410,7 +426,7 @@ impl Client {
}; };
// Commit results // Commit results
let receipts = block.receipts().clone(); let receipts = block.receipts().to_owned();
let traces = From::from(block.traces().clone().unwrap_or_else(Vec::new)); let traces = From::from(block.traces().clone().unwrap_or_else(Vec::new));
// CHECK! I *think* this is fine, even if the state_root is equal to another // CHECK! I *think* this is fine, even if the state_root is equal to another
@ -566,7 +582,9 @@ impl Client {
fn wake_up(&self) { fn wake_up(&self) {
if !self.liveness.load(AtomicOrdering::Relaxed) { if !self.liveness.load(AtomicOrdering::Relaxed) {
self.liveness.store(true, AtomicOrdering::Relaxed); self.liveness.store(true, AtomicOrdering::Relaxed);
self.io_channel.send(NetworkIoMessage::User(SyncMessage::StartNetwork)).unwrap(); if let Some(notify) = self.notify() {
notify.start();
}
trace!(target: "mode", "wake_up: Waking."); trace!(target: "mode", "wake_up: Waking.");
} }
} }
@ -576,7 +594,9 @@ impl Client {
// only sleep if the import queue is mostly empty. // only sleep if the import queue is mostly empty.
if self.queue_info().total_queue_size() <= MAX_QUEUE_SIZE_TO_SLEEP_ON { if self.queue_info().total_queue_size() <= MAX_QUEUE_SIZE_TO_SLEEP_ON {
self.liveness.store(false, AtomicOrdering::Relaxed); self.liveness.store(false, AtomicOrdering::Relaxed);
self.io_channel.send(NetworkIoMessage::User(SyncMessage::StopNetwork)).unwrap(); if let Some(notify) = self.notify() {
notify.stop();
}
trace!(target: "mode", "sleep: Sleeping."); trace!(target: "mode", "sleep: Sleeping.");
} else { } else {
trace!(target: "mode", "sleep: Cannot sleep - syncing ongoing."); trace!(target: "mode", "sleep: Cannot sleep - syncing ongoing.");
@ -901,7 +921,7 @@ impl BlockChainClient for Client {
debug!("Ignoring {} transactions: queue is full", transactions.len()); debug!("Ignoring {} transactions: queue is full", transactions.len());
} else { } else {
let len = transactions.len(); let len = transactions.len();
match self.io_channel.send(NetworkIoMessage::User(SyncMessage::NewTransactions(transactions))) { match self.io_channel.send(ClientIoMessage::NewTransactions(transactions)) {
Ok(_) => { Ok(_) => {
self.queue_transactions.fetch_add(len, AtomicOrdering::SeqCst); self.queue_transactions.fetch_add(len, AtomicOrdering::SeqCst);
} }
@ -969,13 +989,15 @@ impl MiningBlockChainClient for Client {
let (enacted, retracted) = self.calculate_enacted_retracted(&[route]); let (enacted, retracted) = self.calculate_enacted_retracted(&[route]);
self.miner.chain_new_blocks(self, &[h.clone()], &[], &enacted, &retracted); self.miner.chain_new_blocks(self, &[h.clone()], &[], &enacted, &retracted);
self.io_channel.send(NetworkIoMessage::User(SyncMessage::NewChainBlocks { if let Some(notify) = self.notify() {
imported: vec![h.clone()], notify.new_blocks(
invalid: vec![], vec![h.clone()],
enacted: enacted, vec![],
retracted: retracted, enacted,
sealed: vec![h.clone()], retracted,
})).unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e)); vec![h.clone()],
);
}
} }
if self.chain_info().best_block_hash != original_best { if self.chain_info().best_block_hash != original_best {

View File

@ -20,6 +20,7 @@ mod config;
mod error; mod error;
mod test_client; mod test_client;
mod trace; mod trace;
mod chain_notify;
pub use self::client::*; pub use self::client::*;
pub use self::config::{Mode, ClientConfig, DatabaseCompactionProfile, BlockQueueConfig, BlockChainConfig, Switch, VMType}; pub use self::config::{Mode, ClientConfig, DatabaseCompactionProfile, BlockQueueConfig, BlockChainConfig, Switch, VMType};
@ -29,6 +30,7 @@ pub use self::test_client::{TestBlockChainClient, EachBlockWith};
pub use types::trace_filter::Filter as TraceFilter; pub use types::trace_filter::Filter as TraceFilter;
pub use executive::{Executed, Executive, TransactOptions}; pub use executive::{Executed, Executive, TransactOptions};
pub use env_info::{LastHashes, EnvInfo}; pub use env_info::{LastHashes, EnvInfo};
pub use self::chain_notify::ChainNotify;
use util::bytes::Bytes; use util::bytes::Bytes;
use util::hash::{Address, H256, H2048}; use util::hash::{Address, H256, H2048};

View File

@ -143,7 +143,7 @@ impl Header {
/// Get the difficulty field of the header. /// Get the difficulty field of the header.
pub fn difficulty(&self) -> &U256 { &self.difficulty } pub fn difficulty(&self) -> &U256 { &self.difficulty }
/// Get the seal field of the header. /// Get the seal field of the header.
pub fn seal(&self) -> &Vec<Bytes> { &self.seal } pub fn seal(&self) -> &[Bytes] { &self.seal }
// TODO: seal_at, set_seal_at &c. // TODO: seal_at, set_seal_at &c.

View File

@ -665,7 +665,7 @@ impl MinerService for Miner {
}; };
match (&self.options.pending_set, sealing_set) { match (&self.options.pending_set, sealing_set) {
(&PendingSet::AlwaysQueue, _) | (&PendingSet::SealingOrElseQueue, None) => queue.top_transactions(), (&PendingSet::AlwaysQueue, _) | (&PendingSet::SealingOrElseQueue, None) => queue.top_transactions(),
(_, sealing) => sealing.map_or_else(Vec::new, |s| s.transactions().clone()), (_, sealing) => sealing.map_or_else(Vec::new, |s| s.transactions().to_owned()),
} }
} }
@ -702,7 +702,7 @@ impl MinerService for Miner {
.iter() .iter()
.map(|t| t.hash()); .map(|t| t.hash());
let receipts = pending.receipts().clone().into_iter(); let receipts = pending.receipts().iter().cloned();
hashes.zip(receipts).collect() hashes.zip(receipts).collect()
}, },

View File

@ -20,67 +20,50 @@ use util::*;
use util::panics::*; use util::panics::*;
use spec::Spec; use spec::Spec;
use error::*; use error::*;
use client::{Client, ClientConfig}; use client::{Client, ClientConfig, ChainNotify};
use miner::Miner; use miner::Miner;
/// Message type for external and internal events /// Message type for external and internal events
#[derive(Clone)] #[derive(Clone)]
pub enum SyncMessage { pub enum ClientIoMessage {
/// New block has been imported into the blockchain
NewChainBlocks {
/// Hashes of blocks imported to blockchain
imported: Vec<H256>,
/// Hashes of blocks not imported to blockchain (because were invalid)
invalid: Vec<H256>,
/// Hashes of blocks that were removed from canonical chain
retracted: Vec<H256>,
/// Hashes of blocks that are now included in cannonical chain
enacted: Vec<H256>,
/// Hashes of blocks that are sealed by this node
sealed: Vec<H256>,
},
/// Best Block Hash in chain has been changed /// Best Block Hash in chain has been changed
NewChainHead, NewChainHead,
/// A block is ready /// A block is ready
BlockVerified, BlockVerified,
/// New transaction RLPs are ready to be imported /// New transaction RLPs are ready to be imported
NewTransactions(Vec<Bytes>), NewTransactions(Vec<Bytes>),
/// Start network command.
StartNetwork,
/// Stop network command.
StopNetwork,
} }
/// IO Message type used for Network service
pub type NetSyncMessage = NetworkIoMessage<SyncMessage>;
/// Client service setup. Creates and registers client and network services with the IO subsystem. /// Client service setup. Creates and registers client and network services with the IO subsystem.
pub struct ClientService { pub struct ClientService {
net_service: Arc<NetworkService<SyncMessage>>, io_service: Arc<IoService<ClientIoMessage>>,
client: Arc<Client>, client: Arc<Client>,
panic_handler: Arc<PanicHandler> panic_handler: Arc<PanicHandler>
} }
impl ClientService { impl ClientService {
/// Start the service in a separate thread. /// Start the service in a separate thread.
pub fn start(config: ClientConfig, spec: Spec, net_config: NetworkConfiguration, db_path: &Path, miner: Arc<Miner>, enable_network: bool) -> Result<ClientService, Error> { pub fn start(
config: ClientConfig,
spec: Spec,
db_path: &Path,
miner: Arc<Miner>,
) -> Result<ClientService, Error>
{
let panic_handler = PanicHandler::new_in_arc(); let panic_handler = PanicHandler::new_in_arc();
let net_service = try!(NetworkService::new(net_config)); let io_service = try!(IoService::<ClientIoMessage>::start());
panic_handler.forward_from(&net_service); panic_handler.forward_from(&io_service);
if enable_network {
try!(net_service.start());
}
info!("Configured for {} using {} engine", spec.name.clone().apply(Colour::White.bold()), spec.engine.name().apply(Colour::Yellow.bold())); info!("Configured for {} using {} engine", spec.name.clone().apply(Colour::White.bold()), spec.engine.name().apply(Colour::Yellow.bold()));
let client = try!(Client::new(config, spec, db_path, miner, net_service.io().channel())); let client = try!(Client::new(config, spec, db_path, miner, io_service.channel()));
panic_handler.forward_from(client.deref()); panic_handler.forward_from(client.deref());
let client_io = Arc::new(ClientIoHandler { let client_io = Arc::new(ClientIoHandler {
client: client.clone() client: client.clone()
}); });
try!(net_service.io().register_handler(client_io)); try!(io_service.register_handler(client_io));
Ok(ClientService { Ok(ClientService {
net_service: Arc::new(net_service), io_service: Arc::new(io_service),
client: client, client: client,
panic_handler: panic_handler, panic_handler: panic_handler,
}) })
@ -92,8 +75,8 @@ impl ClientService {
} }
/// Get general IO interface /// Get general IO interface
pub fn register_io_handler(&self, handler: Arc<IoHandler<NetSyncMessage> + Send>) -> Result<(), IoError> { pub fn register_io_handler(&self, handler: Arc<IoHandler<ClientIoMessage> + Send>) -> Result<(), IoError> {
self.net_service.io().register_handler(handler) self.io_service.register_handler(handler)
} }
/// Get client interface /// Get client interface
@ -102,8 +85,13 @@ impl ClientService {
} }
/// Get network service component /// Get network service component
pub fn network(&mut self) -> Arc<NetworkService<SyncMessage>> { pub fn io(&self) -> Arc<IoService<ClientIoMessage>> {
self.net_service.clone() self.io_service.clone()
}
/// Set the actor to be notified on certain chain events
pub fn set_notify(&self, notify: &Arc<ChainNotify>) {
self.client.set_notify(notify);
} }
} }
@ -121,26 +109,22 @@ struct ClientIoHandler {
const CLIENT_TICK_TIMER: TimerToken = 0; const CLIENT_TICK_TIMER: TimerToken = 0;
const CLIENT_TICK_MS: u64 = 5000; const CLIENT_TICK_MS: u64 = 5000;
impl IoHandler<NetSyncMessage> for ClientIoHandler { impl IoHandler<ClientIoMessage> for ClientIoHandler {
fn initialize(&self, io: &IoContext<NetSyncMessage>) { fn initialize(&self, io: &IoContext<ClientIoMessage>) {
io.register_timer(CLIENT_TICK_TIMER, CLIENT_TICK_MS).expect("Error registering client timer"); io.register_timer(CLIENT_TICK_TIMER, CLIENT_TICK_MS).expect("Error registering client timer");
} }
fn timeout(&self, _io: &IoContext<NetSyncMessage>, timer: TimerToken) { fn timeout(&self, _io: &IoContext<ClientIoMessage>, timer: TimerToken) {
if timer == CLIENT_TICK_TIMER { if timer == CLIENT_TICK_TIMER {
self.client.tick(); self.client.tick();
} }
} }
#[cfg_attr(feature="dev", allow(single_match))] #[cfg_attr(feature="dev", allow(single_match))]
fn message(&self, io: &IoContext<NetSyncMessage>, net_message: &NetSyncMessage) { fn message(&self, io: &IoContext<ClientIoMessage>, net_message: &ClientIoMessage) {
match *net_message { match *net_message {
UserMessage(ref message) => match *message { ClientIoMessage::BlockVerified => { self.client.import_verified_blocks(&io.channel()); }
SyncMessage::BlockVerified => { self.client.import_verified_blocks(&io.channel()); } ClientIoMessage::NewTransactions(ref transactions) => { self.client.import_queued_transactions(&transactions); }
SyncMessage::NewTransactions(ref transactions) => { self.client.import_queued_transactions(&transactions); }
_ => {} // ignore other messages
},
NetworkIoMessage::NetworkStarted(ref url) => { self.client.network_started(url); }
_ => {} // ignore other messages _ => {} // ignore other messages
} }
} }
@ -150,7 +134,6 @@ impl IoHandler<NetSyncMessage> for ClientIoHandler {
mod tests { mod tests {
use super::*; use super::*;
use tests::helpers::*; use tests::helpers::*;
use util::network::*;
use devtools::*; use devtools::*;
use client::ClientConfig; use client::ClientConfig;
use std::sync::Arc; use std::sync::Arc;
@ -162,10 +145,8 @@ mod tests {
let service = ClientService::start( let service = ClientService::start(
ClientConfig::default(), ClientConfig::default(),
get_test_spec(), get_test_spec(),
NetworkConfiguration::new_local(),
&temp_path.as_path(), &temp_path.as_path(),
Arc::new(Miner::with_spec(get_test_spec())), Arc::new(Miner::with_spec(get_test_spec())),
false
); );
assert!(service.is_ok()); assert!(service.is_ok());
} }

View File

@ -143,7 +143,7 @@ impl Spec {
} }
/// Get the known knodes of the network in enode format. /// Get the known knodes of the network in enode format.
pub fn nodes(&self) -> &Vec<String> { &self.nodes } pub fn nodes(&self) -> &[String] { &self.nodes }
/// Get the configured Network ID. /// Get the configured Network ID.
pub fn network_id(&self) -> U256 { self.params.network_id } pub fn network_id(&self) -> U256 { self.params.network_id }

View File

@ -40,9 +40,9 @@ impl Into<Vec<u8>> for Bytes {
} }
impl Deref for Bytes { impl Deref for Bytes {
type Target = Vec<u8>; type Target = [u8];
fn deref(&self) -> &Vec<u8> { fn deref(&self) -> &[u8] {
&self.0 &self.0
} }
} }

View File

@ -238,7 +238,7 @@ impl Configuration {
}) })
}).collect(), }).collect(),
Some(_) => Vec::new(), Some(_) => Vec::new(),
None => spec.nodes().clone(), None => spec.nodes().to_owned(),
} }
} }

View File

@ -21,8 +21,8 @@ use self::ansi_term::Style;
use std::time::{Instant, Duration}; use std::time::{Instant, Duration};
use std::sync::RwLock; use std::sync::RwLock;
use std::ops::{Deref, DerefMut}; use std::ops::{Deref, DerefMut};
use ethsync::{EthSync, SyncProvider}; use ethsync::SyncStatus;
use util::{Uint, RwLockable, NetworkService}; use util::{Uint, RwLockable, NetworkConfiguration};
use ethcore::client::*; use ethcore::client::*;
use number_prefix::{binary_prefix, Standalone, Prefixed}; use number_prefix::{binary_prefix, Standalone, Prefixed};
@ -75,7 +75,8 @@ impl Informant {
} }
} }
pub fn tick<Message>(&self, client: &Client, maybe_sync: Option<(&EthSync, &NetworkService<Message>)>) where Message: Send + Sync + Clone + 'static { #[cfg_attr(feature="dev", allow(match_bool))]
pub fn tick(&self, client: &Client, maybe_status: Option<(SyncStatus, NetworkConfiguration)>) {
let elapsed = self.last_tick.unwrapped_read().elapsed(); let elapsed = self.last_tick.unwrapped_read().elapsed();
if elapsed < Duration::from_secs(5) { if elapsed < Duration::from_secs(5) {
return; return;
@ -108,10 +109,8 @@ impl Informant {
paint(Yellow.bold(), format!("{:4}", ((report.transactions_applied - last_report.transactions_applied) * 1000) as u64 / elapsed.as_milliseconds())), paint(Yellow.bold(), format!("{:4}", ((report.transactions_applied - last_report.transactions_applied) * 1000) as u64 / elapsed.as_milliseconds())),
paint(Yellow.bold(), format!("{:3}", ((report.gas_processed - last_report.gas_processed) / From::from(elapsed.as_milliseconds() * 1000)).low_u64())), paint(Yellow.bold(), format!("{:3}", ((report.gas_processed - last_report.gas_processed) / From::from(elapsed.as_milliseconds() * 1000)).low_u64())),
match maybe_sync { match maybe_status {
Some((sync, net)) => { Some((ref sync_info, ref net_config)) => {
let sync_info = sync.status();
let net_config = net.config();
format!("{}/{}/{} peers {} ", format!("{}/{}/{} peers {} ",
paint(Green.bold(), format!("{:2}", sync_info.num_active_peers)), paint(Green.bold(), format!("{:2}", sync_info.num_active_peers)),
paint(Green.bold(), format!("{:2}", sync_info.num_peers)), paint(Green.bold(), format!("{:2}", sync_info.num_peers)),
@ -128,13 +127,9 @@ impl Informant {
paint(Purple.bold(), format!("{:>8}", Informant::format_bytes(report.state_db_mem))), paint(Purple.bold(), format!("{:>8}", Informant::format_bytes(report.state_db_mem))),
paint(Purple.bold(), format!("{:>8}", Informant::format_bytes(cache_info.total()))), paint(Purple.bold(), format!("{:>8}", Informant::format_bytes(cache_info.total()))),
paint(Purple.bold(), format!("{:>8}", Informant::format_bytes(queue_info.mem_used))), paint(Purple.bold(), format!("{:>8}", Informant::format_bytes(queue_info.mem_used))),
match maybe_sync { if let Some((ref sync_info, _)) = maybe_status {
Some((sync, _)) => { format!(" {} sync", paint(Purple.bold(), format!("{:>8}", Informant::format_bytes(sync_info.mem_used))))
let sync_info = sync.status(); } else { String::new() },
format!(" {} sync", paint(Purple.bold(), format!("{:>8}", Informant::format_bytes(sync_info.mem_used))))
}
None => String::new()
},
); );
} }

View File

@ -14,12 +14,12 @@
// You should have received a copy of the GNU General Public License // You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>. // along with Parity. If not, see <http://www.gnu.org/licenses/>.
use std::sync::{Arc, Weak}; use std::sync::Arc;
use ethcore::client::Client; use ethcore::client::Client;
use ethcore::service::{NetSyncMessage, SyncMessage}; use ethcore::service::ClientIoMessage;
use ethsync::EthSync; use ethsync::{EthSync, SyncProvider, ManageNetwork};
use ethcore::account_provider::AccountProvider; use ethcore::account_provider::AccountProvider;
use util::{TimerToken, IoHandler, IoContext, NetworkService, NetworkIoMessage}; use util::{TimerToken, IoHandler, IoContext};
use informant::Informant; use informant::Informant;
@ -30,38 +30,18 @@ pub struct ClientIoHandler {
pub sync: Arc<EthSync>, pub sync: Arc<EthSync>,
pub accounts: Arc<AccountProvider>, pub accounts: Arc<AccountProvider>,
pub info: Informant, pub info: Informant,
pub network: Weak<NetworkService<SyncMessage>>,
} }
impl IoHandler<NetSyncMessage> for ClientIoHandler { impl IoHandler<ClientIoMessage> for ClientIoHandler {
fn initialize(&self, io: &IoContext<NetSyncMessage>) { fn initialize(&self, io: &IoContext<ClientIoMessage>) {
io.register_timer(INFO_TIMER, 5000).expect("Error registering timer"); io.register_timer(INFO_TIMER, 5000).expect("Error registering timer");
} }
fn timeout(&self, _io: &IoContext<NetSyncMessage>, timer: TimerToken) { fn timeout(&self, _io: &IoContext<ClientIoMessage>, timer: TimerToken) {
if let INFO_TIMER = timer { if let INFO_TIMER = timer {
if let Some(net) = self.network.upgrade() { let sync_status = self.sync.status();
self.info.tick(&self.client, Some((&self.sync, &net))); let network_config = self.sync.network_config();
} self.info.tick(&self.client, Some((sync_status, network_config)));
}
}
fn message(&self, _io: &IoContext<NetSyncMessage>, message: &NetSyncMessage) {
match *message {
NetworkIoMessage::User(SyncMessage::StartNetwork) => {
if let Some(network) = self.network.upgrade() {
network.start().unwrap_or_else(|e| warn!("Error starting network: {:?}", e));
EthSync::register(&*network, self.sync.clone()).unwrap_or_else(|e| warn!("Error registering eth protocol handler: {}", e));
}
},
NetworkIoMessage::User(SyncMessage::StopNetwork) => {
if let Some(network) = self.network.upgrade() {
network.stop().unwrap_or_else(|e| warn!("Error stopping network: {:?}", e));
}
},
_ => {/* Ignore other messages */},
} }
} }
} }

View File

@ -46,16 +46,14 @@ extern crate hyper; // for price_info.rs
extern crate json_ipc_server as jsonipc; extern crate json_ipc_server as jsonipc;
extern crate ethcore_ipc_hypervisor as hypervisor; extern crate ethcore_ipc_hypervisor as hypervisor;
#[cfg(feature = "rpc")]
extern crate ethcore_rpc; extern crate ethcore_rpc;
extern crate ethcore_signer;
extern crate ansi_term;
#[cfg(feature = "dapps")] #[cfg(feature = "dapps")]
extern crate ethcore_dapps; extern crate ethcore_dapps;
#[cfg(feature = "ethcore-signer")]
extern crate ethcore_signer;
#[macro_use] #[macro_use]
mod die; mod die;
mod upgrade; mod upgrade;
@ -81,9 +79,10 @@ use std::thread::sleep;
use std::time::Duration; use std::time::Duration;
use rustc_serialize::hex::FromHex; use rustc_serialize::hex::FromHex;
use ctrlc::CtrlC; use ctrlc::CtrlC;
use util::{Lockable, H256, ToPretty, NetworkConfiguration, PayloadInfo, Bytes, UtilError, Colour, Applyable, version, journaldb}; use util::{Lockable, H256, ToPretty, PayloadInfo, Bytes, Colour, Applyable, version, journaldb};
use util::panics::{MayPanic, ForwardPanic, PanicHandler}; use util::panics::{MayPanic, ForwardPanic, PanicHandler};
use ethcore::client::{Mode, BlockID, BlockChainClient, ClientConfig, get_db_path, BlockImportError}; use ethcore::client::{BlockID, BlockChainClient, ClientConfig, get_db_path, BlockImportError,
ChainNotify, Mode};
use ethcore::error::{ImportError}; use ethcore::error::{ImportError};
use ethcore::service::ClientService; use ethcore::service::ClientService;
use ethcore::spec::Spec; use ethcore::spec::Spec;
@ -231,13 +230,11 @@ fn execute_client(conf: Configuration, spec: Spec, client_config: ClientConfig)
miner.set_transactions_limit(conf.args.flag_tx_queue_size); miner.set_transactions_limit(conf.args.flag_tx_queue_size);
// Build client // Build client
let mut service = ClientService::start( let service = ClientService::start(
client_config, client_config,
spec, spec,
net_settings,
Path::new(&conf.path()), Path::new(&conf.path()),
miner.clone(), miner.clone(),
match conf.mode() { Mode::Dark(..) => false, _ => !conf.args.flag_no_network }
).unwrap_or_else(|e| die_with_error("Client", e)); ).unwrap_or_else(|e| die_with_error("Client", e));
panic_handler.forward_from(&service); panic_handler.forward_from(&service);
@ -247,8 +244,14 @@ fn execute_client(conf: Configuration, spec: Spec, client_config: ClientConfig)
let network_settings = Arc::new(conf.network_settings()); let network_settings = Arc::new(conf.network_settings());
// Sync // Sync
let sync = EthSync::new(sync_config, client.clone()); let sync = EthSync::new(sync_config, client.clone(), net_settings)
EthSync::register(&*service.network(), sync.clone()).unwrap_or_else(|e| die_with_error("Error registering eth protocol handler", UtilError::from(e).into())); .unwrap_or_else(|e| die_with_error("Sync", ethcore::error::Error::Util(e)));
service.set_notify(&(sync.clone() as Arc<ChainNotify>));
// if network is active by default
if match conf.mode() { Mode::Dark(..) => false, _ => !conf.args.flag_no_network } {
sync.start();
}
let deps_for_rpc_apis = Arc::new(rpc_apis::Dependencies { let deps_for_rpc_apis = Arc::new(rpc_apis::Dependencies {
signer_port: conf.signer_port(), signer_port: conf.signer_port(),
@ -261,7 +264,7 @@ fn execute_client(conf: Configuration, spec: Spec, client_config: ClientConfig)
logger: logger.clone(), logger: logger.clone(),
settings: network_settings.clone(), settings: network_settings.clone(),
allow_pending_receipt_query: !conf.args.flag_geth, allow_pending_receipt_query: !conf.args.flag_geth,
net_service: service.network(), net_service: sync.clone(),
}); });
let dependencies = rpc::Dependencies { let dependencies = rpc::Dependencies {
@ -311,7 +314,6 @@ fn execute_client(conf: Configuration, spec: Spec, client_config: ClientConfig)
info: Informant::new(conf.have_color()), info: Informant::new(conf.have_color()),
sync: sync.clone(), sync: sync.clone(),
accounts: account_service.clone(), accounts: account_service.clone(),
network: Arc::downgrade(&service.network()),
}); });
service.register_io_handler(io_handler).expect("Error registering IO handler"); service.register_io_handler(io_handler).expect("Error registering IO handler");
@ -345,24 +347,11 @@ fn execute_export(conf: Configuration) {
unsafe { ::fdlimit::raise_fd_limit(); } unsafe { ::fdlimit::raise_fd_limit(); }
let spec = conf.spec(); let spec = conf.spec();
let net_settings = NetworkConfiguration {
config_path: None,
listen_address: None,
public_address: None,
udp_port: None,
nat_enabled: false,
discovery_enabled: false,
boot_nodes: Vec::new(),
use_secret: None,
ideal_peers: 0,
reserved_nodes: Vec::new(),
non_reserved_mode: ::util::network::NonReservedPeerMode::Accept,
};
let client_config = conf.client_config(&spec); let client_config = conf.client_config(&spec);
// Build client // Build client
let service = ClientService::start( let service = ClientService::start(
client_config, spec, net_settings, Path::new(&conf.path()), Arc::new(Miner::with_spec(conf.spec())), false client_config, spec, Path::new(&conf.path()), Arc::new(Miner::with_spec(conf.spec()))
).unwrap_or_else(|e| die_with_error("Client", e)); ).unwrap_or_else(|e| die_with_error("Client", e));
panic_handler.forward_from(&service); panic_handler.forward_from(&service);
@ -419,24 +408,11 @@ fn execute_import(conf: Configuration) {
unsafe { ::fdlimit::raise_fd_limit(); } unsafe { ::fdlimit::raise_fd_limit(); }
let spec = conf.spec(); let spec = conf.spec();
let net_settings = NetworkConfiguration {
config_path: None,
listen_address: None,
public_address: None,
udp_port: None,
nat_enabled: false,
discovery_enabled: false,
boot_nodes: Vec::new(),
use_secret: None,
ideal_peers: 0,
reserved_nodes: Vec::new(),
non_reserved_mode: ::util::network::NonReservedPeerMode::Accept,
};
let client_config = conf.client_config(&spec); let client_config = conf.client_config(&spec);
// Build client // Build client
let service = ClientService::start( let service = ClientService::start(
client_config, spec, net_settings, Path::new(&conf.path()), Arc::new(Miner::with_spec(conf.spec())), false client_config, spec, Path::new(&conf.path()), Arc::new(Miner::with_spec(conf.spec()))
).unwrap_or_else(|e| die_with_error("Client", e)); ).unwrap_or_else(|e| die_with_error("Client", e));
panic_handler.forward_from(&service); panic_handler.forward_from(&service);
@ -485,7 +461,7 @@ fn execute_import(conf: Configuration) {
Err(BlockImportError::Import(ImportError::AlreadyInChain)) => { trace!("Skipping block already in chain."); } Err(BlockImportError::Import(ImportError::AlreadyInChain)) => { trace!("Skipping block already in chain."); }
Err(e) => die!("Cannot import block: {:?}", e) Err(e) => die!("Cannot import block: {:?}", e)
} }
informant.tick::<&'static ()>(client.deref(), None); informant.tick(client.deref(), None);
}; };
match format { match format {

View File

@ -24,12 +24,8 @@ use jsonipc;
use rpc_apis; use rpc_apis;
use std::fmt; use std::fmt;
#[cfg(feature = "rpc")]
pub use ethcore_rpc::Server as RpcServer; pub use ethcore_rpc::Server as RpcServer;
#[cfg(feature = "rpc")]
use ethcore_rpc::{RpcServerError, RpcServer as Server}; use ethcore_rpc::{RpcServerError, RpcServer as Server};
#[cfg(not(feature = "rpc"))]
pub struct RpcServer;
pub struct HttpConfiguration { pub struct HttpConfiguration {
pub enabled: bool, pub enabled: bool,
@ -79,17 +75,6 @@ fn setup_rpc_server(apis: Vec<&str>, deps: &Dependencies) -> Server {
rpc_apis::setup_rpc(server, deps.apis.clone(), rpc_apis::ApiSet::List(apis)) rpc_apis::setup_rpc(server, deps.apis.clone(), rpc_apis::ApiSet::List(apis))
} }
#[cfg(not(feature = "rpc"))]
pub fn setup_http_rpc_server(
_deps: &Dependencies,
_url: &SocketAddr,
_cors_domain: Vec<String>,
_apis: Vec<&str>,
) -> ! {
die!("Your Parity version has been compiled without JSON-RPC support.")
}
#[cfg(feature = "rpc")]
pub fn setup_http_rpc_server( pub fn setup_http_rpc_server(
dependencies: &Dependencies, dependencies: &Dependencies,
url: &SocketAddr, url: &SocketAddr,
@ -111,18 +96,12 @@ pub fn setup_http_rpc_server(
} }
} }
#[cfg(not(feature = "rpc"))]
pub fn setup_ipc_rpc_server(_dependencies: &Dependencies, _addr: &str, _apis: Vec<&str>) -> ! {
die!("Your Parity version has been compiled without JSON-RPC support.")
}
pub fn new_ipc(conf: IpcConfiguration, deps: &Dependencies) -> Option<jsonipc::Server> { pub fn new_ipc(conf: IpcConfiguration, deps: &Dependencies) -> Option<jsonipc::Server> {
if !conf.enabled { return None; } if !conf.enabled { return None; }
let apis = conf.apis.split(',').collect(); let apis = conf.apis.split(',').collect();
Some(setup_ipc_rpc_server(deps, &conf.socket_addr, apis)) Some(setup_ipc_rpc_server(deps, &conf.socket_addr, apis))
} }
#[cfg(feature = "rpc")]
pub fn setup_ipc_rpc_server(dependencies: &Dependencies, addr: &str, apis: Vec<&str>) -> jsonipc::Server { pub fn setup_ipc_rpc_server(dependencies: &Dependencies, addr: &str, apis: Vec<&str>) -> jsonipc::Server {
let server = setup_rpc_server(apis, dependencies); let server = setup_rpc_server(apis, dependencies);
match server.start_ipc(addr) { match server.start_ipc(addr) {

View File

@ -18,21 +18,15 @@ use std::collections::BTreeMap;
use std::str::FromStr; use std::str::FromStr;
use std::sync::Arc; use std::sync::Arc;
use ethsync::EthSync; use ethsync::{EthSync, ManageNetwork};
use ethcore::miner::{Miner, ExternalMiner}; use ethcore::miner::{Miner, ExternalMiner};
use ethcore::client::Client; use ethcore::client::Client;
use util::RotatingLogger; use util::RotatingLogger;
use ethcore::account_provider::AccountProvider; use ethcore::account_provider::AccountProvider;
use util::network_settings::NetworkSettings; use util::network_settings::NetworkSettings;
use util::network::NetworkService;
#[cfg(feature="rpc")]
pub use ethcore_rpc::ConfirmationsQueue; pub use ethcore_rpc::ConfirmationsQueue;
#[cfg(not(feature="rpc"))]
#[derive(Default)]
pub struct ConfirmationsQueue;
#[cfg(feature="rpc")]
use ethcore_rpc::Extendable; use ethcore_rpc::Extendable;
pub enum Api { pub enum Api {
@ -89,7 +83,7 @@ pub struct Dependencies {
pub logger: Arc<RotatingLogger>, pub logger: Arc<RotatingLogger>,
pub settings: Arc<NetworkSettings>, pub settings: Arc<NetworkSettings>,
pub allow_pending_receipt_query: bool, pub allow_pending_receipt_query: bool,
pub net_service: Arc<NetworkService<::ethcore::service::SyncMessage>>, pub net_service: Arc<ManageNetwork>,
} }
fn to_modules(apis: &[Api]) -> BTreeMap<String, String> { fn to_modules(apis: &[Api]) -> BTreeMap<String, String> {

View File

@ -15,23 +15,19 @@
// along with Parity. If not, see <http://www.gnu.org/licenses/>. // along with Parity. If not, see <http://www.gnu.org/licenses/>.
use std::io; use std::io;
use std::path::PathBuf;
use std::sync::Arc; use std::sync::Arc;
use util::{Colour, Applyable}; use std::path::PathBuf;
use util::panics::{PanicHandler, ForwardPanic}; use ansi_term::Colour;
use util::panics::{ForwardPanic, PanicHandler};
use util::path::restrict_permissions_owner; use util::path::restrict_permissions_owner;
use die::*; use util::Applyable;
use rpc_apis; use rpc_apis;
const CODES_FILENAME: &'static str = "authcodes";
#[cfg(feature = "ethcore-signer")]
use ethcore_signer as signer; use ethcore_signer as signer;
#[cfg(feature = "ethcore-signer")] use die::*;
pub use ethcore_signer::Server as SignerServer; pub use ethcore_signer::Server as SignerServer;
#[cfg(not(feature = "ethcore-signer"))] const CODES_FILENAME: &'static str = "authcodes";
pub struct SignerServer;
pub struct Configuration { pub struct Configuration {
pub enabled: bool, pub enabled: bool,
@ -59,8 +55,6 @@ fn codes_path(path: String) -> PathBuf {
p p
} }
#[cfg(feature = "ethcore-signer")]
pub fn new_token(path: String) -> io::Result<()> { pub fn new_token(path: String) -> io::Result<()> {
let path = codes_path(path); let path = codes_path(path);
let mut codes = try!(signer::AuthCodes::from_file(&path)); let mut codes = try!(signer::AuthCodes::from_file(&path));
@ -70,7 +64,6 @@ pub fn new_token(path: String) -> io::Result<()> {
Ok(()) Ok(())
} }
#[cfg(feature = "ethcore-signer")]
fn do_start(conf: Configuration, deps: Dependencies) -> SignerServer { fn do_start(conf: Configuration, deps: Dependencies) -> SignerServer {
let addr = format!("127.0.0.1:{}", conf.port).parse().unwrap_or_else(|_| { let addr = format!("127.0.0.1:{}", conf.port).parse().unwrap_or_else(|_| {
die!("Invalid port specified: {}", conf.port) die!("Invalid port specified: {}", conf.port)
@ -95,13 +88,4 @@ fn do_start(conf: Configuration, deps: Dependencies) -> SignerServer {
} }
} }
#[cfg(not(feature = "ethcore-signer"))]
fn do_start(_conf: Configuration) -> ! {
die!("Your Parity version has been compiled without Trusted Signer support.")
}
#[cfg(not(feature = "ethcore-signer"))]
pub fn new_token(_path: String) -> ! {
die!("Your Parity version has been compiled without Trusted Signer support.")
}

View File

@ -508,7 +508,7 @@ impl<C, S, M, EM> Eth for EthClient<C, S, M, EM> where
fn compilers(&self, params: Params) -> Result<Value, Error> { fn compilers(&self, params: Params) -> Result<Value, Error> {
try!(self.active()); try!(self.active());
match params { match params {
Params::None => to_value(&vec![] as &Vec<String>), Params::None => to_value(&(&[] as &[String])),
_ => Err(Error::invalid_params()) _ => Err(Error::invalid_params())
} }
} }

View File

@ -19,26 +19,26 @@ use std::sync::{Arc, Weak};
use jsonrpc_core::*; use jsonrpc_core::*;
use ethcore::miner::MinerService; use ethcore::miner::MinerService;
use ethcore::client::MiningBlockChainClient; use ethcore::client::MiningBlockChainClient;
use ethcore::service::SyncMessage; use ethsync::ManageNetwork;
use util::network::{NetworkService, NonReservedPeerMode}; use util::network::NonReservedPeerMode;
use v1::traits::EthcoreSet; use v1::traits::EthcoreSet;
use v1::types::{Bytes, H160, U256}; use v1::types::{Bytes, H160, U256};
/// Ethcore-specific rpc interface for operations altering the settings. /// Ethcore-specific rpc interface for operations altering the settings.
pub struct EthcoreSetClient<C, M> where pub struct EthcoreSetClient<C, M> where
C: MiningBlockChainClient, C: MiningBlockChainClient,
M: MinerService { M: MinerService
{
client: Weak<C>, client: Weak<C>,
miner: Weak<M>, miner: Weak<M>,
net: Weak<NetworkService<SyncMessage>>, net: Weak<ManageNetwork>,
} }
impl<C, M> EthcoreSetClient<C, M> where impl<C, M> EthcoreSetClient<C, M> where
C: MiningBlockChainClient, C: MiningBlockChainClient,
M: MinerService { M: MinerService {
/// Creates new `EthcoreSetClient`. /// Creates new `EthcoreSetClient`.
pub fn new(client: &Arc<C>, miner: &Arc<M>, net: &Arc<NetworkService<SyncMessage>>) -> Self { pub fn new(client: &Arc<C>, miner: &Arc<M>, net: &Arc<ManageNetwork>) -> Self {
EthcoreSetClient { EthcoreSetClient {
client: Arc::downgrade(client), client: Arc::downgrade(client),
miner: Arc::downgrade(miner), miner: Arc::downgrade(miner),
@ -144,4 +144,14 @@ impl<C, M> EthcoreSet for EthcoreSetClient<C, M> where
take_weak!(self.net).set_non_reserved_mode(NonReservedPeerMode::Accept); take_weak!(self.net).set_non_reserved_mode(NonReservedPeerMode::Accept);
to_value(&true) to_value(&true)
} }
fn start_network(&self, _: Params) -> Result<Value, Error> {
take_weak!(self.net).start_network();
Ok(Value::Bool(true))
}
fn stop_network(&self, _: Params) -> Result<Value, Error> {
take_weak!(self.net).stop_network();
Ok(Value::Bool(true))
}
} }

View File

@ -48,13 +48,4 @@ impl<S> Net for NetClient<S> where S: SyncProvider + 'static {
Ok(Value::Bool(true)) Ok(Value::Bool(true))
} }
fn start_network(&self, _: Params) -> Result<Value, Error> {
take_weak!(self.sync).start_network();
Ok(Value::Bool(true))
}
fn stop_network(&self, _: Params) -> Result<Value, Error> {
take_weak!(self.sync).stop_network();
Ok(Value::Bool(true))
}
} }

View File

@ -59,11 +59,5 @@ impl SyncProvider for TestSyncProvider {
fn status(&self) -> SyncStatus { fn status(&self) -> SyncStatus {
self.status.unwrapped_read().clone() self.status.unwrapped_read().clone()
} }
fn start_network(&self) {
}
fn stop_network(&self) {
}
} }

View File

@ -19,12 +19,12 @@ use std::str::FromStr;
use jsonrpc_core::IoHandler; use jsonrpc_core::IoHandler;
use v1::{EthcoreSet, EthcoreSetClient}; use v1::{EthcoreSet, EthcoreSetClient};
use ethcore::miner::MinerService; use ethcore::miner::MinerService;
use ethcore::service::SyncMessage;
use ethcore::client::TestBlockChainClient; use ethcore::client::TestBlockChainClient;
use v1::tests::helpers::TestMinerService; use v1::tests::helpers::TestMinerService;
use util::numbers::*; use util::numbers::*;
use util::network::{NetworkConfiguration, NetworkService};
use rustc_serialize::hex::FromHex; use rustc_serialize::hex::FromHex;
use super::manage_network::TestManageNetwork;
use ethsync::ManageNetwork;
fn miner_service() -> Arc<TestMinerService> { fn miner_service() -> Arc<TestMinerService> {
Arc::new(TestMinerService::default()) Arc::new(TestMinerService::default())
@ -34,12 +34,12 @@ fn client_service() -> Arc<TestBlockChainClient> {
Arc::new(TestBlockChainClient::default()) Arc::new(TestBlockChainClient::default())
} }
fn network_service() -> Arc<NetworkService<SyncMessage>> { fn network_service() -> Arc<TestManageNetwork> {
Arc::new(NetworkService::new(NetworkConfiguration::new()).unwrap()) Arc::new(TestManageNetwork)
} }
fn ethcore_set_client(client: &Arc<TestBlockChainClient>, miner: &Arc<TestMinerService>, net: &Arc<NetworkService<SyncMessage>>) -> EthcoreSetClient<TestBlockChainClient, TestMinerService> { fn ethcore_set_client(client: &Arc<TestBlockChainClient>, miner: &Arc<TestMinerService>, net: &Arc<TestManageNetwork>) -> EthcoreSetClient<TestBlockChainClient, TestMinerService> {
EthcoreSetClient::new(client, miner, net) EthcoreSetClient::new(client, miner, &(net.clone() as Arc<ManageNetwork>))
} }
#[test] #[test]

View File

@ -0,0 +1,30 @@
// Copyright 2015, 2016 Ethcore (UK) Ltd.
// This file is part of Parity.
// Parity 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.
// Parity 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 Parity. If not, see <http://www.gnu.org/licenses/>.
use ethsync::ManageNetwork;
use util::network::NetworkConfiguration;
pub struct TestManageNetwork;
// TODO: rob, gavin (originally introduced this functions) - proper tests and test state
impl ManageNetwork for TestManageNetwork {
fn set_non_reserved_mode(&self, _mode: ::util::network::NonReservedPeerMode) {}
fn remove_reserved_peer(&self, _peer: &str) -> Result<(), String> { Ok(()) }
fn add_reserved_peer(&self, _peer: &str) -> Result<(), String> { Ok(()) }
fn start_network(&self) {}
fn stop_network(&self) {}
fn network_config(&self) -> NetworkConfiguration { NetworkConfiguration::new_local() }
}

View File

@ -26,3 +26,4 @@ mod personal_signer;
mod ethcore; mod ethcore;
mod ethcore_set; mod ethcore_set;
mod rpc; mod rpc;
mod manage_network;

View File

@ -55,6 +55,12 @@ pub trait EthcoreSet: Sized + Send + Sync + 'static {
/// Accept non-reserved peers (default behavior) /// Accept non-reserved peers (default behavior)
fn accept_non_reserved_peers(&self, _: Params) -> Result<Value, Error>; fn accept_non_reserved_peers(&self, _: Params) -> Result<Value, Error>;
/// Start the network.
fn start_network(&self, _: Params) -> Result<Value, Error>;
/// Stop the network.
fn stop_network(&self, _: Params) -> Result<Value, Error>;
/// Should be used to convert object to io delegate. /// Should be used to convert object to io delegate.
fn to_delegate(self) -> IoDelegate<Self> { fn to_delegate(self) -> IoDelegate<Self> {
let mut delegate = IoDelegate::new(Arc::new(self)); let mut delegate = IoDelegate::new(Arc::new(self));

View File

@ -30,12 +30,6 @@ pub trait Net: Sized + Send + Sync + 'static {
/// Otherwise false. /// Otherwise false.
fn is_listening(&self, _: Params) -> Result<Value, Error>; fn is_listening(&self, _: Params) -> Result<Value, Error>;
/// Start the network.
fn start_network(&self, _: Params) -> Result<Value, Error>;
/// Stop the network.
fn stop_network(&self, _: Params) -> Result<Value, Error>;
/// Should be used to convert object to io delegate. /// Should be used to convert object to io delegate.
fn to_delegate(self) -> IoDelegate<Self> { fn to_delegate(self) -> IoDelegate<Self> {
let mut delegate = IoDelegate::new(Arc::new(self)); let mut delegate = IoDelegate::new(Arc::new(self));

View File

@ -25,11 +25,11 @@ use std::str::FromStr;
use jsonrpc_core::IoHandler; use jsonrpc_core::IoHandler;
use util::H256; use util::H256;
fn origin_is_allowed(self_origin: &str, header: Option<&Vec<u8>>) -> bool { fn origin_is_allowed(self_origin: &str, header: Option<&[u8]>) -> bool {
match header { match header {
None => false, None => false,
Some(h) => { Some(h) => {
let v = String::from_utf8(h.clone()).ok(); let v = String::from_utf8(h.to_owned()).ok();
match v { match v {
Some(ref origin) if origin.starts_with("chrome-extension://") => true, Some(ref origin) if origin.starts_with("chrome-extension://") => true,
Some(ref origin) if origin.starts_with(self_origin) => true, Some(ref origin) if origin.starts_with(self_origin) => true,
@ -84,8 +84,8 @@ pub struct Session {
impl ws::Handler for Session { impl ws::Handler for Session {
fn on_request(&mut self, req: &ws::Request) -> ws::Result<(ws::Response)> { fn on_request(&mut self, req: &ws::Request) -> ws::Result<(ws::Response)> {
let origin = req.header("origin").or_else(|| req.header("Origin")); let origin = req.header("origin").or_else(|| req.header("Origin")).map(|x| &x[..]);
let host = req.header("host").or_else(|| req.header("Host")); let host = req.header("host").or_else(|| req.header("Host")).map(|x| &x[..]);
// Check request origin and host header. // Check request origin and host header.
if !origin_is_allowed(&self.self_origin, origin) && !(origin.is_none() && origin_is_allowed(&self.self_origin, host)) { if !origin_is_allowed(&self.self_origin, origin) && !(origin.is_none() && origin_is_allowed(&self.self_origin, host)) {

View File

@ -16,7 +16,6 @@
use util::{NetworkContext, PeerId, PacketId,}; use util::{NetworkContext, PeerId, PacketId,};
use util::error::UtilError; use util::error::UtilError;
use ethcore::service::SyncMessage;
use ethcore::client::BlockChainClient; use ethcore::client::BlockChainClient;
/// IO interface for the syning handler. /// IO interface for the syning handler.
@ -47,13 +46,13 @@ pub trait SyncIo {
/// Wraps `NetworkContext` and the blockchain client /// Wraps `NetworkContext` and the blockchain client
pub struct NetSyncIo<'s, 'h> where 'h: 's { pub struct NetSyncIo<'s, 'h> where 'h: 's {
network: &'s NetworkContext<'h, SyncMessage>, network: &'s NetworkContext<'h>,
chain: &'s BlockChainClient chain: &'s BlockChainClient
} }
impl<'s, 'h> NetSyncIo<'s, 'h> { impl<'s, 'h> NetSyncIo<'s, 'h> {
/// Creates a new instance from the `NetworkContext` and the blockchain client reference. /// Creates a new instance from the `NetworkContext` and the blockchain client reference.
pub fn new(network: &'s NetworkContext<'h, SyncMessage>, chain: &'s BlockChainClient) -> NetSyncIo<'s, 'h> { pub fn new(network: &'s NetworkContext<'h>, chain: &'s BlockChainClient) -> NetSyncIo<'s, 'h> {
NetSyncIo { NetSyncIo {
network: network, network: network,
chain: chain, chain: chain,

View File

@ -34,15 +34,14 @@
//! extern crate ethsync; //! extern crate ethsync;
//! use std::env; //! use std::env;
//! use std::sync::Arc; //! use std::sync::Arc;
//! use util::network::{NetworkService, NetworkConfiguration}; //! use util::network::{NetworkConfiguration};
//! use util::io::IoChannel;
//! use ethcore::client::{Client, ClientConfig}; //! use ethcore::client::{Client, ClientConfig};
//! use ethsync::{EthSync, SyncConfig}; //! use ethsync::{EthSync, SyncConfig, ManageNetwork};
//! use ethcore::ethereum; //! use ethcore::ethereum;
//! use ethcore::miner::{GasPricer, Miner}; //! use ethcore::miner::{GasPricer, Miner};
//! //!
//! fn main() { //! fn main() {
//! let mut service = NetworkService::new(NetworkConfiguration::new()).unwrap();
//! service.start().unwrap();
//! let dir = env::temp_dir(); //! let dir = env::temp_dir();
//! let miner = Miner::new( //! let miner = Miner::new(
//! Default::default(), //! Default::default(),
@ -55,10 +54,10 @@
//! ethereum::new_frontier(), //! ethereum::new_frontier(),
//! &dir, //! &dir,
//! miner, //! miner,
//! service.io().channel() //! IoChannel::disconnected()
//! ).unwrap(); //! ).unwrap();
//! let sync = EthSync::new(SyncConfig::default(), client); //! let sync = EthSync::new(SyncConfig::default(), client, NetworkConfiguration::new()).unwrap();
//! EthSync::register(&mut service, sync); //! sync.start_network();
//! } //! }
//! ``` //! ```
@ -75,13 +74,10 @@ extern crate heapsize;
use std::ops::*; use std::ops::*;
use std::sync::*; use std::sync::*;
use util::network::{NetworkProtocolHandler, NetworkService, NetworkContext, PeerId}; use util::network::{NetworkProtocolHandler, NetworkService, NetworkContext, PeerId, NetworkConfiguration};
use util::{TimerToken, U256, RwLockable}; use util::{TimerToken, U256, H256, RwLockable, UtilError};
use ethcore::client::Client; use ethcore::client::{Client, ChainNotify};
use ethcore::service::{SyncMessage, NetSyncMessage};
use io::NetSyncIo; use io::NetSyncIo;
use util::io::IoChannel;
use util::{NetworkIoMessage, NetworkError};
use chain::ChainSync; use chain::ChainSync;
mod chain; mod chain;
@ -91,6 +87,9 @@ mod io;
#[cfg(test)] #[cfg(test)]
mod tests; mod tests;
/// Ethereum sync protocol
pub const ETH_PROTOCOL: &'static str = "eth";
/// Sync configuration /// Sync configuration
pub struct SyncConfig { pub struct SyncConfig {
/// Max blocks to download ahead /// Max blocks to download ahead
@ -112,99 +111,142 @@ impl Default for SyncConfig {
pub trait SyncProvider: Send + Sync { pub trait SyncProvider: Send + Sync {
/// Get sync status /// Get sync status
fn status(&self) -> SyncStatus; fn status(&self) -> SyncStatus;
/// Start the network
fn start_network(&self);
/// Stop the network
fn stop_network(&self);
} }
/// Ethereum network protocol handler /// Ethereum network protocol handler
pub struct EthSync { pub struct EthSync {
/// Shared blockchain client. TODO: this should evetually become an IPC endpoint /// Network service
chain: Arc<Client>, network: NetworkService,
/// Sync strategy /// Protocol handler
sync: RwLock<ChainSync>, handler: Arc<SyncProtocolHandler>,
/// IO communication chnnel.
io_channel: RwLock<IoChannel<NetSyncMessage>>,
} }
pub use self::chain::{SyncStatus, SyncState}; pub use self::chain::{SyncStatus, SyncState};
impl EthSync { impl EthSync {
/// Creates and register protocol with the network service /// Creates and register protocol with the network service
pub fn new(config: SyncConfig, chain: Arc<Client>) -> Arc<EthSync> { pub fn new(config: SyncConfig, chain: Arc<Client>, network_config: NetworkConfiguration) -> Result<Arc<EthSync>, UtilError> {
let sync = ChainSync::new(config, chain.deref()); let chain_sync = ChainSync::new(config, chain.deref());
Arc::new(EthSync { let service = try!(NetworkService::new(network_config));
chain: chain, let sync = Arc::new(EthSync{
sync: RwLock::new(sync), network: service,
io_channel: RwLock::new(IoChannel::disconnected()), handler: Arc::new(SyncProtocolHandler { sync: RwLock::new(chain_sync), chain: chain }),
}) });
}
/// Register protocol with the network service Ok(sync)
pub fn register(service: &NetworkService<SyncMessage>, sync: Arc<EthSync>) -> Result<(), NetworkError> {
service.register_protocol(sync.clone(), "eth", &[62u8, 63u8])
}
/// Stop sync
pub fn stop(&mut self, io: &mut NetworkContext<SyncMessage>) {
self.sync.unwrapped_write().abort(&mut NetSyncIo::new(io, self.chain.deref()));
}
/// Restart sync
pub fn restart(&mut self, io: &mut NetworkContext<SyncMessage>) {
self.sync.unwrapped_write().restart(&mut NetSyncIo::new(io, self.chain.deref()));
} }
} }
impl SyncProvider for EthSync { impl SyncProvider for EthSync {
/// Get sync status /// Get sync status
fn status(&self) -> SyncStatus { fn status(&self) -> SyncStatus {
self.sync.unwrapped_read().status() self.handler.sync.unwrapped_read().status()
}
fn start_network(&self) {
self.io_channel.unwrapped_read().send(NetworkIoMessage::User(SyncMessage::StartNetwork))
.unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e));
}
fn stop_network(&self) {
self.io_channel.unwrapped_read().send(NetworkIoMessage::User(SyncMessage::StopNetwork))
.unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e));
} }
} }
impl NetworkProtocolHandler<SyncMessage> for EthSync { struct SyncProtocolHandler {
fn initialize(&self, io: &NetworkContext<SyncMessage>) { /// Shared blockchain client. TODO: this should evetually become an IPC endpoint
chain: Arc<Client>,
/// Sync strategy
sync: RwLock<ChainSync>,
}
impl NetworkProtocolHandler for SyncProtocolHandler {
fn initialize(&self, io: &NetworkContext) {
io.register_timer(0, 1000).expect("Error registering sync timer"); io.register_timer(0, 1000).expect("Error registering sync timer");
*self.io_channel.unwrapped_write() = io.io_channel();
} }
fn read(&self, io: &NetworkContext<SyncMessage>, peer: &PeerId, packet_id: u8, data: &[u8]) { fn read(&self, io: &NetworkContext, peer: &PeerId, packet_id: u8, data: &[u8]) {
ChainSync::dispatch_packet(&self.sync, &mut NetSyncIo::new(io, self.chain.deref()) , *peer, packet_id, data); ChainSync::dispatch_packet(&self.sync, &mut NetSyncIo::new(io, self.chain.deref()), *peer, packet_id, data);
} }
fn connected(&self, io: &NetworkContext<SyncMessage>, peer: &PeerId) { fn connected(&self, io: &NetworkContext, peer: &PeerId) {
self.sync.unwrapped_write().on_peer_connected(&mut NetSyncIo::new(io, self.chain.deref()), *peer); self.sync.unwrapped_write().on_peer_connected(&mut NetSyncIo::new(io, self.chain.deref()), *peer);
} }
fn disconnected(&self, io: &NetworkContext<SyncMessage>, peer: &PeerId) { fn disconnected(&self, io: &NetworkContext, peer: &PeerId) {
self.sync.unwrapped_write().on_peer_aborting(&mut NetSyncIo::new(io, self.chain.deref()), *peer); self.sync.unwrapped_write().on_peer_aborting(&mut NetSyncIo::new(io, self.chain.deref()), *peer);
} }
fn timeout(&self, io: &NetworkContext<SyncMessage>, _timer: TimerToken) { fn timeout(&self, io: &NetworkContext, _timer: TimerToken) {
self.sync.unwrapped_write().maintain_peers(&mut NetSyncIo::new(io, self.chain.deref())); self.sync.unwrapped_write().maintain_peers(&mut NetSyncIo::new(io, self.chain.deref()));
self.sync.unwrapped_write().maintain_sync(&mut NetSyncIo::new(io, self.chain.deref())); self.sync.unwrapped_write().maintain_sync(&mut NetSyncIo::new(io, self.chain.deref()));
} }
}
#[cfg_attr(feature="dev", allow(single_match))] impl ChainNotify for EthSync {
fn message(&self, io: &NetworkContext<SyncMessage>, message: &SyncMessage) { fn new_blocks(&self,
match *message { imported: Vec<H256>,
SyncMessage::NewChainBlocks { ref imported, ref invalid, ref enacted, ref retracted, ref sealed } => { invalid: Vec<H256>,
let mut sync_io = NetSyncIo::new(io, self.chain.deref()); enacted: Vec<H256>,
self.sync.unwrapped_write().chain_new_blocks(&mut sync_io, imported, invalid, enacted, retracted, sealed); retracted: Vec<H256>,
}, sealed: Vec<H256>)
_ => {/* Ignore other messages */}, {
} self.network.with_context(ETH_PROTOCOL, |context| {
let mut sync_io = NetSyncIo::new(context, self.handler.chain.deref());
self.handler.sync.unwrapped_write().chain_new_blocks(
&mut sync_io,
&imported,
&invalid,
&enacted,
&retracted,
&sealed);
});
}
fn start(&self) {
self.network.start().unwrap_or_else(|e| warn!("Error starting network: {:?}", e));
self.network.register_protocol(self.handler.clone(), ETH_PROTOCOL, &[62u8, 63u8])
.unwrap_or_else(|e| warn!("Error registering ethereum protocol: {:?}", e));
}
fn stop(&self) {
self.network.stop().unwrap_or_else(|e| warn!("Error stopping network: {:?}", e));
}
}
/// Trait for managing network
pub trait ManageNetwork : Send + Sync {
/// Set mode for reserved peers (allow/deny peers that are unreserved)
fn set_non_reserved_mode(&self, mode: ::util::network::NonReservedPeerMode);
/// Remove reservation for the peer
fn remove_reserved_peer(&self, peer: &str) -> Result<(), String>;
/// Add reserved peer
fn add_reserved_peer(&self, peer: &str) -> Result<(), String>;
/// Start network
fn start_network(&self);
/// Stop network
fn stop_network(&self);
/// Query the current configuration of the network
fn network_config(&self) -> NetworkConfiguration;
}
impl ManageNetwork for EthSync {
fn set_non_reserved_mode(&self, mode: ::util::network::NonReservedPeerMode) {
self.network.set_non_reserved_mode(mode);
}
fn remove_reserved_peer(&self, peer: &str) -> Result<(), String> {
self.network.remove_reserved_peer(peer).map_err(|e| format!("{:?}", e))
}
fn add_reserved_peer(&self, peer: &str) -> Result<(), String> {
self.network.add_reserved_peer(peer).map_err(|e| format!("{:?}", e))
}
fn start_network(&self) {
self.start();
}
fn stop_network(&self) {
self.network.with_context(ETH_PROTOCOL, |context| {
let mut sync_io = NetSyncIo::new(context, self.handler.chain.deref());
self.handler.sync.unwrapped_write().abort(&mut sync_io);
});
self.stop();
}
fn network_config(&self) -> NetworkConfiguration {
self.network.config().clone()
} }
} }

View File

@ -242,7 +242,7 @@ pub enum FromBytesError {
} }
/// Value that can be serialized from bytes array /// Value that can be serialized from bytes array
pub trait FromRawBytes : Sized { pub trait FromRawBytes: Sized {
/// function that will instantiate and initialize object from slice /// function that will instantiate and initialize object from slice
fn from_bytes(d: &[u8]) -> Result<Self, FromBytesError>; fn from_bytes(d: &[u8]) -> Result<Self, FromBytesError>;
} }
@ -255,7 +255,7 @@ impl<T> FromRawBytes for T where T: FixedHash {
Ordering::Equal => () Ordering::Equal => ()
}; };
let mut res: Self = unsafe { mem::uninitialized() }; let mut res = T::zero();
res.copy_raw(bytes); res.copy_raw(bytes);
Ok(res) Ok(res)
} }
@ -271,7 +271,7 @@ macro_rules! sized_binary_map {
::std::cmp::Ordering::Greater => return Err(FromBytesError::TooLong), ::std::cmp::Ordering::Greater => return Err(FromBytesError::TooLong),
::std::cmp::Ordering::Equal => () ::std::cmp::Ordering::Equal => ()
}; };
let mut res: Self = unsafe { ::std::mem::uninitialized() }; let mut res: Self = 0;
res.copy_raw(bytes); res.copy_raw(bytes);
Ok(res) Ok(res)
} }
@ -298,7 +298,7 @@ sized_binary_map!(u32);
sized_binary_map!(u64); sized_binary_map!(u64);
/// Value that can be serialized from variable-length byte array /// Value that can be serialized from variable-length byte array
pub trait FromRawBytesVariable : Sized { pub trait FromRawBytesVariable: Sized {
/// Create value from slice /// Create value from slice
fn from_bytes_variable(bytes: &[u8]) -> Result<Self, FromBytesError>; fn from_bytes_variable(bytes: &[u8]) -> Result<Self, FromBytesError>;
} }
@ -326,7 +326,7 @@ impl<T> FromRawBytesVariable for Vec<T> where T: FromRawBytes {
let size_of_t = mem::size_of::<T>(); let size_of_t = mem::size_of::<T>();
let length_in_chunks = bytes.len() / size_of_t; let length_in_chunks = bytes.len() / size_of_t;
let mut result = Vec::with_capacity(length_in_chunks ); let mut result = Vec::with_capacity(length_in_chunks);
unsafe { result.set_len(length_in_chunks) }; unsafe { result.set_len(length_in_chunks) };
for i in 0..length_in_chunks { for i in 0..length_in_chunks {
*result.get_mut(i).unwrap() = try!(T::from_bytes( *result.get_mut(i).unwrap() = try!(T::from_bytes(
@ -339,7 +339,7 @@ impl<T> FromRawBytesVariable for Vec<T> where T: FromRawBytes {
impl<V1, T2> FromRawBytes for (V1, T2) where V1: FromRawBytesVariable, T2: FromRawBytes { impl<V1, T2> FromRawBytes for (V1, T2) where V1: FromRawBytesVariable, T2: FromRawBytes {
fn from_bytes(bytes: &[u8]) -> Result<Self, FromBytesError> { fn from_bytes(bytes: &[u8]) -> Result<Self, FromBytesError> {
let header = 8usize; let header = 8usize;
let mut map: (u64, ) = unsafe { mem::uninitialized() }; let mut map: (u64, ) = (0,);
if bytes.len() < header { return Err(FromBytesError::NotLongEnough); } if bytes.len() < header { return Err(FromBytesError::NotLongEnough); }
map.copy_raw(&bytes[0..header]); map.copy_raw(&bytes[0..header]);
@ -358,7 +358,7 @@ impl<V1, V2, T3> FromRawBytes for (V1, V2, T3)
{ {
fn from_bytes(bytes: &[u8]) -> Result<Self, FromBytesError> { fn from_bytes(bytes: &[u8]) -> Result<Self, FromBytesError> {
let header = 16usize; let header = 16usize;
let mut map: (u64, u64, ) = unsafe { mem::uninitialized() }; let mut map: (u64, u64, ) = (0, 0,);
if bytes.len() < header { return Err(FromBytesError::NotLongEnough); } if bytes.len() < header { return Err(FromBytesError::NotLongEnough); }
map.copy_raw(&bytes[0..header]); map.copy_raw(&bytes[0..header]);
@ -373,7 +373,7 @@ impl<V1, V2, T3> FromRawBytes for (V1, V2, T3)
} }
} }
impl<'a, V1, T2> ToBytesWithMap for (&'a Vec<V1>, &'a T2) where V1: ToBytesWithMap, T2: ToBytesWithMap { impl<'a, V1, X1, T2> ToBytesWithMap for (X1, &'a T2) where V1: ToBytesWithMap, X1: Deref<Target=[V1]>, T2: ToBytesWithMap {
fn to_bytes_map(&self) -> Vec<u8> { fn to_bytes_map(&self) -> Vec<u8> {
let header = 8usize; let header = 8usize;
let v1_size = mem::size_of::<V1>(); let v1_size = mem::size_of::<V1>();
@ -390,9 +390,9 @@ impl<'a, V1, T2> ToBytesWithMap for (&'a Vec<V1>, &'a T2) where V1: ToBytesWithM
} }
impl<'a, V1, V2, T3> ToBytesWithMap for (&'a Vec<V1>, &'a Vec<V2>, &'a T3) impl<'a, V1, X1, V2, X2, T3> ToBytesWithMap for (X1, X2, &'a T3)
where V1: ToBytesWithMap, where V1: ToBytesWithMap, X1: Deref<Target=[V1]>,
V2: ToBytesWithMap, V2: ToBytesWithMap, X2: Deref<Target=[V2]>,
T3: ToBytesWithMap T3: ToBytesWithMap
{ {
fn to_bytes_map(&self) -> Vec<u8> { fn to_bytes_map(&self) -> Vec<u8> {
@ -433,7 +433,7 @@ pub trait ToBytesWithMap {
impl<T> ToBytesWithMap for T where T: FixedHash { impl<T> ToBytesWithMap for T where T: FixedHash {
fn to_bytes_map(&self) -> Vec<u8> { fn to_bytes_map(&self) -> Vec<u8> {
self.as_slice().to_vec() self.as_slice().to_owned()
} }
} }
@ -493,7 +493,7 @@ fn populate_big_types() {
fn raw_bytes_from_tuple() { fn raw_bytes_from_tuple() {
type Tup = (Vec<u16>, u16); type Tup = (Vec<u16>, u16);
let tup = (vec![1u16, 1u16, 1u16, 1u16], 10u16); let tup: (&[u16], u16) = (&[1; 4], 10);
let bytes = vec![ let bytes = vec![
// map // map
8u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 8u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8,
@ -505,18 +505,19 @@ fn raw_bytes_from_tuple() {
// 10u16 // 10u16
10u8, 0u8]; 10u8, 0u8];
let tup_from = Tup::from_bytes(&bytes).unwrap(); let (v, x) = Tup::from_bytes(&bytes).unwrap();
assert_eq!(tup, tup_from); assert_eq!(tup, (&v[..], x));
let tup_from = (v, x);
let tup_to = (&tup_from.0, &tup_from.1); let tup_to = (tup_from.0, &tup_from.1);
let bytes_to = tup_to.to_bytes_map(); let bytes_to = tup_to.to_bytes_map();
assert_eq!(bytes_to, bytes); assert_eq!(bytes_to, bytes);
} }
#[test] #[test]
fn bytes_map_from_triple() { fn bytes_map_from_triple() {
let data = (vec![2u16; 6], vec![6u32; 3], 12u64); let data: (&[u16], &[u32], u64) = (&[2; 6], &[6; 3], 12u64);
let bytes_map = (&data.0, &data.1, &data.2).to_bytes_map(); let bytes_map = (data.0, data.1, &data.2).to_bytes_map();
assert_eq!(bytes_map, vec![ assert_eq!(bytes_map, vec![
// data map 2 x u64 // data map 2 x u64
12, 0, 0, 0, 0, 0, 0, 0, 12, 0, 0, 0, 0, 0, 0, 0,

View File

@ -202,7 +202,7 @@ impl JournalDB for ArchiveDB {
fn latest_era(&self) -> Option<u64> { self.latest_era } fn latest_era(&self) -> Option<u64> { self.latest_era }
fn state(&self, id: &H256) -> Option<Bytes> { fn state(&self, id: &H256) -> Option<Bytes> {
self.backing.get_by_prefix(&id[0..12]).and_then(|b| Some(b.to_vec())) self.backing.get_by_prefix(&id[0..DB_PREFIX_LEN]).map(|b| b.to_vec())
} }
fn is_pruned(&self) -> bool { false } fn is_pruned(&self) -> bool { false }

View File

@ -339,6 +339,10 @@ impl JournalDB for EarlyMergeDB {
} }
} }
fn state(&self, id: &H256) -> Option<Bytes> {
self.backing.get_by_prefix(&id[0..DB_PREFIX_LEN]).map(|b| b.to_vec())
}
#[cfg_attr(feature="dev", allow(cyclomatic_complexity))] #[cfg_attr(feature="dev", allow(cyclomatic_complexity))]
fn commit(&mut self, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError> { fn commit(&mut self, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError> {
// journal format: // journal format:

View File

@ -57,7 +57,7 @@ pub enum Algorithm {
} }
impl Default for Algorithm { impl Default for Algorithm {
fn default() -> Algorithm { Algorithm::Archive } fn default() -> Algorithm { Algorithm::OverlayRecent }
} }
impl fmt::Display for Algorithm { impl fmt::Display for Algorithm {

View File

@ -171,7 +171,7 @@ impl OverlayRecentDB {
for r in insertions.iter() { for r in insertions.iter() {
let k: H256 = r.val_at(0); let k: H256 = r.val_at(0);
let v: Bytes = r.val_at(1); let v: Bytes = r.val_at(1);
overlay.emplace(k.clone(), v); overlay.emplace(OverlayRecentDB::to_short_key(&k), v);
inserted_keys.push(k); inserted_keys.push(k);
count += 1; count += 1;
} }
@ -191,6 +191,13 @@ impl OverlayRecentDB {
trace!("Recovered {} overlay entries, {} journal entries", count, journal.len()); trace!("Recovered {} overlay entries, {} journal entries", count, journal.len());
JournalOverlay { backing_overlay: overlay, journal: journal, latest_era: latest_era } JournalOverlay { backing_overlay: overlay, journal: journal, latest_era: latest_era }
} }
#[inline]
fn to_short_key(key: &H256) -> H256 {
let mut k = H256::new();
&mut k[0..DB_PREFIX_LEN].copy_from_slice(&key[0..DB_PREFIX_LEN]);
k
}
} }
impl JournalDB for OverlayRecentDB { impl JournalDB for OverlayRecentDB {
@ -212,6 +219,11 @@ impl JournalDB for OverlayRecentDB {
fn latest_era(&self) -> Option<u64> { self.journal_overlay.unwrapped_read().latest_era } fn latest_era(&self) -> Option<u64> { self.journal_overlay.unwrapped_read().latest_era }
fn state(&self, key: &H256) -> Option<Bytes> {
let v = self.journal_overlay.unwrapped_read().backing_overlay.get(&OverlayRecentDB::to_short_key(key)).map(|v| v.to_vec());
v.or_else(|| self.backing.get_by_prefix(&key[0..DB_PREFIX_LEN]).map(|b| b.to_vec()))
}
fn commit(&mut self, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError> { fn commit(&mut self, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError> {
// record new commit's details. // record new commit's details.
trace!("commit: #{} ({}), end era: {:?}", now, id, end); trace!("commit: #{} ({}), end era: {:?}", now, id, end);
@ -230,7 +242,7 @@ impl JournalDB for OverlayRecentDB {
r.begin_list(2); r.begin_list(2);
r.append(&k); r.append(&k);
r.append(&v); r.append(&v);
journal_overlay.backing_overlay.emplace(k, v); journal_overlay.backing_overlay.emplace(OverlayRecentDB::to_short_key(&k), v);
} }
r.append(&removed_keys); r.append(&removed_keys);
@ -266,7 +278,7 @@ impl JournalDB for OverlayRecentDB {
{ {
if canon_id == journal.id { if canon_id == journal.id {
for h in &journal.insertions { for h in &journal.insertions {
if let Some(&(ref d, rc)) = journal_overlay.backing_overlay.raw(h) { if let Some(&(ref d, rc)) = journal_overlay.backing_overlay.raw(&OverlayRecentDB::to_short_key(h)) {
if rc > 0 { if rc > 0 {
canon_insertions.push((h.clone(), d.clone())); //TODO: optimize this to avoid data copy canon_insertions.push((h.clone(), d.clone())); //TODO: optimize this to avoid data copy
} }
@ -284,11 +296,11 @@ impl JournalDB for OverlayRecentDB {
} }
// update the overlay // update the overlay
for k in overlay_deletions { for k in overlay_deletions {
journal_overlay.backing_overlay.remove(&k); journal_overlay.backing_overlay.remove(&OverlayRecentDB::to_short_key(&k));
} }
// apply canon deletions // apply canon deletions
for k in canon_deletions { for k in canon_deletions {
if !journal_overlay.backing_overlay.contains(&k) { if !journal_overlay.backing_overlay.contains(&OverlayRecentDB::to_short_key(&k)) {
try!(batch.delete(&k)); try!(batch.delete(&k));
} }
} }
@ -322,7 +334,7 @@ impl HashDB for OverlayRecentDB {
match k { match k {
Some(&(ref d, rc)) if rc > 0 => Some(d), Some(&(ref d, rc)) if rc > 0 => Some(d),
_ => { _ => {
let v = self.journal_overlay.unwrapped_read().backing_overlay.get(key).map(|v| v.to_vec()); let v = self.journal_overlay.unwrapped_read().backing_overlay.get(&OverlayRecentDB::to_short_key(key)).map(|v| v.to_vec());
match v { match v {
Some(x) => { Some(x) => {
Some(&self.transaction_overlay.denote(key, x).0) Some(&self.transaction_overlay.denote(key, x).0)

View File

@ -111,6 +111,10 @@ impl JournalDB for RefCountedDB {
fn latest_era(&self) -> Option<u64> { self.latest_era } fn latest_era(&self) -> Option<u64> { self.latest_era }
fn state(&self, id: &H256) -> Option<Bytes> {
self.backing.get_by_prefix(&id[0..DB_PREFIX_LEN]).map(|b| b.to_vec())
}
fn commit(&mut self, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError> { fn commit(&mut self, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError> {
// journal format: // journal format:
// [era, 0] => [ id, [insert_0, ...], [remove_0, ...] ] // [era, 0] => [ id, [insert_0, ...], [remove_0, ...] ]

View File

@ -39,9 +39,7 @@ pub trait JournalDB : HashDB + Send + Sync {
fn commit(&mut self, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError>; fn commit(&mut self, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError>;
/// State data query /// State data query
fn state(&self, _id: &H256) -> Option<Bytes> { fn state(&self, _id: &H256) -> Option<Bytes>;
None
}
/// Whether this database is pruned. /// Whether this database is pruned.
fn is_pruned(&self) -> bool { true } fn is_pruned(&self) -> bool { true }

View File

@ -136,11 +136,11 @@ pub type ProtocolId = &'static str;
/// Messages used to communitate with the event loop from other threads. /// Messages used to communitate with the event loop from other threads.
#[derive(Clone)] #[derive(Clone)]
pub enum NetworkIoMessage<Message> where Message: Send + Sync + Clone { pub enum NetworkIoMessage {
/// Register a new protocol handler. /// Register a new protocol handler.
AddHandler { AddHandler {
/// Handler shared instance. /// Handler shared instance.
handler: Arc<NetworkProtocolHandler<Message> + Sync>, handler: Arc<NetworkProtocolHandler + Sync>,
/// Protocol Id. /// Protocol Id.
protocol: ProtocolId, protocol: ProtocolId,
/// Supported protocol versions. /// Supported protocol versions.
@ -163,8 +163,6 @@ pub enum NetworkIoMessage<Message> where Message: Send + Sync + Clone {
DisablePeer(PeerId), DisablePeer(PeerId),
/// Network has been started with the host as the given enode. /// Network has been started with the host as the given enode.
NetworkStarted(String), NetworkStarted(String),
/// User message
User(Message),
} }
/// Local (temporary) peer session ID. /// Local (temporary) peer session ID.
@ -188,8 +186,8 @@ impl Encodable for CapabilityInfo {
} }
/// IO access point. This is passed to all IO handlers and provides an interface to the IO subsystem. /// IO access point. This is passed to all IO handlers and provides an interface to the IO subsystem.
pub struct NetworkContext<'s, Message> where Message: Send + Sync + Clone + 'static, 's { pub struct NetworkContext<'s> {
io: &'s IoContext<NetworkIoMessage<Message>>, io: &'s IoContext<NetworkIoMessage>,
protocol: ProtocolId, protocol: ProtocolId,
sessions: Arc<RwLock<Slab<SharedSession>>>, sessions: Arc<RwLock<Slab<SharedSession>>>,
session: Option<SharedSession>, session: Option<SharedSession>,
@ -197,12 +195,12 @@ pub struct NetworkContext<'s, Message> where Message: Send + Sync + Clone + 'sta
_reserved_peers: &'s HashSet<NodeId>, _reserved_peers: &'s HashSet<NodeId>,
} }
impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone + 'static, { impl<'s> NetworkContext<'s> {
/// Create a new network IO access point. Takes references to all the data that can be updated within the IO handler. /// Create a new network IO access point. Takes references to all the data that can be updated within the IO handler.
fn new(io: &'s IoContext<NetworkIoMessage<Message>>, fn new(io: &'s IoContext<NetworkIoMessage>,
protocol: ProtocolId, protocol: ProtocolId,
session: Option<SharedSession>, sessions: Arc<RwLock<Slab<SharedSession>>>, session: Option<SharedSession>, sessions: Arc<RwLock<Slab<SharedSession>>>,
reserved_peers: &'s HashSet<NodeId>) -> NetworkContext<'s, Message> { reserved_peers: &'s HashSet<NodeId>) -> NetworkContext<'s> {
let id = session.as_ref().map(|s| s.locked().token()); let id = session.as_ref().map(|s| s.locked().token());
NetworkContext { NetworkContext {
io: io, io: io,
@ -238,13 +236,8 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
self.send(self.session_id.unwrap(), packet_id, data) self.send(self.session_id.unwrap(), packet_id, data)
} }
/// Send an IO message
pub fn message(&self, msg: Message) -> Result<(), UtilError> {
self.io.message(NetworkIoMessage::User(msg))
}
/// Get an IoChannel. /// Get an IoChannel.
pub fn io_channel(&self) -> IoChannel<NetworkIoMessage<Message>> { pub fn io_channel(&self) -> IoChannel<NetworkIoMessage> {
self.io.channel() self.io.channel()
} }
@ -333,13 +326,13 @@ struct ProtocolTimer {
} }
/// Root IO handler. Manages protocol handlers, IO timers and network connections. /// Root IO handler. Manages protocol handlers, IO timers and network connections.
pub struct Host<Message> where Message: Send + Sync + Clone { pub struct Host {
pub info: RwLock<HostInfo>, pub info: RwLock<HostInfo>,
tcp_listener: Mutex<TcpListener>, tcp_listener: Mutex<TcpListener>,
sessions: Arc<RwLock<Slab<SharedSession>>>, sessions: Arc<RwLock<Slab<SharedSession>>>,
discovery: Mutex<Option<Discovery>>, discovery: Mutex<Option<Discovery>>,
nodes: RwLock<NodeTable>, nodes: RwLock<NodeTable>,
handlers: RwLock<HashMap<ProtocolId, Arc<NetworkProtocolHandler<Message>>>>, handlers: RwLock<HashMap<ProtocolId, Arc<NetworkProtocolHandler>>>,
timers: RwLock<HashMap<TimerToken, ProtocolTimer>>, timers: RwLock<HashMap<TimerToken, ProtocolTimer>>,
timer_counter: RwLock<usize>, timer_counter: RwLock<usize>,
stats: Arc<NetworkStats>, stats: Arc<NetworkStats>,
@ -348,9 +341,9 @@ pub struct Host<Message> where Message: Send + Sync + Clone {
stopping: AtomicBool, stopping: AtomicBool,
} }
impl<Message> Host<Message> where Message: Send + Sync + Clone { impl Host {
/// Create a new instance /// Create a new instance
pub fn new(config: NetworkConfiguration, stats: Arc<NetworkStats>) -> Result<Host<Message>, UtilError> { pub fn new(config: NetworkConfiguration, stats: Arc<NetworkStats>) -> Result<Host, UtilError> {
trace!(target: "host", "Creating new Host object"); trace!(target: "host", "Creating new Host object");
let mut listen_address = match config.listen_address { let mut listen_address = match config.listen_address {
@ -381,7 +374,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
let boot_nodes = config.boot_nodes.clone(); let boot_nodes = config.boot_nodes.clone();
let reserved_nodes = config.reserved_nodes.clone(); let reserved_nodes = config.reserved_nodes.clone();
let mut host = Host::<Message> { let mut host = Host {
info: RwLock::new(HostInfo { info: RwLock::new(HostInfo {
keys: keys, keys: keys,
config: config, config: config,
@ -444,7 +437,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
Ok(()) Ok(())
} }
pub fn set_non_reserved_mode(&self, mode: NonReservedPeerMode, io: &IoContext<NetworkIoMessage<Message>>) { pub fn set_non_reserved_mode(&self, mode: NonReservedPeerMode, io: &IoContext<NetworkIoMessage>) {
let mut info = self.info.unwrapped_write(); let mut info = self.info.unwrapped_write();
if info.config.non_reserved_mode != mode { if info.config.non_reserved_mode != mode {
@ -495,7 +488,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
r r
} }
pub fn stop(&self, io: &IoContext<NetworkIoMessage<Message>>) -> Result<(), UtilError> { pub fn stop(&self, io: &IoContext<NetworkIoMessage>) -> Result<(), UtilError> {
self.stopping.store(true, AtomicOrdering::Release); self.stopping.store(true, AtomicOrdering::Release);
let mut to_kill = Vec::new(); let mut to_kill = Vec::new();
for e in self.sessions.unwrapped_write().iter_mut() { for e in self.sessions.unwrapped_write().iter_mut() {
@ -511,7 +504,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
Ok(()) Ok(())
} }
fn init_public_interface(&self, io: &IoContext<NetworkIoMessage<Message>>) -> Result<(), UtilError> { fn init_public_interface(&self, io: &IoContext<NetworkIoMessage>) -> Result<(), UtilError> {
if self.info.unwrapped_read().public_endpoint.is_some() { if self.info.unwrapped_read().public_endpoint.is_some() {
return Ok(()); return Ok(());
} }
@ -567,7 +560,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
Ok(()) Ok(())
} }
fn maintain_network(&self, io: &IoContext<NetworkIoMessage<Message>>) { fn maintain_network(&self, io: &IoContext<NetworkIoMessage>) {
self.keep_alive(io); self.keep_alive(io);
self.connect_peers(io); self.connect_peers(io);
} }
@ -588,7 +581,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
self.sessions.unwrapped_read().count() - self.session_count() self.sessions.unwrapped_read().count() - self.session_count()
} }
fn keep_alive(&self, io: &IoContext<NetworkIoMessage<Message>>) { fn keep_alive(&self, io: &IoContext<NetworkIoMessage>) {
let mut to_kill = Vec::new(); let mut to_kill = Vec::new();
for e in self.sessions.unwrapped_write().iter_mut() { for e in self.sessions.unwrapped_write().iter_mut() {
let mut s = e.locked(); let mut s = e.locked();
@ -603,7 +596,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
} }
} }
fn connect_peers(&self, io: &IoContext<NetworkIoMessage<Message>>) { fn connect_peers(&self, io: &IoContext<NetworkIoMessage>) {
let (ideal_peers, mut pin) = { let (ideal_peers, mut pin) = {
let info = self.info.unwrapped_read(); let info = self.info.unwrapped_read();
if info.capabilities.is_empty() { if info.capabilities.is_empty() {
@ -651,7 +644,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
} }
#[cfg_attr(feature="dev", allow(single_match))] #[cfg_attr(feature="dev", allow(single_match))]
fn connect_peer(&self, id: &NodeId, io: &IoContext<NetworkIoMessage<Message>>) { fn connect_peer(&self, id: &NodeId, io: &IoContext<NetworkIoMessage>) {
if self.have_session(id) if self.have_session(id)
{ {
trace!(target: "network", "Aborted connect. Node already connected."); trace!(target: "network", "Aborted connect. Node already connected.");
@ -688,9 +681,10 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
} }
#[cfg_attr(feature="dev", allow(block_in_if_condition_stmt))] #[cfg_attr(feature="dev", allow(block_in_if_condition_stmt))]
fn create_connection(&self, socket: TcpStream, id: Option<&NodeId>, io: &IoContext<NetworkIoMessage<Message>>) -> Result<(), UtilError> { fn create_connection(&self, socket: TcpStream, id: Option<&NodeId>, io: &IoContext<NetworkIoMessage>) -> Result<(), UtilError> {
let nonce = self.info.unwrapped_write().next_nonce(); let nonce = self.info.unwrapped_write().next_nonce();
let mut sessions = self.sessions.unwrapped_write(); let mut sessions = self.sessions.unwrapped_write();
let token = sessions.insert_with_opt(|token| { let token = sessions.insert_with_opt(|token| {
match Session::new(io, socket, token, id, &nonce, self.stats.clone(), &self.info.unwrapped_read()) { match Session::new(io, socket, token, id, &nonce, self.stats.clone(), &self.info.unwrapped_read()) {
Ok(s) => Some(Arc::new(Mutex::new(s))), Ok(s) => Some(Arc::new(Mutex::new(s))),
@ -710,7 +704,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
} }
} }
fn accept(&self, io: &IoContext<NetworkIoMessage<Message>>) { fn accept(&self, io: &IoContext<NetworkIoMessage>) {
trace!(target: "network", "Accepting incoming connection"); trace!(target: "network", "Accepting incoming connection");
loop { loop {
let socket = match self.tcp_listener.locked().accept() { let socket = match self.tcp_listener.locked().accept() {
@ -727,8 +721,9 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
} }
} }
fn session_writable(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) { fn session_writable(&self, token: StreamToken, io: &IoContext<NetworkIoMessage>) {
let session = { self.sessions.unwrapped_read().get(token).cloned() }; let session = { self.sessions.unwrapped_read().get(token).cloned() };
if let Some(session) = session { if let Some(session) = session {
let mut s = session.locked(); let mut s = session.locked();
if let Err(e) = s.writable(io, &self.info.unwrapped_read()) { if let Err(e) = s.writable(io, &self.info.unwrapped_read()) {
@ -740,13 +735,13 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
} }
} }
fn connection_closed(&self, token: TimerToken, io: &IoContext<NetworkIoMessage<Message>>) { fn connection_closed(&self, token: TimerToken, io: &IoContext<NetworkIoMessage>) {
trace!(target: "network", "Connection closed: {}", token); trace!(target: "network", "Connection closed: {}", token);
self.kill_connection(token, io, true); self.kill_connection(token, io, true);
} }
#[cfg_attr(feature="dev", allow(collapsible_if))] #[cfg_attr(feature="dev", allow(collapsible_if))]
fn session_readable(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) { fn session_readable(&self, token: StreamToken, io: &IoContext<NetworkIoMessage>) {
let mut ready_data: Vec<ProtocolId> = Vec::new(); let mut ready_data: Vec<ProtocolId> = Vec::new();
let mut packet_data: Vec<(ProtocolId, PacketId, Vec<u8>)> = Vec::new(); let mut packet_data: Vec<(ProtocolId, PacketId, Vec<u8>)> = Vec::new();
let mut kill = false; let mut kill = false;
@ -831,12 +826,12 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
} }
} }
fn connection_timeout(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) { fn connection_timeout(&self, token: StreamToken, io: &IoContext<NetworkIoMessage>) {
trace!(target: "network", "Connection timeout: {}", token); trace!(target: "network", "Connection timeout: {}", token);
self.kill_connection(token, io, true) self.kill_connection(token, io, true)
} }
fn kill_connection(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>, remote: bool) { fn kill_connection(&self, token: StreamToken, io: &IoContext<NetworkIoMessage>, remote: bool) {
let mut to_disconnect: Vec<ProtocolId> = Vec::new(); let mut to_disconnect: Vec<ProtocolId> = Vec::new();
let mut failure_id = None; let mut failure_id = None;
let mut deregister = false; let mut deregister = false;
@ -876,7 +871,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
} }
} }
fn update_nodes(&self, io: &IoContext<NetworkIoMessage<Message>>, node_changes: TableUpdates) { fn update_nodes(&self, io: &IoContext<NetworkIoMessage>, node_changes: TableUpdates) {
let mut to_remove: Vec<PeerId> = Vec::new(); let mut to_remove: Vec<PeerId> = Vec::new();
{ {
let sessions = self.sessions.unwrapped_write(); let sessions = self.sessions.unwrapped_write();
@ -895,17 +890,24 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
} }
self.nodes.unwrapped_write().update(node_changes); self.nodes.unwrapped_write().update(node_changes);
} }
pub fn with_context<F>(&self, protocol: ProtocolId, io: &IoContext<NetworkIoMessage>, action: F) where F: Fn(&NetworkContext) {
let reserved = { self.reserved_nodes.unwrapped_read() };
let context = NetworkContext::new(io, protocol, None, self.sessions.clone(), &reserved);
action(&context);
}
} }
impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Message: Send + Sync + Clone + 'static { impl IoHandler<NetworkIoMessage> for Host {
/// Initialize networking /// Initialize networking
fn initialize(&self, io: &IoContext<NetworkIoMessage<Message>>) { fn initialize(&self, io: &IoContext<NetworkIoMessage>) {
io.register_timer(IDLE, MAINTENANCE_TIMEOUT).expect("Error registering Network idle timer"); io.register_timer(IDLE, MAINTENANCE_TIMEOUT).expect("Error registering Network idle timer");
io.message(NetworkIoMessage::InitPublicInterface).unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e)); io.message(NetworkIoMessage::InitPublicInterface).unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e));
self.maintain_network(io) self.maintain_network(io)
} }
fn stream_hup(&self, io: &IoContext<NetworkIoMessage<Message>>, stream: StreamToken) { fn stream_hup(&self, io: &IoContext<NetworkIoMessage>, stream: StreamToken) {
trace!(target: "network", "Hup: {}", stream); trace!(target: "network", "Hup: {}", stream);
match stream { match stream {
FIRST_SESSION ... LAST_SESSION => self.connection_closed(stream, io), FIRST_SESSION ... LAST_SESSION => self.connection_closed(stream, io),
@ -913,7 +915,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
}; };
} }
fn stream_readable(&self, io: &IoContext<NetworkIoMessage<Message>>, stream: StreamToken) { fn stream_readable(&self, io: &IoContext<NetworkIoMessage>, stream: StreamToken) {
if self.stopping.load(AtomicOrdering::Acquire) { if self.stopping.load(AtomicOrdering::Acquire) {
return; return;
} }
@ -930,7 +932,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
} }
} }
fn stream_writable(&self, io: &IoContext<NetworkIoMessage<Message>>, stream: StreamToken) { fn stream_writable(&self, io: &IoContext<NetworkIoMessage>, stream: StreamToken) {
if self.stopping.load(AtomicOrdering::Acquire) { if self.stopping.load(AtomicOrdering::Acquire) {
return; return;
} }
@ -943,7 +945,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
} }
} }
fn timeout(&self, io: &IoContext<NetworkIoMessage<Message>>, token: TimerToken) { fn timeout(&self, io: &IoContext<NetworkIoMessage>, token: TimerToken) {
if self.stopping.load(AtomicOrdering::Acquire) { if self.stopping.load(AtomicOrdering::Acquire) {
return; return;
} }
@ -978,7 +980,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
} }
} }
fn message(&self, io: &IoContext<NetworkIoMessage<Message>>, message: &NetworkIoMessage<Message>) { fn message(&self, io: &IoContext<NetworkIoMessage>, message: &NetworkIoMessage) {
if self.stopping.load(AtomicOrdering::Acquire) { if self.stopping.load(AtomicOrdering::Acquire) {
return; return;
} }
@ -1031,19 +1033,13 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
trace!(target: "network", "Disabling peer {}", peer); trace!(target: "network", "Disabling peer {}", peer);
self.kill_connection(*peer, io, false); self.kill_connection(*peer, io, false);
}, },
NetworkIoMessage::User(ref message) => {
let reserved = self.reserved_nodes.unwrapped_read();
for (p, h) in self.handlers.unwrapped_read().iter() {
h.message(&NetworkContext::new(io, p, None, self.sessions.clone(), &reserved), &message);
}
},
NetworkIoMessage::InitPublicInterface => NetworkIoMessage::InitPublicInterface =>
self.init_public_interface(io).unwrap_or_else(|e| warn!("Error initializing public interface: {:?}", e)), self.init_public_interface(io).unwrap_or_else(|e| warn!("Error initializing public interface: {:?}", e)),
_ => {} // ignore others. _ => {} // ignore others.
} }
} }
fn register_stream(&self, stream: StreamToken, reg: Token, event_loop: &mut EventLoop<IoManager<NetworkIoMessage<Message>>>) { fn register_stream(&self, stream: StreamToken, reg: Token, event_loop: &mut EventLoop<IoManager<NetworkIoMessage>>) {
match stream { match stream {
FIRST_SESSION ... LAST_SESSION => { FIRST_SESSION ... LAST_SESSION => {
let session = { self.sessions.unwrapped_read().get(stream).cloned() }; let session = { self.sessions.unwrapped_read().get(stream).cloned() };
@ -1057,7 +1053,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
} }
} }
fn deregister_stream(&self, stream: StreamToken, event_loop: &mut EventLoop<IoManager<NetworkIoMessage<Message>>>) { fn deregister_stream(&self, stream: StreamToken, event_loop: &mut EventLoop<IoManager<NetworkIoMessage>>) {
match stream { match stream {
FIRST_SESSION ... LAST_SESSION => { FIRST_SESSION ... LAST_SESSION => {
let mut connections = self.sessions.unwrapped_write(); let mut connections = self.sessions.unwrapped_write();
@ -1071,7 +1067,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
} }
} }
fn update_stream(&self, stream: StreamToken, reg: Token, event_loop: &mut EventLoop<IoManager<NetworkIoMessage<Message>>>) { fn update_stream(&self, stream: StreamToken, reg: Token, event_loop: &mut EventLoop<IoManager<NetworkIoMessage>>) {
match stream { match stream {
FIRST_SESSION ... LAST_SESSION => { FIRST_SESSION ... LAST_SESSION => {
let connection = { self.sessions.unwrapped_read().get(stream).cloned() }; let connection = { self.sessions.unwrapped_read().get(stream).cloned() };
@ -1152,6 +1148,6 @@ fn host_client_url() {
let mut config = NetworkConfiguration::new(); let mut config = NetworkConfiguration::new();
let key = h256_from_hex("6f7b0d801bc7b5ce7bbd930b84fd0369b3eb25d09be58d64ba811091046f3aa2"); let key = h256_from_hex("6f7b0d801bc7b5ce7bbd930b84fd0369b3eb25d09be58d64ba811091046f3aa2");
config.use_secret = Some(key); config.use_secret = Some(key);
let host: Host<u32> = Host::new(config, Arc::new(NetworkStats::new())).unwrap(); let host: Host = Host::new(config, Arc::new(NetworkStats::new())).unwrap();
assert!(host.local_url().starts_with("enode://101b3ef5a4ea7a1c7928e24c4c75fd053c235d7b80c22ae5c03d145d0ac7396e2a4ffff9adee3133a7b05044a5cee08115fd65145e5165d646bde371010d803c@")); assert!(host.local_url().starts_with("enode://101b3ef5a4ea7a1c7928e24c4c75fd053c235d7b80c22ae5c03d145d0ac7396e2a4ffff9adee3133a7b05044a5cee08115fd65145e5165d646bde371010d803c@"));
} }

View File

@ -24,39 +24,30 @@
//! //!
//! struct MyHandler; //! struct MyHandler;
//! //!
//! #[derive(Clone)] //! impl NetworkProtocolHandler for MyHandler {
//! struct MyMessage { //! fn initialize(&self, io: &NetworkContext) {
//! data: u32
//! }
//!
//! impl NetworkProtocolHandler<MyMessage> for MyHandler {
//! fn initialize(&self, io: &NetworkContext<MyMessage>) {
//! io.register_timer(0, 1000); //! io.register_timer(0, 1000);
//! } //! }
//! //!
//! fn read(&self, io: &NetworkContext<MyMessage>, peer: &PeerId, packet_id: u8, data: &[u8]) { //! fn read(&self, io: &NetworkContext, peer: &PeerId, packet_id: u8, data: &[u8]) {
//! println!("Received {} ({} bytes) from {}", packet_id, data.len(), peer); //! println!("Received {} ({} bytes) from {}", packet_id, data.len(), peer);
//! } //! }
//! //!
//! fn connected(&self, io: &NetworkContext<MyMessage>, peer: &PeerId) { //! fn connected(&self, io: &NetworkContext, peer: &PeerId) {
//! println!("Connected {}", peer); //! println!("Connected {}", peer);
//! } //! }
//! //!
//! fn disconnected(&self, io: &NetworkContext<MyMessage>, peer: &PeerId) { //! fn disconnected(&self, io: &NetworkContext, peer: &PeerId) {
//! println!("Disconnected {}", peer); //! println!("Disconnected {}", peer);
//! } //! }
//! //!
//! fn timeout(&self, io: &NetworkContext<MyMessage>, timer: TimerToken) { //! fn timeout(&self, io: &NetworkContext, timer: TimerToken) {
//! println!("Timeout {}", timer); //! println!("Timeout {}", timer);
//! } //! }
//!
//! fn message(&self, io: &NetworkContext<MyMessage>, message: &MyMessage) {
//! println!("Message {}", message.data);
//! }
//! } //! }
//! //!
//! fn main () { //! fn main () {
//! let mut service = NetworkService::<MyMessage>::new(NetworkConfiguration::new_local()).expect("Error creating network service"); //! let mut service = NetworkService::new(NetworkConfiguration::new_local()).expect("Error creating network service");
//! service.register_protocol(Arc::new(MyHandler), "myproto", &[1u8]); //! service.register_protocol(Arc::new(MyHandler), "myproto", &[1u8]);
//! service.start().expect("Error starting service"); //! service.start().expect("Error starting service");
//! //!
@ -84,7 +75,6 @@ pub use network::host::PacketId;
pub use network::host::NetworkContext; pub use network::host::NetworkContext;
pub use network::service::NetworkService; pub use network::service::NetworkService;
pub use network::host::NetworkIoMessage; pub use network::host::NetworkIoMessage;
pub use network::host::NetworkIoMessage::User as UserMessage;
pub use network::error::NetworkError; pub use network::error::NetworkError;
pub use network::host::NetworkConfiguration; pub use network::host::NetworkConfiguration;
pub use network::stats::NetworkStats; pub use network::stats::NetworkStats;
@ -97,19 +87,17 @@ const PROTOCOL_VERSION: u32 = 4;
/// Network IO protocol handler. This needs to be implemented for each new subprotocol. /// Network IO protocol handler. This needs to be implemented for each new subprotocol.
/// All the handler function are called from within IO event loop. /// All the handler function are called from within IO event loop.
/// `Message` is the type for message data. /// `Message` is the type for message data.
pub trait NetworkProtocolHandler<Message>: Sync + Send where Message: Send + Sync + Clone { pub trait NetworkProtocolHandler: Sync + Send {
/// Initialize the handler /// Initialize the handler
fn initialize(&self, _io: &NetworkContext<Message>) {} fn initialize(&self, _io: &NetworkContext) {}
/// Called when new network packet received. /// Called when new network packet received.
fn read(&self, io: &NetworkContext<Message>, peer: &PeerId, packet_id: u8, data: &[u8]); fn read(&self, io: &NetworkContext, peer: &PeerId, packet_id: u8, data: &[u8]);
/// Called when new peer is connected. Only called when peer supports the same protocol. /// Called when new peer is connected. Only called when peer supports the same protocol.
fn connected(&self, io: &NetworkContext<Message>, peer: &PeerId); fn connected(&self, io: &NetworkContext, peer: &PeerId);
/// Called when a previously connected peer disconnects. /// Called when a previously connected peer disconnects.
fn disconnected(&self, io: &NetworkContext<Message>, peer: &PeerId); fn disconnected(&self, io: &NetworkContext, peer: &PeerId);
/// Timer function called after a timeout created with `NetworkContext::timeout`. /// Timer function called after a timeout created with `NetworkContext::timeout`.
fn timeout(&self, _io: &NetworkContext<Message>, _timer: TimerToken) {} fn timeout(&self, _io: &NetworkContext, _timer: TimerToken) {}
/// Called when a broadcasted message is received. The message can only be sent from a different IO handler.
fn message(&self, _io: &NetworkContext<Message>, _message: &Message) {}
} }
/// Non-reserved peer modes. /// Non-reserved peer modes.
@ -130,4 +118,4 @@ impl NonReservedPeerMode {
_ => None, _ => None,
} }
} }
} }

View File

@ -20,30 +20,30 @@ use panics::*;
use misc::RwLockable; use misc::RwLockable;
use network::{NetworkProtocolHandler, NetworkConfiguration}; use network::{NetworkProtocolHandler, NetworkConfiguration};
use network::error::NetworkError; use network::error::NetworkError;
use network::host::{Host, NetworkIoMessage, ProtocolId}; use network::host::{Host, NetworkContext, NetworkIoMessage, ProtocolId};
use network::stats::NetworkStats; use network::stats::NetworkStats;
use io::*; use io::*;
/// IO Service with networking /// IO Service with networking
/// `Message` defines a notification data type. /// `Message` defines a notification data type.
pub struct NetworkService<Message> where Message: Send + Sync + Clone + 'static { pub struct NetworkService {
io_service: IoService<NetworkIoMessage<Message>>, io_service: IoService<NetworkIoMessage>,
host_info: String, host_info: String,
host: RwLock<Option<Arc<Host<Message>>>>, host: RwLock<Option<Arc<Host>>>,
stats: Arc<NetworkStats>, stats: Arc<NetworkStats>,
panic_handler: Arc<PanicHandler>, panic_handler: Arc<PanicHandler>,
config: NetworkConfiguration, config: NetworkConfiguration,
} }
impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'static { impl NetworkService {
/// Starts IO event loop /// Starts IO event loop
pub fn new(config: NetworkConfiguration) -> Result<NetworkService<Message>, UtilError> { pub fn new(config: NetworkConfiguration) -> Result<NetworkService, UtilError> {
let panic_handler = PanicHandler::new_in_arc(); let panic_handler = PanicHandler::new_in_arc();
let io_service = try!(IoService::<NetworkIoMessage<Message>>::start()); let io_service = try!(IoService::<NetworkIoMessage>::start());
panic_handler.forward_from(&io_service); panic_handler.forward_from(&io_service);
let stats = Arc::new(NetworkStats::new()); let stats = Arc::new(NetworkStats::new());
let host_info = Host::<Message>::client_version(); let host_info = Host::client_version();
Ok(NetworkService { Ok(NetworkService {
io_service: io_service, io_service: io_service,
host_info: host_info, host_info: host_info,
@ -55,7 +55,7 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
} }
/// Regiter a new protocol handler with the event loop. /// Regiter a new protocol handler with the event loop.
pub fn register_protocol(&self, handler: Arc<NetworkProtocolHandler<Message>+Send + Sync>, protocol: ProtocolId, versions: &[u8]) -> Result<(), NetworkError> { pub fn register_protocol(&self, handler: Arc<NetworkProtocolHandler + Send + Sync>, protocol: ProtocolId, versions: &[u8]) -> Result<(), NetworkError> {
try!(self.io_service.send_message(NetworkIoMessage::AddHandler { try!(self.io_service.send_message(NetworkIoMessage::AddHandler {
handler: handler, handler: handler,
protocol: protocol, protocol: protocol,
@ -70,7 +70,7 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
} }
/// Returns underlying io service. /// Returns underlying io service.
pub fn io(&self) -> &IoService<NetworkIoMessage<Message>> { pub fn io(&self) -> &IoService<NetworkIoMessage> {
&self.io_service &self.io_service
} }
@ -146,9 +146,18 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
host.set_non_reserved_mode(mode, &io_ctxt); host.set_non_reserved_mode(mode, &io_ctxt);
} }
} }
/// Executes action in the network context
pub fn with_context<F>(&self, protocol: ProtocolId, action: F) where F: Fn(&NetworkContext) {
let io = IoContext::new(self.io_service.channel(), 0);
let host = self.host.unwrapped_read();
if let Some(ref host) = host.as_ref() {
host.with_context(protocol, &io, action);
};
}
} }
impl<Message> MayPanic for NetworkService<Message> where Message: Send + Sync + Clone + 'static { impl MayPanic for NetworkService {
fn on_panic<F>(&self, closure: F) where F: OnPanicListener { fn on_panic<F>(&self, closure: F) where F: OnPanicListener {
self.panic_handler.on_panic(closure); self.panic_handler.on_panic(closure);
} }

View File

@ -30,22 +30,17 @@ pub struct TestProtocol {
pub got_disconnect: AtomicBool, pub got_disconnect: AtomicBool,
} }
#[derive(Clone)]
pub struct TestProtocolMessage {
payload: u32,
}
impl TestProtocol { impl TestProtocol {
pub fn new(drop_session: bool) -> Self { pub fn new(drop_session: bool) -> Self {
TestProtocol { TestProtocol {
packet: Mutex::new(Vec::new()), packet: Mutex::new(Vec::new()),
got_timeout: AtomicBool::new(false), got_timeout: AtomicBool::new(false),
got_disconnect: AtomicBool::new(false), got_disconnect: AtomicBool::new(false),
drop_session: drop_session, drop_session: drop_session,
} }
} }
/// Creates and register protocol with the network service /// Creates and register protocol with the network service
pub fn register(service: &mut NetworkService<TestProtocolMessage>, drop_session: bool) -> Arc<TestProtocol> { pub fn register(service: &mut NetworkService, drop_session: bool) -> Arc<TestProtocol> {
let handler = Arc::new(TestProtocol::new(drop_session)); let handler = Arc::new(TestProtocol::new(drop_session));
service.register_protocol(handler.clone(), "test", &[42u8, 43u8]).expect("Error registering test protocol handler"); service.register_protocol(handler.clone(), "test", &[42u8, 43u8]).expect("Error registering test protocol handler");
handler handler
@ -64,17 +59,17 @@ impl TestProtocol {
} }
} }
impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol { impl NetworkProtocolHandler for TestProtocol {
fn initialize(&self, io: &NetworkContext<TestProtocolMessage>) { fn initialize(&self, io: &NetworkContext) {
io.register_timer(0, 10).unwrap(); io.register_timer(0, 10).unwrap();
} }
fn read(&self, _io: &NetworkContext<TestProtocolMessage>, _peer: &PeerId, packet_id: u8, data: &[u8]) { fn read(&self, _io: &NetworkContext, _peer: &PeerId, packet_id: u8, data: &[u8]) {
assert_eq!(packet_id, 33); assert_eq!(packet_id, 33);
self.packet.locked().extend(data); self.packet.locked().extend(data);
} }
fn connected(&self, io: &NetworkContext<TestProtocolMessage>, peer: &PeerId) { fn connected(&self, io: &NetworkContext, peer: &PeerId) {
assert!(io.peer_info(*peer).contains("Parity")); assert!(io.peer_info(*peer).contains("Parity"));
if self.drop_session { if self.drop_session {
io.disconnect_peer(*peer) io.disconnect_peer(*peer)
@ -83,13 +78,12 @@ impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol {
} }
} }
fn disconnected(&self, _io: &NetworkContext<TestProtocolMessage>, _peer: &PeerId) { fn disconnected(&self, _io: &NetworkContext, _peer: &PeerId) {
self.got_disconnect.store(true, AtomicOrdering::Relaxed); self.got_disconnect.store(true, AtomicOrdering::Relaxed);
} }
/// Timer function called after a timeout created with `NetworkContext::timeout`. /// Timer function called after a timeout created with `NetworkContext::timeout`.
fn timeout(&self, io: &NetworkContext<TestProtocolMessage>, timer: TimerToken) { fn timeout(&self, _io: &NetworkContext, timer: TimerToken) {
io.message(TestProtocolMessage { payload: 22 }).unwrap();
assert_eq!(timer, 0); assert_eq!(timer, 0);
self.got_timeout.store(true, AtomicOrdering::Relaxed); self.got_timeout.store(true, AtomicOrdering::Relaxed);
} }
@ -98,7 +92,7 @@ impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol {
#[test] #[test]
fn net_service() { fn net_service() {
let service = NetworkService::<TestProtocolMessage>::new(NetworkConfiguration::new_local()).expect("Error creating network service"); let service = NetworkService::new(NetworkConfiguration::new_local()).expect("Error creating network service");
service.start().unwrap(); service.start().unwrap();
service.register_protocol(Arc::new(TestProtocol::new(false)), "myproto", &[1u8]).unwrap(); service.register_protocol(Arc::new(TestProtocol::new(false)), "myproto", &[1u8]).unwrap();
} }
@ -110,13 +104,13 @@ fn net_connect() {
let mut config1 = NetworkConfiguration::new_local(); let mut config1 = NetworkConfiguration::new_local();
config1.use_secret = Some(key1.secret().clone()); config1.use_secret = Some(key1.secret().clone());
config1.boot_nodes = vec![ ]; config1.boot_nodes = vec![ ];
let mut service1 = NetworkService::<TestProtocolMessage>::new(config1).unwrap(); let mut service1 = NetworkService::new(config1).unwrap();
service1.start().unwrap(); service1.start().unwrap();
let handler1 = TestProtocol::register(&mut service1, false); let handler1 = TestProtocol::register(&mut service1, false);
let mut config2 = NetworkConfiguration::new_local(); let mut config2 = NetworkConfiguration::new_local();
info!("net_connect: local URL: {}", service1.local_url().unwrap()); info!("net_connect: local URL: {}", service1.local_url().unwrap());
config2.boot_nodes = vec![ service1.local_url().unwrap() ]; config2.boot_nodes = vec![ service1.local_url().unwrap() ];
let mut service2 = NetworkService::<TestProtocolMessage>::new(config2).unwrap(); let mut service2 = NetworkService::new(config2).unwrap();
service2.start().unwrap(); service2.start().unwrap();
let handler2 = TestProtocol::register(&mut service2, false); let handler2 = TestProtocol::register(&mut service2, false);
while !handler1.got_packet() && !handler2.got_packet() && (service1.stats().sessions() == 0 || service2.stats().sessions() == 0) { while !handler1.got_packet() && !handler2.got_packet() && (service1.stats().sessions() == 0 || service2.stats().sessions() == 0) {
@ -129,7 +123,7 @@ fn net_connect() {
#[test] #[test]
fn net_start_stop() { fn net_start_stop() {
let config = NetworkConfiguration::new_local(); let config = NetworkConfiguration::new_local();
let service = NetworkService::<TestProtocolMessage>::new(config).unwrap(); let service = NetworkService::new(config).unwrap();
service.start().unwrap(); service.start().unwrap();
service.stop().unwrap(); service.stop().unwrap();
service.start().unwrap(); service.start().unwrap();
@ -141,12 +135,12 @@ fn net_disconnect() {
let mut config1 = NetworkConfiguration::new_local(); let mut config1 = NetworkConfiguration::new_local();
config1.use_secret = Some(key1.secret().clone()); config1.use_secret = Some(key1.secret().clone());
config1.boot_nodes = vec![ ]; config1.boot_nodes = vec![ ];
let mut service1 = NetworkService::<TestProtocolMessage>::new(config1).unwrap(); let mut service1 = NetworkService::new(config1).unwrap();
service1.start().unwrap(); service1.start().unwrap();
let handler1 = TestProtocol::register(&mut service1, false); let handler1 = TestProtocol::register(&mut service1, false);
let mut config2 = NetworkConfiguration::new_local(); let mut config2 = NetworkConfiguration::new_local();
config2.boot_nodes = vec![ service1.local_url().unwrap() ]; config2.boot_nodes = vec![ service1.local_url().unwrap() ];
let mut service2 = NetworkService::<TestProtocolMessage>::new(config2).unwrap(); let mut service2 = NetworkService::new(config2).unwrap();
service2.start().unwrap(); service2.start().unwrap();
let handler2 = TestProtocol::register(&mut service2, true); let handler2 = TestProtocol::register(&mut service2, true);
while !(handler1.got_disconnect() && handler2.got_disconnect()) { while !(handler1.got_disconnect() && handler2.got_disconnect()) {
@ -159,7 +153,7 @@ fn net_disconnect() {
#[test] #[test]
fn net_timeout() { fn net_timeout() {
let config = NetworkConfiguration::new_local(); let config = NetworkConfiguration::new_local();
let mut service = NetworkService::<TestProtocolMessage>::new(config).unwrap(); let mut service = NetworkService::new(config).unwrap();
service.start().unwrap(); service.start().unwrap();
let handler = TestProtocol::register(&mut service, false); let handler = TestProtocol::register(&mut service, false);
while !handler.got_timeout() { while !handler.got_timeout() {

View File

@ -195,7 +195,7 @@ fn hash256rlp(input: &[(Vec<u8>, Vec<u8>)], pre_len: usize, stream: &mut RlpStre
} }
// take slices // take slices
let key: &Vec<u8> = &input[0].0; let key: &[u8] = &input[0].0;
let value: &[u8] = &input[0].1; let value: &[u8] = &input[0].1;
// if the slice contains just one item, append the suffix of the key // if the slice contains just one item, append the suffix of the key

View File

@ -17,23 +17,23 @@
//! Vector extensions. //! Vector extensions.
/// Returns len of prefix shared with elem /// Returns len of prefix shared with elem
/// ///
/// ```rust /// ```rust
/// extern crate ethcore_util as util; /// extern crate ethcore_util as util;
/// use util::vector::SharedPrefix; /// use util::vector::SharedPrefix;
/// ///
/// fn main () { /// fn main () {
/// let a = vec![1,2,3,3,5]; /// let a = vec![1,2,3,3,5];
/// let b = vec![1,2,3]; /// let b = vec![1,2,3];
/// assert_eq!(a.shared_prefix_len(&b), 3); /// assert_eq!(a.shared_prefix_len(&b), 3);
/// } /// }
/// ``` /// ```
pub trait SharedPrefix <T> { pub trait SharedPrefix<T> {
/// Get common prefix length /// Get common prefix length
fn shared_prefix_len(&self, elem: &[T]) -> usize; fn shared_prefix_len(&self, elem: &[T]) -> usize;
} }
impl <T> SharedPrefix<T> for Vec<T> where T: Eq { impl<T> SharedPrefix<T> for [T] where T: Eq {
fn shared_prefix_len(&self, elem: &[T]) -> usize { fn shared_prefix_len(&self, elem: &[T]) -> usize {
use std::cmp; use std::cmp;
let len = cmp::min(self.len(), elem.len()); let len = cmp::min(self.len(), elem.len());
@ -58,7 +58,7 @@ mod test {
let b = vec![1,2,3]; let b = vec![1,2,3];
assert_eq!(a.shared_prefix_len(&b), 3); assert_eq!(a.shared_prefix_len(&b), 3);
} }
#[test] #[test]
fn test_shared_prefix3() { fn test_shared_prefix3() {
let a = vec![1,2,3,4,5,6]; let a = vec![1,2,3,4,5,6];