io_stress/
workloads.rs

1// Copyright 2026 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 crate::stress::MemoryPressureStats;
6use anyhow::Context as _;
7use argh::FromArgs;
8use fidl_fuchsia_io as fio;
9use fuchsia_trace as trace;
10use fuchsiaperf::{Direction, FuchsiaPerfBenchmarkResult, Unit};
11use rand::RngExt as _;
12use std::fs::OpenOptions;
13use std::path::Path;
14use std::sync::Arc;
15use std::sync::atomic::{AtomicBool, Ordering};
16use std::thread;
17use std::time::{Duration, Instant};
18
19pub const KB: u64 = 1024;
20pub const MB: u64 = 1024 * KB;
21pub const PAGE_SIZE: u64 = 4096;
22
23pub const BURST_IO_PREALLOC_SIZE: u64 = 55 * MB;
24
25pub const NS_PER_MS: f64 = 1_000_000.0;
26
27pub const RANDOM_IO_WRITE_PATTERN: u8 = 0xBB;
28pub const BURST_IO_WRITE_PATTERN: u8 = 0xCC;
29
30#[derive(Default, Debug, Clone)]
31pub struct Metrics {
32    pub read_ops: u64,
33    pub write_ops: u64,
34    pub fsync_ops: u64,
35    pub read_bytes: u64,
36    pub write_bytes: u64,
37    pub max_read_latency_ns: u64,
38    pub max_write_latency_ns: u64,
39    pub max_fsync_latency_ns: u64,
40    pub total_write_latency_ns: u64,
41    pub total_read_latency_ns: u64,
42    pub total_fsync_latency_ns: u64,
43    pub op_latencies_ns: Vec<u64>,
44    pub fsync_latencies_ns: Vec<u64>,
45}
46
47impl Metrics {
48    pub fn record_read(&mut self, bytes_read: u64, elapsed_ns: u64) {
49        self.read_ops += 1;
50        self.read_bytes += bytes_read;
51        self.max_read_latency_ns = std::cmp::max(self.max_read_latency_ns, elapsed_ns);
52        self.total_read_latency_ns += elapsed_ns;
53        self.op_latencies_ns.push(elapsed_ns);
54    }
55
56    pub fn record_write(&mut self, bytes_written: u64, elapsed_ns: u64) {
57        self.write_ops += 1;
58        self.write_bytes += bytes_written;
59        self.max_write_latency_ns = std::cmp::max(self.max_write_latency_ns, elapsed_ns);
60        self.total_write_latency_ns += elapsed_ns;
61        self.op_latencies_ns.push(elapsed_ns);
62    }
63
64    pub fn record_fsync(&mut self, elapsed_ns: u64) {
65        self.fsync_ops += 1;
66        self.max_fsync_latency_ns = std::cmp::max(self.max_fsync_latency_ns, elapsed_ns);
67        self.total_fsync_latency_ns += elapsed_ns;
68        self.fsync_latencies_ns.push(elapsed_ns);
69    }
70
71    pub fn merge(&mut self, other: &Metrics) {
72        self.read_ops += other.read_ops;
73        self.write_ops += other.write_ops;
74        self.fsync_ops += other.fsync_ops;
75        self.read_bytes += other.read_bytes;
76        self.write_bytes += other.write_bytes;
77        self.max_read_latency_ns =
78            std::cmp::max(self.max_read_latency_ns, other.max_read_latency_ns);
79        self.max_write_latency_ns =
80            std::cmp::max(self.max_write_latency_ns, other.max_write_latency_ns);
81        self.max_fsync_latency_ns =
82            std::cmp::max(self.max_fsync_latency_ns, other.max_fsync_latency_ns);
83        self.total_write_latency_ns += other.total_write_latency_ns;
84        self.total_read_latency_ns += other.total_read_latency_ns;
85        self.total_fsync_latency_ns += other.total_fsync_latency_ns;
86        self.op_latencies_ns.extend_from_slice(&other.op_latencies_ns);
87        self.fsync_latencies_ns.extend_from_slice(&other.fsync_latencies_ns);
88    }
89}
90
91struct Timer {
92    start: zx::MonotonicInstant,
93}
94
95impl Timer {
96    fn start() -> Self {
97        Self { start: zx::MonotonicInstant::get() }
98    }
99
100    fn elapsed_ns(&self) -> u64 {
101        (zx::MonotonicInstant::get() - self.start).into_nanos() as u64
102    }
103}
104
105fn do_fsync(file_proxy: &fio::FileSynchronousProxy, metrics: &mut Metrics) -> anyhow::Result<()> {
106    let timer = Timer::start();
107    trace::duration!("benchmark", "fsync");
108    let deadline = zx::MonotonicInstant::after(zx::MonotonicDuration::from_seconds(30));
109    file_proxy.sync(deadline).context("FIDL error on fsync")?.map_err(|status| {
110        anyhow::anyhow!("fsync status: {:?}", zx::Status::err_from_raw(status))
111    })?;
112    metrics.record_fsync(timer.elapsed_ns());
113    Ok(())
114}
115
116struct RateLimiter {
117    start_run: Instant,
118    rate_mibs: u64,
119}
120
121impl RateLimiter {
122    fn new(rate_mibs: u64) -> Self {
123        Self { start_run: Instant::now(), rate_mibs }
124    }
125
126    fn sleep_if_needed(&mut self, total_bytes_processed: u64) {
127        if self.rate_mibs > 0 {
128            let target_duration = Duration::from_secs_f64(
129                total_bytes_processed as f64 / (self.rate_mibs * 1024 * 1024) as f64,
130            );
131            let elapsed = self.start_run.elapsed();
132            if elapsed < target_duration {
133                thread::sleep(target_duration - elapsed);
134            } else if elapsed > target_duration + Duration::from_millis(50) {
135                self.start_run = Instant::now() - target_duration;
136            }
137        }
138    }
139}
140
141#[allow(clippy::too_many_arguments)]
142pub fn run_random(
143    vmo: zx::Vmo,
144    file_proxy: &fio::FileSynchronousProxy,
145    op_size_bytes: usize,
146    file_size_bytes: u64,
147    read_percentage: u32,
148    fsync_every_n_ops: u64,
149    rate_mibs: u64,
150    seed: u64,
151    stop_signal: Arc<AtomicBool>,
152    metrics: &mut Metrics,
153) -> anyhow::Result<()> {
154    use rand::SeedableRng as _;
155    let mut rng = if seed == 0 {
156        rand::rngs::StdRng::from_rng(&mut rand::rng())
157    } else {
158        rand::rngs::StdRng::seed_from_u64(seed)
159    };
160    let mut op_count = 0;
161    let mut rate_limiter = RateLimiter::new(rate_mibs);
162    let mut total_bytes_processed = 0u64;
163    let mut io_buffer = vec![0u8; op_size_bytes];
164
165    while !stop_signal.load(Ordering::Relaxed) {
166        let range = file_size_bytes.saturating_sub(op_size_bytes as u64);
167        let max_page_index = range / PAGE_SIZE;
168        let mut offset =
169            if max_page_index == 0 { 0 } else { rng.random_range(0..=max_page_index) * PAGE_SIZE };
170        let is_read = rng.random_range(0..100) < read_percentage;
171        let buf_slice = &mut io_buffer[..op_size_bytes];
172
173        if offset + op_size_bytes as u64 > file_size_bytes {
174            offset = 0;
175        }
176
177        let timer = Timer::start();
178        if is_read {
179            trace::duration!(
180                "benchmark",
181                "vmo_read",
182                "bytes" => op_size_bytes as u64,
183                "offset" => offset
184            );
185            vmo.read(buf_slice, offset).context("VMO read failed")?;
186            metrics.record_read(op_size_bytes as u64, timer.elapsed_ns());
187        } else {
188            buf_slice.fill(RANDOM_IO_WRITE_PATTERN);
189            trace::duration!(
190                "benchmark",
191                "vmo_write",
192                "bytes" => op_size_bytes as u64,
193                "offset" => offset
194            );
195            vmo.write(buf_slice, offset).context("VMO write failed")?;
196            metrics.record_write(op_size_bytes as u64, timer.elapsed_ns());
197        }
198
199        let is_fsync = !is_read && fsync_every_n_ops > 0 && op_count % fsync_every_n_ops == 0;
200        if is_fsync {
201            do_fsync(file_proxy, metrics)?;
202        }
203
204        if !is_read {
205            op_count += 1;
206        }
207
208        total_bytes_processed += op_size_bytes as u64;
209        rate_limiter.sleep_if_needed(total_bytes_processed);
210    }
211
212    if fsync_every_n_ops > 0 {
213        do_fsync(file_proxy, metrics)?;
214    }
215
216    Ok(())
217}
218
219#[allow(clippy::too_many_arguments)]
220pub fn run_sequential(
221    zx_vmo: zx::Vmo,
222    file_proxy: &fio::FileSynchronousProxy,
223    op_size_bytes: usize,
224    file_size_bytes: u64,
225    rate_mibs: u64,
226    fsync_every_n_ops: u64,
227    read: bool,
228    stop_signal: Arc<AtomicBool>,
229    metrics: &mut Metrics,
230) -> anyhow::Result<()> {
231    let mut io_buffer = vec![0u8; op_size_bytes];
232    if !read {
233        io_buffer.fill(0x99);
234    }
235
236    let mut offset = 0;
237    let mut op_count = 0;
238    let mut rate_limiter = RateLimiter::new(rate_mibs);
239    let mut total_bytes_processed = 0u64;
240
241    while !stop_signal.load(Ordering::Relaxed) {
242        if offset + op_size_bytes as u64 > file_size_bytes {
243            if !read && fsync_every_n_ops == 0 {
244                do_fsync(file_proxy, metrics)?;
245            }
246            offset = 0;
247        }
248
249        let timer = Timer::start();
250        if read {
251            trace::duration!(
252                "benchmark",
253                "vmo_read",
254                "bytes" => op_size_bytes as u64,
255                "offset" => offset
256            );
257            zx_vmo.read(&mut io_buffer, offset).context("Sequential VMO read failed")?;
258            metrics.record_read(op_size_bytes as u64, timer.elapsed_ns());
259        } else {
260            trace::duration!(
261                "benchmark",
262                "vmo_write",
263                "bytes" => op_size_bytes as u64,
264                "offset" => offset
265            );
266            zx_vmo.write(&io_buffer, offset).context("Sequential VMO write failed")?;
267            metrics.record_write(op_size_bytes as u64, timer.elapsed_ns());
268        }
269
270        let is_fsync = !read && fsync_every_n_ops > 0 && op_count % fsync_every_n_ops == 0;
271        if is_fsync {
272            do_fsync(file_proxy, metrics)?;
273        }
274
275        if !read {
276            op_count += 1;
277        }
278
279        offset += op_size_bytes as u64;
280        total_bytes_processed += op_size_bytes as u64;
281        rate_limiter.sleep_if_needed(total_bytes_processed);
282    }
283
284    if !read {
285        do_fsync(file_proxy, metrics)?;
286    }
287
288    Ok(())
289}
290
291#[allow(clippy::too_many_arguments)]
292pub fn run_burst(
293    vmo: zx::Vmo,
294    file_proxy: &fio::FileSynchronousProxy,
295    op_size_bytes: usize,
296    burst_ops_count: usize,
297    sleep_between_bursts_ms: u64,
298    periodic_fsync_ms: u64,
299    read: bool,
300    rate_mibs: u64,
301    stop_signal: Arc<AtomicBool>,
302    metrics: &mut Metrics,
303) -> anyhow::Result<()> {
304    let mut io_buffer = vec![0u8; op_size_bytes];
305    let buf_slice = &mut io_buffer[..op_size_bytes];
306    if !read {
307        buf_slice.fill(BURST_IO_WRITE_PATTERN);
308    }
309    let mut offset = 0;
310    let mut last_fsync = Instant::now();
311    let mut rate_limiter = RateLimiter::new(rate_mibs);
312    let mut total_bytes_processed = 0u64;
313
314    while !stop_signal.load(Ordering::Relaxed) {
315        for _ in 0..burst_ops_count {
316            if offset + op_size_bytes as u64 > BURST_IO_PREALLOC_SIZE {
317                offset = 0;
318            }
319            let timer = Timer::start();
320            if read {
321                trace::duration!(
322                    "benchmark",
323                    "vmo_read",
324                    "bytes" => op_size_bytes as u64,
325                    "offset" => offset
326                );
327                vmo.read(buf_slice, offset).context("VMO burst read failed")?;
328                metrics.record_read(op_size_bytes as u64, timer.elapsed_ns());
329            } else {
330                trace::duration!(
331                    "benchmark",
332                    "vmo_write",
333                    "bytes" => op_size_bytes as u64,
334                    "offset" => offset
335                );
336                vmo.write(buf_slice, offset).context("VMO burst write failed")?;
337                metrics.record_write(op_size_bytes as u64, timer.elapsed_ns());
338            }
339            offset += op_size_bytes as u64;
340            total_bytes_processed += op_size_bytes as u64;
341        }
342
343        if !read {
344            let is_fsync = periodic_fsync_ms > 0
345                && last_fsync.elapsed().as_millis() >= periodic_fsync_ms as u128;
346            if is_fsync {
347                do_fsync(file_proxy, metrics)?;
348                last_fsync = Instant::now();
349            }
350        }
351
352        rate_limiter.sleep_if_needed(total_bytes_processed);
353        thread::sleep(Duration::from_millis(sleep_between_bursts_ms));
354    }
355
356    if !read {
357        do_fsync(file_proxy, metrics)?;
358    }
359
360    Ok(())
361}
362
363#[allow(clippy::too_many_arguments)]
364pub fn run_transfer(
365    source_vmo: zx::Vmo,
366    dest_vmo: zx::Vmo,
367    dest_proxy: &fio::FileSynchronousProxy,
368    op_size_bytes: usize,
369    file_size_bytes: u64,
370    xor_transform: bool,
371    rate_mibs: u64,
372    fsync_every_n_ops: u64,
373    stop_signal: Arc<AtomicBool>,
374    metrics: &mut Metrics,
375) -> anyhow::Result<()> {
376    let mut op_count = 0;
377    let mut rate_limiter = RateLimiter::new(rate_mibs);
378    let mut total_bytes_processed = 0u64;
379    let mut io_buffer = vec![0u8; op_size_bytes];
380
381    while !stop_signal.load(Ordering::Relaxed) {
382        let offset = (op_count as u64 * op_size_bytes as u64) % file_size_bytes;
383
384        let read_timer = Timer::start();
385        source_vmo.read(&mut io_buffer, offset).context("Failed to read from source VMO")?;
386        metrics.record_read(op_size_bytes as u64, read_timer.elapsed_ns());
387
388        if xor_transform {
389            for b in &mut io_buffer {
390                *b ^= 0x55;
391            }
392        }
393
394        let write_timer = Timer::start();
395        dest_vmo.write(&io_buffer, offset).context("Failed to write to dest VMO")?;
396        metrics.record_write(op_size_bytes as u64, write_timer.elapsed_ns());
397
398        op_count += 1;
399        total_bytes_processed += op_size_bytes as u64;
400        rate_limiter.sleep_if_needed(total_bytes_processed);
401
402        if fsync_every_n_ops > 0 && op_count % fsync_every_n_ops == 0 {
403            do_fsync(dest_proxy, metrics)?;
404        }
405    }
406
407    if fsync_every_n_ops > 0 {
408        do_fsync(dest_proxy, metrics)?;
409    }
410
411    Ok(())
412}
413
414#[derive(FromArgs, Debug, Clone)]
415#[argh(subcommand, name = "random")]
416/// Run a random write/read workload.
417pub struct RandomArgs {
418    /// block size in bytes (default: 4096)
419    #[argh(option, default = "4096")]
420    pub op_size_bytes: u64,
421
422    /// total backing file size in bytes (default: 67108864)
423    #[argh(option, default = "67108864")]
424    pub file_size_bytes: u64,
425
426    /// percentage of read operations (0 to 100)
427    #[argh(option, default = "50")]
428    pub read_percentage: u32,
429
430    /// trigger fsync every N operations (default: 0)
431    #[argh(option, default = "0")]
432    pub fsync_every_n_ops: u64,
433
434    /// rate limit operations in MiB/s (default: 0)
435    #[argh(option, default = "0")]
436    pub rate_mibs: u64,
437
438    /// RNG seed for reproducibility (default: 0)
439    #[argh(option, default = "0")]
440    pub seed: u64,
441}
442
443#[derive(FromArgs, Debug, Clone)]
444#[argh(subcommand, name = "sequential")]
445/// Run a sequential write workload.
446pub struct SequentialArgs {
447    /// block size in bytes (default: 131072)
448    #[argh(option, default = "131072")]
449    pub op_size_bytes: u64,
450
451    /// total backing file size in bytes (default: 1073741824)
452    #[argh(option, default = "1073741824")]
453    pub file_size_bytes: u64,
454
455    /// perform sequential read instead of write
456    #[argh(switch)]
457    pub read: bool,
458
459    /// trigger fsync every N operations (default: 0)
460    #[argh(option, default = "0")]
461    pub fsync_every_n_ops: u64,
462
463    /// rate limit operations in MiB/s (default: 0)
464    #[argh(option, default = "0")]
465    pub rate_mibs: u64,
466}
467
468#[derive(FromArgs, Debug, Clone)]
469#[argh(subcommand, name = "burst")]
470/// Run a burst workload.
471pub struct BurstArgs {
472    /// block size in bytes (default: 4096)
473    #[argh(option, default = "4096")]
474    pub op_size_bytes: u64,
475
476    /// number of operations in a single burst (default: 1000)
477    #[argh(option, default = "1000")]
478    pub burst_ops_count: u64,
479
480    /// sleep duration between bursts in ms (default: 100)
481    #[argh(option, default = "100")]
482    pub sleep_between_bursts_ms: u64,
483
484    /// periodic fsync interval in ms (default: 1000)
485    #[argh(option, default = "1000")]
486    pub periodic_fsync_ms: u64,
487
488    /// perform burst read instead of write
489    #[argh(switch)]
490    pub read: bool,
491
492    /// rate limit operations in MiB/s (default: 0)
493    #[argh(option, default = "0")]
494    pub rate_mibs: u64,
495}
496
497#[derive(FromArgs, Debug, Clone)]
498#[argh(subcommand, name = "transfer")]
499/// Run a dual-file transfer (copy or compress).
500pub struct TransferArgs {
501    /// block size in bytes (default: 131072)
502    #[argh(option, default = "131072")]
503    pub op_size_bytes: u64,
504
505    /// total backing file size in bytes (default: 67108864)
506    #[argh(option, default = "67108864")]
507    pub file_size_bytes: u64,
508
509    /// perform CPU XOR transformation on data before writing
510    #[argh(switch)]
511    pub xor_transform: bool,
512
513    /// trigger fsync every N operations (default: 0)
514    #[argh(option, default = "0")]
515    pub fsync_every_n_ops: u64,
516
517    /// rate limit operations in MiB/s (default: 0)
518    #[argh(option, default = "0")]
519    pub rate_mibs: u64,
520}
521
522#[derive(FromArgs, Debug, Clone)]
523#[argh(subcommand)]
524pub enum WorkloadSubcommand {
525    Random(RandomArgs),
526    Sequential(SequentialArgs),
527    Burst(BurstArgs),
528    Transfer(TransferArgs),
529}
530
531impl WorkloadSubcommand {
532    pub fn name(&self) -> &'static str {
533        match self {
534            WorkloadSubcommand::Random(_) => "random",
535            WorkloadSubcommand::Sequential(_) => "sequential",
536            WorkloadSubcommand::Burst(_) => "burst",
537            WorkloadSubcommand::Transfer(_) => "transfer",
538        }
539    }
540
541    pub fn persona(&self) -> &'static str {
542        match self {
543            WorkloadSubcommand::Random(args) => {
544                if args.rate_mibs == 0 {
545                    "AppLaunch"
546                } else {
547                    "Database"
548                }
549            }
550            WorkloadSubcommand::Sequential(args) => {
551                if args.fsync_every_n_ops > 0 {
552                    "Media"
553                } else {
554                    "Download"
555                }
556            }
557            WorkloadSubcommand::Transfer(args) => {
558                if args.xor_transform {
559                    "Compress"
560                } else {
561                    "Copy"
562                }
563            }
564            WorkloadSubcommand::Burst(_) => "Media",
565        }
566    }
567}
568
569pub fn get_effective_op_size(op_size_bytes: u64) -> u64 {
570    if op_size_bytes == 0 { PAGE_SIZE } else { op_size_bytes }
571}
572
573pub fn setup_backing_vmo(
574    file_path: &Path,
575    size_bytes: u64,
576) -> anyhow::Result<(fio::FileSynchronousProxy, zx::Vmo)> {
577    let file = OpenOptions::new()
578        .read(true)
579        .write(true)
580        .create(true)
581        .truncate(true)
582        .open(file_path)
583        .context("Failed to open or create target file under /data")?;
584    if size_bytes > 0 {
585        file.set_len(size_bytes).context("Failed to set file size limit")?;
586    }
587
588    let zx_channel = fdio::clone_channel(&file).context("Failed to clone fdio channel")?;
589    let file_proxy = fio::FileSynchronousProxy::new(zx_channel);
590    let vmo_flags = fio::VmoFlags::READ | fio::VmoFlags::WRITE | fio::VmoFlags::SHARED_BUFFER;
591    let deadline = zx::MonotonicInstant::after(zx::MonotonicDuration::from_seconds(30));
592    let vmo_handle = file_proxy
593        .get_backing_memory(vmo_flags, deadline)
594        .context("Failed to get backing memory via FIDL")?
595        .map_err(|status| {
596            anyhow::anyhow!("get_backing_memory status: {:?}", zx::Status::err_from_raw(status))
597        })?;
598    Ok((file_proxy, zx::Vmo::from(vmo_handle)))
599}
600
601pub struct StopSignalGuard(pub Arc<AtomicBool>);
602impl Drop for StopSignalGuard {
603    fn drop(&mut self) {
604        self.0.store(true, Ordering::Relaxed);
605    }
606}
607
608pub struct FileCleanupGuard {
609    pub paths: Vec<String>,
610}
611impl Drop for FileCleanupGuard {
612    fn drop(&mut self) {
613        for path in &self.paths {
614            let _ = std::fs::remove_file(path);
615        }
616    }
617}
618
619fn calc_percentile(mut vals: Vec<u64>, pct: f64) -> f64 {
620    if vals.is_empty() {
621        return 0.0;
622    }
623    vals.sort_unstable();
624    let idx = ((vals.len() as f64 - 1.0) * pct).round() as usize;
625    vals[std::cmp::min(idx, vals.len() - 1)] as f64
626}
627
628pub async fn run_workloads(
629    run_name: &str,
630    subcommands: Vec<WorkloadSubcommand>,
631    duration_secs: u64,
632    mem_stats: Arc<MemoryPressureStats>,
633) -> anyhow::Result<Vec<FuchsiaPerfBenchmarkResult>> {
634    let stop_signal = Arc::new(AtomicBool::new(false));
635    let _stop_guard = StopSignalGuard(stop_signal.clone());
636    let mut created_file_paths = Vec::new();
637
638    let _timer_task = if duration_secs > 0 {
639        let stop_signal_clone = stop_signal.clone();
640        Some(fuchsia_async::Task::spawn(async move {
641            fuchsia_async::Timer::new(Duration::from_secs(duration_secs)).await;
642            log::info!("Configured global duration reached, signalling stop to all threads.");
643            stop_signal_clone.store(true, Ordering::Relaxed);
644        }))
645    } else {
646        None
647    };
648
649    let mut thread_handles = Vec::new();
650    let mut persona_counters = std::collections::HashMap::new();
651
652    for sub in subcommands {
653        let sub_clone = sub.clone();
654        let stop_signal = stop_signal.clone();
655        let persona = sub.persona();
656
657        let entry = persona_counters.entry(persona).or_insert(0);
658        *entry += 1;
659        let display_name = format!("{}_{}", persona, entry);
660
661        let handle = match &sub_clone {
662            WorkloadSubcommand::Transfer(args) => {
663                let src_path_str = format!("/data/stress_target_{}_src", display_name);
664                let dest_path_str = format!("/data/stress_target_{}_dest", display_name);
665                created_file_paths.push(src_path_str.clone());
666                created_file_paths.push(dest_path_str.clone());
667
668                let args_clone = args.clone();
669                thread::Builder::new()
670                    .name(display_name.clone())
671                    .spawn(move || -> anyhow::Result<Metrics> {
672                        let mut metrics = Metrics::default();
673                        let op_size = get_effective_op_size(args_clone.op_size_bytes);
674                        let (_src_proxy, src_vmo) = setup_backing_vmo(
675                            Path::new(&src_path_str),
676                            args_clone.file_size_bytes,
677                        )?;
678                        let (dest_proxy, dest_vmo) = setup_backing_vmo(
679                            Path::new(&dest_path_str),
680                            args_clone.file_size_bytes,
681                        )?;
682
683                        run_transfer(
684                            src_vmo,
685                            dest_vmo,
686                            &dest_proxy,
687                            op_size as usize,
688                            args_clone.file_size_bytes,
689                            args_clone.xor_transform,
690                            args_clone.rate_mibs,
691                            args_clone.fsync_every_n_ops,
692                            stop_signal,
693                            &mut metrics,
694                        )?;
695                        Ok(metrics)
696                    })
697                    .context("Failed to spawn thread")?
698            }
699            WorkloadSubcommand::Random(args) => {
700                let file_path_str = format!("/data/stress_target_{}", display_name);
701                created_file_paths.push(file_path_str.clone());
702
703                let args_clone = args.clone();
704                thread::Builder::new()
705                    .name(display_name.clone())
706                    .spawn(move || -> anyhow::Result<Metrics> {
707                        let mut metrics = Metrics::default();
708                        let file_path = Path::new(&file_path_str);
709                        let op_size = get_effective_op_size(args_clone.op_size_bytes);
710                        let (file_proxy, zx_vmo) =
711                            setup_backing_vmo(&file_path, args_clone.file_size_bytes)?;
712                        run_random(
713                            zx_vmo,
714                            &file_proxy,
715                            op_size as usize,
716                            args_clone.file_size_bytes,
717                            args_clone.read_percentage,
718                            args_clone.fsync_every_n_ops,
719                            args_clone.rate_mibs,
720                            args_clone.seed,
721                            stop_signal,
722                            &mut metrics,
723                        )?;
724                        Ok(metrics)
725                    })
726                    .context("Failed to spawn thread")?
727            }
728            WorkloadSubcommand::Sequential(args) => {
729                let file_path_str = format!("/data/stress_target_{}", display_name);
730                created_file_paths.push(file_path_str.clone());
731
732                let args_clone = args.clone();
733                thread::Builder::new()
734                    .name(display_name.clone())
735                    .spawn(move || -> anyhow::Result<Metrics> {
736                        let mut metrics = Metrics::default();
737                        let file_path = Path::new(&file_path_str);
738                        let op_size = get_effective_op_size(args_clone.op_size_bytes);
739                        let (file_proxy, zx_vmo) =
740                            setup_backing_vmo(&file_path, args_clone.file_size_bytes)?;
741                        run_sequential(
742                            zx_vmo,
743                            &file_proxy,
744                            op_size as usize,
745                            args_clone.file_size_bytes,
746                            args_clone.rate_mibs,
747                            args_clone.fsync_every_n_ops,
748                            args_clone.read,
749                            stop_signal,
750                            &mut metrics,
751                        )?;
752                        Ok(metrics)
753                    })
754                    .context("Failed to spawn thread")?
755            }
756            WorkloadSubcommand::Burst(args) => {
757                let file_path_str = format!("/data/stress_target_{}", display_name);
758                created_file_paths.push(file_path_str.clone());
759
760                let args_clone = args.clone();
761                thread::Builder::new()
762                    .name(display_name.clone())
763                    .spawn(move || -> anyhow::Result<Metrics> {
764                        let mut metrics = Metrics::default();
765                        let file_path = Path::new(&file_path_str);
766                        let op_size = get_effective_op_size(args_clone.op_size_bytes);
767                        let (file_proxy, zx_vmo) =
768                            setup_backing_vmo(&file_path, BURST_IO_PREALLOC_SIZE)?;
769                        run_burst(
770                            zx_vmo,
771                            &file_proxy,
772                            op_size as usize,
773                            args_clone.burst_ops_count as usize,
774                            args_clone.sleep_between_bursts_ms,
775                            args_clone.periodic_fsync_ms,
776                            args_clone.read,
777                            args_clone.rate_mibs,
778                            stop_signal,
779                            &mut metrics,
780                        )?;
781                        Ok(metrics)
782                    })
783                    .context("Failed to spawn thread")?
784            }
785        };
786
787        thread_handles.push((persona, handle));
788    }
789
790    let _file_guard = FileCleanupGuard { paths: created_file_paths };
791
792    let start_time = Instant::now();
793    let mut results: Vec<(&'static str, Metrics)> = Vec::new();
794    let mut first_error = None;
795
796    for (persona, handle) in thread_handles {
797        let join_result = fuchsia_async::unblock(move || handle.join()).await;
798        match join_result {
799            Ok(Ok(metrics)) => results.push((persona, metrics)),
800            Ok(Err(e)) => {
801                stop_signal.store(true, Ordering::Relaxed);
802                if first_error.is_none() {
803                    first_error = Some(e.context(format!("Workload '{}' returned error", persona)));
804                }
805            }
806            Err(_) => {
807                stop_signal.store(true, Ordering::Relaxed);
808                if first_error.is_none() {
809                    first_error = Some(anyhow::anyhow!("Workload '{}' panicked", persona));
810                }
811            }
812        }
813    }
814    if let Some(err) = first_error {
815        return Err(err);
816    }
817
818    let elapsed_secs = start_time.elapsed().as_secs_f64();
819
820    let mut persona_map: std::collections::BTreeMap<&'static str, Metrics> =
821        std::collections::BTreeMap::new();
822    for (persona, m) in results {
823        let entry = persona_map.entry(persona).or_default();
824        entry.merge(&m);
825    }
826
827    let mut total_bytes = 0;
828    let mut perf_results = Vec::new();
829
830    for (persona, m) in &persona_map {
831        total_bytes += m.write_bytes + m.read_bytes;
832        let p95_op = calc_percentile(m.op_latencies_ns.clone(), 0.95);
833        let p99_op = calc_percentile(m.op_latencies_ns.clone(), 0.99);
834        let throughput = if elapsed_secs > 0.0 {
835            (m.write_bytes + m.read_bytes) as f64 / elapsed_secs
836        } else {
837            0.0
838        };
839
840        perf_results.push(FuchsiaPerfBenchmarkResult {
841            label: format!("{}/{}/p95_op_latency", run_name, persona),
842            test_suite: "fuchsia.io_stress".to_string(),
843            unit: Unit::Nanoseconds,
844            direction: Direction::SmallerBetter,
845            values: vec![p95_op],
846        });
847        perf_results.push(FuchsiaPerfBenchmarkResult {
848            label: format!("{}/{}/p99_op_latency", run_name, persona),
849            test_suite: "fuchsia.io_stress".to_string(),
850            unit: Unit::Nanoseconds,
851            direction: Direction::SmallerBetter,
852            values: vec![p99_op],
853        });
854        perf_results.push(FuchsiaPerfBenchmarkResult {
855            label: format!("{}/{}/throughput", run_name, persona),
856            test_suite: "fuchsia.io_stress".to_string(),
857            unit: Unit::BytesPerSecond,
858            direction: Direction::BiggerBetter,
859            values: vec![throughput],
860        });
861        if m.fsync_ops > 0 {
862            let p95_fsync = calc_percentile(m.fsync_latencies_ns.clone(), 0.95);
863            let p99_fsync = calc_percentile(m.fsync_latencies_ns.clone(), 0.99);
864            perf_results.push(FuchsiaPerfBenchmarkResult {
865                label: format!("{}/{}/p95_fsync_latency", run_name, persona),
866                test_suite: "fuchsia.io_stress".to_string(),
867                unit: Unit::Nanoseconds,
868                direction: Direction::SmallerBetter,
869                values: vec![p95_fsync],
870            });
871            perf_results.push(FuchsiaPerfBenchmarkResult {
872                label: format!("{}/{}/p99_fsync_latency", run_name, persona),
873                test_suite: "fuchsia.io_stress".to_string(),
874                unit: Unit::Nanoseconds,
875                direction: Direction::SmallerBetter,
876                values: vec![p99_fsync],
877            });
878        }
879    }
880
881    let total_throughput_bps =
882        if elapsed_secs > 0.0 { total_bytes as f64 / elapsed_secs } else { 0.0 };
883    let mem_score = match mem_stats.max_level() {
884        fidl_fuchsia_memorypressure::Level::Normal => 0.0,
885        fidl_fuchsia_memorypressure::Level::Warning => 1.0,
886        fidl_fuchsia_memorypressure::Level::Critical => 2.0,
887    };
888
889    perf_results.push(FuchsiaPerfBenchmarkResult {
890        label: format!("{}/total_throughput", run_name),
891        test_suite: "fuchsia.io_stress".to_string(),
892        unit: Unit::BytesPerSecond,
893        direction: Direction::BiggerBetter,
894        values: vec![total_throughput_bps],
895    });
896    perf_results.push(FuchsiaPerfBenchmarkResult {
897        label: format!("{}/max_memory_pressure_score", run_name),
898        test_suite: "fuchsia.io_stress".to_string(),
899        unit: Unit::Count,
900        direction: Direction::SmallerBetter,
901        values: vec![mem_score],
902    });
903
904    perf_results.sort_by(|a, b| a.label.cmp(&b.label));
905
906    Ok(perf_results)
907}
io_stress/workloads.rs

io_stress/
workloads.rs
