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

Publish locally-made transactions to peers.

This commit is contained in:
Gav Wood 2016-03-28 18:11:00 +02:00
parent f60d74b231
commit 3023d76a46
4 changed files with 100 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

84
rpc/src/v1/impls/eth.rs
View File

@ -178,6 +178,29 @@ impl<C, S, A, M, EM> EthClient<C, S, A, M, EM>
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<u8>) -> Result<Value, Error> {
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::<Result<Vec<_>, _>>() {
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<C, S, A, M, EM> Eth for EthClient<C, S, A, M, EM>
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::<Result<Vec<_>, _>>() {
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<C, S, A, M, EM> Eth for EthClient<C, S, A, M, EM>
fn send_raw_transaction(&self, params: Params) -> Result<Value, Error> {
from_params::<(Bytes, )>(params)
.and_then(|(raw_transaction, )| {
let decoded: Result<SignedTransaction, _> = 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::<Result<Vec<_>, _>>() {
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()),
}
})
}

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

@ -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) {
}
}

48
sync/src/chain.rs
View File

@ -217,6 +217,10 @@ pub struct ChainSync {
network_id: U256,
/// Miner
miner: Arc<Miner>,
/// Transactions to propagate
// TODO: reconsider where this is in the codebase - seems a little dodgy to have here.
transactions_to_send: Vec<Bytes>,
}
type RlpResponseResult = Result<Option<(PacketId, RlpStream)>, 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::<u32>() < fraction { Some(p.clone()) } else { None })
.collect::<Vec<_>>();
// taking at max of MAX_PEERS_PROPAGATION
lucky_peers.iter().map(|&id| id.clone()).take(min(lucky_peers.len(), MAX_PEERS_PROPAGATION)).collect::<Vec<PeerId>>()
};
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);

10
sync/src/lib.rs
View File

@ -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<SyncMessage> for EthSync {