feat: Procedural overhaul

Signed-off-by: Jonathan Woollett-Light <[email protected]>

Author: Jonathan Woollett-Light

Cargo.toml

@@ -14,19 +14,9 @@ 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"
+rand = "0.8.5"
 
 [[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

benches/main.rs

@@ -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
@@ -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);