Implement a second mmap-based serialization sink that is backed directly by a file.

measureme/Cargo.toml

@@ -13,3 +13,6 @@ travis-ci = { repository = "rust-lang/measureme" }
 byteorder = "1.3"
 rustc-hash = "1.0.1"
 memmap = "0.7.0"
+
+[target.'cfg(unix)'.dependencies]
+libc = "0.2.50"

measureme/benches/serialization_bench.rs

@@ -6,6 +6,8 @@ use measureme::{
     FileSerializationSink, MmapSerializationSink, testing_common
 };
 
+#[cfg(unix)] use measureme::AsyncMmapSerializationSink;
+
 #[bench]
 fn bench_file_serialization_sink(bencher: &mut test::Bencher) {
     bencher.iter(|| {
@@ -19,3 +21,11 @@ fn bench_mmap_serialization_sink(bencher: &mut test::Bencher) {
         testing_common::run_end_to_end_serialization_test::<MmapSerializationSink>("mmap_serialization_sink_test");
     });
 }
+
+#[cfg(unix)]
+#[bench]
+fn bench_async_mmap_serialization_sink(bencher: &mut test::Bencher) {
+    bencher.iter(|| {
+        testing_common::run_end_to_end_serialization_test::<AsyncMmapSerializationSink>("async_mmap_serialization_sink_test");
+    });
+}

measureme/src/async_mmap_serialization_sink.rs

@@ -0,0 +1,108 @@
+use crate::serialization::{Addr, SerializationSink};
+use std::fs::{File, OpenOptions};
+use std::path::{Path};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::os::unix::io::AsRawFd;
+use std::io;
+
+/// Implements a `SerializationSink` that uses a file-backed mmap.
+pub struct AsyncMmapSerializationSink {
+    file: File,
+    current_pos: AtomicUsize,
+    mapping_start: *mut u8,
+    mapping_len: usize,
+}
+
+impl SerializationSink for AsyncMmapSerializationSink {
+    fn from_path(path: &Path) -> Self {
+
+        // Lazily allocate 1 GB
+        let file_size = 1 << 30;
+
+        let file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .truncate(true)
+            .open(path)
+            .unwrap();
+
+        if let Err(e) = file.set_len(file_size as u64) {
+            panic!("Error setting file length: {:?}", e);
+        }
+
+        //
+        let ptr: *mut libc::c_void = unsafe {
+            match libc::mmap(0 as *mut _, file_size, libc::PROT_WRITE, libc::MAP_SHARED, file.as_raw_fd(), 0) {
+                libc::MAP_FAILED => {
+                    panic!("Error creating mmap: {:?}", io::Error::last_os_error())
+                }
+                other => other,
+            }
+        };
+
+        // Hint to the OS that it can write old pages to disk once they are
+        // fully written.
+        unsafe {
+            if libc::madvise(ptr, file_size as _, libc::MADV_SEQUENTIAL) != 0 {
+                eprintln!("Error during `madvise`: {:?}", io::Error::last_os_error());
+            }
+        }
+
+        AsyncMmapSerializationSink {
+            file,
+            current_pos: AtomicUsize::new(0),
+            mapping_start: ptr as *mut u8,
+            mapping_len: file_size as usize,
+        }
+    }
+
+    #[inline]
+    fn write_atomic<W>(&self, num_bytes: usize, write: W) -> Addr
+    where
+        W: FnOnce(&mut [u8]),
+    {
+        // Reserve the range of bytes we'll copy to
+        let pos = self.current_pos.fetch_add(num_bytes, Ordering::SeqCst);
+
+        // Bounds checks
+        assert!(pos.checked_add(num_bytes).unwrap() <= self.mapping_len);
+
+        let bytes: &mut [u8] = unsafe {
+            let start: *mut u8 = self.mapping_start.offset(pos as isize);
+            std::slice::from_raw_parts_mut(start, num_bytes)
+        };
+
+        write(bytes);
+
+        Addr(pos as u32)
+    }
+}
+
+impl Drop for AsyncMmapSerializationSink {
+    fn drop(&mut self) {
+        let actual_size = *self.current_pos.get_mut();
+
+        unsafe {
+            // First use `mremap` to shrink the memory map. Otherwise `munmap`
+            // would write everything to the backing file, including the
+            // memory we never touched.
+            let new_addr = libc::mremap(self.mapping_start as *mut _,
+                         self.mapping_len as _,
+                         actual_size as _,
+                         0);
+
+            if new_addr == libc::MAP_FAILED {
+                eprintln!("mremap failed: {:?}", io::Error::last_os_error())
+            }
+
+            if libc::munmap(new_addr, actual_size as _) != 0 {
+                eprintln!("munmap failed: {:?}", io::Error::last_os_error())
+            }
+        }
+
+        if let Err(e) = self.file.set_len(actual_size as u64) {
+            eprintln!("Error setting file length: {:?}", e);
+        }
+    }
+}

measureme/src/lib.rs

@@ -7,6 +7,9 @@ mod raw_event;
 mod serialization;
 mod stringtable;
 
+#[cfg(unix)]
+mod async_mmap_serialization_sink;
+
 pub mod testing_common;
 
 pub use crate::event::Event;
@@ -19,3 +22,6 @@ pub use crate::serialization::{Addr, SerializationSink};
 pub use crate::stringtable::{
     SerializableString, StringId, StringRef, StringTable, StringTableBuilder,
 };
+
+#[cfg(unix)]
+pub use crate::async_mmap_serialization_sink::AsyncMmapSerializationSink;

measureme/tests/serialization.rs

@@ -1,4 +1,5 @@
 
+#[cfg(unix)] use measureme::AsyncMmapSerializationSink;
 use measureme::{FileSerializationSink, MmapSerializationSink};
 use measureme::testing_common::run_end_to_end_serialization_test;
 
@@ -11,3 +12,9 @@ fn test_file_serialization_sink() {
 fn test_mmap_serialization_sink() {
     run_end_to_end_serialization_test::<MmapSerializationSink>("mmap_serialization_sink_test");
 }
+
+#[cfg(unix)]
+#[test]
+fn test_unix_mmap_serialization_sink() {
+    run_end_to_end_serialization_test::<AsyncMmapSerializationSink>("async_mmap_serialization_sink_test");
+}