tokio: switch the whole library to async tokio for now

We need to dedicate the capacity to individual protocols. We still want to support non-tokio usage in the future.
2nd-Layer · Jan 26, 2022 · aa80407 · aa80407
1 parent 5077db2
commit aa80407
Show file tree

Hide file tree

Showing 12 changed files with 271 additions and 347 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -16,7 +16,6 @@ categories = ["network-programming"]
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
-experimental = ["tokio", "futures"]
 
 [dependencies]
 blake2b_simd = "1.0.0"
@@ -30,8 +29,7 @@ serde = { version = "1.0.117", features = ["derive"] }
 serde_cbor = "0.11.1"
 serde_json = "1.0.59"
 log = "0.4.11"
-tokio = { version = "1.15.0", features = ["full"], optional = true }
-futures = { version = "0.3.19", optional = true }
+tokio = { version = "1.15.0", features = ["full"]}
 
 [dev-dependencies]
 simple_logger = { version = "2.1.0", default-features = false, features = ["colors"] }
@@ -49,7 +47,3 @@ crate-type = ["staticlib"]
 [[example]]
 name = "pooltool"
 crate-type = ["staticlib"]
-
-[[example]]
-name = "async"
-required-features = ["experimental"]
diff --git a/examples/async.rs b/examples/async.rs
diff --git a/examples/common.rs b/examples/common.rs
@@ -8,19 +8,18 @@ SPDX-License-Identifier: GPL-3.0-only OR LGPL-3.0-only
 use simple_logger::SimpleLogger;
 use std::path::PathBuf;
 
+#[derive(Clone)]
 pub struct Config {
     pub db: PathBuf,
     pub host: String,
-    pub port: u16,
     pub magic: u32,
 }
 
 pub fn init() -> Config {
     SimpleLogger::new().init().unwrap();
     Config {
         db: PathBuf::from("sqlite.db"),
-        host: "relays-new.cardano-mainnet.iohk.io".to_string(),
-        port: 3001,
+        host: "relays-new.cardano-mainnet.iohk.io:3001".to_string(),
         magic: 764824073,
     }
 }
diff --git a/examples/local.rs b/examples/local.rs
@@ -5,9 +5,8 @@ SPDX-License-Identifier: GPL-3.0-only OR LGPL-3.0-only
 
 */
 
-use cardano_ouroboros_network::mux;
-use futures::executor::block_on;
-use log::{error, info};
+use cardano_ouroboros_network::mux::Connection;
+use log::info;
 use std::env;
 
 mod common;
@@ -16,21 +15,21 @@ mod common;
  * Test a handshake with the local node's unix socket
  */
 #[cfg(target_family = "unix")]
-fn main() {
+#[tokio::main]
+async fn main() {
     let cfg = common::init();
-    let args: Vec<String> = env::args().collect();
     let magic = cfg.magic;
 
-    let socket_path = &args[1];
-    info!("UNIX socket path set to {} ", socket_path);
+    let mut args = env::args();
+    args.next();
+    let socket_path = &args.next().unwrap_or("test.sock".to_string());
+    info!("UNIX socket path set to {:?} ", socket_path);
 
-    block_on(async {
-        let channel = mux::connection::connect_unix(socket_path)
-            .await
-            .unwrap();
-        channel.handshake(magic).await.unwrap();
-        info!("Ping UNIX socket success");
-    });
+    let mut connection = Connection::unix_connect(socket_path)
+        .await
+        .unwrap();
+    connection.handshake(magic).await.unwrap();
+    info!("Ping UNIX socket success");
 }
 
 #[cfg(target_family = "windows")]

diff --git a/examples/ping.rs b/examples/ping.rs
@@ -5,55 +5,52 @@ SPDX-License-Identifier: GPL-3.0-only OR LGPL-3.0-only
 
 */
 
-use cardano_ouroboros_network::mux;
+use cardano_ouroboros_network::mux::Connection;
 use std::{
     env,
     time::Duration,
 };
 use log::{info, error};
-use futures::{
-    executor::block_on,
-    future::join_all,
-};
+use futures::future::join_all;
 
 mod common;
 
-async fn ping(host: &String, port: u16, magic: u32) -> Result<(Duration, Duration), String> {
-    info!("Pinging host {} port {} magic {}.", host, port, magic);
-    let channel = match mux::connection::connect(&host, port).await {
-        Ok(channel) => channel,
+async fn ping(host: &String, magic: u32) -> Result<(Duration, Duration), String> {
+    info!("Pinging host {} magic {}.", host, magic);
+    let mut connection = match Connection::tcp_connect(&host).await {
+        Ok(connection) => connection,
         Err(_) => { return Err("Could not connect.".to_string()) }
     };
-    let connect_duration = channel.duration();
-    channel.handshake(magic).await?;
-    let total_duration = channel.duration();
+    let connect_duration = connection.duration();
+    connection.handshake(magic).await?;
+    let total_duration = connection.duration();
     Ok((connect_duration, total_duration))
 }
 
-fn main() {
+#[tokio::main]
+async fn main() {
     let cfg = common::init();
-    let port = cfg.port;
-    let magic = cfg.magic;
 
-    block_on(async {
-        let mut args: Vec<String> = env::args().collect();
+    let mut args: Vec<String> = env::args().collect();
 
-        args.remove(0);
+    args.remove(0);
 
-        /* Use configured host by default. */
-        if args.len() == 0 {
-            args = vec![cfg.host.clone()];
-        }
+    /* Use configured host by default. */
+    if args.len() == 0 {
+        args = vec![cfg.host.clone()];
+    }
 
-        join_all(args.iter().map(|host| async move {
-            match ping(&host.clone(), port, magic).await {
+    join_all(args.iter().map(|host| {
+        let cfg = cfg.clone();
+        async move {
+            match ping(&host.clone(), cfg.magic).await {
                 Ok((connect_duration, total_duration)) => {
-                    info!("Ping {}:{} success! : connect_duration: {}, total_duration: {}", &host, port, connect_duration.as_millis(), total_duration.as_millis());
+                    info!("Ping {} success! : connect_duration: {}, total_duration: {}", &host, connect_duration.as_millis(), total_duration.as_millis());
                 }
                 Err(error) => {
-                    error!("Ping {}:{} failed! : {:?}", &host, port, error);
+                    error!("Ping {} failed! : {:?}", &host, error);
                 }
             }
-        })).await;
-    });
+        }
+    })).await;
 }
diff --git a/examples/pingpong.rs b/examples/pingpong.rs
@@ -6,7 +6,7 @@ SPDX-License-Identifier: GPL-3.0-only OR LGPL-3.0-only
 */
 
 use cardano_ouroboros_network::{
-    mux,
+    mux::Connection,
     protocols::pingpong,
 };
 use futures::executor::block_on;
@@ -17,8 +17,8 @@ fn main() {
     let cfg = common::init();
 
     block_on(async {
-        let channel = mux::connection::connect("127.0.0.1", cfg.port).await.unwrap();
-        channel.handshake(cfg.magic).await.unwrap();
-        channel.execute(pingpong::PingPongProtocol::new(0x0100)).await.unwrap();
+        let mut connection = Connection::tcp_connect("127.0.0.1:3001").await.unwrap();
+        connection.handshake(cfg.magic).await.unwrap();
+        connection.execute(&mut pingpong::PingPongProtocol::new(0x0100)).await.unwrap();
     });
 }
diff --git a/examples/server.rs b/examples/server.rs
@@ -6,44 +6,42 @@ SPDX-License-Identifier: GPL-3.0-only OR LGPL-3.0-only
 */
 
 use cardano_ouroboros_network::{
-    mux::connection::Channel,
+    mux::Connection,
     protocols::{
         handshake,
         pingpong,
     },
 };
-use std::net::{TcpListener, TcpStream};
-use log::{info, error};
-use futures::executor::block_on;
-use cardano_ouroboros_network::mux::connection::Stream::Tcp;
+use tokio::net::{TcpListener, TcpStream};
+use log::info;
 
 mod common;
 
-fn main() {
+async fn serve() {
     let cfg = common::init();
-    let listener = TcpListener::bind(format!("127.0.0.1:{}", cfg.port)).unwrap();
-
-    for stream in listener.incoming() {
-        match handle(stream.unwrap(), &cfg) {
-            Ok(_) => info!("connection closed"),
-            Err(e) => error!("connection failed: {}", e),
-        }
+    let listener = TcpListener::bind("127.0.0.1:3001").await.unwrap();
+    loop {
+        let (socket, _addr) = listener.accept().await.unwrap();
+        handle(socket, &cfg).await.unwrap();
     }
 }
 
 type Error = Box<dyn std::error::Error>;
 
-fn handle(stream: TcpStream, cfg: &common::Config) -> Result<(), Error> {
-    let channel = Channel::new(Tcp(stream));
-
+async fn handle(stream: TcpStream, cfg: &common::Config) -> Result<(), Error> {
     info!("new client!");
-    block_on(async {
-        channel.execute(handshake::HandshakeProtocol::builder()
-            .server()
-            .node_to_node()
-            .network_magic(cfg.magic)
-            .build()?).await?;
-        channel.execute(pingpong::PingPongProtocol::expect(0x0100)).await?;
-        Ok(())
-    })
+
+    let mut connection = Connection::from_tcp_stream(stream);
+    connection.execute(&mut handshake::HandshakeProtocol::builder()
+        .server()
+        .node_to_node()
+        .network_magic(cfg.magic)
+        .build()?).await?;
+    connection.execute(&mut pingpong::PingPongProtocol::expect(0x0100)).await?;
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() {
+    serve().await;
 }
diff --git a/examples/sync.rs b/examples/sync.rs
@@ -6,25 +6,23 @@ SPDX-License-Identifier: GPL-3.0-only OR LGPL-3.0-only
 */
 
 use cardano_ouroboros_network::{
-    mux,
+    mux::Connection,
     protocols::chainsync::{ChainSyncProtocol, Mode},
 };
-use futures::executor::block_on;
 
 mod common;
 mod sqlite;
 
-fn main() {
+#[tokio::main]
+async fn main() {
     let cfg = common::init();
 
-    block_on(async {
-        let channel = mux::connection::connect(&cfg.host, cfg.port).await.unwrap();
-        channel.handshake(cfg.magic).await.unwrap();
-        channel.execute({ChainSyncProtocol {
-            mode: Mode::Sync,
-            network_magic: cfg.magic,
-            store: Some(Box::new(sqlite::SQLiteBlockStore::new(&cfg.db).unwrap())),
-            ..Default::default()
-        }}).await.unwrap();
-    });
+    let mut connection = Connection::tcp_connect(&cfg.host).await.unwrap();
+    connection.handshake(cfg.magic).await.unwrap();
+    connection.execute(&mut ChainSyncProtocol {
+        mode: Mode::Sync,
+        network_magic: cfg.magic,
+        store: Some(Box::new(sqlite::SQLiteBlockStore::new(&cfg.db).unwrap())),
+        ..Default::default()
+    }).await.unwrap();
 }
diff --git a/examples/tip.rs b/examples/tip.rs
@@ -6,13 +6,10 @@ SPDX-License-Identifier: GPL-3.0-only OR LGPL-3.0-only
 */
 
 use cardano_ouroboros_network::{
+    mux::Connection,
     BlockHeader,
-    mux,
     protocols::chainsync::{ChainSyncProtocol, Mode, Listener},
 };
-use futures::{
-    executor::block_on,
-};
 use log::info;
 
 mod common;
@@ -25,17 +22,16 @@ impl Listener for Handler {
     }
 }
 
-fn main() {
+#[tokio::main]
+async fn main() {
     let cfg = common::init();
 
-    block_on(async {
-        let channel = mux::connection::connect(&cfg.host, cfg.port).await.unwrap();
-        channel.handshake(cfg.magic).await.unwrap();
-        channel.execute(ChainSyncProtocol {
-            mode: Mode::SendTip,
-            network_magic: cfg.magic,
-            notify: Some(Box::new(Handler {})),
-            ..Default::default()
-        }).await.unwrap();
-    });
+    let mut connection = Connection::tcp_connect(&cfg.host).await.unwrap();
+    connection.handshake(cfg.magic).await.unwrap();
+    connection.execute(&mut ChainSyncProtocol {
+        mode: Mode::SendTip,
+        network_magic: cfg.magic,
+        notify: Some(Box::new(Handler {})),
+        ..Default::default()
+    }).await.unwrap();
 }