test(common): load tests for bounded pool and stream opening

crates/common/src/context.rs

@@ -8,6 +8,7 @@ use std::sync::Arc;
 use futures::{
     AsyncRead, AsyncWrite,
     future::{self, BoxFuture, Either},
+    stream::{self, StreamExt, TryStreamExt},
 };
 
 #[cfg(any(test, feature = "test-utils"))]
@@ -19,7 +20,12 @@ pub use test::{
     test_mt_context, test_mt_context_with_spawn, test_st_context,
 };
 
-use crate::{ContextId, executor::Inner, io::Io, mux::Mux};
+use crate::{ContextId, io::Io, mux::Mux, thread_pool::ThreadPool};
+
+/// Default maximum number of [`map`](Context::map) items processed
+/// concurrently. Both parties must agree on this value, so it is a fixed
+/// constant rather than data- or timing-dependent.
+pub const DEFAULT_CONCURRENCY_LIMIT: usize = 32;
 
 /// A task execution context.
 ///
@@ -42,42 +48,31 @@ enum Mode {
     Single,
     Multi {
         mux: Arc<dyn Mux + Send + Sync>,
-        executor: Option<Arc<Inner>>,
+        /// Pool for parallel execution; `None` runs sub-tasks cooperatively
+        /// on the caller's future.
+        pool: Option<ThreadPool>,
+        /// Maximum number of [`map`](Context::map) items processed
+        /// concurrently.
+        concurrency_limit: usize,
     },
 }
 
 impl std::fmt::Debug for Context {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let mode = match &self.mode {
-            Mode::Single => "single",
-            Mode::Multi {
-                executor: Some(_), ..
-            } => "multi-threaded",
-            Mode::Multi { executor: None, .. } => "multi-cooperative",
-        };
         f.debug_struct("Context")
             .field("id", &self.id)
             .field("io", &self.io)
-            .field("mode", &mode)
-            .finish()
+            .finish_non_exhaustive()
     }
 }
 
 impl Context {
-    /// Creates a new context that uses `mux` to allocate a channel per
-    /// sub-task.
-    ///
-    /// Sub-tasks are executed cooperatively on the calling future. For
-    /// parallel execution, build an [`Executor`](crate::Executor) and use
-    /// [`Executor::new_context`](crate::Executor::new_context) instead.
-    pub fn new<M: Mux + Send + Sync + 'static>(mux: M) -> Result<Self, ContextError> {
-        Self::with_prefix(mux, ContextId::default())
-    }
-
     /// Creates a new context backed by a single I/O channel.
     ///
     /// Sub-tasks spawned via [`join`], [`try_join`], [`map`] etc. share the
-    /// channel and run **sequentially** in the order given.
+    /// channel and run **sequentially** in the order given. For parallel
+    /// execution, build a [`Session`](crate::Session) and use
+    /// [`Session::new_context`](crate::Session::new_context) instead.
     ///
     /// [`join`]: Self::join
     /// [`try_join`]: Self::try_join
@@ -98,45 +93,32 @@ impl Context {
         }
     }
 
-    /// Like [`Context::new`], but namespaces all channels under `prefix` so
-    /// several sub-protocols can share a mux without colliding.
-    pub fn with_prefix<M: Mux + Send + Sync + 'static>(
-        mux: M,
-        prefix: impl AsRef<[u8]>,
-    ) -> Result<Self, ContextError> {
-        let mux: Arc<dyn Mux + Send + Sync> = Arc::new(mux);
-        let id = ContextId::from_prefix(prefix);
-        let io = mux.open(id.as_ref()).map_err(ContextError::mux)?;
-        Ok(Self {
-            id,
-            io,
-            mode: Mode::Multi {
-                mux,
-                executor: None,
-            },
-            fork_counter: 0,
-        })
-    }
-
-    pub(crate) fn with_executor(
+    pub(crate) fn for_session(
         id: ContextId,
         io: Io,
         mux: Arc<dyn Mux + Send + Sync>,
-        executor: Arc<Inner>,
+        pool: Option<ThreadPool>,
+        concurrency_limit: usize,
     ) -> Self {
         Self {
             id,
             io,
             mode: Mode::Multi {
                 mux,
-                executor: Some(executor),
+                pool,
+                concurrency_limit,
             },
             fork_counter: 0,
         }
     }
 
     fn child(&self, id: ContextId) -> Result<Self, ContextError> {
-        let Mode::Multi { mux, executor } = &self.mode else {
+        let Mode::Multi {
+            mux,
+            pool,
+            concurrency_limit,
+        } = &self.mode
+        else {
             unreachable!("child() called on a single-channel context");
         };
         let io = mux.open(id.as_ref()).map_err(ContextError::mux)?;
@@ -145,7 +127,8 @@ impl Context {
             io,
             mode: Mode::Multi {
                 mux: mux.clone(),
-                executor: executor.clone(),
+                pool: pool.clone(),
+                concurrency_limit: *concurrency_limit,
             },
             fork_counter: 0,
         })
@@ -180,27 +163,50 @@ impl Context {
         T: Send + 'static,
         R: Send + 'static,
     {
-        if matches!(self.mode, Mode::Single) {
-            let mut results = Vec::with_capacity(items.len());
-            for item in items {
-                results.push(f(self, item).await);
+        let (mux, pool, concurrency_limit) = match &self.mode {
+            Mode::Single => {
+                let mut results = Vec::with_capacity(items.len());
+                for item in items {
+                    results.push(f(self, item).await);
+                }
+                return Ok(results);
             }
-            return Ok(results);
-        }
+            Mode::Multi {
+                mux,
+                pool,
+                concurrency_limit,
+            } => (mux.clone(), pool.clone(), *concurrency_limit),
+        };
 
         let parent_id = self.next_fork();
-        let executor = self.executor().cloned();
-        let mut tasks = Vec::with_capacity(items.len());
-        for (i, item) in items.into_iter().enumerate() {
-            let i = u32::try_from(i).expect("more than u32::MAX items");
-            let mut ctx = self.child(parent_id.child(i))?;
-            let f = f.clone();
-            tasks.push(run(
-                executor.as_ref(),
-                async move { f(&mut ctx, item).await },
-            ));
-        }
-        Ok(future::join_all(tasks).await)
+
+        // Each item lazily opens its own channel only once `buffered` polls it,
+        // so at most `limit` channels are open at any time. Channel IDs stay
+        // keyed by item index and results are yielded in input order, so the
+        // bound changes neither the wire protocol nor the output ordering.
+        stream::iter(items.into_iter().enumerate())
+            .map(move |(i, item)| {
+                let i = u32::try_from(i).expect("more than u32::MAX items");
+                let id = parent_id.child(i);
+                let (mux, pool, f) = (mux.clone(), pool.clone(), f.clone());
+                async move {
+                    let io = mux.open(id.as_ref()).map_err(ContextError::mux)?;
+                    let mut ctx = Context {
+                        id,
+                        io,
+                        mode: Mode::Multi {
+                            mux,
+                            pool: pool.clone(),
+                            concurrency_limit,
+                        },
+                        fork_counter: 0,
+                    };
+                    Ok(run(pool.as_ref(), async move { f(&mut ctx, item).await }).await)
+                }
+            })
+            .buffered(concurrency_limit)
+            .try_collect()
+            .await
     }
 
     /// Runs `a` and `b` concurrently and returns both results.
@@ -218,12 +224,12 @@ impl Context {
         }
 
         let parent_id = self.next_fork();
-        let executor = self.executor().cloned();
+        let pool = self.pool().cloned();
         let mut ctx_a = self.child(parent_id.child(0))?;
         let mut ctx_b = self.child(parent_id.child(1))?;
 
-        let task_a = run(executor.as_ref(), async move { a(&mut ctx_a).await });
-        let task_b = run(executor.as_ref(), async move { b(&mut ctx_b).await });
+        let task_a = run(pool.as_ref(), async move { a(&mut ctx_a).await });
+        let task_b = run(pool.as_ref(), async move { b(&mut ctx_b).await });
         Ok(future::join(task_a, task_b).await)
     }
 
@@ -251,12 +257,12 @@ impl Context {
         }
 
         let parent_id = self.next_fork();
-        let executor = self.executor().cloned();
+        let pool = self.pool().cloned();
         let mut ctx_a = self.child(parent_id.child(0))?;
         let mut ctx_b = self.child(parent_id.child(1))?;
 
-        let task_a = run(executor.as_ref(), async move { a(&mut ctx_a).await });
-        let task_b = run(executor.as_ref(), async move { b(&mut ctx_b).await });
+        let task_a = run(pool.as_ref(), async move { a(&mut ctx_a).await });
+        let task_b = run(pool.as_ref(), async move { b(&mut ctx_b).await });
         Ok(future::try_join(task_a, task_b).await)
     }
 
@@ -287,14 +293,14 @@ impl Context {
         }
 
         let parent_id = self.next_fork();
-        let executor = self.executor().cloned();
+        let pool = self.pool().cloned();
         let mut ctx_a = self.child(parent_id.child(0))?;
         let mut ctx_b = self.child(parent_id.child(1))?;
         let mut ctx_c = self.child(parent_id.child(2))?;
 
-        let task_a = run(executor.as_ref(), async move { a(&mut ctx_a).await });
-        let task_b = run(executor.as_ref(), async move { b(&mut ctx_b).await });
-        let task_c = run(executor.as_ref(), async move { c(&mut ctx_c).await });
+        let task_a = run(pool.as_ref(), async move { a(&mut ctx_a).await });
+        let task_b = run(pool.as_ref(), async move { b(&mut ctx_b).await });
+        let task_c = run(pool.as_ref(), async move { c(&mut ctx_c).await });
         Ok(future::try_join3(task_a, task_b, task_c).await)
     }
 
@@ -329,40 +335,37 @@ impl Context {
         }
 
         let parent_id = self.next_fork();
-        let executor = self.executor().cloned();
+        let pool = self.pool().cloned();
         let mut ctx_a = self.child(parent_id.child(0))?;
         let mut ctx_b = self.child(parent_id.child(1))?;
         let mut ctx_c = self.child(parent_id.child(2))?;
         let mut ctx_d = self.child(parent_id.child(3))?;
 
-        let task_a = run(executor.as_ref(), async move { a(&mut ctx_a).await });
-        let task_b = run(executor.as_ref(), async move { b(&mut ctx_b).await });
-        let task_c = run(executor.as_ref(), async move { c(&mut ctx_c).await });
-        let task_d = run(executor.as_ref(), async move { d(&mut ctx_d).await });
+        let task_a = run(pool.as_ref(), async move { a(&mut ctx_a).await });
+        let task_b = run(pool.as_ref(), async move { b(&mut ctx_b).await });
+        let task_c = run(pool.as_ref(), async move { c(&mut ctx_c).await });
+        let task_d = run(pool.as_ref(), async move { d(&mut ctx_d).await });
         Ok(future::try_join4(task_a, task_b, task_c, task_d).await)
     }
 
-    fn executor(&self) -> Option<&Arc<Inner>> {
-        if let Mode::Multi { executor, .. } = &self.mode {
-            executor.as_ref()
+    fn pool(&self) -> Option<&ThreadPool> {
+        if let Mode::Multi { pool, .. } = &self.mode {
+            pool.as_ref()
         } else {
             None
         }
     }
 }
 
-/// Spawns `fut` on `executor` if one is provided, otherwise yields the future
+/// Spawns `fut` on `pool` if one is provided, otherwise yields the future
 /// as-is. The output type is identical either way.
-fn run<F>(
-    executor: Option<&Arc<Inner>>,
-    fut: F,
-) -> impl std::future::Future<Output = F::Output> + Send
+fn run<F>(pool: Option<&ThreadPool>, fut: F) -> impl std::future::Future<Output = F::Output> + Send
 where
     F: std::future::Future + Send + 'static,
     F::Output: Send + 'static,
 {
-    match executor {
-        Some(exec) => Either::Left(crate::executor::spawn_on(exec, fut)),
+    match pool {
+        Some(pool) => Either::Left(crate::thread_pool::spawn_on(pool, fut)),
         None => Either::Right(fut),
     }
 }

crates/common/src/context/test/helpers.rs

@@ -4,9 +4,10 @@ use serio::channel::duplex;
 
 use crate::{
     context::Context,
-    executor::{Executor, ExecutorBuilder},
     io::Io,
     mux::test_framed_mux,
+    session::{Session, SessionBuilder},
+    thread_pool::ThreadPool,
 };
 
 /// Creates a pair of single-threaded contexts using memory I/O channels.
@@ -19,19 +20,19 @@ pub fn test_st_context(io_buffer: usize) -> (Context, Context) {
     )
 }
 
-/// Creates a pair of multi-threaded executors sharing multiplexed I/O channels.
-pub fn test_mt_context(io_buffer: usize) -> (Executor, Executor) {
+/// Creates a pair of multi-threaded sessions sharing multiplexed I/O channels.
+pub fn test_mt_context(io_buffer: usize) -> (Session, Session) {
     let (mux_0, mux_1) = test_framed_mux(io_buffer);
 
     (
-        ExecutorBuilder::default().build(mux_0),
-        ExecutorBuilder::default().build(mux_1),
+        SessionBuilder::default().build(mux_0).unwrap(),
+        SessionBuilder::default().build(mux_1).unwrap(),
     )
 }
 
 /// Like [`test_mt_context`], but uses a custom worker spawn callback (e.g.
 /// `web_spawn::spawn` on wasm).
-pub fn test_mt_context_with_spawn<F>(io_buffer: usize, spawn: F) -> (Executor, Executor)
+pub fn test_mt_context_with_spawn<F>(io_buffer: usize, spawn: F) -> (Session, Session)
 where
     F: Fn(Box<dyn FnOnce() + Send + 'static>) -> Result<(), std::io::Error>
         + Clone
@@ -40,9 +41,11 @@ where
         + 'static,
 {
     let (mux_0, mux_1) = test_framed_mux(io_buffer);
+    let pool_0 = ThreadPool::builder().spawn(spawn.clone()).build().unwrap();
+    let pool_1 = ThreadPool::builder().spawn(spawn).build().unwrap();
 
     (
-        ExecutorBuilder::default().spawn(spawn.clone()).build(mux_0),
-        ExecutorBuilder::default().spawn(spawn).build(mux_1),
+        SessionBuilder::default().pool(pool_0).build(mux_0).unwrap(),
+        SessionBuilder::default().pool(pool_1).build(mux_1).unwrap(),
     )
 }