Watch transactions pool #10558

IntegralTeam · 2019-04-02T05:34:01Z

Issue #9713
Still in progress, but would like some review on current approach.
What is not ready:

Implementation for LightChainClient
testing

I will add more if feedback will be positive

parity-cla-bot · 2019-04-02T05:34:04Z

It looks like @IntegralTeam hasn't signed our Contributor License Agreement, yet.

The purpose of a CLA is to ensure that the guardian of a project's outputs has the necessary ownership or grants of rights over all contributions to allow them to distribute under the chosen licence.
Wikipedia

You can read and sign our full Contributor License Agreement at the following URL: https://cla.parity.io

Once you've signed, please reply to this thread with [clabot:check] to prove it.

Many thanks,

Parity Technologies CLA Bot

IntegralTeam · 2019-04-02T05:36:37Z

[clabot:check]

parity-cla-bot · 2019-04-02T05:36:38Z

It looks like @IntegralTeam signed our Contributor License Agreement. 👍

Many thanks,

Parity Technologies CLA Bot

tomusdrw

Looks really good for a WiP, few things to polish.

tomusdrw · 2019-04-02T08:57:24Z

ethcore/src/miner/miner.rs

@@ -262,6 +263,13 @@ impl Miner {
 		self.transaction_queue.add_listener(f);
 	}

+	/// Set a callback to be notified
+	pub fn get_tx_pool_receiver(&self) -> mpsc::Receiver<(H256, String)> {


Yo can ditch get_ imho.

tomusdrw · 2019-04-02T08:58:18Z

miner/src/pool/listener.rs

+/// Transactions pool notifier
+#[derive(Default)]
+pub struct TransactionsPoolNotifier {
+	listeners: Vec<Arc<Mutex<mpsc::Sender<(H256, String)>>>>,


We don't really need Arc<Mutex here afaict.

Suggested change

listeners: Vec<Arc<Mutex<mpsc::Sender<(H256, String)>>>>,

listeners: Vec<mpsc::Sender<(H256, String)>>,

@tomusdrw If I delete the Arc<Mutex<, it leads to errors

Compiling ethcore v1.12.0 (/home/user/Work/parity-ethereum/ethcore) error[E0277]: `std::sync::mpsc::Sender<ethcore_miner::pool::listener::TxStatus>` cannot be shared between threads safely --> ethcore/src/client/client.rs:1305:6 | 1305 | impl snapshot::DatabaseRestore for Client { | ^^^^^^^^^^^^^^^^^^^^^^^^^ `std::sync::mpsc::Sender<ethcore_miner::pool::listener::TxStatus>` cannot be shared between threads safely | = help: the trait `std::marker::Sync` is not implemented for `std::sync::mpsc::Sender<ethcore_miner::pool::listener::TxStatus>` = note: required because of the requirements on the impl of `std::marker::Sync` for `std::ptr::Unique<std::sync::mpsc::Sender<ethcore_miner::pool::listener::TxStatus>>` = note: required because it appears within the type `alloc::raw_vec::RawVec<std::sync::mpsc::Sender<ethcore_miner::pool::listener::TxStatus>>` = note: required because it appears within the type `std::vec::Vec<std::sync::mpsc::Sender<ethcore_miner::pool::listener::TxStatus>>` = note: required because it appears within the type `ethcore_miner::pool::listener::TransactionsPoolNotifier` = note: required because it appears within the type `(ethcore_miner::pool::listener::Logger, ethcore_miner::pool::listener::TransactionsPoolNotifier)` = note: required because it appears within the type `(ethcore_miner::pool::listener::Notifier, (ethcore_miner::pool::listener::Logger, ethcore_miner::pool::listener::TransactionsPoolNotifier))` = note: required because it appears within the type `(ethcore_miner::pool::local_transactions::LocalTransactionsList, (ethcore_miner::pool::listener::Notifier, (ethcore_miner::pool::listener::Logger, ethcore_miner::pool::listener::TransactionsPoolNotifier)))` = note: required because it appears within the type `transaction_pool::pool::Pool<ethcore_miner::pool::VerifiedTransaction, ethcore_miner::pool::scoring::NonceAndGasPrice, (ethcore_miner::pool::local_transactions::LocalTransactionsList, (ethcore_miner::pool::listener::Notifier, (ethcore_miner::pool::listener::Logger, ethcore_miner::pool::listener::TransactionsPoolNotifier)))>` = note: required because of the requirements on the impl of `std::marker::Sync` for `lock_api::rwlock::RwLock<parking_lot::RawRwLock, transaction_pool::pool::Pool<ethcore_miner::pool::VerifiedTransaction, ethcore_miner::pool::scoring::NonceAndGasPrice, (ethcore_miner::pool::local_transactions::LocalTransactionsList, (ethcore_miner::pool::listener::Notifier, (ethcore_miner::pool::listener::Logger, ethcore_miner::pool::listener::TransactionsPoolNotifier)))>>` = note: required because it appears within the type `ethcore_miner::pool::TransactionQueue` = note: required because of the requirements on the impl of `std::marker::Send` for `std::sync::Arc<ethcore_miner::pool::TransactionQueue>` = note: required because it appears within the type `miner::miner::Miner` = note: required because of the requirements on the impl of `std::marker::Sync` for `std::sync::Arc<miner::miner::Miner>` = note: required because it appears within the type `client::client::Importer` = note: required because it appears within the type `client::client::Client` ...

Found this: Allowing Access from Multiple Threads with Sync
And it seems to me that either Mutex is needed here, or I don't understand what am I doing wrong.

Ah, I see. Yes, unfortunatelly we can't satisfy Sync with std::mpsc::Sender. To overcome this I'd go with mpsc::unbounded_channel from futures crate, this will also allow us to get rid of this extra thread responsible for polling the channel.

tomusdrw · 2019-04-02T08:58:39Z

miner/src/pool/listener.rs

+impl TransactionsPoolNotifier {
+	/// Add new listener to receive notifications.
+	pub fn add(&mut self, f: mpsc::Sender<(H256, String)>) {
+		self.listeners.push(Arc::new(Mutex::new(f)));


Suggested change

self.listeners.push(Arc::new(Mutex::new(f)));

self.listeners.push(f);

tomusdrw · 2019-04-02T08:58:48Z

miner/src/pool/listener.rs

+	/// Notify listeners about all currently transactions.
+	fn notify(& mut self, hash: H256, status: String) {
+		for l in &self.listeners {
+			let l = l.lock();


This would also clear dropped listeners:

Suggested change

let l = l.lock();

self.listeners

.retain(|sender| sender.send((hash, status.clone()).is_ok())

tomusdrw · 2019-04-02T09:02:53Z

miner/src/pool/listener.rs

+impl txpool::Listener<Transaction> for TransactionsPoolNotifier {
+	fn added(&mut self, tx: &Arc<Transaction>, _old: Option<&Arc<Transaction>>) {
+		let hash = tx.hash.clone();
+		self.notify(hash, "added".to_string());


I think it would be better to have an enum like this:

#[derive(Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] enum Status { Added, Rejected, ... Culled, }

And the same status enum should be sent in the notification.

tomusdrw · 2019-04-02T09:08:11Z

rpc/src/v1/impls/transactions_pool.rs

+    }
+    pub fn run(&self, pool_receiver: mpsc::Receiver<(H256, String)>) {
+        let handler = self.handler.clone();
+        thread::spawn(move || loop {


Can't we use future-aware channels instead to avoid spawning a new thread?

tomusdrw · 2019-04-02T09:08:20Z

rpc/src/v1/impls/transactions_pool.rs

+
+}
+
+pub struct TransactionsNotificationHandler {


tomusdrw · 2019-04-02T09:10:38Z

rpc/src/v1/impls/transactions_pool.rs

+
+    pub fn notify_transaction(&self, status: (H256, String)) {
+        for subscriber in self.transactions_pool_subscribers.read().values() {
+            let status = (status.0.clone(), status.1.clone());


Suggested change

let status = (status.0.clone(), status.1.clone());

let status = status.clone();

or even inline it into the next line.

tomusdrw · 2019-04-02T09:11:23Z

rpc/src/v1/traits/transactions_pool.rs

+
+    /// Subscribe to Transactions Pool subscription.
+    #[pubsub(subscription = "parity_watchTransactionsPool", subscribe, name = "parity_watchTransactionsPool")]
+    fn subscribe(&self, Self::Metadata, typed::Subscriber<pubsub::Result>, Option<pubsub::Params>);


If you don't want any params just leave this out:

Suggested change

fn subscribe(&self, Self::Metadata, typed::Subscriber<pubsub::Result>, Option<pubsub::Params>);

fn subscribe(&self, Self::Metadata, typed::Subscriber<pubsub::Result>);

tomusdrw · 2019-04-02T09:12:37Z

rpc/src/v1/impls/transactions_pool.rs

+impl TransactionsPool for TransactionsPoolClient {
+    type Metadata = Metadata;
+
+    fn subscribe(&self, _meta: Metadata, subscriber: Subscriber<pubsub::Result>, params: Option<pubsub::Params>) {


using pubsub::Result here kind of suggests that we might get different values of this enum as well. I think it should be safe to just use the tuple directly:

Suggested change

fn subscribe(&self, _meta: Metadata, subscriber: Subscriber<pubsub::Result>, params: Option<pubsub::Params>) {

fn subscribe(&self, _meta: Metadata, subscriber: Subscriber<(H256, Status)>, params: Option<pubsub::Params>) {

tomusdrw · 2019-04-03T13:06:59Z

miner/src/pool/listener.rs

+		self.notify(hash, TxStatus::Added);
+	}
+
+	fn rejected(&mut self, tx: &Arc<Transaction>, _reason: &txpool::ErrorKind) {


I just realised that this might be called really frequently, and currently we need to iterate over all senders and have a unique lock for this, so this will have drastic performance consequences.

Instead I think we should do something very similar that we do with Notifer in the same file, i.e:

We first aggregate all the notifications that should be sent

Then we send them all at once after batch import to the queue is finished.

It might be even worth to remove Notifier and convert this other notification to use this (channel-based) notifier instead.

struct Notifier { pending: HashMap<H256, TxStatus>, listeners: Vec<futures::sync::mpsc::UnboundedSender<Arc<HashMap<H256, TxStatus>>>>, } impl Notifier { fn notify(&mut self) { let to_send = Arc::new(std::mem::replace(&mut self.pending, HashMap::new)); self.listeners.retain(|listener| listener.unbounded_send(to_send.clone()).is_ok()); } } impl txpool::Listener<Transaction> for TransactionPoolNotifier { fn added(&mut self, ...) { self.pending.insert(tx.hash, TxStatus::Added); } ... }

@IntegralTeam Please let me know if you need any more explanations or help with this.

@tomusdrw Thank you, I've got what I need to do.

But, I think, it's not quite right to use HashMap here, because transaction firstly could be added with according Added status, and then before sending notification, same transaction could get status Culled and it will lead that Added status will be erased from HashMap

Maybe it will be better to use tuple vector Vec<(H256, TxStatus)>?

@IntegralTeam good point, yeah if we never want to miss any event and have them in order, then Vec is indeed a good choice here.

Ok, let's do it this way.

tomusdrw

Looks nice now, I'd like to see the other Notifier rewritten to futures as well, but we can leave that as a separate PR - please create an issue for this if that happens to be the case.

tomusdrw · 2019-04-08T09:41:10Z

miner/src/pool/queue.rs

@@ -303,6 +303,8 @@ impl TransactionQueue {
 		// Notify about imported transactions.
 		(self.pool.write().listener_mut().1).0.notify();


This listener now seems obsolete, since it just duplicates the same data and is less efficient than the new channel-based one.

@tomusdrw

Looks nice now, I'd like to see the other Notifier rewritten to futures as well, but we can leave that as a separate PR - please create an issue for this if that happens to be the case.

Ok, then let`s do it in other PR

This listener now seems obsolete, since it just duplicates the same data and is less efficient than the new channel-based one.

Maybe we can make one Notifier by combining old one Notifier and TransactionsPoolNotifier and store two different lists of listeners for parity_watchTransactionsPool and eth_pubsub in it?

# Conflicts: # ethcore/src/miner/miner.rs # miner/src/pool/queue.rs # rpc/src/v1/types/pubsub.rs

IntegralTeam · 2019-04-10T09:50:29Z

@tomusdrw Please review

tomusdrw

One last tiny grumble regarding recently added commit, but looks good otherwise.

Could you also create an issue to merge the two listeners?

tomusdrw · 2019-04-10T13:00:50Z

ethcore/light/src/transaction_queue.rs

+
+		let to_send: Vec<(H256, TxStatus)> = hashes.into_iter().map(|hash| (hash.clone(), status)).collect();
+
+		self.tx_statuses_listeners.retain(| listener| listener.unbounded_send(Arc::new(to_send.clone())).is_ok());


That's incorrect, the reason we have Arc is to avoid cloning the entire Vec. So:

First create the Arc

Then clone Arc not Vec

@tomusdrw I fixed it

IntegralTeam · 2019-04-10T16:50:56Z

@tomusdrw

Could you also create an issue to merge the two listeners?

Yesterday I already created an issuee #10581
Can I change the title and description?

tomusdrw · 2019-04-10T21:00:20Z

Perfect 👌

ethcore/light/src/transaction_queue.rs

* master: CI improvements (#10579) Watch transactions pool (#10558) fix(evmbin): make benches compile again (#10586) fix issue with compilation when 'slow-blocks' feature enabled (#10585) Reject crazy timestamps instead of truncating. (#10574) Node table limiting and cache for node filter (#10288)

IntegralTeam added 3 commits March 28, 2019 19:48

add parity_watchTransactionsPool

2c4034f

change TransactionsPool PubSub for using mpsc::channel

458a85b

add missing comments

79e3afe

soc1c added this to the 2.6 milestone Apr 2, 2019

soc1c added A3-inprogress ⏳ Pull request is in progress. No review needed at this stage. M6-rpcapi 📣 RPC API. labels Apr 2, 2019

soc1c requested a review from tomusdrw April 2, 2019 07:48

tomusdrw reviewed Apr 2, 2019

View reviewed changes

IntegralTeam added 3 commits April 3, 2019 15:58

replace spaces with tabs and remove params

dc6d4e0

add enum TxStatus

d68e4c7

add parity prefix for transaction pool PubSub

055021d

tomusdrw reviewed Apr 3, 2019

View reviewed changes

IntegralTeam added 2 commits April 5, 2019 19:29

reimplement Transactions pool PubSub with using futures::sync::mpsc;

629c590

fix falling tests

d908a39

tomusdrw reviewed Apr 8, 2019

View reviewed changes

IntegralTeam added 2 commits April 9, 2019 09:56

add transactions poll PubSub for LightChainClient

591b14e

Merge branch 'master' into watch-transactions-pool

ea12cd5

# Conflicts: # ethcore/src/miner/miner.rs # miner/src/pool/queue.rs # rpc/src/v1/types/pubsub.rs

IntegralTeam mentioned this pull request Apr 9, 2019

Merge Notifier and TransactionsPoolNotifier #10581

Closed

IntegralTeam changed the title ~~[WIP] Watch transactions pool~~ Watch transactions pool Apr 10, 2019

tomusdrw reviewed Apr 10, 2019

View reviewed changes

tomusdrw added A5-grumble 🔥 Pull request has minor issues that must be addressed before merging. and removed A3-inprogress ⏳ Pull request is in progress. No review needed at this stage. labels Apr 10, 2019

fix Arc clone instead of Vec

286ba53

IntegralTeam force-pushed the watch-transactions-pool branch from 18b0800 to 286ba53 Compare April 10, 2019 16:37

tomusdrw approved these changes Apr 10, 2019

View reviewed changes

tomusdrw added A8-looksgood 🦄 Pull request is reviewed well. and removed A5-grumble 🔥 Pull request has minor issues that must be addressed before merging. labels Apr 10, 2019

seunlanlege suggested changes Apr 12, 2019

View reviewed changes

ethcore/light/src/transaction_queue.rs Show resolved Hide resolved

seunlanlege approved these changes Apr 12, 2019

View reviewed changes

seunlanlege merged commit 944bf6a into openethereum:master Apr 12, 2019

fanatid mentioned this pull request Apr 12, 2019

Feature request: parity_allTransactionHashes / parity_watchPendingTransactions #9713

Closed

jam10o-new mentioned this pull request Apr 12, 2019

I would like to get listtransactions for ether like in bitcoin. #10518

Closed

IntegralTeam deleted the watch-transactions-pool branch April 15, 2019 02:24

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Watch transactions pool #10558

Watch transactions pool #10558

IntegralTeam commented Apr 2, 2019 •

edited

Loading

parity-cla-bot commented Apr 2, 2019

IntegralTeam commented Apr 2, 2019

parity-cla-bot commented Apr 2, 2019

tomusdrw left a comment

tomusdrw Apr 2, 2019

tomusdrw Apr 2, 2019

IntegralTeam Apr 3, 2019

tomusdrw Apr 3, 2019

tomusdrw Apr 2, 2019

tomusdrw Apr 2, 2019

tomusdrw Apr 2, 2019

tomusdrw Apr 2, 2019

tomusdrw Apr 2, 2019

tomusdrw Apr 2, 2019

tomusdrw Apr 2, 2019

tomusdrw Apr 2, 2019

tomusdrw Apr 3, 2019

tomusdrw Apr 3, 2019

IntegralTeam Apr 4, 2019

tomusdrw Apr 4, 2019

IntegralTeam Apr 4, 2019

tomusdrw left a comment

tomusdrw Apr 8, 2019

IntegralTeam Apr 9, 2019

IntegralTeam commented Apr 10, 2019

tomusdrw left a comment

tomusdrw Apr 10, 2019

IntegralTeam Apr 10, 2019

IntegralTeam commented Apr 10, 2019

tomusdrw commented Apr 10, 2019

	listeners: Vec<Arc<Mutex<mpsc::Sender<(H256, String)>>>>,
	listeners: Vec<mpsc::Sender<(H256, String)>>,

	self.listeners.push(Arc::new(Mutex::new(f)));
	self.listeners.push(f);

	let l = l.lock();
	self.listeners
	.retain(\|sender\| sender.send((hash, status.clone()).is_ok())

	let status = (status.0.clone(), status.1.clone());
	let status = status.clone();

	fn subscribe(&self, Self::Metadata, typed::Subscriber<pubsub::Result>, Option<pubsub::Params>);
	fn subscribe(&self, Self::Metadata, typed::Subscriber<pubsub::Result>);

	fn subscribe(&self, _meta: Metadata, subscriber: Subscriber<pubsub::Result>, params: Option<pubsub::Params>) {
	fn subscribe(&self, _meta: Metadata, subscriber: Subscriber<(H256, Status)>, params: Option<pubsub::Params>) {

		@@ -303,6 +303,8 @@ impl TransactionQueue {
		// Notify about imported transactions.
		(self.pool.write().listener_mut().1).0.notify();


		let to_send: Vec<(H256, TxStatus)> = hashes.into_iter().map(\|hash\| (hash.clone(), status)).collect();

		self.tx_statuses_listeners.retain(\| listener\| listener.unbounded_send(Arc::new(to_send.clone())).is_ok());

Watch transactions pool #10558

Watch transactions pool #10558

Conversation

IntegralTeam commented Apr 2, 2019 • edited Loading

parity-cla-bot commented Apr 2, 2019

IntegralTeam commented Apr 2, 2019

parity-cla-bot commented Apr 2, 2019

tomusdrw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tomusdrw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

IntegralTeam commented Apr 10, 2019

tomusdrw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

IntegralTeam commented Apr 10, 2019

tomusdrw commented Apr 10, 2019

IntegralTeam commented Apr 2, 2019 •

edited

Loading