Remove Runtime object and cleanup runtime handling

So far we instantiated a dedicated `Runtime` object which also held the
BP. This was clunky and somewhat error-prone under different
circumstances. As we now got rid of the BP object, we here remove the
dedicated `Runtime` object and just pass around an
`Arc<RwLock<Option<Runtime>>>` upon init, which allows us to get rid of
the `set`/`drop` methods on `Wallet` as an added benefit.

Author: tnull

src/event.rs

@@ -22,7 +22,7 @@ use bitcoin::secp256k1::Secp256k1;
 use rand::{thread_rng, Rng};
 use std::collections::VecDeque;
 use std::ops::Deref;
-use std::sync::{Arc, Condvar, Mutex};
+use std::sync::{Arc, Condvar, Mutex, RwLock};
 use std::time::Duration;
 
 /// An event emitted by [`Node`], which should be handled by the user.
@@ -226,7 +226,7 @@ where
 	network_graph: Arc<NetworkGraph>,
 	keys_manager: Arc<KeysManager>,
 	payment_store: Arc<PaymentStore<K, L>>,
-	tokio_runtime: Arc<tokio::runtime::Runtime>,
+	runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>,
 	logger: L,
 	_config: Arc<Config>,
 }
@@ -240,7 +240,7 @@ where
 		wallet: Arc<Wallet<bdk::database::SqliteDatabase>>, event_queue: Arc<EventQueue<K, L>>,
 		channel_manager: Arc<ChannelManager>, network_graph: Arc<NetworkGraph>,
 		keys_manager: Arc<KeysManager>, payment_store: Arc<PaymentStore<K, L>>,
-		tokio_runtime: Arc<tokio::runtime::Runtime>, logger: L, _config: Arc<Config>,
+		runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>, logger: L, _config: Arc<Config>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -250,7 +250,7 @@ where
 			keys_manager,
 			payment_store,
 			logger,
-			tokio_runtime,
+			runtime,
 			_config,
 		}
 	}
@@ -509,12 +509,17 @@ where
 				let forwarding_channel_manager = self.channel_manager.clone();
 				let min = time_forwardable.as_millis() as u64;
 
-				self.tokio_runtime.spawn(async move {
-					let millis_to_sleep = thread_rng().gen_range(min..min * 5) as u64;
-					tokio::time::sleep(Duration::from_millis(millis_to_sleep)).await;
+				let runtime_lock = self.runtime.read().unwrap();
+				debug_assert!(runtime_lock.is_some());
 
-					forwarding_channel_manager.process_pending_htlc_forwards();
-				});
+				if let Some(runtime) = runtime_lock.as_ref() {
+					runtime.spawn(async move {
+						let millis_to_sleep = thread_rng().gen_range(min..min * 5) as u64;
+						tokio::time::sleep(Duration::from_millis(millis_to_sleep)).await;
+
+						forwarding_channel_manager.process_pending_htlc_forwards();
+					});
+				}
 			}
 			LdkEvent::SpendableOutputs { outputs } => {
 				// TODO: We should eventually remember the outputs and supply them to the wallet's coin selection, once BDK allows us to do so.

src/lib.rs

@@ -329,7 +329,13 @@ impl Builder {
 			EsploraBlockchain::from_client(tx_sync.client().clone(), BDK_CLIENT_STOP_GAP)
 				.with_concurrency(BDK_CLIENT_CONCURRENCY);
 
-		let wallet = Arc::new(Wallet::new(blockchain, bdk_wallet, Arc::clone(&logger)));
+		let runtime = Arc::new(RwLock::new(None));
+		let wallet = Arc::new(Wallet::new(
+			blockchain,
+			bdk_wallet,
+			Arc::clone(&runtime),
+			Arc::clone(&logger),
+		));
 
 		let kv_store = Arc::new(FilesystemStore::new(ldk_data_dir.clone().into()));
 
@@ -538,10 +544,11 @@ impl Builder {
 			}
 		};
 
-		let running = RwLock::new(None);
+		let stop_running = Arc::new(AtomicBool::new(false));
 
 		Node {
-			running,
+			runtime,
+			stop_running,
 			config,
 			wallet,
 			tx_sync,
@@ -561,18 +568,12 @@ impl Builder {
 	}
 }
 
-/// Wraps all objects that need to be preserved during the run time of [`Node`]. Will be dropped
-/// upon [`Node::stop()`].
-struct Runtime {
-	tokio_runtime: Arc<tokio::runtime::Runtime>,
-	stop_runtime: Arc<AtomicBool>,
-}
-
 /// The main interface object of LDK Node, wrapping the necessary LDK and BDK functionalities.
 ///
 /// Needs to be initialized and instantiated through [`Builder::build`].
 pub struct Node {
-	running: RwLock<Option<Runtime>>,
+	runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>,
+	stop_running: Arc<AtomicBool>,
 	config: Arc<Config>,
 	wallet: Arc<Wallet<bdk::database::SqliteDatabase>>,
 	tx_sync: Arc<EsploraSyncClient<Arc<FilesystemLogger>>>,
@@ -598,49 +599,15 @@ impl Node {
 	/// a thread-safe manner.
 	pub fn start(&self) -> Result<(), Error> {
 		// Acquire a run lock and hold it until we're setup.
-		let mut run_lock = self.running.write().unwrap();
-		if run_lock.is_some() {
+		let mut runtime_lock = self.runtime.write().unwrap();
+		if runtime_lock.is_some() {
 			// We're already running.
 			return Err(Error::AlreadyRunning);
 		}
 
-		let runtime = self.setup_runtime()?;
-		*run_lock = Some(runtime);
-		Ok(())
-	}
-
-	/// Disconnects all peers, stops all running background tasks, and shuts down [`Node`].
-	///
-	/// After this returns most API methods will return [`Error::NotRunning`].
-	pub fn stop(&self) -> Result<(), Error> {
-		let mut run_lock = self.running.write().unwrap();
-		if run_lock.is_none() {
-			return Err(Error::NotRunning);
-		}
-
-		let runtime = run_lock.as_ref().unwrap();
-
-		// Stop the runtime.
-		runtime.stop_runtime.store(true, Ordering::Release);
-
-		// Stop disconnect peers.
-		self.peer_manager.disconnect_all_peers();
-
-		// Drop the held runtimes.
-		self.wallet.drop_runtime();
+		let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap();
 
-		// Drop the runtime, which stops the background processor and any possibly remaining tokio threads.
-		*run_lock = None;
-		Ok(())
-	}
-
-	fn setup_runtime(&self) -> Result<Runtime, Error> {
-		let tokio_runtime =
-			Arc::new(tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap());
-
-		self.wallet.set_runtime(Arc::clone(&tokio_runtime));
-
-		let stop_runtime = Arc::new(AtomicBool::new(false));
+		let stop_running = Arc::new(AtomicBool::new(false));
 
 		let event_handler = Arc::new(EventHandler::new(
 			Arc::clone(&self.wallet),
@@ -649,7 +616,7 @@ impl Node {
 			Arc::clone(&self.network_graph),
 			Arc::clone(&self.keys_manager),
 			Arc::clone(&self.payment_store),
-			Arc::clone(&tokio_runtime),
+			Arc::clone(&self.runtime),
 			Arc::clone(&self.logger),
 			Arc::clone(&self.config),
 		));
@@ -660,7 +627,7 @@ impl Node {
 		let sync_cman = Arc::clone(&self.channel_manager);
 		let sync_cmon = Arc::clone(&self.chain_monitor);
 		let sync_logger = Arc::clone(&self.logger);
-		let stop_sync = Arc::clone(&stop_runtime);
+		let stop_sync = Arc::clone(&stop_running);
 
 		std::thread::spawn(move || {
 			tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(
@@ -691,8 +658,8 @@ impl Node {
 		});
 
 		let sync_logger = Arc::clone(&self.logger);
-		let stop_sync = Arc::clone(&stop_runtime);
-		tokio_runtime.spawn(async move {
+		let stop_sync = Arc::clone(&stop_running);
+		runtime.spawn(async move {
 			loop {
 				if stop_sync.load(Ordering::Acquire) {
 					return;
@@ -719,10 +686,10 @@ impl Node {
 		if let Some(listening_address) = &self.config.listening_address {
 			// Setup networking
 			let peer_manager_connection_handler = Arc::clone(&self.peer_manager);
-			let stop_listen = Arc::clone(&stop_runtime);
+			let stop_listen = Arc::clone(&stop_running);
 			let listening_address = listening_address.clone();
 
-			tokio_runtime.spawn(async move {
+			runtime.spawn(async move {
 				let listener =
 					tokio::net::TcpListener::bind(listening_address).await.expect(
 						"Failed to bind to listen address/port - is something else already listening on it?",
@@ -749,8 +716,8 @@ impl Node {
 		let connect_pm = Arc::clone(&self.peer_manager);
 		let connect_logger = Arc::clone(&self.logger);
 		let connect_peer_store = Arc::clone(&self.peer_store);
-		let stop_connect = Arc::clone(&stop_runtime);
-		tokio_runtime.spawn(async move {
+		let stop_connect = Arc::clone(&stop_running);
+		runtime.spawn(async move {
 			let mut interval = tokio::time::interval(PEER_RECONNECTION_INTERVAL);
 			loop {
 				if stop_connect.load(Ordering::Acquire) {
@@ -790,7 +757,7 @@ impl Node {
 		let background_peer_man = Arc::clone(&self.peer_manager);
 		let background_logger = Arc::clone(&self.logger);
 		let background_scorer = Arc::clone(&self.scorer);
-		let stop_background_processing = Arc::clone(&stop_runtime);
+		let stop_background_processing = Arc::clone(&stop_running);
 		let sleeper = move |d| {
 			let stop = Arc::clone(&stop_background_processing);
 			Box::pin(async move {
@@ -803,7 +770,7 @@ impl Node {
 			})
 		};
 
-		tokio_runtime.spawn(async move {
+		runtime.spawn(async move {
 			process_events_async(
 				background_persister,
 				|e| background_event_handler.handle_event(e),
@@ -820,7 +787,23 @@ impl Node {
 			.expect("Failed to process events");
 		});
 
-		Ok(Runtime { tokio_runtime, stop_runtime })
+		*runtime_lock = Some(runtime);
+		Ok(())
+	}
+
+	/// Disconnects all peers, stops all running background tasks, and shuts down [`Node`].
+	///
+	/// After this returns most API methods will return [`Error::NotRunning`].
+	pub fn stop(&self) -> Result<(), Error> {
+		let runtime = self.runtime.write().unwrap().take().ok_or(Error::NotRunning)?;
+		// Stop the runtime.
+		self.stop_running.store(true, Ordering::Release);
+		runtime.shutdown_timeout(Duration::from_secs(10));
+
+		// Stop disconnect peers.
+		self.peer_manager.disconnect_all_peers();
+
+		Ok(())
 	}
 
 	/// Blocks until the next event is available.
@@ -870,12 +853,11 @@ impl Node {
 	pub fn connect_open_channel(
 		&self, node_pubkey_and_address: &str, channel_amount_sats: u64, announce_channel: bool,
 	) -> Result<(), Error> {
-		let runtime_lock = self.running.read().unwrap();
-		if runtime_lock.is_none() {
+		let rt_lock = self.runtime.read().unwrap();
+		if rt_lock.is_none() {
 			return Err(Error::NotRunning);
 		}
-
-		let runtime = runtime_lock.as_ref().unwrap();
+		let runtime = rt_lock.as_ref().unwrap();
 
 		let cur_balance = self.wallet.get_balance()?;
 		if cur_balance.get_spendable() < channel_amount_sats {
@@ -893,7 +875,7 @@ impl Node {
 		let con_pm = Arc::clone(&self.peer_manager);
 
 		tokio::task::block_in_place(move || {
-			runtime.tokio_runtime.block_on(async move {
+			runtime.block_on(async move {
 				let res =
 					connect_peer_if_necessary(con_peer_pubkey, con_peer_addr, con_pm, con_logger)
 						.await;
@@ -947,10 +929,12 @@ impl Node {
 	///
 	/// Note that the wallets will be also synced regularly in the background.
 	pub fn sync_wallets(&self) -> Result<(), Error> {
-		let runtime_lock = self.running.read().unwrap();
-		if runtime_lock.is_none() {
+		let rt_lock = self.runtime.read().unwrap();
+		if rt_lock.is_none() {
 			return Err(Error::NotRunning);
 		}
+		let runtime = rt_lock.as_ref().unwrap();
+
 		let wallet = Arc::clone(&self.wallet);
 		let tx_sync = Arc::clone(&self.tx_sync);
 		let sync_cman = Arc::clone(&self.channel_manager);
@@ -961,7 +945,6 @@ impl Node {
 			&*sync_cmon as &(dyn Confirm + Sync + Send),
 		];
 
-		let runtime = runtime_lock.as_ref().unwrap();
 		tokio::task::block_in_place(move || {
 			tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(
 				async move {
@@ -986,7 +969,7 @@ impl Node {
 
 		let sync_logger = Arc::clone(&self.logger);
 		tokio::task::block_in_place(move || {
-			runtime.tokio_runtime.block_on(async move {
+			runtime.block_on(async move {
 				let now = Instant::now();
 				match tx_sync.sync(confirmables).await {
 					Ok(()) => {
@@ -1021,7 +1004,8 @@ impl Node {
 
 	/// Send a payement given an invoice.
 	pub fn send_payment(&self, invoice: Invoice) -> Result<PaymentHash, Error> {
-		if self.running.read().unwrap().is_none() {
+		let rt_lock = self.runtime.read().unwrap();
+		if rt_lock.is_none() {
 			return Err(Error::NotRunning);
 		}
 
@@ -1087,7 +1071,8 @@ impl Node {
 	pub fn send_payment_using_amount(
 		&self, invoice: Invoice, amount_msat: u64,
 	) -> Result<PaymentHash, Error> {
-		if self.running.read().unwrap().is_none() {
+		let rt_lock = self.runtime.read().unwrap();
+		if rt_lock.is_none() {
 			return Err(Error::NotRunning);
 		}
 
@@ -1175,7 +1160,8 @@ impl Node {
 	pub fn send_spontaneous_payment(
 		&self, amount_msat: u64, node_id: &str,
 	) -> Result<PaymentHash, Error> {
-		if self.running.read().unwrap().is_none() {
+		let rt_lock = self.runtime.read().unwrap();
+		if rt_lock.is_none() {
 			return Err(Error::NotRunning);
 		}
 

src/wallet.rs

@@ -38,7 +38,7 @@ where
 	inner: Mutex<bdk::Wallet<D>>,
 	// A cache storing the most recently retrieved fee rate estimations.
 	fee_rate_cache: RwLock<HashMap<ConfirmationTarget, FeeRate>>,
-	tokio_runtime: RwLock<Option<Arc<tokio::runtime::Runtime>>>,
+	runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>,
 	sync_lock: (Mutex<()>, Condvar),
 	logger: Arc<FilesystemLogger>,
 }
@@ -48,13 +48,13 @@ where
 	D: BatchDatabase,
 {
 	pub(crate) fn new(
-		blockchain: EsploraBlockchain, wallet: bdk::Wallet<D>, logger: Arc<FilesystemLogger>,
+		blockchain: EsploraBlockchain, wallet: bdk::Wallet<D>,
+		runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>, logger: Arc<FilesystemLogger>,
 	) -> Self {
 		let inner = Mutex::new(wallet);
 		let fee_rate_cache = RwLock::new(HashMap::new());
-		let tokio_runtime = RwLock::new(None);
 		let sync_lock = (Mutex::new(()), Condvar::new());
-		Self { blockchain, inner, fee_rate_cache, tokio_runtime, sync_lock, logger }
+		Self { blockchain, inner, fee_rate_cache, runtime, sync_lock, logger }
 	}
 
 	pub(crate) async fn sync(&self) -> Result<(), Error> {
@@ -115,14 +115,6 @@ where
 		res
 	}
 
-	pub(crate) fn set_runtime(&self, tokio_runtime: Arc<tokio::runtime::Runtime>) {
-		*self.tokio_runtime.write().unwrap() = Some(tokio_runtime);
-	}
-
-	pub(crate) fn drop_runtime(&self) {
-		*self.tokio_runtime.write().unwrap() = None;
-	}
-
 	pub(crate) async fn update_fee_estimates(&self) -> Result<(), Error> {
 		let mut locked_fee_rate_cache = self.fee_rate_cache.write().unwrap();
 
@@ -239,7 +231,7 @@ where
 	D: BatchDatabase,
 {
 	fn broadcast_transaction(&self, tx: &Transaction) {
-		let locked_runtime = self.tokio_runtime.read().unwrap();
+		let locked_runtime = self.runtime.read().unwrap();
 		if locked_runtime.as_ref().is_none() {
 			log_error!(self.logger, "Failed to broadcast transaction: No runtime.");
 			return;