diff --git a/Cargo.toml b/Cargo.toml index 6466f6f..649c8ce 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,8 @@ tower = "0.3" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" rand = "0.7" +tempfile = "3" +libc = "0.2" [build-dependencies] tonic-build ={ version = "0.3" } diff --git a/proto/river.proto b/proto/river.proto index 85758af..e7c77eb 100644 --- a/proto/river.proto +++ b/proto/river.proto @@ -45,13 +45,18 @@ message JudgeResponse { CompileError = 6; PresentationError = 7; SystemError = 8; - Pending = 9; - Compiling = 10; - Running = 11; } JudgeResult result = 3; int32 errno = 4; int32 exit_code = 5; string stdout = 6; string stderr = 7; + string errmsg = 8; + enum JudgeStatus { + Pending = 0; + Compiling = 1; + Running = 2; + Ended = 3; + } + JudgeStatus status = 9; } diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 0000000..94ae5b7 --- /dev/null +++ b/src/error.rs @@ -0,0 +1,31 @@ +use crate::river::judge_response::{JudgeResult, JudgeStatus}; +use crate::river::JudgeResponse; +use std::fmt; +use std::result; + +#[derive(Debug)] +pub enum Error {} + +pub type Result = result::Result; + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + match *self { + _ => write!(f, "{:?}", self), + } + } +} + +pub fn system_error(err: Error) -> JudgeResponse { + JudgeResponse { + time_used: 0, + memory_used: 0, + result: JudgeResult::SystemError as i32, + errno: 0, + exit_code: 0, + stdout: "".into(), + stderr: "".into(), + errmsg: format!("{}", err).into(), + status: JudgeStatus::Ended as i32, + } +} diff --git a/src/judger.rs b/src/judger.rs new file mode 100644 index 0000000..0108d78 --- /dev/null +++ b/src/judger.rs @@ -0,0 +1,73 @@ +use super::error::Result; +use crate::river::judge_response::{JudgeResult, JudgeStatus}; +use crate::river::{JudgeRequest, JudgeResponse}; + +pub async fn judger(request: &JudgeRequest) -> Result { + return Ok(JudgeResponse { + time_used: request.time_limit, + memory_used: 2, + result: JudgeResult::Accepted as i32, + errno: 0, + exit_code: 0, + stdout: "stdout".into(), + stderr: "stderr".into(), + errmsg: "".into(), + status: JudgeStatus::Ended as i32, + }); +} + +pub async fn compile(request: &JudgeRequest) -> Result { + return Ok(JudgeResponse { + time_used: request.time_limit, + memory_used: 2, + result: JudgeResult::Accepted as i32, + errno: 0, + exit_code: 0, + stdout: "stdout".into(), + stderr: "stderr".into(), + errmsg: "".into(), + status: JudgeStatus::Ended as i32, + }); +} + +pub fn pending() -> JudgeResponse { + JudgeResponse { + time_used: 0, + memory_used: 0, + result: JudgeResult::Accepted as i32, + errno: 0, + exit_code: 0, + stdout: "".into(), + stderr: "".into(), + errmsg: "".into(), + status: JudgeStatus::Pending as i32, + } +} + +pub fn running() -> JudgeResponse { + JudgeResponse { + time_used: 0, + memory_used: 0, + result: JudgeResult::Accepted as i32, + errno: 0, + exit_code: 0, + stdout: "".into(), + stderr: "".into(), + errmsg: "".into(), + status: JudgeStatus::Running as i32, + } +} + +pub fn compiling() -> JudgeResponse { + JudgeResponse { + time_used: 0, + memory_used: 0, + result: JudgeResult::Accepted as i32, + errno: 0, + exit_code: 0, + stdout: "".into(), + stderr: "".into(), + errmsg: "".into(), + status: JudgeStatus::Compiling as i32, + } +} diff --git a/src/main.rs b/src/main.rs index 1109d0b..6f5e88c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,14 +1,20 @@ +#![recursion_limit = "256"] + +use crate::river::judge_response::JudgeResult; use futures::StreamExt; use futures_core::Stream; use river::river_server::{River, RiverServer}; use river::{JudgeRequest, JudgeResponse}; use std::pin::Pin; -use tokio::time::{delay_for, Duration}; use tonic::transport::Server; use tonic::{Request, Response, Status}; +mod error; +mod judger; +mod process; + pub mod river { - tonic::include_proto!("river"); // The string specified here must match the proto package name + tonic::include_proto!("river"); } #[derive(Debug, Default)] @@ -26,18 +32,36 @@ impl River for RiverService { let mut stream = request.into_inner(); let output = async_stream::try_stream! { - while let Some(note) = stream.next().await { - yield river::JudgeResponse { - time_used: 1, - memory_used: 2, - result: 0, - errno: 0, - exit_code: 0, - stdout: "stdout".into(), - stderr: "stderr".into(), + let mut need_compile = true; + while let Some(req) = stream.next().await { + // TODO: 使用锁或者资源量等机制限制并发 + yield judger::pending(); + let req = req?; + + // 首次获取流进行编译 + if need_compile { + yield judger::compiling(); + let result = match judger::compile(&req).await { + Ok(res) => res, + Err(e) => error::system_error(e) + }; + // 如果编译错误,则不进行后续流程 + if result.result != JudgeResult::Accepted as i32 { + yield result; + break; + } + } + need_compile = false; + + yield judger::running(); + let result = match judger::judger(&req).await { + Ok(res) => res, + Err(e) => error::system_error(e) }; - delay_for(Duration::from_millis(10000)).await; + + yield result; } + while let Some(_) = stream.next().await {} }; Ok(Response::new(Box::pin(output) as Self::JudgeStream)) diff --git a/src/process.rs b/src/process.rs new file mode 100644 index 0000000..f4d1ecf --- /dev/null +++ b/src/process.rs @@ -0,0 +1,41 @@ +use libc; +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::thread; +use std::time::Duration; + +pub struct Process { + pub pid: i32, + pub time_limit: i32, + pub read_time_limit: i32, + pub memory_limit: i32, + pub stdin_fd: i32, + pub stdout_fd: i32, + pub stderr_fd: i32, + pub cmd: str, +} + +impl Future for Process { + type Output = &'static str; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<&'static str> { + let mut status = 0; + unsafe { + // TODO: fork、修改用户、组、设置限制,exec 等等 + // TODO: seccomp 等保证安全 + let _pid = libc::waitpid(self.pid, &mut status, libc::WNOHANG); + } + if status != 0 { + return Poll::Ready(""); + } + let waker = cx.waker().clone(); + + thread::spawn(move || { + thread::sleep(Duration::from_millis(3000)); + waker.wake(); + }); + + Poll::Pending + } +}