fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-06-26 08:11:39 +00:00
Code Issues Projects Releases Wiki Activity

feat(swarm): Make executor for connection tasks explicit (#3097)

Browse Source 
Previously, the executor for connection tasks silently defaulted to a `futures::executor::ThreadPool`. This causes issues such as https://github.com/libp2p/rust-libp2p/issues/2230.

With this patch, we force the user to choose, which executor they want to run the connection tasks on which results in overall simpler API with less footguns.

Closes #3068.
This commit is contained in:
Hannes
2022-11-15 15:26:03 +01:00
committed by GitHub
parent d8fe7bf49f
commit d5ea93dd71
41 changed files with 384 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
swarm/src/connection/pool.rs
View File

@ -54,10 +54,34 @@ use void::Void;
mod concurrent_dial;
mod task;
enum ExecSwitch {
Executor(Box<dyn Executor + Send>),
LocalSpawn(FuturesUnordered<Pin<Box<dyn Future<Output = ()> + Send>>>),
}
impl ExecSwitch {
fn advance_local(&mut self, cx: &mut Context) {
match self {
ExecSwitch::Executor(_) => {}
ExecSwitch::LocalSpawn(local) => {
while let Poll::Ready(Some(())) = local.poll_next_unpin(cx) {}
}
}
}
fn spawn(&mut self, task: BoxFuture<'static, ()>) {
match self {
Self::Executor(executor) => executor.exec(task),
Self::LocalSpawn(local) => local.push(task),
}
}
}
/// A connection `Pool` manages a set of connections for each peer.
pub struct Pool<THandler: IntoConnectionHandler, TTrans>
pub struct Pool<THandler, TTrans>
where
TTrans: Transport,
THandler: IntoConnectionHandler,
{
local_id: PeerId,
@ -93,14 +117,9 @@ where
/// See [`Connection::max_negotiating_inbound_streams`].
max_negotiating_inbound_streams: usize,
/// The executor to use for running the background tasks. If `None`,
/// the tasks are kept in `local_spawns` instead and polled on the
/// current thread when the [`Pool`] is polled for new events.
executor: Option<Box<dyn Executor + Send>>,
/// If no `executor` is configured, tasks are kept in this set and
/// polled on the current thread when the [`Pool`] is polled for new events.
local_spawns: FuturesUnordered<Pin<Box<dyn Future<Output = ()> + Send>>>,
/// The executor to use for running connection tasks. Can either be a global executor
/// or a local queue.
executor: ExecSwitch,
/// Sender distributed to pending tasks for reporting events back
/// to the pool.
@ -299,6 +318,10 @@ where
mpsc::channel(config.task_event_buffer_size);
let (established_connection_events_tx, established_connection_events_rx) =
mpsc::channel(config.task_event_buffer_size);
let executor = match config.executor {
Some(exec) => ExecSwitch::Executor(exec),
None => ExecSwitch::LocalSpawn(Default::default()),
};
Pool {
local_id,
counters: ConnectionCounters::new(limits),
@ -309,8 +332,7 @@ where
dial_concurrency_factor: config.dial_concurrency_factor,
substream_upgrade_protocol_override: config.substream_upgrade_protocol_override,
max_negotiating_inbound_streams: config.max_negotiating_inbound_streams,
executor: config.executor,
local_spawns: FuturesUnordered::new(),
executor,
pending_connection_events_tx,
pending_connection_events_rx,
established_connection_events_tx,
@ -399,11 +421,7 @@ where
}
fn spawn(&mut self, task: BoxFuture<'static, ()>) {
if let Some(executor) = &mut self.executor {
executor.exec(task);
} else {
self.local_spawns.push(task);
}
self.executor.spawn(task)
}
}
@ -820,8 +838,7 @@ where
}
}
// Advance the tasks in `local_spawns`.
while let Poll::Ready(Some(())) = self.local_spawns.poll_next_unpin(cx) {}
self.executor.advance_local(cx);
Poll::Pending
}
@ -1073,34 +1090,21 @@ pub struct PoolConfig {
max_negotiating_inbound_streams: usize,
}
impl Default for PoolConfig {
fn default() -> Self {
PoolConfig {
executor: None,
task_event_buffer_size: 32,
task_command_buffer_size: 7,
// Set to a default of 8 based on frequency of dialer connections
impl PoolConfig {
pub fn new(executor: Option<Box<dyn Executor + Send>>) -> Self {
Self {
executor,
task_command_buffer_size: 32,
task_event_buffer_size: 7,
dial_concurrency_factor: NonZeroU8::new(8).expect("8 > 0"),
substream_upgrade_protocol_override: None,
max_negotiating_inbound_streams: 128,
}
}
}
impl PoolConfig {
/// Configures the executor to use for spawning connection background tasks.
pub fn with_executor(mut self, e: Box<dyn Executor + Send>) -> Self {
self.executor = Some(e);
self
}
/// Configures the executor to use for spawning connection background tasks,
/// only if no executor has already been configured.
pub fn or_else_with_executor<F>(mut self, f: F) -> Self
where
F: FnOnce() -> Option<Box<dyn Executor + Send>>,
{
self.executor = self.executor.or_else(f);
pub fn with_executor(mut self, executor: Box<dyn Executor + Send>) -> Self {
self.executor = Some(executor);
self
}
@ -1174,13 +1178,4 @@ mod tests {
impl Executor for Dummy {
fn exec(&self, _: Pin<Box<dyn Future<Output = ()> + Send>>) {}
}
#[test]
fn set_executor() {
PoolConfig::default()
.with_executor(Box::new(Dummy))
.with_executor(Box::new(|f| {
async_std::task::spawn(f);
}));
}
}

48
swarm/src/executor.rs Normal file
View File

@ -0,0 +1,48 @@
use futures::executor::ThreadPool;
use std::{future::Future, pin::Pin};
/// Implemented on objects that can run a `Future` in the background.
///
/// > **Note**: While it may be tempting to implement this trait on types such as
/// > [`futures::stream::FuturesUnordered`], please note that passing an `Executor` is
/// > optional, and that `FuturesUnordered` (or a similar struct) will automatically
/// > be used as fallback by libp2p. The `Executor` trait should therefore only be
/// > about running `Future`s on a separate task.
pub trait Executor {
/// Run the given future in the background until it ends.
fn exec(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>);
}
impl<F: Fn(Pin<Box<dyn Future<Output = ()> + Send>>)> Executor for F {
fn exec(&self, f: Pin<Box<dyn Future<Output = ()> + Send>>) {
self(f)
}
}
impl Executor for ThreadPool {
fn exec(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
self.spawn_ok(future)
}
}
#[cfg(feature = "tokio")]
#[derive(Default, Debug, Clone, Copy)]
pub(crate) struct TokioExecutor;
#[cfg(feature = "tokio")]
impl Executor for TokioExecutor {
fn exec(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
let _ = tokio::spawn(future);
}
}
#[cfg(feature = "async-std")]
#[derive(Default, Debug, Clone, Copy)]
pub(crate) struct AsyncStdExecutor;
#[cfg(feature = "async-std")]
impl Executor for AsyncStdExecutor {
fn exec(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
let _ = async_std::task::spawn(future);
}
}

166
swarm/src/lib.rs
View File

@ -64,6 +64,7 @@ mod upgrade;
pub mod behaviour;
pub mod dial_opts;
pub mod dummy;
mod executor;
pub mod handler;
pub mod keep_alive;
@ -94,6 +95,7 @@ pub use connection::{
ConnectionError, ConnectionLimit, PendingConnectionError, PendingInboundConnectionError,
PendingOutboundConnectionError,
};
pub use executor::Executor;
pub use handler::{
ConnectionHandler, ConnectionHandlerEvent, ConnectionHandlerSelect, ConnectionHandlerUpgrErr,
IntoConnectionHandler, IntoConnectionHandlerSelect, KeepAlive, OneShotHandler,
@ -117,7 +119,7 @@ use libp2p_core::{
muxing::StreamMuxerBox,
transport::{self, ListenerId, TransportError, TransportEvent},
upgrade::ProtocolName,
Endpoint, Executor, Multiaddr, Negotiated, PeerId, Transport,
Endpoint, Multiaddr, Negotiated, PeerId, Transport,
};
use registry::{AddressIntoIter, Addresses};
use smallvec::SmallVec;
@ -328,12 +330,89 @@ where
TBehaviour: NetworkBehaviour,
{
/// Builds a new `Swarm`.
#[deprecated(
since = "0.41.0",
note = "This constructor is considered ambiguous regarding the executor. Use one of the new, executor-specific constructors or `Swarm::with_threadpool_executor` for the same behaviour."
)]
pub fn new(
transport: transport::Boxed<(PeerId, StreamMuxerBox)>,
behaviour: TBehaviour,
local_peer_id: PeerId,
) -> Self {
SwarmBuilder::new(transport, behaviour, local_peer_id).build()
Self::with_threadpool_executor(transport, behaviour, local_peer_id)
}
/// Builds a new `Swarm` with a provided executor.
pub fn with_executor(
transport: transport::Boxed<(PeerId, StreamMuxerBox)>,
behaviour: TBehaviour,
local_peer_id: PeerId,
executor: impl Executor + Send + 'static,
) -> Self {
SwarmBuilder::with_executor(transport, behaviour, local_peer_id, executor).build()
}
/// Builds a new `Swarm` with a tokio executor.
#[cfg(feature = "tokio")]
pub fn with_tokio_executor(
transport: transport::Boxed<(PeerId, StreamMuxerBox)>,
behaviour: TBehaviour,
local_peer_id: PeerId,
) -> Self {
Self::with_executor(
transport,
behaviour,
local_peer_id,
crate::executor::TokioExecutor,
)
}
/// Builds a new `Swarm` with an async-std executor.
#[cfg(feature = "async-std")]
pub fn with_async_std_executor(
transport: transport::Boxed<(PeerId, StreamMuxerBox)>,
behaviour: TBehaviour,
local_peer_id: PeerId,
) -> Self {
Self::with_executor(
transport,
behaviour,
local_peer_id,
crate::executor::AsyncStdExecutor,
)
}
/// Builds a new `Swarm` with a threadpool executor.
pub fn with_threadpool_executor(
transport: transport::Boxed<(PeerId, StreamMuxerBox)>,
behaviour: TBehaviour,
local_peer_id: PeerId,
) -> Self {
let builder = match ThreadPoolBuilder::new()
.name_prefix("libp2p-swarm-task-")
.create()
{
Ok(tp) => SwarmBuilder::with_executor(transport, behaviour, local_peer_id, tp),
Err(err) => {
log::warn!("Failed to create executor thread pool: {:?}", err);
SwarmBuilder::without_executor(transport, behaviour, local_peer_id)
}
};
builder.build()
}
/// Builds a new `Swarm` without an executor, instead using the current task.
///
/// ## ⚠️ Performance warning
/// All connections will be polled on the current task, thus quite bad performance
/// characteristics should be expected. Whenever possible use an executor and
/// [`Swarm::with_executor`].
pub fn without_executor(
transport: transport::Boxed<(PeerId, StreamMuxerBox)>,
behaviour: TBehaviour,
local_peer_id: PeerId,
) -> Self {
SwarmBuilder::without_executor(transport, behaviour, local_peer_id).build()
}
/// Returns information about the connections underlying the [`Swarm`].
@ -1294,16 +1373,67 @@ where
/// Creates a new `SwarmBuilder` from the given transport, behaviour and
/// local peer ID. The `Swarm` with its underlying `Network` is obtained
/// via [`SwarmBuilder::build`].
#[deprecated(
since = "0.41.0",
note = "Use `SwarmBuilder::with_executor` or `SwarmBuilder::without_executor` instead."
)]
pub fn new(
transport: transport::Boxed<(PeerId, StreamMuxerBox)>,
behaviour: TBehaviour,
local_peer_id: PeerId,
) -> Self {
let executor: Option<Box<dyn Executor + Send>> = match ThreadPoolBuilder::new()
.name_prefix("libp2p-swarm-task-")
.create()
.ok()
{
Some(tp) => Some(Box::new(tp)),
None => None,
};
SwarmBuilder {
local_peer_id,
transport,
behaviour,
pool_config: Default::default(),
pool_config: PoolConfig::new(executor),
connection_limits: Default::default(),
}
}
/// Creates a new [`SwarmBuilder`] from the given transport, behaviour, local peer ID and
/// executor. The `Swarm` with its underlying `Network` is obtained via
/// [`SwarmBuilder::build`].
pub fn with_executor(
transport: transport::Boxed<(PeerId, StreamMuxerBox)>,
behaviour: TBehaviour,
local_peer_id: PeerId,
executor: impl Executor + Send + 'static,
) -> Self {
Self {
local_peer_id,
transport,
behaviour,
pool_config: PoolConfig::new(Some(Box::new(executor))),
connection_limits: Default::default(),
}
}
/// Creates a new [`SwarmBuilder`] from the given transport, behaviour and local peer ID. The
/// `Swarm` with its underlying `Network` is obtained via [`SwarmBuilder::build`].
///
/// ## ⚠️ Performance warning
/// All connections will be polled on the current task, thus quite bad performance
/// characteristics should be expected. Whenever possible use an executor and
/// [`SwarmBuilder::with_executor`].
pub fn without_executor(
transport: transport::Boxed<(PeerId, StreamMuxerBox)>,
behaviour: TBehaviour,
local_peer_id: PeerId,
) -> Self {
Self {
local_peer_id,
transport,
behaviour,
pool_config: PoolConfig::new(None),
connection_limits: Default::default(),
}
}
@ -1313,8 +1443,9 @@ where
/// By default, unless another executor has been configured,
/// [`SwarmBuilder::build`] will try to set up a
/// [`ThreadPool`](futures::executor::ThreadPool).
pub fn executor(mut self, e: Box<dyn Executor + Send>) -> Self {
self.pool_config = self.pool_config.with_executor(e);
#[deprecated(since = "0.41.0", note = "Use `SwarmBuilder::with_executor` instead.")]
pub fn executor(mut self, executor: Box<dyn Executor + Send>) -> Self {
self.pool_config = self.pool_config.with_executor(executor);
self
}
@ -1412,25 +1543,10 @@ where
.map(|info| info.protocol_name().to_vec())
.collect();
// If no executor has been explicitly configured, try to set up a thread pool.
let pool_config =
self.pool_config.or_else_with_executor(|| {
match ThreadPoolBuilder::new()
.name_prefix("libp2p-swarm-task-")
.create()
{
Ok(tp) => Some(Box::new(move |f| tp.spawn_ok(f))),
Err(err) => {
log::warn!("Failed to create executor thread pool: {:?}", err);
None
}
}
});
Swarm {
local_peer_id: self.local_peer_id,
transport: self.transport,
pool: Pool::new(self.local_peer_id, pool_config, self.connection_limits),
pool: Pool::new(self.local_peer_id, self.pool_config, self.connection_limits),
behaviour: self.behaviour,
supported_protocols,
listened_addrs: HashMap::new(),
@ -1586,6 +1702,7 @@ mod tests {
use super::*;
use crate::test::{CallTraceBehaviour, MockBehaviour};
use futures::executor::block_on;
use futures::executor::ThreadPool;
use futures::future::poll_fn;
use futures::future::Either;
use futures::{executor, future, ready};
@ -1622,7 +1739,12 @@ mod tests {
.multiplex(yamux::YamuxConfig::default())
.boxed();
let behaviour = CallTraceBehaviour::new(MockBehaviour::new(handler_proto));
SwarmBuilder::new(transport, behaviour, local_public_key.into())
match ThreadPool::new().ok() {
Some(tp) => {
SwarmBuilder::with_executor(transport, behaviour, local_public_key.into(), tp)
}
None => SwarmBuilder::without_executor(transport, behaviour, local_public_key.into()),
}
}
fn swarms_connected<TBehaviour>(