fix: time and spawn

Author: Frando

Cargo.lock

@@ -122,17 +122,17 @@ async fn print_store_info(store: &Store) -> anyhow::Result<()> {
 }
 
 async fn info_task(store: Store) -> anyhow::Result<()> {
-    tokio::time::sleep(Duration::from_secs(1)).await;
+    n0_future::time::sleep(Duration::from_secs(1)).await;
     loop {
         print_store_info(&store).await?;
-        tokio::time::sleep(Duration::from_secs(5)).await;
+        n0_future::time::sleep(Duration::from_secs(5)).await;
     }
 }
 
 async fn delete_expired_tags_task(store: Store, prefix: &str) -> anyhow::Result<()> {
     loop {
         delete_expired_tags(&store, prefix, false).await?;
-        tokio::time::sleep(Duration::from_secs(5)).await;
+        n0_future::time::sleep(Duration::from_secs(5)).await;
     }
 }
 

examples/expiring-tags.rs

@@ -156,7 +156,7 @@ fn throttle(delay_ms: u64) -> EventSender {
                     );
                     // we could compute the delay from the size of the data to have a fixed rate.
                     // but the size is almost always 16 KiB (16 chunks).
-                    tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
+                    n0_future::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
                     msg.tx.send(Ok(())).await.ok();
                 });
             }

examples/limit.rs

@@ -342,7 +342,7 @@ impl Downloader {
     pub fn new(store: &Store, endpoint: &Endpoint) -> Self {
         let (tx, rx) = tokio::sync::mpsc::channel::<SwarmMsg>(32);
         let actor = DownloaderActor::new(store.clone(), endpoint.clone());
-        tokio::spawn(actor.run(rx));
+        n0_future::task::spawn(actor.run(rx));
         Self { client: tx.into() }
     }
 

src/api/downloader.rs

@@ -18,12 +18,13 @@
 //! [iroh]: https://docs.rs/iroh
 use std::{
     fmt::{self, Debug},
-    time::{Duration, Instant},
+    time::Duration,
 };
 
 use anyhow::Result;
 use bao_tree::{io::fsm::BaoContentItem, ChunkNum};
 use fsm::RequestCounters;
+use n0_future::time::Instant;
 use n0_snafu::SpanTrace;
 use nested_enum_utils::common_fields;
 use serde::{Deserialize, Serialize};

src/get.rs

@@ -308,7 +308,7 @@ pub async fn handle_connection(
         while let Ok(pair) = StreamPair::accept(&connection, progress.clone()).await {
             let span = debug_span!("stream", stream_id = %pair.stream_id());
             let store = store.clone();
-            tokio::spawn(handle_stream(pair, store).instrument(span));
+            n0_future::task::spawn(handle_stream(pair, store).instrument(span));
         }
         progress
             .connection_closed(|| ConnectionClosed { connection_id })

src/provider.rs

@@ -1561,7 +1561,7 @@ pub mod tests {
             let ranges = ChunkRanges::all();
             let (hash, bao) = create_n0_bao(&data, &ranges)?;
             let obs = store.observe(hash);
-            let task = tokio::spawn(async move {
+            let task = n0_future::task::spawn(async move {
                 obs.await_completion().await?;
                 api::Result::Ok(())
             });

src/store/fs.rs

@@ -766,7 +766,7 @@ impl Actor {
                     self.cmds.push_back(cmd.into()).ok();
                     let tx = db.begin_read().context(TransactionSnafu)?;
                     let tables = ReadOnlyTables::new(&tx).context(TableSnafu)?;
-                    let timeout = tokio::time::sleep(self.options.max_read_duration);
+                    let timeout = n0_future::time::sleep(self.options.max_read_duration);
                     pin!(timeout);
                     let mut n = 0;
                     while let Some(cmd) = self.cmds.extract(Command::read_only, &mut timeout).await
@@ -784,7 +784,7 @@ impl Actor {
                     let ftx = self.ds.begin_write();
                     let tx = db.begin_write().context(TransactionSnafu)?;
                     let mut tables = Tables::new(&tx, &ftx).context(TableSnafu)?;
-                    let timeout = tokio::time::sleep(self.options.max_read_duration);
+                    let timeout = n0_future::time::sleep(self.options.max_read_duration);
                     pin!(timeout);
                     let mut n = 0;
                     while let Some(cmd) = self

src/store/fs/meta.rs

@@ -723,7 +723,7 @@ impl<P: Params> EntityManager<P> {
             options.entity_response_inbox_size,
             options.entity_futures_initial_capacity,
         );
-        tokio::spawn(actor.run());
+        n0_future::task::spawn(actor.run());
         Self(send)
     }
 

src/store/fs/util/entity_manager.rs

@@ -221,7 +221,7 @@ pub async fn run_gc(store: Store, config: GcConfig) {
     let mut live = HashSet::new();
     loop {
         live.clear();
-        tokio::time::sleep(config.interval).await;
+        n0_future::time::sleep(config.interval).await;
         if let Some(ref cb) = config.add_protected {
             match (cb)(&mut live).await {
                 ProtectOutcome::Continue => {}