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

Merge branch 'master' of github.com:ethcore/parity

Browse Source
This commit is contained in:
NikVolf 2016-06-14 14:18:31 +03:00
commit b3caeee758
31 changed files with 487 additions and 412 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
Cargo.lock generated
View File

@ -3,7 +3,7 @@ name = "parity"
version = "1.2.0"
dependencies = [
"ansi_term 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.71 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.76 (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)",
@ -20,7 +20,7 @@ dependencies = [
"ethsync 1.2.0",
"fdlimit 0.1.0",
"hyper 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)",
"json-ipc-server 0.1.0 (git+https://github.com/ethcore/json-ipc-server.git)",
"json-ipc-server 0.2.2 (git+https://github.com/ethcore/json-ipc-server.git)",
"log 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",
"num_cpus 0.2.11 (registry+https://github.com/rust-lang/crates.io-index)",
"number_prefix 0.2.5 (registry+https://github.com/rust-lang/crates.io-index)",
@ -130,23 +130,18 @@ dependencies = [
[[package]]
name = "clippy"
version = "0.0.71"
version = "0.0.76"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"clippy_lints 0.0.71 (registry+https://github.com/rust-lang/crates.io-index)",
"quine-mc_cluskey 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
"regex-syntax 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"rustc-serialize 0.3.19 (registry+https://github.com/rust-lang/crates.io-index)",
"semver 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
"toml 0.1.28 (registry+https://github.com/rust-lang/crates.io-index)",
"unicode-normalization 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy_lints 0.0.76 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "clippy_lints"
version = "0.0.71"
version = "0.0.76"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"matches 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
"quine-mc_cluskey 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
"regex-syntax 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"rustc-serialize 0.3.19 (registry+https://github.com/rust-lang/crates.io-index)",
@ -256,7 +251,7 @@ name = "ethcore"
version = "1.2.0"
dependencies = [
"bloomchain 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.71 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.76 (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.2.0",
@ -280,7 +275,7 @@ dependencies = [
name = "ethcore-dapps"
version = "1.2.0"
dependencies = [
"clippy 0.0.71 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.76 (registry+https://github.com/rust-lang/crates.io-index)",
"ethcore-rpc 1.2.0",
"ethcore-util 1.2.0",
"hyper 0.9.3 (git+https://github.com/ethcore/hyper)",
@ -293,7 +288,7 @@ dependencies = [
"parity-dapps-dao 0.4.0 (git+https://github.com/ethcore/parity-dapps-dao-rs.git)",
"parity-dapps-makerotc 0.3.0 (git+https://github.com/ethcore/parity-dapps-makerotc-rs.git)",
"parity-dapps-status 0.5.0 (git+https://github.com/ethcore/parity-dapps-status-rs.git)",
"parity-dapps-wallet 0.6.0 (git+https://github.com/ethcore/parity-dapps-wallet-rs.git)",
"parity-dapps-wallet 0.6.1 (git+https://github.com/ethcore/parity-dapps-wallet-rs.git)",
"rustc-serialize 0.3.19 (registry+https://github.com/rust-lang/crates.io-index)",
"serde 0.7.7 (registry+https://github.com/rust-lang/crates.io-index)",
"serde_codegen 0.7.7 (registry+https://github.com/rust-lang/crates.io-index)",
@ -344,14 +339,14 @@ dependencies = [
name = "ethcore-rpc"
version = "1.2.0"
dependencies = [
"clippy 0.0.71 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.76 (registry+https://github.com/rust-lang/crates.io-index)",
"ethash 1.2.0",
"ethcore 1.2.0",
"ethcore-devtools 1.2.0",
"ethcore-util 1.2.0",
"ethjson 0.1.0",
"ethsync 1.2.0",
"json-ipc-server 0.1.0 (git+https://github.com/ethcore/json-ipc-server.git)",
"json-ipc-server 0.2.2 (git+https://github.com/ethcore/json-ipc-server.git)",
"jsonrpc-core 2.0.5 (registry+https://github.com/rust-lang/crates.io-index)",
"jsonrpc-http-server 5.1.0 (git+https://github.com/ethcore/jsonrpc-http-server.git)",
"log 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",
@ -367,7 +362,7 @@ dependencies = [
name = "ethcore-signer"
version = "1.2.0"
dependencies = [
"clippy 0.0.71 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.76 (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.2.0",
"ethcore-util 1.2.0",
@ -386,7 +381,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.71 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.76 (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)",
@ -399,7 +394,7 @@ dependencies = [
"lazy_static 0.1.16 (registry+https://github.com/rust-lang/crates.io-index)",
"libc 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)",
"log 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",
"mio 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)",
"mio 0.5.1 (git+https://github.com/ethcore/mio?branch=v0.5.x)",
"nix 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)",
"rand 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)",
"rocksdb 0.4.5 (git+https://github.com/ethcore/rust-rocksdb)",
@ -431,7 +426,7 @@ dependencies = [
name = "ethsync"
version = "1.2.0"
dependencies = [
"clippy 0.0.71 (registry+https://github.com/rust-lang/crates.io-index)",
"clippy 0.0.76 (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.2.0",
"ethcore-util 1.2.0",
@ -575,12 +570,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "json-ipc-server"
version = "0.1.0"
source = "git+https://github.com/ethcore/json-ipc-server.git#4f9226c4f84dcce2385a188374e3b5fc66b63e68"
version = "0.2.2"
source = "git+https://github.com/ethcore/json-ipc-server.git#15ef25e5f859d2d27469c92cc13dd1ddea03e444"
dependencies = [
"bytes 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
"env_logger 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
"jsonrpc-core 2.0.5 (registry+https://github.com/rust-lang/crates.io-index)",
"lazy_static 0.1.16 (registry+https://github.com/rust-lang/crates.io-index)",
"log 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",
"mio 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)",
"miow 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
"slab 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
]
@ -709,6 +708,22 @@ dependencies = [
"winapi 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "mio"
version = "0.5.1"
source = "git+https://github.com/ethcore/mio?branch=v0.5.x#1fc881771fb8c2517317b4f805d7b88235be422b"
dependencies = [
"bytes 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
"libc 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)",
"log 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",
"miow 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
"net2 0.2.23 (registry+https://github.com/rust-lang/crates.io-index)",
"nix 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)",
"slab 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
"time 0.1.35 (registry+https://github.com/rust-lang/crates.io-index)",
"winapi 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "mio"
version = "0.5.1"
@ -901,7 +916,7 @@ dependencies = [
[[package]]
name = "parity-dapps-builtins"
version = "0.5.1"
source = "git+https://github.com/ethcore/parity-dapps-builtins-rs.git#d3c95d62ffaa57016b162a9a9f0e6dd629dab423"
source = "git+https://github.com/ethcore/parity-dapps-builtins-rs.git#b3970ff4686a12321365cfafba6552bfaa2e7874"
dependencies = [
"parity-dapps 0.3.0 (git+https://github.com/ethcore/parity-dapps-rs.git)",
]
@ -909,7 +924,7 @@ dependencies = [
[[package]]
name = "parity-dapps-dao"
version = "0.4.0"
source = "git+https://github.com/ethcore/parity-dapps-dao-rs.git#18f4b839b20fbdf8e0d163e14d25aafee603ac4b"
source = "git+https://github.com/ethcore/parity-dapps-dao-rs.git#dd6b9ca7c18fbfa714183a4f570bd75b8391c13d"
dependencies = [
"parity-dapps 0.3.0 (git+https://github.com/ethcore/parity-dapps-rs.git)",
]
@ -917,7 +932,7 @@ dependencies = [
[[package]]
name = "parity-dapps-makerotc"
version = "0.3.0"
source = "git+https://github.com/ethcore/parity-dapps-makerotc-rs.git#7b771f217a3eefeb9a976c7ed470ca49fd9a9daa"
source = "git+https://github.com/ethcore/parity-dapps-makerotc-rs.git#33568ac7209aa765c498bb2322e848f552656303"
dependencies = [
"parity-dapps 0.3.0 (git+https://github.com/ethcore/parity-dapps-rs.git)",
]
@ -932,8 +947,8 @@ dependencies = [
[[package]]
name = "parity-dapps-wallet"
version = "0.6.0"
source = "git+https://github.com/ethcore/parity-dapps-wallet-rs.git#ad23b093d47527333a262c95e6fb20a97d15d6e6"
version = "0.6.1"
source = "git+https://github.com/ethcore/parity-dapps-wallet-rs.git#8923d4c73359c75ce04f0639bbcde46adb846b81"
dependencies = [
"parity-dapps 0.3.0 (git+https://github.com/ethcore/parity-dapps-rs.git)",
]
@ -941,7 +956,7 @@ dependencies = [
[[package]]
name = "parity-minimal-sysui"
version = "0.1.0"
source = "git+https://github.com/ethcore/parity-dapps-minimal-sysui-rs.git#cb27ae09ee18773ccca6ba2ac74fa3128047a652"
source = "git+https://github.com/ethcore/parity-dapps-minimal-sysui-rs.git#4c704913f671060bb0e43b5ce4a68d02281115d5"
[[package]]
name = "phf"

6
Cargo.toml
View File

@ -19,11 +19,10 @@ docopt = "0.6"
time = "0.1"
ctrlc = { git = "https://github.com/ethcore/rust-ctrlc.git" }
fdlimit = { path = "util/fdlimit" }
daemonize = "0.2"
num_cpus = "0.2"
number_prefix = "0.2"
rpassword = "0.2.1"
clippy = { version = "0.0.71", optional = true}
clippy = { version = "0.0.76", optional = true}
ethcore = { path = "ethcore" }
ethcore-util = { path = "util" }
ethsync = { path = "sync" }
@ -37,6 +36,9 @@ ethcore-ipc = { path = "ipc/rpc" }
json-ipc-server = { git = "https://github.com/ethcore/json-ipc-server.git" }
ansi_term = "0.7"
[target.'cfg(not(windows))'.dependencies]
daemonize = "0.2"
[dependencies.hyper]
version = "0.8"
default-features = false

19
README.md
View File

@ -40,18 +40,21 @@ below to build from source.
Parity is fully compatible with Stable Rust.
We recommend installing Rust through [multirust](https://github.com/brson/multirust). If you don't already have multirust, you can install it like this:
We recommend installing Rust through [rustup](https://www.rustup.rs/). If you don't already have rustup, you can install it like this:
- Linux:
- Linux and OSX:
```bash
$ curl -sf https://raw.githubusercontent.com/brson/multirust/master/quick-install.sh | sh
$ curl https://sh.rustup.rs -sSf | sh
```
- OSX with Homebrew:
```bash
$ brew update && brew install multirust
$ multirust default stable
- Windows
Make sure you have Visual Studio 2015 with C++ support installed, download and run [rustup](https://static.rust-lang.org/rustup/dist/x86_64-pc-windows-msvc/rustup-init.exe) and use the following command to setup msvc toolchain:
```
rustup default stable-x86_64-pc-windows-msvc
```
Then, download and build Parity:
@ -72,4 +75,4 @@ To get started, just run
$ parity
```
and parity will begin syncing the Ethereum blockchain.
and parity will begin syncing the Ethereum blockchain.

2
dapps/Cargo.toml
View File

@ -28,7 +28,7 @@ parity-dapps-wallet = { git = "https://github.com/ethcore/parity-dapps-wallet-rs
parity-dapps-dao = { git = "https://github.com/ethcore/parity-dapps-dao-rs.git", version = "0.4.0", optional = true }
parity-dapps-makerotc = { git = "https://github.com/ethcore/parity-dapps-makerotc-rs.git", version = "0.3.0", optional = true }
mime_guess = { version = "1.6.1" }
clippy = { version = "0.0.71", optional = true}
clippy = { version = "0.0.76", optional = true}
[build-dependencies]
serde_codegen = { version = "0.7.0", optional = true }

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.71", optional = true}
clippy = { version = "0.0.76", optional = true}
crossbeam = "0.2.9"
lazy_static = "0.1"
ethcore-devtools = { path = "../devtools" }

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

@ -756,18 +756,16 @@ impl<V> BlockChainClient for Client<V> where V: Verifier {
}
impl<V> MiningBlockChainClient for Client<V> where V: Verifier {
fn prepare_sealing(&self, author: Address, gas_floor_target: U256, extra_data: Bytes, transactions: Vec<SignedTransaction>)
-> (Option<ClosedBlock>, HashSet<H256>) {
fn prepare_open_block(&self, author: Address, gas_floor_target: U256, extra_data: Bytes) -> OpenBlock {
let engine = self.engine.deref().deref();
let h = self.chain.best_block_hash();
let mut invalid_transactions = HashSet::new();
let mut b = OpenBlock::new(
let mut open_block = OpenBlock::new(
engine,
&self.vm_factory,
false, // TODO: this will need to be parameterised once we want to do immediate mining insertion.
self.state_db.lock().unwrap().boxed_clone(),
match self.chain.block_header(&h) { Some(ref x) => x, None => { return (None, invalid_transactions) } },
&self.chain.block_header(&h).expect("h is best block hash: so it's header must exist: qed"),
self.build_last_hashes(h.clone()),
author,
gas_floor_target,
@ -782,48 +780,10 @@ impl<V> MiningBlockChainClient for Client<V> where V: Verifier {
.into_iter()
.take(engine.maximum_uncle_count())
.foreach(|h| {
b.push_uncle(h).unwrap();
open_block.push_uncle(h).unwrap();
});
// Add transactions
let block_number = b.block().header().number();
let min_tx_gas = U256::from(self.engine.schedule(&b.env_info()).tx_gas);
for tx in transactions {
// Push transaction to block
let hash = tx.hash();
let import = b.push_transaction(tx, None);
match import {
Err(Error::Execution(ExecutionError::BlockGasLimitReached { gas_limit, gas_used, .. })) => {
trace!(target: "miner", "Skipping adding transaction to block because of gas limit: {:?}", hash);
// Exit early if gas left is smaller then min_tx_gas
if gas_limit - gas_used < min_tx_gas {
break;
}
},
Err(e) => {
invalid_transactions.insert(hash);
trace!(target: "miner",
"Error adding transaction to block: number={}. transaction_hash={:?}, Error: {:?}",
block_number, hash, e);
},
_ => {}
}
}
// And close
let b = b.close();
trace!(target: "miner", "Sealing: number={}, hash={}, diff={}",
b.block().header().number(),
b.hash(),
b.block().header().difficulty()
);
(Some(b), invalid_transactions)
}
fn try_seal(&self, block: LockedBlock, seal: Vec<Bytes>) -> Result<SealedBlock, LockedBlock> {
block.try_seal(self.engine.deref().deref(), seal)
open_block
}
}

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

@ -31,13 +31,12 @@ pub use self::trace::Filter as TraceFilter;
pub use executive::{Executed, Executive, TransactOptions};
pub use env_info::{LastHashes, EnvInfo};
use std::collections::HashSet;
use util::bytes::Bytes;
use util::hash::{Address, H256, H2048};
use util::numbers::U256;
use blockchain::TreeRoute;
use block_queue::BlockQueueInfo;
use block::{ClosedBlock, LockedBlock, SealedBlock};
use block::OpenBlock;
use header::{BlockNumber, Header};
use transaction::{LocalizedTransaction, SignedTransaction};
use log_entry::LocalizedLogEntry;
@ -198,10 +197,7 @@ pub trait BlockChainClient : Sync + Send {
/// Extended client interface used for mining
pub trait MiningBlockChainClient : BlockChainClient {
/// Attempts to seal given block. Returns `SealedBlock` on success and the same block in case of error.
fn try_seal(&self, block: LockedBlock, seal: Vec<Bytes>) -> Result<SealedBlock, LockedBlock>;
/// Returns ClosedBlock prepared for sealing.
fn prepare_sealing(&self, author: Address, gas_floor_target: U256, extra_data: Bytes, transactions: Vec<SignedTransaction>)
-> (Option<ClosedBlock>, HashSet<H256>);
/// Returns OpenBlock prepared for closing.
fn prepare_open_block(&self, author: Address, gas_floor_target: U256, extra_data: Bytes)
-> OpenBlock;
}

11
ethcore/src/client/test_client.rs
View File

@ -31,7 +31,7 @@ use evm::Factory as EvmFactory;
use miner::{Miner, MinerService};
use block_queue::BlockQueueInfo;
use block::{SealedBlock, ClosedBlock, LockedBlock};
use block::OpenBlock;
use executive::Executed;
use error::{ExecutionError};
use trace::LocalizedTrace;
@ -240,13 +240,8 @@ impl TestBlockChainClient {
}
impl MiningBlockChainClient for TestBlockChainClient {
fn try_seal(&self, block: LockedBlock, _seal: Vec<Bytes>) -> Result<SealedBlock, LockedBlock> {
Err(block)
}
fn prepare_sealing(&self, _author: Address, _gas_floor_target: U256, _extra_data: Bytes, _transactions: Vec<SignedTransaction>) -> (Option<ClosedBlock>, HashSet<H256>) {
(None, HashSet::new())
fn prepare_open_block(&self, _author: Address, _gas_floor_target: U256, _extra_data: Bytes) -> OpenBlock {
unimplemented!();
}
}

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

@ -117,90 +117,89 @@ impl Miner {
// otherwise, leave everything alone.
// otherwise, author a fresh block.
*/
let (b, invalid_transactions) = match sealing_work.pop_if(|b| b.block().fields().header.parent_hash() == &best_hash) {
let mut open_block = match sealing_work.pop_if(|b| b.block().fields().header.parent_hash() == &best_hash) {
Some(old_block) => {
trace!(target: "miner", "Already have previous work; updating and returning");
// add transactions to old_block
let e = self.engine();
let mut invalid_transactions = HashSet::new();
let mut block = old_block.reopen(e, chain.vm_factory());
let block_number = block.block().fields().header.number();
// TODO: push new uncles, too.
// TODO: refactor with chain.prepare_sealing
for tx in transactions {
let hash = tx.hash();
let res = block.push_transaction(tx, None);
match res {
Err(Error::Execution(ExecutionError::BlockGasLimitReached { gas_limit, gas_used, .. })) => {
trace!(target: "miner", "Skipping adding transaction to block because of gas limit: {:?}", hash);
// Exit early if gas left is smaller then min_tx_gas
let min_tx_gas: U256 = 21000.into(); // TODO: figure this out properly.
if gas_limit - gas_used < min_tx_gas {
break;
}
},
Err(Error::Transaction(TransactionError::AlreadyImported)) => {} // already have transaction - ignore
Err(e) => {
invalid_transactions.insert(hash);
trace!(target: "miner",
"Error adding transaction to block: number={}. transaction_hash={:?}, Error: {:?}",
block_number, hash, e);
},
_ => {} // imported ok
}
}
(Some(block.close()), invalid_transactions)
old_block.reopen(e, chain.vm_factory())
}
None => {
// block not found - create it.
trace!(target: "miner", "No existing work - making new block");
chain.prepare_sealing(
chain.prepare_open_block(
self.author(),
self.gas_floor_target(),
self.extra_data(),
transactions,
self.extra_data()
)
}
};
let mut invalid_transactions = HashSet::new();
let block_number = open_block.block().fields().header.number();
// TODO: push new uncles, too.
for tx in transactions {
let hash = tx.hash();
match open_block.push_transaction(tx, None) {
Err(Error::Execution(ExecutionError::BlockGasLimitReached { gas_limit, gas_used, .. })) => {
trace!(target: "miner", "Skipping adding transaction to block because of gas limit: {:?}", hash);
// Exit early if gas left is smaller then min_tx_gas
let min_tx_gas: U256 = 21000.into(); // TODO: figure this out properly.
if gas_limit - gas_used < min_tx_gas {
break;
}
},
Err(Error::Transaction(TransactionError::AlreadyImported)) => {} // already have transaction - ignore
Err(e) => {
invalid_transactions.insert(hash);
trace!(target: "miner",
"Error adding transaction to block: number={}. transaction_hash={:?}, Error: {:?}",
block_number, hash, e);
},
_ => {} // imported ok
}
}
let block = open_block.close();
let mut queue = self.transaction_queue.lock().unwrap();
let fetch_account = |a: &Address| AccountDetails {
nonce: chain.latest_nonce(a),
balance: chain.latest_balance(a),
};
for hash in invalid_transactions.into_iter() {
queue.remove_invalid(&hash, &fetch_account);
}
if let Some(block) = b {
if !block.transactions().is_empty() {
trace!(target: "miner", "prepare_sealing: block has transaction - attempting internal seal.");
// block with transactions - see if we can seal immediately.
let s = self.engine().generate_seal(block.block(), match self.accounts {
Some(ref x) => Some(&**x),
None => None,
});
if let Some(seal) = s {
trace!(target: "miner", "prepare_sealing: managed internal seal. importing...");
if let Ok(sealed) = chain.try_seal(block.lock(), seal) {
if let Ok(_) = chain.import_block(sealed.rlp_bytes()) {
trace!(target: "miner", "prepare_sealing: sealed internally and imported. leaving.");
} else {
warn!("prepare_sealing: ERROR: could not import internally sealed block. WTF?");
}
if !block.transactions().is_empty() {
trace!(target: "miner", "prepare_sealing: block has transaction - attempting internal seal.");
// block with transactions - see if we can seal immediately.
let s = self.engine().generate_seal(block.block(), match self.accounts {
Some(ref x) => Some(&**x),
None => None,
});
if let Some(seal) = s {
trace!(target: "miner", "prepare_sealing: managed internal seal. importing...");
if let Ok(sealed) = block.lock().try_seal(self.engine(), seal) {
if let Ok(_) = chain.import_block(sealed.rlp_bytes()) {
trace!(target: "miner", "prepare_sealing: sealed internally and imported. leaving.");
} else {
warn!("prepare_sealing: ERROR: try_seal failed when given internally generated seal. WTF?");
warn!("prepare_sealing: ERROR: could not import internally sealed block. WTF?");
}
return;
} else {
trace!(target: "miner", "prepare_sealing: unable to generate seal internally");
warn!("prepare_sealing: ERROR: try_seal failed when given internally generated seal. WTF?");
}
}
if sealing_work.peek_last_ref().map_or(true, |pb| pb.block().fields().header.hash() != block.block().fields().header.hash()) {
trace!(target: "miner", "Pushing a new, refreshed or borrowed pending {}...", block.block().fields().header.hash());
sealing_work.push(block);
return;
} else {
trace!(target: "miner", "prepare_sealing: unable to generate seal internally");
}
}
if sealing_work.peek_last_ref().map_or(true, |pb| pb.block().fields().header.hash() != block.block().fields().header.hash()) {
trace!(target: "miner", "Pushing a new, refreshed or borrowed pending {}...", block.block().fields().header.hash());
sealing_work.push(block);
}
trace!(target: "miner", "prepare_sealing: leaving (last={:?})", sealing_work.peek_last_ref().map(|b| b.block().fields().header.hash()));
}
@ -493,7 +492,7 @@ impl MinerService for Miner {
trace!(target: "miner", "Miner sleeping (current {}, last {})", current_no, last_request);
self.sealing_enabled.store(false, atomic::Ordering::Relaxed);
self.sealing_work.lock().unwrap().reset();
} else if self.sealing_enabled.load(atomic::Ordering::Relaxed) {
} else {
self.prepare_sealing(chain);
}
}
@ -511,7 +510,7 @@ impl MinerService for Miner {
fn submit_seal(&self, chain: &MiningBlockChainClient, pow_hash: H256, seal: Vec<Bytes>) -> Result<(), Error> {
if let Some(b) = self.sealing_work.lock().unwrap().take_used_if(|b| &b.hash() == &pow_hash) {
match chain.try_seal(b.lock(), seal) {
match b.lock().try_seal(self.engine(), seal) {
Err(_) => {
info!(target: "miner", "Mined block rejected, PoW was invalid.");
Err(Error::PowInvalid)

3
ethcore/src/tests/client.rs
View File

@ -137,8 +137,7 @@ fn can_mine() {
let client_result = get_test_client_with_blocks(vec![dummy_blocks[0].clone()]);
let client = client_result.reference();
let b = client.prepare_sealing(Address::default(), 31415926.into(), vec![], vec![]).0.unwrap();
let b = client.prepare_open_block(Address::default(), 31415926.into(), vec![]).close();
assert_eq!(*b.block().header().parent_hash(), BlockView::new(&dummy_blocks[0]).header_view().sha3());
assert!(client.try_seal(b.lock(), vec![]).is_ok());
}

2
ipc/codegen/Cargo.toml
View File

@ -15,7 +15,7 @@ with-syntex = ["quasi/with-syntex", "quasi_codegen", "quasi_codegen/with-syntex"
[build-dependencies]
quasi_codegen = { version = "0.11", optional = true }
syntex = { version = "*", optional = true }
syntex = { version = "0.33", optional = true }
[dependencies]
aster = { version = "0.17", default-features = false }

14
ipc/codegen/src/serialization.rs
View File

@ -89,7 +89,7 @@ fn serialize_item(
let (size_expr, read_expr, write_expr) =
(binary_expressions.size, binary_expressions.read, binary_expressions.write);
Ok(quote_item!(cx,
match quote_item!(cx,
impl $generics ::ipc::BinaryConvertable for $ty $where_clause {
fn size(&self) -> usize {
$size_expr
@ -106,8 +106,16 @@ fn serialize_item(
fn len_params() -> usize {
1
}
}
).unwrap())
})
{
Some(item) => Ok(item),
None => {
cx.span_err(
item.span,
"syntax error expanding serialization implementation");
Err(Error)
}
}
}
#[allow(unreachable_code)]

27
ipc/tests/build.rs
View File

@ -19,20 +19,20 @@ extern crate ethcore_ipc_codegen as codegen;
use std::env;
use std::path::Path;
use std::process::exit;
pub fn main() {
let out_dir = env::var_os("OUT_DIR").unwrap();
// ipc pass
{
// rpc pass
if {
let src = Path::new("nested.rs.in");
let dst = Path::new(&out_dir).join("nested_ipc.rs");
let mut registry = syntex::Registry::new();
codegen::register(&mut registry);
registry.expand("", &src, &dst).unwrap();
registry.expand("", &src, &dst).is_ok()
}
// serde pass
// serialization pass
{
let src = Path::new(&out_dir).join("nested_ipc.rs");
let dst = Path::new(&out_dir).join("nested_cg.rs");
@ -41,16 +41,15 @@ pub fn main() {
registry.expand("", &src, &dst).unwrap();
}
// ipc pass
{
// rpc pass
if {
let src = Path::new("service.rs.in");
let dst = Path::new(&out_dir).join("service_ipc.rs");
let mut registry = syntex::Registry::new();
codegen::register(&mut registry);
registry.expand("", &src, &dst).unwrap();
registry.expand("", &src, &dst).is_ok()
}
// serde pass
// serialization pass
{
let src = Path::new(&out_dir).join("service_ipc.rs");
let dst = Path::new(&out_dir).join("service_cg.rs");
@ -59,13 +58,15 @@ pub fn main() {
registry.expand("", &src, &dst).unwrap();
}
// ipc pass
// rpc pass
{
let src = Path::new("binary.rs.in");
let dst = Path::new(&out_dir).join("binary.rs");
let mut registry = syntex::Registry::new();
codegen::register(&mut registry);
registry.expand("", &src, &dst).unwrap();
if let Err(err_msg) = registry.expand("", &src, &dst) {
println!("error: {}", err_msg);
exit(1);
}
}
}

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.71", optional = true}
clippy = { version = "0.0.76", optional = true}
[build-dependencies]
serde_codegen = { version = "0.7.0", optional = true }

32
parity/configuration.rs
View File

@ -287,9 +287,14 @@ impl Configuration {
}
fn geth_ipc_path(&self) -> String {
if self.args.flag_testnet { path::ethereum::with_testnet("geth.ipc") }
else { path::ethereum::with_default("geth.ipc") }
.to_str().unwrap().to_owned()
if cfg!(windows) {
r"\\.\pipe\geth.ipc".to_owned()
}
else {
if self.args.flag_testnet { path::ethereum::with_testnet("geth.ipc") }
else { path::ethereum::with_default("geth.ipc") }
.to_str().unwrap().to_owned()
}
}
pub fn keys_iterations(&self) -> u32 {
@ -358,7 +363,26 @@ impl Configuration {
fn ipc_path(&self) -> String {
if self.args.flag_geth { self.geth_ipc_path() }
else { Configuration::replace_home(&self.args.flag_ipcpath.clone().unwrap_or(self.args.flag_ipc_path.clone())) }
else {
if cfg!(windows) {
r"\\.\pipe\parity.jsonrpc".to_owned()
}
else {
Configuration::replace_home(&self.args.flag_ipcpath.clone().unwrap_or(self.args.flag_ipc_path.clone()))
}
}
}
pub fn have_color(&self) -> bool {
!self.args.flag_no_color && !cfg!(windows)
}
pub fn signer_port(&self) -> Option<u16> {
if self.args.flag_signer {
Some(self.args.flag_signer_port)
} else {
None
}
}
}

47
parity/main.rs
View File

@ -32,6 +32,7 @@ extern crate log as rlog;
extern crate env_logger;
extern crate ctrlc;
extern crate fdlimit;
#[cfg(not(windows))]
extern crate daemonize;
extern crate time;
extern crate number_prefix;
@ -86,7 +87,6 @@ use ethcore::service::ClientService;
use ethcore::spec::Spec;
use ethsync::EthSync;
use ethcore::miner::{Miner, MinerService, ExternalMiner};
use daemonize::Daemonize;
use migration::migrate;
use informant::Informant;
@ -109,17 +109,18 @@ fn execute(conf: Configuration) {
return;
}
if conf.args.cmd_signer {
execute_signer(conf);
return;
}
let spec = conf.spec();
let client_config = conf.client_config(&spec);
execute_upgrades(&conf, &spec, &client_config);
if conf.args.cmd_daemon {
Daemonize::new()
.pid_file(conf.args.arg_pid_file.clone())
.chown_pid_file(true)
.start()
.unwrap_or_else(|e| die!("Couldn't daemonize; {}", e));
daemonize(&conf);
}
if conf.args.cmd_account {
@ -137,14 +138,23 @@ fn execute(conf: Configuration) {
return;
}
if conf.args.cmd_signer {
execute_signer(conf);
return;
}
execute_client(conf, spec, client_config);
}
#[cfg(not(windows))]
fn daemonize(conf: &Configuration) {
use daemonize::Daemonize;
Daemonize::new()
.pid_file(conf.args.arg_pid_file.clone())
.chown_pid_file(true)
.start()
.unwrap_or_else(|e| die!("Couldn't daemonize; {}", e));
}
#[cfg(windows)]
fn daemonize(_conf: &Configuration) {
}
fn execute_upgrades(conf: &Configuration, spec: &Spec, client_config: &ClientConfig) {
match ::upgrade::upgrade(Some(&conf.path())) {
Ok(upgrades_applied) if upgrades_applied > 0 => {
@ -201,7 +211,7 @@ fn execute_client(conf: Configuration, spec: Spec, client_config: ClientConfig)
let sync = EthSync::register(service.network(), sync_config, client.clone());
let deps_for_rpc_apis = Arc::new(rpc_apis::Dependencies {
signer_enabled: conf.args.flag_signer,
signer_port: conf.signer_port(),
signer_queue: Arc::new(rpc_apis::ConfirmationsQueue::default()),
client: client.clone(),
sync: sync.clone(),
@ -228,6 +238,7 @@ fn execute_client(conf: Configuration, spec: Spec, client_config: ClientConfig)
// setup ipc rpc
let _ipc_server = rpc::new_ipc(conf.ipc_settings(), &dependencies);
debug!("IPC: {}", conf.ipc_settings());
if conf.args.flag_webapp { println!("WARNING: Flag -w/--webapp is deprecated. Dapps server is now on by default. Ignoring."); }
let dapps_server = dapps::new(dapps::Configuration {
@ -244,7 +255,7 @@ fn execute_client(conf: Configuration, spec: Spec, client_config: ClientConfig)
// Set up a signer
let signer_server = signer::start(signer::Configuration {
enabled: deps_for_rpc_apis.signer_enabled,
enabled: deps_for_rpc_apis.signer_port.is_some(),
port: conf.args.flag_signer_port,
signer_path: conf.directories().signer,
}, signer::Dependencies {
@ -255,7 +266,7 @@ fn execute_client(conf: Configuration, spec: Spec, client_config: ClientConfig)
// Register IO handler
let io_handler = Arc::new(ClientIoHandler {
client: service.client(),
info: Informant::new(!conf.args.flag_no_color),
info: Informant::new(conf.have_color()),
sync: sync.clone(),
accounts: account_service.clone(),
});
@ -375,8 +386,8 @@ fn execute_import(conf: Configuration) {
panic_handler.forward_from(&service);
let client = service.client();
let mut instream: Box<Read> = if let Some(f) = conf.args.arg_file {
let f = File::open(&f).unwrap_or_else(|_| die!("Cannot open the file given: {}", f));
let mut instream: Box<Read> = if let Some(ref f) = conf.args.arg_file {
let f = File::open(f).unwrap_or_else(|_| die!("Cannot open the file given: {}", f));
Box::new(f)
} else {
Box::new(::std::io::stdin())
@ -386,7 +397,7 @@ fn execute_import(conf: Configuration) {
let mut first_read = 0;
let format = match conf.args.flag_format {
Some(x) => match x.deref() {
Some(ref x) => match x.deref() {
"binary" | "bin" => DataFormat::Binary,
"hex" => DataFormat::Hex,
x => die!("Invalid --format parameter given: {:?}", x),
@ -407,7 +418,7 @@ fn execute_import(conf: Configuration) {
}
};
let informant = Informant::new(!conf.args.flag_no_color);
let informant = Informant::new(conf.have_color());
let do_import = |bytes| {
while client.queue_info().is_full() { sleep(Duration::from_secs(1)); }

26
parity/rpc.rs
View File

@ -22,6 +22,7 @@ use util::panics::PanicHandler;
use die::*;
use jsonipc;
use rpc_apis;
use std::fmt;
#[cfg(feature = "rpc")]
pub use ethcore_rpc::Server as RpcServer;
@ -44,6 +45,17 @@ pub struct IpcConfiguration {
pub apis: String,
}
impl fmt::Display for IpcConfiguration {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
if self.enabled {
write!(f, "endpoint address [{}], api list [{}]", self.socket_addr, self.apis)
}
else {
write!(f, "disabled")
}
}
}
pub struct Dependencies {
pub panic_handler: Arc<PanicHandler>,
pub apis: Arc<rpc_apis::Dependencies>,
@ -66,12 +78,6 @@ pub fn new_http(conf: HttpConfiguration, deps: &Dependencies) -> Option<RpcServe
Some(setup_http_rpc_server(deps, &addr, conf.cors, apis))
}
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))
}
fn setup_rpc_server(apis: Vec<&str>, deps: &Dependencies) -> Server {
let apis = rpc_apis::from_str(apis);
let server = Server::new();
@ -109,10 +115,18 @@ 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);

6
parity/rpc_apis.rs
View File

@ -79,7 +79,7 @@ impl FromStr for Api {
}
pub struct Dependencies {
pub signer_enabled: bool,
pub signer_port: Option<u16>,
pub signer_queue: Arc<ConfirmationsQueue>,
pub client: Arc<Client>,
pub sync: Arc<EthSync>,
@ -146,14 +146,14 @@ pub fn setup_rpc<T: Extendable>(server: T, deps: Arc<Dependencies>, apis: ApiSet
server.add_delegate(EthClient::new(&deps.client, &deps.sync, &deps.secret_store, &deps.miner, &deps.external_miner).to_delegate());
server.add_delegate(EthFilterClient::new(&deps.client, &deps.miner).to_delegate());
if deps.signer_enabled {
if deps.signer_port.is_some() {
server.add_delegate(EthSigningQueueClient::new(&deps.signer_queue).to_delegate());
} else {
server.add_delegate(EthSigningUnsafeClient::new(&deps.client, &deps.secret_store, &deps.miner).to_delegate());
}
},
Api::Personal => {
server.add_delegate(PersonalClient::new(&deps.secret_store, &deps.client, &deps.miner, deps.signer_enabled).to_delegate());
server.add_delegate(PersonalClient::new(&deps.secret_store, &deps.client, &deps.miner, deps.signer_port).to_delegate());
},
Api::Signer => {
server.add_delegate(SignerClient::new(&deps.secret_store, &deps.client, &deps.miner, &deps.signer_queue).to_delegate());

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.71", optional = true}
clippy = { version = "0.0.76", optional = true}
json-ipc-server = { git = "https://github.com/ethcore/json-ipc-server.git" }
[build-dependencies]

10
rpc/src/v1/impls/personal.rs
View File

@ -31,18 +31,18 @@ pub struct PersonalClient<A, C, M>
accounts: Weak<A>,
client: Weak<C>,
miner: Weak<M>,
signer_enabled: bool,
signer_port: Option<u16>,
}
impl<A, C, M> PersonalClient<A, C, M>
where A: AccountProvider, C: MiningBlockChainClient, M: MinerService {
/// Creates new PersonalClient
pub fn new(store: &Arc<A>, client: &Arc<C>, miner: &Arc<M>, signer_enabled: bool) -> Self {
pub fn new(store: &Arc<A>, client: &Arc<C>, miner: &Arc<M>, signer_port: Option<u16>) -> Self {
PersonalClient {
accounts: Arc::downgrade(store),
client: Arc::downgrade(client),
miner: Arc::downgrade(miner),
signer_enabled: signer_enabled,
signer_port: signer_port,
}
}
}
@ -51,7 +51,9 @@ impl<A: 'static, C: 'static, M: 'static> Personal for PersonalClient<A, C, M>
where A: AccountProvider, C: MiningBlockChainClient, M: MinerService {
fn signer_enabled(&self, _: Params) -> Result<Value, Error> {
to_value(&self.signer_enabled)
self.signer_port
.map(|v| to_value(&v))
.unwrap_or_else(|| to_value(&false))
}
fn accounts(&self, _: Params) -> Result<Value, Error> {

28
rpc/src/v1/tests/mocked/personal.rs
View File

@ -49,11 +49,11 @@ fn miner_service() -> Arc<TestMinerService> {
Arc::new(TestMinerService::default())
}
fn setup() -> PersonalTester {
fn setup(signer: Option<u16>) -> PersonalTester {
let accounts = accounts_provider();
let client = blockchain_client();
let miner = miner_service();
let personal = PersonalClient::new(&accounts, &client, &miner, false);
let personal = PersonalClient::new(&accounts, &client, &miner, signer);
let io = IoHandler::new();
io.add_delegate(personal.to_delegate());
@ -71,7 +71,7 @@ fn setup() -> PersonalTester {
#[test]
fn should_return_false_if_signer_is_disabled() {
// given
let tester = setup();
let tester = setup(None);
// when
let request = r#"{"jsonrpc": "2.0", "method": "personal_signerEnabled", "params": [], "id": 1}"#;
@ -82,9 +82,23 @@ fn should_return_false_if_signer_is_disabled() {
assert_eq!(tester.io.handle_request(request), Some(response.to_owned()));
}
#[test]
fn should_return_port_number_if_signer_is_enabled() {
// given
let tester = setup(Some(8180));
// when
let request = r#"{"jsonrpc": "2.0", "method": "personal_signerEnabled", "params": [], "id": 1}"#;
let response = r#"{"jsonrpc":"2.0","result":8180,"id":1}"#;
// then
assert_eq!(tester.io.handle_request(request), Some(response.to_owned()));
}
#[test]
fn accounts() {
let tester = setup();
let tester = setup(None);
tester.accounts.accounts
.write()
.unwrap()
@ -98,7 +112,7 @@ fn accounts() {
#[test]
fn new_account() {
let tester = setup();
let tester = setup(None);
let request = r#"{"jsonrpc": "2.0", "method": "personal_newAccount", "params": ["pass"], "id": 1}"#;
let res = tester.io.handle_request(request);
@ -122,7 +136,7 @@ fn sign_and_send_transaction_with_invalid_password() {
let account = TestAccount::new("password123");
let address = account.address();
let tester = setup();
let tester = setup(None);
tester.accounts.accounts.write().unwrap().insert(address.clone(), account);
let request = r#"{
"jsonrpc": "2.0",
@ -148,7 +162,7 @@ fn sign_and_send_transaction() {
let address = account.address();
let secret = account.secret.clone();
let tester = setup();
let tester = setup(None);
tester.accounts.accounts.write().unwrap().insert(address.clone(), account);
let request = r#"{
"jsonrpc": "2.0",

2
signer/Cargo.toml
View File

@ -20,7 +20,7 @@ ethcore-util = { path = "../util" }
ethcore-rpc = { path = "../rpc" }
parity-minimal-sysui = { git = "https://github.com/ethcore/parity-dapps-minimal-sysui-rs.git" }
clippy = { version = "0.0.71", optional = true}
clippy = { version = "0.0.76", optional = true}
[features]
dev = ["clippy"]

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.71", optional = true}
clippy = { version = "0.0.76", optional = true}
log = "0.3"
env_logger = "0.3"
time = "0.1.34"

4
util/Cargo.toml
View File

@ -12,7 +12,7 @@ log = "0.3"
env_logger = "0.3"
rustc-serialize = "0.3"
arrayvec = "0.3"
mio = "0.5.1"
mio = { git = "https://github.com/ethcore/mio", branch = "v0.5.x" }
nix ="0.5.0"
rand = "0.3.12"
time = "0.1.34"
@ -28,7 +28,7 @@ crossbeam = "0.2"
slab = "0.2"
sha3 = { path = "sha3" }
serde = "0.7.0"
clippy = { version = "0.0.71", optional = true}
clippy = { version = "0.0.76", optional = true}
json-tests = { path = "json-tests" }
igd = "0.4.2"
ethcore-devtools = { path = "../devtools" }

30
util/bigint/src/uint.rs
View File

@ -91,9 +91,9 @@ macro_rules! uint_overflowing_add_reg {
#[cfg(all(asm_available, target_arch="x86_64"))]
macro_rules! uint_overflowing_add {
(U256, $n_words: expr, $self_expr: expr, $other: expr) => ({
let mut result: [u64; 4] = unsafe { mem::uninitialized() };
let self_t: &[u64; 4] = &self.0;
let other_t: &[u64; 4] = &other.0;
let mut result: [u64; $n_words] = unsafe { mem::uninitialized() };
let self_t: &[u64; $n_words] = &$self_expr.0;
let other_t: &[u64; $n_words] = &$other.0;
let overflow: u8;
unsafe {
@ -114,9 +114,9 @@ macro_rules! uint_overflowing_add {
(U256(result), overflow != 0)
});
(U512, $n_words: expr, $self_expr: expr, $other: expr) => ({
let mut result: [u64; 8] = unsafe { mem::uninitialized() };
let self_t: &[u64; 8] = &self.0;
let other_t: &[u64; 8] = &other.0;
let mut result: [u64; $n_words] = unsafe { mem::uninitialized() };
let self_t: &[u64; $n_words] = &$self_expr.0;
let other_t: &[u64; $n_words] = &$other.0;
let overflow: u8;
@ -195,9 +195,9 @@ macro_rules! uint_overflowing_sub_reg {
#[cfg(all(asm_available, target_arch="x86_64"))]
macro_rules! uint_overflowing_sub {
(U256, $n_words: expr, $self_expr: expr, $other: expr) => ({
let mut result: [u64; 4] = unsafe { mem::uninitialized() };
let self_t: &[u64; 4] = &self.0;
let other_t: &[u64; 4] = &other.0;
let mut result: [u64; $n_words] = unsafe { mem::uninitialized() };
let self_t: &[u64; $n_words] = &$self_expr.0;
let other_t: &[u64; $n_words] = &$other.0;
let overflow: u8;
unsafe {
@ -217,9 +217,9 @@ macro_rules! uint_overflowing_sub {
(U256(result), overflow != 0)
});
(U512, $n_words: expr, $self_expr: expr, $other: expr) => ({
let mut result: [u64; 8] = unsafe { mem::uninitialized() };
let self_t: &[u64; 8] = &self.0;
let other_t: &[u64; 8] = &other.0;
let mut result: [u64; $n_words] = unsafe { mem::uninitialized() };
let self_t: &[u64; $n_words] = &$self_expr.0;
let other_t: &[u64; $n_words] = &$other.0;
let overflow: u8;
@ -269,9 +269,9 @@ macro_rules! uint_overflowing_sub {
#[cfg(all(asm_available, target_arch="x86_64"))]
macro_rules! uint_overflowing_mul {
(U256, $n_words: expr, $self_expr: expr, $other: expr) => ({
let mut result: [u64; 4] = unsafe { mem::uninitialized() };
let self_t: &[u64; 4] = &self.0;
let other_t: &[u64; 4] = &self.0;
let mut result: [u64; $n_words] = unsafe { mem::uninitialized() };
let self_t: &[u64; $n_words] = &$self_expr.0;
let other_t: &[u64; $n_words] = &$other.0;
let overflow: u64;
unsafe {

2
util/src/io/service.rs
View File

@ -335,7 +335,7 @@ impl<Message> IoService<Message> where Message: Send + Sync + Clone + 'static {
/// Starts IO event loop
pub fn start() -> Result<IoService<Message>, UtilError> {
let panic_handler = PanicHandler::new_in_arc();
let mut event_loop = EventLoop::new().unwrap();
let mut event_loop = EventLoop::new().unwrap();
let channel = event_loop.channel();
let panic = panic_handler.clone();
let thread = thread::spawn(move || {

8
util/src/keys/directory.rs
View File

@ -466,6 +466,7 @@ pub struct KeyDirectory {
}
/// Restricts the permissions of given path only to the owner.
#[cfg(not(windows))]
pub fn restrict_permissions_owner(file_path: &Path) -> Result<(), i32> {
let cstr = ::std::ffi::CString::new(file_path.to_str().unwrap()).unwrap();
match unsafe { ::libc::chmod(cstr.as_ptr(), ::libc::S_IWUSR | ::libc::S_IRUSR) } {
@ -474,6 +475,13 @@ pub fn restrict_permissions_owner(file_path: &Path) -> Result<(), i32> {
}
}
/// Restricts the permissions of given path only to the owner.
#[cfg(windows)]
pub fn restrict_permissions_owner(_file_path: &Path) -> Result<(), i32> {
//TODO: implement me
Ok(())
}
impl KeyDirectory {
/// Initializes new cache directory context with a given `path`
pub fn new(path: &Path) -> KeyDirectory {

134
util/src/network/connection.rs
View File

@ -17,6 +17,7 @@
use std::sync::Arc;
use std::collections::VecDeque;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
use mio::{Handler, Token, EventSet, EventLoop, PollOpt, TryRead, TryWrite};
use mio::tcp::*;
use hash::*;
@ -60,45 +61,57 @@ pub struct GenericConnection<Socket: GenericSocket> {
interest: EventSet,
/// Shared network statistics
stats: Arc<NetworkStats>,
/// Registered flag
registered: AtomicBool,
}
impl<Socket: GenericSocket> GenericConnection<Socket> {
pub fn expect(&mut self, size: usize) {
trace!(target:"network", "Expect to read {} bytes", size);
if self.rec_size != self.rec_buf.len() {
warn!(target:"net", "Unexpected connection read start");
warn!(target:"network", "Unexpected connection read start");
}
unsafe { self.rec_buf.set_len(0) }
self.rec_size = size;
}
/// Readable IO handler. Called when there is some data to be read.
pub fn readable(&mut self) -> io::Result<Option<Bytes>> {
if self.rec_size == 0 || self.rec_buf.len() >= self.rec_size {
warn!(target:"net", "Unexpected connection read");
warn!(target:"network", "Unexpected connection read");
}
let max = self.rec_size - self.rec_buf.len();
// resolve "multiple applicable items in scope [E0034]" error
let sock_ref = <Socket as Read>::by_ref(&mut self.socket);
match sock_ref.take(max as u64).try_read_buf(&mut self.rec_buf) {
Ok(Some(size)) if size != 0 => {
self.stats.inc_recv(size);
if self.rec_size != 0 && self.rec_buf.len() == self.rec_size {
self.rec_size = 0;
Ok(Some(::std::mem::replace(&mut self.rec_buf, Bytes::new())))
} else { Ok(None) }
},
Ok(_) => Ok(None),
Err(e) => Err(e),
}
}
loop {
let max = self.rec_size - self.rec_buf.len();
match sock_ref.take(max as u64).try_read_buf(&mut self.rec_buf) {
Ok(Some(size)) if size != 0 => {
self.stats.inc_recv(size);
trace!(target:"network", "{}: Read {} of {} bytes", self.token, self.rec_buf.len(), self.rec_size);
if self.rec_size != 0 && self.rec_buf.len() == self.rec_size {
self.rec_size = 0;
return Ok(Some(::std::mem::replace(&mut self.rec_buf, Bytes::new())))
}
else if self.rec_buf.len() > self.rec_size {
warn!(target:"network", "Read past buffer {} bytes", self.rec_buf.len() - self.rec_size);
return Ok(Some(::std::mem::replace(&mut self.rec_buf, Bytes::new())))
}
},
Ok(_) => return Ok(None),
Err(e) => {
debug!(target:"network", "Read error {} ({})", self.token, e);
return Err(e)
}
}
}
}
/// Add a packet to send queue.
pub fn send(&mut self, data: Bytes) {
pub fn send<Message>(&mut self, io: &IoContext<Message>, data: Bytes) where Message: Send + Clone {
if !data.is_empty() {
self.send_queue.push_back(Cursor::new(data));
}
if !self.interest.is_writable() {
self.interest.insert(EventSet::writable());
io.update_registration(self.token).ok();
}
}
@ -108,7 +121,7 @@ impl<Socket: GenericSocket> GenericConnection<Socket> {
}
/// Writable IO handler. Called when the socket is ready to send.
pub fn writable(&mut self) -> io::Result<WriteStatus> {
pub fn writable<Message>(&mut self, io: &IoContext<Message>) -> Result<WriteStatus, UtilError> where Message: Send + Clone {
if self.send_queue.is_empty() {
return Ok(WriteStatus::Complete)
}
@ -121,7 +134,6 @@ impl<Socket: GenericSocket> GenericConnection<Socket> {
}
match self.socket.try_write_buf(buf) {
Ok(Some(size)) if (buf.position() as usize) < send_size => {
self.interest.insert(EventSet::writable());
self.stats.inc_send(size);
Ok(WriteStatus::Ongoing)
},
@ -131,7 +143,7 @@ impl<Socket: GenericSocket> GenericConnection<Socket> {
},
Ok(Some(_)) => { panic!("Wrote past buffer");},
Ok(None) => Ok(WriteStatus::Ongoing),
Err(e) => Err(e)
Err(e) => try!(Err(e))
}
}.and_then(|r| {
if r == WriteStatus::Complete {
@ -139,9 +151,7 @@ impl<Socket: GenericSocket> GenericConnection<Socket> {
}
if self.send_queue.is_empty() {
self.interest.remove(EventSet::writable());
}
else {
self.interest.insert(EventSet::writable());
try!(io.update_registration(self.token));
}
Ok(r)
})
@ -162,6 +172,7 @@ impl Connection {
rec_size: 0,
interest: EventSet::hup() | EventSet::readable(),
stats: stats,
registered: AtomicBool::new(false),
}
}
@ -188,27 +199,36 @@ impl Connection {
rec_buf: Vec::new(),
rec_size: 0,
send_queue: self.send_queue.clone(),
interest: EventSet::hup() | EventSet::readable(),
interest: EventSet::hup(),
stats: self.stats.clone(),
registered: AtomicBool::new(false),
})
}
/// Register this connection with the IO event loop.
pub fn register_socket<Host: Handler>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> io::Result<()> {
if self.registered.load(AtomicOrdering::SeqCst) {
return Ok(());
}
trace!(target: "network", "connection register; token={:?}", reg);
if let Err(e) = event_loop.register(&self.socket, reg, self.interest, PollOpt::edge() /* | PollOpt::oneshot() */) { // TODO: oneshot is broken on windows
trace!(target: "network", "Failed to register {:?}, {:?}", reg, e);
}
self.registered.store(true, AtomicOrdering::SeqCst);
Ok(())
}
/// Update connection registration. Should be called at the end of the IO handler.
pub fn update_socket<Host: Handler>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> io::Result<()> {
trace!(target: "network", "connection reregister; token={:?}", reg);
event_loop.reregister( &self.socket, reg, self.interest, PollOpt::edge() /* | PollOpt::oneshot() */ ).or_else(|e| { // TODO: oneshot is broken on windows
trace!(target: "network", "Failed to reregister {:?}, {:?}", reg, e);
if !self.registered.load(AtomicOrdering::SeqCst) {
self.register_socket(reg, event_loop)
} else {
event_loop.reregister(&self.socket, reg, self.interest, PollOpt::edge() /* | PollOpt::oneshot() */ ).unwrap_or_else(|e| { // TODO: oneshot is broken on windows
trace!(target: "network", "Failed to reregister {:?}, {:?}", reg, e);
});
Ok(())
})
}
}
/// Delete connection registration. Should be called at the end of the IO handler.
@ -266,7 +286,7 @@ pub struct EncryptedConnection {
}
impl EncryptedConnection {
/// Create an encrypted connection out of the handshake. Consumes a handshake object.
/// Create an encrypted connection out of the handshake.
pub fn new(handshake: &mut Handshake) -> Result<EncryptedConnection, UtilError> {
let shared = try!(crypto::ecdh::agree(handshake.ecdhe.secret(), &handshake.remote_ephemeral));
let mut nonce_material = H512::new();
@ -320,7 +340,7 @@ impl EncryptedConnection {
}
/// Send a packet
pub fn send_packet(&mut self, payload: &[u8]) -> Result<(), UtilError> {
pub fn send_packet<Message>(&mut self, io: &IoContext<Message>, payload: &[u8]) -> Result<(), UtilError> where Message: Send + Clone {
let mut header = RlpStream::new();
let len = payload.len() as usize;
header.append_raw(&[(len >> 16) as u8, (len >> 8) as u8, len as u8], 1);
@ -342,7 +362,7 @@ impl EncryptedConnection {
self.egress_mac.update(&packet[32..(32 + len + padding)]);
EncryptedConnection::update_mac(&mut self.egress_mac, &mut self.mac_encoder, &[0u8; 0]);
self.egress_mac.clone().finalize(&mut packet[(32 + len + padding)..]);
self.connection.send(packet);
self.connection.send(io, packet);
Ok(())
}
@ -417,31 +437,29 @@ impl EncryptedConnection {
/// Readable IO handler. Tracker receive status and returns decoded packet if avaialable.
pub fn readable<Message>(&mut self, io: &IoContext<Message>) -> Result<Option<Packet>, UtilError> where Message: Send + Clone{
io.clear_timer(self.connection.token).unwrap();
match self.read_state {
EncryptedConnectionState::Header => {
if let Some(data) = try!(self.connection.readable()) {
try!(self.read_header(&data));
try!(io.register_timer(self.connection.token, RECIEVE_PAYLOAD_TIMEOUT));
}
Ok(None)
},
EncryptedConnectionState::Payload => {
match try!(self.connection.readable()) {
Some(data) => {
self.read_state = EncryptedConnectionState::Header;
self.connection.expect(ENCRYPTED_HEADER_LEN);
Ok(Some(try!(self.read_payload(&data))))
},
None => Ok(None)
}
if let EncryptedConnectionState::Header = self.read_state {
if let Some(data) = try!(self.connection.readable()) {
try!(self.read_header(&data));
try!(io.register_timer(self.connection.token, RECIEVE_PAYLOAD_TIMEOUT));
}
};
if let EncryptedConnectionState::Payload = self.read_state {
match try!(self.connection.readable()) {
Some(data) => {
self.read_state = EncryptedConnectionState::Header;
self.connection.expect(ENCRYPTED_HEADER_LEN);
Ok(Some(try!(self.read_payload(&data))))
},
None => Ok(None)
}
} else {
Ok(None)
}
}
/// Writable IO handler. Processes send queeue.
pub fn writable<Message>(&mut self, io: &IoContext<Message>) -> Result<(), UtilError> where Message: Send + Clone {
io.clear_timer(self.connection.token).unwrap();
try!(self.connection.writable());
try!(self.connection.writable(io));
Ok(())
}
}
@ -472,12 +490,14 @@ pub fn test_encryption() {
mod tests {
use super::*;
use std::sync::*;
use std::sync::atomic::AtomicBool;
use super::super::stats::*;
use std::io::{Read, Write, Error, Cursor, ErrorKind};
use mio::{EventSet};
use std::collections::VecDeque;
use bytes::*;
use devtools::*;
use io::*;
impl GenericSocket for TestSocket {}
@ -521,6 +541,7 @@ mod tests {
rec_size: 0,
interest: EventSet::hup() | EventSet::readable(),
stats: Arc::<NetworkStats>::new(NetworkStats::new()),
registered: AtomicBool::new(false),
}
}
}
@ -543,10 +564,15 @@ mod tests {
rec_size: 0,
interest: EventSet::hup() | EventSet::readable(),
stats: Arc::<NetworkStats>::new(NetworkStats::new()),
registered: AtomicBool::new(false),
}
}
}
fn test_io() -> IoContext<i32> {
IoContext::new(IoChannel::disconnected(), 0)
}
#[test]
fn connection_expect() {
let mut connection = TestConnection::new();
@ -557,7 +583,7 @@ mod tests {
#[test]
fn connection_write_empty() {
let mut connection = TestConnection::new();
let status = connection.writable();
let status = connection.writable(&test_io());
assert!(status.is_ok());
assert!(WriteStatus::Complete == status.unwrap());
}
@ -568,7 +594,7 @@ mod tests {
let data = Cursor::new(vec![0; 10240]);
connection.send_queue.push_back(data);
let status = connection.writable();
let status = connection.writable(&test_io());
assert!(status.is_ok());
assert!(WriteStatus::Complete == status.unwrap());
assert_eq!(10240, connection.socket.write_buffer.len());
@ -581,7 +607,7 @@ mod tests {
let data = Cursor::new(vec![0; 10240]);
connection.send_queue.push_back(data);
let status = connection.writable();
let status = connection.writable(&test_io());
assert!(status.is_ok());
assert!(WriteStatus::Ongoing == status.unwrap());
@ -594,7 +620,7 @@ mod tests {
let data = Cursor::new(vec![0; 10240]);
connection.send_queue.push_back(data);
let status = connection.writable();
let status = connection.writable(&test_io());
assert!(!status.is_ok());
assert_eq!(1, connection.send_queue.len());

54
util/src/network/handshake.rs
View File

@ -111,7 +111,7 @@ impl Handshake {
self.originated = originated;
io.register_timer(self.connection.token, HANDSHAKE_TIMEOUT).ok();
if originated {
try!(self.write_auth(host.secret(), host.id()));
try!(self.write_auth(io, host.secret(), host.id()));
}
else {
self.state = HandshakeState::ReadingAuth;
@ -128,17 +128,17 @@ impl Handshake {
/// Readable IO handler. Drives the state change.
pub fn readable<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo) -> Result<(), UtilError> where Message: Send + Clone {
if !self.expired() {
io.clear_timer(self.connection.token).unwrap();
io.clear_timer(self.connection.token).ok();
match self.state {
HandshakeState::New => {}
HandshakeState::ReadingAuth => {
if let Some(data) = try!(self.connection.readable()) {
try!(self.read_auth(host.secret(), &data));
try!(self.read_auth(io, host.secret(), &data));
};
},
HandshakeState::ReadingAuthEip8 => {
if let Some(data) = try!(self.connection.readable()) {
try!(self.read_auth_eip8(host.secret(), &data));
try!(self.read_auth_eip8(io, host.secret(), &data));
};
},
HandshakeState::ReadingAck => {
@ -153,9 +153,6 @@ impl Handshake {
},
HandshakeState::StartSession => {},
}
if self.state != HandshakeState::StartSession {
try!(io.update_registration(self.connection.token));
}
}
Ok(())
}
@ -163,11 +160,7 @@ impl Handshake {
/// Writabe IO handler.
pub fn writable<Message>(&mut self, io: &IoContext<Message>) -> Result<(), UtilError> where Message: Send + Clone {
if !self.expired() {
io.clear_timer(self.connection.token).unwrap();
try!(self.connection.writable());
if self.state != HandshakeState::StartSession {
io.update_registration(self.connection.token).unwrap();
}
try!(self.connection.writable(io));
}
Ok(())
}
@ -183,7 +176,7 @@ impl Handshake {
}
/// Parse, validate and confirm auth message
fn read_auth(&mut self, secret: &Secret, data: &[u8]) -> Result<(), UtilError> {
fn read_auth<Message>(&mut self, io: &IoContext<Message>, secret: &Secret, data: &[u8]) -> Result<(), UtilError> where Message: Send + Clone {
trace!(target:"network", "Received handshake auth from {:?}", self.connection.remote_addr_str());
if data.len() != V4_AUTH_PACKET_SIZE {
debug!(target:"net", "Wrong auth packet size");
@ -197,7 +190,7 @@ impl Handshake {
let (pubk, rest) = rest.split_at(64);
let (nonce, _) = rest.split_at(32);
try!(self.set_auth(secret, sig, pubk, nonce, PROTOCOL_VERSION));
try!(self.write_ack());
try!(self.write_ack(io));
}
Err(_) => {
// Try to interpret as EIP-8 packet
@ -214,7 +207,7 @@ impl Handshake {
Ok(())
}
fn read_auth_eip8(&mut self, secret: &Secret, data: &[u8]) -> Result<(), UtilError> {
fn read_auth_eip8<Message>(&mut self, io: &IoContext<Message>, secret: &Secret, data: &[u8]) -> Result<(), UtilError> where Message: Send + Clone {
trace!(target:"network", "Received EIP8 handshake auth from {:?}", self.connection.remote_addr_str());
self.auth_cipher.extend_from_slice(data);
let auth = try!(ecies::decrypt(secret, &self.auth_cipher[0..2], &self.auth_cipher[2..]));
@ -224,13 +217,13 @@ impl Handshake {
let remote_nonce: H256 = try!(rlp.val_at(2));
let remote_version: u64 = try!(rlp.val_at(3));
try!(self.set_auth(secret, &signature, &remote_public, &remote_nonce, remote_version));
try!(self.write_ack_eip8());
try!(self.write_ack_eip8(io));
Ok(())
}
/// Parse and validate ack message
fn read_ack(&mut self, secret: &Secret, data: &[u8]) -> Result<(), UtilError> {
trace!(target:"network", "Received handshake auth to {:?}", self.connection.remote_addr_str());
trace!(target:"network", "Received handshake ack from {:?}", self.connection.remote_addr_str());
if data.len() != V4_ACK_PACKET_SIZE {
debug!(target:"net", "Wrong ack packet size");
return Err(From::from(NetworkError::BadProtocol));
@ -270,7 +263,7 @@ impl Handshake {
}
/// Sends auth message
fn write_auth(&mut self, secret: &Secret, public: &Public) -> Result<(), UtilError> {
fn write_auth<Message>(&mut self, io: &IoContext<Message>, secret: &Secret, public: &Public) -> Result<(), UtilError> where Message: Send + Clone {
trace!(target:"network", "Sending handshake auth to {:?}", self.connection.remote_addr_str());
let mut data = [0u8; /*Signature::SIZE*/ 65 + /*H256::SIZE*/ 32 + /*Public::SIZE*/ 64 + /*H256::SIZE*/ 32 + 1]; //TODO: use associated constants
let len = data.len();
@ -290,14 +283,14 @@ impl Handshake {
}
let message = try!(crypto::ecies::encrypt(&self.id, &[], &data));
self.auth_cipher = message.clone();
self.connection.send(message);
self.connection.send(io, message);
self.connection.expect(V4_ACK_PACKET_SIZE);
self.state = HandshakeState::ReadingAck;
Ok(())
}
/// Sends ack message
fn write_ack(&mut self) -> Result<(), UtilError> {
fn write_ack<Message>(&mut self, io: &IoContext<Message>) -> Result<(), UtilError> where Message: Send + Clone {
trace!(target:"network", "Sending handshake ack to {:?}", self.connection.remote_addr_str());
let mut data = [0u8; 1 + /*Public::SIZE*/ 64 + /*H256::SIZE*/ 32]; //TODO: use associated constants
let len = data.len();
@ -310,13 +303,13 @@ impl Handshake {
}
let message = try!(crypto::ecies::encrypt(&self.id, &[], &data));
self.ack_cipher = message.clone();
self.connection.send(message);
self.connection.send(io, message);
self.state = HandshakeState::StartSession;
Ok(())
}
/// Sends EIP8 ack message
fn write_ack_eip8(&mut self) -> Result<(), UtilError> {
fn write_ack_eip8<Message>(&mut self, io: &IoContext<Message>) -> Result<(), UtilError> where Message: Send + Clone {
trace!(target:"network", "Sending EIP8 handshake ack to {:?}", self.connection.remote_addr_str());
let mut rlp = RlpStream::new_list(3);
rlp.append(self.ecdhe.public());
@ -333,7 +326,7 @@ impl Handshake {
let message = try!(crypto::ecies::encrypt(&self.id, &prefix, &encoded));
self.ack_cipher.extend_from_slice(&prefix);
self.ack_cipher.extend_from_slice(&message);
self.connection.send(self.ack_cipher.clone());
self.connection.send(io, self.ack_cipher.clone());
self.state = HandshakeState::StartSession;
Ok(())
}
@ -347,6 +340,7 @@ mod test {
use super::*;
use crypto::*;
use hash::*;
use io::*;
use std::net::SocketAddr;
use mio::tcp::TcpStream;
use network::stats::NetworkStats;
@ -371,6 +365,10 @@ mod test {
Handshake::new(0, to, socket, &nonce, Arc::new(NetworkStats::new())).unwrap()
}
fn test_io() -> IoContext<i32> {
IoContext::new(IoChannel::disconnected(), 0)
}
#[test]
fn test_handshake_auth_plain() {
let mut h = create_handshake(None);
@ -387,7 +385,7 @@ mod test {
a4592ee77e2bd94d0be3691f3b406f9bba9b591fc63facc016bfa8\
".from_hex().unwrap();
h.read_auth(&secret, &auth).unwrap();
h.read_auth(&test_io(), &secret, &auth).unwrap();
assert_eq!(h.state, super::HandshakeState::StartSession);
check_auth(&h, 4);
}
@ -411,9 +409,9 @@ mod test {
3bf7678318e2d5b5340c9e488eefea198576344afbdf66db5f51204a6961a63ce072c8926c\
".from_hex().unwrap();
h.read_auth(&secret, &auth[0..super::V4_AUTH_PACKET_SIZE]).unwrap();
h.read_auth(&test_io(), &secret, &auth[0..super::V4_AUTH_PACKET_SIZE]).unwrap();
assert_eq!(h.state, super::HandshakeState::ReadingAuthEip8);
h.read_auth_eip8(&secret, &auth[super::V4_AUTH_PACKET_SIZE..]).unwrap();
h.read_auth_eip8(&test_io(), &secret, &auth[super::V4_AUTH_PACKET_SIZE..]).unwrap();
assert_eq!(h.state, super::HandshakeState::StartSession);
check_auth(&h, 4);
}
@ -438,9 +436,9 @@ mod test {
d490\
".from_hex().unwrap();
h.read_auth(&secret, &auth[0..super::V4_AUTH_PACKET_SIZE]).unwrap();
h.read_auth(&test_io(), &secret, &auth[0..super::V4_AUTH_PACKET_SIZE]).unwrap();
assert_eq!(h.state, super::HandshakeState::ReadingAuthEip8);
h.read_auth_eip8(&secret, &auth[super::V4_AUTH_PACKET_SIZE..]).unwrap();
h.read_auth_eip8(&test_io(), &secret, &auth[super::V4_AUTH_PACKET_SIZE..]).unwrap();
assert_eq!(h.state, super::HandshakeState::StartSession);
check_auth(&h, 56);
let ack = h.ack_cipher.clone();

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

@ -215,8 +215,7 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
pub fn send(&self, peer: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), UtilError> {
let session = self.resolve_session(peer);
if let Some(session) = session {
try!(session.lock().unwrap().deref_mut().send_packet(self.protocol, packet_id as u8, &data));
try!(self.io.update_registration(peer));
try!(session.lock().unwrap().deref_mut().send_packet(self.io, self.protocol, packet_id as u8, &data));
} else {
trace!(target: "network", "Send: Peer no longer exist")
}
@ -494,7 +493,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
for e in self.sessions.write().unwrap().iter_mut() {
let mut s = e.lock().unwrap();
if !s.keep_alive(io) {
s.disconnect(DisconnectReason::PingTimeout);
s.disconnect(io, DisconnectReason::PingTimeout);
to_kill.push(s.token());
}
}
@ -616,10 +615,8 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
trace!(target: "network", "Session write error: {}: {:?}", token, e);
}
if s.done() {
io.deregister_stream(token).expect("Error deregistering stream");
} else {
io.update_registration(token).unwrap_or_else(|e| debug!(target: "network", "Session registration error: {:?}", e));
}
io.deregister_stream(token).unwrap_or_else(|e| debug!("Error deregistering stream: {:?}", e));
}
}
}
@ -630,62 +627,65 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
fn session_readable(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) {
let mut ready_data: Vec<ProtocolId> = Vec::new();
let mut packet_data: Option<(ProtocolId, PacketId, Vec<u8>)> = None;
let mut packet_data: Vec<(ProtocolId, PacketId, Vec<u8>)> = Vec::new();
let mut kill = false;
let session = { self.sessions.read().unwrap().get(token).cloned() };
if let Some(session) = session.clone() {
let mut s = session.lock().unwrap();
match s.readable(io, &self.info.read().unwrap()) {
Err(e) => {
trace!(target: "network", "Session read error: {}:{:?} ({:?}) {:?}", token, s.id(), s.remote_addr(), e);
match e {
UtilError::Network(NetworkError::Disconnect(DisconnectReason::UselessPeer)) |
UtilError::Network(NetworkError::Disconnect(DisconnectReason::IncompatibleProtocol)) => {
if let Some(id) = s.id() {
self.nodes.write().unwrap().mark_as_useless(id);
loop {
match s.readable(io, &self.info.read().unwrap()) {
Err(e) => {
trace!(target: "network", "Session read error: {}:{:?} ({:?}) {:?}", token, s.id(), s.remote_addr(), e);
match e {
UtilError::Network(NetworkError::Disconnect(DisconnectReason::UselessPeer)) |
UtilError::Network(NetworkError::Disconnect(DisconnectReason::IncompatibleProtocol)) => {
if let Some(id) = s.id() {
self.nodes.write().unwrap().mark_as_useless(id);
}
}
_ => (),
}
kill = true;
break;
},
Ok(SessionData::Ready) => {
if !s.info.originated {
let session_count = self.session_count();
let ideal_peers = { self.info.read().unwrap().deref().config.ideal_peers };
if session_count >= ideal_peers as usize {
s.disconnect(io, DisconnectReason::TooManyPeers);
return;
}
// Add it no node table
if let Ok(address) = s.remote_addr() {
let entry = NodeEntry { id: s.id().unwrap().clone(), endpoint: NodeEndpoint { address: address, udp_port: address.port() } };
self.nodes.write().unwrap().add_node(Node::new(entry.id.clone(), entry.endpoint.clone()));
let mut discovery = self.discovery.lock().unwrap();
if let Some(ref mut discovery) = *discovery.deref_mut() {
discovery.add_node(entry);
}
}
}
_ => (),
}
kill = true;
},
Ok(SessionData::Ready) => {
if !s.info.originated {
let session_count = self.session_count();
let ideal_peers = { self.info.read().unwrap().deref().config.ideal_peers };
if session_count >= ideal_peers as usize {
s.disconnect(DisconnectReason::TooManyPeers);
return;
}
// Add it no node table
if let Ok(address) = s.remote_addr() {
let entry = NodeEntry { id: s.id().unwrap().clone(), endpoint: NodeEndpoint { address: address, udp_port: address.port() } };
self.nodes.write().unwrap().add_node(Node::new(entry.id.clone(), entry.endpoint.clone()));
let mut discovery = self.discovery.lock().unwrap();
if let Some(ref mut discovery) = *discovery.deref_mut() {
discovery.add_node(entry);
self.num_sessions.fetch_add(1, AtomicOrdering::SeqCst);
for (p, _) in self.handlers.read().unwrap().iter() {
if s.have_capability(p) {
ready_data.push(p);
}
}
}
self.num_sessions.fetch_add(1, AtomicOrdering::SeqCst);
for (p, _) in self.handlers.read().unwrap().iter() {
if s.have_capability(p) {
ready_data.push(p);
},
Ok(SessionData::Packet {
data,
protocol,
packet_id,
}) => {
match self.handlers.read().unwrap().get(protocol) {
None => { warn!(target: "network", "No handler found for protocol: {:?}", protocol) },
Some(_) => packet_data.push((protocol, packet_id, data)),
}
}
},
Ok(SessionData::Packet {
data,
protocol,
packet_id,
}) => {
match self.handlers.read().unwrap().get(protocol) {
None => { warn!(target: "network", "No handler found for protocol: {:?}", protocol) },
Some(_) => packet_data = Some((protocol, packet_id, data)),
}
},
Ok(SessionData::None) => {},
}
},
Ok(SessionData::None) => break,
}
}
}
if kill {
self.kill_connection(token, io, true);
@ -695,11 +695,10 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
self.stats.inc_sessions();
h.connected(&NetworkContext::new(io, p, session.clone(), self.sessions.clone()), &token);
}
if let Some((p, packet_id, data)) = packet_data {
for (p, packet_id, data) in packet_data {
let h = self.handlers.read().unwrap().get(p).unwrap().clone();
h.read(&NetworkContext::new(io, p, session.clone(), self.sessions.clone()), &token, packet_id, &data[1..]);
}
io.update_registration(token).unwrap_or_else(|e| debug!(target: "network", "Token registration error: {:?}", e));
}
fn connection_timeout(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) {
@ -742,10 +741,8 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
h.disconnected(&NetworkContext::new(io, p, expired_session.clone(), self.sessions.clone()), &token);
}
if deregister {
io.deregister_stream(token).expect("Error deregistering stream");
} else if expired_session.is_some() {
io.update_registration(token).unwrap_or_else(|e| debug!(target: "network", "Connection registration error: {:?}", e));
}
io.deregister_stream(token).unwrap_or_else(|e| debug!("Error deregistering stream: {:?}", e));
}
}
fn update_nodes(&self, io: &IoContext<NetworkIoMessage<Message>>, node_changes: TableUpdates) {
@ -874,7 +871,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
NetworkIoMessage::Disconnect(ref peer) => {
let session = { self.sessions.read().unwrap().get(*peer).cloned() };
if let Some(session) = session {
session.lock().unwrap().disconnect(DisconnectReason::DisconnectRequested);
session.lock().unwrap().disconnect(io, DisconnectReason::DisconnectRequested);
}
trace!(target: "network", "Disconnect requested {}", peer);
self.kill_connection(*peer, io, false);
@ -882,7 +879,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
NetworkIoMessage::DisablePeer(ref peer) => {
let session = { self.sessions.read().unwrap().get(*peer).cloned() };
if let Some(session) = session {
session.lock().unwrap().disconnect(DisconnectReason::DisconnectRequested);
session.lock().unwrap().disconnect(io, DisconnectReason::DisconnectRequested);
if let Some(id) = session.lock().unwrap().id() {
self.nodes.write().unwrap().mark_as_useless(id)
}

53
util/src/network/session.rs
View File

@ -145,7 +145,7 @@ impl Session {
})
}
fn complete_handshake(&mut self, host: &HostInfo) -> Result<(), UtilError> {
fn complete_handshake<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo) -> Result<(), UtilError> where Message: Send + Sync + Clone {
let connection = if let State::Handshake(ref mut h) = self.state {
self.info.id = Some(h.id.clone());
try!(EncryptedConnection::new(h))
@ -153,8 +153,8 @@ impl Session {
panic!("Unexpected state");
};
self.state = State::Session(connection);
try!(self.write_hello(host));
try!(self.send_ping());
try!(self.write_hello(io, host));
try!(self.send_ping(io));
Ok(())
}
@ -220,10 +220,11 @@ impl Session {
}
}
if let Some(data) = packet_data {
return Ok(try!(self.read_packet(data, host)));
return Ok(try!(self.read_packet(io, data, host)));
}
if create_session {
try!(self.complete_handshake(host));
try!(self.complete_handshake(io, host));
io.update_registration(self.token()).unwrap_or_else(|e| debug!(target: "network", "Token registration error: {:?}", e));
}
Ok(SessionData::None)
}
@ -263,7 +264,8 @@ impl Session {
}
/// Send a protocol packet to peer.
pub fn send_packet(&mut self, protocol: &str, packet_id: u8, data: &[u8]) -> Result<(), UtilError> {
pub fn send_packet<Message>(&mut self, io: &IoContext<Message>, protocol: &str, packet_id: u8, data: &[u8]) -> Result<(), UtilError>
where Message: Send + Sync + Clone {
if self.info.capabilities.is_empty() || !self.had_hello {
debug!(target: "network", "Sending to unconfirmed session {}, protocol: {}, packet: {}", self.token(), protocol, packet_id);
return Err(From::from(NetworkError::BadProtocol));
@ -283,7 +285,7 @@ impl Session {
let mut rlp = RlpStream::new();
rlp.append(&(pid as u32));
rlp.append_raw(data, 1);
self.send(rlp)
self.send(io, rlp)
}
/// Keep this session alive. Returns false if ping timeout happened
@ -298,10 +300,9 @@ impl Session {
};
if !timed_out && time::precise_time_ns() - self.ping_time_ns > PING_INTERVAL_SEC * 1000_000_000 {
if let Err(e) = self.send_ping() {
if let Err(e) = self.send_ping(io) {
debug!("Error sending ping message: {:?}", e);
}
io.update_registration(self.token()).unwrap_or_else(|e| debug!(target: "network", "Session registration error: {:?}", e));
}
!timed_out
}
@ -310,7 +311,8 @@ impl Session {
self.connection().token()
}
fn read_packet(&mut self, packet: Packet, host: &HostInfo) -> Result<SessionData, UtilError> {
fn read_packet<Message>(&mut self, io: &IoContext<Message>, packet: Packet, host: &HostInfo) -> Result<SessionData, UtilError>
where Message: Send + Sync + Clone {
if packet.data.len() < 2 {
return Err(From::from(NetworkError::BadProtocol));
}
@ -321,7 +323,7 @@ impl Session {
match packet_id {
PACKET_HELLO => {
let rlp = UntrustedRlp::new(&packet.data[1..]); //TODO: validate rlp expected size
try!(self.read_hello(&rlp, host));
try!(self.read_hello(io, &rlp, host));
Ok(SessionData::Ready)
},
PACKET_DISCONNECT => {
@ -330,7 +332,7 @@ impl Session {
Err(From::from(NetworkError::Disconnect(DisconnectReason::from_u8(reason))))
}
PACKET_PING => {
try!(self.send_pong());
try!(self.send_pong(io));
Ok(SessionData::None)
},
PACKET_PONG => {
@ -362,7 +364,7 @@ impl Session {
}
}
fn write_hello(&mut self, host: &HostInfo) -> Result<(), UtilError> {
fn write_hello<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo) -> Result<(), UtilError> where Message: Send + Sync + Clone {
let mut rlp = RlpStream::new();
rlp.append_raw(&[PACKET_HELLO as u8], 0);
rlp.begin_list(5)
@ -371,10 +373,11 @@ impl Session {
.append(&host.capabilities)
.append(&host.local_endpoint.address.port())
.append(host.id());
self.send(rlp)
self.send(io, rlp)
}
fn read_hello(&mut self, rlp: &UntrustedRlp, host: &HostInfo) -> Result<(), UtilError> {
fn read_hello<Message>(&mut self, io: &IoContext<Message>, rlp: &UntrustedRlp, host: &HostInfo) -> Result<(), UtilError>
where Message: Send + Sync + Clone {
let protocol = try!(rlp.val_at::<u32>(0));
let client_version = try!(rlp.val_at::<String>(1));
let peer_caps = try!(rlp.val_at::<Vec<PeerCapabilityInfo>>(2));
@ -417,36 +420,36 @@ impl Session {
self.info.capabilities = caps;
if self.info.capabilities.is_empty() {
trace!(target: "network", "No common capabilities with peer.");
return Err(From::from(self.disconnect(DisconnectReason::UselessPeer)));
return Err(From::from(self.disconnect(io, DisconnectReason::UselessPeer)));
}
if protocol != host.protocol_version {
trace!(target: "network", "Peer protocol version mismatch: {}", protocol);
return Err(From::from(self.disconnect(DisconnectReason::UselessPeer)));
return Err(From::from(self.disconnect(io, DisconnectReason::UselessPeer)));
}
self.had_hello = true;
Ok(())
}
/// Senf ping packet
pub fn send_ping(&mut self) -> Result<(), UtilError> {
try!(self.send(try!(Session::prepare(PACKET_PING))));
pub fn send_ping<Message>(&mut self, io: &IoContext<Message>) -> Result<(), UtilError> where Message: Send + Sync + Clone {
try!(self.send(io, try!(Session::prepare(PACKET_PING))));
self.ping_time_ns = time::precise_time_ns();
self.pong_time_ns = None;
Ok(())
}
fn send_pong(&mut self) -> Result<(), UtilError> {
self.send(try!(Session::prepare(PACKET_PONG)))
fn send_pong<Message>(&mut self, io: &IoContext<Message>) -> Result<(), UtilError> where Message: Send + Sync + Clone {
self.send(io, try!(Session::prepare(PACKET_PONG)))
}
/// Disconnect this session
pub fn disconnect(&mut self, reason: DisconnectReason) -> NetworkError {
pub fn disconnect<Message>(&mut self, io: &IoContext<Message>, reason: DisconnectReason) -> NetworkError where Message: Send + Sync + Clone {
if let State::Session(_) = self.state {
let mut rlp = RlpStream::new();
rlp.append(&(PACKET_DISCONNECT as u32));
rlp.begin_list(1);
rlp.append(&(reason as u32));
self.send(rlp).ok();
self.send(io, rlp).ok();
}
NetworkError::Disconnect(reason)
}
@ -458,13 +461,13 @@ impl Session {
Ok(rlp)
}
fn send(&mut self, rlp: RlpStream) -> Result<(), UtilError> {
fn send<Message>(&mut self, io: &IoContext<Message>, rlp: RlpStream) -> Result<(), UtilError> where Message: Send + Sync + Clone {
match self.state {
State::Handshake(_) => {
warn!(target:"network", "Unexpected send request");
},
State::Session(ref mut s) => {
try!(s.send_packet(&rlp.out()))
try!(s.send_packet(io, &rlp.out()))
},
}
Ok(())