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

Propagate transaction queue

Browse Source
This commit is contained in:
arkpar
2016-04-06 23:03:07 +02:00
parent 5685fde606
commit 3438cda432
7 changed files with 29 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
sync/src/chain.rs
View File

@@ -217,10 +217,6 @@ 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>;
@@ -247,7 +243,6 @@ 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![],
}
}
@@ -950,11 +945,6 @@ impl ChainSync {
}
}
/// 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
@@ -1296,11 +1286,16 @@ impl ChainSync {
return 0;
}
let mut packet = RlpStream::new_list(self.transactions_to_send.len());
for tx in &self.transactions_to_send {
packet.append_raw(tx, 1);
let mut transactions = self.miner.pending_transactions();
if transactions.is_empty() {
return 0;
}
let mut packet = RlpStream::new_list(transactions.len());
let tx_count = transactions.len();
for tx in transactions.drain(..) {
packet.append(&tx);
}
self.transactions_to_send.clear();
let rlp = packet.out();
let lucky_peers = {
@@ -1319,13 +1314,12 @@ impl ChainSync {
for peer_id in lucky_peers {
self.send_packet(io, peer_id, TRANSACTIONS_PACKET, rlp.clone());
}
trace!(target: "sync", "Sent {} transactions to {} peers.", tx_count, sent);
sent
}
fn propagate_latest_blocks(&mut self, io: &mut SyncIo) {
if !self.transactions_to_send.is_empty() {
self.propagate_new_transactions(io);
}
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, Bytes, ONE_U256};
use util::{U256, ONE_U256};
use ethcore::client::Client;
use ethcore::service::SyncMessage;
use ethminer::Miner;
@@ -101,9 +101,6 @@ 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
@@ -143,11 +140,6 @@ 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 {