chore(libp2p-mplex): use tokio instead of async_std

ref #4449

Pull-Request: #6045.

Author: kamuik16

Cargo.lock

@@ -24,7 +24,7 @@ tracing = { workspace = true }
 unsigned-varint = { workspace = true, features = ["asynchronous_codec"] }
 
 [dev-dependencies]
-async-std = { version = "1.7.0", features = ["attributes"] }
+tokio = { workspace = true, features = ["rt", "rt-multi-thread", "macros"] }
 criterion = "0.5"
 futures = { workspace = true }
 libp2p-identity = { workspace = true, features = ["rand"] }

muxers/mplex/Cargo.toml

@@ -23,13 +23,8 @@
 
 use std::{pin::Pin, time::Duration};
 
-use async_std::task;
 use criterion::{black_box, criterion_group, criterion_main, Criterion, Throughput};
-use futures::{
-    channel::oneshot,
-    future::{join, poll_fn},
-    prelude::*,
-};
+use futures::{channel::oneshot, future::poll_fn, prelude::*};
 use libp2p_core::{
     multiaddr::multiaddr, muxing, muxing::StreamMuxerExt, transport, transport::ListenerId,
     upgrade, Endpoint, Multiaddr, Transport,
@@ -38,6 +33,7 @@ use libp2p_identity as identity;
 use libp2p_identity::PeerId;
 use libp2p_mplex as mplex;
 use libp2p_plaintext as plaintext;
+use tokio::runtime::Runtime;
 use tracing_subscriber::EnvFilter;
 
 type BenchTransport = transport::Boxed<(PeerId, muxing::StreamMuxerBox)>;
@@ -180,7 +176,10 @@ fn run(
     };
 
     // Wait for all data to be received.
-    task::block_on(join(sender, receiver));
+    let rt = Runtime::new().unwrap();
+    rt.block_on(async {
+        tokio::join!(sender, receiver);
+    });
 }
 
 fn tcp_transport(split_send_size: usize) -> BenchTransport {

muxers/mplex/benches/split_send_size.rs

@@ -1150,10 +1150,10 @@ const EXTRA_PENDING_FRAMES: usize = 1000;
 mod tests {
     use std::{collections::HashSet, num::NonZeroU8, ops::DerefMut, pin::Pin};
 
-    use async_std::task;
     use asynchronous_codec::{Decoder, Encoder};
     use bytes::BytesMut;
     use quickcheck::*;
+    use tokio::runtime::Runtime;
 
     use super::*;
 
@@ -1265,7 +1265,8 @@ mod tests {
             };
             let mut m = Multiplexed::new(conn, cfg.clone());
 
-            task::block_on(future::poll_fn(move |cx| {
+            let rt = Runtime::new().unwrap();
+            rt.block_on(future::poll_fn(move |cx| {
                 // Receive all inbound streams.
                 for i in 0..cfg.max_substreams {
                     match m.poll_next_stream(cx) {
@@ -1383,7 +1384,8 @@ mod tests {
 
             // Run the test.
             let mut opened = HashSet::new();
-            task::block_on(future::poll_fn(move |cx| {
+            let rt = Runtime::new().unwrap();
+            rt.block_on(future::poll_fn(move |cx| {
                 // Open a number of streams.
                 for _ in 0..num_streams {
                     let id = ready!(m.poll_open_stream(cx)).unwrap();

muxers/mplex/src/io.rs

@@ -1,14 +1,14 @@
 use libp2p_mplex::Config;
 
-#[async_std::test]
+#[tokio::test]
 async fn close_implies_flush() {
     let (alice, bob) =
         libp2p_muxer_test_harness::connected_muxers_on_memory_ring_buffer::<Config, _, _>().await;
 
     libp2p_muxer_test_harness::close_implies_flush(alice, bob).await;
 }
 
-#[async_std::test]
+#[tokio::test]
 async fn read_after_close() {
     let (alice, bob) =
         libp2p_muxer_test_harness::connected_muxers_on_memory_ring_buffer::<Config, _, _>().await;