one child for each channel

Author: shepmaster

orchestrator/src/coordinator.rs

@@ -26,7 +26,7 @@ use crate::{
         CoordinatorMessage, JobId, Multiplexed, OneToOneResponse, ReadFileRequest,
         ReadFileResponse, SerializedError, WorkerMessage, WriteFileRequest,
     },
-    sandbox::{CompileRequest, CompileResponse, CompileResponseWithOutput},
+    sandbox::{CompileRequest, CompileResponse, CompileResponseWithOutput, Channel},
     DropErrorDetailsExt,
 };
 
@@ -38,12 +38,99 @@ enum DemultiplexCommand {
 
 #[derive(Debug)]
 pub struct Container {
-    task: JoinHandle<Result<()>>,
-    commander: Commander,
+    stable: ContainerCore,
+    beta: ContainerCore,
+    nightly: ContainerCore,
     token: CancellationToken,
 }
 
 impl Container {
+    pub async fn compile(
+        &self,
+        request: CompileRequest,
+    ) -> Result<CompileResponseWithOutput, CompileError> {
+        self.select_channel(request.channel).compile(request).await
+    }
+
+    pub async fn begin_compile(
+        &self,
+        request: CompileRequest,
+    ) -> Result<ActiveCompilation, CompileError> {
+        self.select_channel(request.channel).begin_compile(request).await
+    }
+
+    pub async fn shutdown(self) -> Result<()> {
+        let Self {
+            stable,
+            beta,
+            nightly,
+            token,
+        } = self;
+        token.cancel();
+
+        let (stable, beta, nightly) = join!(stable.shutdown(), beta.shutdown(), nightly.shutdown());
+
+        stable.unwrap();
+        beta.unwrap();
+        nightly.unwrap();
+
+        Ok(())
+//        task.await.context(ContainerTaskPanickedSnafu)?
+    }
+
+
+    fn select_channel(&self, channel: Channel) -> &ContainerCore {
+        match channel {
+            Channel::Stable => &self.stable,
+            Channel::Beta => &self.beta,
+            Channel::Nightly => &self.nightly,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct ContainerCore {
+    task: JoinHandle<Result<()>>,
+    commander: Commander,
+}
+
+impl ContainerCore {
+    fn new(channel: Channel, token: CancellationToken, backend: &impl Backend) -> Result<Self> {
+        let (mut child, stdin, stdout) = backend.run_worker_in_background(channel)?;
+        let IoQueue {
+            mut tasks,
+            to_worker_tx,
+            from_worker_rx,
+        } = spawn_io_queue(stdin, stdout, token);
+
+        let (command_tx, command_rx) = mpsc::channel(8);
+        let demultiplex_task = tokio::spawn(Commander::demultiplex(command_rx, from_worker_rx));
+
+        let task = tokio::spawn(async move {
+            let (c, d, t) = join!(child.wait(), demultiplex_task, tasks.join_next());
+            c.context(JoinWorkerSnafu)?;
+            d.context(DemultiplexerTaskPanickedSnafu)?
+                .context(DemultiplexerTaskFailedSnafu)?;
+            if let Some(t) = t {
+                t.context(IoQueuePanickedSnafu)??;
+            }
+
+            Ok(())
+        });
+
+        let commander = Commander {
+            to_worker_tx,
+            to_demultiplexer_tx: command_tx,
+            id: Default::default(),
+        };
+
+        Ok(ContainerCore {
+            task,
+            commander,
+        })
+
+    }
+
     pub async fn compile(
         &self,
         request: CompileRequest,
@@ -150,10 +237,8 @@ impl Container {
         let Self {
             task,
             commander,
-            token,
         } = self;
         drop(commander);
-        token.cancel();
         task.await.context(ContainerTaskPanickedSnafu)?
     }
 }
@@ -423,37 +508,14 @@ impl<B: Backend> Coordinator<B> {
     pub fn allocate(&mut self) -> Result<Container, Error> {
         let token = CancellationToken::new();
 
-        let (mut child, stdin, stdout) = self.backend.run_worker_in_background()?;
-        let IoQueue {
-            mut tasks,
-            to_worker_tx,
-            from_worker_rx,
-        } = spawn_io_queue(stdin, stdout, token.clone());
-
-        let (command_tx, command_rx) = mpsc::channel(8);
-        let demultiplex_task = tokio::spawn(Commander::demultiplex(command_rx, from_worker_rx));
-
-        let task = tokio::spawn(async move {
-            let (c, d, t) = join!(child.wait(), demultiplex_task, tasks.join_next());
-            c.context(JoinWorkerSnafu)?;
-            d.context(DemultiplexerTaskPanickedSnafu)?
-                .context(DemultiplexerTaskFailedSnafu)?;
-            if let Some(t) = t {
-                t.context(IoQueuePanickedSnafu)??;
-            }
-
-            Ok(())
+        let [stable, beta, nightly] = Channel::ALL.map(|channel| {
+            ContainerCore::new(channel, token.clone(), &self.backend)
         });
 
-        let commander = Commander {
-            to_worker_tx,
-            to_demultiplexer_tx: command_tx,
-            id: Default::default(),
-        };
-
         Ok(Container {
-            task,
-            commander,
+            stable: stable?,
+            beta: beta?,
+            nightly: nightly?,
             token,
         })
     }
@@ -466,9 +528,9 @@ impl Coordinator<DockerBackend> {
 }
 
 pub trait Backend {
-    fn run_worker_in_background(&self) -> Result<(Child, ChildStdin, ChildStdout)> {
+    fn run_worker_in_background(&self, channel: Channel) -> Result<(Child, ChildStdin, ChildStdout)> {
         let mut child = self
-            .prepare_worker_command()
+            .prepare_worker_command(channel)
             .stdin(Stdio::piped())
             .stdout(Stdio::piped())
             .stderr(Stdio::inherit())
@@ -479,7 +541,7 @@ pub trait Backend {
         Ok((child, stdin, stdout))
     }
 
-    fn prepare_worker_command(&self) -> Command;
+    fn prepare_worker_command(&self, channel: Channel) -> Command;
 }
 
 macro_rules! docker_command {
@@ -516,19 +578,29 @@ fn basic_secure_docker_command() -> Command {
 pub struct DockerBackend(());
 
 impl Backend for DockerBackend {
-    fn prepare_worker_command(&self) -> Command {
+    fn prepare_worker_command(&self, channel: Channel) -> Command {
         let mut command = basic_secure_docker_command();
         command
             .arg("-i")
             .args(["-a", "stdin", "-a", "stdout", "-a", "stderr"])
             .arg("--rm")
-            .arg("orchestrator")
+            .arg(channel.to_container_name())
             .arg("worker")
             .arg("/playground");
         command
     }
 }
 
+impl Channel {
+    fn to_container_name(&self) -> &'static str {
+        match self {
+            Channel::Stable => "orchestrator-stable",
+            Channel::Beta => "orchestrator-beta",
+            Channel::Nightly => "orchestrator-nightly",
+        }
+    }
+}
+
 pub type Result<T, E = Error> = ::std::result::Result<T, E>;
 
 #[derive(Debug, Snafu)]
@@ -681,7 +753,14 @@ mod tests {
     }
 
     impl Backend for TestBackend {
-        fn prepare_worker_command(&self) -> Command {
+        fn prepare_worker_command(&self, channel: Channel) -> Command {
+            let toolchain_file = format!(r#"
+                [toolchain]
+                channel = "{}"
+            "#, channel.to_str());
+            let path = self.project_dir.path().join("rust-toolchain.toml");
+            std::fs::write(path, toolchain_file).expect("Couldn't write toolchain file");
+
             let mut command = Command::new("./target/debug/worker");
             command.arg(self.project_dir.path());
             command

orchestrator/src/sandbox.rs

@@ -35,6 +35,18 @@ pub enum Channel {
     Nightly,
 }
 
+impl Channel {
+    pub(crate) const ALL: [Self; 3] = [Self::Stable, Self::Beta, Self::Nightly];
+
+    pub(crate) fn to_str(&self) -> &'static str {
+        match self {
+            Channel::Stable => "stable",
+            Channel::Beta => "beta",
+            Channel::Nightly => "nightly",
+        }
+    }
+}
+
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub enum Mode {
     Debug,