diff --git a/rpc/src/v1/impls/eth.rs b/rpc/src/v1/impls/eth.rs index 1c417dcbc..565b255d9 100644 --- a/rpc/src/v1/impls/eth.rs +++ b/rpc/src/v1/impls/eth.rs @@ -178,6 +178,29 @@ impl EthClient data: request.data.map_or_else(Vec::new, |d| d.to_vec()) }.fake_sign(from) } + + fn dispatch_transaction(&self, signed_transaction: SignedTransaction, raw_transaction: Vec) -> Result { + let hash = signed_transaction.hash(); + + let import = { + let client = take_weak!(self.client); + take_weak!(self.miner).import_transactions(vec![signed_transaction], |a: &Address| AccountDetails { + nonce: client.nonce(a), + balance: client.balance(a), + }) + }; + + match import.into_iter().collect::, _>>() { + Ok(_) => { + take_weak!(self.sync).new_transaction(raw_transaction); + to_value(&hash) + } + Err(e) => { + warn!("Error sending transaction: {:?}", e); + to_value(&H256::zero()) + } + } + } } const MAX_QUEUE_SIZE_TO_MINE_ON: usize = 4; // because uncles go back 6. @@ -459,32 +482,19 @@ impl Eth for EthClient let accounts = take_weak!(self.accounts); match accounts.account_secret(&request.from) { Ok(secret) => { - let miner = take_weak!(self.miner); - let client = take_weak!(self.client); - - let transaction = EthTransaction { - nonce: request.nonce.unwrap_or_else(|| client.nonce(&request.from)), - action: request.to.map_or(Action::Create, Action::Call), - gas: request.gas.unwrap_or_else(default_gas), - gas_price: request.gas_price.unwrap_or_else(default_gas_price), - value: request.value.unwrap_or_else(U256::zero), - data: request.data.map_or_else(Vec::new, |d| d.to_vec()) + let signed_transaction = { + let client = take_weak!(self.client); + EthTransaction { + nonce: request.nonce.unwrap_or_else(|| client.nonce(&request.from)), + action: request.to.map_or(Action::Create, Action::Call), + gas: request.gas.unwrap_or_else(default_gas), + gas_price: request.gas_price.unwrap_or_else(default_gas_price), + value: request.value.unwrap_or_else(U256::zero), + data: request.data.map_or_else(Vec::new, |d| d.to_vec()), + }.sign(&secret) }; - - let signed_transaction = transaction.sign(&secret); - let hash = signed_transaction.hash(); - - let import = miner.import_transactions(vec![signed_transaction], |a: &Address| AccountDetails { - nonce: client.nonce(a), - balance: client.balance(a), - }); - match import.into_iter().collect::, _>>() { - Ok(_) => to_value(&hash), - Err(e) => { - warn!("Error sending transaction: {:?}", e); - to_value(&H256::zero()) - } - } + let raw_transaction = encode(&signed_transaction).to_vec(); + self.dispatch_transaction(signed_transaction, raw_transaction) }, Err(_) => { to_value(&H256::zero()) } } @@ -494,26 +504,10 @@ impl Eth for EthClient fn send_raw_transaction(&self, params: Params) -> Result { from_params::<(Bytes, )>(params) .and_then(|(raw_transaction, )| { - let decoded: Result = UntrustedRlp::new(&raw_transaction.to_vec()).as_val(); - match decoded { - Ok(signed_tx) => { - let miner = take_weak!(self.miner); - let client = take_weak!(self.client); - - let hash = signed_tx.hash(); - let import = miner.import_transactions(vec![signed_tx], |a: &Address| AccountDetails { - nonce: client.nonce(a), - balance: client.balance(a), - }); - match import.into_iter().collect::, _>>() { - Ok(_) => to_value(&hash), - Err(e) => { - warn!("Error sending transaction: {:?}", e); - to_value(&H256::zero()) - } - } - }, - Err(_) => { to_value(&H256::zero()) } + let raw_transaction = raw_transaction.to_vec(); + match UntrustedRlp::new(&raw_transaction).as_val() { + Ok(signed_transaction) => self.dispatch_transaction(signed_transaction, raw_transaction), + Err(_) => to_value(&H256::zero()), } }) } diff --git a/rpc/src/v1/tests/helpers/sync_provider.rs b/rpc/src/v1/tests/helpers/sync_provider.rs index 527974140..59188f0a7 100644 --- a/rpc/src/v1/tests/helpers/sync_provider.rs +++ b/rpc/src/v1/tests/helpers/sync_provider.rs @@ -16,7 +16,7 @@ //! Test implementation of SyncProvider. -use util::U256; +use util::{U256, Bytes}; use ethsync::{SyncProvider, SyncStatus, SyncState}; use std::sync::{RwLock}; @@ -59,5 +59,8 @@ impl SyncProvider for TestSyncProvider { fn status(&self) -> SyncStatus { self.status.read().unwrap().clone() } + + fn new_transaction(&self, _raw_transaction: Bytes) { + } } diff --git a/sync/src/chain.rs b/sync/src/chain.rs index f84f91f77..eccc8ae98 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -217,6 +217,10 @@ pub struct ChainSync { network_id: U256, /// Miner miner: Arc, + + /// Transactions to propagate + // TODO: reconsider where this is in the codebase - seems a little dodgy to have here. + transactions_to_send: Vec, } type RlpResponseResult = Result, PacketDecodeError>; @@ -243,6 +247,7 @@ impl ChainSync { max_download_ahead_blocks: max(MAX_HEADERS_TO_REQUEST, config.max_download_ahead_blocks), network_id: config.network_id, miner: miner, + transactions_to_send: vec![], } } @@ -944,6 +949,12 @@ impl ChainSync { sync.disable_peer(peer_id); } } + + /// Place a new transaction on the wire. + pub fn new_transaction(&mut self, raw_transaction: Bytes) { + self.transactions_to_send.push(raw_transaction); + } + /// Called when peer sends us new transactions fn on_peer_transactions(&mut self, io: &mut SyncIo, peer_id: PeerId, r: &UntrustedRlp) -> Result<(), PacketDecodeError> { // accepting transactions once only fully synced @@ -1277,7 +1288,44 @@ impl ChainSync { sent } + /// propagates new transactions to all peers + fn propagate_new_transactions(&mut self, io: &mut SyncIo) -> usize { + + // Early out of nobody to send to. + if self.peers.len() == 0 { + return 0; + } + + let mut packet = RlpStream::new_list(self.transactions_to_send.len()); + for tx in self.transactions_to_send.iter() { + packet.append_raw(tx, 1); + } + self.transactions_to_send.clear(); + let rlp = packet.out(); + + let lucky_peers = { + // sqrt(x)/x scaled to max u32 + let fraction = (self.peers.len() as f64).powf(-0.5).mul(u32::max_value() as f64).round() as u32; + let small = self.peers.len() < MIN_PEERS_PROPAGATION; + let lucky_peers = self.peers.iter() + .filter_map(|(&p, _)| if small || ::rand::random::() < fraction { Some(p.clone()) } else { None }) + .collect::>(); + + // taking at max of MAX_PEERS_PROPAGATION + lucky_peers.iter().map(|&id| id.clone()).take(min(lucky_peers.len(), MAX_PEERS_PROPAGATION)).collect::>() + }; + + let sent = lucky_peers.len(); + for peer_id in lucky_peers { + self.send_packet(io, peer_id, TRANSACTIONS_PACKET, rlp.clone()); + } + sent + } + fn propagate_latest_blocks(&mut self, io: &mut SyncIo) { + if !self.transactions_to_send.is_empty() { + self.propagate_new_transactions(io); + } let chain_info = io.chain().chain_info(); if (((chain_info.best_block_number as i64) - (self.last_sent_block_number as i64)).abs() as BlockNumber) < MAX_PEER_LAG_PROPAGATION { let blocks = self.propagate_blocks(&chain_info, io); diff --git a/sync/src/lib.rs b/sync/src/lib.rs index a4f6eff38..ea4a1daea 100644 --- a/sync/src/lib.rs +++ b/sync/src/lib.rs @@ -66,7 +66,7 @@ use std::ops::*; use std::sync::*; use util::network::{NetworkProtocolHandler, NetworkService, NetworkContext, PeerId}; use util::TimerToken; -use util::{U256, ONE_U256}; +use util::{U256, Bytes, ONE_U256}; use ethcore::client::Client; use ethcore::service::SyncMessage; use ethminer::Miner; @@ -101,6 +101,9 @@ impl Default for SyncConfig { pub trait SyncProvider: Send + Sync { /// Get sync status fn status(&self) -> SyncStatus; + + /// Note that a user has submitted a new transaction. + fn new_transaction(&self, raw_transaction: Bytes); } /// Ethereum network protocol handler @@ -140,6 +143,11 @@ impl SyncProvider for EthSync { fn status(&self) -> SyncStatus { self.sync.read().unwrap().status() } + + /// Note that a user has submitted a new transaction. + fn new_transaction(&self, raw_transaction: Bytes) { + self.sync.write().unwrap().new_transaction(raw_transaction); + } } impl NetworkProtocolHandler for EthSync {