完成异步子进程运行与等待功能

This commit is contained in:
MeiK 2020-08-31 17:46:44 +08:00
parent 55d090af9d
commit 45af94494a
3 changed files with 126 additions and 45 deletions

View File

@ -11,12 +11,12 @@ pub enum Error {
CreateTempDirError(io::Error),
LanguageNotFound(i32),
FileWriteError(io::Error),
ForkError(Option<i32>),
ChannelRecvError,
}
pub type Result<T> = result::Result<T, Error>;
pub fn errno_str(errno: Option<i32>) -> String {
pub fn _errno_str(errno: Option<i32>) -> String {
match errno {
Some(no) => {
let stre = unsafe { strerror(no) };
@ -30,10 +30,6 @@ pub fn errno_str(errno: Option<i32>) -> 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),
}
}

View File

@ -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<JudgeRespons
return Err(Error::FileWriteError(e));
};
let process = Process::new();
let status = process.await?;
println!("{}", status.exit_code);
return Ok(JudgeResponse {
time_used: request.time_limit,
memory_used: 2,

View File

@ -1,10 +1,12 @@
use super::error::{Error, Result};
use libc;
use std::future::Future;
use std::io;
use std::pin::Pin;
use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use std::thread;
use std::time::SystemTime;
#[derive(Clone)]
pub struct Process {
@ -15,10 +17,13 @@ pub struct Process {
pub stdout_fd: i32,
pub stderr_fd: i32,
pub cmd: String,
tx: Arc<Mutex<mpsc::Sender<ProcessStatus>>>,
rx: Arc<Mutex<mpsc::Receiver<ProcessStatus>>>,
}
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<ProcessStatus>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<ProcessStatus>> {
// 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<Result<ProcessStatus>> {
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,
}