Merge pull request #82 from orxfun/concurrent-iterator

Concurrent and Parallel Iterator over References

Author: orxfun

.github/workflows/ci.yml

@@ -52,7 +52,7 @@ jobs:
       run: cargo clippy -- -D warnings --verbose
 
     - name: Miri
-      run: cargo +nightly miri test --verbose
+      run: cargo +nightly miri test --lib --bins --tests --verbose
 
     - name: NoStd
       run: cargo +nightly no-std-check

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "orx-split-vec"
-version = "3.16.0"
+version = "3.17.0"
 edition = "2024"
 authors = ["orxfun <orx.ugur.arikan@gmail.com>"]
 description = "An efficient dynamic capacity vector with pinned element guarantees."
@@ -13,13 +13,18 @@ categories = ["data-structures", "rust-patterns", "no-std"]
 orx-iterable = { version = "1.3.0", default-features = false }
 orx-pseudo-default = { version = "2.1.0", default-features = false }
 orx-pinned-vec = { version = "3.16.0", default-features = false }
+orx-concurrent-iter = { version = "2.1.0", default-features = false }
 
 [[bench]]
-name = "serial_access"
+name = "par_collect_map_filter_ref"
 harness = false
 
 [dev-dependencies]
-criterion = { version = "0.5", default-features = false }
+clap = { version = "4.5.38", features = ["derive"] }
+criterion = "0.6.0"
 rand = { version = "0.9", default-features = false }
 rand_chacha = { version = "0.9", default-features = false }
 test-case = "3.3.1"
+orx-concurrent-bag = "2.12.0"
+orx-parallel = { version = "2.0.1", default-features = false }
+rayon = { version = "1.10.0", default-features = false }

README.md

@@ -63,6 +63,20 @@ Recursive strategy is a specialized variant of the Doubling, which works identic
 
 It is equivalent to Doubling strategy in terms of sequential access performance. However, due to the additional flexibility, it cannot implement `GrowthWithConstantTimeAccess`. Its random access time complexity is **O(f)** where **f** is the number of fragments in the split vector. 
 
+## Parallelization
+
+`SplitVec` implements [`ConcurrentCollection`](https://docs.rs/orx-concurrent-iter/latest/orx_concurrent_iter/trait.ConcurrentCollection.html) for all above-mentioned growth strategies.
+
+Therefore, when [orx_parallel](https://crates.io/crates/orx-parallel) crate is included, `SplitVec` also automatically implements [`ParallelizableCollection`](https://docs.rs/orx-parallel/latest/orx_parallel/trait.ParallelizableCollection.html).
+
+This means that computations over the split vector can be efficiently parallelized:
+
+* `split_vec.par()` returns a parallel iterator over references to its elements, and
+* `split_vec.into_par()` consumes the vector and returns a parallel iterator of the owned elements.
+
+You may find demonstrations in [`demo_parallelization`](https://github.com/orxfun/orx-split-vec/blob/main/examples/demo_parallelization.rs) and [`bench_parallelization`](https://github.com/orxfun/orx-split-vec/blob/main/examples/bench_parallelization.rs) examples.
+
+
 ## Examples
 
 SplitVec api resembles and aims to cover as much as possible the standard vector's api.

benches/append.rs

@@ -1,7 +1,8 @@
-use criterion::{BatchSize, BenchmarkId, Criterion, black_box, criterion_group, criterion_main};
+use criterion::{BatchSize, BenchmarkId, Criterion, criterion_group, criterion_main};
 use orx_split_vec::*;
 use rand::prelude::*;
 use rand_chacha::ChaCha8Rng;
+use std::hint::black_box;
 
 const NUM_APPEND_OPS: usize = 32;
 

benches/grow.rs

@@ -1,8 +1,8 @@
 use criterion::{
-    black_box, criterion_group, criterion_main, measurement::WallTime, BenchmarkGroup, BenchmarkId,
-    Criterion,
+    BenchmarkGroup, BenchmarkId, Criterion, criterion_group, criterion_main, measurement::WallTime,
 };
 use orx_split_vec::*;
+use std::hint::black_box;
 
 fn get_value<const N: usize>(i: usize) -> [u64; N] {
     let modulo = i % 3;

benches/par_collect_map_filter_owned.rs

@@ -0,0 +1,138 @@
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use orx_parallel::*;
+use orx_split_vec::*;
+use rand::prelude::*;
+use rand_chacha::ChaCha8Rng;
+use std::hint::black_box;
+
+const TEST_LARGE_OUTPUT: bool = false;
+
+const LARGE_OUTPUT_LEN: usize = match TEST_LARGE_OUTPUT {
+    true => 64,
+    false => 0,
+};
+const SEED: u64 = 5426;
+const FIB_UPPER_BOUND: u32 = 201;
+
+#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
+struct Output {
+    name: String,
+    numbers: [i64; LARGE_OUTPUT_LEN],
+}
+
+fn map(idx: usize) -> Output {
+    let prefix = match idx % 7 {
+        0 => "zero-",
+        3 => "three-",
+        _ => "sth-",
+    };
+    let fib = fibonacci(&(idx as u32));
+    let name = format!("{}-fib-{}", prefix, fib);
+
+    let mut numbers = [0i64; LARGE_OUTPUT_LEN];
+    for (i, x) in numbers.iter_mut().enumerate() {
+        *x = match (idx * 7 + i) % 3 {
+            0 => idx as i64 + i as i64,
+            _ => idx as i64 - i as i64,
+        };
+    }
+
+    Output { name, numbers }
+}
+
+fn filter(output: &Output) -> bool {
+    let last_char = output.name.chars().last().unwrap();
+    let last_digit: u32 = last_char.to_string().parse().unwrap();
+    last_digit < 4
+}
+
+fn fibonacci(n: &u32) -> u32 {
+    let mut a = 0;
+    let mut b = 1;
+    for _ in 0..*n {
+        let c = a + b;
+        a = b;
+        b = c;
+    }
+    a
+}
+
+fn input(len: usize) -> impl Iterator<Item = usize> {
+    let mut rng = ChaCha8Rng::seed_from_u64(SEED);
+    (0..len).map(move |_| rng.random_range(0..FIB_UPPER_BOUND) as usize)
+}
+
+fn seq(inputs: Vec<usize>) -> Vec<Output> {
+    inputs.into_iter().map(map).filter(filter).collect()
+}
+
+fn par_over_vec(inputs: Vec<usize>) -> Vec<Output> {
+    inputs.into_par().map(map).filter(filter).collect()
+}
+
+fn par_over_split_vec<G: ParGrowth>(inputs: SplitVec<usize, G>) -> Vec<Output> {
+    inputs.into_par().map(map).filter(filter).collect()
+}
+
+fn run(c: &mut Criterion) {
+    let treatments = [65_536, 65_536 * 4];
+
+    #[allow(unused_mut)]
+    let mut group = c.benchmark_group("par_collect_map_filter_owned");
+
+    for n in &treatments {
+        let expected = seq(input(*n).collect());
+
+        let input_doubling = || input(*n).collect::<SplitVec<_, Doubling>>();
+
+        let input_recursive = || input(*n).collect::<SplitVec<_, Recursive>>();
+
+        let input_linear = || {
+            let mut input_linear = SplitVec::with_linear_growth(10);
+            input_linear.extend(input(*n));
+            input_linear
+        };
+
+        group.bench_with_input(BenchmarkId::new("seq", n), n, |b, _| {
+            assert_eq!(&expected, &seq(input(*n).collect()));
+            b.iter(|| seq(black_box(input(*n).collect())))
+        });
+
+        group.bench_with_input(BenchmarkId::new("par_over_vec", n), n, |b, _| {
+            assert_eq!(&expected, &par_over_vec(input(*n).collect()));
+            b.iter(|| par_over_vec(black_box(input(*n).collect())))
+        });
+
+        group.bench_with_input(
+            BenchmarkId::new("par_over_split_vec_doubling", n),
+            n,
+            |b, _| {
+                assert_eq!(&expected, &par_over_split_vec(input_doubling()));
+                b.iter(|| par_over_split_vec(black_box(input_doubling())))
+            },
+        );
+
+        group.bench_with_input(
+            BenchmarkId::new("par_over_split_vec_linear", n),
+            n,
+            |b, _| {
+                assert_eq!(&expected, &par_over_split_vec(input_linear()));
+                b.iter(|| par_over_split_vec(black_box(input_linear())))
+            },
+        );
+
+        group.bench_with_input(
+            BenchmarkId::new("par_over_split_vec_recursive", n),
+            n,
+            |b, _| {
+                assert_eq!(&expected, &par_over_split_vec(input_recursive()));
+                b.iter(|| par_over_split_vec(black_box(input_recursive())))
+            },
+        );
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, run);
+criterion_main!(benches);

benches/par_collect_map_filter_ref.rs

@@ -0,0 +1,140 @@
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use orx_parallel::*;
+use orx_split_vec::*;
+use rand::prelude::*;
+use rand_chacha::ChaCha8Rng;
+use std::hint::black_box;
+
+const TEST_LARGE_OUTPUT: bool = false;
+
+const LARGE_OUTPUT_LEN: usize = match TEST_LARGE_OUTPUT {
+    true => 64,
+    false => 0,
+};
+const SEED: u64 = 5426;
+const FIB_UPPER_BOUND: u32 = 201;
+
+#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
+struct Output {
+    name: String,
+    numbers: [i64; LARGE_OUTPUT_LEN],
+}
+
+fn map(idx: &usize) -> Output {
+    let idx = *idx;
+    let prefix = match idx % 7 {
+        0 => "zero-",
+        3 => "three-",
+        _ => "sth-",
+    };
+    let fib = fibonacci(&(idx as u32));
+    let name = format!("{}-fib-{}", prefix, fib);
+
+    let mut numbers = [0i64; LARGE_OUTPUT_LEN];
+    for (i, x) in numbers.iter_mut().enumerate() {
+        *x = match (idx * 7 + i) % 3 {
+            0 => idx as i64 + i as i64,
+            _ => idx as i64 - i as i64,
+        };
+    }
+
+    Output { name, numbers }
+}
+
+fn filter(output: &Output) -> bool {
+    let last_char = output.name.chars().last().unwrap();
+    let last_digit: u32 = last_char.to_string().parse().unwrap();
+    last_digit < 4
+}
+
+fn fibonacci(n: &u32) -> u32 {
+    let mut a = 0;
+    let mut b = 1;
+    for _ in 0..*n {
+        let c = a + b;
+        a = b;
+        b = c;
+    }
+    a
+}
+
+fn get_input(len: usize) -> impl Iterator<Item = usize> {
+    let mut rng = ChaCha8Rng::seed_from_u64(SEED);
+    (0..len).map(move |_| rng.random_range(0..FIB_UPPER_BOUND) as usize)
+}
+
+fn seq(inputs: &[usize]) -> Vec<Output> {
+    inputs.iter().map(map).filter(filter).collect()
+}
+
+fn par_over_vec(inputs: &[usize]) -> Vec<Output> {
+    inputs.par().map(map).filter(filter).collect()
+}
+
+fn par_over_split_vec<G: ParGrowth>(inputs: &SplitVec<usize, G>) -> Vec<Output> {
+    inputs.par().map(map).filter(filter).collect()
+}
+
+fn run(c: &mut Criterion) {
+    let treatments = [65_536, 65_536 * 4];
+
+    #[allow(unused_mut)]
+    let mut group = c.benchmark_group("par_collect_map_filter_ref");
+
+    for n in &treatments {
+        let input: Vec<_> = get_input(*n).collect();
+        let expected = seq(&input);
+
+        let input_doubling = get_input(*n).collect::<SplitVec<_, Doubling>>();
+
+        let input_recursive = get_input(*n).collect::<SplitVec<_, Recursive>>();
+
+        let input_linear = {
+            let mut input_linear = SplitVec::with_linear_growth(10);
+            input_linear.extend_from_slice(&input);
+            input_linear
+        };
+
+        group.bench_with_input(BenchmarkId::new("seq", n), n, |b, _| {
+            assert_eq!(&expected, &seq(&input));
+            b.iter(|| seq(black_box(&input)))
+        });
+
+        group.bench_with_input(BenchmarkId::new("par_over_vec", n), n, |b, _| {
+            assert_eq!(&expected, &par_over_vec(&input));
+            b.iter(|| par_over_vec(black_box(&input)))
+        });
+
+        group.bench_with_input(
+            BenchmarkId::new("par_over_split_vec_doubling", n),
+            n,
+            |b, _| {
+                assert_eq!(&expected, &par_over_split_vec(&input_doubling));
+                b.iter(|| par_over_split_vec(black_box(&input_doubling)))
+            },
+        );
+
+        group.bench_with_input(
+            BenchmarkId::new("par_over_split_vec_linear", n),
+            n,
+            |b, _| {
+                assert_eq!(&expected, &par_over_split_vec(&input_linear));
+                b.iter(|| par_over_split_vec(black_box(&input_linear)))
+            },
+        );
+
+        group.bench_with_input(
+            BenchmarkId::new("par_over_split_vec_recursive", n),
+            n,
+            |b, _| {
+                assert_eq!(&expected, &par_over_split_vec(&input_recursive));
+                b.iter(|| par_over_split_vec(black_box(&input_recursive)))
+            },
+        );
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, run);
+criterion_main!(benches);

benches/random_access.rs

@@ -1,10 +1,10 @@
 use criterion::{
-    BenchmarkGroup, BenchmarkId, Criterion, black_box, criterion_group, criterion_main,
-    measurement::WallTime,
+    BenchmarkGroup, BenchmarkId, Criterion, criterion_group, criterion_main, measurement::WallTime,
 };
 use orx_split_vec::*;
 use rand::prelude::*;
 use rand_chacha::ChaCha8Rng;
+use std::hint::black_box;
 
 fn get_indices(n: usize) -> Vec<usize> {
     let mut rng = ChaCha8Rng::seed_from_u64(7541);

benches/serial_access.rs

@@ -1,8 +1,8 @@
 use criterion::{
-    black_box, criterion_group, criterion_main, measurement::WallTime, BenchmarkGroup, BenchmarkId,
-    Criterion,
+    BenchmarkGroup, BenchmarkId, Criterion, criterion_group, criterion_main, measurement::WallTime,
 };
 use orx_split_vec::*;
+use std::hint::black_box;
 
 fn get_value<const N: usize>(i: usize) -> [u64; N] {
     let modulo = i % 3;