Remove the futures dependencies (#468)

* No longer use futures::lock::Mutex

* Directly use the `futures-channel` dependency for channels

* Use futures-util::select! instead of futures::select!

* Import from futures_util rather than futures::prelude

* Forgot one usage of futures::lock

* Revert accidental usage of async_executor

* Final clean-ups

Author: tomaka

Cargo.lock

@@ -15,6 +15,7 @@ path = "src/main.rs"
 
 [dependencies]
 async-std = "1.12.0"
+async-lock = { version = "2.7.0" }
 atty = "0.2.14"
 blake2-rfc = { version = "0.2.18", default-features = false }
 clap = { version = "4.2.4", default-features = false, features = ["color", "derive", "help", "std", "suggestions", "usage"] }  # Note: enabling/disabling some features modifies the internal behavior of clap, be careful
@@ -25,8 +26,10 @@ either = { version = "1.8.1", default-features = false }
 env_logger = { version = "0.10.0", default-features = false, features = ["auto-color", "humantime"] }
 event-listener = { version = "2.5.3" }
 fnv = { version = "1.0.7", default-features = false }
-futures = { version = "0.3.27", default-features = false, features = ["std", "thread-pool"] }
+futures-channel = "0.3.27"
+futures-executor = { version = "0.3.28", default-features = false, features = ["std", "thread-pool"] }
 futures-timer = "3.0"
+futures-util = { version = "0.3.27", default-features = false }
 hashbrown = { version = "0.13.2", default-features = false }
 hex = { version = "0.4.3", default-features = false }
 log = { version = "0.4.17", default-features = false }

full-node/Cargo.toml

@@ -22,7 +22,7 @@ mod cli;
 mod run;
 
 fn main() {
-    futures::executor::block_on(async_main())
+    futures_executor::block_on(async_main())
 }
 
 async fn async_main() {

full-node/src/main.rs

@@ -17,7 +17,8 @@
 
 use crate::cli;
 
-use futures::{channel::oneshot, prelude::*};
+use futures_channel::oneshot;
+use futures_util::{future, stream, FutureExt as _, StreamExt as _};
 use smoldot::{
     chain, chain_spec,
     database::full_sqlite,
@@ -189,7 +190,7 @@ pub async fn run(cli_options: cli::CliOptionsRun) {
         .as_ref()
         .map(|relay_chain_spec| relay_chain_spec.as_chain_information().unwrap().0);
 
-    let threads_pool = futures::executor::ThreadPool::builder()
+    let threads_pool = futures_executor::ThreadPool::builder()
         .name_prefix("tasks-pool-")
         .create()
         .unwrap();
@@ -592,7 +593,7 @@ pub async fn run(cli_options: cli::CliOptionsRun) {
     debug_assert!(network_events_receivers.next().is_none());
 
     loop {
-        futures::select! {
+        futures_util::select! {
             _ = informant_timer.next() => {
                 if matches!(cli_output, cli::Output::Informant) {
                     // We end the informant line with a `\r` so that it overwrites itself every time.
@@ -826,7 +827,7 @@ async fn background_open_database(
     let mut next_progress_icon = ['-', '\\', '|', '/'].iter().copied().cycle();
 
     loop {
-        futures::select! {
+        futures_util::select! {
             res = rx => return res.unwrap(),
             _ = progress_timer.next() => {
                 if show_progress {

full-node/src/run.rs

@@ -26,8 +26,9 @@
 
 use crate::run::{database_thread, jaeger_service, network_service};
 
+use async_lock::Mutex;
 use core::{num::NonZeroU32, ops};
-use futures::{lock::Mutex, prelude::*};
+use futures_util::{future, stream, FutureExt as _, StreamExt as _};
 use hashbrown::HashSet;
 use smoldot::{
     author,
@@ -495,7 +496,7 @@ impl SyncBackground {
                 }
             };
 
-            futures::select! {
+            futures_util::select! {
                 () = authoring_ready_future => {
                     // Ready to author a block. Call `author_block()`.
                     // While a block is being authored, the whole syncing state machine is

full-node/src/run/consensus_service.rs

@@ -18,11 +18,9 @@
 //! As explained in the documentation of smoldot, the database uses synchronous I/O operations.
 //! For this reason, it is undesirable to access it from an asynchronous context.
 
-use futures::{
-    channel::{mpsc, oneshot},
-    lock::Mutex,
-    prelude::*,
-};
+use async_lock::Mutex;
+use futures_channel::{mpsc, oneshot};
+use futures_util::{SinkExt as _, StreamExt as _};
 use smoldot::database::full_sqlite::SqliteFullDatabase;
 use std::thread;
 
@@ -82,7 +80,7 @@ impl From<SqliteFullDatabase> for DatabaseThread {
             .spawn(move || {
                 // When the `DatabaseThread` is dropped, the sender will close, `rx.next()`
                 // will return `None`, and the closure here will finish, ending the thread.
-                while let Some(closure) = futures::executor::block_on(rx.next()) {
+                while let Some(closure) = futures_executor::block_on(rx.next()) {
                     closure(&db)
                 }
             })

full-node/src/run/database_thread.rs

@@ -35,7 +35,7 @@
 // TODO: more documentation
 
 use async_std::net::UdpSocket;
-use futures::prelude::*;
+use futures_util::future;
 use smoldot::libp2p::PeerId;
 use std::{convert::TryFrom as _, io, net::SocketAddr, num::NonZeroU128, sync::Arc};
 

full-node/src/run/jaeger_service.rs

@@ -15,7 +15,8 @@
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-use futures::{channel::oneshot, prelude::*};
+use futures_channel::oneshot;
+use futures_util::{future, FutureExt as _};
 use smoldot::json_rpc::{self, methods, websocket_server};
 use std::{io, net::SocketAddr};
 
@@ -93,7 +94,7 @@ struct JsonRpcBackground {
 impl JsonRpcBackground {
     async fn run(mut self) {
         loop {
-            let event = futures::select! {
+            let event = futures_util::select! {
                 _ = &mut self.client_still_alive => return,
                 event = self.server.next_event().fuse() => event,
             };

full-node/src/run/json_rpc_service.rs

@@ -29,12 +29,10 @@
 
 use crate::run::{database_thread, jaeger_service};
 
-use core::{cmp, mem, task::Poll, time::Duration};
-use futures::{
-    channel::{mpsc, oneshot},
-    lock::Mutex,
-    prelude::*,
-};
+use async_lock::Mutex;
+use core::{cmp, future::Future, mem, task::Poll, time::Duration};
+use futures_channel::{mpsc, oneshot};
+use futures_util::{future, stream, FutureExt as _, SinkExt as _, StreamExt as _};
 use hashbrown::HashMap;
 use smoldot::{
     database::full_sqlite,
@@ -462,7 +460,7 @@ impl NetworkService {
             let future = async move {
                 let mut connections = stream::FuturesUnordered::new();
                 loop {
-                    futures::select! {
+                    futures_util::select! {
                         new_connec = conn_tasks_rx.select_next_some() => {
                             connections.push(new_connec);
                         },

full-node/src/run/network_service.rs

@@ -17,9 +17,10 @@
 
 use super::Inner;
 
-use core::{pin, time::Duration};
-use futures::{channel::mpsc, prelude::*};
+use core::{future::Future, pin, time::Duration};
+use futures_channel::mpsc;
 use futures_timer::Delay;
+use futures_util::{future, AsyncRead, AsyncWrite, FutureExt as _, StreamExt as _};
 use smoldot::{
     libp2p::{
         async_std_connection::with_buffers,
@@ -74,7 +75,7 @@ pub(super) async fn opening_connection_task(
         })
         .fuse();
         let mut socket = pin::pin!(socket.fuse());
-        futures::select! {
+        futures_util::select! {
             _ = timeout => {
                 let mut guarded = inner.guarded.lock().await;
                 guarded.num_pending_out_attempts -= 1;
@@ -247,7 +248,7 @@ pub(super) async fn established_connection_task(
             // always remain ready until we push an element. While waiting, we process
             // incoming messages.
             loop {
-                futures::select! {
+                futures_util::select! {
                     _ = future::poll_fn(|cx| connection_to_coordinator.poll_ready(cx)).fuse() => break,
                     message = coordinator_to_connection.next() => {
                         if let Some(message) = message {