chore: use async-rt in place of rt utils (#362)

CHANGELOG.md

@@ -1,5 +1,6 @@
 # 0.14.2
 - feat: Add reconnect option to address book. [PR 356](https://github.com/dariusc93/rust-ipfs/pull/356)
+- chore: use async-rt in place of rt utils. [PR 362](https://github.com/dariusc93/rust-ipfs/pull/362)
 
 # 0.14.1
 - fix: remove expect when session failed to get next block.

Cargo.toml

@@ -21,6 +21,7 @@ test_js_interop = []
 
 [workspace.dependencies]
 anyhow = "1.0.94"
+async-rt = "0.1.2"
 async-stream = { version = "0.3.6" }
 async-trait = { version = "0.1.82" }
 asynchronous-codec = "0.7.0"
@@ -94,6 +95,7 @@ zeroize = "1.8.1"
 
 [dependencies]
 anyhow.workspace = true
+async-rt.workspace = true
 async-stream.workspace = true
 async-trait.workspace = true
 asynchronous-codec.workspace = true

src/lib.rs

@@ -32,7 +32,6 @@ pub mod p2p;
 pub mod path;
 pub mod refs;
 pub mod repo;
-pub(crate) mod rt;
 mod task;
 pub mod unixfs;
 
@@ -66,19 +65,20 @@ use repo::{
     BlockStore, DataStore, GCConfig, GCTrigger, Lock, RepoFetch, RepoInsertPin, RepoRemovePin,
 };
 
-use rt::{AbortableJoinHandle, Executor, ExecutorSwitch};
 use tracing::Span;
 use tracing_futures::Instrument;
 
 use unixfs::UnixfsGet;
 use unixfs::{AddOpt, IpfsUnixfs, UnixfsAdd, UnixfsCat, UnixfsLs};
 
+pub use self::p2p::gossipsub::SubscriptionStream;
 use self::{
     dag::IpldDag,
     ipns::Ipns,
     p2p::{create_swarm, TSwarm},
     repo::Repo,
 };
+use async_rt::AbortableJoinHandle;
 use ipld_core::cid::Cid;
 use ipld_core::ipld::Ipld;
 use std::borrow::Borrow;
@@ -91,8 +91,6 @@ use std::{
     time::Duration,
 };
 
-pub use self::p2p::gossipsub::SubscriptionStream;
-
 pub use self::{
     error::Error,
     p2p::BehaviourEvent,
@@ -342,7 +340,6 @@ pub struct Ipfs {
     record_key_validator: HashMap<String, Arc<dyn Fn(&str) -> anyhow::Result<Key> + Sync + Send>>,
     _guard: AbortableJoinHandle<()>,
     _gc_guard: AbortableJoinHandle<()>,
-    executor: ExecutorSwitch,
 }
 
 impl std::fmt::Debug for Ipfs {
@@ -916,8 +913,6 @@ impl<C: NetworkBehaviour<ToSwarm = void::Void> + Send> UninitializedIpfs<C> {
             ..
         } = self;
 
-        let executor = ExecutorSwitch;
-
         let keys = keys.unwrap_or(Keypair::generate_ed25519());
 
         let root_span = Option::take(&mut options.span)
@@ -999,7 +994,6 @@ impl<C: NetworkBehaviour<ToSwarm = void::Void> + Send> UninitializedIpfs<C> {
             record_key_validator,
             _guard,
             _gc_guard,
-            executor,
         };
 
         //Note: If `All` or `Pinned` are used, we would have to auto adjust the amount of
@@ -1043,7 +1037,6 @@ impl<C: NetworkBehaviour<ToSwarm = void::Void> + Send> UninitializedIpfs<C> {
         let swarm = create_swarm(
             &keys,
             &options,
-            executor,
             &ipfs.repo,
             exec_span,
             (custom_behaviour, custom_transport),
@@ -1054,7 +1047,7 @@ impl<C: NetworkBehaviour<ToSwarm = void::Void> + Send> UninitializedIpfs<C> {
         } = options;
 
         let gc_handle = gc_config.map(|config| {
-            executor.spawn_abortable({
+            async_rt::task::spawn_abortable({
                 let repo = ipfs.repo.clone();
                 async move {
                     let GCConfig { duration, trigger } = config;
@@ -1160,7 +1153,7 @@ impl<C: NetworkBehaviour<ToSwarm = void::Void> + Send> UninitializedIpfs<C> {
             }
         }
 
-        ipfs._guard.replace(executor.spawn_abortable({
+        let main_handle = async_rt::task::spawn_abortable({
             async move {
                 //Note: For now this is not configurable as its meant for internal testing purposes but may change in the future
                 let as_fut = false;
@@ -1174,8 +1167,12 @@ impl<C: NetworkBehaviour<ToSwarm = void::Void> + Send> UninitializedIpfs<C> {
                 fut.await
             }
             .instrument(swarm_span)
-        }));
-        ipfs._gc_guard.replace(gc_handle);
+        });
+
+        unsafe {
+            ipfs._guard.replace(main_handle);
+            ipfs._gc_guard.replace(gc_handle);
+        }
         Ok(ipfs)
     }
 }
@@ -2547,7 +2544,7 @@ impl Ipfs {
         self.to_task.clone().send(IpfsEvent::Bootstrap(tx)).await?;
         let fut = rx.await??;
 
-        self.executor.dispatch(async move {
+        async_rt::task::dispatch(async move {
             if let Err(e) = fut.await.map_err(|e| anyhow!(e)) {
                 tracing::error!(error = %e, "failed to bootstrap");
             }

src/p2p/mod.rs

@@ -1,7 +1,6 @@
 //! P2P handling for IPFS nodes.
 use crate::error::Error;
 use crate::repo::Repo;
-use crate::rt::{Executor, ExecutorSwitch};
 use crate::{IpfsOptions, TTransportFn};
 use std::convert::TryInto;
 use std::num::{NonZeroU8, NonZeroUsize};
@@ -227,7 +226,6 @@ impl Default for SwarmConfig {
 pub(crate) fn create_swarm<C>(
     keypair: &Keypair,
     options: &IpfsOptions,
-    executor: ExecutorSwitch,
     repo: &Repo,
     span: Span,
     (custom, custom_transport): (Option<C>, Option<TTransportFn>),
@@ -256,7 +254,7 @@ where
         transport,
         behaviour,
         peer_id,
-        libp2p::swarm::Config::with_executor(SpannedExecutor { executor, span })
+        libp2p::swarm::Config::with_executor(SpannedExecutor { span })
             .with_notify_handler_buffer_size(swarm_config.notify_handler_buffer_size)
             .with_per_connection_event_buffer_size(swarm_config.connection_event_buffer_size)
             .with_dial_concurrency_factor(swarm_config.dial_concurrency_factor)
@@ -268,7 +266,6 @@ where
 }
 
 struct SpannedExecutor {
-    executor: ExecutorSwitch,
     span: Span,
 }
 
@@ -278,6 +275,6 @@ impl libp2p::swarm::Executor for SpannedExecutor {
         future: std::pin::Pin<Box<dyn std::future::Future<Output = ()> + 'static + Send>>,
     ) {
         use tracing_futures::Instrument;
-        self.executor.dispatch(future.instrument(self.span.clone()));
+        async_rt::task::dispatch(future.instrument(self.span.clone()));
     }
 }