Skip to content

Commit f43c4df

Browse files
asder8215asder8215
committed
phased out pin policy in sharded ring buf to be in its own file (all other policies belong in exp_shardedringbuf.rs)
1 parent 41fca91 commit f43c4df

15 files changed

Lines changed: 2256 additions & 597 deletions

‎Cargo.lock‎

Lines changed: 4 additions & 4 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎Cargo.toml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,8 @@ documentation = "https://docs.rs/sharded_ringbuf/latest/sharded_ringbuf/"
1919
async-executor = "1.13.3"
2020
crossbeam-utils = "^0.8.21"
2121
fastrand = "^2.3.0"
22-
futures-util = "0.3.31"
23-
smol = "2.0.2"
2422
tokio = {version = "^1.45.1", features = ["rt", "sync"] }
23+
futures-util = "0.3.31"
2524

2625
[dev-dependencies]
2726
criterion = {version = "^0.6.0", features =["async_tokio"] }
@@ -30,6 +29,7 @@ kanal = "^0.1.1"
3029
async-channel = "2.5.0"
3130
crossbeam = "0.8"
3231
flume = {version="0.11", features= ["async"] }
32+
smol = "2.0.2"
3333

3434
[[bench]]
3535
name = "srb_shiftby"

‎benches/kanal_async.rs‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -89,23 +89,23 @@ async fn kanal_async(c: usize, task_count: usize) {
8989
let (s, r) = bounded_async(c);
9090
let mut handles = Vec::new();
9191

92-
for _ in 0..1 {
92+
for _ in 0..task_count {
9393
let rx = r.clone();
9494
handles.push(task::spawn(async move {
95-
for _ in 0..task_count {
96-
for _ in 0..1 {
95+
// for _ in 0..task_count {
96+
for _ in 0..250000 {
9797
let x = rx.recv().await.unwrap();
9898
// test_func(x as u128);
9999
}
100-
}
100+
// }
101101
}));
102102
}
103103

104104
for _ in 0..task_count {
105105
let tx = s.clone();
106106
// let msg = Message::default();
107107
handles.push(task::spawn(async move {
108-
for i in 0..1 {
108+
for i in 0..250000 {
109109
// tx.send(msg).await.unwrap();
110110
tx.send(i).await.unwrap();
111111
}
@@ -130,7 +130,7 @@ async fn kanal_async_with_msg_vec(
130130
for _ in 0..8 {
131131
let rx = r.clone();
132132
handles.push(task::spawn(async move {
133-
for _ in 0..task_count/8 {
133+
for _ in 0..task_count / 8 {
134134
for _ in 0..msg_count {
135135
let x = rx.recv().await.unwrap();
136136
// test_func(x.item_one as u128);
@@ -156,9 +156,9 @@ async fn kanal_async_with_msg_vec(
156156
}
157157

158158
fn benchmark_kanal_async(c: &mut Criterion) {
159-
const MAX_THREADS: [usize; 1] = [8];
160-
const CAPACITY: usize = 128;
161-
const TASKS: [usize; 1] = [100];
159+
const MAX_THREADS: [usize; 1] = [4];
160+
const CAPACITY: usize = 1024;
161+
const TASKS: [usize; 1] = [4];
162162

163163
for thread_num in MAX_THREADS {
164164
let runtime = tokio::runtime::Builder::new_multi_thread()

‎benches/srb_cft.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
22
use sharded_ringbuf::{
3-
ShardedRingBuf, cft_spawn_dequeuer_full_unbounded, cft_spawn_dequeuer_unbounded,
3+
ExpShardedRingBuf, cft_spawn_dequeuer_full_unbounded, cft_spawn_dequeuer_unbounded,
44
cft_spawn_enqueuer_with_iterator, spawn_assigner, terminate_assigner,
55
};
66
use std::sync::Arc;
@@ -16,7 +16,7 @@ fn test_add(x: usize) -> usize {
1616
async fn srb_cft(capacity: usize, shards: usize, task_count: usize) {
1717
let max_items: usize = capacity;
1818

19-
let rb = Arc::new(ShardedRingBuf::new(max_items, shards));
19+
let rb = Arc::new(ExpShardedRingBuf::new(max_items, shards));
2020

2121
let mut deq_tasks = Vec::with_capacity(shards);
2222
let mut enq_tasks = Vec::with_capacity(task_count);
@@ -55,7 +55,7 @@ async fn srb_cft(capacity: usize, shards: usize, task_count: usize) {
5555
async fn srb_cft_deq_full(capacity: usize, shards: usize, task_count: usize) {
5656
let max_items: usize = capacity;
5757

58-
let rb = Arc::new(ShardedRingBuf::new(max_items, shards));
58+
let rb = Arc::new(ExpShardedRingBuf::new(max_items, shards));
5959

6060
let mut deq_tasks = Vec::with_capacity(shards);
6161
let mut enq_tasks = Vec::with_capacity(task_count);

‎benches/srb_pin.rs‎

Lines changed: 62 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,10 @@
11
use criterion::async_executor::AsyncExecutor;
22
use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
3+
use sharded_ringbuf::{ExpShardedRingBuf, spawn_dequeuer_unbounded, spawn_enqueuer_with_iterator};
34
use sharded_ringbuf::{
45
MLFShardedRingBuf, ShardPolicy, mlf_spawn_dequeuer_unbounded, mlf_spawn_enqueuer_with_iterator,
56
spawn_dequeuer_full_unbounded, spawn_enqueuer_full_with_iterator,
67
};
7-
use sharded_ringbuf::{ShardedRingBuf, spawn_dequeuer_unbounded, spawn_enqueuer_with_iterator};
88
use smol::future;
99
use std::future::Future;
1010
use std::sync::Arc;
@@ -93,24 +93,15 @@ fn mul_stress(iter: usize) -> u128 {
9393
acc
9494
}
9595

96-
/// Minimal executor adapter so Criterion can call smol::block_on
97-
#[derive(Clone, Copy)]
98-
struct SmolExecutor;
99-
100-
impl AsyncExecutor for SmolExecutor {
101-
fn block_on<T>(&self, future: impl Future<Output = T>) -> T {
102-
future::block_on(future)
103-
}
104-
}
105-
10696
async fn lfsrb_pin(capacity: usize, shards: usize, task_count: usize) {
10797
let max_items: usize = capacity;
10898

109-
let rb = Arc::new(ShardedRingBuf::new(max_items, shards));
99+
let rb = Arc::new(ExpShardedRingBuf::new(max_items, shards));
110100

111101
let mut deq_tasks = Vec::with_capacity(shards);
112102
let mut enq_tasks = Vec::with_capacity(task_count);
113103

104+
// necessary for poison
114105
let notifier_task = spawn({
115106
let rb_clone = rb.clone();
116107
async move {
@@ -133,18 +124,35 @@ async fn lfsrb_pin(capacity: usize, shards: usize, task_count: usize) {
133124
// );
134125
// enq_tasks.push(handle);
135126
// }
127+
// for i in 0..task_count {
128+
// let handle = spawn_enqueuer_with_iterator(
129+
// rb.clone(),
130+
// ShardPolicy::Pin { initial_index: i },
131+
// 0..250000,
132+
// );
133+
// enq_tasks.push(handle);
134+
// }
135+
136136
for i in 0..task_count {
137-
let handle = spawn_enqueuer_with_iterator(
137+
let handle = spawn_enqueuer_full_with_iterator(
138138
rb.clone(),
139139
ShardPolicy::Pin { initial_index: i },
140-
0..=250000,
140+
0..250000,
141141
);
142142
enq_tasks.push(handle);
143143
}
144144

145-
for i in 0..shards {
145+
// for i in 0..task_count {
146+
// let handle =
147+
// spawn_dequeuer_unbounded(rb.clone(), ShardPolicy::Pin { initial_index: i }, |x| {
148+
// // test_func(x as u128);
149+
// });
150+
// deq_tasks.push(handle);
151+
// }
152+
153+
for i in 0..task_count {
146154
let handle =
147-
spawn_dequeuer_unbounded(rb.clone(), ShardPolicy::Pin { initial_index: i }, |x| {
155+
spawn_dequeuer_full_unbounded(rb.clone(), ShardPolicy::Pin { initial_index: i }, |x| {
148156
// test_func(x as u128);
149157
});
150158
deq_tasks.push(handle);
@@ -212,13 +220,15 @@ async fn mlfsrb_pin(capacity: usize, shards: usize, task_count: usize) {
212220
let handle = tokio::spawn({
213221
let rb_clone = rb.clone();
214222
async move {
215-
rb_clone.enqueue(i).await;
223+
for j in 0..250000 {
224+
rb_clone.enqueue(j).await;
225+
}
216226
}
217227
});
218228
enq_tasks.push(handle);
219229
}
220230

221-
for i in 0..shards {
231+
for i in 0..task_count {
222232
let handle = tokio::spawn({
223233
let rb_clone = rb.clone();
224234
async move {
@@ -227,9 +237,7 @@ async fn mlfsrb_pin(capacity: usize, shards: usize, task_count: usize) {
227237
let item = rb_clone.dequeue_in_shard(i).await;
228238
match item {
229239
None => break,
230-
Some(val) => {
231-
232-
}
240+
Some(val) => {}
233241
}
234242
}
235243
}
@@ -242,7 +250,7 @@ async fn mlfsrb_pin(capacity: usize, shards: usize, task_count: usize) {
242250
enq.await.unwrap();
243251
}
244252

245-
for i in 0..shards {
253+
for i in 0..task_count {
246254
{
247255
let rb = rb.clone();
248256
spawn(async move {
@@ -262,7 +270,7 @@ async fn mlfsrb_pin(capacity: usize, shards: usize, task_count: usize) {
262270
async fn lfsrb_pin_deq_full(capacity: usize, shards: usize, task_count: usize) {
263271
let max_items: usize = capacity;
264272

265-
let rb = Arc::new(ShardedRingBuf::new(max_items, shards));
273+
let rb = Arc::new(ExpShardedRingBuf::new(max_items, shards));
266274

267275
let mut deq_tasks = Vec::with_capacity(shards);
268276
let mut enq_tasks = Vec::with_capacity(task_count);
@@ -380,7 +388,7 @@ async fn lfsrb_pin_with_msg_vec(
380388
) {
381389
let max_items: usize = capacity;
382390

383-
let rb = Arc::new(ShardedRingBuf::new(max_items, shards));
391+
let rb = Arc::new(ExpShardedRingBuf::new(max_items, shards));
384392

385393
let mut deq_tasks = Vec::with_capacity(shards);
386394
let mut enq_tasks = Vec::with_capacity(task_count);
@@ -568,13 +576,13 @@ async fn mlfsrb_pin_with_msg_vec(
568576

569577
fn benchmark_pin(c: &mut Criterion) {
570578
// const MAX_THREADS: [usize; 2] = [4, 8];
571-
const MAX_THREADS: [usize; 1] = [8];
572-
const CAPACITY: usize = 128;
579+
const MAX_THREADS: [usize; 1] = [4];
580+
const CAPACITY: usize = 1000000;
573581
// const CAPACITY: usize = 200000;
574582
// const SHARDS: [usize; 5] = [1, 2, 4, 8, 16];
575583
// const TASKS: [usize; 5] = [1, 2, 4, 8, 16];
576-
const SHARDS: [usize; 1] = [1];
577-
const TASKS: [usize; 1] = [100];
584+
const SHARDS: [usize; 1] = [4];
585+
const TASKS: [usize; 1] = [4];
578586

579587
// const MSG_COUNT: usize = 250000;
580588
// // let msg = BigData { buf: Box::new([0; 1 * 1024]) };
@@ -590,34 +598,34 @@ fn benchmark_pin(c: &mut Criterion) {
590598
// }
591599
// }
592600

593-
// for thread_num in MAX_THREADS {
594-
// let runtime = tokio::runtime::Builder::new_multi_thread()
595-
// .enable_all()
596-
// .worker_threads(thread_num)
597-
// .build()
598-
// .unwrap();
601+
for thread_num in MAX_THREADS {
602+
let runtime = tokio::runtime::Builder::new_multi_thread()
603+
.enable_all()
604+
.worker_threads(thread_num)
605+
.build()
606+
.unwrap();
599607

600-
// for shard_num in SHARDS {
601-
// for task_count in TASKS {
602-
// let func_name = format!(
603-
// "Pin: {} threads, {} shards, {} enq tasks enqueuing 1 million items, {} looping deq task",
604-
// thread_num, shard_num, task_count, shard_num
605-
// );
608+
for shard_num in SHARDS {
609+
for task_count in TASKS {
610+
let func_name = format!(
611+
"Pin: {} threads, {} shards, {} enq tasks enqueuing 1 million items, {} looping deq task",
612+
thread_num, shard_num, task_count, shard_num
613+
);
606614

607-
// c.bench_with_input(
608-
// BenchmarkId::new(func_name, CAPACITY),
609-
// &(CAPACITY),
610-
// |b, &cap| {
611-
// // Insert a call to `to_async` to convert the bencher to async mode.
612-
// // The timing loops are the same as with the normal bencher.
613-
// b.to_async(&runtime).iter(async || {
614-
// lfsrb_pin(cap, shard_num, task_count).await;
615-
// });
616-
// },
617-
// );
618-
// }
619-
// }
620-
// }
615+
c.bench_with_input(
616+
BenchmarkId::new(func_name, CAPACITY),
617+
&(CAPACITY),
618+
|b, &cap| {
619+
// Insert a call to `to_async` to convert the bencher to async mode.
620+
// The timing loops are the same as with the normal bencher.
621+
b.to_async(&runtime).iter(async || {
622+
lfsrb_pin(cap, shard_num, task_count).await;
623+
});
624+
},
625+
);
626+
}
627+
}
628+
}
621629

622630
// for thread_num in MAX_THREADS {
623631
// let runtime = tokio::runtime::Builder::new_multi_thread()

‎benches/srb_randomsweep.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
2+
use sharded_ringbuf::{ExpShardedRingBuf, spawn_dequeuer_unbounded, spawn_enqueuer_with_iterator};
23
use sharded_ringbuf::{ShardPolicy, spawn_dequeuer_full_unbounded};
3-
use sharded_ringbuf::{ShardedRingBuf, spawn_dequeuer_unbounded, spawn_enqueuer_with_iterator};
44
use std::sync::Arc;
55

66
fn test_add(x: usize) -> usize {
@@ -14,7 +14,7 @@ fn test_add(x: usize) -> usize {
1414
async fn lfsrb_random_and_sweep(capacity: usize, shards: usize, task_count: usize) {
1515
let max_items: usize = capacity;
1616

17-
let rb = Arc::new(ShardedRingBuf::new(max_items, shards));
17+
let rb = Arc::new(ExpShardedRingBuf::new(max_items, shards));
1818

1919
let mut deq_tasks = Vec::with_capacity(shards);
2020
let mut enq_tasks = Vec::with_capacity(task_count);
@@ -49,7 +49,7 @@ async fn lfsrb_random_and_sweep(capacity: usize, shards: usize, task_count: usiz
4949
async fn lfsrb_random_and_sweep_deq_full(capacity: usize, shards: usize, task_count: usize) {
5050
let max_items: usize = capacity;
5151

52-
let rb = Arc::new(ShardedRingBuf::new(max_items, shards));
52+
let rb = Arc::new(ExpShardedRingBuf::new(max_items, shards));
5353

5454
let mut deq_tasks = Vec::with_capacity(shards);
5555
let mut enq_tasks = Vec::with_capacity(task_count);

‎benches/srb_shiftby.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
2+
use sharded_ringbuf::{ExpShardedRingBuf, spawn_dequeuer_unbounded, spawn_enqueuer_with_iterator};
23
use sharded_ringbuf::{ShardPolicy, spawn_dequeuer_full_unbounded};
3-
use sharded_ringbuf::{ShardedRingBuf, spawn_dequeuer_unbounded, spawn_enqueuer_with_iterator};
44
use std::sync::Arc;
55

66
fn test_add(x: usize) -> usize {
@@ -14,7 +14,7 @@ fn test_add(x: usize) -> usize {
1414
async fn lfsrb_shiftby(capacity: usize, shards: usize, task_count: usize) {
1515
let max_items: usize = capacity;
1616

17-
let rb = Arc::new(ShardedRingBuf::new(max_items, shards));
17+
let rb = Arc::new(ExpShardedRingBuf::new(max_items, shards));
1818

1919
let mut deq_tasks = Vec::with_capacity(shards);
2020
let mut enq_tasks = Vec::with_capacity(task_count);
@@ -62,7 +62,7 @@ async fn lfsrb_shiftby(capacity: usize, shards: usize, task_count: usize) {
6262
async fn lfsrb_shiftby_deq_full(capacity: usize, shards: usize, task_count: usize) {
6363
let max_items: usize = capacity;
6464

65-
let rb = Arc::new(ShardedRingBuf::new(max_items, shards));
65+
let rb = Arc::new(ExpShardedRingBuf::new(max_items, shards));
6666

6767
let mut deq_tasks = Vec::with_capacity(shards);
6868
let mut enq_tasks = Vec::with_capacity(task_count);

0 commit comments

Comments
 (0)