use lightning::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler, NetAddress};
use lightning::util::logger::Logger;
+use std::ops::Deref;
use std::task;
use std::net::SocketAddr;
use std::net::TcpStream as StdTcpStream;
id: u64,
}
impl Connection {
- async fn schedule_read<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, us: Arc<Mutex<Self>>, mut reader: io::ReadHalf<TcpStream>, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>) where
- CMH: ChannelMessageHandler + 'static,
- RMH: RoutingMessageHandler + 'static,
- L: Logger + 'static + ?Sized,
- UMH: CustomMessageHandler + 'static {
+ async fn poll_event_process<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, CMH, RMH, L, UMH>>, mut event_receiver: mpsc::Receiver<()>) where
+ CMH: Deref + 'static + Send + Sync,
+ RMH: Deref + 'static + Send + Sync,
+ L: Deref + 'static + Send + Sync,
+ UMH: Deref + 'static + Send + Sync,
+ CMH::Target: ChannelMessageHandler + Send + Sync,
+ RMH::Target: RoutingMessageHandler + Send + Sync,
+ L::Target: Logger + Send + Sync,
+ UMH::Target: CustomMessageHandler + Send + Sync,
+ {
+ loop {
+ if event_receiver.recv().await.is_none() {
+ return;
+ }
+ peer_manager.process_events();
+ }
+ }
+
+ async fn schedule_read<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, CMH, RMH, L, UMH>>, us: Arc<Mutex<Self>>, mut reader: io::ReadHalf<TcpStream>, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>) where
+ CMH: Deref + 'static + Send + Sync,
+ RMH: Deref + 'static + Send + Sync,
+ L: Deref + 'static + Send + Sync,
+ UMH: Deref + 'static + Send + Sync,
+ CMH::Target: ChannelMessageHandler + 'static + Send + Sync,
+ RMH::Target: RoutingMessageHandler + 'static + Send + Sync,
+ L::Target: Logger + 'static + Send + Sync,
+ UMH::Target: CustomMessageHandler + 'static + Send + Sync,
+ {
+ // Create a waker to wake up poll_event_process, above
+ let (event_waker, event_receiver) = mpsc::channel(1);
+ tokio::spawn(Self::poll_event_process(Arc::clone(&peer_manager), event_receiver));
+
// 8KB is nice and big but also should never cause any issues with stack overflowing.
let mut buf = [0; 8192];
Err(_) => break Disconnect::PeerDisconnected,
},
}
- peer_manager.process_events();
+ let _ = event_waker.try_send(());
+
+ // At this point we've processed a message or two, and reset the ping timer for this
+ // peer, at least in the "are we still receiving messages" context, if we don't give up
+ // our timeslice to another task we may just spin on this peer, starving other peers
+ // and eventually disconnecting them for ping timeouts. Instead, we explicitly yield
+ // here.
+ tokio::task::yield_now().await;
};
let writer_option = us.lock().unwrap().writer.take();
if let Some(mut writer) = writer_option {
/// The returned future will complete when the peer is disconnected and associated handling
/// futures are freed, though, because all processing futures are spawned with tokio::spawn, you do
/// not need to poll the provided future in order to make progress.
-pub fn setup_inbound<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
- CMH: ChannelMessageHandler + 'static + Send + Sync,
- RMH: RoutingMessageHandler + 'static + Send + Sync,
- L: Logger + 'static + ?Sized + Send + Sync,
- UMH: CustomMessageHandler + 'static + Send + Sync {
+pub fn setup_inbound<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, CMH, RMH, L, UMH>>, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
+ CMH: Deref + 'static + Send + Sync,
+ RMH: Deref + 'static + Send + Sync,
+ L: Deref + 'static + Send + Sync,
+ UMH: Deref + 'static + Send + Sync,
+ CMH::Target: ChannelMessageHandler + Send + Sync,
+ RMH::Target: RoutingMessageHandler + Send + Sync,
+ L::Target: Logger + Send + Sync,
+ UMH::Target: CustomMessageHandler + Send + Sync,
+{
let remote_addr = get_addr_from_stream(&stream);
let (reader, write_receiver, read_receiver, us) = Connection::new(stream);
#[cfg(debug_assertions)]
/// The returned future will complete when the peer is disconnected and associated handling
/// futures are freed, though, because all processing futures are spawned with tokio::spawn, you do
/// not need to poll the provided future in order to make progress.
-pub fn setup_outbound<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, their_node_id: PublicKey, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
- CMH: ChannelMessageHandler + 'static + Send + Sync,
- RMH: RoutingMessageHandler + 'static + Send + Sync,
- L: Logger + 'static + ?Sized + Send + Sync,
- UMH: CustomMessageHandler + 'static + Send + Sync {
+pub fn setup_outbound<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, CMH, RMH, L, UMH>>, their_node_id: PublicKey, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
+ CMH: Deref + 'static + Send + Sync,
+ RMH: Deref + 'static + Send + Sync,
+ L: Deref + 'static + Send + Sync,
+ UMH: Deref + 'static + Send + Sync,
+ CMH::Target: ChannelMessageHandler + Send + Sync,
+ RMH::Target: RoutingMessageHandler + Send + Sync,
+ L::Target: Logger + Send + Sync,
+ UMH::Target: CustomMessageHandler + Send + Sync,
+{
let remote_addr = get_addr_from_stream(&stream);
let (reader, mut write_receiver, read_receiver, us) = Connection::new(stream);
#[cfg(debug_assertions)]
/// disconnected and associated handling futures are freed, though, because all processing in said
/// futures are spawned with tokio::spawn, you do not need to poll the second future in order to
/// make progress.
-pub async fn connect_outbound<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, their_node_id: PublicKey, addr: SocketAddr) -> Option<impl std::future::Future<Output=()>> where
- CMH: ChannelMessageHandler + 'static + Send + Sync,
- RMH: RoutingMessageHandler + 'static + Send + Sync,
- L: Logger + 'static + ?Sized + Send + Sync,
- UMH: CustomMessageHandler + 'static + Send + Sync {
+pub async fn connect_outbound<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, CMH, RMH, L, UMH>>, their_node_id: PublicKey, addr: SocketAddr) -> Option<impl std::future::Future<Output=()>> where
+ CMH: Deref + 'static + Send + Sync,
+ RMH: Deref + 'static + Send + Sync,
+ L: Deref + 'static + Send + Sync,
+ UMH: Deref + 'static + Send + Sync,
+ CMH::Target: ChannelMessageHandler + Send + Sync,
+ RMH::Target: RoutingMessageHandler + Send + Sync,
+ L::Target: Logger + Send + Sync,
+ UMH::Target: CustomMessageHandler + Send + Sync,
+{
if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), async { TcpStream::connect(&addr).await.map(|s| s.into_std().unwrap()) }).await {
Some(setup_outbound(peer_manager, their_node_id, stream))
} else { None }
// pause read given we're now waiting on the remote end to ACK (and in
// accordance with the send_data() docs).
us.read_paused = true;
+ // Further, to avoid any current pending read causing a `read_event` call, wake
+ // up the read_waker and restart its loop.
+ let _ = us.read_waker.try_send(());
return written_len;
},
}
fn handle_accept_channel(&self, _their_node_id: &PublicKey, _their_features: InitFeatures, _msg: &AcceptChannel) {}
fn handle_funding_created(&self, _their_node_id: &PublicKey, _msg: &FundingCreated) {}
fn handle_funding_signed(&self, _their_node_id: &PublicKey, _msg: &FundingSigned) {}
- fn handle_funding_locked(&self, _their_node_id: &PublicKey, _msg: &FundingLocked) {}
+ fn handle_channel_ready(&self, _their_node_id: &PublicKey, _msg: &ChannelReady) {}
fn handle_shutdown(&self, _their_node_id: &PublicKey, _their_features: &InitFeatures, _msg: &Shutdown) {}
fn handle_closing_signed(&self, _their_node_id: &PublicKey, _msg: &ClosingSigned) {}
fn handle_update_add_htlc(&self, _their_node_id: &PublicKey, _msg: &UpdateAddHTLC) {}
projects / rust-lightning / blobdiff