diff --git a/Cargo.lock b/Cargo.lock index 089f5b65..8914f5e7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -890,7 +890,7 @@ dependencies = [ "delegate", "env_logger", "extend", - "futures", + "futures-lite", "libp2p", "log", "networking", @@ -914,6 +914,12 @@ dependencies = [ "syn 2.0.111", ] +[[package]] +name = "fastrand" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" + [[package]] name = "ff" version = "0.13.1" @@ -1022,7 +1028,10 @@ version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" dependencies = [ + "fastrand", "futures-core", + "futures-io", + "parking", "pin-project-lite", ] @@ -2753,7 +2762,7 @@ dependencies = [ "delegate", "either", "extend", - "futures", + "futures-lite", "futures-timer", "keccak-const", "libp2p", diff --git a/Cargo.toml b/Cargo.toml index 4c0b8721..ffa9022c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,14 +29,13 @@ util = { path = "rust/util" } # Macro dependecies extend = "1.2" delegate = "0.13" -pin-project = "1" # Utility dependencies keccak-const = "0.2" # Async dependencies tokio = "1.46" -futures = "0.3" +futures-lite = "2.6.1" futures-timer = "3.0" # Data structures diff --git a/flake.nix b/flake.nix index e90e0bd2..3bf7d912 100644 --- a/flake.nix +++ b/flake.nix @@ -74,7 +74,6 @@ perSystem = { config, self', inputs', pkgs, lib, system, ... }: let - fenixToolchain = inputs'.fenix.packages.complete; # Use pinned nixpkgs for swift-format (swift is broken on x86_64-linux in newer nixpkgs) pkgsSwift = import inputs.nixpkgs-swift { inherit system; }; in diff --git a/rust/clippy.toml b/rust/clippy.toml deleted file mode 100644 index 6d5a6187..00000000 --- a/rust/clippy.toml +++ /dev/null @@ -1,2 +0,0 @@ -# we can manually exclude false-positive lint errors for dual packages (if in dependencies) -#allowed-duplicate-crates = ["hashbrown"] \ No newline at end of file diff --git a/rust/exo_pyo3_bindings/Cargo.toml b/rust/exo_pyo3_bindings/Cargo.toml index e97b9148..77777055 100644 --- a/rust/exo_pyo3_bindings/Cargo.toml +++ b/rust/exo_pyo3_bindings/Cargo.toml @@ -27,7 +27,7 @@ networking = { workspace = true } # interop pyo3 = { version = "0.27.2", features = [ # "abi3-py313", # tells pyo3 (and maturin) to build using the stable ABI with minimum Python version 3.13 - "nightly", # enables better-supported GIL integration + # "nightly", # enables better-supported GIL integration "experimental-async", # async support in #[pyfunction] & #[pymethods] #"experimental-inspect", # inspection of generated binary => easier to automate type-hint generation #"py-clone", # adding Clone-ing of `Py` without GIL (may cause panics - remove if panics happen) @@ -45,11 +45,10 @@ pyo3-log = "0.13.2" # macro dependencies extend = { workspace = true } delegate = { workspace = true } -pin-project = { workspace = true } # async runtime tokio = { workspace = true, features = ["full", "tracing"] } -futures = { workspace = true } +futures-lite = { workspace = true } # utility dependencies util = { workspace = true } @@ -60,3 +59,4 @@ env_logger = "0.11" # Networking libp2p = { workspace = true, features = ["full"] } +pin-project = "1.1.10" diff --git a/rust/exo_pyo3_bindings/src/allow_threading.rs b/rust/exo_pyo3_bindings/src/allow_threading.rs index 18a426aa..142ff0c6 100644 --- a/rust/exo_pyo3_bindings/src/allow_threading.rs +++ b/rust/exo_pyo3_bindings/src/allow_threading.rs @@ -2,7 +2,6 @@ //! use pin_project::pin_project; -use pyo3::marker::Ungil; use pyo3::prelude::*; use std::{ future::Future, @@ -26,8 +25,8 @@ where impl Future for AllowThreads where - F: Future + Ungil, - F::Output: Ungil, + F: Future + Send, + F::Output: Send, { type Output = F::Output; diff --git a/rust/exo_pyo3_bindings/src/pylibp2p/ident.rs b/rust/exo_pyo3_bindings/src/ident.rs similarity index 100% rename from rust/exo_pyo3_bindings/src/pylibp2p/ident.rs rename to rust/exo_pyo3_bindings/src/ident.rs diff --git a/rust/exo_pyo3_bindings/src/lib.rs b/rust/exo_pyo3_bindings/src/lib.rs index 25f2865e..45825b21 100644 --- a/rust/exo_pyo3_bindings/src/lib.rs +++ b/rust/exo_pyo3_bindings/src/lib.rs @@ -4,25 +4,12 @@ //! //! -// enable Rust-unstable features for convenience -#![feature(trait_alias)] -#![feature(tuple_trait)] -#![feature(unboxed_closures)] -// #![feature(stmt_expr_attributes)] -// #![feature(assert_matches)] -// #![feature(async_fn_in_dyn_trait)] -// #![feature(async_for_loop)] -// #![feature(auto_traits)] -// #![feature(negative_impls)] - -extern crate core; mod allow_threading; -pub(crate) mod networking; -pub(crate) mod pylibp2p; +mod ident; +mod networking; +use crate::ident::ident_submodule; use crate::networking::networking_submodule; -use crate::pylibp2p::ident::ident_submodule; -use crate::pylibp2p::multiaddr::multiaddr_submodule; use pyo3::prelude::PyModule; use pyo3::{Bound, PyResult, pyclass, pymodule}; use pyo3_stub_gen::define_stub_info_gatherer; @@ -32,14 +19,6 @@ pub(crate) mod r#const { pub const MPSC_CHANNEL_SIZE: usize = 1024; } -/// Namespace for all the type/trait aliases used by this crate. -pub(crate) mod alias { - use std::marker::Tuple; - - pub trait SendFn = - Fn + Send + 'static; -} - /// Namespace for crate-wide extension traits/methods pub(crate) mod ext { use crate::allow_threading::AllowThreads; @@ -180,7 +159,6 @@ fn main_module(m: &Bound<'_, PyModule>) -> PyResult<()> { // work with maturin, where the types generate correctly, in the right folder, without // too many importing issues... ident_submodule(m)?; - multiaddr_submodule(m)?; networking_submodule(m)?; // top-level constructs diff --git a/rust/exo_pyo3_bindings/src/networking.rs b/rust/exo_pyo3_bindings/src/networking.rs index 03a0dcb6..b864d876 100644 --- a/rust/exo_pyo3_bindings/src/networking.rs +++ b/rust/exo_pyo3_bindings/src/networking.rs @@ -8,8 +8,8 @@ use crate::r#const::MPSC_CHANNEL_SIZE; use crate::ext::{ByteArrayExt as _, FutureExt, PyErrExt as _}; use crate::ext::{ResultExt as _, TokioMpscReceiverExt as _, TokioMpscSenderExt as _}; +use crate::ident::{PyKeypair, PyPeerId}; use crate::pyclass; -use crate::pylibp2p::ident::{PyKeypair, PyPeerId}; use libp2p::futures::StreamExt as _; use libp2p::gossipsub; use libp2p::gossipsub::{IdentTopic, Message, MessageId, PublishError}; diff --git a/rust/exo_pyo3_bindings/src/pylibp2p/mod.rs b/rust/exo_pyo3_bindings/src/pylibp2p/mod.rs deleted file mode 100644 index 8eb1bdc0..00000000 --- a/rust/exo_pyo3_bindings/src/pylibp2p/mod.rs +++ /dev/null @@ -1,8 +0,0 @@ -//! A module for exposing Rust's libp2p datatypes over Pyo3 -//! -//! TODO: right now we are coupled to libp2p's identity, but eventually we want to create our own -//! independent identity type of some kind or another. This may require handshaking. -//! - -pub mod ident; -pub mod multiaddr; diff --git a/rust/exo_pyo3_bindings/src/pylibp2p/multiaddr.rs b/rust/exo_pyo3_bindings/src/pylibp2p/multiaddr.rs deleted file mode 100644 index 4d398b53..00000000 --- a/rust/exo_pyo3_bindings/src/pylibp2p/multiaddr.rs +++ /dev/null @@ -1,81 +0,0 @@ -use crate::ext::ResultExt as _; -use libp2p::Multiaddr; -use pyo3::prelude::{PyBytesMethods as _, PyModule, PyModuleMethods as _}; -use pyo3::types::PyBytes; -use pyo3::{Bound, PyResult, Python, pyclass, pymethods}; -use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; -use std::str::FromStr as _; - -/// Representation of a Multiaddr. -#[gen_stub_pyclass] -#[pyclass(name = "Multiaddr", frozen)] -#[derive(Debug, Clone)] -#[repr(transparent)] -pub struct PyMultiaddr(pub Multiaddr); - -#[gen_stub_pymethods] -#[pymethods] -#[allow(clippy::needless_pass_by_value)] -impl PyMultiaddr { - /// Create a new, empty multiaddress. - #[staticmethod] - fn empty() -> Self { - Self(Multiaddr::empty()) - } - - /// Create a new, empty multiaddress with the given capacity. - #[staticmethod] - fn with_capacity(n: usize) -> Self { - Self(Multiaddr::with_capacity(n)) - } - - /// Parse a `Multiaddr` value from its byte slice representation. - #[staticmethod] - fn from_bytes(bytes: Bound<'_, PyBytes>) -> PyResult { - let bytes = Vec::from(bytes.as_bytes()); - Ok(Self(Multiaddr::try_from(bytes).pyerr()?)) - } - - /// Parse a `Multiaddr` value from its string representation. - #[staticmethod] - fn from_string(string: String) -> PyResult { - Ok(Self(Multiaddr::from_str(&string).pyerr()?)) - } - - /// Return the length in bytes of this multiaddress. - fn len(&self) -> usize { - self.0.len() - } - - /// Returns true if the length of this multiaddress is 0. - fn is_empty(&self) -> bool { - self.0.is_empty() - } - - /// Return a copy of this [`Multiaddr`]'s byte representation. - fn to_bytes<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> { - let bytes = self.0.to_vec(); - PyBytes::new(py, &bytes) - } - - /// Convert a Multiaddr to a string. - fn to_string(&self) -> String { - self.0.to_string() - } - - #[gen_stub(skip)] - fn __repr__(&self) -> String { - format!("Multiaddr({})", self.0) - } - - #[gen_stub(skip)] - fn __str__(&self) -> String { - self.to_string() - } -} - -pub fn multiaddr_submodule(m: &Bound<'_, PyModule>) -> PyResult<()> { - m.add_class::()?; - - Ok(()) -} diff --git a/rust/networking/Cargo.toml b/rust/networking/Cargo.toml index fd5f1b1f..58ca4e0a 100644 --- a/rust/networking/Cargo.toml +++ b/rust/networking/Cargo.toml @@ -22,7 +22,7 @@ delegate = { workspace = true } # async tokio = { workspace = true, features = ["full"] } -futures = { workspace = true } +futures-lite = { workspace = true } futures-timer = { workspace = true } # utility dependencies diff --git a/rust/networking/examples/chatroom.rs b/rust/networking/examples/chatroom.rs index 3371b46d..99067e9b 100644 --- a/rust/networking/examples/chatroom.rs +++ b/rust/networking/examples/chatroom.rs @@ -1,4 +1,4 @@ -use futures::stream::StreamExt as _; +use futures_lite::StreamExt; use libp2p::{gossipsub, identity, swarm::SwarmEvent}; use networking::{discovery, swarm}; use tokio::{io, io::AsyncBufReadExt as _, select}; @@ -38,19 +38,19 @@ async fn main() { println!("Publish error: {e:?}"); } } - event = swarm.select_next_some() => match event { + event = swarm.next() => match event { // on gossipsub incoming - SwarmEvent::Behaviour(swarm::BehaviourEvent::Gossipsub(gossipsub::Event::Message { + Some(SwarmEvent::Behaviour(swarm::BehaviourEvent::Gossipsub(gossipsub::Event::Message { propagation_source: peer_id, message_id: id, message, - })) => println!( + }))) => println!( "\n\nGot message: '{}' with id: {id} from peer: {peer_id}\n\n", String::from_utf8_lossy(&message.data), ), // on discovery - SwarmEvent::Behaviour(swarm::BehaviourEvent::Discovery(e)) => match e { + Some(SwarmEvent::Behaviour(swarm::BehaviourEvent::Discovery(e)) )=> match e { discovery::Event::ConnectionEstablished { peer_id, connection_id, remote_ip, remote_tcp_port } => { @@ -64,7 +64,7 @@ async fn main() { } // ignore outgoing errors: those are normal - e@SwarmEvent::OutgoingConnectionError { .. } => { log::debug!("Outgoing connection error: {e:?}"); } + e@Some(SwarmEvent::OutgoingConnectionError { .. }) => { log::debug!("Outgoing connection error: {e:?}"); } // otherwise log any other event e => { log::info!("Other event {e:?}"); } diff --git a/rust/networking/examples/chatroom_manual.rs b/rust/networking/examples/chatroom_manual.rs deleted file mode 100644 index e9f10a1f..00000000 --- a/rust/networking/examples/chatroom_manual.rs +++ /dev/null @@ -1,127 +0,0 @@ -// Copyright 2018 Parity Technologies (UK) Ltd. -// -// Permission is hereby granted, free of charge, to any person obtaining a -// copy of this software and associated documentation files (the "Software"), -// to deal in the Software without restriction, including without limitation -// the rights to use, copy, modify, merge, publish, distribute, sublicense, -// and/or sell copies of the Software, and to permit persons to whom the -// Software is furnished to do so, subject to the following conditions: -// -// The above copyright notice and this permission notice shall be included in -// all copies or substantial portions of the Software. -// -// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -// DEALINGS IN THE SOFTWARE. - -use futures::stream::StreamExt; -use libp2p::{ - gossipsub, mdns, noise, - swarm::{NetworkBehaviour, SwarmEvent}, - tcp, yamux, -}; -use std::error::Error; -use std::time::Duration; -use tokio::{io, io::AsyncBufReadExt, select}; -use tracing_subscriber::EnvFilter; - -// We create a custom network behaviour that combines Gossipsub and Mdns. -#[derive(NetworkBehaviour)] -struct MyBehaviour { - gossipsub: gossipsub::Behaviour, - mdns: mdns::tokio::Behaviour, -} - -#[tokio::main] -async fn main() -> Result<(), Box> { - let _ = tracing_subscriber::fmt() - .with_env_filter(EnvFilter::from_default_env()) - .try_init(); - - let mut swarm = libp2p::SwarmBuilder::with_new_identity() - .with_tokio() - .with_tcp( - tcp::Config::default(), - noise::Config::new, - yamux::Config::default, - )? - .with_behaviour(|key| { - // Set a custom gossipsub configuration - let gossipsub_config = gossipsub::ConfigBuilder::default() - .heartbeat_interval(Duration::from_secs(10)) - .validation_mode(gossipsub::ValidationMode::Strict) // This sets the kind of message validation. The default is Strict (enforce message signing) - .build() - .map_err(io::Error::other)?; // Temporary hack because `build` does not return a proper `std::error::Error`. - - // build a gossipsub network behaviour - let gossipsub = gossipsub::Behaviour::new( - gossipsub::MessageAuthenticity::Signed(key.clone()), - gossipsub_config, - )?; - - let mdns = - mdns::tokio::Behaviour::new(mdns::Config::default(), key.public().to_peer_id())?; - Ok(MyBehaviour { gossipsub, mdns }) - })? - .build(); - - println!("Running swarm with identity {}", swarm.local_peer_id()); - - // Create a Gossipsub topic - let topic = gossipsub::IdentTopic::new("test-net"); - // subscribes to our topic - swarm.behaviour_mut().gossipsub.subscribe(&topic)?; - - // Read full lines from stdin - let mut stdin = io::BufReader::new(io::stdin()).lines(); - - // Listen on all interfaces and whatever port the OS assigns - swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?; - - println!("Enter messages via STDIN and they will be sent to connected peers using Gossipsub"); - - // Kick it off - loop { - select! { - Ok(Some(line)) = stdin.next_line() => { - if let Err(e) = swarm - .behaviour_mut().gossipsub - .publish(topic.clone(), line.as_bytes()) { - println!("Publish error: {e:?}"); - } - } - event = swarm.select_next_some() => match event { - SwarmEvent::Behaviour(MyBehaviourEvent::Mdns(mdns::Event::Discovered(list))) => { - for (peer_id, multiaddr) in list { - println!("mDNS discovered a new peer: {peer_id} on {multiaddr}"); - swarm.behaviour_mut().gossipsub.add_explicit_peer(&peer_id); - } - }, - SwarmEvent::Behaviour(MyBehaviourEvent::Mdns(mdns::Event::Expired(list))) => { - for (peer_id, multiaddr) in list { - println!("mDNS discover peer has expired: {peer_id} on {multiaddr}"); - swarm.behaviour_mut().gossipsub.remove_explicit_peer(&peer_id); - } - }, - SwarmEvent::Behaviour(MyBehaviourEvent::Gossipsub(gossipsub::Event::Message { - propagation_source: peer_id, - message_id: id, - message, - })) => println!( - "Got message: '{}' with id: {id} from peer: {peer_id}", - String::from_utf8_lossy(&message.data), - ), - SwarmEvent::NewListenAddr { address, .. } => { - println!("Local node is listening on {address}"); - } - e => { - println!("Other swarm event: {:?}", e); - } - } - } - } -} diff --git a/rust/networking/src/discovery.rs b/rust/networking/src/discovery.rs index 581f2200..a6cbd0e0 100644 --- a/rust/networking/src/discovery.rs +++ b/rust/networking/src/discovery.rs @@ -1,7 +1,7 @@ use crate::ext::MultiaddrExt; use delegate::delegate; use either::Either; -use futures::FutureExt; +use futures_lite::FutureExt; use futures_timer::Delay; use libp2p::core::transport::PortUse; use libp2p::core::{ConnectedPoint, Endpoint}; @@ -362,7 +362,7 @@ impl NetworkBehaviour for Behaviour { } // retry connecting to all mDNS peers periodically (fails safely if already connected) - if self.retry_delay.poll_unpin(cx).is_ready() { + if self.retry_delay.poll(cx).is_ready() { for (p, mas) in self.mdns_discovered.clone() { for ma in mas { self.dial(p, ma) diff --git a/rust/networking/src/swarm.rs b/rust/networking/src/swarm.rs index 33cf8966..fc734dc3 100644 --- a/rust/networking/src/swarm.rs +++ b/rust/networking/src/swarm.rs @@ -31,7 +31,7 @@ pub fn create_swarm(keypair: identity::Keypair) -> alias::AnyResult { mod transport { use crate::alias; use crate::swarm::{NETWORK_VERSION, OVERRIDE_VERSION_ENV_VAR}; - use futures::{AsyncRead, AsyncWrite}; + use futures_lite::{AsyncRead, AsyncWrite}; use keccak_const::Sha3_256; use libp2p::core::muxing; use libp2p::core::transport::Boxed; diff --git a/rust/parts.nix b/rust/parts.nix index f160da8e..f5272b72 100644 --- a/rust/parts.nix +++ b/rust/parts.nix @@ -1,11 +1,10 @@ { inputs, ... }: { perSystem = - { config, self', inputs', pkgs, lib, ... }: + { inputs', pkgs, lib, ... }: let # Fenix nightly toolchain with all components - fenixPkgs = inputs'.fenix.packages; - rustToolchain = fenixPkgs.complete.withComponents [ + rustToolchain = inputs'.fenix.packages.stable.withComponents [ "cargo" "rustc" "clippy" diff --git a/rust/rust-toolchain.toml b/rust/rust-toolchain.toml deleted file mode 100644 index 271800cb..00000000 --- a/rust/rust-toolchain.toml +++ /dev/null @@ -1,2 +0,0 @@ -[toolchain] -channel = "nightly" \ No newline at end of file