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

Removing a transaction from queue now removes all from this sender with lower nonces. (#950)

Browse Source 
* Changing  to wipe-out all transactions from particular sender lower then given nonce.

* Changing given nonce to be client_nonce

* Fixing test_client to support proper nonces when transactions are added to blockchain

* Fixing logic for transactions from new blocks in chain
This commit is contained in:
Tomasz Drwięga 2016-04-15 07:38:23 +02:00 committed by Gav Wood
parent c4c17aa1da
commit 00372cf747
4 changed files with 116 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -48,6 +48,8 @@ pub struct TestBlockChainClient {
pub difficulty: RwLock<U256>, pub difficulty: RwLock<U256>,
/// Balances. /// Balances.
pub balances: RwLock<HashMap<Address, U256>>, pub balances: RwLock<HashMap<Address, U256>>,
/// Nonces.
pub nonces: RwLock<HashMap<Address, U256>>,
/// Storage. /// Storage.
pub storage: RwLock<HashMap<(Address, H256), H256>>, pub storage: RwLock<HashMap<(Address, H256), H256>>,
/// Code. /// Code.
@ -90,6 +92,7 @@ impl TestBlockChainClient {
last_hash: RwLock::new(H256::new()), last_hash: RwLock::new(H256::new()),
difficulty: RwLock::new(From::from(0)), difficulty: RwLock::new(From::from(0)),
balances: RwLock::new(HashMap::new()), balances: RwLock::new(HashMap::new()),
nonces: RwLock::new(HashMap::new()),
storage: RwLock::new(HashMap::new()), storage: RwLock::new(HashMap::new()),
code: RwLock::new(HashMap::new()), code: RwLock::new(HashMap::new()),
execution_result: RwLock::new(None), execution_result: RwLock::new(None),
@ -116,6 +119,11 @@ impl TestBlockChainClient {
self.balances.write().unwrap().insert(address, balance); self.balances.write().unwrap().insert(address, balance);
} }
/// Set nonce of account `address` to `nonce`.
pub fn set_nonce(&self, address: Address, nonce: U256) {
self.nonces.write().unwrap().insert(address, nonce);
}
/// Set `code` at `address`. /// Set `code` at `address`.
pub fn set_code(&self, address: Address, code: Bytes) { pub fn set_code(&self, address: Address, code: Bytes) {
self.code.write().unwrap().insert(address, code); self.code.write().unwrap().insert(address, code);
@ -157,6 +165,8 @@ impl TestBlockChainClient {
EachBlockWith::Transaction | EachBlockWith::UncleAndTransaction => { EachBlockWith::Transaction | EachBlockWith::UncleAndTransaction => {
let mut txs = RlpStream::new_list(1); let mut txs = RlpStream::new_list(1);
let keypair = KeyPair::create().unwrap(); let keypair = KeyPair::create().unwrap();
// Update nonces value
self.nonces.write().unwrap().insert(keypair.address(), U256::one());
let tx = Transaction { let tx = Transaction {
action: Action::Create, action: Action::Create,
value: U256::from(100), value: U256::from(100),
@ -222,8 +232,8 @@ impl BlockChainClient for TestBlockChainClient {
unimplemented!(); unimplemented!();
} }
fn nonce(&self, _address: &Address) -> U256 { fn nonce(&self, address: &Address) -> U256 {
U256::zero() self.nonces.read().unwrap().get(address).cloned().unwrap_or_else(U256::zero)
} }
fn code(&self, address: &Address) -> Option<Bytes> { fn code(&self, address: &Address) -> Option<Bytes> {

45
miner/src/miner.rs
View File

@ -133,13 +133,13 @@ impl Miner {
} }
}; };
let mut queue = self.transaction_queue.lock().unwrap(); let mut queue = self.transaction_queue.lock().unwrap();
queue.remove_all( let fetch_account = |a: &Address| AccountDetails {
&invalid_transactions.into_iter().collect::<Vec<H256>>(),
|a: &Address| AccountDetails {
nonce: chain.nonce(a), nonce: chain.nonce(a),
balance: chain.balance(a), balance: chain.balance(a),
};
for hash in invalid_transactions.into_iter() {
queue.remove_invalid(&hash, &fetch_account);
} }
);
if let Some(block) = b { if let Some(block) = b {
if sealing_work.peek_last_ref().map_or(true, |pb| pb.block().fields().header.hash() != block.block().fields().header.hash()) { 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()); trace!(target: "miner", "Pushing a new, refreshed or borrowed pending {}...", block.block().fields().header.hash());
@ -299,7 +299,7 @@ impl MinerService for Miner {
} }
} }
fn chain_new_blocks(&self, chain: &BlockChainClient, imported: &[H256], invalid: &[H256], enacted: &[H256], retracted: &[H256]) { fn chain_new_blocks(&self, chain: &BlockChainClient, _imported: &[H256], _invalid: &[H256], enacted: &[H256], retracted: &[H256]) {
fn fetch_transactions(chain: &BlockChainClient, hash: &H256) -> Vec<SignedTransaction> { fn fetch_transactions(chain: &BlockChainClient, hash: &H256) -> Vec<SignedTransaction> {
let block = chain let block = chain
.block(BlockId::Hash(*hash)) .block(BlockId::Hash(*hash))
@ -309,6 +309,11 @@ impl MinerService for Miner {
block.transactions() block.transactions()
} }
// 1. We ignore blocks that were `imported` (because it means that they are not in canon-chain, and transactions
// should be still available in the queue.
// 2. We ignore blocks that are `invalid` because it doesn't have any meaning in terms of the transactions that
// are in those blocks
// First update gas limit in transaction queue // First update gas limit in transaction queue
self.update_gas_limit(chain); self.update_gas_limit(chain);
@ -330,29 +335,23 @@ impl MinerService for Miner {
}); });
} }
// ...and after that remove old ones // ...and at the end remove old ones
{ {
let in_chain = { let in_chain = enacted
let mut in_chain = HashSet::new();
in_chain.extend(imported);
in_chain.extend(enacted);
in_chain.extend(invalid);
in_chain
.into_iter()
.collect::<Vec<H256>>()
};
let in_chain = in_chain
.par_iter() .par_iter()
.map(|h: &H256| fetch_transactions(chain, h)); .map(|h: &H256| fetch_transactions(chain, h));
in_chain.for_each(|txs| { in_chain.for_each(|mut txs| {
let hashes = txs.iter().map(|tx| tx.hash()).collect::<Vec<H256>>();
let mut transaction_queue = self.transaction_queue.lock().unwrap(); let mut transaction_queue = self.transaction_queue.lock().unwrap();
transaction_queue.remove_all(&hashes, |a| AccountDetails {
nonce: chain.nonce(a), let to_remove = txs.drain(..)
balance: chain.balance(a) .map(|tx| {
}); tx.sender().expect("Transaction is in block, so sender has to be defined.")
})
.collect::<HashSet<Address>>();
for sender in to_remove.into_iter() {
transaction_queue.remove_all(sender, chain.nonce(&sender));
}
}); });
} }

88
miner/src/transaction_queue.rs
View File

@ -65,8 +65,8 @@
//! assert_eq!(top[1], st2); //! assert_eq!(top[1], st2);
//! //!
//! // And when transaction is removed (but nonce haven't changed) //! // And when transaction is removed (but nonce haven't changed)
//! // it will move invalid transactions to future //! // it will move subsequent transactions to future
//! txq.remove(&st1.hash(), &default_nonce); //! txq.remove_invalid(&st1.hash(), &default_nonce);
//! assert_eq!(txq.status().pending, 0); //! assert_eq!(txq.status().pending, 0);
//! assert_eq!(txq.status().future, 1); //! assert_eq!(txq.status().future, 1);
//! assert_eq!(txq.top_transactions().len(), 0); //! assert_eq!(txq.top_transactions().len(), 0);
@ -76,11 +76,13 @@
//! # Maintaing valid state //! # Maintaing valid state
//! //!
//! 1. Whenever transaction is imported to queue (to queue) all other transactions from this sender are revalidated in current. It means that they are moved to future and back again (height recalculation & gap filling). //! 1. Whenever transaction is imported to queue (to queue) all other transactions from this sender are revalidated in current. It means that they are moved to future and back again (height recalculation & gap filling).
//! 2. Whenever transaction is removed: //! 2. Whenever invalid transaction is removed:
//! - When it's removed from `future` - all `future` transactions heights are recalculated and then //! - When it's removed from `future` - all `future` transactions heights are recalculated and then
//! we check if the transactions should go to `current` (comparing state nonce) //! we check if the transactions should go to `current` (comparing state nonce)
//! - When it's removed from `current` - all transactions from this sender (`current` & `future`) are recalculated. //! - When it's removed from `current` - all transactions from this sender (`current` & `future`) are recalculated.
//! //! 3. `remove_all` is used to inform the queue about client (state) nonce changes.
//! - It removes all transactions (either from `current` or `future`) with nonce < client nonce
//! - It moves matching `future` transactions to `current`
use std::default::Default; use std::default::Default;
use std::cmp::{Ordering}; use std::cmp::{Ordering};
@ -398,22 +400,28 @@ impl TransactionQueue {
self.import_tx(vtx, client_account.nonce).map_err(Error::Transaction) self.import_tx(vtx, client_account.nonce).map_err(Error::Transaction)
} }
/// Removes all transactions identified by hashes given in slice /// Removes all transactions from particular sender up to (excluding) given client (state) nonce.
/// /// Client (State) Nonce = next valid nonce for this sender.
/// If gap is introduced marks subsequent transactions as future pub fn remove_all(&mut self, sender: Address, client_nonce: U256) {
pub fn remove_all<T>(&mut self, transaction_hashes: &[H256], fetch_account: T) // We will either move transaction to future or remove it completely
where T: Fn(&Address) -> AccountDetails { // so there will be no transactions from this sender in current
for hash in transaction_hashes { self.last_nonces.remove(&sender);
self.remove(&hash, &fetch_account); // First update height of transactions in future to avoid collisions
} self.update_future(&sender, client_nonce);
// This should move all current transactions to future and remove old transactions
self.move_all_to_future(&sender, client_nonce);
// And now lets check if there is some batch of transactions in future
// that should be placed in current. It should also update last_nonces.
self.move_matching_future_to_current(sender, client_nonce, client_nonce);
} }
/// Removes transaction identified by hashes from queue. /// Removes invalid transaction identified by hash from queue.
/// Assumption is that this transaction nonce is not related to client nonce,
/// so transactions left in queue are processed according to client nonce.
/// ///
/// If gap is introduced marks subsequent transactions as future /// If gap is introduced marks subsequent transactions as future
pub fn remove<T>(&mut self, transaction_hash: &H256, fetch_account: &T) pub fn remove_invalid<T>(&mut self, transaction_hash: &H256, fetch_account: &T)
where T: Fn(&Address) -> AccountDetails { where T: Fn(&Address) -> AccountDetails {
let transaction = self.by_hash.remove(transaction_hash); let transaction = self.by_hash.remove(transaction_hash);
if transaction.is_none() { if transaction.is_none() {
// We don't know this transaction // We don't know this transaction
@ -425,7 +433,6 @@ impl TransactionQueue {
let nonce = transaction.nonce(); let nonce = transaction.nonce();
let current_nonce = fetch_account(&sender).nonce; let current_nonce = fetch_account(&sender).nonce;
// Remove from future // Remove from future
let order = self.future.drop(&sender, &nonce); let order = self.future.drop(&sender, &nonce);
if order.is_some() { if order.is_some() {
@ -465,7 +472,7 @@ impl TransactionQueue {
if k >= current_nonce { if k >= current_nonce {
self.future.insert(*sender, k, order.update_height(k, current_nonce)); self.future.insert(*sender, k, order.update_height(k, current_nonce));
} else { } else {
trace!(target: "miner", "Dropping old transaction: {:?} (nonce: {} < {})", order.hash, k, current_nonce); trace!(target: "miner", "Removing old transaction: {:?} (nonce: {} < {})", order.hash, k, current_nonce);
// Remove the transaction completely // Remove the transaction completely
self.by_hash.remove(&order.hash); self.by_hash.remove(&order.hash);
} }
@ -486,7 +493,7 @@ impl TransactionQueue {
if k >= current_nonce { if k >= current_nonce {
self.future.insert(*sender, k, order.update_height(k, current_nonce)); self.future.insert(*sender, k, order.update_height(k, current_nonce));
} else { } else {
trace!(target: "miner", "Dropping old transaction: {:?} (nonce: {} < {})", order.hash, k, current_nonce); trace!(target: "miner", "Removing old transaction: {:?} (nonce: {} < {})", order.hash, k, current_nonce);
self.by_hash.remove(&order.hash); self.by_hash.remove(&order.hash);
} }
} }
@ -665,9 +672,14 @@ mod test {
new_unsigned_tx(U256::from(123)).sign(&keypair.secret()) new_unsigned_tx(U256::from(123)).sign(&keypair.secret())
} }
fn default_nonce_val() -> U256 {
U256::from(123)
}
fn default_nonce(_address: &Address) -> AccountDetails { fn default_nonce(_address: &Address) -> AccountDetails {
AccountDetails { AccountDetails {
nonce: U256::from(123), nonce: default_nonce_val(),
balance: !U256::zero() balance: !U256::zero()
} }
} }
@ -965,8 +977,7 @@ mod test {
// given // given
let prev_nonce = |a: &Address| AccountDetails{ nonce: default_nonce(a).nonce - U256::one(), balance: let prev_nonce = |a: &Address| AccountDetails{ nonce: default_nonce(a).nonce - U256::one(), balance:
!U256::zero() }; !U256::zero() };
let next2_nonce = |a: &Address| AccountDetails{ nonce: default_nonce(a).nonce + U256::from(2), balance: let next2_nonce = default_nonce_val() + U256::from(3);
!U256::zero() };
let mut txq = TransactionQueue::new(); let mut txq = TransactionQueue::new();
@ -976,7 +987,7 @@ mod test {
assert_eq!(txq.status().future, 2); assert_eq!(txq.status().future, 2);
// when // when
txq.remove(&tx.hash(), &next2_nonce); txq.remove_all(tx.sender().unwrap(), next2_nonce);
// should remove both transactions since they are not valid // should remove both transactions since they are not valid
// then // then
@ -1019,8 +1030,8 @@ mod test {
assert_eq!(txq2.status().future, 1); assert_eq!(txq2.status().future, 1);
// when // when
txq2.remove(&tx.hash(), &default_nonce); txq2.remove_all(tx.sender().unwrap(), tx.nonce + U256::one());
txq2.remove(&tx2.hash(), &default_nonce); txq2.remove_all(tx2.sender().unwrap(), tx2.nonce + U256::one());
// then // then
@ -1042,7 +1053,7 @@ mod test {
assert_eq!(txq.status().pending, 3); assert_eq!(txq.status().pending, 3);
// when // when
txq.remove(&tx.hash(), &default_nonce); txq.remove_invalid(&tx.hash(), &default_nonce);
// then // then
let stats = txq.status(); let stats = txq.status();
@ -1152,7 +1163,7 @@ mod test {
assert_eq!(txq.status().pending, 2); assert_eq!(txq.status().pending, 2);
// when // when
txq.remove(&tx1.hash(), &default_nonce); txq.remove_invalid(&tx1.hash(), &default_nonce);
assert_eq!(txq.status().pending, 0); assert_eq!(txq.status().pending, 0);
assert_eq!(txq.status().future, 1); assert_eq!(txq.status().future, 1);
txq.add(tx1.clone(), &default_nonce).unwrap(); txq.add(tx1.clone(), &default_nonce).unwrap();
@ -1166,8 +1177,6 @@ mod test {
#[test] #[test]
fn should_not_move_to_future_if_state_nonce_is_higher() { fn should_not_move_to_future_if_state_nonce_is_higher() {
// given // given
let next_nonce = |a: &Address| AccountDetails { nonce: default_nonce(a).nonce + U256::one(), balance:
!U256::zero() };
let mut txq = TransactionQueue::new(); let mut txq = TransactionQueue::new();
let (tx, tx2) = new_txs(U256::from(1)); let (tx, tx2) = new_txs(U256::from(1));
let tx3 = new_tx(); let tx3 = new_tx();
@ -1178,7 +1187,8 @@ mod test {
assert_eq!(txq.status().pending, 3); assert_eq!(txq.status().pending, 3);
// when // when
txq.remove(&tx.hash(), &next_nonce); let sender = tx.sender().unwrap();
txq.remove_all(sender, default_nonce_val() + U256::one());
// then // then
let stats = txq.status(); let stats = txq.status();
@ -1254,7 +1264,7 @@ mod test {
assert_eq!(txq.status().future, 2); assert_eq!(txq.status().future, 2);
// when // when
txq.remove(&tx1.hash(), &next_nonce); txq.remove_invalid(&tx1.hash(), &next_nonce);
// then // then
let stats = txq.status(); let stats = txq.status();
@ -1286,4 +1296,22 @@ mod test {
// then // then
assert_eq!(txq.last_nonce(&from), Some(nonce)); assert_eq!(txq.last_nonce(&from), Some(nonce));
} }
#[test]
fn should_remove_old_transaction_even_if_newer_transaction_was_not_known() {
// given
let mut txq = TransactionQueue::new();
let (tx1, tx2) = new_txs(U256::one());
let (nonce1, nonce2) = (tx1.nonce, tx2.nonce);
let details1 = |_a: &Address| AccountDetails { nonce: nonce1, balance: !U256::zero() };
// Insert first transaction
txq.add(tx1, &details1).unwrap();
// when
txq.remove_all(tx2.sender().unwrap(), nonce2 + U256::one());
// then
assert!(txq.top_transactions().is_empty());
}
} }

23
sync/src/chain.rs
View File

@ -1694,21 +1694,34 @@ mod tests {
let good_blocks = vec![client.block_hash_delta_minus(2)]; let good_blocks = vec![client.block_hash_delta_minus(2)];
let retracted_blocks = vec![client.block_hash_delta_minus(1)]; let retracted_blocks = vec![client.block_hash_delta_minus(1)];
// Add some balance to clients // Add some balance to clients and reset nonces
for h in &[good_blocks[0], retracted_blocks[0]] { for h in &[good_blocks[0], retracted_blocks[0]] {
let block = client.block(BlockId::Hash(*h)).unwrap(); let block = client.block(BlockId::Hash(*h)).unwrap();
let view = BlockView::new(&block); let view = BlockView::new(&block);
client.set_balance(view.transactions()[0].sender().unwrap(), U256::from(1_000_000_000)); client.set_balance(view.transactions()[0].sender().unwrap(), U256::from(1_000_000_000));
client.set_nonce(view.transactions()[0].sender().unwrap(), U256::from(0));
} }
let mut queue = VecDeque::new();
let mut io = TestIo::new(&mut client, &mut queue, None);
// when // when
{
let mut queue = VecDeque::new();
let mut io = TestIo::new(&mut client, &mut queue, None);
sync.chain_new_blocks(&mut io, &[], &[], &[], &good_blocks); sync.chain_new_blocks(&mut io, &[], &[], &[], &good_blocks);
assert_eq!(sync.miner.status().transactions_in_future_queue, 0); assert_eq!(sync.miner.status().transactions_in_future_queue, 0);
assert_eq!(sync.miner.status().transactions_in_pending_queue, 1); assert_eq!(sync.miner.status().transactions_in_pending_queue, 1);
sync.chain_new_blocks(&mut io, &good_blocks, &[], &[], &retracted_blocks); }
// We need to update nonce status (because we say that the block has been imported)
for h in &[good_blocks[0]] {
let block = client.block(BlockId::Hash(*h)).unwrap();
let view = BlockView::new(&block);
client.set_nonce(view.transactions()[0].sender().unwrap(), U256::from(1));
}
{
let mut queue = VecDeque::new();
let mut io = TestIo::new(&mut client, &mut queue, None);
sync.chain_new_blocks(&mut io, &[], &[], &good_blocks, &retracted_blocks);
}
// then // then
let status = sync.miner.status(); let status = sync.miner.status();
@ -1735,7 +1748,7 @@ mod tests {
sync.chain_new_blocks(&mut io, &[], &[], &[], &good_blocks); sync.chain_new_blocks(&mut io, &[], &[], &[], &good_blocks);
assert_eq!(sync.miner.status().transactions_in_future_queue, 0); assert_eq!(sync.miner.status().transactions_in_future_queue, 0);
assert_eq!(sync.miner.status().transactions_in_pending_queue, 0); assert_eq!(sync.miner.status().transactions_in_pending_queue, 0);
sync.chain_new_blocks(&mut io, &good_blocks, &[], &[], &retracted_blocks); sync.chain_new_blocks(&mut io, &[], &[], &good_blocks, &retracted_blocks);
// then // then
let status = sync.miner.status(); let status = sync.miner.status();