cargo-fuzztest: support parallel fuzzing jobs.

rust/cargo_fuzztest/src/lib.rs

@@ -227,7 +227,7 @@ impl FuzztestRunner {
             }
 
             ExecutionMode::Fuzz(fuzz_options) => {
-                let FuzzOptions { fuzz_for, jobs: _ } = fuzz_options;
+                let FuzzOptions { fuzz_for, jobs } = fuzz_options;
                 match fuzz_for {
                     FuzzFor::Indefinitely => {
                         cmd.env("FUZZTEST_FUZZ_FOR", "inf");
@@ -236,7 +236,9 @@ impl FuzztestRunner {
                         cmd.env("FUZZTEST_FUZZ_FOR", duration.to_string());
                     }
                 }
-                // TODO(the-shank): support parallel jobs
+                if let Some(jobs) = jobs {
+                    cmd.env("FUZZTEST_JOBS", jobs.to_string());
+                }
             }
 
             ExecutionMode::SmokeTest => {

rust/cargo_fuzztest/test_crates/sample_fuzz_crate/src/lib.rs

@@ -10,3 +10,9 @@ fn sample_fuzztest_target(data: i32) {
 fn another_sample_fuzztest_target(data: i32) {
     let _ = data;
 }
+
+#[fuzztest(_a = Arbitrary::<i32>::default())]
+fn jobs_test(_a: i32) {
+    println!("JOBS_TEST_PID: {}", std::process::id());
+    std::thread::sleep(std::time::Duration::from_millis(250));
+}

rust/cargo_fuzztest/tests/e2e_cli_test.rs

@@ -102,3 +102,41 @@ fn test_cargo_fuzztest_e2e_specific_target() {
     expect_true!(stdout_str.contains("sample_fuzztest_target"));
     expect_false!(stdout_str.contains("another_sample_fuzztest_target"));
 }
+
+#[gtest]
+fn test_cargo_fuzztest_e2e_parallel_jobs() {
+    let sample_crate_path = get_sample_crate_path("sample_fuzz_crate");
+
+    let temp_target_dir = TempDir::new().expect("Failed to create temporary target directory");
+
+    let centipede_bin = env::var("FUZZTEST_CENTIPEDE_BINARY_PATH")
+        .expect("FUZZTEST_CENTIPEDE_BINARY_PATH needs to be set for the test");
+
+    let mut cmd = setup_cargo_fuzztest_command(&sample_crate_path, temp_target_dir.path());
+    cmd.arg("__fuzztest_mod__jobs_test::jobs_test")
+        .arg("--jobs")
+        .arg("4")
+        .arg("--fuzz-for")
+        .arg("5s")
+        .env_remove("FUZZTEST_CENTIPEDE_BINARY_PATH")
+        .arg("--centipede-binary-path")
+        .arg(centipede_bin)
+        .env("FUZZTEST_PRINT_SUBPROCESS_LOG", "true");
+
+    let output = cmd.output().expect("running local cargo-fuzztest sub-process should complete");
+
+    expect_true!(output.status.success());
+
+    let stderr_str = String::from_utf8_lossy(&output.stderr);
+    use std::collections::HashSet;
+    let mut pids = HashSet::new();
+    for line in stderr_str.lines() {
+        if let Some(pos) = line.find("LOG: JOBS_TEST_PID: ") {
+            let pid_str = &line[pos + "LOG: JOBS_TEST_PID: ".len()..];
+            if let Ok(pid) = pid_str.parse::<u32>() {
+                pids.insert(pid);
+            }
+        }
+    }
+    expect_that!(pids.len(), eq(4));
+}

rust/cargo_fuzztest/tests/runner_test.rs

@@ -103,6 +103,46 @@ fn test_runner_build_run_command_with_centipede_binary_path() {
     )));
 }
 
+#[gtest]
+fn test_runner_build_run_command_with_jobs() {
+    let binary_path = get_sample_fuzz_test_bin_path();
+    let fuzztest_options = FuzzTestOptions {
+        jobs: Some(4),
+        fuzz_for: Some(FuzzFor::Duration("10s".parse().expect("static valid duration string"))),
+        ..Default::default()
+    };
+    let options = CargoFuzzTestOptions {
+        fuzztest_options,
+        centipede_binary_path: Some("/custom/path/to/centipede".to_string()),
+        ..Default::default()
+    };
+    let runner = FuzztestRunner::new("x86_64-unknown-linux-gnu".to_string(), options);
+    let cmd = runner.build_run_command(&binary_path).expect("valid run command");
+
+    let envs: Vec<(String, Option<String>)> = cmd
+        .get_envs()
+        .map(|(k, v)| (k.to_string_lossy().to_string(), v.map(|s| s.to_string_lossy().to_string())))
+        .collect();
+    expect_true!(envs.contains(&("FUZZTEST_JOBS".to_string(), Some("4".to_string()))));
+    expect_true!(envs.contains(&("FUZZTEST_FUZZ_FOR".to_string(), Some("10s".to_string()))));
+}
+
+#[gtest]
+fn test_runner_build_run_command_with_jobs_only() {
+    let binary_path = get_sample_fuzz_test_bin_path();
+    let fuzztest_options = FuzzTestOptions { jobs: Some(4), ..Default::default() };
+    let options = CargoFuzzTestOptions { fuzztest_options, ..Default::default() };
+    let runner = FuzztestRunner::new("x86_64-unknown-linux-gnu".to_string(), options);
+    let cmd = runner.build_run_command(&binary_path).expect("valid run command");
+
+    let envs: Vec<(String, Option<String>)> = cmd
+        .get_envs()
+        .map(|(k, v)| (k.to_string_lossy().to_string(), v.map(|s| s.to_string_lossy().to_string())))
+        .collect();
+    expect_false!(envs.iter().any(|(k, _)| k == "FUZZTEST_JOBS"));
+    expect_false!(envs.iter().any(|(k, _)| k == "FUZZTEST_FUZZ_FOR"));
+}
+
 #[gtest]
 fn test_execution_mode_without_centipede_binary_path_errors() {
     let fuzztest_options =

rust/options/src/lib.rs

@@ -124,6 +124,7 @@ impl ExecutionMode {
             return ExecutionMode::ReplayCorpus(ReplayCorpusOptions {
                 replay_corpus_for,
                 time_budget_type: options.time_budget_type,
+                jobs: options.jobs.clone(),
             });
         }
 
@@ -135,8 +136,12 @@ impl ExecutionMode {
             return ExecutionMode::ReplayCrash(ReplayCrashOptions { replay_id: replay_id.clone() });
         }
 
-        if let Some(fuzz_for) = options.fuzz_for {
-            return ExecutionMode::Fuzz(FuzzOptions { fuzz_for, jobs: options.jobs.clone() });
+        // Continuous fuzzing mode is selected if an explicit duration/budget (`fuzz_for`) is specified.
+        if let Some(fuzz_for) = &options.fuzz_for {
+            return ExecutionMode::Fuzz(FuzzOptions {
+                fuzz_for: *fuzz_for,
+                jobs: options.jobs.clone(),
+            });
         }
 
         ExecutionMode::SmokeTest
@@ -174,6 +179,9 @@ pub struct ReplayCrashOptions {
 pub struct ReplayCorpusOptions {
     pub replay_corpus_for: Duration,
     pub time_budget_type: TimeBudgetType,
+    /// If `jobs` is `None`, we won't specify the number of jobs while invoking Centipede and it
+    /// will use its own default value.
+    pub jobs: Option<usize>,
 }
 
 #[cfg(test)]
@@ -222,4 +230,78 @@ mod tests {
         expect_that!(options.replay_id.as_deref(), eq(Some("my_crash_123")));
         expect_that!(options.corpus_db.as_deref(), eq(Some("/tmp/corpus_db")));
     }
+
+    #[gtest]
+    fn test_jobs_options_parsing_env() {
+        // SAFETY: Testing environment parsing in single-threaded context.
+        unsafe {
+            std::env::set_var("FUZZTEST_JOBS", "4");
+        }
+
+        let options = FuzzTestOptions::parse_from(std::iter::empty::<OsString>());
+
+        expect_that!(options.jobs, eq(Some(4)));
+        // Setting jobs alone should not enter fuzzing mode; it defaults to smoke test mode.
+        expect_that!(ExecutionMode::from_fuzztest_options(&options), eq(&ExecutionMode::SmokeTest));
+
+        // SAFETY: Cleaning up environment variables.
+        unsafe {
+            std::env::remove_var("FUZZTEST_JOBS");
+        }
+    }
+
+    #[gtest]
+    fn test_jobs_with_fuzz_for_parsing_env() {
+        // SAFETY: Testing environment parsing in single-threaded context.
+        unsafe {
+            std::env::set_var("FUZZTEST_JOBS", "4");
+            std::env::set_var("FUZZTEST_FUZZ_FOR", "5s");
+        }
+
+        let options = FuzzTestOptions::parse_from(std::iter::empty::<OsString>());
+
+        expect_that!(options.jobs, eq(Some(4)));
+        let expected_duration = "5s".parse().expect("valid duration");
+        expect_that!(
+            ExecutionMode::from_fuzztest_options(&options),
+            eq(&ExecutionMode::Fuzz(FuzzOptions {
+                fuzz_for: FuzzFor::Duration(expected_duration),
+                jobs: Some(4),
+            }))
+        );
+
+        // SAFETY: Cleaning up environment variables.
+        unsafe {
+            std::env::remove_var("FUZZTEST_JOBS");
+            std::env::remove_var("FUZZTEST_FUZZ_FOR");
+        }
+    }
+
+    #[gtest]
+    fn test_jobs_with_replay_corpus_parsing_env() {
+        // SAFETY: Testing environment parsing in single-threaded context.
+        unsafe {
+            std::env::set_var("FUZZTEST_JOBS", "4");
+            std::env::set_var("FUZZTEST_REPLAY_CORPUS_FOR", "10s");
+        }
+
+        let options = FuzzTestOptions::parse_from(std::iter::empty::<OsString>());
+
+        expect_that!(options.jobs, eq(Some(4)));
+        let expected_duration = "10s".parse().expect("valid duration");
+        expect_that!(
+            ExecutionMode::from_fuzztest_options(&options),
+            eq(&ExecutionMode::ReplayCorpus(ReplayCorpusOptions {
+                replay_corpus_for: expected_duration,
+                time_budget_type: TimeBudgetType::PerTest,
+                jobs: Some(4),
+            }))
+        );
+
+        // SAFETY: Cleaning up environment variables.
+        unsafe {
+            std::env::remove_var("FUZZTEST_JOBS");
+            std::env::remove_var("FUZZTEST_REPLAY_CORPUS_FOR");
+        }
+    }
 }

rust/src/options.rs

@@ -255,6 +255,9 @@ impl CentipedeArgs {
                 add_arg("--fuzztest_replay_coverage_inputs=true".to_string())?;
                 add_arg("--load_shards_only=true".to_string())?;
                 add_arg(format!("--fuzztest_time_limit_per_test={time_limit}"))?;
+                if let Some(jobs) = &replay_corpus_opts.jobs {
+                    add_arg(format!("--j={jobs}"))?;
+                }
             }
             ExecutionMode::SmokeTest => unreachable!(),
             ExecutionMode::ListFuzzTests => unreachable!(),
@@ -604,6 +607,30 @@ mod tests {
             .any(|s| s.starts_with("--binary=") && s.contains("my_mod::my_test --exact")));
         assert!(args_str.contains(&"--fuzztest_only_replay=true"));
         assert!(args_str.contains(&"--fuzztest_time_limit_per_test=10s"));
+        assert!(!args_str.iter().any(|s| s.starts_with("--j=")));
+    }
+
+    #[gtest]
+    fn test_determine_execution_action_replay_corpus_with_jobs() {
+        let expected_duration = "10s".parse().expect("failed to parse duration");
+        let options = FuzzTestOptions {
+            replay_corpus_for: Some(expected_duration),
+            jobs: Some(4),
+            time_budget_type: TimeBudgetType::PerTest,
+            ..Default::default()
+        };
+        let action = determine_execution_action_internal(&options, "my_mod::my_test");
+
+        let ExecutionAction::Standalone(args) = action else {
+            panic!("Expected Standalone action");
+        };
+
+        let args_str: Vec<&str> =
+            args._c_strings.iter().map(|s| s.to_str().expect("invalid utf8")).collect();
+
+        assert!(args_str.contains(&"--j=4"));
+        assert!(args_str.contains(&"--fuzztest_only_replay=true"));
+        assert!(args_str.contains(&"--fuzztest_time_limit_per_test=10s"));
     }
 
     #[gtest]