use std::fmt;
use std::sync::Arc;
use std::collections::{BTreeMap, HashMap};
use std::collections::hash_map::Entry;
use common_types::transaction::{self, Condition, PendingTransaction, SignedTransaction};
use vapory_types::{H256, U256, Address};
use fastmap::H256FastMap;
use futures::sync::mpsc;
use miner::pool::TxStatus;
#[derive(Debug, Clone, PartialEq, Eq)]
enum CurrentNonce {
Assumed(U256),
Known(U256),
}
impl CurrentNonce {
fn is_assumed(&self) -> bool {
match *self {
CurrentNonce::Assumed(_) => true,
CurrentNonce::Known(_) => false,
}
}
fn is_known(&self) -> bool {
!self.is_assumed()
}
fn value(&self) -> &U256 {
match *self {
CurrentNonce::Assumed(ref val) => val,
CurrentNonce::Known(ref val) => val,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct TransactionInfo {
hash: H256,
nonce: U256,
condition: Option<Condition>,
}
impl<'a> From<&'a PendingTransaction> for TransactionInfo {
fn from(tx: &'a PendingTransaction) -> Self {
TransactionInfo {
hash: tx.hash(),
nonce: tx.nonce,
condition: tx.condition.clone(),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct AccountTransactions {
cur_nonce: CurrentNonce,
current: Vec<TransactionInfo>,
future: BTreeMap<U256, TransactionInfo>,
}
impl AccountTransactions {
fn is_empty(&self) -> bool {
self.current.is_empty() && self.future.is_empty()
}
fn next_nonce(&self) -> U256 {
self.current.last().map(|last| last.nonce.saturating_add(1.into()))
.unwrap_or_else(|| *self.cur_nonce.value())
}
fn adjust_future(&mut self) -> Vec<H256> {
let mut promoted = Vec::new();
let mut next_nonce = self.next_nonce();
while let Some(tx) = self.future.remove(&next_nonce) {
promoted.push(tx.hash);
self.current.push(tx);
next_nonce = next_nonce.saturating_add(1.into());
}
promoted
}
}
pub enum ImportDestination {
Current,
Future,
}
#[derive(Default)]
pub struct TransactionQueue {
by_account: HashMap<Address, AccountTransactions>,
by_hash: H256FastMap<PendingTransaction>,
pending_listeners: Vec<mpsc::UnboundedSender<Arc<Vec<H256>>>>,
full_listeners: Vec<mpsc::UnboundedSender<Arc<Vec<(H256, TxStatus)>>>>,
}
impl fmt::Debug for TransactionQueue {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.debug_struct("TransactionQueue")
.field("by_account", &self.by_account)
.field("by_hash", &self.by_hash)
.field("pending_listeners", &self.pending_listeners.len())
.field("full_listeners", &self.pending_listeners.len())
.finish()
}
}
impl TransactionQueue {
pub fn import(&mut self, tx: PendingTransaction) -> Result<ImportDestination, transaction::Error> {
let sender = tx.sender();
let hash = tx.hash();
let nonce = tx.nonce;
let tx_info = TransactionInfo::from(&tx);
if self.by_hash.contains_key(&hash) { return Err(transaction::Error::AlreadyImported) }
let (res, promoted) = match self.by_account.entry(sender) {
Entry::Vacant(entry) => {
entry.insert(AccountTransactions {
cur_nonce: CurrentNonce::Assumed(nonce),
current: vec![tx_info],
future: BTreeMap::new(),
});
(ImportDestination::Current, vec![hash])
}
Entry::Occupied(mut entry) => {
let acct_txs = entry.get_mut();
if nonce < *acct_txs.cur_nonce.value() {
if acct_txs.cur_nonce.is_known() {
return Err(transaction::Error::Old)
}
acct_txs.cur_nonce = CurrentNonce::Assumed(nonce);
}
match acct_txs.current.binary_search_by(|x| x.nonce.cmp(&nonce)) {
Ok(idx) => {
trace!(target: "txqueue", "Replacing existing transaction from {} with nonce {}",
sender, nonce);
let old = ::std::mem::replace(&mut acct_txs.current[idx], tx_info);
self.by_hash.remove(&old.hash);
(ImportDestination::Current, vec![hash])
}
Err(idx) => {
let cur_len = acct_txs.current.len();
let incr_nonce = nonce + 1;
assert!(idx == 0 || idx == cur_len);
if idx == 0 && acct_txs.current.first().map_or(false, |f| f.nonce != incr_nonce) {
let old_cur = ::std::mem::replace(&mut acct_txs.current, vec![tx_info]);
trace!(target: "txqueue", "Moving {} transactions with nonce > {} to future",
old_cur.len(), incr_nonce);
for future in old_cur {
let future_nonce = future.nonce;
acct_txs.future.insert(future_nonce, future);
}
(ImportDestination::Current, vec![hash])
} else if idx == cur_len && acct_txs.current.last().map_or(false, |f| f.nonce + 1 != nonce) {
trace!(target: "txqueue", "Queued future transaction for {}, nonce={}", sender, nonce);
let future_nonce = nonce;
acct_txs.future.insert(future_nonce, tx_info);
(ImportDestination::Future, vec![])
} else {
trace!(target: "txqueue", "Queued current transaction for {}, nonce={}", sender, nonce);
acct_txs.current.insert(idx, tx_info);
let mut promoted = acct_txs.adjust_future();
promoted.insert(0, hash);
(ImportDestination::Current, promoted)
}
}
}
}
};
self.by_hash.insert(hash, tx);
self.notify(&promoted, TxStatus::Added);
Ok(res)
}
pub fn transaction(&self, hash: &H256) -> Option<SignedTransaction> {
self.by_hash.get(hash).map(|tx| (&**tx).clone())
}
pub fn next_nonce(&self, address: &Address) -> Option<U256> {
self.by_account.get(address).map(AccountTransactions::next_nonce)
}
pub fn ready_transactions(&self, best_block_number: u64, best_block_timestamp: u64) -> Vec<PendingTransaction> {
self.by_account.values()
.flat_map(|acct_txs| {
acct_txs.current.iter().take_while(|tx| match tx.condition {
None => true,
Some(Condition::Number(blk_num)) => blk_num <= best_block_number,
Some(Condition::Timestamp(time)) => time <= best_block_timestamp,
}).map(|info| info.hash)
})
.filter_map(|hash| match self.by_hash.get(&hash) {
Some(tx) => Some(tx.clone()),
None => {
warn!(target: "txqueue", "Inconsistency detected between `by_hash` and `by_account`: {} not stored.",
hash);
None
}
})
.collect()
}
pub fn future_transactions(&self, best_block_number: u64, best_block_timestamp: u64) -> Vec<PendingTransaction> {
self.by_account.values()
.flat_map(|acct_txs| {
acct_txs.current.iter().skip_while(|tx| match tx.condition {
None => true,
Some(Condition::Number(blk_num)) => blk_num <= best_block_number,
Some(Condition::Timestamp(time)) => time <= best_block_timestamp,
}).chain(acct_txs.future.values()).map(|info| info.hash)
})
.filter_map(|hash| match self.by_hash.get(&hash) {
Some(tx) => Some(tx.clone()),
None => {
warn!(target: "txqueue", "Inconsistency detected between `by_hash` and `by_account`: {} not stored.",
hash);
None
}
})
.collect()
}
pub fn queued_senders(&self) -> Vec<Address> {
self.by_account.keys().cloned().collect()
}
pub fn cull(&mut self, address: Address, cur_nonce: U256) {
let mut removed_hashes = vec![];
if let Entry::Occupied(mut entry) = self.by_account.entry(address) {
{
let acct_txs = entry.get_mut();
acct_txs.cur_nonce = CurrentNonce::Known(cur_nonce);
let old_future: Vec<_> = acct_txs.future.keys().take_while(|&&k| k < cur_nonce).cloned().collect();
for old in old_future {
let hash = acct_txs.future.remove(&old)
.expect("key extracted from keys iterator; known to exist; qed")
.hash;
removed_hashes.push(hash);
}
let valid_pos = acct_txs.current.iter().position(|tx| tx.nonce >= cur_nonce);
match valid_pos {
None =>
removed_hashes.extend(acct_txs.current.drain(..).map(|tx| tx.hash)),
Some(valid) =>
removed_hashes.extend(acct_txs.current.drain(..valid).map(|tx| tx.hash)),
}
acct_txs.adjust_future();
}
if entry.get_mut().is_empty() {
trace!(target: "txqueue", "No more queued transactions for {} after nonce {}",
address, cur_nonce);
entry.remove();
}
}
trace!(target: "txqueue", "Culled {} old transactions from sender {} (nonce={})",
removed_hashes.len(), address, cur_nonce);
self.notify(&removed_hashes, TxStatus::Culled);
for hash in removed_hashes {
self.by_hash.remove(&hash);
}
}
pub fn get(&self, hash: &H256) -> Option<&PendingTransaction> {
self.by_hash.get(&hash)
}
pub fn pending_transactions_receiver(&mut self) -> mpsc::UnboundedReceiver<Arc<Vec<H256>>> {
let (sender, receiver) = mpsc::unbounded();
self.pending_listeners.push(sender);
receiver
}
pub fn full_transactions_receiver(&mut self) -> mpsc::UnboundedReceiver<Arc<Vec<(H256, TxStatus)>>> {
let (sender, receiver) = mpsc::unbounded();
self.full_listeners.push(sender);
receiver
}
fn notify(&mut self, hashes: &[H256], status: TxStatus) {
if status == TxStatus::Added {
let to_pending_send: Arc<Vec<H256>> = Arc::new(
hashes
.into_iter()
.map(|hash| hash.clone())
.collect()
);
self.pending_listeners.retain(|listener| listener.unbounded_send(to_pending_send.clone()).is_ok());
}
let to_full_send: Arc<Vec<(H256, TxStatus)>> = Arc::new(
hashes
.into_iter()
.map(|hash| (hash.clone(), status))
.collect()
);
self.full_listeners.retain(|listener| listener.unbounded_send(to_full_send.clone()).is_ok());
}
}
#[cfg(test)]
mod tests {
use super::TransactionQueue;
use vapory_types::Address;
use common_types::transaction::{Transaction, PendingTransaction, Condition};
#[test]
fn queued_senders() {
let sender = Address::zero();
let mut txq = TransactionQueue::default();
let tx = Transaction::default().fake_sign(sender);
txq.import(tx.into()).unwrap();
assert_eq!(txq.queued_senders(), vec![sender]);
txq.cull(sender, 1.into());
assert_eq!(txq.queued_senders(), vec![]);
assert!(txq.by_hash.is_empty());
}
#[test]
fn next_nonce() {
let sender = Address::zero();
let mut txq = TransactionQueue::default();
for i in (0..5).chain(10..15) {
let mut tx = Transaction::default();
tx.nonce = i.into();
let tx = tx.fake_sign(sender);
txq.import(tx.into()).unwrap();
}
assert_eq!(txq.ready_transactions(0, 0).len(), 5);
assert_eq!(txq.next_nonce(&sender).unwrap(), 5.into());
txq.cull(sender, 8.into());
assert_eq!(txq.ready_transactions(0, 0).len(), 0);
assert_eq!(txq.next_nonce(&sender).unwrap(), 8.into());
txq.cull(sender, 10.into());
assert_eq!(txq.ready_transactions(0, 0).len(), 5);
assert_eq!(txq.next_nonce(&sender).unwrap(), 15.into());
}
#[test]
fn current_to_future() {
let sender = Address::zero();
let mut txq = TransactionQueue::default();
for i in 5..10 {
let mut tx = Transaction::default();
tx.nonce = i.into();
let tx = tx.fake_sign(sender);
txq.import(tx.into()).unwrap();
}
assert_eq!(txq.ready_transactions(0, 0).len(), 5);
assert_eq!(txq.next_nonce(&sender).unwrap(), 10.into());
for i in 0..3 {
let mut tx = Transaction::default();
tx.nonce = i.into();
let tx = tx.fake_sign(sender);
txq.import(tx.into()).unwrap();
}
assert_eq!(txq.ready_transactions(0, 0).len(), 3);
assert_eq!(txq.next_nonce(&sender).unwrap(), 3.into());
for i in 3..5 {
let mut tx = Transaction::default();
tx.nonce = i.into();
let tx = tx.fake_sign(sender);
txq.import(tx.into()).unwrap();
}
assert_eq!(txq.ready_transactions(0, 0).len(), 10);
assert_eq!(txq.next_nonce(&sender).unwrap(), 10.into());
}
#[test]
fn conditional() {
let mut txq = TransactionQueue::default();
let sender = Address::zero();
for i in 0..5 {
let mut tx = Transaction::default();
tx.nonce = i.into();
let tx = tx.fake_sign(sender);
txq.import(match i {
3 => PendingTransaction::new(tx, Some(Condition::Number(100))),
4 => PendingTransaction::new(tx, Some(Condition::Timestamp(1234))),
_ => tx.into(),
}).unwrap();
}
assert_eq!(txq.ready_transactions(0, 0).len(), 3);
assert_eq!(txq.ready_transactions(0, 1234).len(), 3);
assert_eq!(txq.ready_transactions(100, 0).len(), 4);
assert_eq!(txq.ready_transactions(100, 1234).len(), 5);
}
#[test]
fn cull_from_future() {
let sender = Address::zero();
let mut txq = TransactionQueue::default();
for i in (0..1).chain(3..10) {
let mut tx = Transaction::default();
tx.nonce = i.into();
let tx = tx.fake_sign(sender);
txq.import(tx.into()).unwrap();
}
txq.cull(sender, 6.into());
assert_eq!(txq.ready_transactions(0, 0).len(), 4);
assert_eq!(txq.next_nonce(&sender).unwrap(), 10.into());
}
#[test]
fn import_old() {
let sender = Address::zero();
let mut txq = TransactionQueue::default();
let mut tx_a = Transaction::default();
tx_a.nonce = 3.into();
let mut tx_b = Transaction::default();
tx_b.nonce = 2.into();
txq.import(tx_a.fake_sign(sender).into()).unwrap();
txq.cull(sender, 3.into());
assert!(txq.import(tx_b.fake_sign(sender).into()).is_err())
}
#[test]
fn replace_is_removed() {
let sender = Address::zero();
let mut txq = TransactionQueue::default();
let tx_b: PendingTransaction = Transaction::default().fake_sign(sender).into();
let tx_a: PendingTransaction = {
let mut tx_a = Transaction::default();
tx_a.gas_price = tx_b.gas_price + 1;
tx_a.fake_sign(sender).into()
};
let hash = tx_a.hash();
txq.import(tx_a).unwrap();
txq.import(tx_b).unwrap();
assert!(txq.transaction(&hash).is_none());
}
#[test]
fn future_transactions() {
let sender = Address::zero();
let mut txq = TransactionQueue::default();
for i in (0..1).chain(3..10) {
let mut tx = Transaction::default();
tx.nonce = i.into();
let tx = tx.fake_sign(sender);
txq.import(tx.into()).unwrap();
}
assert_eq!(txq.future_transactions(0, 0).len(), 7);
assert_eq!(txq.next_nonce(&sender).unwrap(), 1.into());
}
}