work

Author: MikailBag

src/judge/src/controller.rs

@@ -164,21 +164,21 @@ impl Controller {
         let engine = self.invoker_set.clone();
         let judge_cx = crate::request_handler::JudgeContext {
             events_tx: judge_events_tx,
-            invoker: rpc::Client::new(rpc::box_engine(engine), "http://does-not-matter".to_string()),
+            invoker: rpc::Client::new(
+                rpc::box_engine(engine),
+                "http://does-not-matter".to_string(),
+            ),
         };
 
         // TODO: can we split into LoweredJudgeRequest and Extensions?
-        let mut responses = worker
-            .send(Request::Judge(low_req))
-            .await
-            .context("failed to submit lowered judge request")?;
+        crate::request_handler::do_judge(judge_cx, low_req);
         loop {
-            let message = responses
+            let message = judge_events_rx
                 .next()
                 .await
                 .context("failed to receive next worker message")?;
             match message {
-                Response::JudgeDone(judge_outcome) => {
+                Event::JudgeDone(judge_outcome) => {
                     debug!("Publising: JudgeOutcome {:?}", &judge_outcome);
                     let reason = match judge_outcome {
                         JudgeOutcome::Fault => InvocationFinishReason::Fault,
@@ -191,13 +191,13 @@ impl Controller {
                         .context("failed to set run outcome in DB")?;
                     break;
                 }
-                Response::LiveScore(score) => {
+                Event::LiveScore(score) => {
                     exts.notifier.set_score(score).await;
                 }
-                Response::LiveTest(test) => {
+                Event::LiveTest(test) => {
                     exts.notifier.set_test(test).await;
                 }
-                Response::OutcomeHeader(header) => {
+                Event::OutcomeHeader(header) => {
                     req.callbacks
                         .add_outcome_header(req.request.request_id, header)
                         .await?;

src/judge/src/controller/task_loading.rs

@@ -4,7 +4,6 @@ use super::{
 };
 use crate::{
     request_handler::{Command, LoweredJudgeRequest},
-    worker,
 };
 use anyhow::Context;
 use std::{

src/judge/src/lib.rs

@@ -5,5 +5,4 @@ pub mod controller;
 pub mod init;
 mod invoker_set;
 pub mod sources;
-pub mod worker;
 mod request_handler;

src/judge/src/request_handler.rs

@@ -8,29 +8,22 @@ use self::{
     compiler::{BuildOutcome, Compiler},
     exec_test::{exec, ExecRequest},
 };
-use judging_apis::Status;
+use anyhow::Context;
+use judging_apis::{valuer_proto::ValuerResponse, Status};
 use serde::{Deserialize, Serialize};
 use std::{
     borrow::Cow,
     future::Future,
     path::{Path, PathBuf},
     pin::Pin,
 };
-use tracing::debug;
-
+use tracing::{debug, error, instrument};
 /// Allows sending requests to invoker
 // TODO upstream BoxedEngine to jjs-commons
 pub type InvokerClient = rpc::Client<rpc::BoxEngine>;
 
-pub(crate) struct JudgeContext {
-    /// Can be used to send requests to invoker
-    pub(crate) invoker: InvokerClient,
-    /// Channel that should be used for sending updates
-    pub(crate) events_tx: async_channel::Sender<Event>,
-}
-
 /// Note: this is not `judging_apis::invoke::Command`, it is higher-level.
-#[derive(Debug)]
+#[derive(Debug, Default)]
 pub(crate) struct Command {
     pub argv: Vec<String>,
     pub env: Vec<String>,
@@ -100,3 +93,182 @@ pub(crate) enum JudgeOutcome {
     /// Maybe, several protocols were emitted, but results are neither precise nor complete
     Fault,
 }
+
+pub(crate) struct JudgeContext {
+    /// Can be used to send requests to invoker
+    pub(crate) invoker: InvokerClient,
+    /// Channel that should be used for sending updates
+    pub(crate) events_tx: async_channel::Sender<Event>,
+}
+
+#[instrument(skip(cx, judge_req), fields(id = %judge_req.judge_request_id))]
+pub fn do_judge(mut cx: JudgeContext, judge_req: LoweredJudgeRequest) {
+    debug!("Got LoweredJudgeRequest: {:?}", &judge_req);
+    tokio::task::spawn(async move {
+        let outcome = match cx.judge(&judge_req).await {
+            Ok(o) => o,
+            Err(err) => {
+                error!("Invoke failed: {:#}", err);
+                cx.create_fake_protocols(
+                    &judge_req,
+                    &judging_apis::Status {
+                        kind: judging_apis::StatusKind::InternalError,
+                        code: judging_apis::status_codes::JUDGE_FAULT.to_string(),
+                    },
+                )
+                .await
+                .ok();
+                JudgeOutcome::Fault
+            }
+        };
+        debug!("JudgeOutcome: {:?}", &outcome);
+        cx.events_tx.send(Event::JudgeDone(outcome)).await;
+    });
+}
+
+impl JudgeContext {
+    async fn judge(&mut self, req: &LoweredJudgeRequest) -> anyhow::Result<JudgeOutcome> {
+        let compiler = Compiler { req };
+
+        if !req.run_source.exists() {
+            anyhow::bail!("Run source file not exists");
+        }
+
+        if !req.out_dir.exists() {
+            anyhow::bail!("Run output dir not exists");
+        }
+
+        let compiler_response = compiler.compile();
+
+        let outcome;
+
+        match compiler_response {
+            Err(err) => return Err(err).context("compilation error"),
+            Ok(BuildOutcome::Error(st)) => {
+                self.create_fake_protocols(req, &st).await?;
+                outcome = JudgeOutcome::CompileError(st);
+            }
+            Ok(BuildOutcome::Success) => {
+                self.run_tests(req).await.context("failed to run tests")?;
+
+                outcome = JudgeOutcome::TestingDone;
+            }
+        };
+        Ok(outcome)
+    }
+
+    /// Used when we are unable to produce protocols, i.e. on compilation errors
+    /// and judge faults.
+    async fn create_fake_protocols(
+        &mut self,
+        req: &LoweredJudgeRequest,
+        status: &judging_apis::Status,
+    ) -> anyhow::Result<()> {
+        for kind in judging_apis::judge_log::JudgeLogKind::list() {
+            let pseudo_valuer_proto = judging_apis::valuer_proto::JudgeLog {
+                kind,
+                tests: vec![],
+                subtasks: vec![],
+                score: 0,
+                is_full: false,
+            };
+            let mut protocol = self.process_judge_log(&pseudo_valuer_proto, req, &[])?;
+            protocol.status = status.clone();
+            self.put_protocol(req, protocol).await?;
+        }
+        Ok(())
+    }
+
+    async fn put_outcome(
+        &mut self,
+        score: u32,
+        status: judging_apis::Status,
+        kind: judging_apis::judge_log::JudgeLogKind,
+    ) {
+        let header = judging_apis::JudgeOutcomeHeader {
+            score: Some(score),
+            status,
+            kind,
+        };
+        self.events_tx.send(Event::OutcomeHeader(header)).await.ok();
+    }
+
+    async fn put_protocol(
+        &mut self,
+        req: &LoweredJudgeRequest,
+        protocol: judging_apis::judge_log::JudgeLog,
+    ) -> anyhow::Result<()> {
+        let protocol_file_name = format!("protocol-{}.json", protocol.kind.as_str());
+        let protocol_path = req.out_dir.join(protocol_file_name);
+        debug!("Writing protocol to {}", protocol_path.display());
+        let protocol_file = std::fs::File::create(&protocol_path)?;
+        let protocol_file = std::io::BufWriter::new(protocol_file);
+        serde_json::to_writer(protocol_file, &protocol)
+            .context("failed to write judge log to file")?;
+        self.put_outcome(protocol.score, protocol.status, protocol.kind)
+            .await;
+        Ok(())
+    }
+
+    async fn run_tests(&mut self, req: &LoweredJudgeRequest) -> anyhow::Result<()> {
+        let mut test_results = vec![];
+
+        let mut valuer = valuer::Valuer::new(req).context("failed to init valuer")?;
+        valuer
+            .write_problem_data(req)
+            .await
+            .context("failed to send problem data")?;
+        loop {
+            match valuer.poll().await? {
+                ValuerResponse::Test { test_id: tid, live } => {
+                    if live {
+                        self.events_tx.send(Event::LiveTest(tid.get())).await;
+                    }
+                    let tid_u32: u32 = tid.into();
+                    let test = &req.problem.tests[(tid_u32 - 1u32) as usize];
+                    let exec_request = ExecRequest {
+                        test,
+                        test_id: tid.into(),
+                    };
+
+                    let judge_response = exec_test::exec(&req, exec_request, self)
+                        .with_context(|| format!("failed to judge solution on test {}", tid))?;
+                    test_results.push((tid, judge_response.clone()));
+                    valuer
+                        .notify_test_done(judging_apis::valuer_proto::TestDoneNotification {
+                            test_id: tid,
+                            test_status: judge_response.status,
+                        })
+                        .await
+                        .with_context(|| {
+                            format!("failed to notify valuer that test {} is done", tid)
+                        })?;
+                }
+                ValuerResponse::Finish => {
+                    break;
+                }
+                ValuerResponse::LiveScore { score } => {
+                    self.events_tx.send(Event::LiveScore(score)).await;
+                }
+                ValuerResponse::JudgeLog(judge_log) => {
+                    let converted_judge_log = self
+                        .process_judge_log(&judge_log, req, &test_results)
+                        .context("failed to convert valuer judge log to invoker judge log")?;
+                    self.put_protocol(req, converted_judge_log)
+                        .await
+                        .context("failed to save protocol")?;
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Creates `InputSource` that can be further sent to invoker
+    async fn intern(&self, data: &[u8]) -> anyhow::Result<judging_apis::invoke::InputSource> {
+        // TODO: optimize
+        Ok(judging_apis::invoke::InputSource::Inline {
+            data: data.to_vec(),
+        })
+    }
+}

src/judge/src/request_handler/compiler.rs

@@ -1,4 +1,4 @@
-use crate::request_handler::{invoke_util, LoweredJudgeRequest};
+use crate::request_handler::LoweredJudgeRequest;
 use anyhow::Context;
 use judging_apis::{status_codes, Status, StatusKind};
 use std::fs;
@@ -11,14 +11,17 @@ pub(crate) enum BuildOutcome {
 /// Compiler turns SubmissionInfo into Artifact
 pub(crate) struct Compiler<'a> {
     pub(crate) req: &'a LoweredJudgeRequest,
-    pub(crate) minion: &'a dyn minion::erased::Backend,
-    pub(crate) config: &'a crate::config::JudgeConfig,
+    // pub(crate) config: &'a crate::config::JudgeConfig,
 }
 
 impl<'a> Compiler<'a> {
     pub(crate) fn compile(&self) -> anyhow::Result<BuildOutcome> {
-        let sandbox = invoke_util::create_sandbox(self.req, None, self.minion, self.config)
-            .context("failed to create sandbox")?;
+        let mut graph = judging_apis::invoke::InvokeRequest {
+            inputs: vec![],
+            outputs: vec![],
+            steps: vec![],
+        };
+
         let step_dir = self.req.step_dir(None);
         fs::copy(
             &self.req.run_source,

src/judge/src/request_handler/exec_test.rs

@@ -1,9 +1,9 @@
 mod checker_proto;
 
-use super::{invoke_util, JudgeContext, LoweredJudgeRequest};
+use super::{JudgeContext, LoweredJudgeRequest};
 use anyhow::Context;
 use judging_apis::{
-    invoke::{Command, Action, Step, Stdio, FileId},
+    invoke::{Action, Command, FileId, Input, InputSource, Stdio, Step},
     status_codes, Status, StatusKind,
 };
 use std::{fs, io::Write, path::PathBuf};
@@ -51,17 +51,47 @@ fn map_checker_outcome_to_status(out: checker_proto::Output) -> Status {
 }
 
 /// Runs Artifact on one test and produces output
-pub fn exec(judge_req: &LoweredJudgeRequest, ctx: &JudgeContext) -> anyhow::Result<ExecOutcome> {
+pub async fn exec(
+    judge_req: &LoweredJudgeRequest,
+    exec_req: ExecRequest<'_>,
+    cx: &JudgeContext,
+) -> anyhow::Result<ExecOutcome> {
     let mut invoke_request = judging_apis::invoke::InvokeRequest {
         steps: vec![],
         inputs: vec![],
         outputs: vec![],
     };
+    let input_file = judge_req.resolve_asset(&exec_req.test.path);
+    let test_data = std::fs::read(input_file).context("failed to read test")?;
     const EXEC_SOLUTION_GENERATION: u32 = 0;
-    const EXEC_SOLUTION_OUTPUT_FILE: FileId = FileId(0);
-    const EXEC_SOLUTION_ERROR_FILE: FileId = FileId(1);
-    
+    const TEST_DATA_INPUT_FILE: &str = "test-data";
+    const EXEC_SOLUTION_OUTPUT_FILE: &str = "solution-output";
+    const EXEC_SOLUTION_ERROR_FILE: &str = "solution-error";
+
     const EXEC_CHECKER_GENERATION: u32 = 1;
+    // create an input with the test date
+    {
+        let test_data_input = Input {
+            id: FileId(TEST_DATA_INPUT_FILE.to_string()),
+            source: cx.intern(&test_data).await?,
+        };
+        invoke_request.inputs.push(test_data_input);
+    }
+    // prepare files for stdout & stderr
+    {
+        invoke_request.steps.push(Step {
+            generation: EXEC_SOLUTION_GENERATION,
+            action: Action::CreateFile {
+                id: FileId(EXEC_SOLUTION_OUTPUT_FILE.to_string()),
+            },
+        });
+        invoke_request.steps.push(Step {
+            generation: EXEC_SOLUTION_GENERATION,
+            action: Action::CreateFile {
+                id: FileId(EXEC_SOLUTION_ERROR_FILE.to_string()),
+            },
+        })
+    }
     // produce a step for executing solution
     {
         let exec_solution_step = Step {
@@ -71,13 +101,14 @@ pub fn exec(judge_req: &LoweredJudgeRequest, ctx: &JudgeContext) -> anyhow::Resu
                 env: judge_req.execute_command.env.clone(),
                 cwd: judge_req.execute_command.cwd.clone(),
                 stdio: Stdio {
-
-                }
+                    stdin: FileId(TEST_DATA_INPUT_FILE.to_string()),
+                    stdout: FileId(EXEC_SOLUTION_OUTPUT_FILE.to_string()),
+                    stderr: FileId(EXEC_SOLUTION_ERROR_FILE.to_string()),
+                },
             }),
         };
+        invoke_request.steps.push(exec_solution_step);
     }
-    let input_file = self.req.resolve_asset(&self.exec.test.path);
-    let test_data = std::fs::read(input_file).context("failed to read test")?;
     let run_outcome = self.run_solution(&test_data, self.exec.test_id)?;
     let sol_file_path = match run_outcome.var {
         RunOutcomeVar::Success { out_data_path } => out_data_path,