starnix_core/perf/
event.rs

1// Copyright 2024 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use std::sync::Arc;
6use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
7
8use crate::task::Kernel;
9use crate::vfs::OutputBuffer;
10use fuchsia_inspect::Property;
11use fuchsia_rcu::RcuDroppableArc;
12use paged_ring_buffer::LocklessRingBuffer;
13use starnix_logging::log_error;
14use starnix_sync::{LockDepMutex, PerfTraceEventQueueListLock};
15use starnix_types::PAGE_SIZE;
16use starnix_uapi::errors::Errno;
17use starnix_uapi::{errno, error};
18use zerocopy::native_endian::{I32, U16, U32, U64};
19use zerocopy::{Immutable, IntoBytes, Unaligned};
20use zx::BootTimeline;
21
22// The default ring buffer size (2MB).
23const DEFAULT_RING_BUFFER_SIZE_BYTES: usize = 2097152;
24
25// The event id for atrace events.
26const FTRACE_PRINT_ID: U16 = U16::new(5);
27
28// Used for inspect tracking.
29const DROPPED_PAGES: &str = "dropped_pages";
30
31#[repr(C)]
32#[derive(Debug, Default, IntoBytes, Immutable, Unaligned)]
33struct PrintEventHeader {
34    common_type: U16,
35    common_flags: u8,
36    common_preempt_count: u8,
37    common_pid: I32,
38    ip: U64,
39}
40
41#[repr(C)]
42#[derive(Debug, IntoBytes, Immutable, Unaligned)]
43struct PrintEvent {
44    header: PrintEventHeader,
45}
46
47impl PrintEvent {
48    fn new(pid: i32) -> Self {
49        Self {
50            header: PrintEventHeader {
51                common_type: FTRACE_PRINT_ID,
52                common_pid: I32::new(pid),
53                // Perfetto doesn't care about any other field.
54                ..Default::default()
55            },
56        }
57    }
58
59    fn size(&self) -> usize {
60        std::mem::size_of::<PrintEventHeader>()
61    }
62}
63
64#[repr(C)]
65#[derive(Debug, Default, IntoBytes, PartialEq, Immutable, Unaligned)]
66struct TraceEventHeader {
67    // u32 where:
68    //   type_or_length: bottom 5 bits. If 0, `data` is read for length. Always set to 0 for now.
69    //   time_delta: top 27 bits
70    time_delta: U32,
71
72    // If type_or_length is 0, holds the length of the trace message.
73    // We always write length here for simplicity.
74    data: U32,
75}
76
77impl TraceEventHeader {
78    fn new(size: usize) -> Self {
79        // The size reported in the event's header includes the size of `size` (a u32) and the size
80        // of the event data.
81        let size = (std::mem::size_of::<u32>() + size) as u32;
82        Self { time_delta: U32::new(0), data: U32::new(size) }
83    }
84
85    fn set_time_delta(&mut self, nanos: u32) {
86        // `Reservation::header_time_delta()` already fits in 27 bits: the ring buffer puts a
87        // TIME_STAMP record before a record whose delta doesn't. Saturate anyway, so that a bug
88        // can't spill into the `type_or_length` bits.
89        let saturated_nanos = u64::from(nanos).min(LocklessRingBuffer::MAX_TIME_DELTA_NANOS) as u32;
90        // Move into the high 27 bits reserving the lower 5 bits for the type_or_length value.
91        self.time_delta = U32::new(saturated_nanos << LocklessRingBuffer::TYPE_OR_LENGTH_BITS);
92    }
93}
94
95#[repr(C)]
96#[derive(Debug, IntoBytes, Immutable, Unaligned)]
97pub struct TraceEvent {
98    /// Common metadata among all trace event types.
99    header: TraceEventHeader, // u64
100
101    /// The event data.
102    ///
103    /// Atrace events are reported as PrintFtraceEvents. When we support multiple types of events,
104    /// this can be updated to be more generic.
105    event: PrintEvent,
106}
107
108impl TraceEvent {
109    pub fn new(pid: i32, data_len: usize) -> Self {
110        let event = PrintEvent::new(pid);
111        // +1 because we append a trailing '\n' to the data when we serialize.
112        let header = TraceEventHeader::new(event.size() + data_len + 1);
113        Self { header, event }
114    }
115
116    fn size(&self) -> usize {
117        // The header's data size doesn't include the time_delta size.
118        std::mem::size_of::<u32>() + self.header.data.get() as usize
119    }
120}
121
122/// Stores all trace events.
123pub struct TraceEventQueue {
124    /// The trace events.
125    ring_buffer: RcuDroppableArc<LocklessRingBuffer>,
126
127    /// Async ID for read track grouping.
128    pub async_id_read: fuchsia_trace::Id,
129
130    /// Async ID for write track grouping.
131    pub async_id_write: fuchsia_trace::Id,
132
133    /// CPU ID for trace tracks.
134    pub cpu_id: u32,
135}
136
137const STATIC_READ_TRACK_NAMES: [&str; 8] = [
138    "Tracefs read 0",
139    "Tracefs read 1",
140    "Tracefs read 2",
141    "Tracefs read 3",
142    "Tracefs read 4",
143    "Tracefs read 5",
144    "Tracefs read 6",
145    "Tracefs read 7",
146];
147
148const STATIC_WRITE_TRACK_NAMES: [&str; 8] = [
149    "Tracefs write 0",
150    "Tracefs write 1",
151    "Tracefs write 2",
152    "Tracefs write 3",
153    "Tracefs write 4",
154    "Tracefs write 5",
155    "Tracefs write 6",
156    "Tracefs write 7",
157];
158
159impl<'a> TraceEventQueue {
160    pub(crate) fn new(cpu_id: u32) -> Result<Self, Errno> {
161        Self::new_with_size(cpu_id, DEFAULT_RING_BUFFER_SIZE_BYTES)
162    }
163
164    pub(crate) fn new_with_size(cpu_id: u32, size_bytes: usize) -> Result<Self, Errno> {
165        let async_id_read = fuchsia_trace::Id::new();
166        let async_id_write = fuchsia_trace::Id::new();
167        let ring_buffer = Arc::new(
168            LocklessRingBuffer::new(size_bytes, true, async_id_write)
169                .map_err(|_| errno!(ENOMEM))?,
170        );
171        ring_buffer.disable()?;
172
173        Ok(Self {
174            ring_buffer: RcuDroppableArc::new(ring_buffer),
175            async_id_read,
176            async_id_write,
177            cpu_id,
178        })
179    }
180
181    pub fn read_track_name(&self) -> std::borrow::Cow<'static, str> {
182        if let Some(&name) = STATIC_READ_TRACK_NAMES.get(self.cpu_id as usize) {
183            std::borrow::Cow::Borrowed(name)
184        } else {
185            std::borrow::Cow::Owned(format!("Tracefs read {}", self.cpu_id))
186        }
187    }
188
189    pub fn write_track_name(&self) -> std::borrow::Cow<'static, str> {
190        if let Some(&name) = STATIC_WRITE_TRACK_NAMES.get(self.cpu_id as usize) {
191            std::borrow::Cow::Borrowed(name)
192        } else {
193            std::borrow::Cow::Owned(format!("Tracefs write {}", self.cpu_id))
194        }
195    }
196
197    fn enable(&self) -> Result<zx::BootInstant, Errno> {
198        self.ring_buffer.read().enable()
199    }
200
201    /// Disables the event queue and resets it to empty.
202    /// The number of dropped pages are recorded for reading via tracefs.
203    fn disable(&self) -> Result<u64, Errno> {
204        self.ring_buffer.read().disable()
205    }
206
207    /// Reads a page worth of events. Currently only reads pages that are full.
208    ///
209    /// From https://docs.kernel.org/trace/ring-buffer-design.html, when memory is mapped, a reader
210    /// page can be swapped with the header page to avoid copying memory.
211    pub fn read(&self, buf: &mut dyn OutputBuffer) -> Result<usize, Errno> {
212        self.ring_buffer.read().read(&mut RingOutputBuffer(buf))
213    }
214
215    /// Write `event` into `ring_buffer`.
216    /// If `event` does not fit in the current page, move on to the next.
217    ///
218    /// Should eventually allow for a writer to preempt another writer.
219    /// See https://docs.kernel.org/trace/ring-buffer-design.html.
220    /// Returns the delta duration between this event and the previous event written.
221    pub fn push_event(
222        &self,
223        event: TraceEvent,
224        data: &[u8],
225    ) -> Result<zx::Duration<BootTimeline>, Errno> {
226        self.push_event_with_clock(event, data, zx::BootInstant::get)
227    }
228
229    /// Like `push_event()`, but reads the current time from `current_boot_time`, so that tests can
230    /// control timestamps.
231    fn push_event_with_clock(
232        &self,
233        mut event: TraceEvent,
234        data: &[u8],
235        current_boot_time: impl Fn() -> zx::BootInstant,
236    ) -> Result<zx::Duration<BootTimeline>, Errno> {
237        let size = event.size();
238        let ring_buffer = self.ring_buffer.read();
239
240        let (res, _timestamp, delta) = match ring_buffer.reserve_with_clock(size, current_boot_time)
241        {
242            Ok(res) => res,
243            Err(e) if e == errno!(EINVAL) => {
244                log_error!("Invalid reservation size: {}", size);
245                return Err(errno!(EINVAL));
246            }
247            Err(e) if e == errno!(ENOSPC) => {
248                log_error!("Ring buffer full, dropping event of size: {}", size);
249                return Err(errno!(ENOSPC));
250            }
251            Err(e) => return Err(e),
252        };
253
254        event.header.set_time_delta(res.header_time_delta());
255
256        let bytes = event.as_bytes();
257        res.write_at(0, bytes);
258        res.write_at(bytes.len(), data);
259        res.write_at(bytes.len() + data.len(), b"\n");
260
261        ring_buffer.commit(res);
262
263        Ok(delta)
264    }
265}
266
267pub struct TraceEventQueueList {
268    pub queues: Vec<Arc<TraceEventQueue>>,
269    // Serializes `enable()`, `disable()`, and `set_buffer_size_kb()`, so that the queues,
270    // `tracing_enabled`, and `configured_buffer_size_kb` change together, as Linux
271    // `trace_types_lock` does. Readers of `tracing_enabled` and `configured_buffer_size_kb`
272    // and writers to the queues do not take it.
273    state_mutex: LockDepMutex<(), PerfTraceEventQueueListLock>,
274    tracing_enabled: Arc<AtomicBool>,
275    configured_buffer_size_kb: AtomicUsize,
276    tracefs_node: fuchsia_inspect::Node,
277    buffer_size_kb_property: fuchsia_inspect::UintProperty,
278    total_capacity_bytes_property: fuchsia_inspect::UintProperty,
279}
280
281impl TraceEventQueueList {
282    pub fn from(kernel: &Kernel) -> Arc<Self> {
283        kernel.expando.get_or_init(|| {
284            let num_cpus = zx::system_get_num_cpus() as usize;
285            Self::new(num_cpus, kernel.inspect_node.create_child("tracefs"))
286        })
287    }
288
289    #[cfg(test)]
290    pub(crate) fn new_for_testing(
291        num_cpus: usize,
292        inspect_parent: &fuchsia_inspect::Node,
293    ) -> Arc<Self> {
294        Arc::new(Self::new(num_cpus, inspect_parent.create_child("tracefs")))
295    }
296
297    fn new(num_cpus: usize, tracefs_node: fuchsia_inspect::Node) -> Self {
298        let default_size_kb = DEFAULT_RING_BUFFER_SIZE_BYTES / 1024;
299        let buffer_size_kb_property =
300            tracefs_node.create_uint("buffer_size_kb", default_size_kb as u64);
301        let total_capacity_bytes_property = tracefs_node.create_uint(
302            "total_capacity_bytes",
303            (DEFAULT_RING_BUFFER_SIZE_BYTES * num_cpus) as u64,
304        );
305        let queues = (0..num_cpus)
306            .map(|cpu| Arc::new(TraceEventQueue::new(cpu as u32).expect("create queue")))
307            .collect();
308        Self {
309            queues,
310            state_mutex: Default::default(),
311            tracing_enabled: Arc::new(AtomicBool::new(false)),
312            configured_buffer_size_kb: AtomicUsize::new(default_size_kb),
313            tracefs_node,
314            buffer_size_kb_property,
315            total_capacity_bytes_property,
316        }
317    }
318
319    pub fn is_enabled(&self) -> bool {
320        self.tracing_enabled.load(Ordering::Acquire)
321    }
322
323    pub fn buffer_size_kb(&self) -> usize {
324        self.configured_buffer_size_kb.load(Ordering::Acquire)
325    }
326
327    /// Replaces every queue's ring buffer with an empty one of at least `size_kb` KiB, rounded up
328    /// to whole pages and to `LocklessRingBuffer`'s 3-page minimum, and returns the size used.
329    /// The new buffers keep the current `tracing_on` state.
330    ///
331    /// Unlike Linux `ring_buffer_resize()`, which adds or removes pages in place, this discards
332    /// every unread event, including events from writers still holding the old buffer. The old
333    /// buffer's dropped-page count is discarded too, so `DROPPED_PAGES` misses it.
334    // TODO(https://fxbug.dev/574072344): Count the events and pages a resize discards.
335    pub fn set_buffer_size_kb(&self, size_kb: usize) -> Result<usize, Errno> {
336        if size_kb == 0 {
337            return error!(EINVAL);
338        }
339        let _lock = self.state_mutex.lock();
340        let enabled = self.is_enabled();
341
342        let page_size_kb = (*PAGE_SIZE as usize) / 1024;
343        let requested_pages = size_kb.div_ceil(page_size_kb);
344        // LocklessRingBuffer requires at least 3 pages (read page, head page, and swap page).
345        let pages = std::cmp::max(3, requested_pages);
346        let actual_size_kb = pages.checked_mul(page_size_kb).ok_or_else(|| errno!(ENOMEM))?;
347        let target_bytes = actual_size_kb.checked_mul(1024).ok_or_else(|| errno!(ENOMEM))?;
348        let total_capacity_bytes =
349            target_bytes.checked_mul(self.queues.len()).ok_or_else(|| errno!(ENOMEM))?;
350
351        // Transactional allocation: pre-allocate all per-CPU ring buffers first.
352        let mut new_buffers = Vec::with_capacity(self.queues.len());
353        for queue in &self.queues {
354            let ring_buffer = Arc::new(
355                LocklessRingBuffer::new(target_bytes, true, queue.async_id_write)
356                    .map_err(|_| errno!(ENOMEM))?,
357            );
358            if !enabled {
359                ring_buffer.disable()?;
360            }
361            new_buffers.push(ring_buffer);
362        }
363
364        // All succeeded; swap in the new buffers.
365        for (queue, new_buffer) in self.queues.iter().zip(new_buffers) {
366            queue.ring_buffer.update(new_buffer);
367        }
368
369        self.configured_buffer_size_kb.store(actual_size_kb, Ordering::Release);
370        self.buffer_size_kb_property.set(actual_size_kb as u64);
371        self.total_capacity_bytes_property.set(total_capacity_bytes as u64);
372
373        Ok(actual_size_kb)
374    }
375
376    /// Enables every queue. A queue that is already enabled is left alone, as Linux
377    /// `rb_simple_write()` does when `tracing_on` is set to its current state. Unlike Linux, which
378    /// keeps the buffer readable after `tracing_on` is cleared, `disable()` discards each queue's
379    /// contents, so enabling a disabled queue starts from empty.
380    pub fn enable(&self) -> Result<(), Errno> {
381        let _lock = self.state_mutex.lock();
382        let mut first_error = None;
383        for queue in &self.queues {
384            if let Err(e) = queue.enable() {
385                if first_error.is_none() {
386                    first_error = Some(e);
387                }
388            }
389        }
390        if let Some(e) = first_error {
391            return Err(e);
392        }
393        self.tracing_enabled.store(true, Ordering::Release);
394        Ok(())
395    }
396
397    pub fn disable(&self) -> Result<(), Errno> {
398        let _lock = self.state_mutex.lock();
399        // Set disabled to stop new access to the queue, then clean up each one.
400        self.tracing_enabled.store(false, Ordering::Release);
401        let mut first_error = None;
402        let mut total_dropped = 0;
403        for queue in &self.queues {
404            match queue.disable() {
405                Ok(dropped) => total_dropped += dropped,
406                Err(e) => {
407                    if first_error.is_none() {
408                        first_error = Some(e);
409                    }
410                }
411            }
412        }
413
414        self.tracefs_node.record_uint(DROPPED_PAGES, total_dropped);
415
416        if let Some(e) = first_error {
417            return Err(e);
418        }
419        Ok(())
420    }
421}
422
423/// Adapts a starnix_core `OutputBuffer` to the `paged_ring_buffer::OutputBuffer` that
424/// `LocklessRingBuffer::read()` takes.
425struct RingOutputBuffer<'a>(&'a mut dyn OutputBuffer);
426
427impl paged_ring_buffer::OutputBuffer for RingOutputBuffer<'_> {
428    fn available(&self) -> usize {
429        self.0.available()
430    }
431    fn write_each(
432        &mut self,
433        callback: &mut paged_ring_buffer::OutputBufferCallback<'_>,
434    ) -> Result<usize, Errno> {
435        self.0.write_each(callback)
436    }
437}
438
439#[cfg(test)]
440mod tests {
441    use super::{
442        DEFAULT_RING_BUFFER_SIZE_BYTES, LocklessRingBuffer, PrintEvent, TraceEvent,
443        TraceEventQueue, TraceEventQueueList,
444    };
445    use crate::vfs::OutputBuffer;
446    use crate::vfs::buffers::VecOutputBuffer;
447    use diagnostics_assertions::assert_data_tree;
448    use std::sync::Arc;
449
450    use starnix_types::PAGE_SIZE;
451    use starnix_uapi::error;
452    use zerocopy::IntoBytes;
453
454    #[fuchsia::test]
455    fn trace_event_queue_empty_errors() {
456        let queue = TraceEventQueue::new(0).unwrap();
457
458        let mut buffer = VecOutputBuffer::new(*PAGE_SIZE as usize);
459        assert_eq!(queue.read(&mut buffer), error!(EAGAIN));
460
461        let data = b"B|1234|slice_name";
462        let event = TraceEvent::new(1234, data.len());
463        assert_eq!(queue.push_event(event, data), error!(ENOMEM));
464    }
465
466    #[fuchsia::test]
467    fn read_empty_queue() {
468        let queue = TraceEventQueue::new(0).expect("create queue");
469        let mut buffer = VecOutputBuffer::new(*PAGE_SIZE as usize);
470        assert_eq!(queue.read(&mut buffer), error!(EAGAIN));
471    }
472
473    #[fuchsia::test]
474    fn enable_disable_queue() {
475        let queue = TraceEventQueue::new(0).expect("create queue");
476        assert!(!queue.ring_buffer.read().is_enabled());
477
478        // Enable tracing.
479        assert!(queue.enable().is_ok());
480
481        // Confirm we can push an event.
482        let data = b"B|1234|slice_name";
483        let event = TraceEvent::new(1234, data.len());
484        let result = queue.push_event(event, data);
485
486        assert!(result.is_ok());
487        assert_eq!(result.as_ref().unwrap().into_nanos(), 0);
488
489        // Disable tracing and check that the queue's state has been reset.
490        assert!(queue.disable().is_ok());
491        assert!(!queue.ring_buffer.read().is_enabled());
492    }
493
494    #[fuchsia::test]
495    fn buffer_size_kb_configuration() {
496        let inspector = fuchsia_inspect::Inspector::default();
497        let list = TraceEventQueueList::new_for_testing(2, inspector.root());
498        assert_eq!(list.buffer_size_kb(), DEFAULT_RING_BUFFER_SIZE_BYTES / 1024);
499
500        // Resize when disabled succeeds.
501        let new_size_kb = list.set_buffer_size_kb(4096).expect("set buffer size");
502        assert_eq!(new_size_kb, 4096);
503        assert_eq!(list.buffer_size_kb(), 4096);
504        for queue in &list.queues {
505            assert_eq!(queue.ring_buffer.read().size_bytes(), 4096 * 1024);
506            assert!(!queue.ring_buffer.read().is_enabled());
507        }
508
509        // Small size is clamped to minimum 3 pages (12 KB on 4KB page system).
510        let clamped_size_kb = list.set_buffer_size_kb(4).expect("clamp buffer size");
511        assert_eq!(clamped_size_kb, 12);
512        assert_eq!(list.buffer_size_kb(), 12);
513        for queue in &list.queues {
514            assert_eq!(queue.ring_buffer.read().size_bytes(), 12 * 1024);
515            assert!(!queue.ring_buffer.read().is_enabled());
516        }
517
518        // Zero returns EINVAL.
519        assert_eq!(list.set_buffer_size_kb(0), error!(EINVAL));
520
521        // When enabled, resizing succeeds and keeps the new buffers enabled.
522        list.enable().expect("enable tracing");
523        let enabled_size_kb = list.set_buffer_size_kb(2048).expect("set buffer size while enabled");
524        assert_eq!(enabled_size_kb, 2048);
525        assert_eq!(list.buffer_size_kb(), 2048);
526        for queue in &list.queues {
527            assert_eq!(queue.ring_buffer.read().size_bytes(), 2048 * 1024);
528            assert!(queue.ring_buffer.read().is_enabled());
529        }
530        list.disable().expect("disable tracing");
531    }
532
533    #[fuchsia::test]
534    async fn buffer_size_kb_inspect_updates_in_place() {
535        let inspector = fuchsia_inspect::Inspector::default();
536        let list = TraceEventQueueList::new_for_testing(2, inspector.root());
537        assert_data_tree!(inspector, root: {
538            tracefs: {
539                buffer_size_kb: (DEFAULT_RING_BUFFER_SIZE_BYTES / 1024) as u64,
540                total_capacity_bytes: (DEFAULT_RING_BUFFER_SIZE_BYTES * 2) as u64,
541            }
542        });
543
544        // Each resize must update the existing properties rather than add new ones, so the tree
545        // must still have exactly one of each.
546        list.set_buffer_size_kb(4096).expect("set buffer size");
547        list.set_buffer_size_kb(1024).expect("set buffer size");
548        assert_data_tree!(inspector, root: {
549            tracefs: {
550                buffer_size_kb: 1024u64,
551                total_capacity_bytes: (1024 * 1024 * 2) as u64,
552            }
553        });
554    }
555
556    #[fuchsia::test]
557    fn create_trace_event() {
558        // Create an event.
559        let event = TraceEvent::new(1234, b"B|1234|slice_name".len());
560        let event_size = event.size();
561        assert_eq!(event_size, 42);
562    }
563
564    // This can be removed when we support reading incomplete pages.
565    #[fuchsia::test]
566    fn single_trace_event_fails_read() {
567        let queue = TraceEventQueue::new(0).expect("create queue");
568        queue.enable().expect("enable queue");
569        // Create an event.
570        let data = b"B|1234|slice_name";
571        let event = TraceEvent::new(1234, data.len());
572
573        // Push the event into the queue.
574        let result = queue.push_event(event, data);
575        assert!(result.is_ok());
576        assert_eq!(result.ok().expect("delta").into_nanos(), 0);
577
578        let mut buffer = VecOutputBuffer::new(*PAGE_SIZE as usize);
579        assert_eq!(queue.read(&mut buffer), error!(EAGAIN));
580    }
581
582    const TYPE_OR_LENGTH_MASK: u32 = (1 << LocklessRingBuffer::TYPE_OR_LENGTH_BITS) - 1;
583    // The `type_or_length` of a data record whose length is in the u32 after the header.
584    const TYPE_OR_LENGTH_DATA: u32 = 0;
585
586    /// A data record from an ftrace page.
587    struct PageRecord {
588        /// The absolute time from the `TIME_STAMP` record right before this record, if any.
589        time_stamp: Option<u64>,
590        /// The record header's time delta field.
591        time_delta: u32,
592        /// The length word, which counts itself and the body.
593        length: u32,
594        /// The bytes after the length word.
595        body: Vec<u8>,
596    }
597
598    /// Splits an ftrace page into its data records. Panics on any other record type, on a
599    /// `TIME_STAMP` record that no data record follows, or on a record that runs past the page's
600    /// data size.
601    fn page_records(page: &[u8]) -> Vec<PageRecord> {
602        let data_size = u64::from_le_bytes(page[8..16].try_into().unwrap()) as usize;
603        let end = LocklessRingBuffer::PAGE_HEADER_SIZE + data_size;
604        assert!(end <= page.len(), "data size {data_size} runs past the page");
605        let word = |at: usize| {
606            assert!(at + 4 <= end, "read at {at} runs past the page data ({end})");
607            u32::from_le_bytes(page[at..at + 4].try_into().unwrap())
608        };
609        let mut records = vec![];
610        let mut time_stamp = None;
611        let mut offset = LocklessRingBuffer::PAGE_HEADER_SIZE;
612        while offset < end {
613            let header = word(offset);
614            let time_delta = header >> LocklessRingBuffer::TYPE_OR_LENGTH_BITS;
615            match header & TYPE_OR_LENGTH_MASK {
616                LocklessRingBuffer::RINGBUF_TYPE_TIME_STAMP => {
617                    assert!(time_stamp.is_none(), "two TIME_STAMP records in a row at {offset}");
618                    let high = word(offset + 4);
619                    time_stamp = Some(
620                        (u64::from(high) << LocklessRingBuffer::TIME_DELTA_BITS)
621                            | u64::from(time_delta),
622                    );
623                    offset += LocklessRingBuffer::TIME_STAMP_RECORD_SIZE;
624                }
625                TYPE_OR_LENGTH_DATA => {
626                    let length = word(offset + 4);
627                    let record_end = offset + 4 + length as usize;
628                    assert!(record_end <= end, "record at {offset} runs past the page data");
629                    let body = page[offset + 8..record_end].to_vec();
630                    records.push(PageRecord {
631                        time_stamp: time_stamp.take(),
632                        time_delta,
633                        length,
634                        body,
635                    });
636                    offset = record_end;
637                }
638                other => panic!("unexpected record type {other} at offset {offset}"),
639            }
640        }
641        assert!(time_stamp.is_none(), "no data record follows the last TIME_STAMP record");
642        records
643    }
644
645    #[fuchsia::test]
646    fn page_overflow() {
647        let queue = TraceEventQueue::new(0).expect("create queue");
648        let queue_start_timestamp = queue.enable().expect("enable queue").into_nanos();
649
650        let pid = 1234;
651        let data = b"B|1234|loooooooooooooooooooooooooooooooooooooooooooooooooooooooooo\
652        ooooooooooooooooooooooooooooooooooooooooooooooooooooooooongevent";
653        let expected_event = TraceEvent::new(pid, data.len());
654        assert_eq!(expected_event.size(), 155);
655        let page_capacity = *PAGE_SIZE as usize - LocklessRingBuffer::PAGE_HEADER_SIZE;
656        let events_per_page = page_capacity / expected_event.size();
657        // Page 0 holds `events_per_page` events even with a TIME_STAMP record among them.
658        assert!(
659            events_per_page * expected_event.size() + LocklessRingBuffer::TIME_STAMP_RECORD_SIZE
660                <= page_capacity
661        );
662
663        // Events are STEP_NANOS apart, except that event GAP_EVENT is more than
664        // MAX_TIME_DELTA_NANOS after the one before it. Push one more event than Page 0 holds, so
665        // that Page 0 is finalized and can be read.
666        const STEP_NANOS: i64 = 1000;
667        const GAP_EVENT: usize = 5;
668        let gap_nanos = LocklessRingBuffer::MAX_TIME_DELTA_NANOS as i64 + 1;
669        let times: Vec<i64> = (0..=events_per_page)
670            .scan(queue_start_timestamp, |time, i| {
671                *time += if i == GAP_EVENT { gap_nanos } else { STEP_NANOS };
672                Some(*time)
673            })
674            .collect();
675        let deltas: Vec<i64> = times
676            .iter()
677            .map(|&time| {
678                let event = TraceEvent::new(pid, data.len());
679                queue
680                    .push_event_with_clock(event, data, || zx::BootInstant::from_nanos(time))
681                    .expect("push event")
682                    .into_nanos()
683            })
684            .collect();
685
686        let mut buffer = VecOutputBuffer::new(*PAGE_SIZE as usize);
687        assert_eq!(queue.read(&mut buffer), Ok(*PAGE_SIZE as usize));
688        assert_eq!(buffer.bytes_written() as u64, *PAGE_SIZE);
689        let page = buffer.data();
690        // The page header holds the time of the first event on the page.
691        let page_timestamp = u64::from_le_bytes(page[0..8].try_into().unwrap());
692        assert_eq!(page_timestamp, times[0] as u64);
693        // Page 0 holds the first `events_per_page` events and GAP_EVENT's TIME_STAMP record. The
694        // next event doesn't fit.
695        let data_size = u64::from_le_bytes(page[8..16].try_into().unwrap()) as usize;
696        assert_eq!(
697            data_size,
698            events_per_page * expected_event.size() + LocklessRingBuffer::TIME_STAMP_RECORD_SIZE
699        );
700
701        // Each event is one ftrace record, in push order. push_event() returns the delta from the
702        // previous event, and the record header holds it, except for the first event on the page,
703        // whose time the page header carries, and GAP_EVENT, whose time a TIME_STAMP record
704        // carries. Their header deltas are 0.
705        let records = page_records(page);
706        assert_eq!(records.len(), events_per_page);
707        let mut expected_body = PrintEvent::new(pid).as_bytes().to_vec();
708        expected_body.extend_from_slice(data);
709        expected_body.push(b'\n');
710        for (i, record) in records.iter().enumerate() {
711            assert_eq!(record.length as usize, expected_event.size() - 4, "event {i}");
712            assert_eq!(record.body, expected_body, "event {i}");
713            let (expected_delta, expected_time_delta, expected_time_stamp) = match i {
714                0 => (0, 0, None),
715                GAP_EVENT => (gap_nanos, 0, Some(times[i] as u64)),
716                _ => (STEP_NANOS, STEP_NANOS as u32, None),
717            };
718            assert_eq!(deltas[i], expected_delta, "event {i}");
719            assert_eq!(record.time_delta, expected_time_delta, "event {i}");
720            assert_eq!(record.time_stamp, expected_time_stamp, "event {i}");
721        }
722        // The next event is the first on Page 1.
723        assert_eq!(deltas[events_per_page], 0);
724
725        // Page 1 is not full, so it can't be read yet.
726        let mut buffer = VecOutputBuffer::new(*PAGE_SIZE as usize);
727        assert_eq!(queue.read(&mut buffer), error!(EAGAIN));
728    }
729
730    // Races list-level enable() against disable() and checks that the queues and
731    // `tracing_enabled` agree afterwards. Without `state_mutex`, enable() can enable every queue,
732    // then disable() can clear `tracing_enabled` and disable every queue, and then enable() can
733    // set `tracing_enabled`, leaving it set with every queue disabled. That needs disable() to
734    // run in the short window between enable()'s queue loop and its store, so this test catches
735    // a regression only on some runs.
736    #[fuchsia::test]
737    fn concurrent_enable_disable_keeps_list_consistent() {
738        let list = TraceEventQueueList::new_for_testing(4, &fuchsia_inspect::Node::default());
739        for _ in 0..200 {
740            let barrier = Arc::new(std::sync::Barrier::new(2));
741            let enabler = {
742                let list = Arc::clone(&list);
743                let barrier = Arc::clone(&barrier);
744                std::thread::spawn(move || {
745                    barrier.wait();
746                    list.enable().unwrap();
747                })
748            };
749            barrier.wait();
750            list.disable().unwrap();
751            enabler.join().unwrap();
752
753            let enabled = list.is_enabled();
754            for queue in &list.queues {
755                assert_eq!(queue.ring_buffer.read().is_enabled(), enabled);
756            }
757        }
758    }
759}
starnix_core/perf/event.rs

starnix_core/perf/
event.rs
