Optimize: Threaded record writer configs

Author: arjunr2

crates/cli-flags/src/lib.rs

@@ -508,6 +508,15 @@ wasmtime_option_group! {
         pub validation_metadata: Option<bool>,
         /// Writer buffer size (in bytes) for internal buffering during recording. Defaults to 8 KiB.
         pub buffer_size: Option<usize>,
+        /// Enable threaded (async) recording writer. Decouples serialization from file I/O
+        /// by shipping buffers to a background thread.
+        pub threaded: Option<bool>,
+        /// Buffer capacity (in bytes) for the threaded writer before shipping to the
+        /// background thread. Defaults to 64 KiB. Only used when `threaded` is enabled.
+        pub threaded_buffer_capacity: Option<usize>,
+        /// Channel bound for the threaded writer (max number of buffers queued).
+        /// Defaults to 8. Only used when `threaded` is enabled.
+        pub threaded_channel_bound: Option<usize>,
     }
 
     enum Record {

crates/wasmtime/src/runtime.rs

@@ -95,6 +95,8 @@ pub use resources::*;
 pub use rr::{
     RecordSettings, RecordWriter, ReplayEnvironment, ReplayInstance, ReplayReader, ReplaySettings,
 };
+#[cfg(all(feature = "rr", feature = "std"))]
+pub use rr::{ThreadedWriter, ThreadedWriterConfig};
 #[cfg(all(feature = "async", feature = "call-hook"))]
 pub use store::CallHookHandler;
 pub use store::{

crates/wasmtime/src/runtime/rr.rs

@@ -66,6 +66,12 @@ mod backend;
 #[cfg(feature = "rr")]
 pub use backend::*;
 
+/// Threaded writer for async file I/O during recording (requires `std` for threads and channels)
+#[cfg(all(feature = "rr", feature = "std"))]
+mod threaded_writer;
+#[cfg(all(feature = "rr", feature = "std"))]
+pub use threaded_writer::*;
+
 /// Driver capabilities for executing replays
 #[cfg(feature = "rr")]
 mod replay_driver;

crates/wasmtime/src/runtime/rr/backend.rs

@@ -583,11 +583,8 @@ mod tests {
     #[test]
     fn lower_flat_events() -> Result<()> {
         use component_events::{LowerFlatEntryEvent, LowerFlatReturnEvent};
-        use wasmtime_environ::component::InterfaceType;
 
-        let entry = LowerFlatEntryEvent {
-            ty: InterfaceType::U32,
-        };
+        let entry = LowerFlatEntryEvent;
         let return_event = LowerFlatReturnEvent(ResultEvent::from_anyhow_result(&Ok(())));
 
         rr_harness(
@@ -597,10 +594,7 @@ mod tests {
                 Ok(())
             },
             |replayer| {
-                replayer.next_event_and(|e: LowerFlatEntryEvent| {
-                    assert_eq!(e.ty, InterfaceType::U32);
-                    Ok(())
-                })?;
+                replayer.next_event_and(|e: LowerFlatEntryEvent| Ok(()))?;
                 replayer.next_event_and(|e: LowerFlatReturnEvent| {
                     assert!(e.0.ret().is_ok());
                     Ok(())
@@ -613,12 +607,8 @@ mod tests {
     #[test]
     fn lower_memory_events() -> Result<()> {
         use component_events::{LowerMemoryEntryEvent, LowerMemoryReturnEvent};
-        use wasmtime_environ::component::InterfaceType;
 
-        let entry = LowerMemoryEntryEvent {
-            ty: InterfaceType::String,
-            offset: 1024,
-        };
+        let entry = LowerMemoryEntryEvent { offset: 1024 };
         let return_event = LowerMemoryReturnEvent(ResultEvent::from_anyhow_result(&Ok(())));
 
         rr_harness(
@@ -629,7 +619,6 @@ mod tests {
             },
             |replayer| {
                 replayer.next_event_and(|e: LowerMemoryEntryEvent| {
-                    assert_eq!(e.ty, InterfaceType::String);
                     assert_eq!(e.offset, 1024);
                     Ok(())
                 })?;