Capture block I/O operations into CBOR

docs/src/ostd/data-capture.md

@@ -0,0 +1,15 @@
+# Data OQueues available in OSTD
+
+## Scheduler
+
+Scheduling events can placed on an OQueue. Unlike most OQueues, this is disabled by default, because
+of the potential for unknown overhead in a very sensitive part of the system. It can be enabled with
+the `capture_scheduling` feature. You can enable this feature with `--features
+ostd/capture_scheduling` on the `cargo osdk` command line. (Or the
+`FEATURES=ostd/capture_scheduling` environment variable for OSTDs own makefiles.)
+
+(NOTE: Once we are confident that the overhead is low enough, `capture_scheduling` will be enabled
+by default.)
+
+To have the the Mariposa kernel capture the scheduling events to a file, add this *and* the kernel
+command line argument `scheduler.capture_data=true`.

kernel/comps/block/Cargo.toml

@@ -13,6 +13,7 @@ int-to-c-enum = { path = "../../libs/int-to-c-enum" }
 component = { path = "../../libs/comp-sys/component" }
 aster-time = { path = "../time" }
 log = "0.4"
+serde = { version = "1.0", default-features = false, features = ["derive", "alloc"]}
 bitvec = { version = "1.0.1", default-features = false, features = ["alloc"] }
 
 [lints]

kernel/comps/block/src/bio.rs

@@ -17,6 +17,7 @@ use ostd::{
     },
     sync::{LocalIrqDisabled, SpinLock, WaitQueue},
 };
+use serde::Serialize;
 use spin::{Mutex, Once};
 
 use super::{BlockDevice, id::Sid};
@@ -25,7 +26,7 @@ use crate::{BLOCK_SIZE, SECTOR_SIZE, prelude::*, request_queue::BioRequestSingle
 /// Trace data for block device I/O completion.
 ///
 /// This struct captures performance metrics when a block I/O request completes.
-#[derive(Clone)]
+#[derive(Clone, Copy, Serialize)]
 pub struct BlockDeviceCompletionStats {
     /// The latency of the I/O request (time from submission to completion).
     pub latency: Duration,

kernel/comps/time/Cargo.toml

@@ -11,6 +11,7 @@ aster-util = { path = "../../libs/aster-util" }
 component = { path = "../../libs/comp-sys/component" }
 log = "0.4"
 spin = "0.9.4"
+serde = { version = "1.0", default-features = false, features = ["derive", "alloc"]}
 
 [target.riscv64gc-unknown-none-elf.dependencies]
 chrono = { version = "0.4.38", default-features = false }

kernel/comps/time/src/clocksource.rs

@@ -15,6 +15,7 @@ use core::{cmp::max, ops::Add, time::Duration};
 
 use aster_util::coeff::Coeff;
 use ostd::sync::{LocalIrqDisabled, RwLock};
+use serde::Serialize;
 
 use crate::NANOS_PER_SECOND;
 
@@ -172,7 +173,7 @@ impl ClockSource {
 /// elapsed since a reference point (typically the system boot time).
 /// The [`Instant`] is expressed in seconds and the fractional part is
 /// expressed in nanoseconds.
-#[derive(Debug, Default, Copy, Clone)]
+#[derive(Debug, Default, Copy, Clone, Serialize)]
 pub struct Instant {
     secs: u64,
     nanos: u32,

kernel/src/event.rs

@@ -0,0 +1,48 @@
+// SPDX-License-Identifier: MPL-2.0
+
+use aster_time::Instant;
+use ostd::task::Task;
+use serde::Serialize;
+
+use crate::{process::posix_thread::AsPosixThread as _, thread::Tid};
+
+#[derive(Debug, Clone, Copy, Serialize)]
+pub enum TaskId {
+    KernelTask(usize),
+    PosixThread(Tid),
+    Unknown,
+}
+
+impl TaskId {
+    pub fn new(task: &Task) -> Self {
+        if let Some(t) = task.as_posix_thread() {
+            Self::PosixThread(t.tid())
+        } else {
+            Self::KernelTask(task.id().into())
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, Serialize)]
+pub struct EventContext {
+    pub task: TaskId,
+    pub timestamp: Instant,
+}
+
+impl EventContext {
+    /// Creates a new EventContext from the current context
+    pub fn new() -> Self {
+        EventContext {
+            task: Task::current()
+                .map(|t| TaskId::new(&t))
+                .unwrap_or(TaskId::Unknown),
+            timestamp: aster_time::read_monotonic_time().into(),
+        }
+    }
+}
+
+impl Default for EventContext {
+    fn default() -> Self {
+        Self::new()
+    }
+}

kernel/src/lib.rs

@@ -23,22 +23,37 @@
 #![feature(closure_track_caller)]
 #![register_tool(component_access_control)]
 
+use aster_block::bio::{BlockDeviceCompletionStats, SubmittedBio};
 use aster_framebuffer::FRAMEBUFFER_CONSOLE;
 #[cfg(not(baseline_asterinas))]
 mod data_capture;
+pub mod event;
+use aster_time::Instant;
 #[cfg(not(baseline_asterinas))]
 pub use data_capture::{new_data_capture_file, new_legacy_data_capture_file};
 use kcmdline::KCmdlineArg;
+#[cfg(not(baseline_asterinas))]
+use mariposa_data_capture::ObserverRegistration;
 use ostd::{
     arch::qemu::{QemuExitCode, exit_qemu},
     boot::boot_info,
     cpu::{CpuId, CpuSet},
+    ignore_err,
+    orpc::oqueue::registry::lookup_by_type,
+    task::scheduler::{SchedulingEvent, SchedulingEventKind},
+};
+#[cfg(not(baseline_asterinas))]
+use ostd::{
+    orpc::oqueue::{OQueueBase as _, ObservationQuery, registry::lookup_by_path},
+    path,
 };
 use process::{Process, spawn_init_process};
 use sched::SchedPolicy;
+use serde::Serialize;
 use spin::Once;
 
 use crate::{
+    event::{EventContext, TaskId},
     kcmdline::set_kernel_cmd_line,
     prelude::*,
     thread::kernel_thread::ThreadOptions,
@@ -225,6 +240,144 @@ fn init_thread() {
         pmu.start();
     }
 
+    #[cfg(not(baseline_asterinas))]
+    if karg
+        .get_module_arg_by_name::<bool>("scheduler", "capture_data")
+        .unwrap_or(false)
+    {
+        if let Some(oqueue) = lookup_by_path::<SchedulingEvent>(&path!(scheduler.events)) {
+            #[derive(Debug, Clone, Copy, Serialize)]
+            struct KernelSchedulingEvent {
+                timestamp: Instant,
+                task: TaskId,
+                kind: SchedulingEventKind,
+            }
+
+            let capture_file = new_data_capture_file::<KernelSchedulingEvent>(
+                mariposa_data_capture::FileDescriptor {
+                    path: path!(scheduler.events),
+                    length: 500 * 1024 * 1024,
+                },
+            );
+
+            ignore_err!(
+                capture_file.register_observer(ObserverRegistration {
+                    path: path!(scheduler.events),
+                    observer: oqueue
+                        .attach_strong_observer(ObservationQuery::new(|e: &SchedulingEvent| {
+                            let context = EventContext::new();
+                            KernelSchedulingEvent {
+                                timestamp: context.timestamp,
+                                kind: e.kind,
+                                task: TaskId::new(&e.task),
+                            }
+                        }))
+                        .unwrap(),
+                })
+            );
+            ignore_err!(capture_file.start());
+        } else {
+            error!("Could not find scheduler.events OQueue. Scheduler events will not be captured.")
+        }
+    }
+
+    if karg
+        .get_module_arg_by_name("io", "capture_block_io")
+        .unwrap_or(false)
+    {
+        {
+            // Setup submitted bio recording
+            let oqueues = lookup_by_type::<SubmittedBio>();
+            if !oqueues.is_empty() {
+                #[derive(Serialize, Clone, Copy)]
+                struct SubmittedBioEvent {
+                    byte_range: (usize, usize),
+                    timestamp: Option<Instant>,
+                    context: EventContext,
+                }
+
+                let capture_file = new_data_capture_file::<SubmittedBioEvent>(
+                    mariposa_data_capture::FileDescriptor {
+                        path: path!(io.block.submitted),
+                        length: 500 * 1024 * 1024,
+                    },
+                );
+
+                for oqueue in oqueues {
+                    let Some(path) = oqueue.path().cloned() else {
+                        log::warn!("Found anonymous SubmittedBio OQueue. Not capturing.");
+                        continue;
+                    };
+                    ignore_err!(
+                        capture_file.register_observer(ObserverRegistration {
+                            path,
+                            observer: oqueue
+                                .attach_strong_observer(ObservationQuery::new(
+                                    |e: &SubmittedBio| {
+                                        let sid_range = e.sid_range();
+                                        let context = EventContext::new();
+                                        SubmittedBioEvent {
+                                            byte_range: (
+                                                sid_range.start.to_offset(),
+                                                sid_range.end.to_offset(),
+                                            ),
+                                            timestamp: e.submission_time().map(|t| t.into()),
+                                            context,
+                                        }
+                                    },
+                                ))
+                                .unwrap(),
+                        })
+                    );
+                }
+                ignore_err!(capture_file.start());
+            }
+        }
+
+        {
+            // Setup submitted bio recording
+            let oqueues = lookup_by_type::<BlockDeviceCompletionStats>();
+            if !oqueues.is_empty() {
+                #[derive(Clone, Copy, Serialize)]
+                struct BlockDeviceCompletionEvent {
+                    stats: BlockDeviceCompletionStats,
+                    context: EventContext,
+                }
+
+                let capture_file = new_data_capture_file::<BlockDeviceCompletionEvent>(
+                    mariposa_data_capture::FileDescriptor {
+                        path: path!(io.block.completion),
+                        length: 500 * 1024 * 1024,
+                    },
+                );
+
+                for oqueue in oqueues {
+                    let Some(path) = oqueue.path().cloned() else {
+                        log::warn!(
+                            "Found anonymous BlockDeviceCompletionStats OQueue. Not capturing."
+                        );
+                        continue;
+                    };
+                    ignore_err!(
+                        capture_file.register_observer(ObserverRegistration {
+                            path,
+                            observer: oqueue
+                                .attach_strong_observer(ObservationQuery::new(|stats| {
+                                    let context = EventContext::new();
+                                    BlockDeviceCompletionEvent {
+                                        stats: *stats,
+                                        context,
+                                    }
+                                }))
+                                .unwrap(),
+                        })
+                    );
+                }
+                ignore_err!(capture_file.start());
+            }
+        }
+    }
+
     // Wait till initproc become zombie.
     while !initproc.status().is_zombie() {
         ostd::task::halt_cpu();

ostd/Cargo.toml

@@ -66,6 +66,7 @@ fdt = { version = "0.1.5", features = ["pretty-printing"] }
 default = ["cvm_guest"]
 capture_stacks = []
 track_mutex = []
+capture_scheduling = []
 # The guest OS support for Confidential VMs (CVMs), e.g., Intel TDX
 cvm_guest = ["dep:tdx-guest", "dep:iced-x86"]
 

ostd/src/lib.rs

@@ -138,6 +138,8 @@ unsafe fn init() {
 
     bus::init();
 
+    task::scheduler::init();
+
     arch::irq::enable_local();
 
     invoke_ffi_init_funcs();

ostd/src/orpc/oqueue/implementation.rs

@@ -37,6 +37,7 @@ use crate::{
             Cursor, InlineStrongObserver, OQueueError, ObservationQuery, ResourceUnavailableSnafu,
             single_thread_ring_buffer::RingBuffer,
         },
+        path::Path,
     },
     sync::{LocalIrqDisabled, SpinLock, WaitQueue, WakerKey},
 };
@@ -65,6 +66,7 @@ pub(crate) struct OQueueImplementation<T: ?Sized> {
     /// The size to use for the consumer and strong-observer ring-buffers.
     len: usize,
     supports_consume: bool,
+    path: Option<Path>,
     pub(super) put_wait_queue: WaitQueue,
     pub(super) read_wait_queue: WaitQueue,
 }
@@ -74,7 +76,9 @@ impl<T: ?Sized + 'static> OQueueImplementation<T> {
     ///
     /// * `len` is the ring buffer length used for consumers and strong-observers.
     /// * `supports_consume` specifies the attachment it allows later.
-    pub(crate) fn new(mut len: usize, supports_consume: bool) -> Self {
+    /// * `paths` are the paths associated with this OQueue; pass an empty `Vec` for anonymous
+    ///   queues.
+    pub(crate) fn new(mut len: usize, supports_consume: bool, path: Option<Path>) -> Self {
         if len < 2 {
             warn!(
                 "Creating an OQueue with length {len} is automatically increased to 2. Ring buffers smaller than 2 are not supported."
@@ -91,11 +95,17 @@ impl<T: ?Sized + 'static> OQueueImplementation<T> {
             }),
             len,
             supports_consume,
+            path,
             put_wait_queue: WaitQueue::new(),
             read_wait_queue: WaitQueue::new(),
         }
     }
 
+    /// Return the path associated with this OQueue, or `None` for anonymous queues.
+    pub(super) fn path(&self) -> Option<&Path> {
+        self.path.as_ref()
+    }
+
     /// Detach a consumer. This will free the consumer ring buffer if there are no consumers left.
     pub(super) fn detach_consumer(self: &Arc<Self>) {
         let mut inner = self.inner.lock();