Subsystems memory tracking: 1. Transaction pool

client/Cargo.toml

@@ -19,7 +19,7 @@ hash-db = { version = "0.15.2" }
 hex-literal = { version = "0.2.1" }
 sp-inherents = { version = "2.0.0", path = "../primitives/inherents" }
 sp-keyring = { version = "2.0.0", path = "../primitives/keyring" }
-kvdb = "0.3.0"
+kvdb = "0.4.0"
 log = { version = "0.4.8" }
 parking_lot = "0.10.0"
 sp-core = { version = "2.0.0", path = "../primitives/core" }
@@ -37,5 +37,5 @@ tracing = "0.1.10"
 env_logger = "0.7.0"
 tempfile = "3.1.0"
 substrate-test-runtime-client = { version = "2.0.0", path = "../test-utils/runtime/client" }
-kvdb-memorydb = "0.3.0"
+kvdb-memorydb = "0.4.0"
 sp-panic-handler = { version = "2.0.0", path = "../primitives/panic-handler" }

client/api/Cargo.toml

@@ -18,7 +18,7 @@ sp-blockchain = { version = "2.0.0", path = "../../primitives/blockchain" }
 hex-literal = { version = "0.2.1" }
 sp-inherents = { version = "2.0.0", default-features = false, path = "../../primitives/inherents" }
 sp-keyring = { version = "2.0.0", path = "../../primitives/keyring" }
-kvdb = "0.3.0"
+kvdb = "0.4.0"
 log = { version = "0.4.8" }
 parking_lot = "0.10.0"
 sp-core = { version = "2.0.0", default-features = false, path = "../../primitives/core" }

client/cli/Cargo.toml

@@ -35,6 +35,7 @@ names = "0.11.0"
 structopt = "0.3.8"
 sc-tracing = { version = "2.0.0", path = "../tracing" }
 chrono = "0.4.10"
+parity-util-mem = { version = "0.5.1", default-features = false, features = ["primitive-types"] }
 
 [target.'cfg(not(target_os = "unknown"))'.dependencies]
 rpassword = "4.0.1"

client/cli/src/informant.rs

@@ -28,6 +28,7 @@ mod display;
 /// Creates an informant in the form of a `Future` that must be polled regularly.
 pub fn build(service: &impl AbstractService) -> impl futures::Future<Output = ()> {
 	let client = service.client();
+	let pool = service.transaction_pool();
 
 	let mut display = display::InformantDisplay::new();
 
@@ -40,6 +41,11 @@ pub fn build(service: &impl AbstractService) -> impl futures::Future<Output = ()
 			} else {
 				trace!(target: "usage", "Usage statistics not displayed as backend does not provide it")
 			}
+			trace!(
+				target: "usage",
+				"Subsystems memory [txpool: {} kB]",
+				parity_util_mem::malloc_size(&*pool) / 1024,
+			);
 			display.display(&info, net_status);
 			future::ready(())
 		});

client/db/Cargo.toml

@@ -8,12 +8,12 @@ license = "GPL-3.0"
 [dependencies]
 parking_lot = "0.10.0"
 log = "0.4.8"
-kvdb = "0.3.0"
-kvdb-rocksdb = { version = "0.4", optional = true }
-kvdb-memorydb = "0.3.0"
+kvdb = "0.4.0"
+kvdb-rocksdb = { version = "0.5", optional = true }
+kvdb-memorydb = "0.4.0"
 linked-hash-map = "0.5.2"
 hash-db = "0.15.2"
-parity-util-mem = { version = "0.4", default-features = false, features = ["std"] }
+parity-util-mem = { version = "0.5.1", default-features = false, features = ["std"] }
 codec = { package = "parity-scale-codec", version = "1.0.0", features = ["derive"] }
 
 sc-client-api = { version = "2.0.0", path = "../api" }
@@ -32,7 +32,7 @@ sp-keyring = { version = "2.0.0", path = "../../primitives/keyring" }
 substrate-test-runtime-client = { version = "2.0.0", path = "../../test-utils/runtime/client" }
 env_logger = "0.7.0"
 quickcheck = "0.9"
-kvdb-rocksdb = "0.4"
+kvdb-rocksdb = "0.5"
 tempfile = "3"
 
 [features]

client/service/Cargo.toml

@@ -56,6 +56,7 @@ parity-multiaddr = { package = "parity-multiaddr", version = "0.5.0" }
 grafana-data-source = { version = "0.8", path = "../../utils/grafana-data-source" }
 sc-tracing = { version = "2.0.0", path = "../tracing" }
 tracing = "0.1.10"
+parity-util-mem = { version = "0.5.1", default-features = false, features = ["primitive-types"] }
 
 [dev-dependencies]
 substrate-test-runtime-client = { version = "2.0.0", path = "../../test-utils/runtime/client" }

client/service/src/builder.rs

@@ -995,10 +995,12 @@ ServiceBuilder<
 				"used_db_cache_size" => info.usage.as_ref().map(|usage| usage.memory.database_cache).unwrap_or(0),
 				"disk_read_per_sec" => info.usage.as_ref().map(|usage| usage.io.bytes_read).unwrap_or(0),
 				"disk_write_per_sec" => info.usage.as_ref().map(|usage| usage.io.bytes_written).unwrap_or(0),
+				"memory_transaction_pool" => parity_util_mem::malloc_size(&*transaction_pool_),
 			);
 
 			ready(())
 		});
+
 		let _ = to_spawn_tx.unbounded_send((
 			Box::pin(select(tel_task, exit.clone()).map(drop)),
 			From::from("telemetry-periodic-send")

client/transaction-pool/Cargo.toml

@@ -19,6 +19,7 @@ sc-transaction-graph = { version = "2.0.0", path = "./graph" }
 sp-transaction-pool = { version = "2.0.0", path = "../../primitives/transaction-pool" }
 sc-client-api = { version = "2.0.0", path = "../api" }
 sp-blockchain = { version = "2.0.0", path = "../../primitives/blockchain" }
+parity-util-mem = { version = "0.5.1", default-features = false, features = ["primitive-types"] }
 
 [dev-dependencies]
 sp-keyring = { version = "2.0.0", path = "../../primitives/keyring" }

client/transaction-pool/graph/Cargo.toml

@@ -14,6 +14,7 @@ serde = { version = "1.0.101", features = ["derive"] }
 sp-core = { version = "2.0.0", path = "../../../primitives/core" }
 sp-runtime = { version = "2.0.0", path = "../../../primitives/runtime" }
 sp-transaction-pool = { version = "2.0.0", path = "../../../primitives/transaction-pool" }
+parity-util-mem = { version = "0.5.1", default-features = false, features = ["primitive-types"] }
 
 [dev-dependencies]
 assert_matches = "1.3.0"

client/transaction-pool/graph/src/base_pool.rs

@@ -84,7 +84,7 @@ pub struct PruneStatus<Hash, Ex> {
 
 /// Immutable transaction
 #[cfg_attr(test, derive(Clone))]
-#[derive(PartialEq, Eq)]
+#[derive(PartialEq, Eq, parity_util_mem::MallocSizeOf)]
 pub struct Transaction<Hash, Extrinsic> {
 	/// Raw extrinsic representing that transaction.
 	pub data: Extrinsic,
@@ -209,7 +209,7 @@ const RECENTLY_PRUNED_TAGS: usize = 2;
 /// as-is for the second time will fail or produce unwanted results.
 /// Most likely it is required to revalidate them and recompute set of
 /// required tags.
-#[derive(Debug)]
+#[derive(Debug, parity_util_mem::MallocSizeOf)]
 pub struct BasePool<Hash: hash::Hash + Eq, Ex> {
 	reject_future_transactions: bool,
 	future: FutureTransactions<Hash, Ex>,
@@ -846,6 +846,33 @@ mod tests {
 		}
 	}
 
+	#[test]
+	fn can_track_heap_size() {
+		let mut pool = pool();
+		pool.import(Transaction {
+			data: vec![5u8; 1024],
+			bytes: 1,
+			hash: 5,
+			priority: 5u64,
+			valid_till: 64u64,
+			requires: vec![],
+			provides: vec![vec![0], vec![4]],
+			propagate: true,
+		}).expect("import 1 should be ok");
+		pool.import(Transaction {
+			data: vec![3u8; 1024],
+			bytes: 1,
+			hash: 7,
+			priority: 5u64,
+			valid_till: 64u64,
+			requires: vec![],
+			provides: vec![vec![2], vec![7]],
+			propagate: true,
+		}).expect("import 2 should be ok");
+
+		assert!(parity_util_mem::malloc_size(&pool) > 5000);
+	}
+
 	#[test]
 	fn should_remove_invalid_transactions() {
 		// given

client/transaction-pool/graph/src/future.rs

@@ -29,6 +29,7 @@ use sp_runtime::transaction_validity::{
 
 use crate::base_pool::Transaction;
 
+#[derive(parity_util_mem::MallocSizeOf)]
 /// Transaction with partially satisfied dependencies.
 pub struct WaitingTransaction<Hash, Ex> {
 	/// Transaction details.
@@ -109,7 +110,7 @@ impl<Hash, Ex> WaitingTransaction<Hash, Ex> {
 ///
 /// Contains transactions that are still awaiting for some other transactions that
 /// could provide a tag that they require.
-#[derive(Debug)]
+#[derive(Debug, parity_util_mem::MallocSizeOf)]
 pub struct FutureTransactions<Hash: hash::Hash + Eq, Ex> {
 	/// tags that are not yet provided by any transaction and we await for them
 	wanted_tags: HashMap<Tag, HashSet<Hash>>,
@@ -243,3 +244,30 @@ impl<Hash: hash::Hash + Eq + Clone, Ex> FutureTransactions<Hash, Ex> {
 		self.waiting.values().fold(0, |acc, tx| acc + tx.transaction.bytes)
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn can_track_heap_size() {
+		let mut future = FutureTransactions::default();
+		future.import(WaitingTransaction {
+			transaction: Transaction {
+				data: vec![0u8; 1024],
+				bytes: 1,
+				hash: 1,
+				priority: 1,
+				valid_till: 2,
+				requires: vec![vec![1], vec![2]],
+				provides: vec![vec![3], vec![4]],
+				propagate: true,
+			}.into(),
+			missing_tags: vec![vec![1u8], vec![2u8]].into_iter().collect(),
+			imported_at: std::time::Instant::now(),
+		});
+
+		// data is at least 1024!
+		assert!(parity_util_mem::malloc_size(&future) > 1024);
+	}
+}

client/transaction-pool/graph/src/pool.rs

@@ -122,6 +122,16 @@ pub struct Pool<B: ChainApi> {
 	validated_pool: Arc<ValidatedPool<B>>,
 }
 
+impl<B: ChainApi> parity_util_mem::MallocSizeOf for Pool<B>
+where
+	B::Hash: parity_util_mem::MallocSizeOf,
+	ExtrinsicFor<B>: parity_util_mem::MallocSizeOf,
+{
+	fn size_of(&self, ops: &mut parity_util_mem::MallocSizeOfOps) -> usize {
+		self.validated_pool.size_of(ops)
+	}
+}
+
 impl<B: ChainApi> Pool<B> {
 	/// Create a new transaction pool.
 	pub fn new(options: Options, api: Arc<B>) -> Self {