grassrootseconomics
/
openethereum
5
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge branch 'master' into streamlined-ui

This commit is contained in:
Tomasz Drwięga 2016-07-12 12:06:01 +02:00
parent 986c28efb7 fbc0e0039a
commit 61c405a014
65 changed files with 879 additions and 601 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
Cargo.lock generated
View File

@ -3,7 +3,7 @@ name = "parity"
version = "1.3.0"
dependencies = [
"ansi_term 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.78 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.79 (registry+https://github.com/rust-lang/crates.io-index)",
"ctrlc 1.1.1 (git+https://github.com/ethcore/rust-ctrlc.git)",
"daemonize 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
"docopt 0.6.80 (registry+https://github.com/rust-lang/crates.io-index)",
@ -129,15 +129,15 @@ dependencies = [
[[package]]
name = "clippy"
version = "0.0.78"
version = "0.0.79"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"clippy_lints 0.0.78 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy_lints 0.0.79 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "clippy_lints"
version = "0.0.78"
version = "0.0.79"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"matches 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
@ -239,7 +239,7 @@ name = "ethcore"
version = "1.3.0"
dependencies = [
"bloomchain 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.78 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.79 (registry+https://github.com/rust-lang/crates.io-index)",
"crossbeam 0.2.9 (registry+https://github.com/rust-lang/crates.io-index)",
"env_logger 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
"ethash 1.3.0",
@ -267,7 +267,7 @@ dependencies = [
name = "ethcore-dapps"
version = "1.3.0"
dependencies = [
"clippy 0.0.78 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.79 (registry+https://github.com/rust-lang/crates.io-index)",
"ethcore-rpc 1.3.0",
"ethcore-util 1.3.0",
"hyper 0.9.4 (git+https://github.com/ethcore/hyper)",
@ -343,7 +343,7 @@ dependencies = [
name = "ethcore-rpc"
version = "1.3.0"
dependencies = [
"clippy 0.0.78 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.79 (registry+https://github.com/rust-lang/crates.io-index)",
"ethash 1.3.0",
"ethcore 1.3.0",
"ethcore-devtools 1.3.0",
@ -367,7 +367,7 @@ dependencies = [
name = "ethcore-signer"
version = "1.3.0"
dependencies = [
"clippy 0.0.78 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.79 (registry+https://github.com/rust-lang/crates.io-index)",
"env_logger 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
"ethcore-rpc 1.3.0",
"ethcore-util 1.3.0",
@ -387,7 +387,7 @@ dependencies = [
"arrayvec 0.3.16 (registry+https://github.com/rust-lang/crates.io-index)",
"bigint 0.1.0",
"chrono 0.2.22 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.78 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.79 (registry+https://github.com/rust-lang/crates.io-index)",
"crossbeam 0.2.9 (registry+https://github.com/rust-lang/crates.io-index)",
"elastic-array 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
"env_logger 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
@ -459,7 +459,7 @@ dependencies = [
name = "ethsync"
version = "1.3.0"
dependencies = [
"clippy 0.0.78 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.79 (registry+https://github.com/rust-lang/crates.io-index)",
"env_logger 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
"ethcore 1.3.0",
"ethcore-util 1.3.0",

9
Cargo.toml
View File

@ -22,13 +22,13 @@ fdlimit = { path = "util/fdlimit" }
num_cpus = "0.2"
number_prefix = "0.2"
rpassword = "0.2.1"
clippy = { version = "0.0.78", optional = true}
clippy = { version = "0.0.79", optional = true}
ethcore = { path = "ethcore" }
ethcore-util = { path = "util" }
ethsync = { path = "sync" }
ethcore-devtools = { path = "devtools" }
ethcore-rpc = { path = "rpc", optional = true }
ethcore-signer = { path = "signer", optional = true }
ethcore-rpc = { path = "rpc" }
ethcore-signer = { path = "signer" }
ethcore-dapps = { path = "dapps", optional = true }
semver = "0.2"
ethcore-ipc-nano = { path = "ipc/nano" }
@ -48,8 +48,7 @@ version = "0.8"
default-features = false
[features]
default = ["rpc", "dapps", "ethcore-signer"]
rpc = ["ethcore-rpc"]
default = ["dapps"]
dapps = ["ethcore-dapps"]
dev = ["clippy", "ethcore/dev", "ethcore-util/dev", "ethsync/dev", "ethcore-rpc/dev",
"ethcore-dapps/dev", "ethcore-signer/dev"]

2
dapps/Cargo.toml
View File

@ -27,7 +27,7 @@ parity-dapps-status = { git = "https://github.com/ethcore/parity-ui.git", versio
parity-dapps-home = { git = "https://github.com/ethcore/parity-ui.git", version = "0.5.2" }
parity-dapps-wallet = { git = "https://github.com/ethcore/parity-ui.git", version = "0.6.0", optional = true }
mime_guess = { version = "1.6.1" }
clippy = { version = "0.0.78", optional = true}
clippy = { version = "0.0.79", optional = true}
[build-dependencies]
serde_codegen = { version = "0.7.0", optional = true }

2
db/Cargo.toml
View File

@ -12,7 +12,7 @@ syntex = "*"
ethcore-ipc-codegen = { path = "../ipc/codegen" }
[dependencies]
clippy = { version = "0.0.78", optional = true}
clippy = { version = "0.0.79", optional = true}
ethcore-devtools = { path = "../devtools" }
ethcore-ipc = { path = "../ipc/rpc" }
rocksdb = { git = "https://github.com/ethcore/rust-rocksdb" }

2
db/src/database.rs
View File

@ -60,7 +60,7 @@ impl WriteCache {
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(
|vec_ref| match vec_ref {
&WriteCacheEntry::Write(ref val) => Some(val.clone()),

2
ethcore/Cargo.toml
View File

@ -22,7 +22,7 @@ ethcore-util = { path = "../util" }
evmjit = { path = "../evmjit", optional = true }
ethash = { path = "../ethash" }
num_cpus = "0.2"
clippy = { version = "0.0.78", optional = true}
clippy = { version = "0.0.79", optional = true}
crossbeam = "0.2.9"
lazy_static = "0.2"
ethcore-devtools = { path = "../devtools" }

20
ethcore/src/block.rs
View File

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

6
ethcore/src/block_queue.rs
View File

@ -104,7 +104,7 @@ struct VerifyingBlock {
struct QueueSignal {
deleting: Arc<AtomicBool>,
signalled: AtomicBool,
message_channel: IoChannel<NetSyncMessage>,
message_channel: IoChannel<ClientIoMessage>,
}
impl QueueSignal {
@ -116,7 +116,7 @@ impl QueueSignal {
}
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);
}
}
@ -137,7 +137,7 @@ struct Verification {
impl BlockQueue {
/// 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 {
unverified: Mutex::new(VecDeque::new()),
verified: Mutex::new(VecDeque::new()),

40
ethcore/src/client/chain_notify.rs Normal file
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
}
}

144
ethcore/src/client/client.rs
View File

@ -28,7 +28,6 @@ use std::time::Instant;
// util
use util::numbers::*;
use util::panics::*;
use util::network::*;
use util::io::*;
use util::rlp;
use util::sha3::*;
@ -47,7 +46,7 @@ use state::State;
use spec::Spec;
use engine::Engine;
use views::HeaderView;
use service::{NetSyncMessage, SyncMessage};
use service::ClientIoMessage;
use env_info::LastHashes;
use verification;
use verification::{PreverifiedBlock, Verifier};
@ -60,7 +59,7 @@ use block_queue::{BlockQueue, BlockQueueInfo};
use blockchain::{BlockChain, BlockProvider, TreeRoute, ImportRoute};
use client::{BlockID, TransactionID, UncleID, TraceId, ClientConfig,
DatabaseCompactionProfile, BlockChainClient, MiningBlockChainClient,
TraceFilter, CallAnalytics, BlockImportError, Mode};
TraceFilter, CallAnalytics, BlockImportError, Mode, ChainNotify};
use client::Error as ClientError;
use env_info::EnvInfo;
use executive::{Executive, Executed, TransactOptions, contract_address};
@ -141,7 +140,8 @@ pub struct Client {
miner: Arc<Miner>,
sleep_state: Mutex<SleepState>,
liveness: AtomicBool,
io_channel: IoChannel<NetSyncMessage>,
io_channel: IoChannel<ClientIoMessage>,
notify: RwLock<Option<Weak<ChainNotify>>>,
queue_transactions: AtomicUsize,
previous_enode: Mutex<Option<String>>,
}
@ -178,7 +178,7 @@ impl Client {
spec: Spec,
path: &Path,
miner: Arc<Miner>,
message_channel: IoChannel<NetSyncMessage>
message_channel: IoChannel<ClientIoMessage>,
) -> Result<Arc<Client>, ClientError> {
let path = get_db_path(path, config.pruning, spec.genesis_header().hash());
let gb = spec.genesis_block();
@ -228,12 +228,24 @@ impl Client {
trie_factory: TrieFactory::new(config.trie_spec),
miner: miner,
io_channel: message_channel,
notify: RwLock::new(None),
queue_transactions: AtomicUsize::new(0),
previous_enode: Mutex::new(None),
};
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.
pub fn flush_queue(&self) {
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
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 (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 mut invalid_blocks = HashSet::new();
let mut import_results = Vec::with_capacity(max_blocks_to_import);
let _import_lock = self.import_lock.lock();
let _timer = PerfTimer::new("import_verified_blocks");
let blocks = self.block_queue.drain(max_blocks_to_import);
let _import_lock = self.import_lock.lock();
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;
let original_best = self.chain_info().best_block_hash;
for block in blocks {
let header = &block.header;
for block in blocks {
let header = &block.header;
if invalid_blocks.contains(&header.parent_hash) {
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) {
invalid_blocks.insert(header.hash());
continue;
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 closed_block = self.check_and_close_block(&block);
if let Err(_) = closed_block {
invalid_blocks.insert(header.hash());
continue;
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);
}
}
let closed_block = closed_block.unwrap();
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);
}
}
(imported_blocks, import_results, invalid_blocks, original_best, imported)
};
{
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);
}
io.send(NetworkIoMessage::User(SyncMessage::NewChainBlocks {
imported: imported_blocks,
invalid: invalid_blocks,
enacted: enacted,
retracted: retracted,
sealed: Vec::new(),
})).unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e));
if let Some(notify) = self.notify() {
notify.new_blocks(
imported_blocks,
invalid_blocks,
enacted,
retracted,
Vec::new(),
);
}
}
}
@ -410,7 +426,7 @@ impl Client {
};
// 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));
// 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) {
if !self.liveness.load(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.");
}
}
@ -576,7 +594,9 @@ impl Client {
// only sleep if the import queue is mostly empty.
if self.queue_info().total_queue_size() <= MAX_QUEUE_SIZE_TO_SLEEP_ON {
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.");
} else {
trace!(target: "mode", "sleep: Cannot sleep - syncing ongoing.");
@ -901,7 +921,7 @@ impl BlockChainClient for Client {
debug!("Ignoring {} transactions: queue is full", transactions.len());
} else {
let len = transactions.len();
match self.io_channel.send(NetworkIoMessage::User(SyncMessage::NewTransactions(transactions))) {
match self.io_channel.send(ClientIoMessage::NewTransactions(transactions)) {
Ok(_) => {
self.queue_transactions.fetch_add(len, AtomicOrdering::SeqCst);
}
@ -969,13 +989,15 @@ impl MiningBlockChainClient for Client {
let (enacted, retracted) = self.calculate_enacted_retracted(&[route]);
self.miner.chain_new_blocks(self, &[h.clone()], &[], &enacted, &retracted);
self.io_channel.send(NetworkIoMessage::User(SyncMessage::NewChainBlocks {
imported: vec![h.clone()],
invalid: vec![],
enacted: enacted,
retracted: retracted,
sealed: vec![h.clone()],
})).unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e));
if let Some(notify) = self.notify() {
notify.new_blocks(
vec![h.clone()],
vec![],
enacted,
retracted,
vec![h.clone()],
);
}
}
if self.chain_info().best_block_hash != original_best {

2
ethcore/src/client/mod.rs
View File

@ -20,6 +20,7 @@ mod config;
mod error;
mod test_client;
mod trace;
mod chain_notify;
pub use self::client::*;
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 executive::{Executed, Executive, TransactOptions};
pub use env_info::{LastHashes, EnvInfo};
pub use self::chain_notify::ChainNotify;
use util::bytes::Bytes;
use util::hash::{Address, H256, H2048};

60
ethcore/src/evm/evm.rs
View File

@ -107,6 +107,9 @@ pub trait CostType: ops::Mul<Output=Self> + ops::Div<Output=Self> + ops::Add<Out
fn overflow_add(self, other: Self) -> (Self, bool);
/// Multiple with overflow
fn overflow_mul(self, other: Self) -> (Self, bool);
/// Single-step full multiplication and division: `self*other/div`
/// Should not overflow on intermediate steps
fn overflow_mul_div(self, other: Self, div: Self) -> (Self, bool);
}
impl CostType for U256 {
@ -129,6 +132,17 @@ impl CostType for U256 {
fn overflow_mul(self, other: Self) -> (Self, bool) {
Uint::overflowing_mul(self, other)
}
fn overflow_mul_div(self, other: Self, div: Self) -> (Self, bool) {
let x = self.full_mul(other);
let (U512(parts), o) = Uint::overflowing_div(x, U512::from(div));
let overflow = (parts[4] | parts[5] | parts[6] | parts[7]) > 0;
(
U256([parts[0], parts[1], parts[2], parts[3]]),
o | overflow
)
}
}
impl CostType for usize {
@ -154,6 +168,14 @@ impl CostType for usize {
fn overflow_mul(self, other: Self) -> (Self, bool) {
self.overflowing_mul(other)
}
fn overflow_mul_div(self, other: Self, div: Self) -> (Self, bool) {
let (c, o) = U128::from(self).overflowing_mul(U128::from(other));
let (U128(parts), o1) = c.overflowing_div(U128::from(div));
let result = parts[0] as usize;
let overflow = o | o1 | (parts[1] > 0) | (parts[0] > result as u64);
(result, overflow)
}
}
/// Evm interface
@ -164,3 +186,41 @@ pub trait Evm {
/// to compute the final gas left.
fn exec(&mut self, params: ActionParams, ext: &mut Ext) -> Result<GasLeft>;
}
#[test]
fn should_calculate_overflow_mul_div_without_overflow() {
// given
let num = 10_000_000;
// when
let (res1, o1) = U256::from(num).overflow_mul_div(U256::from(num), U256::from(num));
let (res2, o2) = num.overflow_mul_div(num, num);
// then
assert_eq!(res1, U256::from(num));
assert!(!o1);
assert_eq!(res2, num);
assert!(!o2);
}
#[test]
fn should_calculate_overflow_mul_div_with_overflow() {
// given
let max = ::std::u64::MAX;
let num1 = U256([max, max, max, max]);
let num2 = ::std::usize::MAX;
// when
let (res1, o1) = num1.overflow_mul_div(num1, num1 - U256::from(2));
let (res2, o2) = num2.overflow_mul_div(num2, num2 - 2);
// then
// (x+2)^2/x = (x^2 + 4x + 4)/x = x + 4 + 4/x ~ (MAX-2) + 4 + 0 = 1
assert_eq!(res2, 1);
assert!(o2);
assert_eq!(res1, U256::from(1));
assert!(o1);
}

18
ethcore/src/evm/interpreter/gasometer.rs
View File

@ -68,6 +68,9 @@ impl<Gas: CostType> Gasometer<Gas> {
let default_gas = Gas::from(schedule.tier_step_gas[tier]);
let cost = match instruction {
instructions::JUMPDEST => {
InstructionCost::Gas(Gas::from(1))
},
instructions::SSTORE => {
let address = H256::from(stack.peek(0));
let newval = stack.peek(1);
@ -106,9 +109,6 @@ impl<Gas: CostType> Gasometer<Gas> {
instructions::EXTCODECOPY => {
InstructionCost::GasMemCopy(default_gas, try!(self.mem_needed(stack.peek(1), stack.peek(3))), try!(Gas::from_u256(*stack.peek(3))))
},
instructions::JUMPDEST => {
InstructionCost::Gas(Gas::from(1))
},
instructions::LOG0...instructions::LOG4 => {
let no_of_topics = instructions::get_log_topics(instruction);
let log_gas = schedule.log_gas + schedule.log_topic_gas * no_of_topics;
@ -199,14 +199,12 @@ impl<Gas: CostType> Gasometer<Gas> {
let s = mem_size >> 5;
// s * memory_gas + s * s / quad_coeff_div
let a = overflowing!(s.overflow_mul(Gas::from(schedule.memory_gas)));
// We need to go to U512 to calculate s*s/quad_coeff_div
let b = U512::from(s.as_u256()) * U512::from(s.as_u256()) / U512::from(schedule.quad_coeff_div);
if b > U512::from(!U256::zero()) {
Err(evm::Error::OutOfGas)
} else {
Ok(overflowing!(a.overflow_add(try!(Gas::from_u256(U256::from(b))))))
}
// Calculate s*s/quad_coeff_div
let b = overflowing!(s.overflow_mul_div(s, Gas::from(schedule.quad_coeff_div)));
Ok(overflowing!(a.overflow_add(b)))
};
let current_mem_size = Gas::from(current_mem_size);
let req_mem_size_rounded = (overflowing!(mem_size.overflow_add(Gas::from(31 as usize))) >> 5) << 5;

2
ethcore/src/header.rs
View File

@ -143,7 +143,7 @@ impl Header {
/// Get the difficulty field of the header.
pub fn difficulty(&self) -> &U256 { &self.difficulty }
/// 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.

6
ethcore/src/migrations/state/v7.rs
View File

@ -61,7 +61,7 @@ fn attempt_migrate(mut key_h: H256, val: &[u8]) -> Option<H256> {
}
}
/// Version for ArchiveDB.
/// Version for `ArchiveDB`.
pub struct ArchiveV7;
impl SimpleMigration for ArchiveV7 {
@ -91,7 +91,7 @@ const V7_VERSION_KEY: &'static [u8] = &[ b'j', b'v', b'e', b'r', 0, 0, 0, 0, 0,
const DB_VERSION: u32 = 0x203;
const PADDING : [u8; 10] = [0u8; 10];
/// Version for OverlayRecent database.
/// Version for `OverlayRecent` database.
/// more involved than the archive version because of journaling.
#[derive(Default)]
pub struct OverlayRecentV7 {
@ -244,4 +244,4 @@ impl Migration for OverlayRecentV7 {
try!(self.walk_journal(source));
self.migrate_journal(source, batch, dest)
}
}
}

4
ethcore/src/miner/miner.rs
View File

@ -665,7 +665,7 @@ impl MinerService for Miner {
};
match (&self.options.pending_set, sealing_set) {
(&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()
.map(|t| t.hash());
let receipts = pending.receipts().clone().into_iter();
let receipts = pending.receipts().iter().cloned();
hashes.zip(receipts).collect()
},

79
ethcore/src/service.rs
View File

@ -20,67 +20,50 @@ use util::*;
use util::panics::*;
use spec::Spec;
use error::*;
use client::{Client, ClientConfig};
use client::{Client, ClientConfig, ChainNotify};
use miner::Miner;
/// Message type for external and internal events
#[derive(Clone)]
pub enum SyncMessage {
/// 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>,
},
pub enum ClientIoMessage {
/// Best Block Hash in chain has been changed
NewChainHead,
/// A block is ready
BlockVerified,
/// New transaction RLPs are ready to be imported
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.
pub struct ClientService {
net_service: Arc<NetworkService<SyncMessage>>,
io_service: Arc<IoService<ClientIoMessage>>,
client: Arc<Client>,
panic_handler: Arc<PanicHandler>
}
impl ClientService {
/// 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 net_service = try!(NetworkService::new(net_config));
panic_handler.forward_from(&net_service);
if enable_network {
try!(net_service.start());
}
let io_service = try!(IoService::<ClientIoMessage>::start());
panic_handler.forward_from(&io_service);
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());
let client_io = Arc::new(ClientIoHandler {
client: client.clone()
});
try!(net_service.io().register_handler(client_io));
try!(io_service.register_handler(client_io));
Ok(ClientService {
net_service: Arc::new(net_service),
io_service: Arc::new(io_service),
client: client,
panic_handler: panic_handler,
})
@ -92,8 +75,8 @@ impl ClientService {
}
/// Get general IO interface
pub fn register_io_handler(&self, handler: Arc<IoHandler<NetSyncMessage> + Send>) -> Result<(), IoError> {
self.net_service.io().register_handler(handler)
pub fn register_io_handler(&self, handler: Arc<IoHandler<ClientIoMessage> + Send>) -> Result<(), IoError> {
self.io_service.register_handler(handler)
}
/// Get client interface
@ -102,8 +85,13 @@ impl ClientService {
}
/// Get network service component
pub fn network(&mut self) -> Arc<NetworkService<SyncMessage>> {
self.net_service.clone()
pub fn io(&self) -> Arc<IoService<ClientIoMessage>> {
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_MS: u64 = 5000;
impl IoHandler<NetSyncMessage> for ClientIoHandler {
fn initialize(&self, io: &IoContext<NetSyncMessage>) {
impl IoHandler<ClientIoMessage> for ClientIoHandler {
fn initialize(&self, io: &IoContext<ClientIoMessage>) {
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 {
self.client.tick();
}
}
#[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 {
UserMessage(ref message) => match *message {
SyncMessage::BlockVerified => { self.client.import_verified_blocks(&io.channel()); }
SyncMessage::NewTransactions(ref transactions) => { self.client.import_queued_transactions(&transactions); }
_ => {} // ignore other messages
},
NetworkIoMessage::NetworkStarted(ref url) => { self.client.network_started(url); }
ClientIoMessage::BlockVerified => { self.client.import_verified_blocks(&io.channel()); }
ClientIoMessage::NewTransactions(ref transactions) => { self.client.import_queued_transactions(&transactions); }
_ => {} // ignore other messages
}
}
@ -150,7 +134,6 @@ impl IoHandler<NetSyncMessage> for ClientIoHandler {
mod tests {
use super::*;
use tests::helpers::*;
use util::network::*;
use devtools::*;
use client::ClientConfig;
use std::sync::Arc;
@ -162,10 +145,8 @@ mod tests {
let service = ClientService::start(
ClientConfig::default(),
get_test_spec(),
NetworkConfiguration::new_local(),
&temp_path.as_path(),
Arc::new(Miner::with_spec(get_test_spec())),
false
);
assert!(service.is_ok());
}

2
ethcore/src/spec/spec.rs
View File

@ -143,7 +143,7 @@ impl Spec {
}
/// 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.
pub fn network_id(&self) -> U256 { self.params.network_id }

8
evmbin/Cargo.toml
View File

@ -4,6 +4,14 @@ description = "Parity's EVM implementation"
version = "0.1.0"
authors = ["Ethcore <admin@ethcore.io>"]
[lib]
name = "evm"
path = "./src/main.rs"
[[bin]]
name = "evm"
path = "./src/main.rs"
[dependencies]
rustc-serialize = "0.3"
docopt = { version = "0.6" }

24
evmbin/bench.sh Executable file
View File

@ -0,0 +1,24 @@
#!/bin/bash
set -x
set -e
cargo build --release
# LOOP TEST
CODE1=606060405260005b620f42408112156019575b6001016007565b600081905550600680602b6000396000f3606060405200
ethvm --code $CODE1
echo "^^^^ ethvm"
./target/release/evm stats --code $CODE1 --gas 4402000
echo "^^^^ usize"
./target/release/evm stats --code $CODE1
echo "^^^^ U256"
# RNG TEST
CODE2=6060604052600360056007600b60005b620f4240811215607f5767ffe7649d5eca84179490940267f47ed85c4b9a6379019367f8e5dd9a5c994bba9390930267f91d87e4b8b74e55019267ff97f6f3b29cda529290920267f393ada8dd75c938019167fe8d437c45bb3735830267f47d9a7b5428ffec019150600101600f565b838518831882186000555050505050600680609a6000396000f3606060405200
ethvm --code $CODE2
echo "^^^^ ethvm"
./target/release/evm stats --code $CODE2 --gas 143020115
echo "^^^^ usize"
./target/release/evm stats --code $CODE2
echo "^^^^ U256"

85
evmbin/benches/mod.rs Normal file
View File

@ -0,0 +1,85 @@
// 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/>.
//! benchmarking for EVM
//! should be started with:
//! ```bash
//! multirust run nightly cargo bench
//! ```
#![feature(test)]
extern crate test;
extern crate ethcore;
extern crate evm;
extern crate ethcore_util;
extern crate rustc_serialize;
use self::test::{Bencher, black_box};
use evm::run_vm;
use ethcore::action_params::ActionParams;
use ethcore_util::{U256, Uint};
use rustc_serialize::hex::FromHex;
#[bench]
fn simple_loop_usize(b: &mut Bencher) {
simple_loop(U256::from(::std::usize::MAX), b)
}
#[bench]
fn simple_loop_u256(b: &mut Bencher) {
simple_loop(!U256::zero(), b)
}
fn simple_loop(gas: U256, b: &mut Bencher) {
let code = black_box(
"606060405260005b620042408112156019575b6001016007565b600081905550600680602b6000396000f3606060405200".from_hex().unwrap()
);
b.iter(|| {
let mut params = ActionParams::default();
params.gas = gas;
params.code = Some(code.clone());
run_vm(params)
});
}
#[bench]
fn rng_usize(b: &mut Bencher) {
rng(U256::from(::std::usize::MAX), b)
}
#[bench]
fn rng_u256(b: &mut Bencher) {
rng(!U256::zero(), b)
}
fn rng(gas: U256, b: &mut Bencher) {
let code = black_box(
"6060604052600360056007600b60005b62004240811215607f5767ffe7649d5eca84179490940267f47ed85c4b9a6379019367f8e5dd9a5c994bba9390930267f91d87e4b8b74e55019267ff97f6f3b29cda529290920267f393ada8dd75c938019167fe8d437c45bb3735830267f47d9a7b5428ffec019150600101600f565b838518831882186000555050505050600680609a6000396000f3606060405200".from_hex().unwrap()
);
b.iter(|| {
let mut params = ActionParams::default();
params.gas = gas;
params.code = Some(code.clone());
run_vm(params)
});
}

30
evmbin/src/main.rs
View File

@ -17,6 +17,7 @@
//! Parity EVM interpreter binary.
#![warn(missing_docs)]
#![allow(dead_code)]
extern crate ethcore;
extern crate rustc_serialize;
extern crate docopt;
@ -25,7 +26,7 @@ extern crate ethcore_util as util;
mod ext;
use std::time::Instant;
use std::time::{Instant, Duration};
use std::str::FromStr;
use docopt::Docopt;
use util::{U256, FromHex, Uint, Bytes};
@ -58,6 +59,15 @@ fn main() {
params.code = Some(args.code());
params.data = args.data();
let result = run_vm(params);
println!("Gas used: {:?}", result.gas_used);
println!("Output: {:?}", result.output);
println!("Time: {}.{:.9}s", result.time.as_secs(), result.time.subsec_nanos());
}
/// Execute VM with given `ActionParams`
pub fn run_vm(params: ActionParams) -> ExecutionResults {
let initial_gas = params.gas;
let factory = Factory::new(VMType::Interpreter);
let mut vm = factory.create(params.gas);
let mut ext = ext::FakeExt::default();
@ -66,9 +76,21 @@ fn main() {
let gas_left = vm.exec(params, &mut ext).finalize(ext).expect("OK");
let duration = start.elapsed();
println!("Gas used: {:?}", args.gas() - gas_left);
println!("Output: {:?}", "");
println!("Time: {}.{:.9}s", duration.as_secs(), duration.subsec_nanos());
ExecutionResults {
gas_used: initial_gas - gas_left,
output: Vec::new(),
time: duration,
}
}
/// VM execution results
pub struct ExecutionResults {
/// Used gas
pub gas_used: U256,
/// Output as bytes
pub output: Vec<u8>,
/// Time Taken
pub time: Duration,
}
#[derive(Debug, RustcDecodable)]

32
ipc/codegen/src/codegen.rs
View File

@ -62,24 +62,10 @@ pub fn expand_ipc_implementation(
};
push_client(cx, &builder, &interface_map, push);
push_handshake_struct(cx, push);
push(Annotatable::Item(interface_map.item));
}
fn push_handshake_struct(cx: &ExtCtxt, push: &mut FnMut(Annotatable)) {
let handshake_item = quote_item!(cx,
#[derive(Binary)]
pub struct BinHandshake {
api_version: String,
protocol_version: String,
reserved: Vec<u8>,
}
).unwrap();
push(Annotatable::Item(handshake_item));
}
fn field_name(builder: &aster::AstBuilder, arg: &Arg) -> ast::Ident {
match arg.pat.node {
PatKind::Ident(_, ref ident, _) => builder.id(ident.node),
@ -601,15 +587,14 @@ fn push_client_implementation(
let handshake_item = quote_impl_item!(cx,
pub fn handshake(&self) -> Result<(), ::ipc::Error> {
let payload = BinHandshake {
protocol_version: $item_ident::protocol_version().to_string(),
api_version: $item_ident::api_version().to_string(),
reserved: vec![0u8; 64],
let payload = ::ipc::Handshake {
protocol_version: $item_ident::protocol_version(),
api_version: $item_ident::api_version(),
};
::ipc::invoke(
0,
&Some(::ipc::binary::serialize(&payload).unwrap()),
&Some(::ipc::binary::serialize(&::ipc::BinHandshake::from(payload)).unwrap()),
&mut *self.socket.write().unwrap());
let mut result = vec![0u8; 1];
@ -673,18 +658,15 @@ fn implement_handshake_arm(
) -> (ast::Arm, ast::Arm)
{
let handshake_deserialize = quote_stmt!(&cx,
let handshake_payload = ::ipc::binary::deserialize_from::<BinHandshake, _>(r).unwrap();
let handshake_payload = ::ipc::binary::deserialize_from::<::ipc::BinHandshake, _>(r).unwrap();
);
let handshake_deserialize_buf = quote_stmt!(&cx,
let handshake_payload = ::ipc::binary::deserialize::<BinHandshake>(buf).unwrap();
let handshake_payload = ::ipc::binary::deserialize::<::ipc::BinHandshake>(buf).unwrap();
);
let handshake_serialize = quote_expr!(&cx,
::ipc::binary::serialize::<bool>(&Self::handshake(&::ipc::Handshake {
api_version: ::semver::Version::parse(&handshake_payload.api_version).unwrap(),
protocol_version: ::semver::Version::parse(&handshake_payload.protocol_version).unwrap(),
})).unwrap()
::ipc::binary::serialize::<bool>(&Self::handshake(&handshake_payload.to_semver())).unwrap()
);
(

2
ipc/rpc/Cargo.toml
View File

@ -8,6 +8,6 @@ license = "GPL-3.0"
[dependencies]
ethcore-devtools = { path = "../../devtools" }
semver = "0.2.0"
nanomsg = { git = "https://github.com/ethcore/nanomsg.rs.git" }
ethcore-util = { path = "../../util" }
semver = "0.2"

79
ipc/rpc/src/binary.rs
View File

@ -21,6 +21,7 @@ use util::numbers::{U256, U512, H256, H2048, Address};
use std::mem;
use std::collections::{VecDeque, BTreeMap};
use std::ops::Range;
use super::Handshake;
#[derive(Debug)]
pub struct BinaryConvertError;
@ -554,6 +555,61 @@ macro_rules! binary_fixed_size {
}
}
/// Fixed-sized version of Handshake struct
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BinHandshake {
api_version: BinVersion,
protocol_version: BinVersion,
}
/// Shorten version of semver Version without `pre` and `build` information
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BinVersion {
pub major: u64,
pub minor: u64,
pub patch: u64,
}
impl From<Handshake> for BinHandshake {
fn from(other: Handshake) -> Self {
BinHandshake {
api_version: BinVersion::from(other.api_version),
protocol_version: BinVersion::from(other.protocol_version),
}
}
}
impl BinHandshake {
pub fn to_semver(self) -> Handshake {
Handshake {
api_version: self.api_version.to_semver(),
protocol_version: self.protocol_version.to_semver(),
}
}
}
impl BinVersion {
pub fn to_semver(self) -> ::semver::Version {
::semver::Version {
major: self.major,
minor: self.minor,
patch: self.patch,
pre: vec![],
build: vec![],
}
}
}
impl From<::semver::Version> for BinVersion {
fn from(other: ::semver::Version) -> Self {
BinVersion {
major: other.major,
minor: other.minor,
patch: other.patch,
}
}
}
binary_fixed_size!(u64);
binary_fixed_size!(u32);
binary_fixed_size!(usize);
@ -564,6 +620,7 @@ binary_fixed_size!(U512);
binary_fixed_size!(H256);
binary_fixed_size!(H2048);
binary_fixed_size!(Address);
binary_fixed_size!(BinHandshake);
#[test]
fn vec_serialize() {
@ -706,8 +763,6 @@ fn serialize_opt_vec() {
#[test]
fn serialize_opt_vec_payload() {
use std::io::Cursor;
let optional_vec: Option<Vec<u8>> = None;
let payload = serialize(&optional_vec).unwrap();
@ -776,3 +831,23 @@ fn serialize_btree() {
assert_eq!(res[&1u64], 5u64);
}
#[test]
fn serialize_handshake() {
use std::io::{Cursor, SeekFrom, Seek};
let mut buff = Cursor::new(Vec::new());
let handshake = Handshake {
api_version: ::semver::Version::parse("1.2.0").unwrap(),
protocol_version: ::semver::Version::parse("1.2.0").unwrap(),
};
serialize_into(&BinHandshake::from(handshake.clone()), &mut buff).unwrap();
buff.seek(SeekFrom::Start(0)).unwrap();
let res = deserialize_from::<BinHandshake, _>(&mut buff).unwrap().to_semver();
assert_eq!(res, handshake);
}

1
ipc/rpc/src/interface.rs
View File

@ -20,6 +20,7 @@ use std::io::{Read, Write};
use std::marker::Sync;
use semver::Version;
#[derive(Debug, PartialEq, Eq, Clone)]
/// Handshake for client and server to negotiate api/protocol version
pub struct Handshake {
pub protocol_version: Version,

2
ipc/rpc/src/lib.rs
View File

@ -24,4 +24,4 @@ extern crate ethcore_util as util;
pub mod interface;
pub mod binary;
pub use interface::{IpcInterface, IpcSocket, invoke, IpcConfig, Handshake, Error, WithSocket};
pub use binary::{BinaryConvertable, BinaryConvertError};
pub use binary::{BinaryConvertable, BinaryConvertError, BinHandshake};

2
json/Cargo.toml
View File

@ -10,7 +10,7 @@ rustc-serialize = "0.3"
serde = "0.7.0"
serde_json = "0.7.0"
serde_macros = { version = "0.7.0", optional = true }
clippy = { version = "0.0.78", optional = true}
clippy = { version = "0.0.79", optional = true}
[build-dependencies]
serde_codegen = { version = "0.7.0", optional = true }

4
json/src/bytes.rs
View File

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

2
parity/configuration.rs
View File

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

23
parity/informant.rs
View File

@ -21,8 +21,8 @@ use self::ansi_term::Style;
use std::time::{Instant, Duration};
use std::sync::RwLock;
use std::ops::{Deref, DerefMut};
use ethsync::{EthSync, SyncProvider};
use util::{Uint, RwLockable, NetworkService};
use ethsync::SyncStatus;
use util::{Uint, RwLockable, NetworkConfiguration};
use ethcore::client::*;
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();
if elapsed < Duration::from_secs(5) {
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!("{:3}", ((report.gas_processed - last_report.gas_processed) / From::from(elapsed.as_milliseconds() * 1000)).low_u64())),
match maybe_sync {
Some((sync, net)) => {
let sync_info = sync.status();
let net_config = net.config();
match maybe_status {
Some((ref sync_info, ref net_config)) => {
format!("{}/{}/{} peers {} ",
paint(Green.bold(), format!("{:2}", sync_info.num_active_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(cache_info.total()))),
paint(Purple.bold(), format!("{:>8}", Informant::format_bytes(queue_info.mem_used))),
match maybe_sync {
Some((sync, _)) => {
let sync_info = sync.status();
format!(" {} sync", paint(Purple.bold(), format!("{:>8}", Informant::format_bytes(sync_info.mem_used))))
}
None => String::new()
},
if let Some((ref sync_info, _)) = maybe_status {
format!(" {} sync", paint(Purple.bold(), format!("{:>8}", Informant::format_bytes(sync_info.mem_used))))
} else { String::new() },
);
}

40
parity/io_handler.rs
View File

@ -14,12 +14,12 @@
// You should have received a copy of the GNU General Public License
// 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::service::{NetSyncMessage, SyncMessage};
use ethsync::EthSync;
use ethcore::service::ClientIoMessage;
use ethsync::{EthSync, SyncProvider, ManageNetwork};
use ethcore::account_provider::AccountProvider;
use util::{TimerToken, IoHandler, IoContext, NetworkService, NetworkIoMessage};
use util::{TimerToken, IoHandler, IoContext};
use informant::Informant;
@ -30,38 +30,18 @@ pub struct ClientIoHandler {
pub sync: Arc<EthSync>,
pub accounts: Arc<AccountProvider>,
pub info: Informant,
pub network: Weak<NetworkService<SyncMessage>>,
}
impl IoHandler<NetSyncMessage> for ClientIoHandler {
fn initialize(&self, io: &IoContext<NetSyncMessage>) {
impl IoHandler<ClientIoMessage> for ClientIoHandler {
fn initialize(&self, io: &IoContext<ClientIoMessage>) {
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 Some(net) = self.network.upgrade() {
self.info.tick(&self.client, Some((&self.sync, &net)));
}
}
}
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 */},
let sync_status = self.sync.status();
let network_config = self.sync.network_config();
self.info.tick(&self.client, Some((sync_status, network_config)));
}
}
}

62
parity/main.rs
View File

@ -46,16 +46,14 @@ extern crate hyper; // for price_info.rs
extern crate json_ipc_server as jsonipc;
extern crate ethcore_ipc_hypervisor as hypervisor;
#[cfg(feature = "rpc")]
extern crate ethcore_rpc;
extern crate ethcore_signer;
extern crate ansi_term;
#[cfg(feature = "dapps")]
extern crate ethcore_dapps;
#[cfg(feature = "ethcore-signer")]
extern crate ethcore_signer;
#[macro_use]
mod die;
mod upgrade;
@ -81,9 +79,10 @@ use std::thread::sleep;
use std::time::Duration;
use rustc_serialize::hex::FromHex;
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 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::service::ClientService;
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);
// Build client
let mut service = ClientService::start(
let service = ClientService::start(
client_config,
spec,
net_settings,
Path::new(&conf.path()),
miner.clone(),
match conf.mode() { Mode::Dark(..) => false, _ => !conf.args.flag_no_network }
).unwrap_or_else(|e| die_with_error("Client", e));
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());
// Sync
let sync = EthSync::new(sync_config, client.clone());
EthSync::register(&*service.network(), sync.clone()).unwrap_or_else(|e| die_with_error("Error registering eth protocol handler", UtilError::from(e).into()));
let sync = EthSync::new(sync_config, client.clone(), net_settings)
.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 {
signer_port: conf.signer_port(),
@ -261,7 +264,7 @@ fn execute_client(conf: Configuration, spec: Spec, client_config: ClientConfig)
logger: logger.clone(),
settings: network_settings.clone(),
allow_pending_receipt_query: !conf.args.flag_geth,
net_service: service.network(),
net_service: sync.clone(),
});
let dependencies = rpc::Dependencies {
@ -311,7 +314,6 @@ fn execute_client(conf: Configuration, spec: Spec, client_config: ClientConfig)
info: Informant::new(conf.have_color()),
sync: sync.clone(),
accounts: account_service.clone(),
network: Arc::downgrade(&service.network()),
});
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(); }
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);
// Build client
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));
panic_handler.forward_from(&service);
@ -419,24 +408,11 @@ fn execute_import(conf: Configuration) {
unsafe { ::fdlimit::raise_fd_limit(); }
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);
// Build client
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));
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(e) => die!("Cannot import block: {:?}", e)
}
informant.tick::<&'static ()>(client.deref(), None);
informant.tick(client.deref(), None);
};
match format {

21
parity/rpc.rs
View File

@ -24,12 +24,8 @@ use jsonipc;
use rpc_apis;
use std::fmt;
#[cfg(feature = "rpc")]
pub use ethcore_rpc::Server as RpcServer;
#[cfg(feature = "rpc")]
use ethcore_rpc::{RpcServerError, RpcServer as Server};
#[cfg(not(feature = "rpc"))]
pub struct RpcServer;
pub struct HttpConfiguration {
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))
}
#[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(
dependencies: &Dependencies,
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> {
if !conf.enabled { return None; }
let apis = conf.apis.split(',').collect();
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 {
let server = setup_rpc_server(apis, dependencies);
match server.start_ipc(addr) {

10
parity/rpc_apis.rs
View File

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

30
parity/signer.rs
View File

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

2
rpc/Cargo.toml
View File

@ -23,7 +23,7 @@ ethcore-devtools = { path = "../devtools" }
rustc-serialize = "0.3"
transient-hashmap = "0.1"
serde_macros = { version = "0.7.0", optional = true }
clippy = { version = "0.0.78", optional = true}
clippy = { version = "0.0.79", optional = true}
json-ipc-server = { git = "https://github.com/ethcore/json-ipc-server.git" }
ethcore-ipc = { path = "../ipc/rpc" }

2
rpc/src/v1/impls/eth.rs
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> {
try!(self.active());
match params {
Params::None => to_value(&vec![] as &Vec<String>),
Params::None => to_value(&(&[] as &[String])),
_ => Err(Error::invalid_params())
}
}

22
rpc/src/v1/impls/ethcore_set.rs
View File

@ -19,26 +19,26 @@ use std::sync::{Arc, Weak};
use jsonrpc_core::*;
use ethcore::miner::MinerService;
use ethcore::client::MiningBlockChainClient;
use ethcore::service::SyncMessage;
use util::network::{NetworkService, NonReservedPeerMode};
use ethsync::ManageNetwork;
use util::network::NonReservedPeerMode;
use v1::traits::EthcoreSet;
use v1::types::{Bytes, H160, U256};
/// Ethcore-specific rpc interface for operations altering the settings.
pub struct EthcoreSetClient<C, M> where
C: MiningBlockChainClient,
M: MinerService {
M: MinerService
{
client: Weak<C>,
miner: Weak<M>,
net: Weak<NetworkService<SyncMessage>>,
net: Weak<ManageNetwork>,
}
impl<C, M> EthcoreSetClient<C, M> where
C: MiningBlockChainClient,
M: MinerService {
/// 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 {
client: Arc::downgrade(client),
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);
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))
}
}

9
rpc/src/v1/impls/net.rs
View File

@ -48,13 +48,4 @@ impl<S> Net for NetClient<S> where S: SyncProvider + 'static {
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))
}
}

6
rpc/src/v1/tests/helpers/sync_provider.rs
View File

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

12
rpc/src/v1/tests/mocked/ethcore_set.rs
View File

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

30
rpc/src/v1/tests/mocked/manage_network.rs Normal file
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() }
}

1
rpc/src/v1/tests/mocked/mod.rs
View File

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

6
rpc/src/v1/traits/ethcore_set.rs
View File

@ -55,6 +55,12 @@ pub trait EthcoreSet: Sized + Send + Sync + 'static {
/// Accept non-reserved peers (default behavior)
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.
fn to_delegate(self) -> IoDelegate<Self> {
let mut delegate = IoDelegate::new(Arc::new(self));

6
rpc/src/v1/traits/net.rs
View File

@ -30,12 +30,6 @@ pub trait Net: Sized + Send + Sync + 'static {
/// Otherwise false.
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.
fn to_delegate(self) -> IoDelegate<Self> {
let mut delegate = IoDelegate::new(Arc::new(self));

2
signer/Cargo.toml
View File

@ -20,7 +20,7 @@ ethcore-util = { path = "../util" }
ethcore-rpc = { path = "../rpc" }
parity-dapps-signer = { git = "https://github.com/ethcore/parity-ui.git", version = "0.2.0" }
clippy = { version = "0.0.78", optional = true}
clippy = { version = "0.0.79", optional = true}
[features]
dev = ["clippy"]

8
signer/src/ws_server/session.rs
View File

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

2
sync/Cargo.toml
View File

@ -10,7 +10,7 @@ authors = ["Ethcore <admin@ethcore.io"]
[dependencies]
ethcore-util = { path = "../util" }
ethcore = { path = "../ethcore" }
clippy = { version = "0.0.78", optional = true}
clippy = { version = "0.0.79", optional = true}
log = "0.3"
env_logger = "0.3"
time = "0.1.34"

5
sync/src/io.rs
View File

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

186
sync/src/lib.rs
View File

@ -34,15 +34,14 @@
//! extern crate ethsync;
//! use std::env;
//! use std::sync::Arc;
//! use util::network::{NetworkService, NetworkConfiguration};
//! use util::network::{NetworkConfiguration};
//! use util::io::IoChannel;
//! use ethcore::client::{Client, ClientConfig};
//! use ethsync::{EthSync, SyncConfig};
//! use ethsync::{EthSync, SyncConfig, ManageNetwork};
//! use ethcore::ethereum;
//! use ethcore::miner::{GasPricer, Miner};
//!
//! fn main() {
//! let mut service = NetworkService::new(NetworkConfiguration::new()).unwrap();
//! service.start().unwrap();
//! let dir = env::temp_dir();
//! let miner = Miner::new(
//! Default::default(),
@ -55,10 +54,10 @@
//! ethereum::new_frontier(),
//! &dir,
//! miner,
//! service.io().channel()
//! IoChannel::disconnected()
//! ).unwrap();
//! let sync = EthSync::new(SyncConfig::default(), client);
//! EthSync::register(&mut service, sync);
//! let sync = EthSync::new(SyncConfig::default(), client, NetworkConfiguration::new()).unwrap();
//! sync.start_network();
//! }
//! ```
@ -75,13 +74,10 @@ extern crate heapsize;
use std::ops::*;
use std::sync::*;
use util::network::{NetworkProtocolHandler, NetworkService, NetworkContext, PeerId};
use util::{TimerToken, U256, RwLockable};
use ethcore::client::Client;
use ethcore::service::{SyncMessage, NetSyncMessage};
use util::network::{NetworkProtocolHandler, NetworkService, NetworkContext, PeerId, NetworkConfiguration};
use util::{TimerToken, U256, H256, RwLockable, UtilError};
use ethcore::client::{Client, ChainNotify};
use io::NetSyncIo;
use util::io::IoChannel;
use util::{NetworkIoMessage, NetworkError};
use chain::ChainSync;
mod chain;
@ -91,6 +87,9 @@ mod io;
#[cfg(test)]
mod tests;
/// Ethereum sync protocol
pub const ETH_PROTOCOL: &'static str = "eth";
/// Sync configuration
pub struct SyncConfig {
/// Max blocks to download ahead
@ -112,99 +111,142 @@ impl Default for SyncConfig {
pub trait SyncProvider: Send + Sync {
/// Get sync status
fn status(&self) -> SyncStatus;
/// Start the network
fn start_network(&self);
/// Stop the network
fn stop_network(&self);
}
/// Ethereum network protocol handler
pub struct EthSync {
/// Shared blockchain client. TODO: this should evetually become an IPC endpoint
chain: Arc<Client>,
/// Sync strategy
sync: RwLock<ChainSync>,
/// IO communication chnnel.
io_channel: RwLock<IoChannel<NetSyncMessage>>,
/// Network service
network: NetworkService,
/// Protocol handler
handler: Arc<SyncProtocolHandler>,
}
pub use self::chain::{SyncStatus, SyncState};
impl EthSync {
/// Creates and register protocol with the network service
pub fn new(config: SyncConfig, chain: Arc<Client>) -> Arc<EthSync> {
let sync = ChainSync::new(config, chain.deref());
Arc::new(EthSync {
chain: chain,
sync: RwLock::new(sync),
io_channel: RwLock::new(IoChannel::disconnected()),
})
}
pub fn new(config: SyncConfig, chain: Arc<Client>, network_config: NetworkConfiguration) -> Result<Arc<EthSync>, UtilError> {
let chain_sync = ChainSync::new(config, chain.deref());
let service = try!(NetworkService::new(network_config));
let sync = Arc::new(EthSync{
network: service,
handler: Arc::new(SyncProtocolHandler { sync: RwLock::new(chain_sync), chain: chain }),
});
/// Register protocol with the network service
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()));
Ok(sync)
}
}
impl SyncProvider for EthSync {
/// Get sync status
fn status(&self) -> SyncStatus {
self.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));
self.handler.sync.unwrapped_read().status()
}
}
impl NetworkProtocolHandler<SyncMessage> for EthSync {
fn initialize(&self, io: &NetworkContext<SyncMessage>) {
struct SyncProtocolHandler {
/// 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");
*self.io_channel.unwrapped_write() = io.io_channel();
}
fn read(&self, io: &NetworkContext<SyncMessage>, peer: &PeerId, packet_id: u8, data: &[u8]) {
ChainSync::dispatch_packet(&self.sync, &mut NetSyncIo::new(io, self.chain.deref()) , *peer, packet_id, data);
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);
}
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);
}
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);
}
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_sync(&mut NetSyncIo::new(io, self.chain.deref()));
}
}
#[cfg_attr(feature="dev", allow(single_match))]
fn message(&self, io: &NetworkContext<SyncMessage>, message: &SyncMessage) {
match *message {
SyncMessage::NewChainBlocks { ref imported, ref invalid, ref enacted, ref retracted, ref sealed } => {
let mut sync_io = NetSyncIo::new(io, self.chain.deref());
self.sync.unwrapped_write().chain_new_blocks(&mut sync_io, imported, invalid, enacted, retracted, sealed);
},
_ => {/* Ignore other messages */},
}
impl ChainNotify for EthSync {
fn new_blocks(&self,
imported: Vec<H256>,
invalid: Vec<H256>,
enacted: Vec<H256>,
retracted: Vec<H256>,
sealed: Vec<H256>)
{
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()
}
}

2
util/Cargo.toml
View File

@ -27,7 +27,7 @@ itertools = "0.4"
crossbeam = "0.2"
slab = "0.2"
sha3 = { path = "sha3" }
clippy = { version = "0.0.78", optional = true}
clippy = { version = "0.0.79", optional = true}
igd = "0.5.0"
ethcore-devtools = { path = "../devtools" }
libc = "0.2.7"

37
util/src/bytes.rs
View File

@ -242,7 +242,7 @@ pub enum FromBytesError {
}
/// 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
fn from_bytes(d: &[u8]) -> Result<Self, FromBytesError>;
}
@ -255,7 +255,7 @@ impl<T> FromRawBytes for T where T: FixedHash {
Ordering::Equal => ()
};
let mut res: Self = unsafe { mem::uninitialized() };
let mut res = T::zero();
res.copy_raw(bytes);
Ok(res)
}
@ -271,7 +271,7 @@ macro_rules! sized_binary_map {
::std::cmp::Ordering::Greater => return Err(FromBytesError::TooLong),
::std::cmp::Ordering::Equal => ()
};
let mut res: Self = unsafe { ::std::mem::uninitialized() };
let mut res: Self = 0;
res.copy_raw(bytes);
Ok(res)
}
@ -298,7 +298,7 @@ sized_binary_map!(u32);
sized_binary_map!(u64);
/// Value that can be serialized from variable-length byte array
pub trait FromRawBytesVariable : Sized {
pub trait FromRawBytesVariable: Sized {
/// Create value from slice
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 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) };
for i in 0..length_in_chunks {
*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 {
fn from_bytes(bytes: &[u8]) -> Result<Self, FromBytesError> {
let header = 8usize;
let mut map: (u64, ) = unsafe { mem::uninitialized() };
let mut map: (u64, ) = (0,);
if bytes.len() < header { return Err(FromBytesError::NotLongEnough); }
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> {
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); }
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> {
let header = 8usize;
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)
where V1: ToBytesWithMap,
V2: ToBytesWithMap,
impl<'a, V1, X1, V2, X2, T3> ToBytesWithMap for (X1, X2, &'a T3)
where V1: ToBytesWithMap, X1: Deref<Target=[V1]>,
V2: ToBytesWithMap, X2: Deref<Target=[V2]>,
T3: ToBytesWithMap
{
fn to_bytes_map(&self) -> Vec<u8> {
@ -433,7 +433,7 @@ pub trait ToBytesWithMap {
impl<T> ToBytesWithMap for T where T: FixedHash {
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() {
type Tup = (Vec<u16>, u16);
let tup = (vec![1u16, 1u16, 1u16, 1u16], 10u16);
let tup: (&[u16], u16) = (&[1; 4], 10);
let bytes = vec![
// map
8u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8,
@ -505,18 +505,19 @@ fn raw_bytes_from_tuple() {
// 10u16
10u8, 0u8];
let tup_from = Tup::from_bytes(&bytes).unwrap();
assert_eq!(tup, tup_from);
let (v, x) = Tup::from_bytes(&bytes).unwrap();
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();
assert_eq!(bytes_to, bytes);
}
#[test]
fn bytes_map_from_triple() {
let data = (vec![2u16; 6], vec![6u32; 3], 12u64);
let bytes_map = (&data.0, &data.1, &data.2).to_bytes_map();
let data: (&[u16], &[u32], u64) = (&[2; 6], &[6; 3], 12u64);
let bytes_map = (data.0, data.1, &data.2).to_bytes_map();
assert_eq!(bytes_map, vec![
// data map 2 x u64
12, 0, 0, 0, 0, 0, 0, 0,

2
util/src/journaldb/archivedb.rs
View File

@ -202,7 +202,7 @@ impl JournalDB for ArchiveDB {
fn latest_era(&self) -> Option<u64> { self.latest_era }
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 }

4
util/src/journaldb/earlymergedb.rs
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))]
fn commit(&mut self, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError> {
// journal format:

2
util/src/journaldb/mod.rs
View File

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

24
util/src/journaldb/overlayrecentdb.rs
View File

@ -171,7 +171,7 @@ impl OverlayRecentDB {
for r in insertions.iter() {
let k: H256 = r.val_at(0);
let v: Bytes = r.val_at(1);
overlay.emplace(k.clone(), v);
overlay.emplace(OverlayRecentDB::to_short_key(&k), v);
inserted_keys.push(k);
count += 1;
}
@ -191,6 +191,13 @@ impl OverlayRecentDB {
trace!("Recovered {} overlay entries, {} journal entries", count, journal.len());
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 {
@ -212,6 +219,11 @@ impl JournalDB for OverlayRecentDB {
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> {
// record new commit's details.
trace!("commit: #{} ({}), end era: {:?}", now, id, end);
@ -230,7 +242,7 @@ impl JournalDB for OverlayRecentDB {
r.begin_list(2);
r.append(&k);
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);
@ -266,7 +278,7 @@ impl JournalDB for OverlayRecentDB {
{
if canon_id == journal.id {
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 {
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
for k in overlay_deletions {
journal_overlay.backing_overlay.remove(&k);
journal_overlay.backing_overlay.remove(&OverlayRecentDB::to_short_key(&k));
}
// apply 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));
}
}
@ -322,7 +334,7 @@ impl HashDB for OverlayRecentDB {
match k {
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 {
Some(x) => {
Some(&self.transaction_overlay.denote(key, x).0)

4
util/src/journaldb/refcounteddb.rs
View File

@ -111,6 +111,10 @@ impl JournalDB for RefCountedDB {
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> {
// journal format:
// [era, 0] => [ id, [insert_0, ...], [remove_0, ...] ]

4
util/src/journaldb/traits.rs
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>;
/// State data query
fn state(&self, _id: &H256) -> Option<Bytes> {
None
}
fn state(&self, _id: &H256) -> Option<Bytes>;
/// Whether this database is pruned.
fn is_pruned(&self) -> bool { true }

100
util/src/network/host.rs
View File

@ -136,11 +136,11 @@ pub type ProtocolId = &'static str;
/// Messages used to communitate with the event loop from other threads.
#[derive(Clone)]
pub enum NetworkIoMessage<Message> where Message: Send + Sync + Clone {
pub enum NetworkIoMessage {
/// Register a new protocol handler.
AddHandler {
/// Handler shared instance.
handler: Arc<NetworkProtocolHandler<Message> + Sync>,
handler: Arc<NetworkProtocolHandler + Sync>,
/// Protocol Id.
protocol: ProtocolId,
/// Supported protocol versions.
@ -163,8 +163,6 @@ pub enum NetworkIoMessage<Message> where Message: Send + Sync + Clone {
DisablePeer(PeerId),
/// Network has been started with the host as the given enode.
NetworkStarted(String),
/// User message
User(Message),
}
/// 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.
pub struct NetworkContext<'s, Message> where Message: Send + Sync + Clone + 'static, 's {
io: &'s IoContext<NetworkIoMessage<Message>>,
pub struct NetworkContext<'s> {
io: &'s IoContext<NetworkIoMessage>,
protocol: ProtocolId,
sessions: Arc<RwLock<Slab<SharedSession>>>,
session: Option<SharedSession>,
@ -197,12 +195,12 @@ pub struct NetworkContext<'s, Message> where Message: Send + Sync + Clone + 'sta
_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.
fn new(io: &'s IoContext<NetworkIoMessage<Message>>,
fn new(io: &'s IoContext<NetworkIoMessage>,
protocol: ProtocolId,
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());
NetworkContext {
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)
}
/// Send an IO message
pub fn message(&self, msg: Message) -> Result<(), UtilError> {
self.io.message(NetworkIoMessage::User(msg))
}
/// Get an IoChannel.
pub fn io_channel(&self) -> IoChannel<NetworkIoMessage<Message>> {
pub fn io_channel(&self) -> IoChannel<NetworkIoMessage> {
self.io.channel()
}
@ -333,13 +326,13 @@ struct ProtocolTimer {
}
/// 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>,
tcp_listener: Mutex<TcpListener>,
sessions: Arc<RwLock<Slab<SharedSession>>>,
discovery: Mutex<Option<Discovery>>,
nodes: RwLock<NodeTable>,
handlers: RwLock<HashMap<ProtocolId, Arc<NetworkProtocolHandler<Message>>>>,
handlers: RwLock<HashMap<ProtocolId, Arc<NetworkProtocolHandler>>>,
timers: RwLock<HashMap<TimerToken, ProtocolTimer>>,
timer_counter: RwLock<usize>,
stats: Arc<NetworkStats>,
@ -348,9 +341,9 @@ pub struct Host<Message> where Message: Send + Sync + Clone {
stopping: AtomicBool,
}
impl<Message> Host<Message> where Message: Send + Sync + Clone {
impl Host {
/// 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");
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 reserved_nodes = config.reserved_nodes.clone();
let mut host = Host::<Message> {
let mut host = Host {
info: RwLock::new(HostInfo {
keys: keys,
config: config,
@ -444,7 +437,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
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();
if info.config.non_reserved_mode != mode {
@ -495,7 +488,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
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);
let mut to_kill = Vec::new();
for e in self.sessions.unwrapped_write().iter_mut() {
@ -511,7 +504,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
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() {
return Ok(());
}
@ -567,7 +560,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
Ok(())
}
fn maintain_network(&self, io: &IoContext<NetworkIoMessage<Message>>) {
fn maintain_network(&self, io: &IoContext<NetworkIoMessage>) {
self.keep_alive(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()
}
fn keep_alive(&self, io: &IoContext<NetworkIoMessage<Message>>) {
fn keep_alive(&self, io: &IoContext<NetworkIoMessage>) {
let mut to_kill = Vec::new();
for e in self.sessions.unwrapped_write().iter_mut() {
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 info = self.info.unwrapped_read();
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))]
fn connect_peer(&self, id: &NodeId, io: &IoContext<NetworkIoMessage<Message>>) {
fn connect_peer(&self, id: &NodeId, io: &IoContext<NetworkIoMessage>) {
if self.have_session(id)
{
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))]
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 mut sessions = self.sessions.unwrapped_write();
let token = sessions.insert_with_opt(|token| {
match Session::new(io, socket, token, id, &nonce, self.stats.clone(), &self.info.unwrapped_read()) {
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");
loop {
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() };
if let Some(session) = session {
let mut s = session.locked();
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);
self.kill_connection(token, io, true);
}
#[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 packet_data: Vec<(ProtocolId, PacketId, Vec<u8>)> = Vec::new();
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);
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 failure_id = None;
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 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);
}
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
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.message(NetworkIoMessage::InitPublicInterface).unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e));
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);
match stream {
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) {
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) {
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) {
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) {
return;
}
@ -1031,19 +1033,13 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
trace!(target: "network", "Disabling peer {}", peer);
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 =>
self.init_public_interface(io).unwrap_or_else(|e| warn!("Error initializing public interface: {:?}", e)),
_ => {} // 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 {
FIRST_SESSION ... LAST_SESSION => {
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 {
FIRST_SESSION ... LAST_SESSION => {
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 {
FIRST_SESSION ... LAST_SESSION => {
let connection = { self.sessions.unwrapped_read().get(stream).cloned() };
@ -1152,6 +1148,6 @@ fn host_client_url() {
let mut config = NetworkConfiguration::new();
let key = h256_from_hex("6f7b0d801bc7b5ce7bbd930b84fd0369b3eb25d09be58d64ba811091046f3aa2");
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@"));
}

40
util/src/network/mod.rs
View File

@ -24,39 +24,30 @@
//!
//! struct MyHandler;
//!
//! #[derive(Clone)]
//! struct MyMessage {
//! data: u32
//! }
//!
//! impl NetworkProtocolHandler<MyMessage> for MyHandler {
//! fn initialize(&self, io: &NetworkContext<MyMessage>) {
//! impl NetworkProtocolHandler for MyHandler {
//! fn initialize(&self, io: &NetworkContext) {
//! 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);
//! }
//!
//! fn connected(&self, io: &NetworkContext<MyMessage>, peer: &PeerId) {
//! fn connected(&self, io: &NetworkContext, peer: &PeerId) {
//! println!("Connected {}", peer);
//! }
//!
//! fn disconnected(&self, io: &NetworkContext<MyMessage>, peer: &PeerId) {
//! fn disconnected(&self, io: &NetworkContext, peer: &PeerId) {
//! println!("Disconnected {}", peer);
//! }
//!
//! fn timeout(&self, io: &NetworkContext<MyMessage>, timer: TimerToken) {
//! fn timeout(&self, io: &NetworkContext, timer: TimerToken) {
//! println!("Timeout {}", timer);
//! }
//!
//! fn message(&self, io: &NetworkContext<MyMessage>, message: &MyMessage) {
//! println!("Message {}", message.data);
//! }
//! }
//!
//! 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.start().expect("Error starting service");
//!
@ -84,7 +75,6 @@ pub use network::host::PacketId;
pub use network::host::NetworkContext;
pub use network::service::NetworkService;
pub use network::host::NetworkIoMessage;
pub use network::host::NetworkIoMessage::User as UserMessage;
pub use network::error::NetworkError;
pub use network::host::NetworkConfiguration;
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.
/// All the handler function are called from within IO event loop.
/// `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
fn initialize(&self, _io: &NetworkContext<Message>) {}
fn initialize(&self, _io: &NetworkContext) {}
/// 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.
fn connected(&self, io: &NetworkContext<Message>, peer: &PeerId);
fn connected(&self, io: &NetworkContext, peer: &PeerId);
/// 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`.
fn timeout(&self, _io: &NetworkContext<Message>, _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) {}
fn timeout(&self, _io: &NetworkContext, _timer: TimerToken) {}
}
/// Non-reserved peer modes.
@ -130,4 +118,4 @@ impl NonReservedPeerMode {
_ => None,
}
}
}
}

31
util/src/network/service.rs
View File

@ -20,30 +20,30 @@ use panics::*;
use misc::RwLockable;
use network::{NetworkProtocolHandler, NetworkConfiguration};
use network::error::NetworkError;
use network::host::{Host, NetworkIoMessage, ProtocolId};
use network::host::{Host, NetworkContext, NetworkIoMessage, ProtocolId};
use network::stats::NetworkStats;
use io::*;
/// IO Service with networking
/// `Message` defines a notification data type.
pub struct NetworkService<Message> where Message: Send + Sync + Clone + 'static {
io_service: IoService<NetworkIoMessage<Message>>,
pub struct NetworkService {
io_service: IoService<NetworkIoMessage>,
host_info: String,
host: RwLock<Option<Arc<Host<Message>>>>,
host: RwLock<Option<Arc<Host>>>,
stats: Arc<NetworkStats>,
panic_handler: Arc<PanicHandler>,
config: NetworkConfiguration,
}
impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'static {
impl NetworkService {
/// 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 io_service = try!(IoService::<NetworkIoMessage<Message>>::start());
let io_service = try!(IoService::<NetworkIoMessage>::start());
panic_handler.forward_from(&io_service);
let stats = Arc::new(NetworkStats::new());
let host_info = Host::<Message>::client_version();
let host_info = Host::client_version();
Ok(NetworkService {
io_service: io_service,
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.
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 {
handler: handler,
protocol: protocol,
@ -70,7 +70,7 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
}
/// Returns underlying io service.
pub fn io(&self) -> &IoService<NetworkIoMessage<Message>> {
pub fn io(&self) -> &IoService<NetworkIoMessage> {
&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);
}
}
/// 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 {
self.panic_handler.on_panic(closure);
}

42
util/src/network/tests.rs
View File

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

2
util/src/triehash.rs
View File

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

10
util/src/vector.rs
View File

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