Watch transactions pool by IntegralTeam · Pull Request #10558 · openethereum/parity-ethereum

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

 	}

+	/// 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

+/// 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

+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

+	/// 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

+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

+    }
+    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

+
+}
+
+pub struct TransactionsNotificationHandler {


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

+
+    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

+
+    /// 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

+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

+		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

@@ -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

+
+		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 👌

* 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

Comment thread ethcore/light/src/transaction_queue.rs

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

	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());

Conversation

IntegralTeam commented Apr 2, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

parity-cla-bot commented Apr 2, 2019

Uh oh!

IntegralTeam commented Apr 2, 2019

Uh oh!

parity-cla-bot commented Apr 2, 2019

Uh oh!

tomusdrw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tomusdrw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

IntegralTeam commented Apr 10, 2019

Uh oh!

tomusdrw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

IntegralTeam commented Apr 10, 2019

Uh oh!

tomusdrw commented Apr 10, 2019

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

IntegralTeam commented Apr 2, 2019 •

edited

Loading