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

@@ -23,7 +23,7 @@ use bitcoin::OutPoint;
 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.
@@ -247,7 +247,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>,
 }
@@ -261,7 +261,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,
@@ -271,7 +271,7 @@ where
 			keys_manager,
 			payment_store,
 			logger,
-			tokio_runtime,
+			runtime,
 			_config,
 		}
 	}
@@ -531,12 +531,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

@@ -347,7 +347,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()));
 
@@ -556,10 +562,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,
@@ -579,18 +586,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>>>,
@@ -616,49 +617,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();
-
-		// 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 runtime = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap();
 
-		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),
@@ -667,7 +634,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),
 		));
@@ -678,7 +645,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(
@@ -709,8 +676,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;
@@ -737,10 +704,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?",
@@ -767,8 +734,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) {
@@ -808,7 +775,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 {
@@ -821,7 +788,7 @@ impl Node {
 			})
 		};
 
-		tokio_runtime.spawn(async move {
+		runtime.spawn(async move {
 			process_events_async(
 				background_persister,
 				|e| background_event_handler.handle_event(e),
@@ -838,7 +805,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);
+
+		// Stop disconnect peers.
+		self.peer_manager.disconnect_all_peers();
+
+		runtime.shutdown_timeout(Duration::from_secs(10));
+		Ok(())
 	}
 
 	/// Blocks until the next event is available.
@@ -915,12 +898,11 @@ impl Node {
 	pub fn connect(
 		&self, node_id: PublicKey, address: SocketAddr, permanently: 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 peer_info = PeerInfo { pubkey: node_id, address };
 
@@ -932,7 +914,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;
@@ -958,8 +940,8 @@ impl Node {
 	/// Will also remove the peer from the peer store, i.e., after this has been called we won't
 	/// try to reconnect on restart.
 	pub fn disconnect(&self, counterparty_node_id: &PublicKey) -> 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);
 		}
 
@@ -989,12 +971,11 @@ impl Node {
 		&self, node_id: PublicKey, address: SocketAddr, channel_amount_sats: u64,
 		push_to_counterparty_msat: Option<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 {
@@ -1012,7 +993,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;
@@ -1067,10 +1048,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);
@@ -1081,7 +1064,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 {
@@ -1106,7 +1088,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(()) => {
@@ -1141,7 +1123,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);
 		}
 
@@ -1207,7 +1190,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);
 		}
 
@@ -1295,7 +1279,8 @@ impl Node {
 	pub fn send_spontaneous_payment(
 		&self, amount_msat: u64, node_id: &PublicKey,
 	) -> 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);
 		}