Small optim of frame serialization for puffin_http

puffin/Cargo.toml

@@ -60,6 +60,9 @@ web-time = { version = "0.2", optional = true }
 [dev-dependencies]
 criterion = "0.5"
 
+[target.'cfg(target_os = "linux")'.dev-dependencies]
+memfile = "0.3"
+
 [[bench]]
 name = "benchmark"
 harness = false

puffin/src/frame_data.rs

@@ -569,16 +569,13 @@ impl FrameData {
     pub fn write_into(
         &self,
         scope_collection: Option<&crate::ScopeCollection>,
-        write: &mut impl std::io::Write,
+        mut write: &mut impl std::io::Write,
     ) -> anyhow::Result<()> {
         use bincode::Options as _;
-        use byteorder::{LE, WriteBytesExt as _};
+        use byteorder::WriteBytesExt as _;
 
-        let meta_serialized = bincode::options().serialize(&self.meta)?;
-
-        write.write_all(b"PFD4")?;
-        write.write_all(&(meta_serialized.len() as u32).to_le_bytes())?;
-        write.write_all(&meta_serialized)?;
+        write.write_all(b"PFD5")?;
+        bincode::options().serialize_into(&mut write, &self.meta)?;
 
         self.create_packed();
         let packed_streams_lock = self.data.read();
@@ -588,15 +585,12 @@ impl FrameData {
         write.write_u8(packed_streams.compression_kind as u8)?;
         write.write_all(&packed_streams.bytes)?;
 
-        let to_serialize_scopes: Vec<_> = if let Some(scope_collection) = scope_collection {
-            scope_collection.scopes_by_id().values().cloned().collect()
+        if let Some(scope_collection) = scope_collection {
+            bincode::options().serialize_into(&mut write, &scope_collection.serializable())?;
         } else {
-            self.scope_delta.clone()
-        };
+            bincode::options().serialize_into(write, &self.scope_delta)?;
+        }
 
-        let serialized_scopes = bincode::options().serialize(&to_serialize_scopes)?;
-        write.write_u32::<LE>(serialized_scopes.len() as u32)?;
-        write.write_all(&serialized_scopes)?;
         Ok(())
     }
 
@@ -605,7 +599,7 @@ impl FrameData {
     /// [`None`] is returned if the end of the stream is reached (EOF),
     /// or an end-of-stream sentinel of `0u32` is read.
     #[cfg(feature = "serialization")]
-    pub fn read_next(read: &mut impl std::io::Read) -> anyhow::Result<Option<Self>> {
+    pub fn read_next(mut read: &mut impl std::io::Read) -> anyhow::Result<Option<Self>> {
         use anyhow::Context as _;
         use bincode::Options as _;
         use byteorder::{LE, ReadBytesExt};
@@ -785,6 +779,34 @@ impl FrameData {
                     scope_delta: new_scopes,
                     full_delta: false,
                 }))
+            } else if &header == b"PFD5" {
+                // Added 2024-12-22: remove useless manual sequence size serialization and temporary vector.
+                let meta = {
+                    bincode::options()
+                        .deserialize_from(&mut read)
+                        .context("bincode deserialize")?
+                };
+
+                let streams_compressed_length = read.read_u32::<LE>()? as usize;
+                let compression_kind = CompressionKind::from_u8(read.read_u8()?)?;
+                let streams_compressed = {
+                    let mut streams_compressed = vec![0_u8; streams_compressed_length];
+                    read.read_exact(&mut streams_compressed)?;
+                    PackedStreams::new(compression_kind, streams_compressed)
+                };
+
+                let deserialized_scopes: Vec<Arc<crate::ScopeDetails>> = {
+                    bincode::options()
+                        .deserialize_from(read) // serialized_scopes.as_slice()
+                        .context("Can not deserialize scope details")?
+                };
+
+                Ok(Some(Self {
+                    meta,
+                    data: RwLock::new(FrameDataState::Packed(streams_compressed)),
+                    scope_delta: deserialized_scopes,
+                    full_delta: false,
+                }))
             } else {
                 anyhow::bail!(
                     "Failed to decode: this data is newer than this reader. Please update your puffin version!"

puffin/src/profile_view.rs

@@ -441,3 +441,87 @@ impl FrameStats {
         self.total_ram_used = 0;
     }
 }
+
+#[cfg(all(test, feature = "serialization"))]
+mod tests {
+    use std::{io::Seek, ops::DerefMut, sync::Arc, thread, time::Duration};
+
+    use memfile::MemFile;
+    use parking_lot::Mutex;
+
+    use crate::{GlobalProfiler, profile_scope, set_scopes_on};
+
+    use super::FrameView;
+
+    #[test]
+    fn read_pfd4_file() -> anyhow::Result<()> {
+        let mut file = std::fs::File::open("tests/data/capture_PFD4.puffin")?;
+        let _ = FrameView::read(&mut file)?;
+        Ok(())
+    }
+
+    #[test]
+    fn read_pfd3_file() -> anyhow::Result<()> {
+        let mut file = std::fs::File::open("tests/data/capture_PFD3.puffin")?;
+        let _ = FrameView::read(&mut file)?;
+        Ok(())
+    }
+
+    #[test]
+    fn read_pfd2_file() -> anyhow::Result<()> {
+        let mut file = std::fs::File::open("tests/data/capture_PFD2.puffin")?;
+        let _ = FrameView::read(&mut file)?;
+        Ok(())
+    }
+
+    #[test]
+    fn read_pfd1_file() -> anyhow::Result<()> {
+        let mut file = std::fs::File::open("tests/data/capture_PFD1.puffin")?;
+        let _ = FrameView::read(&mut file)?;
+        Ok(())
+    }
+
+    fn run_write(file: MemFile) {
+        // Init profiler sink with sync wrinting
+        let writer = Arc::new(Mutex::new(file));
+        let sink = GlobalProfiler::lock().add_sink(Box::new(move |frame_data| {
+            let mut writer = writer.lock();
+            frame_data.write_into(None, writer.deref_mut()).unwrap();
+        }));
+
+        set_scopes_on(true); // need this to enable capture
+        // run frames
+        for idx in 0..4 {
+            profile_scope!("main", idx.to_string());
+            {
+                profile_scope!("sleep 1ms");
+                let sleep_duration = Duration::from_millis(1);
+                thread::sleep(sleep_duration);
+            }
+            {
+                profile_scope!("sleep 2ms");
+                let sleep_duration = Duration::from_millis(2);
+                thread::sleep(sleep_duration);
+            }
+            GlobalProfiler::lock().new_frame();
+        }
+
+        set_scopes_on(false);
+        GlobalProfiler::lock().new_frame(); //Force to get last frame
+        GlobalProfiler::lock().remove_sink(sink);
+    }
+
+    fn run_read(mut file: MemFile) {
+        file.rewind().unwrap();
+        let _ = FrameView::read(&mut file).expect("read :");
+    }
+
+    #[test]
+    #[cfg(target_os = "linux")]
+    fn deserialize_serialized() {
+        let file = MemFile::create_default("deserialize_serialized.puffin").unwrap();
+        run_write(file.try_clone().unwrap());
+        thread::sleep(Duration::from_secs(1));
+        run_read(file);
+    }
+}

puffin/src/scope_details.rs

@@ -58,6 +58,28 @@ impl ScopeCollection {
     pub fn scopes_by_id(&self) -> &HashMap<ScopeId, Arc<ScopeDetails>> {
         &self.0.scope_id_to_details
     }
+
+    /// A wrapper than allow `Serialize` all the scopes values of `ScopeCollection`.
+    #[cfg(feature = "serialization")]
+    pub fn serializable(&self) -> Serializable<'_> {
+        Serializable(self)
+    }
+}
+
+/// A wrapper than impl `Serialize` for `ScopeCollection`.
+/// This `struct` is created by the [`serializable`] method on `ScopeCollection`.
+#[cfg(feature = "serialization")]
+pub struct Serializable<'a>(&'a crate::ScopeCollection);
+
+#[cfg(feature = "serialization")]
+impl serde::Serialize for Serializable<'_> {
+    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let iter = self.0.scopes_by_id().values();
+        serializer.collect_seq(iter)
+    }
 }
 
 /// Scopes are identified by user-provided name while functions are identified by the function name.

puffin/tests/common/mod.rs

@@ -0,0 +1,111 @@
+#![allow(dead_code)]
+
+use std::{
+    io::Write,
+    sync::Arc,
+    thread::{self, JoinHandle},
+    time::Duration,
+};
+
+use parking_lot::Mutex;
+#[cfg(feature = "serialization")]
+use puffin::FrameData;
+use puffin::{FrameSinkId, GlobalProfiler};
+
+pub fn process_1() {
+    puffin::profile_function!();
+    sub_process_1_1();
+    (0..2).for_each(|_| sub_process_1_2());
+}
+
+fn sub_process_1_1() {
+    puffin::profile_function!();
+    thread::sleep(Duration::from_millis(1));
+}
+
+fn sub_process_1_2() {
+    puffin::profile_function!();
+    thread::sleep(Duration::from_micros(2));
+}
+
+pub fn example_run() {
+    for idx in 0..4 {
+        puffin::profile_scope!("main", idx.to_string());
+
+        {
+            puffin::profile_scope!("sleep 1ms");
+            let sleep_duration = Duration::from_millis(1);
+            thread::sleep(sleep_duration);
+        }
+
+        {
+            puffin::profile_scope!("sleep 2ms");
+            let sleep_duration = Duration::from_millis(2);
+            thread::sleep(sleep_duration);
+        }
+        //println!("before new_frame {idx}");
+        puffin::GlobalProfiler::lock().new_frame();
+        //println!("after new_frame {idx}");
+    }
+}
+
+pub struct FrameWriterImpl<W: Write> {
+    writer: Arc<Mutex<W>>,
+}
+
+impl<W: Write> FrameWriterImpl<W> {
+    pub fn from_writer(mut writer: W) -> Self {
+        writer.write_all(b"PUF0").unwrap(); //Hack: should not be duplicated
+        Self {
+            writer: Arc::new(Mutex::new(writer)),
+        }
+    }
+
+    #[cfg(feature = "serialization")]
+    fn write_frame(&self, frame_data: Arc<FrameData>) {
+        use std::ops::DerefMut;
+
+        let mut writer = self.writer.lock();
+        frame_data.write_into(None, writer.deref_mut()).unwrap();
+    }
+}
+
+pub struct FrameWriterSink {
+    sink_id: FrameSinkId,
+    write_thread: Option<JoinHandle<()>>,
+}
+impl Drop for FrameWriterSink {
+    fn drop(&mut self) {
+        GlobalProfiler::lock().remove_sink(self.sink_id);
+        if let Some(write_handle) = self.write_thread.take() {
+            let _ = write_handle.join();
+        }
+    }
+}
+
+#[cfg(feature = "serialization")]
+#[must_use]
+pub fn init_frames_writer(writer: impl Write + Send + 'static) -> FrameWriterSink {
+    use std::sync::mpsc;
+
+    let frame_writer = FrameWriterImpl::from_writer(writer);
+    let (frame_sender, frames_recv) = mpsc::channel();
+
+    let write_thread = thread::Builder::new()
+        .name("frame_writer".into())
+        .spawn(move || {
+            while let Ok(frame_data) = frames_recv.recv() {
+                frame_writer.write_frame(frame_data);
+            }
+        })
+        .unwrap();
+
+    // Init profiler sink and enable capture
+    let sink_id = GlobalProfiler::lock().add_sink(Box::new(move |frame_data| {
+        frame_sender.send(frame_data).unwrap()
+    }));
+    FrameWriterSink {
+        sink_id,
+        write_thread: Some(write_thread),
+    }
+}

puffin/tests/frame_serialization.rs

@@ -0,0 +1,17 @@
+mod common;
+
+#[cfg(feature = "serialization")]
+#[test]
+fn frame_serialization() {
+    let frame_data = Vec::new();
+    let _frame_writer = common::init_frames_writer(frame_data);
+
+    //println!("set_scopes_on(true)");
+    puffin::set_scopes_on(true); // need this to enable capture
+
+    common::example_run();
+
+    //println!("set_scopes_on(false)");
+    puffin::set_scopes_on(false);
+    puffin::GlobalProfiler::lock().new_frame(); //Force to get last frame
+}