jleony/nanoreth
1
0
Fork 0
mirror of https://github.com/hl-archive-node/nanoreth.git synced 2025-12-06 10:59:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: dont await changes (#7760)

Browse Source
This commit is contained in:
Matthias Seitz
2024-04-19 23:18:33 +02:00
committed by GitHub
parent 16b747e20c
commit d599ce4683
1 changed files with 12 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
crates/exex/src/manager.rs
View File

@ -1,3 +1,9 @@
use crate::ExExEvent;
use metrics::Gauge;
use reth_metrics::{metrics::Counter, Metrics};
use reth_primitives::{BlockNumber, FinishedExExHeight};
use reth_provider::CanonStateNotification;
use reth_tracing::tracing::debug;
use std::{
collections::VecDeque,
future::{poll_fn, Future},
@ -8,14 +14,6 @@ use std::{
},
task::{Context, Poll},
};
use crate::ExExEvent;
use futures::StreamExt;
use metrics::Gauge;
use reth_metrics::{metrics::Counter, Metrics};
use reth_primitives::{BlockNumber, FinishedExExHeight};
use reth_provider::CanonStateNotification;
use reth_tracing::tracing::debug;
use tokio::sync::{
mpsc::{self, error::SendError, Receiver, UnboundedReceiver, UnboundedSender},
watch,
@ -347,6 +345,7 @@ pub struct ExExManagerHandle {
/// otherwise unused.
is_ready_receiver: watch::Receiver<bool>,
/// A stream of bools denoting whether the manager is ready for new notifications.
#[allow(unused)]
is_ready: WatchStream<bool>,
/// The current capacity of the manager's internal notification buffer.
current_capacity: Arc<AtomicUsize>,
@ -426,15 +425,14 @@ impl ExExManagerHandle {
}
/// Wait until the manager is ready for new notifications.
#[allow(clippy::needless_pass_by_ref_mut)]
pub fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<()> {
// if this returns `Poll::Ready(None)` the stream is exhausted, which means the underlying
// channel is closed.
//
// this can only happen if the manager died, and the node is shutting down, so we ignore it
let mut pinned = std::pin::pin!(&mut self.is_ready);
if pinned.poll_next_unpin(cx) == Poll::Ready(Some(true)) {
use futures as _;
// FIXME: if not ready this must be polled
if *self.is_ready_receiver.borrow() {
Poll::Ready(())
} else {
cx.waker().wake_by_ref();
Poll::Pending
}
}