-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
11 changed files
with
428 additions
and
142 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,122 @@ | ||
| //! Multicast single-producer, multiple-consumers | ||
|
|
||
| use chute::LendingReader; | ||
| use arrayvec::ArrayVec; | ||
| use criterion::{criterion_group, criterion_main, Criterion, Throughput}; | ||
| use std::sync::Arc; | ||
|
|
||
| mod common; | ||
| use common::*; | ||
|
|
||
| fn chute_spmc(reader_threads: usize){ | ||
| let mut queue: chute::spmc::Queue<_> = Default::default(); | ||
|
|
||
| let mut joins: ArrayVec<_, 64> = Default::default(); | ||
|
|
||
| // READ | ||
| for _ in 0..reader_threads { | ||
| let mut reader = queue.reader(); | ||
| joins.push(std::thread::spawn(move || { | ||
| for _ in 0..COUNT { | ||
| let msg = loop { | ||
| if let Some(msg) = reader.next() { | ||
| break msg; | ||
| } | ||
| yield_fn(); | ||
| }; | ||
| } | ||
| })); | ||
| } | ||
|
|
||
| // WRITE | ||
| for i in 0..COUNT { | ||
| queue.push(message::new(i)); | ||
| } | ||
|
|
||
| for join in joins{ | ||
| join.join().unwrap(); | ||
| } | ||
| } | ||
|
|
||
| fn chute_mpmc(reader_threads: usize){ | ||
| let queue = chute::mpmc::Queue::new(); | ||
|
|
||
| let mut joins: ArrayVec<_, 64> = Default::default(); | ||
|
|
||
| // READ | ||
| for _ in 0..reader_threads { | ||
| let mut reader = queue.reader(); | ||
| joins.push(std::thread::spawn(move || { | ||
| for _ in 0..COUNT { | ||
| let msg = loop { | ||
| if let Some(msg) = reader.next() { | ||
| break msg; | ||
| } | ||
| yield_fn(); | ||
| }; | ||
| } | ||
| })); | ||
| } | ||
|
|
||
| // WRITE | ||
| let mut writer = queue.writer(); | ||
| for i in 0..COUNT { | ||
| writer.push(message::new(i)); | ||
| } | ||
|
|
||
| for join in joins{ | ||
| join.join().unwrap(); | ||
| } | ||
| } | ||
|
|
||
| fn tokio_broadcast(reader_threads: usize){ | ||
| use tokio::sync::broadcast; | ||
| let (tx, _) = broadcast::channel(COUNT); | ||
|
|
||
| let mut joins: ArrayVec<_, 64> = Default::default(); | ||
|
|
||
| // READ | ||
| for _ in 0..reader_threads { | ||
| let mut reader = tx.subscribe(); | ||
| joins.push(std::thread::spawn(move || { | ||
| for _ in 0..COUNT { | ||
| reader.blocking_recv().unwrap(); | ||
| } | ||
| })); | ||
| } | ||
|
|
||
| // WRITE | ||
| let mut writer = tx; | ||
| for i in 0..COUNT { | ||
| writer.send(message::new(i)); | ||
| } | ||
|
|
||
| for join in joins{ | ||
| join.join().unwrap(); | ||
| } | ||
| } | ||
|
|
||
| fn criterion_benchmark(c: &mut Criterion) { | ||
| use criterion::BenchmarkId; | ||
|
|
||
| let mut group = c.benchmark_group("spmc"); | ||
| for reader_threads in [1, 2, 4, 8] { | ||
| let parameter_string = format!("w:1 r:{}", reader_threads); | ||
|
|
||
| group.bench_with_input(BenchmarkId::new("chute::spmc", parameter_string.clone()), &reader_threads | ||
| , |b, rt| b.iter(|| chute_spmc(*rt)) | ||
| ); | ||
|
|
||
| group.bench_with_input(BenchmarkId::new("chute::mpmc", parameter_string.clone()), &reader_threads | ||
| , |b, rt| b.iter(|| chute_mpmc(*rt)) | ||
| ); | ||
|
|
||
| group.bench_with_input(BenchmarkId::new("tokio::broadcast", parameter_string.clone()), &reader_threads | ||
| , |b, rt| b.iter(|| tokio_broadcast(*rt)) | ||
| ); | ||
| } | ||
| group.finish(); | ||
| } | ||
|
|
||
| criterion_group!(benches, criterion_benchmark); | ||
| criterion_main!(benches); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,123 +1,31 @@ | ||
| use std::collections::BTreeMap; | ||
| use std::path::Path; | ||
| use charming::{component::{Grid, Axis}, Chart, ImageRenderer}; | ||
| use charming::element::{AxisLabel, AxisType, Color, Formatter}; | ||
| use charming::element::LabelPosition; | ||
| use charming::element::Label; | ||
| use charming::series::{Bar, Series}; | ||
| use charming::component::{Legend}; | ||
| use charming::component::Title; | ||
| use str_macro::str; | ||
| use std::string::String; | ||
| use charming::element::font_settings::{FontFamily, FontStyle, FontWeight}; | ||
| use crate::{read_group, EstimatesMPMC}; | ||
| use crate::CHART_WIDTH; | ||
| use crate::CHART_THEME; | ||
| use crate::CHART_BACKGROUND; | ||
| use crate::{read_group}; | ||
| use crate::multi_chart::{multi_chart, MultiChartData, Visual}; | ||
|
|
||
| pub fn mpmc(dir_name: impl AsRef<Path>) { | ||
| let rt = 4; | ||
| let wts = [1,2,4,8]; | ||
| let rts = [1,2,4,8]; | ||
| let read = |dir: &str| -> EstimatesMPMC { | ||
| read_group( | ||
| let read = |dir: &str| -> BTreeMap<usize, f64> { | ||
| let data = read_group( | ||
| &std::path::Path::new(dir_name.as_ref()).join(dir) | ||
| ,&wts, &rts | ||
| ) | ||
| ,&wts, &[rt] | ||
| ); | ||
| data.iter().map(|(&wt, readers)| (wt, readers[&rt])).collect() | ||
| }; | ||
|
|
||
| let all: Vec<(String, EstimatesMPMC)> = vec![ | ||
| let all: MultiChartData = vec![ | ||
| (str!("chute::spmc\nw/ mutex"), read("chute__spmc_mutex")), | ||
| (str!("chute::mpmc"), read("chute__mpmc")), | ||
| (str!("tokio::\nbroadcast"), read("tokio__broadcast")), | ||
| ]; | ||
|
|
||
| chart(&all, 4, str!("mpmc (4 readers)"), "out/mpmc"); | ||
| } | ||
|
|
||
| /// `rt` - read thread count | ||
| pub fn chart( | ||
| all_estimates: &Vec<(String, EstimatesMPMC)>, | ||
| rt: usize, | ||
| title: String, | ||
| fname: impl AsRef<Path> | ||
| ) { | ||
| let wts: Vec<usize> = all_estimates.first().unwrap().1 | ||
| .iter().map(|(wt, _)| *wt) | ||
| .collect(); | ||
|
|
||
| let unit = String::from("ms"); | ||
| let ns_to_unit = 1.0 / 1_000_000.0; | ||
|
|
||
| let mut chart = | ||
| Chart::new() | ||
| .background_color(CHART_BACKGROUND) | ||
| .title( | ||
| Title::new() | ||
| .text(title) | ||
| .left("center") | ||
| ) | ||
| .legend( | ||
| Legend::new().top("bottom") | ||
| ) | ||
| .grid( | ||
| Grid::new() | ||
| .left(100) | ||
| .right(40) | ||
| .top(40) | ||
| .bottom(60) | ||
| ) | ||
| .y_axis( | ||
| Axis::new() | ||
| .type_(AxisType::Category) | ||
| .data( | ||
| all_estimates.iter() | ||
| .map(|(name,_)| name.clone()) | ||
| .collect() | ||
| ) | ||
| .axis_label( | ||
| AxisLabel::new().show(true) | ||
| .font_size(13) | ||
| .font_weight(FontWeight::Bolder) | ||
| .color("#666666") | ||
| ) | ||
| ) | ||
| .x_axis( | ||
| Axis::new() | ||
| .type_(AxisType::Value) | ||
| .axis_label(AxisLabel::new().formatter( | ||
| Formatter::String( | ||
| "{value}".to_string() + &unit | ||
| ) | ||
| )) | ||
| ); | ||
|
|
||
| for wt in wts { | ||
| let mut bar = | ||
| Bar::new() | ||
| .name(format!("{wt} writers")) | ||
| .label( | ||
| Label::new() | ||
| .show(true) | ||
| .font_size(11) | ||
| .font_weight(FontWeight::Bold) | ||
| .font_family(FontFamily::MonoSpace) | ||
| .position(LabelPosition::InsideRight) | ||
| .formatter(Formatter::Function( | ||
| ( | ||
| "function (param) { return param.data.toFixed(2); }" | ||
| ).into() | ||
| )) | ||
| ); | ||
| let mut datas = Vec::new(); | ||
| for (_, estimates) in all_estimates { | ||
| let data_ns = estimates[&wt][&rt]; | ||
| datas.push(data_ns * ns_to_unit); | ||
| } | ||
| bar = bar.data(datas); | ||
| chart = chart.series(Series::Bar(bar)); | ||
| } | ||
|
|
||
| let height = all_estimates.len() as u32 * 80 + 100; | ||
| let mut renderer = ImageRenderer::new(CHART_WIDTH, height).theme(CHART_THEME); | ||
| renderer.save(&chart, fname.as_ref().with_extension("svg")).unwrap(); | ||
| renderer.save_format(charming::ImageFormat::Png, &chart, fname.as_ref().with_extension("png")).unwrap(); | ||
| let visual = Visual{ | ||
| title: format!("broadcast mpmc ({rt} readers)"), | ||
| sub_chart_name: str!("writers"), | ||
| label_pos: LabelPosition::Right, | ||
| }; | ||
| multi_chart(&all, "out/mpmc", visual); | ||
| } |
Oops, something went wrong.