X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning-background-processor%2Fsrc%2Flib.rs;h=b2ee865471c89f3ca1d537088c8e3425c9d4cbf8;hb=e59b3847a363afba5d30e1f1271341e3163f7591;hp=199295b9f0c92e5dd1139480ebeed419ee44a701;hpb=682bb9b0ae4df2cd3042316a9cb99811fbb8595c;p=rust-lightning diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs index 199295b9..b2ee8654 100644 --- a/lightning-background-processor/src/lib.rs +++ b/lightning-background-processor/src/lib.rs @@ -17,7 +17,7 @@ extern crate lightning_rapid_gossip_sync; use lightning::chain; use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator}; use lightning::chain::chainmonitor::{ChainMonitor, Persist}; -use lightning::chain::keysinterface::{KeysInterface, SignerProvider}; +use lightning::chain::keysinterface::{EntropySource, NodeSigner, SignerProvider}; use lightning::ln::channelmanager::ChannelManager; use lightning::ln::msgs::{ChannelMessageHandler, OnionMessageHandler, RoutingMessageHandler}; use lightning::ln::peer_handler::{CustomMessageHandler, PeerManager, SocketDescriptor}; @@ -36,7 +36,7 @@ use std::time::{Duration, Instant}; use std::ops::Deref; #[cfg(feature = "futures")] -use futures_util::{select_biased, future::FutureExt}; +use futures_util::{select_biased, future::FutureExt, task}; /// `BackgroundProcessor` takes care of tasks that (1) need to happen periodically to keep /// Rust-Lightning running properly, and (2) either can or should be run in the background. Its @@ -207,15 +207,15 @@ macro_rules! define_run_body { ($persister: ident, $chain_monitor: ident, $process_chain_monitor_events: expr, $channel_manager: ident, $process_channel_manager_events: expr, $gossip_sync: ident, $peer_manager: ident, $logger: ident, $scorer: ident, - $loop_exit_check: expr, $await: expr) + $loop_exit_check: expr, $await: expr, $get_timer: expr, $timer_elapsed: expr) => { { log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup"); $channel_manager.timer_tick_occurred(); - let mut last_freshness_call = Instant::now(); - let mut last_ping_call = Instant::now(); - let mut last_prune_call = Instant::now(); - let mut last_scorer_persist_call = Instant::now(); + let mut last_freshness_call = $get_timer(FRESHNESS_TIMER); + let mut last_ping_call = $get_timer(PING_TIMER); + let mut last_prune_call = $get_timer(FIRST_NETWORK_PRUNE_TIMER); + let mut last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER); let mut have_pruned = false; loop { @@ -237,9 +237,9 @@ macro_rules! define_run_body { // We wait up to 100ms, but track how long it takes to detect being put to sleep, // see `await_start`'s use below. - let await_start = Instant::now(); + let mut await_start = $get_timer(1); let updates_available = $await; - let await_time = await_start.elapsed(); + let await_slow = $timer_elapsed(&mut await_start, 1); if updates_available { log_trace!($logger, "Persisting ChannelManager..."); @@ -251,12 +251,12 @@ macro_rules! define_run_body { log_trace!($logger, "Terminating background processor."); break; } - if last_freshness_call.elapsed().as_secs() > FRESHNESS_TIMER { + if $timer_elapsed(&mut last_freshness_call, FRESHNESS_TIMER) { log_trace!($logger, "Calling ChannelManager's timer_tick_occurred"); $channel_manager.timer_tick_occurred(); - last_freshness_call = Instant::now(); + last_freshness_call = $get_timer(FRESHNESS_TIMER); } - if await_time > Duration::from_secs(1) { + if await_slow { // On various platforms, we may be starved of CPU cycles for several reasons. // E.g. on iOS, if we've been in the background, we will be entirely paused. // Similarly, if we're on a desktop platform and the device has been asleep, we @@ -271,18 +271,18 @@ macro_rules! define_run_body { // peers. log_trace!($logger, "100ms sleep took more than a second, disconnecting peers."); $peer_manager.disconnect_all_peers(); - last_ping_call = Instant::now(); - } else if last_ping_call.elapsed().as_secs() > PING_TIMER { + last_ping_call = $get_timer(PING_TIMER); + } else if $timer_elapsed(&mut last_ping_call, PING_TIMER) { log_trace!($logger, "Calling PeerManager's timer_tick_occurred"); $peer_manager.timer_tick_occurred(); - last_ping_call = Instant::now(); + last_ping_call = $get_timer(PING_TIMER); } // Note that we want to run a graph prune once not long after startup before // falling back to our usual hourly prunes. This avoids short-lived clients never // pruning their network graph. We run once 60 seconds after startup before // continuing our normal cadence. - if last_prune_call.elapsed().as_secs() > if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER } { + if $timer_elapsed(&mut last_prune_call, if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER }) { // The network graph must not be pruned while rapid sync completion is pending if let Some(network_graph) = $gossip_sync.prunable_network_graph() { log_trace!($logger, "Pruning and persisting network graph."); @@ -292,19 +292,19 @@ macro_rules! define_run_body { log_error!($logger, "Error: Failed to persist network graph, check your disk and permissions {}", e) } - last_prune_call = Instant::now(); + last_prune_call = $get_timer(NETWORK_PRUNE_TIMER); have_pruned = true; } } - if last_scorer_persist_call.elapsed().as_secs() > SCORER_PERSIST_TIMER { + if $timer_elapsed(&mut last_scorer_persist_call, SCORER_PERSIST_TIMER) { if let Some(ref scorer) = $scorer { log_trace!($logger, "Persisting scorer"); if let Err(e) = $persister.persist_scorer(&scorer) { log_error!($logger, "Error: Failed to persist scorer, check your disk and permissions {}", e) } } - last_scorer_persist_call = Instant::now(); + last_scorer_persist_call = $get_timer(SCORER_PERSIST_TIMER); } } @@ -341,7 +341,9 @@ pub async fn process_events_async< CF: 'static + Deref + Send + Sync, CW: 'static + Deref + Send + Sync, T: 'static + Deref + Send + Sync, - K: 'static + Deref + Send + Sync, + ES: 'static + Deref + Send + Sync, + NS: 'static + Deref + Send + Sync, + SP: 'static + Deref + Send + Sync, F: 'static + Deref + Send + Sync, R: 'static + Deref + Send + Sync, G: 'static + Deref> + Send + Sync, @@ -354,15 +356,15 @@ pub async fn process_events_async< EventHandlerFuture: core::future::Future, EventHandler: Fn(Event) -> EventHandlerFuture, PS: 'static + Deref + Send, - M: 'static + Deref::Signer, CF, T, F, L, P>> + Send + Sync, - CM: 'static + Deref> + Send + Sync, + M: 'static + Deref::Signer, CF, T, F, L, P>> + Send + Sync, + CM: 'static + Deref> + Send + Sync, PGS: 'static + Deref> + Send + Sync, RGS: 'static + Deref> + Send, UMH: 'static + Deref + Send + Sync, PM: 'static + Deref> + Send + Sync, S: 'static + Deref + Send + Sync, SC: WriteableScore<'a>, - SleepFuture: core::future::Future, + SleepFuture: core::future::Future + core::marker::Unpin, Sleeper: Fn(Duration) -> SleepFuture >( persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM, @@ -372,18 +374,20 @@ pub async fn process_events_async< where CA::Target: 'static + chain::Access, CF::Target: 'static + chain::Filter, - CW::Target: 'static + chain::Watch<::Signer>, + CW::Target: 'static + chain::Watch<::Signer>, T::Target: 'static + BroadcasterInterface, - K::Target: 'static + KeysInterface, + ES::Target: 'static + EntropySource, + NS::Target: 'static + NodeSigner, + SP::Target: 'static + SignerProvider, F::Target: 'static + FeeEstimator, R::Target: 'static + Router, L::Target: 'static + Logger, - P::Target: 'static + Persist<::Signer>, + P::Target: 'static + Persist<::Signer>, CMH::Target: 'static + ChannelMessageHandler, OMH::Target: 'static + OnionMessageHandler, RMH::Target: 'static + RoutingMessageHandler, UMH::Target: 'static + CustomMessageHandler, - PS::Target: 'static + Persister<'a, CW, T, K, F, R, L, SC>, + PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>, { let mut should_break = true; let async_event_handler = |event| { @@ -407,6 +411,11 @@ where false } } + }, |t| sleeper(Duration::from_secs(t)), + |fut: &mut SleepFuture, _| { + let mut waker = task::noop_waker(); + let mut ctx = task::Context::from_waker(&mut waker); + core::pin::Pin::new(fut).poll(&mut ctx).is_ready() }) } @@ -461,7 +470,9 @@ impl BackgroundProcessor { CF: 'static + Deref + Send + Sync, CW: 'static + Deref + Send + Sync, T: 'static + Deref + Send + Sync, - K: 'static + Deref + Send + Sync, + ES: 'static + Deref + Send + Sync, + NS: 'static + Deref + Send + Sync, + SP: 'static + Deref + Send + Sync, F: 'static + Deref + Send + Sync, R: 'static + Deref + Send + Sync, G: 'static + Deref> + Send + Sync, @@ -473,8 +484,8 @@ impl BackgroundProcessor { RMH: 'static + Deref + Send + Sync, EH: 'static + EventHandler + Send, PS: 'static + Deref + Send, - M: 'static + Deref::Signer, CF, T, F, L, P>> + Send + Sync, - CM: 'static + Deref> + Send + Sync, + M: 'static + Deref::Signer, CF, T, F, L, P>> + Send + Sync, + CM: 'static + Deref> + Send + Sync, PGS: 'static + Deref> + Send + Sync, RGS: 'static + Deref> + Send, UMH: 'static + Deref + Send + Sync, @@ -488,18 +499,20 @@ impl BackgroundProcessor { where CA::Target: 'static + chain::Access, CF::Target: 'static + chain::Filter, - CW::Target: 'static + chain::Watch<::Signer>, + CW::Target: 'static + chain::Watch<::Signer>, T::Target: 'static + BroadcasterInterface, - K::Target: 'static + KeysInterface, + ES::Target: 'static + EntropySource, + NS::Target: 'static + NodeSigner, + SP::Target: 'static + SignerProvider, F::Target: 'static + FeeEstimator, R::Target: 'static + Router, L::Target: 'static + Logger, - P::Target: 'static + Persist<::Signer>, + P::Target: 'static + Persist<::Signer>, CMH::Target: 'static + ChannelMessageHandler, OMH::Target: 'static + OnionMessageHandler, RMH::Target: 'static + RoutingMessageHandler, UMH::Target: 'static + CustomMessageHandler, - PS::Target: 'static + Persister<'a, CW, T, K, F, R, L, SC>, + PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>, { let stop_thread = Arc::new(AtomicBool::new(false)); let stop_thread_clone = stop_thread.clone(); @@ -514,7 +527,8 @@ impl BackgroundProcessor { define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler), channel_manager, channel_manager.process_pending_events(&event_handler), gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire), - channel_manager.await_persistable_update_timeout(Duration::from_millis(100))) + channel_manager.await_persistable_update_timeout(Duration::from_millis(100)), + |_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur) }); Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) } } @@ -575,10 +589,10 @@ mod tests { use bitcoin::network::constants::Network; use lightning::chain::{BestBlock, Confirm, chainmonitor}; use lightning::chain::channelmonitor::ANTI_REORG_DELAY; - use lightning::chain::keysinterface::{InMemorySigner, Recipient, EntropySource, KeysInterface, KeysManager, NodeSigner}; + use lightning::chain::keysinterface::{InMemorySigner, Recipient, EntropySource, KeysManager, NodeSigner}; use lightning::chain::transaction::OutPoint; use lightning::get_event_msg; - use lightning::ln::channelmanager::{self, BREAKDOWN_TIMEOUT, ChainParameters, ChannelManager, SimpleArcChannelManager}; + use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, ChannelManager, SimpleArcChannelManager}; use lightning::ln::features::ChannelFeatures; use lightning::ln::msgs::{ChannelMessageHandler, Init}; use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler}; @@ -742,7 +756,7 @@ mod tests { let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), persister.clone())); let best_block = BestBlock::from_genesis(network); let params = ChainParameters { network, best_block }; - let manager = Arc::new(ChannelManager::new(fee_estimator.clone(), chain_monitor.clone(), tx_broadcaster.clone(), router.clone(), logger.clone(), keys_manager.clone(), UserConfig::default(), params)); + let manager = Arc::new(ChannelManager::new(fee_estimator.clone(), chain_monitor.clone(), tx_broadcaster.clone(), router.clone(), logger.clone(), keys_manager.clone(), keys_manager.clone(), keys_manager.clone(), UserConfig::default(), params)); let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone())); let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone())); let msg_handler = MessageHandler { chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new()), route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()), onion_message_handler: IgnoringMessageHandler{}}; @@ -753,8 +767,8 @@ mod tests { for i in 0..num_nodes { for j in (i+1)..num_nodes { - nodes[i].node.peer_connected(&nodes[j].node.get_our_node_id(), &Init { features: channelmanager::provided_init_features(), remote_network_address: None }).unwrap(); - nodes[j].node.peer_connected(&nodes[i].node.get_our_node_id(), &Init { features: channelmanager::provided_init_features(), remote_network_address: None }).unwrap(); + nodes[i].node.peer_connected(&nodes[j].node.get_our_node_id(), &Init { features: nodes[j].node.init_features(), remote_network_address: None }).unwrap(); + nodes[j].node.peer_connected(&nodes[i].node.get_our_node_id(), &Init { features: nodes[i].node.init_features(), remote_network_address: None }).unwrap(); } } @@ -775,8 +789,8 @@ mod tests { macro_rules! begin_open_channel { ($node_a: expr, $node_b: expr, $channel_value: expr) => {{ $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap(); - $node_b.node.handle_open_channel(&$node_a.node.get_our_node_id(), channelmanager::provided_init_features(), &get_event_msg!($node_a, MessageSendEvent::SendOpenChannel, $node_b.node.get_our_node_id())); - $node_a.node.handle_accept_channel(&$node_b.node.get_our_node_id(), channelmanager::provided_init_features(), &get_event_msg!($node_b, MessageSendEvent::SendAcceptChannel, $node_a.node.get_our_node_id())); + $node_b.node.handle_open_channel(&$node_a.node.get_our_node_id(), $node_a.node.init_features(), &get_event_msg!($node_a, MessageSendEvent::SendOpenChannel, $node_b.node.get_our_node_id())); + $node_a.node.handle_accept_channel(&$node_b.node.get_our_node_id(), $node_b.node.init_features(), &get_event_msg!($node_b, MessageSendEvent::SendAcceptChannel, $node_a.node.get_our_node_id())); }} }