Introduce Runtime object allowng to detect outer runtime context

Instead of holding an `Arc<RwLock<Option<Arc<tokio::runtime::Runtime>>>`
and dealing with stuff like `tokio::task::block_in_place` at all
callsites, we introduce a `Runtime` object that takes care of the state
transitions, and allows to detect and reuse an outer runtime context.

We also adjust the `with_runtime` API to take a `tokio::runtime::Handle`
rather than an `Arc<Runtime>`.

Author: tnull

bindings/ldk_node.udl

@@ -64,7 +64,7 @@ dictionary LogRecord {
 
 [Trait, WithForeign]
 interface LogWriter {
-    void log(LogRecord record);
+	void log(LogRecord record);
 };
 
 interface Builder {
@@ -161,8 +161,8 @@ interface Node {
 
 [Enum]
 interface Bolt11InvoiceDescription {
-  Hash(string hash);
-  Direct(string description);
+	Hash(string hash);
+	Direct(string description);
 };
 
 interface Bolt11Payment {
@@ -335,6 +335,7 @@ enum BuildError {
 	"InvalidListeningAddresses",
 	"InvalidAnnouncementAddresses",
 	"InvalidNodeAlias",
+	"RuntimeSetupFailed",
 	"ReadFailed",
 	"WriteFailed",
 	"StoragePathAccessFailed",

src/builder.rs

@@ -28,6 +28,7 @@ use crate::liquidity::{
 use crate::logger::{log_error, log_info, LdkLogger, LogLevel, LogWriter, Logger};
 use crate::message_handler::NodeCustomMessageHandler;
 use crate::peer_store::PeerStore;
+use crate::runtime::Runtime;
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
 	ChainMonitor, ChannelManager, DynStore, GossipSync, Graph, KeysManager, MessageRouter,
@@ -168,6 +169,8 @@ pub enum BuildError {
 	InvalidAnnouncementAddresses,
 	/// The provided alias is invalid.
 	InvalidNodeAlias,
+	/// An attempt to setup a runtime has failed.
+	RuntimeSetupFailed,
 	/// We failed to read data from the [`KVStore`].
 	///
 	/// [`KVStore`]: lightning::util::persist::KVStore
@@ -205,6 +208,7 @@ impl fmt::Display for BuildError {
 			Self::InvalidAnnouncementAddresses => {
 				write!(f, "Given announcement addresses are invalid.")
 			},
+			Self::RuntimeSetupFailed => write!(f, "Failed to setup a runtime."),
 			Self::ReadFailed => write!(f, "Failed to read from store."),
 			Self::WriteFailed => write!(f, "Failed to write to store."),
 			Self::StoragePathAccessFailed => write!(f, "Failed to access the given storage path."),
@@ -236,6 +240,7 @@ pub struct NodeBuilder {
 	gossip_source_config: Option<GossipSourceConfig>,
 	liquidity_source_config: Option<LiquiditySourceConfig>,
 	log_writer_config: Option<LogWriterConfig>,
+	runtime_handle: Option<tokio::runtime::Handle>,
 }
 
 impl NodeBuilder {
@@ -252,16 +257,28 @@ impl NodeBuilder {
 		let gossip_source_config = None;
 		let liquidity_source_config = None;
 		let log_writer_config = None;
+		let runtime_handle = None;
 		Self {
 			config,
 			entropy_source_config,
 			chain_data_source_config,
 			gossip_source_config,
 			liquidity_source_config,
 			log_writer_config,
+			runtime_handle,
 		}
 	}
 
+	/// Configures the [`Node`] instance to (re-)use a specific `tokio` runtime.
+	///
+	/// If not provided, the node will spawn its own runtime or reuse any outer runtime context it
+	/// can detect.
+	#[cfg_attr(feature = "uniffi", allow(dead_code))]
+	pub fn set_runtime(&mut self, runtime_handle: tokio::runtime::Handle) -> &mut Self {
+		self.runtime_handle = Some(runtime_handle);
+		self
+	}
+
 	/// Configures the [`Node`] instance to source its wallet entropy from a seed file on disk.
 	///
 	/// If the given file does not exist a new random seed file will be generated and
@@ -650,6 +667,15 @@ impl NodeBuilder {
 	) -> Result<Node, BuildError> {
 		let logger = setup_logger(&self.log_writer_config, &self.config)?;
 
+		let runtime = if let Some(handle) = self.runtime_handle.as_ref() {
+			Arc::new(Runtime::with_handle(handle.clone()))
+		} else {
+			Arc::new(Runtime::new().map_err(|e| {
+				log_error!(logger, "Failed to setup tokio runtime: {}", e);
+				BuildError::RuntimeSetupFailed
+			})?)
+		};
+
 		let seed_bytes = seed_bytes_from_config(
 			&self.config,
 			self.entropy_source_config.as_ref(),
@@ -678,6 +704,7 @@ impl NodeBuilder {
 			self.gossip_source_config.as_ref(),
 			self.liquidity_source_config.as_ref(),
 			seed_bytes,
+			runtime,
 			logger,
 			Arc::new(vss_store),
 		)
@@ -687,6 +714,15 @@ impl NodeBuilder {
 	pub fn build_with_store(&self, kv_store: Arc<DynStore>) -> Result<Node, BuildError> {
 		let logger = setup_logger(&self.log_writer_config, &self.config)?;
 
+		let runtime = if let Some(handle) = self.runtime_handle.as_ref() {
+			Arc::new(Runtime::with_handle(handle.clone()))
+		} else {
+			Arc::new(Runtime::new().map_err(|e| {
+				log_error!(logger, "Failed to setup tokio runtime: {}", e);
+				BuildError::RuntimeSetupFailed
+			})?)
+		};
+
 		let seed_bytes = seed_bytes_from_config(
 			&self.config,
 			self.entropy_source_config.as_ref(),
@@ -700,6 +736,7 @@ impl NodeBuilder {
 			self.gossip_source_config.as_ref(),
 			self.liquidity_source_config.as_ref(),
 			seed_bytes,
+			runtime,
 			logger,
 			kv_store,
 		)
@@ -1049,7 +1086,7 @@ fn build_with_store_internal(
 	config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
 	gossip_source_config: Option<&GossipSourceConfig>,
 	liquidity_source_config: Option<&LiquiditySourceConfig>, seed_bytes: [u8; 64],
-	logger: Arc<Logger>, kv_store: Arc<DynStore>,
+	runtime: Arc<Runtime>, logger: Arc<Logger>, kv_store: Arc<DynStore>,
 ) -> Result<Node, BuildError> {
 	if let Err(err) = may_announce_channel(&config) {
 		if config.announcement_addresses.is_some() {
@@ -1239,8 +1276,6 @@ fn build_with_store_internal(
 		},
 	};
 
-	let runtime = Arc::new(RwLock::new(None));
-
 	// Initialize the ChainMonitor
 	let chain_monitor: Arc<ChainMonitor> = Arc::new(chainmonitor::ChainMonitor::new(
 		Some(Arc::clone(&chain_source)),
@@ -1633,6 +1668,8 @@ fn build_with_store_internal(
 	let (stop_sender, _) = tokio::sync::watch::channel(());
 	let background_processor_task = Mutex::new(None);
 
+	let is_running = Arc::new(RwLock::new(false));
+
 	Ok(Node {
 		runtime,
 		stop_sender,
@@ -1658,6 +1695,7 @@ fn build_with_store_internal(
 		scorer,
 		peer_store,
 		payment_store,
+		is_running,
 		is_listening,
 		node_metrics,
 	})

src/chain/electrum.rs

@@ -18,6 +18,7 @@ use crate::fee_estimator::{
 };
 use crate::io::utils::write_node_metrics;
 use crate::logger::{log_bytes, log_error, log_info, log_trace, LdkLogger, Logger};
+use crate::runtime::Runtime;
 use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::NodeMetrics;
 
@@ -89,7 +90,7 @@ impl ElectrumChainSource {
 		}
 	}
 
-	pub(super) fn start(&self, runtime: Arc<tokio::runtime::Runtime>) -> Result<(), Error> {
+	pub(super) fn start(&self, runtime: Arc<Runtime>) -> Result<(), Error> {
 		self.electrum_runtime_status.write().unwrap().start(
 			self.server_url.clone(),
 			Arc::clone(&runtime),
@@ -345,7 +346,7 @@ impl ElectrumRuntimeStatus {
 	}
 
 	pub(super) fn start(
-		&mut self, server_url: String, runtime: Arc<tokio::runtime::Runtime>, config: Arc<Config>,
+		&mut self, server_url: String, runtime: Arc<Runtime>, config: Arc<Config>,
 		logger: Arc<Logger>,
 	) -> Result<(), Error> {
 		match self {
@@ -409,15 +410,14 @@ struct ElectrumRuntimeClient {
 	electrum_client: Arc<ElectrumClient>,
 	bdk_electrum_client: Arc<BdkElectrumClient<ElectrumClient>>,
 	tx_sync: Arc<ElectrumSyncClient<Arc<Logger>>>,
-	runtime: Arc<tokio::runtime::Runtime>,
+	runtime: Arc<Runtime>,
 	config: Arc<Config>,
 	logger: Arc<Logger>,
 }
 
 impl ElectrumRuntimeClient {
 	fn new(
-		server_url: String, runtime: Arc<tokio::runtime::Runtime>, config: Arc<Config>,
-		logger: Arc<Logger>,
+		server_url: String, runtime: Arc<Runtime>, config: Arc<Config>, logger: Arc<Logger>,
 	) -> Result<Self, Error> {
 		let electrum_config = ElectrumConfigBuilder::new()
 			.retry(ELECTRUM_CLIENT_NUM_RETRIES)
@@ -550,7 +550,6 @@ impl ElectrumRuntimeClient {
 
 		let spawn_fut =
 			self.runtime.spawn_blocking(move || electrum_client.transaction_broadcast(&tx));
-
 		let timeout_fut =
 			tokio::time::timeout(Duration::from_secs(TX_BROADCAST_TIMEOUT_SECS), spawn_fut);
 

src/chain/mod.rs

@@ -19,6 +19,7 @@ use crate::config::{
 use crate::fee_estimator::OnchainFeeEstimator;
 use crate::io::utils::write_node_metrics;
 use crate::logger::{log_info, log_trace, LdkLogger, Logger};
+use crate::runtime::Runtime;
 use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, NodeMetrics};
 
@@ -188,7 +189,7 @@ impl ChainSource {
 		Self { kind, logger }
 	}
 
-	pub(crate) fn start(&self, runtime: Arc<tokio::runtime::Runtime>) -> Result<(), Error> {
+	pub(crate) fn start(&self, runtime: Arc<Runtime>) -> Result<(), Error> {
 		match &self.kind {
 			ChainSourceKind::Electrum(electrum_chain_source) => {
 				electrum_chain_source.start(runtime)?

src/event.rs

@@ -29,6 +29,8 @@ use crate::io::{
 };
 use crate::logger::{log_debug, log_error, log_info, LdkLogger};
 
+use crate::runtime::Runtime;
+
 use lightning::events::bump_transaction::BumpTransactionEvent;
 use lightning::events::{ClosureReason, PaymentPurpose, ReplayEvent};
 use lightning::events::{Event as LdkEvent, PaymentFailureReason};
@@ -53,7 +55,7 @@ use core::future::Future;
 use core::task::{Poll, Waker};
 use std::collections::VecDeque;
 use std::ops::Deref;
-use std::sync::{Arc, Condvar, Mutex, RwLock};
+use std::sync::{Arc, Condvar, Mutex};
 use std::time::Duration;
 
 /// An event emitted by [`Node`], which should be handled by the user.
@@ -451,7 +453,7 @@ where
 	liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
 	payment_store: Arc<PaymentStore>,
 	peer_store: Arc<PeerStore<L>>,
-	runtime: Arc<RwLock<Option<Arc<tokio::runtime::Runtime>>>>,
+	runtime: Arc<Runtime>,
 	logger: L,
 	config: Arc<Config>,
 }
@@ -466,8 +468,8 @@ where
 		channel_manager: Arc<ChannelManager>, connection_manager: Arc<ConnectionManager<L>>,
 		output_sweeper: Arc<Sweeper>, network_graph: Arc<Graph>,
 		liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
-		payment_store: Arc<PaymentStore>, peer_store: Arc<PeerStore<L>>,
-		runtime: Arc<RwLock<Option<Arc<tokio::runtime::Runtime>>>>, logger: L, config: Arc<Config>,
+		payment_store: Arc<PaymentStore>, peer_store: Arc<PeerStore<L>>, runtime: Arc<Runtime>,
+		logger: L, config: Arc<Config>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -1049,17 +1051,14 @@ where
 				let forwarding_channel_manager = self.channel_manager.clone();
 				let min = time_forwardable.as_millis() as u64;
 
-				let runtime_lock = self.runtime.read().unwrap();
-				debug_assert!(runtime_lock.is_some());
+				let future = 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;
 
-				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();
+				};
 
-						forwarding_channel_manager.process_pending_htlc_forwards();
-					});
-				}
+				self.runtime.spawn(future);
 			},
 			LdkEvent::SpendableOutputs { outputs, channel_id } => {
 				match self.output_sweeper.track_spendable_outputs(outputs, channel_id, true, None) {
@@ -1421,31 +1420,27 @@ where
 				debug_assert!(false, "We currently don't handle BOLT12 invoices manually, so this event should never be emitted.");
 			},
 			LdkEvent::ConnectionNeeded { node_id, addresses } => {
-				let runtime_lock = self.runtime.read().unwrap();
-				debug_assert!(runtime_lock.is_some());
-
-				if let Some(runtime) = runtime_lock.as_ref() {
-					let spawn_logger = self.logger.clone();
-					let spawn_cm = Arc::clone(&self.connection_manager);
-					runtime.spawn(async move {
-						for addr in &addresses {
-							match spawn_cm.connect_peer_if_necessary(node_id, addr.clone()).await {
-								Ok(()) => {
-									return;
-								},
-								Err(e) => {
-									log_error!(
-										spawn_logger,
-										"Failed to establish connection to peer {}@{}: {}",
-										node_id,
-										addr,
-										e
-									);
-								},
-							}
+				let spawn_logger = self.logger.clone();
+				let spawn_cm = Arc::clone(&self.connection_manager);
+				let future = async move {
+					for addr in &addresses {
+						match spawn_cm.connect_peer_if_necessary(node_id, addr.clone()).await {
+							Ok(()) => {
+								return;
+							},
+							Err(e) => {
+								log_error!(
+									spawn_logger,
+									"Failed to establish connection to peer {}@{}: {}",
+									node_id,
+									addr,
+									e
+								);
+							},
 						}
-					});
-				}
+					}
+				};
+				self.runtime.spawn(future);
 			},
 			LdkEvent::BumpTransaction(bte) => {
 				match bte {