feat(txpool) feed new pending transactions to BestTxns iterator #4053
Conversation
Codecov Report
... and 21 files with indirect coverage changes
Flags with carried forward coverage won't be shown. Click here to find out more.
|
There was a problem hiding this comment.
this looks great and is exactly what I had in mind.
maybe @supernovahs can take a look as well.
and sorry that issues keep getting sniped lol
| /// Used to recieve any new pending transactions that have been added to the pool after this | ||
| /// iterator was snapshotted | ||
| /// | ||
| /// We prioritize these new transactions over existing ones by first inserting these into the |
There was a problem hiding this comment.
I think prioritize is not correct here, since they get the same treatment as txs already included in this snapshot, it's just an extra data source.
There was a problem hiding this comment.
ah yeah given what priority actually means in the pool data structures - prioritize is the wrong word here.
makes sense, i removed this comment completely.
| /// pool, and then yielding. This breaks the contract that the iterator yields the directly | ||
| /// "next" best transaction from the POV of the consumer, but we assume more often than not | ||
| /// newer transactions are better |
There was a problem hiding this comment.
let's move these docs to the trait docs, that there is no guarantee that the iterator will only ever return txs with decreasing priority, and instead always returns the best transaction that it currently knows
There was a problem hiding this comment.
yeah that makes sense and your wording is clearer - modified and moved to trait docs.
| // check for new txs that have come in after this iterator was snapshotted and add them | ||
| if let Some(pending_tx) = self.try_recv() { | ||
| let tx = pending_tx.transaction.clone(); | ||
| // same logic as PendingPool::add_transaction/PendingPool::best_with_unlocked | ||
| let tx_id = *tx.id(); | ||
| if self.ancestor(&tx_id).is_none() { | ||
| self.independent.insert(pending_tx.clone()); | ||
| } | ||
| self.all.insert(tx_id, pending_tx); | ||
| } |
There was a problem hiding this comment.
I'd like to extract this into a standalone function
| @@ -89,6 +113,16 @@ impl<T: TransactionOrdering> Iterator for BestTransactions<T> { | |||
|
|
|||
| fn next(&mut self) -> Option<Self::Item> { | |||
| loop { | |||
| // check for new txs that have come in after this iterator was snapshotted and add them | |||
| if let Some(pending_tx) = self.try_recv() { | |||
There was a problem hiding this comment.
this could be made into a while loop
| } | ||
|
|
||
| // === impl PendingPool === | ||
|
|
||
| impl<T: TransactionOrdering> PendingPool<T> { | ||
| /// Create a new pool instance. | ||
| pub(crate) fn new(ordering: T) -> Self { | ||
| let (new_transaction_notifier, _) = broadcast::channel(1000); |
There was a problem hiding this comment.
1000 is a bit large, I think 200 would be appropriate
| // send the new transaction to any existing pendingpool snapshot iterators | ||
| // drop the Result value returned as Err effectively means there are no subscribers | ||
| // and Ok just returns the # of subscribed recievers | ||
| let _ = self.new_transaction_notifier.send(tx); |
There was a problem hiding this comment.
we assume the most common case is that there are no receivers in which case we don't need to send anything, so lets move this above the self.by_id.insert and check for receiver_count > 0 this prevents 1 clone
There was a problem hiding this comment.
makes sense, modified
|
looks good to me |
yep makes sense, added in the |
creating a draft pr for #4045
feel free to close this out in favor of @supernovahs solution, since they had asked to be assigned - mainly would love any feedback as to whether my understanding of the issue was correct, and if not, where I'm thinking about this solution wrong