process: rewrite and simplify the issue_42 test
#![cfg(feature = "full")]
#![cfg(unix)]
-use tokio::process::Command;
-use tokio::runtime;
-
-use futures::future::FutureExt;
-use futures::stream::FuturesOrdered;
+use futures::future::join_all;
use std::process::Stdio;
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::Arc;
-use std::thread;
-use std::time::Duration;
-
-fn run_test() {
- let finished = Arc::new(AtomicBool::new(false));
- let finished_clone = finished.clone();
-
- thread::spawn(move || {
- let mut rt = runtime::Builder::new()
- .basic_scheduler()
- .enable_all()
- .build()
- .unwrap();
-
- let mut futures = FuturesOrdered::new();
- rt.block_on(async {
- for i in 0..2 {
- futures.push(
+use tokio::process::Command;
+use tokio::task;
+
+#[tokio::test]
+async fn issue_42() {
+ // We spawn a many batches of processes which should exit at roughly the
+ // same time (modulo OS scheduling delays), to make sure that consuming
+ // a readiness event for one process doesn't inadvertently starve another.
+ // We then do this many times (in parallel) in an effort to stress test the
+ // implementation to ensure there are no race conditions.
+ // See alexcrichton/tokio-process#42 for background
+ let join_handles = (0..10usize).into_iter().map(|_| {
+ task::spawn(async {
+ let processes = (0..10usize).into_iter().map(|i| {
Command::new("echo")
.arg(format!("I am spawned process #{}", i))
.stdin(Stdio::null())
@@ -36,27 +26,11 @@ fn run_test() {
.kill_on_drop(true)
.spawn()
.unwrap()
- .boxed(),
- )
- }
});
- drop(rt);
- finished_clone.store(true, Ordering::SeqCst);
+ join_all(processes).await;
+ })
});
- thread::sleep(Duration::from_millis(1000));
- assert!(
- finished.load(Ordering::SeqCst),
- "FINISHED flag not set, maybe we deadlocked?"
- );
-}
-
-#[test]
-fn issue_42() {
- let max = 10;
- for i in 0..max {
- println!("running {}/{}", i, max);
- run_test()
- }
+ join_all(join_handles).await;
}
