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

io_stress/
workloads.rs
