From 45af94494a0551e5c430b194160a7e95610b65a7 Mon Sep 17 00:00:00 2001 From: MeiK Date: Mon, 31 Aug 2020 17:46:44 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E6=88=90=E5=BC=82=E6=AD=A5=E5=AD=90?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E8=BF=90=E8=A1=8C=E4=B8=8E=E7=AD=89=E5=BE=85?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/error.rs | 8 +-- src/judger.rs | 5 ++ src/process.rs | 158 +++++++++++++++++++++++++++++++++++++------------ 3 files changed, 126 insertions(+), 45 deletions(-) diff --git a/src/error.rs b/src/error.rs index 7d5785d..3d11dba 100644 --- a/src/error.rs +++ b/src/error.rs @@ -11,12 +11,12 @@ pub enum Error { CreateTempDirError(io::Error), LanguageNotFound(i32), FileWriteError(io::Error), - ForkError(Option), + ChannelRecvError, } pub type Result = result::Result; -pub fn errno_str(errno: Option) -> String { +pub fn _errno_str(errno: Option) -> String { match errno { Some(no) => { let stre = unsafe { strerror(no) }; @@ -30,10 +30,6 @@ pub fn errno_str(errno: Option) -> String { impl fmt::Display for Error { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { match *self { - Error::ForkError(errno) => { - let reason = errno_str(errno); - write!(f, "ForkError: {}", reason) - } _ => write!(f, "{:?}", self), } } diff --git a/src/judger.rs b/src/judger.rs index 58853df..71ddce2 100644 --- a/src/judger.rs +++ b/src/judger.rs @@ -1,4 +1,5 @@ use super::error::{Error, Result}; +use super::process::Process; use crate::river::judge_request::Language; use crate::river::judge_response::{JudgeResult, JudgeStatus}; use crate::river::{JudgeRequest, JudgeResponse}; @@ -35,6 +36,10 @@ pub async fn compile(request: &JudgeRequest, path: &Path) -> Result>>, + rx: Arc>>, } impl Process { pub fn new() -> Process { + let (tx, rx) = mpsc::channel(); Process { pid: -1, time_limit: -1, @@ -27,6 +32,8 @@ impl Process { stdout_fd: -1, stderr_fd: -1, cmd: "".to_string(), + tx: Arc::new(Mutex::new(tx)), + rx: Arc::new(Mutex::new(rx)), } } pub fn set_pid(&mut self, pid: i32) { @@ -37,55 +44,128 @@ impl Process { impl Future for Process { type Output = Result; - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - // pid == -1 时为第一次触发,此时开始运行 - // 直到进程执行完毕, wake 通知进入下次检查 - if self.pid == -1 { - let pid; - unsafe { - pid = libc::fork(); - } - if pid == 0 { - // 子进程 - // TODO: fork、修改用户、组、设置限制,exec 等等 - // TODO: seccomp 等保证安全 - let process = Pin::into_inner(self).clone(); - run(process); - } else if pid > 0 { - // 父进程 - self.as_mut().set_pid(pid); - let waker = cx.waker().clone(); - let pid = self.pid; - - thread::spawn(move || { - // 等待子进程结束 - wait(pid); - // 触发唤醒异步 + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let process = Pin::into_inner(self); + // 如果 pid == -1,则说明子进程还没有运行,开始进程 + if process.pid == -1 { + let (tx, rx) = mpsc::channel(); + let process_clone = process.clone(); + let waker = cx.waker().clone(); + thread::spawn(move || { + let pid; + unsafe { + pid = libc::fork(); + } + if pid == 0 { + run(process_clone); + } else if pid > 0 { + tx.send(pid).unwrap(); + let status = wait(pid); + let status_tx = process_clone.tx.lock().unwrap(); + status_tx.send(status).unwrap(); waker.wake(); - }); - } else { - // 出错 - return Poll::Ready(Err(Error::ForkError( - io::Error::last_os_error().raw_os_error(), - ))); - } + } else { + panic!("How dare you!"); + } + }); + + // 等待子线程启动子进程并返回 pid + let pid = match rx.recv() { + Ok(val) => val, + Err(_) => return Poll::Ready(Err(Error::ChannelRecvError)), + }; + process.set_pid(pid); return Poll::Pending; } else { - // 子进程结束后被唤醒 - // 收集运行信息,返回数据,结束异步 - wait(self.pid); - // TODO: Poll::Ready - return Poll::Pending; + // 再次进入 poll,说明子进程已经结束,通知了 wake + // 此时 channel 应该是有数据的 + let status = match process.rx.lock() { + Ok(rx) => match rx.recv() { + Ok(val) => val, + Err(_) => return Poll::Ready(Err(Error::ChannelRecvError)), + }, + Err(_) => return Poll::Ready(Err(Error::ChannelRecvError)), + }; + return Poll::Ready(Ok(status)); } } } -fn run(process: Process) {} +fn run(_process: Process) { + unsafe { + // TODO + println!("run"); + libc::exit(1); + } +} -fn wait(pid: i32) {} +fn wait(pid: i32) -> ProcessStatus { + let start = SystemTime::now(); + let mut status = 0; + let mut rusage = libc::rusage { + ru_utime: libc::timeval { + tv_sec: 0 as libc::time_t, + tv_usec: 0 as libc::suseconds_t, + }, + ru_stime: libc::timeval { + tv_sec: 0 as libc::time_t, + tv_usec: 0 as libc::suseconds_t, + }, + ru_maxrss: 0 as libc::c_long, + ru_ixrss: 0 as libc::c_long, + ru_idrss: 0 as libc::c_long, + ru_isrss: 0 as libc::c_long, + ru_minflt: 0 as libc::c_long, + ru_majflt: 0 as libc::c_long, + ru_nswap: 0 as libc::c_long, + ru_inblock: 0 as libc::c_long, + ru_oublock: 0 as libc::c_long, + ru_msgsnd: 0 as libc::c_long, + ru_msgrcv: 0 as libc::c_long, + ru_nsignals: 0 as libc::c_long, + ru_nvcsw: 0 as libc::c_long, + ru_nivcsw: 0 as libc::c_long, + }; + unsafe { + libc::waitpid(pid, &mut status, 0); + libc::getrusage(pid, &mut rusage); + } + let mut exit_code = 0; + let exited = unsafe { libc::WIFEXITED(status) }; + if exited { + exit_code = unsafe { libc::WEXITSTATUS(status) }; + } + let signal = unsafe { + if libc::WIFSIGNALED(status) { + libc::WTERMSIG(status) + } else if libc::WIFSTOPPED(status) { + libc::WSTOPSIG(status) + } else { + 0 + } + }; + let real_time_used = match start.elapsed() { + Ok(elapsed) => elapsed.as_millis(), + // 这种地方如果出错了,确实没有办法解决 + // 只能崩溃再见了 + // How dare you! + Err(_) => panic!("How dare you!"), + }; + return ProcessStatus { + rusage: rusage, + exit_code: exit_code, + status: status, + signal: signal, + real_time_used: real_time_used, + }; +} + +#[derive(Clone)] pub struct ProcessStatus { pub rusage: libc::rusage, pub exit_code: i32, pub status: i32, + pub signal: i32, + pub real_time_used: u128, }