重构 grpc 格式

This commit is contained in:
MeiK 2020-09-08 10:47:36 +08:00
parent 57e6285013
commit 86395283d0
5 changed files with 81 additions and 73 deletions

View File

@ -5,58 +5,68 @@ service River {
rpc Judge(stream JudgeRequest) returns (stream JudgeResponse) {} rpc Judge(stream JudgeRequest) returns (stream JudgeResponse) {}
} }
message CompileData {
bytes code = 1;
}
message JudgeData {
bytes in_data = 1;
bytes out_data = 2;
int32 time_limit = 3;
int32 memory_limit = 4;
}
enum Language {
C = 0;
Cpp = 1;
Python = 2;
Rust = 3;
Node = 4;
TypeScript = 5;
Go = 6;
}
enum JudgeType {
Standard = 0;
// Special = 1;
}
message JudgeRequest { message JudgeRequest {
enum Language {
C = 0;
Cpp = 1;
Python = 2;
Rust = 3;
Node = 4;
TypeScript = 5;
Go = 6;
}
Language language = 1; Language language = 1;
JudgeType judge_type = 2;
string code = 2; oneof data {
CompileData compile_data = 3;
enum JudgeType { JudgeData judge_data = 4;
Standard = 0;
// Special = 1;
} }
JudgeType judge_type = 3; }
bytes in_data = 4; enum JudgeResult {
bytes out_data = 5; Accepted = 0;
WrongAnswer = 1;
TimeLimitExceeded = 2;
MemoryLimitExceeded = 3;
RuntimeError = 4;
OutputLimitExceeded = 5;
CompileError = 6;
PresentationError = 7;
SystemError = 8;
}
int32 time_limit = 6; enum JudgeStatus {
int32 memory_limit = 7; Pending = 0;
Running = 1;
Ended = 2;
} }
message JudgeResponse { message JudgeResponse {
int64 time_used = 1; int64 time_used = 1;
int64 memory_used = 2; int64 memory_used = 2;
enum JudgeResult {
Accepted = 0;
WrongAnswer = 1;
TimeLimitExceeded = 2;
MemoryLimitExceeded = 3;
RuntimeError = 4;
OutputLimitExceeded = 5;
CompileError = 6;
PresentationError = 7;
SystemError = 8;
}
JudgeResult result = 3; JudgeResult result = 3;
// int32 errno = 4; // int32 errno = 4;
// int32 exit_code = 5; // int32 exit_code = 5;
// string stdout = 6; // string stdout = 6;
// string stderr = 7; // string stderr = 7;
string errmsg = 8; string errmsg = 8;
enum JudgeStatus {
Pending = 0;
Compiling = 1;
Running = 2;
Ended = 3;
}
JudgeStatus status = 9; JudgeStatus status = 9;
} }

View File

@ -1,4 +1,4 @@
use crate::river::judge_response::{JudgeResult, JudgeStatus}; use crate::river::{JudgeResult, JudgeStatus};
use crate::river::JudgeResponse; use crate::river::JudgeResponse;
use libc::strerror; use libc::strerror;
use std::ffi::{CStr, NulError, OsString}; use std::ffi::{CStr, NulError, OsString};
@ -16,6 +16,7 @@ pub enum Error {
StringToCStringError(NulError), StringToCStringError(NulError),
OsStringToStringError(OsString), OsStringToStringError(OsString),
RemoveFileError(PathBuf), RemoveFileError(PathBuf),
UnknownRequestData,
} }
pub type Result<T> = result::Result<T, Error>; pub type Result<T> = result::Result<T, Error>;

View File

@ -1,7 +1,7 @@
use super::error::{Error, Result}; use super::error::{Error, Result};
use super::process::{Process, ProcessStatus}; use super::process::{Process, ProcessStatus};
use crate::river::judge_request::Language; use crate::river::Language;
use crate::river::judge_response::{JudgeResult, JudgeStatus}; use crate::river::{JudgeResult, JudgeStatus, CompileData, JudgeData};
use crate::river::{JudgeRequest, JudgeResponse}; use crate::river::{JudgeRequest, JudgeResponse};
use std::path::Path; use std::path::Path;
use tokio::fs; use tokio::fs;
@ -23,7 +23,7 @@ impl JudgeResponse {
} }
} }
pub async fn judger(request: &JudgeRequest, path: &Path) -> Result<JudgeResponse> { pub async fn judger(request: &JudgeRequest, data: &JudgeData, path: &Path) -> Result<JudgeResponse> {
let mut resp = JudgeResponse::new(); let mut resp = JudgeResponse::new();
let cmd = match Language::from_i32(request.language) { let cmd = match Language::from_i32(request.language) {
Some(Language::C) => "./a.out", Some(Language::C) => "./a.out",
@ -38,12 +38,13 @@ pub async fn judger(request: &JudgeRequest, path: &Path) -> Result<JudgeResponse
let mut process = Process::new(); let mut process = Process::new();
process.cmd = cmd.to_string(); process.cmd = cmd.to_string();
process.workdir = path.to_path_buf(); process.workdir = path.to_path_buf();
process.time_limit = request.time_limit; process.time_limit = data.time_limit;
process.memory_limit = request.memory_limit; process.memory_limit = data.memory_limit;
// TODO: 使用内存流替换,尽可能减少文件读写与复制
// 写入输入文件 // 写入输入文件
let in_file = path.join("stdin.txt"); let in_file = path.join("stdin.txt");
if let Err(e) = fs::write(in_file.clone(), request.in_data.clone()).await { if let Err(e) = fs::write(in_file.clone(), data.in_data.clone()).await {
return Err(Error::FileWriteError(e)); return Err(Error::FileWriteError(e));
}; };
process.stdin_file = Some(in_file.clone()); process.stdin_file = Some(in_file.clone());
@ -65,7 +66,7 @@ pub async fn judger(request: &JudgeRequest, path: &Path) -> Result<JudgeResponse
Ok(resp) Ok(resp)
} }
pub async fn compile(request: &JudgeRequest, path: &Path) -> Result<JudgeResponse> { pub async fn compile(request: &JudgeRequest, data: &CompileData, path: &Path) -> Result<JudgeResponse> {
let mut resp = JudgeResponse::new(); let mut resp = JudgeResponse::new();
// 写入代码 // 写入代码
let filename = match Language::from_i32(request.language) { let filename = match Language::from_i32(request.language) {
@ -78,7 +79,7 @@ pub async fn compile(request: &JudgeRequest, path: &Path) -> Result<JudgeRespons
Some(Language::Go) => "main.go", Some(Language::Go) => "main.go",
None => return Err(Error::LanguageNotFound(request.language)), None => return Err(Error::LanguageNotFound(request.language)),
}; };
if let Err(e) = fs::write(path.join(filename), &request.code).await { if let Err(e) = fs::write(path.join(filename), &data.code).await {
return Err(Error::FileWriteError(e)); return Err(Error::FileWriteError(e));
}; };
@ -121,9 +122,3 @@ pub fn running() -> JudgeResponse {
let resp = JudgeResponse::new(); let resp = JudgeResponse::new();
resp resp
} }
pub fn compiling() -> JudgeResponse {
let mut resp = JudgeResponse::new();
resp.status = JudgeStatus::Compiling as i32;
resp
}

View File

@ -1,8 +1,8 @@
#![recursion_limit = "256"] #![recursion_limit = "256"]
use crate::river::judge_response::JudgeResult;
use futures::StreamExt; use futures::StreamExt;
use futures_core::Stream; use futures_core::Stream;
use river::judge_request::Data;
use river::river_server::{River, RiverServer}; use river::river_server::{River, RiverServer};
use river::{JudgeRequest, JudgeResponse}; use river::{JudgeRequest, JudgeResponse};
use std::pin::Pin; use std::pin::Pin;
@ -33,9 +33,6 @@ impl River for RiverService {
let mut stream = request.into_inner(); let mut stream = request.into_inner();
let output = async_stream::try_stream! { let output = async_stream::try_stream! {
let mut need_compile = true;
// TODO: 使用锁或者资源量等机制限制并发
yield judger::pending();
// 创建评测使用的临时目录 // 创建评测使用的临时目录
// 无需主动删除,变量在 drop 之后会自动删除临时文件夹 // 无需主动删除,变量在 drop 之后会自动删除临时文件夹
@ -48,29 +45,21 @@ impl River for RiverService {
}; };
while let Some(req) = stream.next().await { while let Some(req) = stream.next().await {
// TODO: 使用锁或者资源量等机制限制并发
yield judger::pending();
let req = req?; let req = req?;
// 首次获取流进行编译
if need_compile {
yield judger::compiling();
let result = match judger::compile(&req, &pwd.path()).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(); yield judger::running();
let result = match judger::judger(&req, &pwd.path()).await { let result = match &req.data {
Some(Data::CompileData(data)) => judger::compile(&req, &data, &pwd.path()).await,
Some(Data::JudgeData(data)) => judger::judger(&req, &data, &pwd.path()).await,
_ => Err(error::Error::UnknownRequestData),
};
let result = match result {
Ok(res) => res, Ok(res) => res,
Err(e) => error::system_error(e) Err(e) => error::system_error(e)
}; };
yield result; yield result;
} }
while let Some(_) = stream.next().await {} while let Some(_) = stream.next().await {}

View File

@ -173,6 +173,19 @@ impl Drop for ExecArgs {
} }
} }
const ITIMER_REAL: libc::c_int = 0;
extern "C" {
#[cfg_attr(
all(target_os = "macos", target_arch = "x86"),
link_name = "setitimer$UNIX2003"
)]
fn setitimer(
which: libc::c_int,
new_value: *const libc::itimerval,
old_value: *mut libc::itimerval,
) -> libc::c_int;
}
fn run(process: Process) { fn run(process: Process) {
// 子进程里崩溃也无法返回,崩溃就直接崩溃了 // 子进程里崩溃也无法返回,崩溃就直接崩溃了
let exec_args = ExecArgs::build(&process.cmd).unwrap(); let exec_args = ExecArgs::build(&process.cmd).unwrap();
@ -212,7 +225,7 @@ fn run(process: Process) {
0o644, 0o644,
); );
// 墙上时钟限制 // 墙上时钟限制
if libc::setitimer(libc::ITIMER_REAL, &rt, ptr::null_mut()) == -1 { if setitimer(ITIMER_REAL, &rt, ptr::null_mut()) == -1 {
eprintln!("setitimer failure!"); eprintln!("setitimer failure!");
eprintln!("{:?}", io::Error::last_os_error().raw_os_error()); eprintln!("{:?}", io::Error::last_os_error().raw_os_error());
panic!("How dare you!"); panic!("How dare you!");