Skip to content

Commit 4f0be77

Browse files
asder8215asder8215
committed
change some benchmarking
1 parent 634f791 commit 4f0be77

2 files changed

Lines changed: 85 additions & 80 deletions

File tree

‎benches/kanal_async.rs‎

Lines changed: 78 additions & 76 deletions
Original file line numberDiff line numberDiff line change
@@ -97,19 +97,20 @@ async fn kanal_async(c: usize, task_count: usize) {
9797
let rx = r.clone();
9898
handles.push(task::spawn(async move {
9999
// for _ in 0..task_count {
100-
for _ in 0..250000 {
100+
// for _ in 0..10_000_000 {
101+
for _ in 0..10_000_000 / task_count {
101102
let x = rx.recv().await.unwrap();
102-
test_func(x as u128);
103+
// test_func(x as u128);
103104
}
104105
// }
105106
}));
106107
}
107108

108-
for _ in 0..task_count {
109+
for _ in 0..1 {
109110
let tx = s.clone();
110111
// let msg = Message::default();
111112
handles.push(task::spawn(async move {
112-
for i in 0..250000 {
113+
for i in 0 as i64..10_000_000 {
113114
// tx.send(msg).await.unwrap();
114115
tx.send(i).await.unwrap();
115116
}
@@ -161,38 +162,11 @@ async fn kanal_async_with_msg_vec(
161162

162163
fn benchmark_kanal_async(c: &mut Criterion) {
163164
const MAX_THREADS: [usize; 1] = [8];
164-
const CAPACITY: usize = 1024;
165+
// const CAPACITY: usize = 1024;
166+
const CAPACITY: usize = 32768;
165167
const TASKS: [usize; 1] = [5];
166168

167169
let mut group = c.benchmark_group("Kanal Async");
168-
// for thread_num in MAX_THREADS {
169-
// let runtime = tokio::runtime::Builder::new_multi_thread()
170-
// .enable_all()
171-
// .worker_threads(thread_num)
172-
// .build()
173-
// .unwrap();
174-
175-
// for task_count in TASKS {
176-
// let func_name = format!(
177-
// "Kanal Async: {thread_num} threads, {task_count} enq tasks enqueuing 1 million items, 1 looping deq task"
178-
// );
179-
180-
// c.bench_with_input(
181-
// BenchmarkId::new(func_name, CAPACITY),
182-
// &(CAPACITY),
183-
// |b, &cap| {
184-
// // Insert a call to `to_async` to convert the bencher to async mode.
185-
// // The timing loops are the same as with the normal bencher.
186-
// b.to_async(&runtime).iter(async || {
187-
// kanal_async(cap, task_count).await;
188-
// });
189-
// },
190-
// );
191-
// }
192-
// }
193-
const MSG_SIZES: [usize; 8] = [1, 2, 4, 8, 16, 32, 64, 128];
194-
// const MSG_COUNT: usize = 128;
195-
196170
for thread_num in MAX_THREADS {
197171
let runtime = tokio::runtime::Builder::new_multi_thread()
198172
.enable_all()
@@ -201,51 +175,79 @@ fn benchmark_kanal_async(c: &mut Criterion) {
201175
.unwrap();
202176

203177
for task_count in TASKS {
204-
for msg_size in MSG_SIZES {
205-
let msg = BigData {
206-
buf: Box::new([0; 8]),
207-
};
208-
209-
let mut msg_vecs = Vec::with_capacity(TASKS[0]);
210-
for _ in 0..TASKS[0] {
211-
msg_vecs.push(Vec::with_capacity(msg_size));
212-
let msg_vecs_len = msg_vecs.len();
213-
for _ in 0..msg_size {
214-
msg_vecs[msg_vecs_len - 1].push(msg.clone());
215-
}
216-
}
217-
218-
// let func_name = format!(
219-
// "Kanal Async: {thread_num} threads, {task_count} enq tasks enqueuing 1 million items, 1 looping deq task"
220-
// );
221-
let func_name = format!("Batching {msg_size} 8-byte items");
222-
223-
group.bench_with_input(
224-
BenchmarkId::new(func_name, msg_size),
225-
&(CAPACITY),
226-
|b, &cap| {
227-
// Insert a call to `to_async` to convert the bencher to async mode.
228-
// The timing loops are the same as with the normal bencher.
229-
b.to_async(&runtime).iter_custom(|iters| {
230-
let msg_vecs = msg_vecs.clone();
231-
async move {
232-
let mut total = Duration::ZERO;
233-
for _ in 0..iters {
234-
let msg_vecs = msg_vecs.clone();
235-
let start = Instant::now();
236-
kanal_async_with_msg_vec(msg_vecs, cap, task_count, msg_size)
237-
.await;
238-
let end = Instant::now();
239-
total += end - start;
240-
}
241-
total
242-
}
243-
});
244-
},
245-
);
246-
}
178+
let func_name = format!(
179+
"Kanal Async: {thread_num} threads, {task_count} enq tasks enqueuing 1 million items, 1 looping deq task"
180+
);
181+
182+
group.bench_with_input(
183+
BenchmarkId::new(func_name, CAPACITY),
184+
&(CAPACITY),
185+
|b, &cap| {
186+
// Insert a call to `to_async` to convert the bencher to async mode.
187+
// The timing loops are the same as with the normal bencher.
188+
b.to_async(&runtime).iter(async || {
189+
kanal_async(cap, task_count).await;
190+
});
191+
},
192+
);
247193
}
248194
}
195+
// const MSG_SIZES: [usize; 8] = [1, 2, 4, 8, 16, 32, 64, 128];
196+
// const MSG_COUNT: usize = 128;
197+
198+
// for thread_num in MAX_THREADS {
199+
// let runtime = tokio::runtime::Builder::new_multi_thread()
200+
// .enable_all()
201+
// .worker_threads(thread_num)
202+
// .build()
203+
// .unwrap();
204+
205+
// for task_count in TASKS {
206+
// for msg_size in MSG_SIZES {
207+
// let msg = BigData {
208+
// buf: Box::new([0; 8]),
209+
// };
210+
211+
// let mut msg_vecs = Vec::with_capacity(TASKS[0]);
212+
// for _ in 0..TASKS[0] {
213+
// msg_vecs.push(Vec::with_capacity(msg_size));
214+
// let msg_vecs_len = msg_vecs.len();
215+
// for _ in 0..msg_size {
216+
// msg_vecs[msg_vecs_len - 1].push(msg.clone());
217+
// }
218+
// }
219+
220+
// // let func_name = format!(
221+
// // "Kanal Async: {thread_num} threads, {task_count} enq tasks enqueuing 1 million items, 1 looping deq task"
222+
// // );
223+
// let func_name = format!("Batching {msg_size} 8-byte items");
224+
225+
// group.bench_with_input(
226+
// BenchmarkId::new(func_name, msg_size),
227+
// &(CAPACITY),
228+
// |b, &cap| {
229+
// // Insert a call to `to_async` to convert the bencher to async mode.
230+
// // The timing loops are the same as with the normal bencher.
231+
// b.to_async(&runtime).iter_custom(|iters| {
232+
// let msg_vecs = msg_vecs.clone();
233+
// async move {
234+
// let mut total = Duration::ZERO;
235+
// for _ in 0..iters {
236+
// let msg_vecs = msg_vecs.clone();
237+
// let start = Instant::now();
238+
// kanal_async_with_msg_vec(msg_vecs, cap, task_count, msg_size)
239+
// .await;
240+
// let end = Instant::now();
241+
// total += end - start;
242+
// }
243+
// total
244+
// }
245+
// });
246+
// },
247+
// );
248+
// }
249+
// }
250+
// }
249251
}
250252

251253
criterion_group!(benches, benchmark_kanal_async);

‎benches/srb.rs‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,8 @@ async fn srb_bench(capacity: usize, shards: usize, task_count: usize) {
115115
for item in items {
116116
if counter != 0 && counter % full_enq == 0 {
117117
// let _ = rb_clone.enqueue_full_in_shard(enq_vec, i).await;
118-
let _ = rb_clone.enqueue_full(enq_vec).await;
118+
// let _ = rb_clone.enqueue_full(enq_vec).await;
119+
let _ = rb_clone.enqueue(enq_vec).await;
119120
enq_vec = Vec::with_capacity(full_enq);
120121
enq_vec.push(item);
121122
counter += 1;
@@ -126,7 +127,8 @@ async fn srb_bench(capacity: usize, shards: usize, task_count: usize) {
126127
}
127128
if !enq_vec.is_empty() {
128129
// let _ = rb_clone.enqueue_full_in_shard(enq_vec, i).await;
129-
let _ = rb_clone.enqueue_full(enq_vec).await;
130+
// let _ = rb_clone.enqueue_full(enq_vec).await;
131+
let _ = rb_clone.enqueue(enq_vec).await;
130132
}
131133
}
132134
});
@@ -138,7 +140,8 @@ async fn srb_bench(capacity: usize, shards: usize, task_count: usize) {
138140
let rb_clone = rb.clone();
139141
async move {
140142
loop {
141-
match rb_clone.dequeue_full_in_shard(i).await {
143+
// match rb_clone.dequeue_full_in_shard(i).await {
144+
match rb_clone.dequeue_in_shard(i).await {
142145
Some(items) => {
143146
for _j in items {
144147
// println!("{j}");
@@ -260,7 +263,7 @@ fn benchmark_srb(c: &mut Criterion) {
260263
// const SHARDS: [usize; 5] = [1, 2, 4, 8, 16];
261264
// const TASKS: [usize; 5] = [1, 2, 4, 8, 16];
262265
const SHARDS: [usize; 1] = [8];
263-
const TASKS: [usize; 1] = [1];
266+
const TASKS: [usize; 1] = [8];
264267

265268
for thread_num in MAX_THREADS {
266269
let runtime = tokio::runtime::Builder::new_multi_thread()

0 commit comments

Comments
 (0)