Transactions dependency graph

Cargo.toml

@@ -26,7 +26,6 @@ members = [
 	"core/client",
 	"core/client/db",
 	"core/executor",
-	"core/transaction-pool",
 	"core/keyring",
 	"core/misbehavior-check",
 	"core/network",
@@ -37,6 +36,8 @@ members = [
 	"core/sr-sandbox",
 	"core/sr-std",
 	"core/sr-version",
+	"core/transaction-graph",
+	"core/transaction-pool",
 	"srml/support",
 	"srml/balances",
 	"srml/consensus",

core/transaction-graph/Cargo.toml

@@ -0,0 +1,9 @@
+[package]
+name = "substrate-transaction-graph"
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+
+[dependencies]
+error-chain = "0.12"
+log = "0.3.0"
+sr-primitives = { path = "../sr-primitives" }

core/transaction-graph/src/error.rs

@@ -0,0 +1,37 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+use sr_primitives::transaction_validity::TransactionPriority as Priority;
+
+error_chain! {
+	errors {
+		/// The transaction is already in the pool.
+		AlreadyImported {
+			description("Transaction is already in the pool."),
+			display("Already imported"),
+		}
+		/// The transaction cannot be imported cause it's a replacement and has too low priority.
+		TooLowPriority(old: Priority, new: Priority) {
+			description("The priority is too low to replace transactions already in the pool."),
+			display("Too low priority ({} > {})", old, new)
+		}
+		/// Deps cycle detected and we couldn't import transaction.
+		CycleDetected {
+			description("Transaction was not imported because of detected cycle."),
+			display("Cycle Detected"),
+		}
+	}
+}

core/transaction-graph/src/future.rs

@@ -0,0 +1,177 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+use std::{
+	collections::{HashMap, HashSet},
+	hash,
+};
+
+use sr_primitives::transaction_validity::{
+	TransactionTag as Tag,
+};
+
+use pool::Transaction;
+
+/// Transaction with partially satisfied dependencies.
+#[derive(Debug)]
+pub struct WaitingTransaction<Hash> {
+	/// Transaction details.
+	pub transaction: Transaction<Hash>,
+	/// Tags that are required and have not been satisfied yet by other transactions in the pool.
+	pub missing_tags: HashSet<Tag>,
+}
+
+impl<Hash> WaitingTransaction<Hash> {
+	/// Creates a new `WaitingTransaction`.
+	///
+	/// Computes the set of missing tags based on the requirements and tags that
+	/// are provided by all transactions in the ready queue.
+	pub fn new(transaction: Transaction<Hash>, provided: &HashMap<Tag, Hash>) -> Self {
+		let missing_tags = transaction.requires
+			.iter()
+			.filter(|tag| !provided.contains_key(&**tag))
+			.cloned()
+			.collect();
+
+		WaitingTransaction {
+			transaction,
+			missing_tags,
+		}
+	}
+
+	/// Marks the tag as satisfied.
+	pub fn satisfy_tag(&mut self, tag: &Tag) {
+		self.missing_tags.remove(tag);
+	}
+
+	/// Returns true if transaction has all requirements satisfied.
+	pub fn is_ready(&self) -> bool {
+		self.missing_tags.is_empty()
+	}
+}
+
+/// A pool of transactions that are not yet ready to be included in the block.
+///
+/// Contains transactions that are still awaiting for some other transactions that
+/// could provide a tag that they require.
+#[derive(Debug)]
+pub struct FutureTransactions<Hash: hash::Hash + Eq> {
+	/// tags that are not yet provided by any transaction and we await for them
+	wanted_tags: HashMap<Tag, HashSet<Hash>>,
+	/// Transactions waiting for a particular other transaction
+	waiting: HashMap<Hash, WaitingTransaction<Hash>>,
+}
+
+impl<Hash: hash::Hash + Eq> Default for FutureTransactions<Hash> {
+	fn default() -> Self {
+		FutureTransactions {
+			wanted_tags: Default::default(),
+			waiting: Default::default(),
+		}
+	}
+}
+
+const WAITING_PROOF: &str = r"#
+In import we always insert to `waiting` if we push to `wanted_tags`;
+when removing from `waiting` we always clear `wanted_tags`;
+every hash from `wanted_tags` is always present in `waiting`;
+qed
+#";
+
+impl<Hash: hash::Hash + Eq + Clone> FutureTransactions<Hash> {
+	/// Import transaction to Future queue.
+	///
+	/// Only transactions that don't have all their tags satisfied should occupy
+	/// the Future queue.
+	/// As soon as required tags are provided by some other transactions that are ready
+	/// we should remove the transactions from here and move them to the Ready queue.
+	pub fn import(&mut self, tx: WaitingTransaction<Hash>) {
+		assert!(!tx.is_ready(), "Transaction is ready.");
+		assert!(!self.waiting.contains_key(&tx.transaction.hash), "Transaction is already imported.");
+
+		// Add all tags that are missing
+		for tag in &tx.missing_tags {
+			let mut entry = self.wanted_tags.entry(tag.clone()).or_insert_with(HashSet::new);
+			entry.insert(tx.transaction.hash.clone());
+		}
+
+		// Add the transaction to a by-hash waiting map
+		self.waiting.insert(tx.transaction.hash.clone(), tx);
+	}
+
+	/// Returns true if given hash is part of the queue.
+	pub fn contains(&self, hash: &Hash) -> bool {
+		self.waiting.contains_key(hash)
+	}
+
+	/// Satisfies provided tags in transactions that are waiting for them.
+	///
+	/// Returns (and removes) transactions that became ready after their last tag got
+	/// satisfied and now we can remove them from Future and move to Ready queue.
+	pub fn satisfy_tags<T: AsRef<Tag>>(&mut self, tags: impl IntoIterator<Item=T>) -> Vec<WaitingTransaction<Hash>> {
+		let mut became_ready = vec![];
+
+		for tag in tags {
+			if let Some(hashes) = self.wanted_tags.remove(tag.as_ref()) {
+				for hash in hashes {
+					let is_ready = {
+						let mut tx = self.waiting.get_mut(&hash)
+							.expect(WAITING_PROOF);
+						tx.satisfy_tag(tag.as_ref());
+						tx.is_ready()
+					};
+
+					if is_ready {
+						let tx = self.waiting.remove(&hash).expect(WAITING_PROOF);
+						became_ready.push(tx);
+					}
+				}
+			}
+		}
+
+		became_ready
+	}
+
+	/// Removes transactions for given list of hashes.
+	///
+	/// Returns a list of actually removed transactions.
+	pub fn remove(&mut self, hashes: &[Hash]) -> Vec<Transaction<Hash>> {
+		let mut removed = vec![];
+		for hash in hashes {
+			if let Some(waiting_tx) = self.waiting.remove(hash) {
+				// remove from wanted_tags as well
+				for tag in waiting_tx.missing_tags {
+					let remove = if let Some(mut wanted) = self.wanted_tags.get_mut(&tag) {
+						wanted.remove(hash);
+						wanted.is_empty()
+					} else { false };
+					if remove {
+						self.wanted_tags.remove(&tag);
+					}
+				}
+				// add to result
+				removed.push(waiting_tx.transaction)
+			}
+		}
+		removed
+	}
+
+	/// Returns number of transactions in the Future queue.
+	#[cfg(test)]
+	pub fn len(&self) -> usize {
+		self.waiting.len()
+	}
+}

core/transaction-graph/src/lib.rs

@@ -0,0 +1,45 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Generic Transaction Pool
+//!
+//! The pool is based on dependency graph between transactions
+//! and their priority.
+//! The pool is able to return an iterator that traverses transaction
+//! graph in the correct order taking into account priorities and dependencies.
+//!
+//! TODO [ToDr]
+//! - [ ] Longevity handling (remove obsolete transactions periodically)
+//! - [ ] Banning / Future-rotation (once rejected (as invalid) should not be accepted for some time)
+//! - [ ] Multi-threading (getting ready transactions should not block the pool)
+
+#![warn(missing_docs)]
+#![warn(unused_extern_crates)]
+
+extern crate sr_primitives;
+
+#[macro_use]
+extern crate error_chain;
+
+#[macro_use]
+extern crate log;
+
+mod error;
+mod future;
+mod pool;
+mod ready;
+
+pub use self::pool::{Transaction, Pool};