diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d12a7ff7..f693c8a2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -2,7 +2,7 @@ on: push: branches: - - main + - "*" tags: - "*" pull_request: @@ -82,28 +82,29 @@ jobs: - name: cargo doc run: cargo doc --locked -p puffin -p puffin_egui -p puffin_http -p --lib --no-deps --all-features - cargo-vet: - name: Vet Dependencies - runs-on: ubuntu-latest - env: - CARGO_VET_VERSION: 0.9.1 - steps: - - uses: actions/checkout@v4 - - uses: dtolnay/rust-toolchain@stable - - uses: actions/cache@v3 - with: - path: ${{ runner.tool_cache }}/cargo-vet - key: cargo-vet-bin-${{ env.CARGO_VET_VERSION }} - - name: Add the tool cache directory to the search path - run: echo "${{ runner.tool_cache }}/cargo-vet/bin" >> $GITHUB_PATH - - name: Ensure that the tool cache is populated with the cargo-vet binary - # build from source, as are not published binaries yet :( - # tracked in https://github.com/mozilla/cargo-vet/issues/484 - run: cargo +stable install --root ${{ runner.tool_cache }}/cargo-vet --version ${{ env.CARGO_VET_VERSION }} cargo-vet - - name: Invoke cargo-vet - run: | - cargo vet --locked - cargo vet --locked >> $GITHUB_STEP_SUMMARY + # cargo-vet: + # name: Vet Dependencies + # runs-on: ubuntu-latest + # env: + # CARGO_VET_VERSION: 0.9.1 + # steps: + # - uses: actions/checkout@v4 + # - uses: dtolnay/rust-toolchain@stable + # - uses: actions/cache@v3 + # with: + # path: ${{ runner.tool_cache }}/cargo-vet + # key: cargo-vet-bin-${{ env.CARGO_VET_VERSION }} + # - name: Add the tool cache directory to the search path + # run: echo "${{ runner.tool_cache }}/cargo-vet/bin" >> $GITHUB_PATH + # - name: Ensure that the tool cache is populated with the cargo-vet binary + # # build from source, as are not published binaries yet :( + # # tracked in https://github.com/mozilla/cargo-vet/issues/484 + # # TODO: cargo-vet now have prebuild binaries https://github.com/mozilla/cargo-vet/releases/tag/v0.10.0 + # run: cargo +stable install --root ${{ runner.tool_cache }}/cargo-vet --version ${{ env.CARGO_VET_VERSION }} cargo-vet + # - name: Invoke cargo-vet + # run: | + # cargo vet --locked + # cargo vet --locked >> $GITHUB_STEP_SUMMARY taplo: name: Toml format check @@ -112,4 +113,4 @@ jobs: - uses: actions/checkout@v4 - uses: gwen-lg/taplo-action@v1 with: - format: true \ No newline at end of file + format: true diff --git a/puffin/src/data_header.rs b/puffin/src/data_header.rs new file mode 100644 index 00000000..45b59c51 --- /dev/null +++ b/puffin/src/data_header.rs @@ -0,0 +1,50 @@ +use std::{fmt::Display, str::from_utf8}; + +/// Header of serialized data. +/// +/// Used to differentiate data of `ScopeCollection` and `FrameData` by example. +#[derive(Debug, Clone, Copy)] +pub struct DataHeader([u8; 4]); + +impl DataHeader { + /// Tried to read header from reader. + pub fn try_read(read: &mut impl std::io::Read) -> std::result::Result { + let mut header = [0_u8; 4]; + read.read_exact(&mut header)?; + Ok(DataHeader(header)) + } + + /// Return a slice containing the entire header. + pub fn as_slice(&self) -> &[u8] { + &self.0 + } + + /// Return the header as array. + pub fn bytes(&self) -> [u8; 4] { + self.0 + } +} + +impl Display for DataHeader { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let header = from_utf8(&self.0).unwrap_or("????"); + write!(f, "{header}") + } +} + +impl From for [u8; 4] { + fn from(val: DataHeader) -> Self { + val.0 + } +} + +impl PartialEq<[u8; 4]> for &DataHeader { + fn eq(&self, other: &[u8; 4]) -> bool { + &self.0 == other + } +} +impl PartialEq<&[u8]> for &DataHeader { + fn eq(&self, other: &&[u8]) -> bool { + &self.0 == other + } +} diff --git a/puffin/src/frame_data.rs b/puffin/src/frame_data.rs index d1fa0992..0c219322 100644 --- a/puffin/src/frame_data.rs +++ b/puffin/src/frame_data.rs @@ -1,3 +1,6 @@ +#[cfg(feature = "packing")] +#[cfg(feature = "serialization")] +use crate::DataHeader; use crate::ScopeDetails; use crate::{Error, FrameIndex, NanoSecond, Result, StreamInfo, ThreadInfo}; #[cfg(feature = "packing")] @@ -566,11 +569,7 @@ impl FrameData { /// Writes one [`FrameData`] into a stream, prefixed by its length ([`u32`] le). #[cfg(not(target_arch = "wasm32"))] // compression not supported on wasm #[cfg(feature = "serialization")] - pub fn write_into( - &self, - scope_collection: Option<&crate::ScopeCollection>, - write: &mut impl std::io::Write, - ) -> anyhow::Result<()> { + pub fn write_into(&self, write: &mut impl std::io::Write) -> anyhow::Result<()> { use bincode::Options as _; use byteorder::{LE, WriteBytesExt as _}; @@ -588,13 +587,7 @@ 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() - } else { - self.scope_delta.clone() - }; - - let serialized_scopes = bincode::options().serialize(&to_serialize_scopes)?; + let serialized_scopes = bincode::options().serialize(&self.scope_delta)?; write.write_u32::(serialized_scopes.len() as u32)?; write.write_all(&serialized_scopes)?; Ok(()) @@ -605,20 +598,14 @@ 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> { + pub fn read_next( + read: &mut impl std::io::Read, + header: &DataHeader, + ) -> anyhow::Result> { use anyhow::Context as _; use bincode::Options as _; use byteorder::{LE, ReadBytesExt}; - let mut header = [0_u8; 4]; - if let Err(err) = read.read_exact(&mut header) { - if err.kind() == std::io::ErrorKind::UnexpectedEof { - return Ok(None); - } else { - return Err(err.into()); - } - } - #[derive(Clone, serde::Deserialize, serde::Serialize)] pub struct LegacyFrameData { pub frame_index: FrameIndex, @@ -657,9 +644,9 @@ impl FrameData { } } - if header == [0_u8; 4] { + if header.bytes() == [0_u8; 4] { Ok(None) // end-of-stream sentinel. - } else if header.starts_with(b"PFD") { + } else if header.as_slice().starts_with(b"PFD") { if &header == b"PFD0" { // Like PDF1, but compressed with `lz4_flex`. // We stopped supporting this in 2021-11-16 in order to remove `lz4_flex` dependency. @@ -792,7 +779,7 @@ impl FrameData { } } else { // Very old packet without magic header - let mut bytes = vec![0_u8; u32::from_le_bytes(header) as usize]; + let mut bytes = vec![0_u8; u32::from_le_bytes(header.bytes()) as usize]; read.read_exact(&mut bytes)?; use bincode::Options as _; diff --git a/puffin/src/lib.rs b/puffin/src/lib.rs index e964b55a..6c937502 100644 --- a/puffin/src/lib.rs +++ b/puffin/src/lib.rs @@ -23,6 +23,8 @@ #![deny(missing_docs)] mod data; +#[cfg(feature = "serialization")] +mod data_header; mod frame_data; mod global_profiler; mod merge; @@ -36,6 +38,8 @@ use std::sync::atomic::{AtomicBool, Ordering}; /// TODO: Improve encapsulation. pub use data::{Error, Reader, Result, Scope, ScopeRecord, Stream, StreamInfo, StreamInfoRef}; +#[cfg(feature = "serialization")] +pub use data_header::DataHeader; pub use frame_data::{FrameData, FrameMeta, UnpackedFrameData}; pub use global_profiler::{FrameSink, GlobalProfiler}; pub use merge::{MergeScope, merge_scopes_for_thread}; diff --git a/puffin/src/profile_view.rs b/puffin/src/profile_view.rs index 21999b48..046b8e4c 100644 --- a/puffin/src/profile_view.rs +++ b/puffin/src/profile_view.rs @@ -5,6 +5,8 @@ use std::{ sync::Arc, }; +#[cfg(feature = "serialization")] +use crate::DataHeader; use crate::{FrameData, FrameSinkId, ScopeCollection}; /// A view of recent and slowest frames, used by GUIs. @@ -60,6 +62,15 @@ impl FrameView { &self.scope_collection } + /// Allow init scope_collection of the FrameView + pub fn init_scope_collection(&mut self, scope_collection: ScopeCollection) { + assert!( + self.scope_collection.scopes_by_id().is_empty() + && self.scope_collection.scopes_by_name().is_empty() + ); + self.scope_collection = scope_collection; + } + /// Adds a new frame to the view. pub fn add_frame(&mut self, new_frame: Arc) { // Register all scopes from the new frame into the scope collection. @@ -227,10 +238,12 @@ impl FrameView { #[cfg(feature = "serialization")] #[cfg(not(target_arch = "wasm32"))] // compression not supported on wasm pub fn write(&self, write: &mut impl std::io::Write) -> anyhow::Result<()> { - write.write_all(b"PUF0")?; + write.write_all(b"PUF1")?; + + self.scope_collection.write_into(write)?; for frame in self.all_uniq() { - frame.write_into(None, write)?; + frame.write_into(write)?; } Ok(()) } @@ -238,22 +251,51 @@ impl FrameView { /// Import profile data from a `.puffin` file/stream. #[cfg(feature = "serialization")] pub fn read(read: &mut impl std::io::Read) -> anyhow::Result { + const MAGIC_0: &[u8; 4] = b"PUF0"; + const MAGIC_1: &[u8; 4] = b"PUF1"; + let mut magic = [0_u8; 4]; read.read_exact(&mut magic)?; - if &magic != b"PUF0" { - anyhow::bail!("Expected .puffin magic header of 'PUF0', found {:?}", magic); + if &magic != MAGIC_0 && &magic != MAGIC_1 { + anyhow::bail!( + "Expected .puffin magic header of '{:?}' ok `{:?}`, found {:?}", + MAGIC_0, + MAGIC_1, + magic + ); } let mut slf = Self { max_recent: usize::MAX, ..Default::default() }; - while let Some(frame) = FrameData::read_next(read)? { - slf.add_frame(frame.into()); + + while let Some(header) = Self::read_header(read)? { + if header.bytes().starts_with(b"PSC") { + slf.init_scope_collection(ScopeCollection::read(read, &header)?); + } else if header.bytes().starts_with(b"PFD") { + if let Some(frame) = FrameData::read_next(read, &header)? { + slf.add_frame(frame.into()); + } + } } Ok(slf) } + + #[cfg(feature = "serialization")] + fn read_header(read: &mut impl std::io::Read) -> Result, anyhow::Error> { + match DataHeader::try_read(read) { + Ok(header) => Ok(Some(header)), + Err(err) => { + if err.kind() == std::io::ErrorKind::UnexpectedEof { + Ok(None) + } else { + Err(err.into()) + } + } + } + } } // ---------------------------------------------------------------------------- diff --git a/puffin/src/scope_details.rs b/puffin/src/scope_details.rs index f85e9da6..8a63f3af 100644 --- a/puffin/src/scope_details.rs +++ b/puffin/src/scope_details.rs @@ -58,6 +58,48 @@ impl ScopeCollection { pub fn scopes_by_id(&self) -> &HashMap> { &self.0.scope_id_to_details } + + #[cfg(feature = "serialization")] + const PSC1: &[u8] = b"PSC1"; + + /// Writes [`ScopeCollection`] into a stream. + #[cfg(feature = "serialization")] + pub fn write_into(&self, write: &mut impl std::io::Write) -> anyhow::Result<()> { + use bincode::Options as _; + + write.write_all(Self::PSC1)?; + let scope_collection = self.scopes_by_id().values().collect::>(); + bincode::options().serialize_into(write, &scope_collection)?; + + Ok(()) + } + + /// Read [`ScopeCollection`] from a stream. + #[cfg(feature = "serialization")] + pub fn read(read: &mut impl std::io::Read, header: &crate::DataHeader) -> anyhow::Result { + use anyhow::anyhow; + match header.as_slice() { + Self::PSC1 => Self::read_scope_collection_format1(read), + _ => Err(anyhow!("`{header}` is not a valid ScopeCollection header")), + } + } + + #[cfg(feature = "serialization")] + fn read_scope_collection_format1(read: &mut impl std::io::Read) -> Result { + use anyhow::Context; + use bincode::Options; + + let scopes: Vec = bincode::options() + .deserialize_from(read) + .context("read scopes collection")?; + + let mut scope_collection = Self::default(); + for scope in scopes { + scope_collection.insert(scope.into()); + } + + Ok(scope_collection) + } } /// Scopes are identified by user-provided name while functions are identified by the function name. diff --git a/puffin_http/src/client.rs b/puffin_http/src/client.rs index a8f37fc8..5fb97cc2 100644 --- a/puffin_http/src/client.rs +++ b/puffin_http/src/client.rs @@ -1,9 +1,15 @@ +use anyhow::Context as _; use std::sync::{ Arc, atomic::{AtomicBool, Ordering::SeqCst}, }; -use puffin::{FrameData, FrameView}; +use puffin::{DataHeader, FrameData, FrameView, ScopeCollection}; + +enum MessageContent { + FrameData(FrameData), + ScopeCollection(ScopeCollection), +} /// Connect to a [`crate::Server`], reading profile data /// and feeding it to a [`puffin::FrameView`]. @@ -56,11 +62,16 @@ impl Client { connected.store(true, SeqCst); while alive.load(SeqCst) { match consume_message(&mut stream) { - Ok(frame_data) => { - frame_view + Ok(message_content) => match message_content { + MessageContent::FrameData(frame_data) => frame_view .lock() - .add_frame(std::sync::Arc::new(frame_data)); - } + .add_frame(std::sync::Arc::new(frame_data)), + MessageContent::ScopeCollection(scope_collection) => { + frame_view + .lock() + .init_scope_collection(scope_collection); + } + }, Err(err) => { log::warn!( "Connection to puffin server closed: {}", @@ -101,7 +112,7 @@ impl Client { } /// Read a `puffin_http` message from a stream. -pub fn consume_message(stream: &mut impl std::io::Read) -> anyhow::Result { +fn consume_message(stream: &mut impl std::io::Read) -> anyhow::Result { let mut server_version = [0_u8; 2]; stream.read_exact(&mut server_version)?; let server_version = u16::from_le_bytes(server_version); @@ -124,11 +135,21 @@ pub fn consume_message(stream: &mut impl std::io::Read) -> anyhow::Result add `ScopeCollection` dedicated messages +pub const PROTOCOL_VERSION: u16 = 3; /// The default TCP port used. pub const DEFAULT_PORT: u16 = 8585; diff --git a/puffin_http/src/server.rs b/puffin_http/src/server.rs index 833b120c..00112025 100644 --- a/puffin_http/src/server.rs +++ b/puffin_http/src/server.rs @@ -257,17 +257,16 @@ impl Server { tcp_listener, clients: Default::default(), num_clients: num_clients_cloned, - send_all_scopes: false, scope_collection: Default::default(), }; while let Ok(frame) = rx.recv() { if let Err(err) = server_impl.accept_new_clients() { - log::warn!("puffin server failure: {err}"); + log::warn!("puffin server `accept new clients` failure: {err}"); } if let Err(err) = server_impl.send(&frame) { - log::warn!("puffin server failure: {err}"); + log::warn!("puffin server `send FrameData` failure: {err}"); } } }) @@ -332,7 +331,6 @@ struct PuffinServerImpl { tcp_listener: TcpListener, clients: Vec, num_clients: Arc, - send_all_scopes: bool, scope_collection: ScopeCollection, } @@ -354,13 +352,16 @@ impl PuffinServerImpl { .spawn(move || client_loop(packet_rx, client_addr, tcp_stream)) .context("Couldn't spawn thread")?; - // Send all scopes when new client connects. - self.send_all_scopes = true; - self.clients.push(Client { + let new_client = Client { client_addr, packet_tx: Some(packet_tx), join_handle: Some(join_handle), - }); + }; + // Send all scopes to new connected client. + if let Err(err) = self.send_scopes_to_client(&new_client) { + log::warn!("puffin server `send scopes` failure: {err}"); + } + self.clients.push(new_client); self.num_clients.store(self.clients.len(), Ordering::SeqCst); } Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { @@ -393,16 +394,9 @@ impl PuffinServerImpl { .write_all(&crate::PROTOCOL_VERSION.to_le_bytes()) .context("Encode puffin `PROTOCOL_VERSION` in packet to be send to client.")?; - let scope_collection = if self.send_all_scopes { - Some(&self.scope_collection) - } else { - None - }; - frame - .write_into(scope_collection, &mut packet) + .write_into(&mut packet) .context("Encode puffin frame")?; - self.send_all_scopes = false; let packet: Packet = packet.into(); @@ -424,6 +418,22 @@ impl PuffinServerImpl { Ok(()) } + + pub fn send_scopes_to_client(&self, client: &Client) -> anyhow::Result<(), anyhow::Error> { + let mut packet = vec![]; + packet + .write_all(&crate::PROTOCOL_VERSION.to_le_bytes()) + .context("Encode puffin `PROTOCOL_VERSION` in packet to be send to client.")?; + self.scope_collection.write_into(&mut packet)?; + + let packet: Packet = packet.into(); + match &client.packet_tx { + None => Ok(()), + Some(packet_tx) => packet_tx + .try_send(packet.clone()) + .map_err(anyhow::Error::from), + } + } } #[expect(clippy::needless_pass_by_value)]