Skip to content

Commit

Permalink
feat: Procedural overhaul
Browse files Browse the repository at this point in the history 
Signed-off-by: Jonathan Woollett-Light <[email protected]>
  • Loading branch information
Jonathan Woollett-Light committed May 30, 2023
1 parent 4a9ba12 commit a778171
Show file tree
Hide file tree
Showing 19 changed files with 503 additions and 2,497 deletions.
15 changes: 2 additions & 13 deletions Cargo.toml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,19 +14,8 @@ vmm-sys-util = "0.11.0"
libc = "0.2.39"

[dev-dependencies]
criterion = "0.3.5"

[features]
remote_endpoint = []
test_utilities = []
criterion = "0.5.1"

[[bench]]
name = "main"
harness = false

[lib]
bench = false # https://bheisler.github.io/criterion.rs/book/faq.html#cargo-bench-gives-unrecognized-option-errors-for-valid-command-line-options

[profile.bench]
lto = true
codegen-units = 1
harness = false
299 changes: 235 additions & 64 deletions benches/main.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,71 +3,154 @@

use criterion::{criterion_group, criterion_main, Criterion};

use event_manager::utilities::subscribers::{
CounterInnerMutSubscriber, CounterSubscriber, CounterSubscriberWithData,
};
use event_manager::{EventManager, EventSubscriber, MutEventSubscriber, SubscriberOps};
use event_manager::{BufferedEventManager, EventManager};
use std::os::fd::AsFd;
use std::os::fd::FromRawFd;
use std::os::fd::OwnedFd;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use vmm_sys_util::epoll::EventSet;

// Test the performance of event manager when it manages a single subscriber type.
// The performance is assessed under stress, all added subscribers have active events.
fn run_basic_subscriber(c: &mut Criterion) {
let no_of_subscribers = 200;
let no_of_subscribers = 200i32;

let mut event_manager = EventManager::<CounterSubscriber>::new().unwrap();
for _ in 0..no_of_subscribers {
let mut counter_subscriber = CounterSubscriber::default();
counter_subscriber.trigger_event();
event_manager.add_subscriber(counter_subscriber);
}
let mut event_manager =
BufferedEventManager::with_capacity(false, no_of_subscribers as usize).unwrap();

let subscribers = (0..no_of_subscribers).map(|_| {
// Create an eventfd that is initialized with 1 waiting event.
let event_fd = unsafe {
let raw_fd = libc::eventfd(1,0);
assert_ne!(raw_fd, -1);
OwnedFd::from_raw_fd(raw_fd)
};

event_manager.add(event_fd.as_fd(),EventSet::IN | EventSet::ERROR | EventSet::HANG_UP,Box::new(move |_:&mut EventManager, event_set: EventSet| {
match event_set {
EventSet::IN => (),
EventSet::ERROR => {
eprintln!("Got error on the monitored event.");
},
EventSet::HANG_UP => {
// TODO Do this: https://github.com/rust-vmm/event-manager/blob/main/src/utilities/subscribers.rs#L116:L118
panic!("Cannot continue execution. Associated fd was closed.");
},
_ => {
eprintln!("Received spurious event from the event manager {event_set:#?}.");
}
}
})).unwrap();

event_fd
}).collect::<Vec<_>>();

c.bench_function("process_basic", |b| {
b.iter(|| {
let ev_count = event_manager.run().unwrap();
assert_eq!(ev_count, no_of_subscribers)
assert_eq!(event_manager.wait(Some(0)), Ok(no_of_subscribers));
})
});

drop(subscribers);
}

// Test the performance of event manager when the subscribers are wrapped in an Arc<Mutex>.
// The performance is assessed under stress, all added subscribers have active events.
fn run_arc_mutex_subscriber(c: &mut Criterion) {
let no_of_subscribers = 200;
let no_of_subscribers = 200i32;

let mut event_manager =
BufferedEventManager::with_capacity(false, no_of_subscribers as usize).unwrap();

let subscribers = (0..no_of_subscribers).map(|_| {
// Create an eventfd that is initialized with 1 waiting event.
let event_fd = unsafe {
let raw_fd = libc::eventfd(1,0);
assert_ne!(raw_fd, -1);
OwnedFd::from_raw_fd(raw_fd)
};
let counter = Arc::new(Mutex::new(0u64));
let counter_clone = counter.clone();

event_manager.add(event_fd.as_fd(),EventSet::IN | EventSet::ERROR | EventSet::HANG_UP,Box::new(move |_:&mut EventManager, event_set: EventSet| {
match event_set {
EventSet::IN => {
*counter_clone.lock().unwrap() += 1;
},
EventSet::ERROR => {
eprintln!("Got error on the monitored event.");
},
EventSet::HANG_UP => {
// TODO Do this: https://github.com/rust-vmm/event-manager/blob/main/src/utilities/subscribers.rs#L116:L118
panic!("Cannot continue execution. Associated fd was closed.");
},
_ => {
eprintln!("Received spurious event from the event manager {event_set:#?}.");
}
}
})).unwrap();

let mut event_manager = EventManager::<Arc<Mutex<CounterSubscriber>>>::new().unwrap();
for _ in 0..no_of_subscribers {
let counter_subscriber = Arc::new(Mutex::new(CounterSubscriber::default()));
counter_subscriber.lock().unwrap().trigger_event();
event_manager.add_subscriber(counter_subscriber);
}
(event_fd,counter)
}).collect::<Vec<_>>();

c.bench_function("process_with_arc_mutex", |b| {
b.iter(|| {
let ev_count = event_manager.run().unwrap();
assert_eq!(ev_count, no_of_subscribers)
assert_eq!(event_manager.wait(Some(0)), Ok(no_of_subscribers));
})
});

drop(subscribers);
}

// Test the performance of event manager when the subscribers are wrapped in an Arc, and they
// leverage inner mutability to update their internal state.
// The performance is assessed under stress, all added subscribers have active events.
fn run_subscriber_with_inner_mut(c: &mut Criterion) {
let no_of_subscribers = 200;
let no_of_subscribers = 200i32;

let mut event_manager =
BufferedEventManager::with_capacity(false, no_of_subscribers as usize).unwrap();

let subscribers = (0..no_of_subscribers).map(|_| {
// Create an eventfd that is initialized with 1 waiting event.
let event_fd = unsafe {
let raw_fd = libc::eventfd(1,0);
assert_ne!(raw_fd, -1);
OwnedFd::from_raw_fd(raw_fd)
};
let counter = Arc::new(AtomicU64::new(0));
let counter_clone = counter.clone();

event_manager.add(event_fd.as_fd(),EventSet::IN | EventSet::ERROR | EventSet::HANG_UP,Box::new(move |_:&mut EventManager, event_set: EventSet| {
match event_set {
EventSet::IN => {
counter_clone.fetch_add(1, Ordering::SeqCst);
},
EventSet::ERROR => {
eprintln!("Got error on the monitored event.");
},
EventSet::HANG_UP => {
// TODO Do this: https://github.com/rust-vmm/event-manager/blob/main/src/utilities/subscribers.rs#L116:L118
panic!("Cannot continue execution. Associated fd was closed.");
},
_ => {
eprintln!("Received spurious event from the event manager {event_set:#?}.");
}
}
})).unwrap();

let mut event_manager = EventManager::<Arc<dyn EventSubscriber + Send + Sync>>::new().unwrap();
for _ in 0..no_of_subscribers {
let counter_subscriber = CounterInnerMutSubscriber::default();
counter_subscriber.trigger_event();
event_manager.add_subscriber(Arc::new(counter_subscriber));
}
(event_fd,counter)
}).collect::<Vec<_>>();

c.bench_function("process_with_inner_mut", |b| {
b.iter(|| {
let ev_count = event_manager.run().unwrap();
assert_eq!(ev_count, no_of_subscribers)
assert_eq!(event_manager.wait(Some(0)), Ok(no_of_subscribers));
})
});

drop(subscribers);
}

// Test the performance of event manager when it manages subscribers of different types, that are
Expand All @@ -76,63 +159,151 @@ fn run_subscriber_with_inner_mut(c: &mut Criterion) {
// The performance is assessed under stress, all added subscribers have active events, and the
// CounterSubscriberWithData subscribers have multiple active events.
fn run_multiple_subscriber_types(c: &mut Criterion) {
let no_of_subscribers = 100;
let no_of_subscribers = 100i32;

let total = no_of_subscribers + (no_of_subscribers * i32::try_from(EVENTS).unwrap());

let mut event_manager =
BufferedEventManager::with_capacity(false, usize::try_from(total).unwrap()).unwrap();

let mut event_manager = EventManager::<Arc<Mutex<dyn MutEventSubscriber>>>::new()
.expect("Cannot create event manager.");
let subscribers = (0..no_of_subscribers).map(|_| {
// Create an eventfd that is initialized with 1 waiting event.
let event_fd = unsafe {
let raw_fd = libc::eventfd(1,0);
assert_ne!(raw_fd, -1);
OwnedFd::from_raw_fd(raw_fd)
};
let counter = Arc::new(AtomicU64::new(0));
let counter_clone = counter.clone();

for i in 0..no_of_subscribers {
// The `CounterSubscriberWithData` expects to receive a number as a parameter that
// represents the number it can use as its inner Events data.
let mut data_subscriber = CounterSubscriberWithData::new(i * no_of_subscribers);
data_subscriber.trigger_all_counters();
event_manager.add_subscriber(Arc::new(Mutex::new(data_subscriber)));
event_manager.add(event_fd.as_fd(),EventSet::IN | EventSet::ERROR | EventSet::HANG_UP,Box::new(move |_:&mut EventManager, event_set: EventSet| {
match event_set {
EventSet::IN => {
counter_clone.fetch_add(1, Ordering::SeqCst);
},
EventSet::ERROR => {
eprintln!("Got error on the monitored event.");
},
EventSet::HANG_UP => {
// TODO Do this: https://github.com/rust-vmm/event-manager/blob/main/src/utilities/subscribers.rs#L116:L118
panic!("Cannot continue execution. Associated fd was closed.");
},
_ => {
eprintln!("Received spurious event from the event manager {event_set:#?}.");
}
}
})).unwrap();

let mut counter_subscriber = CounterSubscriber::default();
counter_subscriber.trigger_event();
event_manager.add_subscriber(Arc::new(Mutex::new(counter_subscriber)));
}
(event_fd,counter)
}).collect::<Vec<_>>();

const EVENTS: usize = 3;

let subscribers_with_data = (0..no_of_subscribers)
.map(|_| {
let data = Arc::new([AtomicU64::new(0), AtomicU64::new(0), AtomicU64::new(0)]);
assert_eq!(data.len(), EVENTS);

// Create eventfd's that are initialized with 1 waiting event.
let inner_subscribers = (0..EVENTS)
.map(|_| unsafe {
let raw_fd = libc::eventfd(1, 0);
assert_ne!(raw_fd, -1);
OwnedFd::from_raw_fd(raw_fd)
})
.collect::<Vec<_>>();

for i in 0..EVENTS {
let data_clone = data.clone();

event_manager
.add(
inner_subscribers[i].as_fd(),
EventSet::IN | EventSet::ERROR | EventSet::HANG_UP,
Box::new(move |_: &mut EventManager, event_set: EventSet| {
match event_set {
EventSet::IN => {
data_clone[i].fetch_add(1, Ordering::SeqCst);
}
EventSet::ERROR => {
eprintln!("Got error on the monitored event.");
}
EventSet::HANG_UP => {
// TODO Do this: https://github.com/rust-vmm/event-manager/blob/main/src/utilities/subscribers.rs#L116:L118
panic!("Cannot continue execution. Associated fd was closed.");
}
_ => {}
}
}),
)
.unwrap();
}

(inner_subscribers, data)
})
.collect::<Vec<_>>();

c.bench_function("process_dynamic_dispatch", |b| {
b.iter(|| {
let _ = event_manager.run().unwrap();
assert_eq!(event_manager.wait(Some(0)), Ok(total));
})
});

drop(subscribers);
drop(subscribers_with_data);
}

// Test the performance of event manager when it manages a single subscriber type.
// Just a few of the events are active in this test scenario.
fn run_with_few_active_events(c: &mut Criterion) {
let no_of_subscribers = 200;
let no_of_subscribers = 200i32;
let active = 1 + no_of_subscribers / 23;

let mut event_manager =
BufferedEventManager::with_capacity(false, no_of_subscribers as usize).unwrap();

let mut event_manager = EventManager::<CounterSubscriber>::new().unwrap();
let subscribers = (0..no_of_subscribers).map(|i| {
// Create an eventfd that is initialized with 1 waiting event.
let event_fd = unsafe {
let raw_fd = libc::eventfd((i % 23 == 0) as u8 as u32,0);
assert_ne!(raw_fd, -1);
OwnedFd::from_raw_fd(raw_fd)
};

for i in 0..no_of_subscribers {
let mut counter_subscriber = CounterSubscriber::default();
// Let's activate the events for a few subscribers (i.e. only the ones that are
// divisible by 23). 23 is a random number that I just happen to like.
if i % 23 == 0 {
counter_subscriber.trigger_event();
}
event_manager.add_subscriber(counter_subscriber);
}
event_manager.add(event_fd.as_fd(),EventSet::IN | EventSet::ERROR | EventSet::HANG_UP,Box::new(move |_:&mut EventManager, event_set: EventSet| {
match event_set {
EventSet::IN => (),
EventSet::ERROR => {
eprintln!("Got error on the monitored event.");
},
EventSet::HANG_UP => {
// TODO Do this: https://github.com/rust-vmm/event-manager/blob/main/src/utilities/subscribers.rs#L116:L118
panic!("Cannot continue execution. Associated fd was closed.");
},
_ => {
eprintln!("Received spurious event from the event manager {event_set:#?}.");
}
}
})).unwrap();

event_fd
}).collect::<Vec<_>>();

c.bench_function("process_dispatch_few_events", |b| {
b.iter(|| {
let _ = event_manager.run().unwrap();
assert_eq!(event_manager.wait(Some(0)), Ok(active));
})
});

drop(subscribers);
}

criterion_group! {
criterion_group!(
name = benches;
config = Criterion::default()
.sample_size(200)
.measurement_time(std::time::Duration::from_secs(40));
targets = run_basic_subscriber, run_arc_mutex_subscriber, run_subscriber_with_inner_mut,
run_multiple_subscriber_types, run_with_few_active_events
}

criterion_main! {
benches
}
run_multiple_subscriber_types, run_with_few_active_events
);
criterion_main!(benches);
Loading

0 comments on commit a778171

Please sign in to comment.