Improve SerializationSink benchmarks and tests.

Author: michaelwoerister

analyzeme/benches/serialization_bench.rs

@@ -8,17 +8,35 @@ use measureme::{FileSerializationSink, MmapSerializationSink};
 #[bench]
 fn bench_file_serialization_sink(bencher: &mut test::Bencher) {
     bencher.iter(|| {
-        testing_common::run_end_to_end_serialization_test::<FileSerializationSink>(
-            "file_serialization_sink_test",
+        testing_common::run_serialization_bench::<FileSerializationSink>(
+            "file_serialization_sink_test", 500_000, 1
         );
     });
 }
 
 #[bench]
 fn bench_mmap_serialization_sink(bencher: &mut test::Bencher) {
     bencher.iter(|| {
-        testing_common::run_end_to_end_serialization_test::<MmapSerializationSink>(
-            "mmap_serialization_sink_test",
+        testing_common::run_serialization_bench::<MmapSerializationSink>(
+            "mmap_serialization_sink_test", 500_000, 1
+        );
+    });
+}
+
+#[bench]
+fn bench_file_serialization_sink_8_threads(bencher: &mut test::Bencher) {
+    bencher.iter(|| {
+        testing_common::run_serialization_bench::<FileSerializationSink>(
+            "file_serialization_sink_test", 50_000, 8
+        );
+    });
+}
+
+#[bench]
+fn bench_mmap_serialization_sink_8_threads(bencher: &mut test::Bencher) {
+    bencher.iter(|| {
+        testing_common::run_serialization_bench::<MmapSerializationSink>(
+            "mmap_serialization_sink_test", 50_000, 8
         );
     });
 }

analyzeme/src/testing_common.rs

@@ -18,12 +18,16 @@ fn mk_filestem(file_name_stem: &str) -> PathBuf {
 }
 
 // Generate some profiling data. This is the part that would run in rustc.
-fn generate_profiling_data<S: SerializationSink>(filestem: &Path) -> Vec<Event<'static>> {
+fn generate_profiling_data<S: SerializationSink>(
+    filestem: &Path,
+    num_stacks: usize,
+    num_threads: usize,
+) -> Vec<Event<'static>> {
     let profiler = Arc::new(Profiler::<S>::new(Path::new(filestem)).unwrap());
 
     let event_id_reserved = StringId::reserved(42);
 
-    let event_ids = &[
+    let event_ids = vec![
         (
             profiler.alloc_string("Generic"),
             profiler.alloc_string("SomeGenericActivity"),
@@ -38,20 +42,33 @@ fn generate_profiling_data<S: SerializationSink>(filestem: &Path) -> Vec<Event<'
     event_ids_as_str.insert(event_ids[1].0, "Query");
     event_ids_as_str.insert(event_ids[1].1, "SomeQuery");
 
-    let mut expected_events = Vec::new();
+    let threads: Vec<_> = (0.. num_threads).map(|thread_id| {
+        let event_ids = event_ids.clone();
+        let profiler = profiler.clone();
+        let event_ids_as_str = event_ids_as_str.clone();
 
-    for i in 0..10_000 {
-        // Allocate some invocation stacks
+        std::thread::spawn(move || {
+            let mut expected_events = Vec::new();
 
-        pseudo_invocation(
-            &profiler,
-            i,
-            4,
-            event_ids,
-            &event_ids_as_str,
-            &mut expected_events,
-        );
-    }
+            for i in 0..num_stacks {
+                // Allocate some invocation stacks
+
+                pseudo_invocation(
+                    &profiler,
+                    i,
+                    thread_id as u64,
+                    4,
+                    &event_ids[..],
+                    &event_ids_as_str,
+                    &mut expected_events,
+                );
+            }
+
+            expected_events
+        })
+    }).collect();
+
+    let expected_events: Vec<_> = threads.into_iter().flat_map(|t| t.join().unwrap()).collect();
 
     // An example of allocating the string contents of an event id that has
     // already been used
@@ -84,34 +101,65 @@ fn check_profiling_data(
 ) {
     let mut count = 0;
 
+    // This assertion makes sure that the ExactSizeIterator impl works as expected.
     assert_eq!(
         (num_expected_events, Some(num_expected_events)),
         actual_events.size_hint()
     );
 
-    for (actual_event, expected_event) in actual_events.zip(expected_events) {
-        assert_eq!(actual_event.event_kind, expected_event.event_kind);
-        assert_eq!(actual_event.label, expected_event.label);
-        assert_eq!(actual_event.additional_data, expected_event.additional_data);
-        assert_eq!(
-            actual_event.timestamp.is_instant(),
-            expected_event.timestamp.is_instant()
-        );
+    let actual_events_per_thread = collect_events_per_thread(actual_events);
+    let expected_events_per_thread = collect_events_per_thread(expected_events);
+
+    let thread_ids: Vec<_> = actual_events_per_thread.keys().collect();
+    assert_eq!(thread_ids, expected_events_per_thread.keys().collect::<Vec<_>>());
 
-        count += 1;
+    for thread_id in thread_ids {
+        let actual_events = &actual_events_per_thread[thread_id];
+        let expected_events = &expected_events_per_thread[thread_id];
+
+        assert_eq!(actual_events.len(), expected_events.len());
+
+        for (actual_event, expected_event) in actual_events.iter().zip(expected_events.iter()) {
+            assert_eq!(actual_event.event_kind, expected_event.event_kind);
+            assert_eq!(actual_event.label, expected_event.label);
+            assert_eq!(actual_event.additional_data, expected_event.additional_data);
+            assert_eq!(
+                actual_event.timestamp.is_instant(),
+                expected_event.timestamp.is_instant()
+            );
+
+            count += 1;
+        }
     }
+
     assert_eq!(count, num_expected_events);
 }
 
-pub fn run_end_to_end_serialization_test<S: SerializationSink>(file_name_stem: &str) {
+fn collect_events_per_thread<'a>(events: &mut dyn Iterator<Item = Event<'a>>) -> FxHashMap<u64, Vec<Event<'a>>> {
+    let mut per_thread: FxHashMap<_, _> = Default::default();
+
+    for event in events {
+        per_thread.entry(event.thread_id).or_insert(Vec::new()).push(event);
+    }
+
+    per_thread
+}
+
+pub fn run_serialization_bench<S: SerializationSink>(file_name_stem: &str, num_events: usize, num_threads: usize) {
     let filestem = mk_filestem(file_name_stem);
-    let expected_events = generate_profiling_data::<S>(&filestem);
+    generate_profiling_data::<S>(&filestem, num_events, num_threads);
+}
+
+pub fn run_end_to_end_serialization_test<S: SerializationSink>(file_name_stem: &str, num_threads: usize) {
+    let filestem = mk_filestem(file_name_stem);
+    let expected_events = generate_profiling_data::<S>(&filestem, 10_000, num_threads);
     process_profiling_data(&filestem, &expected_events);
 }
 
 fn pseudo_invocation<S: SerializationSink>(
     profiler: &Profiler<S>,
     random: usize,
+    thread_id: u64,
     recursions_left: usize,
     event_ids: &[(StringId, StringId)],
     event_ids_as_str: &FxHashMap<StringId, &'static str>,
@@ -121,15 +169,14 @@ fn pseudo_invocation<S: SerializationSink>(
         return;
     }
 
-    let thread_id = (random % 3) as u64;
-
     let (event_kind, event_id) = event_ids[random % event_ids.len()];
 
     let _prof_guard = profiler.start_recording_interval_event(event_kind, event_id, thread_id);
 
     pseudo_invocation(
         profiler,
         random,
+        thread_id,
         recursions_left - 1,
         event_ids,
         event_ids_as_str,

analyzeme/tests/serialization.rs

@@ -2,11 +2,21 @@ use analyzeme::testing_common::run_end_to_end_serialization_test;
 use measureme::{FileSerializationSink, MmapSerializationSink};
 
 #[test]
-fn test_file_serialization_sink() {
-    run_end_to_end_serialization_test::<FileSerializationSink>("file_serialization_sink_test");
+fn test_file_serialization_sink_1_thread() {
+    run_end_to_end_serialization_test::<FileSerializationSink>("file_serialization_sink_test_1_thread", 1);
 }
 
 #[test]
-fn test_mmap_serialization_sink() {
-    run_end_to_end_serialization_test::<MmapSerializationSink>("mmap_serialization_sink_test");
+fn test_file_serialization_sink_8_threads() {
+    run_end_to_end_serialization_test::<FileSerializationSink>("file_serialization_sink_test_8_threads", 8);
+}
+
+#[test]
+fn test_mmap_serialization_sink_1_thread() {
+    run_end_to_end_serialization_test::<MmapSerializationSink>("mmap_serialization_sink_test_1_thread", 1);
+}
+
+#[test]
+fn test_mmap_serialization_sink_8_threads() {
+    run_end_to_end_serialization_test::<MmapSerializationSink>("mmap_serialization_sink_test_8_threads", 8);
 }

measureme/src/serialization.rs

@@ -11,7 +11,7 @@ impl Addr {
     }
 }
 
-pub trait SerializationSink: Sized {
+pub trait SerializationSink: Sized + Send + Sync + 'static {
     fn from_path(path: &Path) -> Result<Self, Box<dyn Error>>;
 
     fn write_atomic<W>(&self, num_bytes: usize, write: W) -> Addr