turbo_tasks_fuzz/
symlink_stress.rs

1use std::{
2    path::{Path, PathBuf},
3    time::{Duration, Instant},
4};
5
6const PROGRESS_INTERVAL: Duration = Duration::from_secs(1);
7
8use clap::Args;
9use rand::{RngExt, SeedableRng};
10use turbo_rcstr::{RcStr, rcstr};
11use turbo_tasks::{
12    Effects, OperationVc, ResolvedVc, TryJoinIterExt, Vc,
13    read_strongly_consistent_and_apply_effects, take_effects,
14};
15use turbo_tasks_backend::{BackendOptions, TurboTasksBackend, noop_backing_storage};
16use turbo_tasks_fs::{DiskFileSystem, FileSystem, FileSystemPath, LinkContent, LinkType};
17
18#[derive(Args)]
19pub struct SymlinkStress {
20    #[arg(long)]
21    fs_root: PathBuf,
22    /// Number of target directories symlinks can point to.
23    #[arg(long, default_value_t = 20)]
24    target_count: usize,
25    /// Number of symlinks to create and update.
26    #[arg(long, default_value_t = 50)]
27    symlink_count: usize,
28    /// Number of symlink writes to perform in parallel.
29    #[arg(long, default_value_t = 16)]
30    parallelism: usize,
31    /// How long to run the stress test for.
32    #[arg(long, default_value_t = 5)]
33    duration_secs: u64,
34}
35
36#[turbo_tasks::function(operation, root)]
37async fn extract_effects_operation(op: OperationVc<()>) -> anyhow::Result<Vc<Effects>> {
38    let _ = op.resolve().strongly_consistent().await?;
39    Ok(take_effects(op).await?.cell())
40}
41
42pub async fn run(args: SymlinkStress) -> anyhow::Result<()> {
43    std::fs::create_dir(&args.fs_root)?;
44    let fs_root = args.fs_root.canonicalize()?;
45    let _guard = FsCleanup {
46        path: &fs_root.clone(),
47    };
48
49    // Create target directories that symlinks will point to
50    let targets_dir = fs_root.join("_targets");
51    std::fs::create_dir(&targets_dir)?;
52    for i in 0..args.target_count {
53        std::fs::create_dir(targets_dir.join(i.to_string()))?;
54    }
55
56    // Create symlinks directory
57    let symlinks_dir = fs_root.join("_symlinks");
58    std::fs::create_dir(&symlinks_dir)?;
59
60    let tt = turbo_tasks::TurboTasks::new(TurboTasksBackend::new(
61        BackendOptions::default(),
62        noop_backing_storage(),
63    ));
64
65    let target_count = args.target_count;
66    let symlink_count = args.symlink_count;
67    let parallelism = args.parallelism;
68    let duration = Duration::from_secs(args.duration_secs);
69
70    tt.run_once(async move {
71        let project_fs = disk_file_system_operation(RcStr::from(fs_root.to_str().unwrap()))
72            .resolve()
73            .strongly_consistent()
74            .await?;
75        let project_root = disk_file_system_root_operation(project_fs)
76            .resolve()
77            .strongly_consistent()
78            .await?
79            .owned()
80            .await?;
81
82        // Create initial symlinks via turbo-tasks, all pointing to target 0
83        let symlinks_path = project_root.join("_symlinks")?;
84        let initial_target = RcStr::from("../_targets/0");
85
86        println!("creating {symlink_count} initial symlinks...");
87
88        read_strongly_consistent_and_apply_effects(
89            extract_effects_operation(create_initial_symlinks_operation(
90                symlinks_path.clone(),
91                symlink_count,
92                initial_target,
93            )),
94            |e| e,
95        )
96        .await?;
97
98        println!(
99            "starting stress test with parallelism={} for {}s...",
100            parallelism,
101            duration.as_secs()
102        );
103
104        let mut rng = rand::rngs::SmallRng::from_rng(&mut rand::rng());
105        let mut total_writes: u64 = 0;
106        let mut last_progress_writes: u64 = 0;
107        let start_time = Instant::now();
108        let mut last_progress_time = start_time;
109
110        loop {
111            // Check if we've reached the duration limit
112            if start_time.elapsed() >= duration {
113                break;
114            }
115
116            // Generate random symlink updates for this batch
117            let updates: Vec<(usize, usize)> = (0..parallelism)
118                .map(|_| {
119                    let symlink_idx = rng.random_range(0..symlink_count);
120                    let target_idx = rng.random_range(0..target_count);
121                    (symlink_idx, target_idx)
122                })
123                .collect();
124
125            // Execute writes in parallel via turbo-tasks
126            read_strongly_consistent_and_apply_effects(
127                extract_effects_operation(write_symlinks_batch_operation(
128                    symlinks_path.clone(),
129                    updates,
130                )),
131                |e| e,
132            )
133            .await?;
134
135            total_writes += parallelism as u64;
136
137            // Print progress every PROGRESS_INTERVAL
138            let now = Instant::now();
139            if now.duration_since(last_progress_time) >= PROGRESS_INTERVAL {
140                let interval_writes = total_writes - last_progress_writes;
141                let interval_duration = now.duration_since(last_progress_time);
142                let writes_per_sec = interval_writes as f64 / interval_duration.as_secs_f64();
143                println!(
144                    "{:.1}s: {} writes, {:.0} writes/sec",
145                    start_time.elapsed().as_secs_f64(),
146                    total_writes,
147                    writes_per_sec
148                );
149                last_progress_time = now;
150                last_progress_writes = total_writes;
151            }
152        }
153
154        // Final summary
155        let elapsed = start_time.elapsed();
156        let writes_per_sec = total_writes as f64 / elapsed.as_secs_f64();
157        println!(
158            "completed {} symlink writes in {:.2}s ({:.0} writes/sec)",
159            total_writes,
160            elapsed.as_secs_f64(),
161            writes_per_sec
162        );
163
164        Ok(())
165    })
166    .await?;
167
168    tt.stop_and_wait().await;
169    Ok(())
170}
171
172#[turbo_tasks::function(operation)]
173fn disk_file_system_operation(fs_root: RcStr) -> Vc<DiskFileSystem> {
174    DiskFileSystem::new(rcstr!("project"), Vc::cell(fs_root))
175}
176
177#[turbo_tasks::function(operation)]
178fn disk_file_system_root_operation(fs: ResolvedVc<DiskFileSystem>) -> Vc<FileSystemPath> {
179    fs.root()
180}
181
182#[turbo_tasks::function(operation)]
183async fn create_initial_symlinks_operation(
184    symlinks_dir: FileSystemPath,
185    count: usize,
186    target: RcStr,
187) -> anyhow::Result<()> {
188    (0..count)
189        .map(|i| write_symlink(symlinks_dir.clone(), i, target.clone()))
190        .try_join()
191        .await?;
192    Ok(())
193}
194
195#[turbo_tasks::function(operation)]
196async fn write_symlinks_batch_operation(
197    symlinks_dir: FileSystemPath,
198    updates: Vec<(usize, usize)>,
199) -> anyhow::Result<()> {
200    updates
201        .into_iter()
202        .map(|(symlink_idx, target_idx)| {
203            let target = RcStr::from(format!("../_targets/{}", target_idx));
204            write_symlink(symlinks_dir.clone(), symlink_idx, target)
205        })
206        .try_join()
207        .await?;
208    Ok(())
209}
210
211#[turbo_tasks::function]
212async fn write_symlink(
213    symlinks_dir: FileSystemPath,
214    symlink_idx: usize,
215    target: RcStr,
216) -> anyhow::Result<()> {
217    let symlink_path = symlinks_dir.join(&symlink_idx.to_string())?;
218    let link_content = LinkContent::Link {
219        target,
220        link_type: LinkType::DIRECTORY,
221    };
222    symlink_path
223        .fs()
224        .write_link(symlink_path.clone(), link_content.cell())
225        .await?;
226    Ok(())
227}
228
229struct FsCleanup<'a> {
230    path: &'a Path,
231}
232
233impl Drop for FsCleanup<'_> {
234    fn drop(&mut self) {
235        std::fs::remove_dir_all(self.path).unwrap();
236    }
237}
turbo_tasks_fuzz/symlink_stress.rs

turbo_tasks_fuzz/
symlink_stress.rs
