-
Notifications
You must be signed in to change notification settings - Fork 58
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge #1010: Update our state upon broadcasting a transaction
58c71c7 lib: gate the RPC server availability on the 'daemon' feature (Antoine Poinsot) b7fde6a commands: update our state immediately after broadcasting a tx (Antoine Poinsot) 1cf42d9 poller: introduce a communication channel with the poller thread (Antoine Poinsot) f6ce85c lib: remove the panic hook. (Antoine Poinsot) b4fe963 lib: encapsulate the handling of both threads (poller and RPC server) (Antoine Poinsot) fd5387f poller: use the same database connection across one update round (Antoine Poinsot) ea6923e poller: make the updating process into its own function. (Antoine Poinsot) Pull request description: Fixes #887. This takes a couple commits from #909 but takes the approach from there in another direction: we don't externalize the poller, since only a single instance must be ran. Instead we properly keep track of the (up to) two threads we manage in the `DaemonHandle` and provide a way for a user of the library to check for errors in any of the threads. This approach allows us to 1) communicate with the poller thread from inside the Liana library/daemon (here we leverage this to tell it to poll) 2) eventually (#909) expose all internal errors from the library to the user instead of panic'ing internally. See the commit messages for details. ACKs for top commit: darosior: ACK 58c71c7 -- did another pass and Edouard tested this in the GUI. Tree-SHA512: 0ab436b2a187f9d124ed8861a47f03bb1e9252cdc4f3b5c4308db07be738c78b2ea3f07dc0a9586e3d5bd34f071a1e2a2569cad30676c9cc004e39260ebb94ca
- Loading branch information
Showing
7 changed files
with
359 additions
and
237 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,60 +1,128 @@ | ||
| mod looper; | ||
|
|
||
| use crate::{ | ||
| bitcoin::{poller::looper::looper, BitcoinInterface}, | ||
| database::DatabaseInterface, | ||
| descriptors, | ||
| }; | ||
| use crate::{bitcoin::BitcoinInterface, database::DatabaseInterface, descriptors}; | ||
|
|
||
| use std::{ | ||
| sync::{self, atomic}, | ||
| thread, time, | ||
| sync::{self, mpsc}, | ||
| time, | ||
| }; | ||
|
|
||
| use miniscript::bitcoin::secp256k1; | ||
|
|
||
| #[derive(Debug, Clone)] | ||
| pub enum PollerMessage { | ||
| Shutdown, | ||
| /// Ask the Bitcoin poller to poll immediately, get notified through the passed channel once | ||
| /// it's done. | ||
| PollNow(mpsc::SyncSender<()>), | ||
| } | ||
|
|
||
| /// The Bitcoin poller handler. | ||
| pub struct Poller { | ||
| handle: thread::JoinHandle<()>, | ||
| shutdown: sync::Arc<atomic::AtomicBool>, | ||
| bit: sync::Arc<sync::Mutex<dyn BitcoinInterface>>, | ||
| db: sync::Arc<sync::Mutex<dyn DatabaseInterface>>, | ||
| secp: secp256k1::Secp256k1<secp256k1::VerifyOnly>, | ||
| // The receive and change descriptors (in this order). | ||
| descs: [descriptors::SinglePathLianaDesc; 2], | ||
| } | ||
|
|
||
| impl Poller { | ||
| pub fn start( | ||
| pub fn new( | ||
| bit: sync::Arc<sync::Mutex<dyn BitcoinInterface>>, | ||
| db: sync::Arc<sync::Mutex<dyn DatabaseInterface>>, | ||
| poll_interval: time::Duration, | ||
| desc: descriptors::LianaDescriptor, | ||
| ) -> Poller { | ||
| let shutdown = sync::Arc::from(atomic::AtomicBool::from(false)); | ||
| let handle = thread::Builder::new() | ||
| .name("Bitcoin poller".to_string()) | ||
| .spawn({ | ||
| let shutdown = shutdown.clone(); | ||
| move || looper(bit, db, shutdown, poll_interval, desc) | ||
| }) | ||
| .expect("Must not fail"); | ||
|
|
||
| Poller { shutdown, handle } | ||
| } | ||
| let secp = secp256k1::Secp256k1::verification_only(); | ||
| let descs = [ | ||
| desc.receive_descriptor().clone(), | ||
| desc.change_descriptor().clone(), | ||
| ]; | ||
|
|
||
| pub fn trigger_stop(&self) { | ||
| self.shutdown.store(true, atomic::Ordering::Relaxed); | ||
| } | ||
| // On first startup the tip may be NULL. Make sure it's set as the poller relies on it. | ||
| looper::maybe_initialize_tip(&bit, &db); | ||
|
|
||
| pub fn stop(self) { | ||
| self.trigger_stop(); | ||
| self.handle.join().expect("The poller loop must not fail"); | ||
| Poller { | ||
| bit, | ||
| db, | ||
| secp, | ||
| descs, | ||
| } | ||
| } | ||
|
|
||
| #[cfg(feature = "nonblocking_shutdown")] | ||
| pub fn is_stopped(&self) -> bool { | ||
| // Doc says "This might return true for a brief moment after the thread’s main function has | ||
| // returned, but before the thread itself has stopped running.". But it's not an issue for | ||
| // us, as long as the main poller function has returned we are good. | ||
| self.handle.is_finished() | ||
| } | ||
| /// Continuously update our state from the Bitcoin backend. | ||
| /// - `poll_interval`: how frequently to perform an update. | ||
| /// - `shutdown`: set to true to stop continuously updating and make this function return. | ||
| /// | ||
| /// Typically this would run for the whole duration of the program in a thread, and the main | ||
| /// thread would set the `shutdown` atomic to `true` when shutting down. | ||
| pub fn poll_forever( | ||
| &self, | ||
| poll_interval: time::Duration, | ||
| receiver: mpsc::Receiver<PollerMessage>, | ||
| ) { | ||
| let mut last_poll = None; | ||
| let mut synced = false; | ||
|
|
||
| loop { | ||
| // How long to wait before the next poll. | ||
| let time_before_poll = if let Some(last_poll) = last_poll { | ||
| let time_since_poll = time::Instant::now().duration_since(last_poll); | ||
| // Until we are synced we poll less often to avoid harassing bitcoind and impeding | ||
| // the sync. As a function since it's mocked for the tests. | ||
| let poll_interval = if synced { | ||
| poll_interval | ||
| } else { | ||
| looper::sync_poll_interval() | ||
| }; | ||
| poll_interval.saturating_sub(time_since_poll) | ||
| } else { | ||
| // Don't wait before doing the first poll. | ||
| time::Duration::ZERO | ||
| }; | ||
|
|
||
| // Wait for the duration of the interval between polls, but listen to messages in the | ||
| // meantime. | ||
| match receiver.recv_timeout(time_before_poll) { | ||
| Ok(PollerMessage::Shutdown) => { | ||
| log::info!("Bitcoin poller was told to shut down."); | ||
| return; | ||
| } | ||
| Ok(PollerMessage::PollNow(sender)) => { | ||
| // We've been asked to poll, don't wait any further and signal completion to | ||
| // the caller. | ||
| last_poll = Some(time::Instant::now()); | ||
| looper::poll(&self.bit, &self.db, &self.secp, &self.descs); | ||
| if let Err(e) = sender.send(()) { | ||
| log::error!("Error sending immediate poll completion signal: {}.", e); | ||
| } | ||
| continue; | ||
| } | ||
| Err(mpsc::RecvTimeoutError::Timeout) => { | ||
| // It's been long enough since the last poll. | ||
| } | ||
| Err(mpsc::RecvTimeoutError::Disconnected) => { | ||
| log::error!("Bitcoin poller communication channel got disconnected. Exiting."); | ||
| return; | ||
| } | ||
| } | ||
| last_poll = Some(time::Instant::now()); | ||
|
|
||
| // Don't poll until the Bitcoin backend is fully synced. | ||
| if !synced { | ||
| let progress = self.bit.sync_progress(); | ||
| log::info!( | ||
| "Block chain synchronization progress: {:.2}% ({} blocks / {} headers)", | ||
| progress.rounded_up_progress() * 100.0, | ||
| progress.blocks, | ||
| progress.headers | ||
| ); | ||
| synced = progress.is_complete(); | ||
| if !synced { | ||
| continue; | ||
| } | ||
| } | ||
|
|
||
| #[cfg(test)] | ||
| pub fn test_stop(&mut self) { | ||
| self.shutdown.store(true, atomic::Ordering::Relaxed); | ||
| looper::poll(&self.bit, &self.db, &self.secp, &self.descs); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.