Towards thread pool

.scripts/run_benchmark.sh

@@ -1,10 +1,10 @@
-original_bench=find
+original_bench=find_iter_into_par
 bench=$1
 
 sed -i "s/$original_bench/$bench/g" Cargo.toml
 
 rm -f benches/results/$bench.txt
 
-cargo bench >> benches/results/$bench.txt
+cargo bench --all-features >> benches/results/$bench.txt
 
 sed -i "s/$bench/$original_bench/g" Cargo.toml

.scripts/run_benchmarks.sh

@@ -1,36 +1,38 @@
-# allBenches=(collect_filter sum)
-
 allBenches=(
+    chain_collect_map
+    chain3_collect_map
+    chain4_collect_map
     collect_filter
     collect_filtermap
     collect_flatmap
     collect_iter_into_par
     collect_long_chain
-    collect_map
-    collect_map_filter
     collect_map_filter_hash_set
-    collect_result
+    collect_map_filter
+    collect_map
     count_filtermap
     count_flatmap
-    count_map
     count_map_filter
+    count_map
     drain_vec_collect_map_filter
-    find
     find_any
     find_flatmap
     find_iter_into_par
     find_map_filter
+    find
     mut_for_each_iter
     mut_for_each_slice
-    reduce
     reduce_iter_into_par
     reduce_long_chain
-    reduce_map
     reduce_map_filter
-    sum
+    reduce_map
+    reduce
+    result_collect_map
+    result_reduce_map
     sum_filtermap
     sum_flatmap
     sum_map_filter
+    sum
     vec_deque_collect_map_filter
     vec_deque_collect_map_filter_owned
 )

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "orx-parallel"
-version = "3.2.0"
+version = "3.3.0"
 edition = "2024"
 authors = ["orxfun <orx.ugur.arikan@gmail.com>"]
 readme = "README.md"
@@ -11,33 +11,47 @@ keywords = ["parallel", "concurrency", "performance", "thread", "iterator"]
 categories = ["concurrency", "algorithms"]
 
 [dependencies]
-orx-pseudo-default = { version = "2.1.0", default-features = false }
 orx-pinned-vec = { version = "3.17.0", default-features = false }
-orx-fixed-vec = { version = "3.18.0", default-features = false }
-orx-split-vec = { version = "3.18.0", default-features = false }
-orx-pinned-concurrent-col = { version = "2.14.0", default-features = false }
-orx-concurrent-bag = { version = "3.0.0", default-features = false }
-orx-concurrent-ordered-bag = { version = "3.0.0", default-features = false }
+orx-fixed-vec = { version = "3.19.0", default-features = false }
+orx-split-vec = { version = "3.19.0", default-features = false }
+orx-concurrent-iter = { version = "3.1.0", default-features = false }
+orx-concurrent-bag = { version = "3.1.0", default-features = false }
+orx-concurrent-ordered-bag = { version = "3.1.0", default-features = false }
 orx-iterable = { version = "1.3.0", default-features = false }
+orx-pinned-concurrent-col = { version = "2.15.0", default-features = false }
 orx-priority-queue = { version = "1.7.0", default-features = false }
-orx-concurrent-iter = { version = "3.1.0", default-features = false }
-rayon = { version = "1.10.0", optional = true }
+orx-pseudo-default = { version = "2.1.0", default-features = false }
+
+# optional: generic iterator
+rayon = { version = "1.11.0", optional = true, default-features = false }
+
+# optional: thread pool
+pond = { version = "0.3.1", optional = true, default-features = false }
+poolite = { version = "0.7.1", optional = true, default-features = false }
+rayon-core = { version = "1.13.0", optional = true, default-features = false }
+scoped-pool = { version = "1.0.0", optional = true, default-features = false }
+scoped_threadpool = { version = "0.1.9", optional = true, default-features = false }
+yastl = { version = "0.1.2", optional = true, default-features = false }
 
 [dev-dependencies]
-chrono = "0.4.39"
-clap = { version = "4.5.36", features = ["derive"] }
-criterion = "0.5.1"
+chrono = "0.4.42"
+clap = { version = "4.5.47", features = ["derive"] }
+criterion = "0.7.0"
 orx-concurrent-option = { version = "1.5.0", default-features = false }
-orx-concurrent-vec = "3.6.0"
-rand = "0.9"
+orx-concurrent-vec = "3.8.0"
+rand = "0.9.2"
 rand_chacha = "0.9"
-rayon = "1.10.0"
+rayon = "1.11.0"
 test-case = "3.3.1"
 
 [[bench]]
-name = "find"
+name = "find_iter_into_par"
 harness = false
 
+[package.metadata.docs.rs]
+all-features = true
+
 [features]
-default = []
+default = ["std"]
+std = []
 generic_iterator = ["rayon"]

README.md

@@ -12,7 +12,7 @@
 * [Fallible Parallel Iterators](#fallible-parallel-iterators)
 * [Using Mutable Variables](#using-mutable-variables)
 * [Configurations](#configurations)
-* [Underlying Approach and Parallel Runners](#underlying-approach-and-parallel-runners)
+* [Runner: Pools and Executors](#runner-pools-and-executors)
 * [Contributing](#contributing)
 
 ## Parallel Computation by Iterators
@@ -420,16 +420,101 @@ This is guaranteed by the fact that both consuming computation calls and configu
 Additionally, maximum number of threads that can be used by parallel computations can be globally bounded by the environment variable `ORX_PARALLEL_MAX_NUM_THREADS`. Please see the corresponding [example](https://github.com/orxfun/orx-parallel/blob/main/examples/max_num_threads_config.rs) for details.
 
 
-## Underlying Approach and Parallel Runners
+## Runner: Pools and Executors
 
 This crate defines parallel computation by combining two basic components.
 
-* Pulling **inputs** in parallel is achieved through [`ConcurrentIter`](https://crates.io/crates/orx-concurrent-iter). Concurrent iterator implementations are lock-free, efficient and support pull-by-chunks optimization to reduce the parallelization overhead. A thread can pull any number of inputs from the concurrent iterator every time it becomes idle. This provides the means to dynamically decide on the chunk sizes.
-* Writing **outputs** in parallel is handled using thread-safe containers such as [`ConcurrentBag`](https://crates.io/crates/orx-concurrent-bag) and [`ConcurrentOrderedBag`](https://crates.io/crates/orx-concurrent-ordered-bag). Similarly, these are lock-free collections that aim for high performance collection of results.
+**Pulling inputs**
+* Pulling inputs in parallel is achieved through [`ConcurrentIter`](https://crates.io/crates/orx-concurrent-iter). Concurrent iterator implementations are lock-free, efficient and support pull-by-chunks optimization to reduce the parallelization overhead. A thread can pull any number of inputs from the concurrent iterator every time it becomes idle. This provides the means to dynamically decide on the chunk sizes.
+* Furthermore, this allows to reduce the overhead of defining creating tasks. To illustrate, provided that the computation will be handled by `n` threads, a closure holding a reference to the input concurrent iterator is defined to represent the computation. This same closure is passed to `n` threads; i.e., `n` spawn calls are made. Each of these threads keep pulling elements from the input until the computation is completed, without requiring to define another task.
 
-Finally, [`ParallelRunner`](https://docs.rs/orx-parallel/latest/orx_parallel/runner/trait.ParallelRunner.html) trait manages parallelization of the given computation with desired configuration. The objective of the parallel runner is to optimize the chunk sizes to solve the tradeoff between impact of heterogeneity of individual computations and overhead of parallelization.
+**Writing outputs**
+* When we collect results, writing outputs is handled using lock-free containers such as [`ConcurrentBag`](https://crates.io/crates/orx-concurrent-bag) and [`ConcurrentOrderedBag`](https://crates.io/crates/orx-concurrent-ordered-bag) which aim for high performance collection of results.
 
-Since it is a trait, parallel runner is customizable. It is possible to implement and use your *own runner* by calling [`with_runner`](https://docs.rs/orx-parallel/latest/orx_parallel/trait.ParIter.html#tymethod.with_runner) transformation method on the parallel iterator. Default parallel runner targets to be efficient in general. When we have a use case with special characteristics, we can implement a `ParallelRunner` optimized for this scenario and use with the parallel iterators.
+There are two main decisions to be taken while executing these components:
+* how many threads do we use?
+* what is the chunk size; i.e., how many input items does a thread pull each time?
+
+A [`ParallelRunner`](https://docs.rs/orx-parallel/latest/orx_parallel/trait.ParallelRunner) is a combination of a `ParThreadPool` and a `ParallelExecutor` that are responsible for these decisions, respectively.
+
+### ParThreadPool: number of threads
+
+[`ParThreadPool`](https://docs.rs/orx-parallel/latest/orx_parallel/trait.ParThreadPool) trait generalizes thread pools that can be used for parallel computations. This allows the parallel computation to be generic over thread pools.
+
+When not explicitly set, [`DefaultPool`](https://docs.rs/orx-parallel/latest/orx_parallel/type.DefaultPool) is used:
+* When **std** feature is enabled, default pool is the [`StdDefaultPool`](https://docs.rs/orx-parallel/latest/orx_parallel/struct.StdDefaultPool). In other words, all available native threads can be used by the parallel computation. This number can globally bounded by "ORX_PARALLEL_MAX_NUM_THREADS" environment variable when set.
+* When working in a **no-std** environment, default pool is the [`SequentialPool`](https://docs.rs/orx-parallel/latest/orx_parallel/struct.SequentialPool). As the name suggests, this pool executes the parallel computation sequentially on the main thread. It can be considered as a placeholder to be overwritten by `with_pool` or `with_runner` methods to achieve parallelism.
+
+*Note that thread pool defines the resource, or upper bound. This upper bound can further be bounded by the [`num_threads`](https://docs.rs/orx-parallel/latest/orx_parallel/trait.ParIter.html#tymethod.num_threads) configuration. Finally, parallel executor might choose not to use all available threads if it decides that the computation is small enough.*
+
+To overwrite the defaults and explicitly set the thread pool to be used for the computation, [`with_pool`](https://docs.rs/orx-parallel/latest/orx_parallel/trait.ParIter.html#tymethod.with_pool) or [`with_runner`](https://docs.rs/orx-parallel/latest/orx_parallel/trait.ParIter.html#tymethod.with_runner) methods are used.
+
+```rust
+use orx_parallel::*;
+
+let inputs: Vec<_> = (0..42).collect();
+
+// uses the DefaultPool
+// assuming "std" enabled, StdDefaultPool will be used; i.e., native threads
+let sum = inputs.par().sum();
+
+// equivalent to:
+let sum2 = inputs.par().with_pool(StdDefaultPool::default()).sum();
+assert_eq!(sum, sum2);
+
+#[cfg(feature = "scoped_threadpool")]
+{
+    let mut pool = scoped_threadpool::Pool::new(8);
+    // uses the scoped_threadpool::Pool created with 8 threads
+    let sum2 = inputs.par().with_pool(&mut pool).sum();
+    assert_eq!(sum, sum2);
+}
+
+#[cfg(feature = "rayon-core")]
+{
+    let pool = rayon_core::ThreadPoolBuilder::new()
+        .num_threads(8)
+        .build()
+        .unwrap();
+    // uses the rayon-core::ThreadPool created with 8 threads
+    let sum2 = inputs.par().with_pool(&pool).sum();
+    assert_eq!(sum, sum2);
+}
+
+#[cfg(feature = "yastl")]
+{
+    let pool = YastlPool::new(8);
+    // uses the yastl::Pool created with 8 threads
+    let sum2 = inputs.par().with_pool(&pool).sum();
+    assert_eq!(sum, sum2);
+}
+```
+
+`ParThreadPool` implementations of several thread pools are provided in this crate as optional features (see [features](#features) section). Provided that the pool supports scoped computations, it is trivial to implement this trait in most cases (see [implementations](https://github.com/orxfun/orx-parallel/tree/main/src/runner/implementations) for examples).
+
+In most of the cases, *rayon-core*, *scoped_threadpool* and *scoped_pool* perform better than others, and get close to native threads performance with `StdDefaultPool`.
+
+Since parallel computations are generic over the thread pools, performances can be conveniently compared for specific use cases. Such an example benchmark can be found in [collect_filter_map](https://github.com/orxfun/orx-parallel/blob/main/benches/collect_filter_map.rs) file. To have quick tests, you may also use the example [benchmark_pools](https://github.com/orxfun/orx-parallel/blob/main/examples/benchmark_pools.rs).
+
+### ParallelExecutor: chunk size
+
+Once thread pool provides the computation resources, it is [`ParallelExecutor`](https://docs.rs/orx-parallel/latest/orx_parallel/trait.ParallelExecutor)'s task to distribute work to available threads. As mentioned above, all threads receive exactly the same closure. This closure continues to pull elements from the input concurrent iterator and operate on the inputs until all elements are processed.
+
+The critical decision that parallel executor makes is the chunk size. Depending on the state of the computation, it can dynamically decide on number of elements to pull from the input iterator. The tradeoff it tries to solve is as follows:
+
+* the larger the chunk size,
+  * the smaller the parallelization overhead; but also
+  * the larger the risk of imbalance in cases of heterogeneity.
+
+## Features
+
+* **std**: This is a **no-std** crate while *std* is included as a default feature. Please use `--no-default-features` flag for no-std use cases. **std** feature enables `StdDefaultPool` as the default thread provider which uses native threads.
+* **rayon-core**: This feature enables using `rayon_core::ThreadPool` for parallel computations.
+* **scoped_threadpool**: This feature enables using `scoped_threadpool::Pool`.
+* **scoped-pool**: This feature enables using `scoped-pool::Pool`.
+* **yastl**: This feature enables using `yastl::Pool`.
+* **pond**: This feature enables using `pond::Pool`.
+* **poolite**: This feature enables using `poolite::Pool`.
 
 ## Contributing
 
@@ -439,7 +524,8 @@ Please open an [issue](https://github.com/orxfun/orx-parallel/issues/new) or cre
 
 * if you notice an error,
 * have a question or think something could be improved,
-* have an input collection or generator that needs to be parallelized, or
+* have an input collection or generator that needs to be parallelized,
+* want to use a particular thread pool with parallel iterators,
 * having trouble representing a particular parallel computation with parallel iterators,
 * or anything else:)
 

benches/chain3_collect_map.rs

@@ -1,9 +1,10 @@
-use criterion::{BenchmarkId, Criterion, black_box, criterion_group, criterion_main};
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
 use orx_parallel::*;
 use orx_split_vec::SplitVec;
 use rand::prelude::*;
 use rand_chacha::ChaCha8Rng;
 use rayon::iter::IntoParallelIterator;
+use std::hint::black_box;
 
 const TEST_LARGE_OUTPUT: bool = false;
 

benches/chain4_collect_map.rs

@@ -1,9 +1,10 @@
-use criterion::{BenchmarkId, Criterion, black_box, criterion_group, criterion_main};
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
 use orx_parallel::*;
 use orx_split_vec::SplitVec;
 use rand::prelude::*;
 use rand_chacha::ChaCha8Rng;
 use rayon::iter::IntoParallelIterator;
+use std::hint::black_box;
 
 const TEST_LARGE_OUTPUT: bool = false;
 

benches/chain_collect_map.rs

@@ -1,9 +1,10 @@
-use criterion::{BenchmarkId, Criterion, black_box, criterion_group, criterion_main};
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
 use orx_parallel::*;
 use orx_split_vec::SplitVec;
 use rand::prelude::*;
 use rand_chacha::ChaCha8Rng;
 use rayon::iter::IntoParallelIterator;
+use std::hint::black_box;
 
 const TEST_LARGE_OUTPUT: bool = false;
 

benches/collect_filter.rs

@@ -1,9 +1,10 @@
-use criterion::{BenchmarkId, Criterion, black_box, criterion_group, criterion_main};
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
 use orx_parallel::*;
 use orx_split_vec::SplitVec;
 use rand::prelude::*;
 use rand_chacha::ChaCha8Rng;
 use rayon::iter::IntoParallelIterator;
+use std::hint::black_box;
 
 const TEST_LARGE_OUTPUT: bool = false;
 
@@ -83,6 +84,11 @@ fn orx_into_split_vec(inputs: &[Output]) -> SplitVec<&Output> {
     inputs.into_par().filter(filter).collect()
 }
 
+#[allow(dead_code)]
+fn orx_into_vec_with<P: ParThreadPool>(inputs: &[Output], pool: P) -> Vec<&Output> {
+    inputs.into_par().with_pool(pool).filter(filter).collect()
+}
+
 fn run(c: &mut Criterion) {
     let treatments = [65_536 * 2];
 
@@ -113,6 +119,56 @@ fn run(c: &mut Criterion) {
             assert_eq!(&expected, &orx_into_split_vec(&input));
             b.iter(|| orx_into_split_vec(black_box(&input)))
         });
+
+        #[cfg(feature = "rayon-core")]
+        group.bench_with_input(
+            BenchmarkId::new("orx-vec (rayon-core::ThreadPool)", n),
+            n,
+            |b, _| {
+                let pool = rayon_core::ThreadPoolBuilder::new()
+                    .num_threads(32)
+                    .build()
+                    .unwrap();
+                assert_eq!(&expected, &orx_into_vec_with(&input, &pool));
+                b.iter(|| orx_into_vec_with(black_box(&input), &pool))
+            },
+        );
+
+        #[cfg(feature = "scoped-pool")]
+        group.bench_with_input(
+            BenchmarkId::new("orx-vec (scoped-pool::Pool)", n),
+            n,
+            |b, _| {
+                let pool = scoped_pool::Pool::new(32);
+                assert_eq!(&expected, &orx_into_vec_with(&input, &pool));
+                b.iter(|| orx_into_vec_with(black_box(&input), &pool))
+            },
+        );
+
+        #[cfg(feature = "scoped_threadpool")]
+        group.bench_with_input(
+            BenchmarkId::new("orx-vec (scoped_threadpool::Pool)", n),
+            n,
+            |b, _| {
+                let pool = || scoped_threadpool::Pool::new(32);
+                assert_eq!(&expected, &orx_into_vec_with(&input, pool()));
+                b.iter(|| orx_into_vec_with(black_box(&input), pool()))
+            },
+        );
+
+        #[cfg(feature = "yastl")]
+        group.bench_with_input(BenchmarkId::new("orx-vec (yastl::Pool)", n), n, |b, _| {
+            let pool = YastlPool::new(32);
+            assert_eq!(&expected, &orx_into_vec_with(&input, &pool));
+            b.iter(|| orx_into_vec_with(black_box(&input), &pool))
+        });
+
+        #[cfg(feature = "pond")]
+        group.bench_with_input(BenchmarkId::new("orx-vec (pond::Pool)", n), n, |b, _| {
+            let pool = || PondPool::new_threads_unbounded(32);
+            assert_eq!(&expected, &orx_into_vec_with(&input, pool()));
+            b.iter(|| orx_into_vec_with(black_box(&input), pool()))
+        });
     }
 
     group.finish();

benches/collect_filtermap.rs

@@ -1,9 +1,10 @@
-use criterion::{BenchmarkId, Criterion, black_box, criterion_group, criterion_main};
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
 use orx_parallel::*;
 use orx_split_vec::SplitVec;
 use rand::prelude::*;
 use rand_chacha::ChaCha8Rng;
 use rayon::iter::IntoParallelIterator;
+use std::hint::black_box;
 
 const TEST_LARGE_OUTPUT: bool = false;