/// for unilateral chain closure fees are at risk.
pub struct BackgroundProcessor {
stop_thread: Arc<AtomicBool>,
- /// May be used to retrieve and handle the error if `BackgroundProcessor`'s thread
- /// exits due to an error while persisting.
- pub thread_handle: JoinHandle<Result<(), std::io::Error>>,
+ thread_handle: Option<JoinHandle<Result<(), std::io::Error>>>,
}
#[cfg(not(test))]
#[cfg(test)]
const FRESHNESS_TIMER: u64 = 1;
+const PING_TIMER: u64 = 5;
+
/// Trait which handles persisting a [`ChannelManager`] to disk.
///
/// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
}
impl BackgroundProcessor {
- /// Start a background thread that takes care of responsibilities enumerated in the top-level
- /// documentation.
+ /// Start a background thread that takes care of responsibilities enumerated in the [top-level
+ /// documentation].
+ ///
+ /// The thread runs indefinitely unless the object is dropped, [`stop`] is called, or
+ /// `persist_manager` returns an error. In case of an error, the error is retrieved by calling
+ /// either [`join`] or [`stop`].
///
- /// If `persist_manager` returns an error, then this thread will return said error (and
- /// `start()` will need to be called again to restart the `BackgroundProcessor`). Users should
- /// wait on [`thread_handle`]'s `join()` method to be able to tell if and when an error is
- /// returned, or implement `persist_manager` such that an error is never returned to the
- /// `BackgroundProcessor`
+ /// Typically, users should either implement [`ChannelManagerPersister`] to never return an
+ /// error or call [`join`] and handle any error that may arise. For the latter case, the
+ /// `BackgroundProcessor` must be restarted by calling `start` again after handling the error.
///
/// `persist_manager` is responsible for writing out the [`ChannelManager`] to disk, and/or
/// uploading to one or more backup services. See [`ChannelManager::write`] for writing out a
/// [`ChannelManager`]. See [`FilesystemPersister::persist_manager`] for Rust-Lightning's
/// provided implementation.
///
- /// [`thread_handle`]: BackgroundProcessor::thread_handle
+ /// [top-level documentation]: Self
+ /// [`join`]: Self::join
+ /// [`stop`]: Self::stop
/// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
/// [`ChannelManager::write`]: lightning::ln::channelmanager::ChannelManager#impl-Writeable
/// [`FilesystemPersister::persist_manager`]: lightning_persister::FilesystemPersister::persist_manager
let stop_thread = Arc::new(AtomicBool::new(false));
let stop_thread_clone = stop_thread.clone();
let handle = thread::spawn(move || -> Result<(), std::io::Error> {
- let mut current_time = Instant::now();
+ let mut last_freshness_call = Instant::now();
+ let mut last_ping_call = Instant::now();
loop {
peer_manager.process_events();
channel_manager.process_pending_events(&event_handler);
log_trace!(logger, "Terminating background processor.");
return Ok(());
}
- if current_time.elapsed().as_secs() > FRESHNESS_TIMER {
- log_trace!(logger, "Calling ChannelManager's and PeerManager's timer_tick_occurred");
+ if last_freshness_call.elapsed().as_secs() > FRESHNESS_TIMER {
+ log_trace!(logger, "Calling ChannelManager's timer_tick_occurred");
channel_manager.timer_tick_occurred();
+ last_freshness_call = Instant::now();
+ }
+ if last_ping_call.elapsed().as_secs() > PING_TIMER {
+ log_trace!(logger, "Calling PeerManager's timer_tick_occurred");
peer_manager.timer_tick_occurred();
- current_time = Instant::now();
+ last_ping_call = Instant::now();
}
}
});
- Self { stop_thread: stop_thread_clone, thread_handle: handle }
+ Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }
}
- /// Stop `BackgroundProcessor`'s thread.
- pub fn stop(self) -> Result<(), std::io::Error> {
+ /// Join `BackgroundProcessor`'s thread, returning any error that occurred while persisting
+ /// [`ChannelManager`].
+ ///
+ /// # Panics
+ ///
+ /// This function panics if the background thread has panicked such as while persisting or
+ /// handling events.
+ ///
+ /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
+ pub fn join(mut self) -> Result<(), std::io::Error> {
+ assert!(self.thread_handle.is_some());
+ self.join_thread()
+ }
+
+ /// Stop `BackgroundProcessor`'s thread, returning any error that occurred while persisting
+ /// [`ChannelManager`].
+ ///
+ /// # Panics
+ ///
+ /// This function panics if the background thread has panicked such as while persisting or
+ /// handling events.
+ ///
+ /// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
+ pub fn stop(mut self) -> Result<(), std::io::Error> {
+ assert!(self.thread_handle.is_some());
+ self.stop_and_join_thread()
+ }
+
+ fn stop_and_join_thread(&mut self) -> Result<(), std::io::Error> {
self.stop_thread.store(true, Ordering::Release);
- self.thread_handle.join().unwrap()
+ self.join_thread()
+ }
+
+ fn join_thread(&mut self) -> Result<(), std::io::Error> {
+ match self.thread_handle.take() {
+ Some(handle) => handle.join().unwrap(),
+ None => Ok(()),
+ }
+ }
+}
+
+impl Drop for BackgroundProcessor {
+ fn drop(&mut self) {
+ self.stop_and_join_thread().unwrap();
}
}
let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].peer_manager.clone(), nodes[0].logger.clone());
loop {
let log_entries = nodes[0].logger.lines.lock().unwrap();
- let desired_log = "Calling ChannelManager's and PeerManager's timer_tick_occurred".to_string();
- if log_entries.get(&("lightning_background_processor".to_string(), desired_log)).is_some() {
+ let desired_log = "Calling ChannelManager's timer_tick_occurred".to_string();
+ let second_desired_log = "Calling PeerManager's timer_tick_occurred".to_string();
+ if log_entries.get(&("lightning_background_processor".to_string(), desired_log)).is_some() &&
+ log_entries.get(&("lightning_background_processor".to_string(), second_desired_log)).is_some() {
break
}
}
let persister = |_: &_| Err(std::io::Error::new(std::io::ErrorKind::Other, "test"));
let event_handler = |_| {};
let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].peer_manager.clone(), nodes[0].logger.clone());
- match bg_processor.stop() {
+ match bg_processor.join() {
Ok(_) => panic!("Expected error persisting manager"),
Err(e) => {
assert_eq!(e.kind(), std::io::ErrorKind::Other);
projects / rust-lightning / blobdiff