Fuse one-time cancellation tokens instead of using a separate flag

Author: shepmaster

compiler/base/orchestrator/src/coordinator.rs

@@ -4,6 +4,7 @@ use snafu::prelude::*;
 use std::{
     collections::{BTreeSet, HashMap},
     fmt, mem, ops,
+    pin::pin,
     process::Stdio,
     sync::{
         atomic::{AtomicU64, Ordering},
@@ -1839,14 +1840,12 @@ impl Container {
 
         let task = tokio::spawn({
             async move {
-                let mut already_cancelled = false;
+                let mut cancelled = pin!(token.cancelled().fuse());
                 let mut stdin_open = true;
 
                 loop {
                     select! {
-                        () = token.cancelled(), if !already_cancelled => {
-                            already_cancelled = true;
-
+                        () = &mut cancelled => {
                             let msg = CoordinatorMessage::Kill;
                             trace!(msg_name = msg.as_ref(), "processing");
                             to_worker_tx.send(msg).await.context(KillSnafu)?;

compiler/base/orchestrator/src/worker.rs

@@ -31,11 +31,13 @@
 //!   - [`tokio::process::ChildStderr`][]
 //!
 
+use futures::FutureExt as _;
 use snafu::prelude::*;
 use std::{
     collections::HashMap,
     io,
     path::{Path, PathBuf},
+    pin::pin,
     process::{ExitStatus, Stdio},
 };
 use tokio::{
@@ -615,14 +617,13 @@ async fn process_end(
 ) -> Result<ExecuteCommandResponse, ProcessError> {
     use process_error::*;
 
-    let mut killed = false;
+    let mut cancelled = pin!(token.cancelled().fuse());
 
     let status = loop {
         select! {
             // The user requested that the process be killed
-            () = token.cancelled(), if !killed => {
+            () = &mut cancelled => {
                 child.kill().await.context(KillChildSnafu)?;
-                killed = true;
             },
 
             // The process exited normally