Skip to content

Commit 992c064

Browse files
orxfunorxfun
committed
clone and index traits
* Thread safe `Clone` method is implemented. * `Index` and `IndexMut` traits are implemented for `usize` indices. * Due to possibly fragmented nature of the underlying pinned vec, and since we cannot return a reference to a temporarily created collection, currently index over a range is not possible. This would be possible with `IndexMove` (see rust-lang/rfcs#997). * Debug trait implementation is made mode convenient, revealing the current len and capacity in addition to the elements. * Upgraded dependencies to pinned-vec (3.7) and pinned-concurrent-col (2.6) versions. * Tests are extended: * concurrent clone is tested. * in addition to concurrent push & read, safety of concurrent extend & read is also tested. * boxcar::Vec is added to the benchmarks.
1 parent 430efe4 commit 992c064

23 files changed

+683
-353
lines changed

‎Cargo.toml

+7-6
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
[package]
22
name = "orx-concurrent-vec"
3-
version = "2.5.0"
3+
version = "2.6.0"
44
edition = "2021"
55
authors = ["orxfun <[email protected]>"]
66
description = "An efficient, convenient and lightweight grow-only read & write concurrent data structure allowing high performance concurrent collection."
@@ -11,19 +11,20 @@ categories = ["data-structures", "concurrency", "rust-patterns"]
1111

1212
[dependencies]
1313
orx-pseudo-default = "1.4"
14-
orx-pinned-vec = "3.4"
15-
orx-fixed-vec = "3.4"
16-
orx-split-vec = "3.4"
17-
orx-pinned-concurrent-col = "2.4"
14+
orx-pinned-vec = "3.7"
15+
orx-fixed-vec = "3.7"
16+
orx-split-vec = "3.7"
17+
orx-pinned-concurrent-col = "2.6"
1818
orx-concurrent-option = "1.1"
1919

2020
[dev-dependencies]
21-
orx-concurrent-bag = "2.4"
21+
orx-concurrent-bag = "2.6"
2222
criterion = "0.5.1"
2323
rand = "0.8.5"
2424
rayon = "1.9.0"
2525
test-case = "3.3.1"
2626
append-only-vec = "0.1.5"
27+
boxcar = "0.2.5"
2728

2829
[[bench]]
2930
name = "collect_with_extend"

‎README.md

+11-60
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,11 @@ assert_eq!(measurements.len(), 100);
9696
assert_eq!(averages.len(), 10);
9797
```
9898

99+
You may find more concurrent grow & read examples in the respective test files:
100+
* [tests/concurrent_push_read.rs](https://github.com/orxfun/orx-concurrent-vec/blob/main/tests/concurrent_push_read.rs)
101+
* [tests/concurrent_extend_read.rs](https://github.com/orxfun/orx-concurrent-vec/blob/main/tests/concurrent_extend_read.rs)
102+
* [tests/concurrent_clone.rs](https://github.com/orxfun/orx-concurrent-vec/blob/main/tests/concurrent_clone.rs)
103+
99104
## Properties of the Concurrent Model
100105

101106
ConcurrentVec wraps a [`PinnedVec`](https://crates.io/crates/orx-pinned-vex) of [`ConcurrentOption`](https://crates.io/crates/orx-concurrent-option) elements. This composition leads to the following safety guarantees:
@@ -118,35 +123,15 @@ Together, concurrency model of the `ConcurrentVec` has the following properties:
118123

119124
*You may find the details of the benchmarks at [benches/collect_with_push.rs](https://github.com/orxfun/orx-concurrent-vec/blob/main/benches/collect_with_push.rs).*
120125

121-
In the experiment, `rayon`s parallel iterator, `AppendOnlyVec`s and `ConcurrentVec`s `push` methods are used to collect results from multiple threads. Further, different underlying pinned vectors of the `ConcurrentVec` are tested. We observe that:
126+
In the experiment, `rayon`s parallel iterator, and push methods of `AppendOnlyVec`, `boxcar::Vec` and `ConcurrentVec` are used to collect results from multiple threads. Further, different underlying pinned vectors of the `ConcurrentVec` are evaluated.
127+
128+
<img src="https://raw.githubusercontent.com/orxfun/orx-concurrent-vec/main/docs/img/bench_collect_with_push.PNG" alt="https://raw.githubusercontent.com/orxfun/orx-concurrent-vec/main/docs/img/bench_collect_with_push.PNG" />
129+
130+
We observe that:
122131
* The default `Doubling` growth strategy leads to efficient concurrent collection of results. Note that this variant does not require any input to construct.
123132
* On the other hand, `Linear` growth strategy performs significantly better. Note that value of this argument means that each fragment of the underlying `SplitVec` will have a capacity of 2^12 (4096) elements. The underlying reason of improvement is potentially be due to less waste and could be preferred with minor knowledge of the data to be pushed.
124133
* Finally, `Fixed` growth strategy is the least flexible and requires perfect knowledge about the hard-constrained capacity (will panic if we exceed). Since it does not outperform `Linear`, we do not necessarily prefer `Fixed` even if we have the perfect knowledge.
125134

126-
```bash
127-
rayon/num_threads=8,num_items_per_thread-type=[16384]
128-
time: [16.057 ms 16.390 ms 16.755 ms]
129-
append_only_vec/num_threads=8,num_items_per_thread-type=[16384]
130-
time: [23.679 ms 24.480 ms 25.439 ms]
131-
concurrent_vec(Doubling)/num_threads=8,num_items_per_thread-type=[16384]
132-
time: [14.055 ms 14.287 ms 14.526 ms]
133-
concurrent_vec(Linear(12))/num_threads=8,num_items_per_thread-type=[16384]
134-
time: [8.4686 ms 8.6396 ms 8.8373 ms]
135-
concurrent_vec(Fixed)/num_threads=8,num_items_per_thread-type=[16384]
136-
time: [9.8297 ms 9.9945 ms 10.151 ms]
137-
138-
rayon/num_threads=8,num_items_per_thread-type=[65536]
139-
time: [43.118 ms 44.065 ms 45.143 ms]
140-
append_only_vec/num_threads=8,num_items_per_thread-type=[65536]
141-
time: [110.66 ms 114.09 ms 117.94 ms]
142-
concurrent_vec(Doubling)/num_threads=8,num_items_per_thread-type=[65536]
143-
time: [61.461 ms 62.547 ms 63.790 ms]
144-
concurrent_vec(Linear(12))/num_threads=8,num_items_per_thread-type=[65536]
145-
time: [37.420 ms 37.740 ms 38.060 ms]
146-
concurrent_vec(Fixed)/num_threads=8,num_items_per_thread-type=[65536]
147-
time: [43.017 ms 43.584 ms 44.160 ms]
148-
```
149-
150135
The performance can further be improved by using `extend` method instead of `push`. You may see results in the next subsection and details in the [performance notes](https://docs.rs/orx-concurrent-bag/2.3.0/orx_concurrent_bag/#section-performance-notes) of `ConcurrentBag` which has similar characteristics.
151136

152137
### Performance with `extend`
@@ -158,41 +143,7 @@ The only difference in this follow up experiment is that we use `extend` rather
158143
* There is not a significant difference between extending by batches of 64 elements or batches of 65536 elements. We do not need a well tuned number, a large enough batch size seems to be just fine.
159144
* Not all scenarios allow to extend in batches; however, the significant performance improvement makes it preferable whenever possible.
160145

161-
```bash
162-
rayon/num_threads=8,num_items_per_thread-type=[16384]
163-
time: [16.102 ms 16.379 ms 16.669 ms]
164-
append_only_vec/num_threads=8,num_items_per_thread-type=[16384]
165-
time: [27.922 ms 28.611 ms 29.356 ms]
166-
concurrent_vec(Doubling) | batch-size=64/num_threads=8,num_items_per_thread-type=[16384]
167-
time: [8.7361 ms 8.8347 ms 8.9388 ms]
168-
concurrent_vec(Linear(12)) | batch-size=64/num_threads=8,num_items_per_thread-type=[16384]
169-
time: [4.2035 ms 4.2975 ms 4.4012 ms]
170-
concurrent_vec(Fixed) | batch-size=64/num_threads=8,num_items_per_thread-type=[16384]
171-
time: [4.9670 ms 5.0928 ms 5.2217 ms]
172-
concurrent_vec(Doubling) | batch-size=16384/num_threads=8,num_items_per_thread-type=[16384]
173-
time: [9.2441 ms 9.3988 ms 9.5594 ms]
174-
concurrent_vec(Linear(12)) | batch-size=16384/num_threads=8,num_items_per_thread-type=[16384]
175-
time: [3.5663 ms 3.6527 ms 3.7405 ms]
176-
concurrent_vec(Fixed) | batch-size=16384/num_threads=8,num_items_per_thread-type=[16384]
177-
time: [5.0839 ms 5.2169 ms 5.3576 ms]
178-
179-
rayon/num_threads=8,num_items_per_thread-type=[65536]
180-
time: [47.861 ms 48.836 ms 49.843 ms]
181-
append_only_vec/num_threads=8,num_items_per_thread-type=[65536]
182-
time: [125.52 ms 128.89 ms 132.41 ms]
183-
concurrent_vec(Doubling) | batch-size=64/num_threads=8,num_items_per_thread-type=[65536]
184-
time: [42.516 ms 43.097 ms 43.715 ms]
185-
concurrent_vec(Linear(12)) | batch-size=64/num_threads=8,num_items_per_thread-type=[65536]
186-
time: [20.025 ms 20.269 ms 20.521 ms]
187-
concurrent_vec(Fixed) | batch-size=64/num_threads=8,num_items_per_thread-type=[65536]
188-
time: [25.284 ms 25.818 ms 26.375 ms]
189-
concurrent_vec(Doubling) | batch-size=65536/num_threads=8,num_items_per_thread-type=[65536]
190-
time: [39.371 ms 39.887 ms 40.470 ms]
191-
concurrent_vec(Linear(12)) | batch-size=65536/num_threads=8,num_items_per_thread-type=[65536]
192-
time: [17.808 ms 17.923 ms 18.046 ms]
193-
concurrent_vec(Fixed) | batch-size=65536/num_threads=8,num_items_per_thread-type=[65536]
194-
time: [24.291 ms 24.702 ms 25.133 ms]
195-
```
146+
<img src="https://raw.githubusercontent.com/orxfun/orx-concurrent-vec/main/docs/img/bench_collect_with_extend.PNG" alt="https://raw.githubusercontent.com/orxfun/orx-concurrent-vec/main/docs/img/bench_collect_with_extend.PNG" />
196147

197148
## Concurrent Friend Collections
198149

‎benches/collect_with_extend.rs

+42-9
Original file line numberDiff line numberDiff line change
@@ -43,8 +43,7 @@ fn with_concurrent_vec<T: Sync, P: IntoConcurrentPinnedVec<ConcurrentOption<T>>>
4343
for _ in 0..num_threads {
4444
s.spawn(|| {
4545
for j in (0..num_items_per_thread).step_by(batch_size) {
46-
let into_iter =
47-
(j..(j + batch_size)).map(|j| std::hint::black_box(compute(j, j + 1)));
46+
let into_iter = (j..(j + batch_size)).map(|j| compute(j, j + 1));
4847
vec.extend(into_iter);
4948
}
5049
});
@@ -65,7 +64,7 @@ fn with_rayon<T: Send + Sync + Clone + Copy>(
6564
.into_par_iter()
6665
.flat_map(|_| {
6766
(0..num_items_per_thread)
68-
.map(move |j| std::hint::black_box(compute(j, j + 1)))
67+
.map(move |j| compute(j, j + 1))
6968
.collect::<Vec<_>>()
7069
})
7170
.collect();
@@ -83,7 +82,26 @@ fn with_append_only_vec<T: Send + Sync + Clone + Copy>(
8382
for _ in 0..num_threads {
8483
s.spawn(|| {
8584
for j in 0..num_items_per_thread {
86-
vec.push(std::hint::black_box(compute(j, j + 1)));
85+
vec.push(compute(j, j + 1));
86+
}
87+
});
88+
}
89+
});
90+
91+
vec
92+
}
93+
94+
fn with_boxcar<T: Send + Sync + Clone + Copy>(
95+
num_threads: usize,
96+
num_items_per_thread: usize,
97+
compute: fn(usize, usize) -> T,
98+
vec: boxcar::Vec<T>,
99+
) -> boxcar::Vec<T> {
100+
std::thread::scope(|s| {
101+
for _ in 0..num_threads {
102+
s.spawn(|| {
103+
for j in 0..num_items_per_thread {
104+
vec.push(compute(j, j + 1));
87105
}
88106
});
89107
}
@@ -105,14 +123,16 @@ fn bench_grow(c: &mut Criterion) {
105123

106124
let max_len = num_threads * num_items_per_thread;
107125

126+
let compute = compute_large_data;
127+
108128
// rayon
109129

110130
group.bench_with_input(BenchmarkId::new("rayon", &treatment), &(), |b, _| {
111131
b.iter(|| {
112132
black_box(with_rayon(
113133
black_box(num_threads),
114134
black_box(num_items_per_thread),
115-
compute_large_data,
135+
compute,
116136
))
117137
})
118138
});
@@ -127,13 +147,26 @@ fn bench_grow(c: &mut Criterion) {
127147
black_box(with_append_only_vec(
128148
black_box(num_threads),
129149
black_box(num_items_per_thread),
130-
compute_large_data,
150+
compute,
131151
AppendOnlyVec::new(),
132152
))
133153
})
134154
},
135155
);
136156

157+
// BOXCAR
158+
159+
group.bench_with_input(BenchmarkId::new("boxcar", &treatment), &(), |b, _| {
160+
b.iter(|| {
161+
black_box(with_boxcar(
162+
black_box(num_threads),
163+
black_box(num_items_per_thread),
164+
compute,
165+
boxcar::Vec::new(),
166+
))
167+
})
168+
});
169+
137170
// ConcurrentVec
138171

139172
let batch_sizes = vec![64, num_items_per_thread];
@@ -154,7 +187,7 @@ fn bench_grow(c: &mut Criterion) {
154187
black_box(with_concurrent_vec(
155188
black_box(num_threads),
156189
black_box(num_items_per_thread),
157-
compute_large_data,
190+
compute,
158191
batch_size,
159192
ConcurrentVec::with_doubling_growth(),
160193
))
@@ -172,7 +205,7 @@ fn bench_grow(c: &mut Criterion) {
172205
black_box(with_concurrent_vec(
173206
black_box(num_threads),
174207
black_box(num_items_per_thread),
175-
compute_large_data,
208+
compute,
176209
batch_size,
177210
ConcurrentVec::with_linear_growth(12, num_linear_fragments),
178211
))
@@ -185,7 +218,7 @@ fn bench_grow(c: &mut Criterion) {
185218
black_box(with_concurrent_vec(
186219
black_box(num_threads),
187220
black_box(num_items_per_thread),
188-
compute_large_data,
221+
compute,
189222
batch_size,
190223
ConcurrentVec::with_fixed_capacity(num_threads * num_items_per_thread),
191224
))

‎benches/collect_with_push.rs

+35-3
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ fn with_concurrent_vec<T: Sync, P: IntoConcurrentPinnedVec<ConcurrentOption<T>>>
4242
for _ in 0..num_threads {
4343
s.spawn(|| {
4444
for j in 0..num_items_per_thread {
45-
vec.push(std::hint::black_box(compute(j, j + 1)));
45+
vec.push(compute(j, j + 1));
4646
}
4747
});
4848
}
@@ -62,7 +62,7 @@ fn with_rayon<T: Send + Sync + Clone + Copy>(
6262
.into_par_iter()
6363
.flat_map(|_| {
6464
(0..num_items_per_thread)
65-
.map(move |j| std::hint::black_box(compute(j, j + 1)))
65+
.map(move |j| compute(j, j + 1))
6666
.collect::<Vec<_>>()
6767
})
6868
.collect();
@@ -80,7 +80,26 @@ fn with_append_only_vec<T: Send + Sync + Clone + Copy>(
8080
for _ in 0..num_threads {
8181
s.spawn(|| {
8282
for j in 0..num_items_per_thread {
83-
vec.push(std::hint::black_box(compute(j, j + 1)));
83+
vec.push(compute(j, j + 1));
84+
}
85+
});
86+
}
87+
});
88+
89+
vec
90+
}
91+
92+
fn with_boxcar<T: Send + Sync + Clone + Copy>(
93+
num_threads: usize,
94+
num_items_per_thread: usize,
95+
compute: fn(usize, usize) -> T,
96+
vec: boxcar::Vec<T>,
97+
) -> boxcar::Vec<T> {
98+
std::thread::scope(|s| {
99+
for _ in 0..num_threads {
100+
s.spawn(|| {
101+
for j in 0..num_items_per_thread {
102+
vec.push(compute(j, j + 1));
84103
}
85104
});
86105
}
@@ -133,6 +152,19 @@ fn bench_grow(c: &mut Criterion) {
133152
},
134153
);
135154

155+
// BOXCAR
156+
157+
group.bench_with_input(BenchmarkId::new("boxcar", &treatment), &(), |b, _| {
158+
b.iter(|| {
159+
black_box(with_boxcar(
160+
black_box(num_threads),
161+
black_box(num_items_per_thread),
162+
compute,
163+
boxcar::Vec::new(),
164+
))
165+
})
166+
});
167+
136168
// WITH-SCOPE
137169

138170
group.bench_with_input(

‎benches/results/collect.xlsx

-5.25 KB
Binary file not shown.

‎docs/img/bench_collect_with_extend.PNG

6.34 KB
Loading

‎docs/img/bench_collect_with_push.PNG

6.58 KB
Loading

‎src/common_traits/clone.rs

+14
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
use crate::ConcurrentVec;
2+
use orx_concurrent_option::ConcurrentOption;
3+
use orx_fixed_vec::IntoConcurrentPinnedVec;
4+
5+
impl<T, P> Clone for ConcurrentVec<T, P>
6+
where
7+
P: IntoConcurrentPinnedVec<ConcurrentOption<T>>,
8+
T: Clone,
9+
{
10+
fn clone(&self) -> Self {
11+
let core = unsafe { self.core.clone_with_len(self.len()) };
12+
Self { core }
13+
}
14+
}

‎src/common_traits/debug.rs

+46
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
use crate::ConcurrentVec;
2+
use orx_concurrent_option::ConcurrentOption;
3+
use orx_fixed_vec::IntoConcurrentPinnedVec;
4+
use std::fmt::Debug;
5+
6+
const ELEM_PER_LINE: usize = 8;
7+
8+
impl<T, P> Debug for ConcurrentVec<T, P>
9+
where
10+
P: IntoConcurrentPinnedVec<ConcurrentOption<T>>,
11+
T: Debug,
12+
{
13+
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
14+
let len = self.len();
15+
let capacity = self.capacity();
16+
17+
write!(f, "ConcurrentVec {{")?;
18+
write!(f, "\n len: {},", len)?;
19+
write!(f, "\n capacity: {},", capacity)?;
20+
write!(f, "\n data: [,")?;
21+
for i in 0..len {
22+
if i % ELEM_PER_LINE == 0 {
23+
write!(f, "\n ")?;
24+
}
25+
match self.get(i) {
26+
Some(x) => write!(f, "{:?}, ", x)?,
27+
None => write!(f, "*, ")?,
28+
}
29+
}
30+
write!(f, "\n ],")?;
31+
write!(f, "\n}}")
32+
}
33+
}
34+
35+
#[cfg(test)]
36+
mod tests {
37+
use super::*;
38+
39+
#[test]
40+
fn debug() {
41+
let vec = ConcurrentVec::new();
42+
vec.extend([0, 4, 1, 2, 5, 6, 32, 5, 1, 121, 2, 42]);
43+
let dbg_str = format!("{:?}", &vec);
44+
assert_eq!(dbg_str, "ConcurrentVec {\n len: 12,\n capacity: 12,\n data: [,\n 0, 4, 1, 2, 5, 6, 32, 5, \n 1, 121, 2, 42, \n ],\n}");
45+
}
46+
}

0 commit comments

Comments
 (0)