פתרון מפקח תהליכים עם מסוף ותקציב זמן

נבנה גרסת ייחוס ללינוקס באמצעות חבילת libc. הקוד הבא הוא קובץ src/main.rs מלא בפרויקט Cargo עם מהדורת 2024. בסביבות שבהן openpty נמצאת בספריית libutil נדרש לקשר אותה, ולכן הדוגמה מציינת זאת במפורש.

use std::{collections::VecDeque, env, fs::File, io, mem::MaybeUninit};
use std::os::fd::{AsFd, AsRawFd, BorrowedFd, FromRawFd, OwnedFd};
use std::os::unix::process::CommandExt;
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};

#[link(name = "util")]
unsafe extern "C" {}
const LIMIT: usize = 64 * 1024;

struct Flags<'a> { fd: BorrowedFd<'a>, original: i32 }
impl<'a> Flags<'a> {
    fn nonblocking(fd: BorrowedFd<'a>) -> io::Result<Self> {
        let original = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_GETFL) };
        if original == -1 { return Err(io::Error::last_os_error()); }
        if unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_SETFL,
            original | libc::O_NONBLOCK) } == -1 {
            return Err(io::Error::last_os_error());
        }
        Ok(Self { fd, original })
    }
}
impl Drop for Flags<'_> {
    fn drop(&mut self) {
        unsafe { libc::fcntl(self.fd.as_raw_fd(), libc::F_SETFL, self.original); }
    }
}
struct Terminal<'a> { fd: BorrowedFd<'a>, original: Option<libc::termios> }
impl<'a> Terminal<'a> {
    fn raw_if_tty(fd: BorrowedFd<'a>) -> io::Result<Self> {
        if unsafe { libc::isatty(fd.as_raw_fd()) } != 1 {
            return Ok(Self { fd, original: None });
        }
        let mut value = MaybeUninit::<libc::termios>::uninit();
        if unsafe { libc::tcgetattr(fd.as_raw_fd(), value.as_mut_ptr()) } == -1 {
            return Err(io::Error::last_os_error());
        }
        let original = unsafe { value.assume_init() };
        let mut raw = original;
        unsafe { libc::cfmakeraw(&mut raw); }
        if unsafe { libc::tcsetattr(fd.as_raw_fd(), libc::TCSANOW, &raw) } == -1 {
            return Err(io::Error::last_os_error());
        }
        Ok(Self { fd, original: Some(original) })
    }
}
impl Drop for Terminal<'_> {
    fn drop(&mut self) {
        if let Some(original) = &self.original {
            unsafe { libc::tcsetattr(self.fd.as_raw_fd(), libc::TCSANOW, original); }
        }
    }
}
struct Running { child: Child, group: libc::pid_t, reaped: bool }
impl Running {
    fn signal(&self, signal: i32) -> io::Result<()> {
        if unsafe { libc::kill(-self.group, signal) } == 0 { return Ok(()); }
        let error = io::Error::last_os_error();
        if error.raw_os_error() == Some(libc::ESRCH) { Ok(()) } else { Err(error) }
    }
}
impl Drop for Running {
    fn drop(&mut self) {
        if !self.reaped {
            let _ = self.signal(libc::SIGKILL);
            let _ = self.child.wait();
        }
    }
}
fn read_queue(fd: i32, queue: &mut VecDeque<u8>, pty: bool) -> io::Result<bool> {
    let mut buffer = [0u8; 4096];
    let count = buffer.len().min(LIMIT - queue.len());
    if count == 0 { return Ok(false); }
    let result = unsafe { libc::read(fd, buffer.as_mut_ptr().cast(), count) };
    if result > 0 { queue.extend(&buffer[..result as usize]); return Ok(false); }
    if result == 0 { return Ok(true); }
    let error = io::Error::last_os_error();
    if pty && error.raw_os_error() == Some(libc::EIO) { return Ok(true); }
    if matches!(error.kind(), io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted) {
        Ok(false)
    } else { Err(error) }
}
fn write_queue(fd: i32, queue: &mut VecDeque<u8>) -> io::Result<()> {
    let bytes = queue.as_slices().0;
    if bytes.is_empty() { return Ok(()); }
    let result = unsafe { libc::write(fd, bytes.as_ptr().cast(), bytes.len()) };
    if result > 0 { queue.drain(..result as usize); return Ok(()); }
    if result == 0 { return Err(io::ErrorKind::WriteZero.into()); }
    let error = io::Error::last_os_error();
    if matches!(error.kind(), io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted) {
        Ok(())
    } else { Err(error) }
}
fn run() -> io::Result<()> {
    let mut args = env::args_os().skip(1);
    let milliseconds: u64 = args.next().and_then(|x| x.into_string().ok())
        .and_then(|x| x.parse().ok()).ok_or_else(||
            io::Error::new(io::ErrorKind::InvalidInput, "expected timeout milliseconds"))?;
    let program = args.next().ok_or_else(||
        io::Error::new(io::ErrorKind::InvalidInput, "expected program"))?;
    let duration = Duration::from_millis(milliseconds);
    let deadline = Instant::now().checked_add(duration).ok_or_else(||
        io::Error::new(io::ErrorKind::InvalidInput, "timeout too large"))?;
    let (mut master, mut slave) = (-1, -1);
    if unsafe { libc::openpty(&mut master, &mut slave, std::ptr::null_mut(),
        std::ptr::null(), std::ptr::null()) } == -1 {
        return Err(io::Error::last_os_error());
    }
    let master = unsafe { OwnedFd::from_raw_fd(master) };
    let slave = File::from(unsafe { OwnedFd::from_raw_fd(slave) });
    for fd in [master.as_raw_fd(), slave.as_raw_fd()] {
        if unsafe { libc::fcntl(fd, libc::F_SETFD, libc::FD_CLOEXEC) } == -1 {
            return Err(io::Error::last_os_error());
        }
    }
    let mut command = Command::new(program);
    command.args(args).stdin(Stdio::from(slave.try_clone()?))
        .stdout(Stdio::from(slave.try_clone()?))
        .stderr(Stdio::from(slave.try_clone()?));
    unsafe {
        command.pre_exec(|| {
            if libc::setsid() == -1 { return Err(io::Error::last_os_error()); }
            if libc::ioctl(0, libc::TIOCSCTTY, 0) == -1 {
                return Err(io::Error::last_os_error());
            }
            if libc::tcsetpgrp(0, libc::getpgrp()) == -1 {
                return Err(io::Error::last_os_error());
            }
            Ok(())
        });
    }
    let child = command.spawn()?;
    let group = child.id() as libc::pid_t;
    let mut running = Running { child, group, reaped: false };
    drop(command);
    drop(slave);
    let stdin = io::stdin();
    let stdout = io::stdout();
    let _terminal = Terminal::raw_if_tty(stdin.as_fd())?;
    let _input_flags = Flags::nonblocking(stdin.as_fd())?;
    let _output_flags = Flags::nonblocking(stdout.as_fd())?;
    let _master_flags = Flags::nonblocking(master.as_fd())?;
    let mut to_child = VecDeque::new();
    let mut to_output = VecDeque::new();
    let (mut input_eof, mut pty_eof) = (false, false);
    let mut terminate_at = None;
    let mut killed = false;
    let mut drain_until = None;
    let mut status = None;
    let mut truncated = false;
    loop {
        let now = Instant::now();
        if !running.reaped {
            if let Some(done) = running.child.try_wait()? {
                running.reaped = true;
                status = Some(done);
                drain_until = Some(now + Duration::from_secs(1));
                to_child.clear();
            } else if now >= deadline && terminate_at.is_none() {
                running.signal(libc::SIGTERM)?;
                terminate_at = Some(now + Duration::from_millis(500));
            } else if terminate_at.is_some_and(|time| now >= time) && !killed {
                running.signal(libc::SIGKILL)?;
                killed = true;
            }
        }
        if running.reaped && pty_eof && to_output.is_empty() { break; }
        if drain_until.is_some_and(|time| now >= time) {
            truncated = !pty_eof || !to_output.is_empty();
            break;
        }
        let mut descriptors = [
            libc::pollfd { fd: master.as_raw_fd(), events: 0, revents: 0 },
            libc::pollfd { fd: if input_eof || running.reaped { -1 } else { 0 },
                           events: 0, revents: 0 },
            libc::pollfd { fd: 1, events: 0, revents: 0 },
        ];
        if !pty_eof && to_output.len() < LIMIT { descriptors[0].events |= libc::POLLIN; }
        if !pty_eof && !to_child.is_empty() { descriptors[0].events |= libc::POLLOUT; }
        if to_child.len() < LIMIT { descriptors[1].events |= libc::POLLIN; }
        if !to_output.is_empty() { descriptors[2].events |= libc::POLLOUT; }
        if descriptors[0].events == 0 { descriptors[0].fd = -1; }
        let result = unsafe { libc::poll(descriptors.as_mut_ptr(), 3, 25) };
        if result == -1 {
            let error = io::Error::last_os_error();
            if error.kind() == io::ErrorKind::Interrupted { continue; }
            return Err(error);
        }
        if descriptors.iter().any(|p| p.revents & libc::POLLNVAL != 0) {
            return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid descriptor"));
        }
        if descriptors[0].revents & (libc::POLLIN | libc::POLLHUP | libc::POLLERR) != 0
            && !pty_eof && to_output.len() < LIMIT {
            pty_eof = read_queue(master.as_raw_fd(), &mut to_output, true)?;
        }
        if descriptors[1].revents & (libc::POLLIN | libc::POLLHUP) != 0 {
            input_eof = read_queue(0, &mut to_child, false)?;
        }
        if descriptors[0].revents & libc::POLLOUT != 0 && !pty_eof {
            write_queue(master.as_raw_fd(), &mut to_child)?;
        }
        if descriptors[2].revents & libc::POLLOUT != 0 {
            write_queue(1, &mut to_output)?;
        }
        if descriptors[2].revents & (libc::POLLERR | libc::POLLHUP) != 0 {
            return Err(io::Error::new(io::ErrorKind::BrokenPipe, "output closed"));
        }
    }
    drop(_master_flags);
    drop(_output_flags);
    drop(_input_flags);
    drop(_terminal);
    eprintln!("status={status:?} timeout={} truncated={truncated}", terminate_at.is_some());
    if truncated { Err(io::Error::other("output drain budget exhausted")) } else { Ok(()) }
}
fn main() {
    if let Err(error) = run() { eprintln!("supervisor: {error}"); std::process::exit(1); }
}

הקוד משנה את דגלי החסימה של הפתיחות שקשורות ל־stdin ול־stdout, ולכן הוא מיועד לכלי מסוף עצמאי ולא לפונקציה שמוטמעת בתוך תהליך עם משתמשים מקבילים באותם מתארים. השומרים מחזירים את הדגלים לפני הדפסת הדוח, וה־guard של הילד מטפל בכשל שמתרחש אחרי ההפעלה. השחזור ב־Drop הוא ניסיון ניקוי ואינו מחליף דיווח מפורש על כשל שחזור במוצר.

הקריאה openpty והגדרת close-on-exec נעשות לפני יצירת תהליכונים ובתוכנית הזאת אין יוצר תהליכים מקבילי. לכן החלון שבין היצירה להגדרת הדגל אינו מאפשר כאן למישהו אחר לבצע fork במקביל. אם נהפוך את הכלי לספרייה בתוך שרת רב־תהליכוני, נצטרך לשנות את פרוטוקול יצירת המתארים ולא להעתיק את ההנחה הזאת.

ה־callback של pre_exec מבצע רק הכנת session ומסוף דרך ממשקי לינוקס, ללא הקצאה או לוג. השימוש ב־ioctl כאן הוא מסלול לינוקס מסוים, ולכן אינו טענה שכל פונקציית ioctl בכל ספרייה או יעד מתאימה לחלון הזה. כל הנתונים ושכפולי הקבצים הוכנו לפני יצירת הילד.

התורים מוגבלים ל־64 קילובייט, וכאשר תור הפלט מלא אנחנו מפסיקים לקרוא מה־PTY עד שהתפנה מקום. כך stdout איטי מפעיל backpressure במקום להגדיל זיכרון בלי גבול. המתארים אינם חוסמים, ולכן מוכנות שהתיישנה אינה משאירה את הלולאה תקועה בכתיבה. תוצאה חלקית מסירה רק את הבייטים שנכתבו בפועל.

סיום המנהיג אינו מוכיח שכל צאצא סיים. גרסת הייחוס מפסיקה לשלוח אותות לקבוצה אחרי איסוף המנהיג, ומאפשרת שנייה לניקוז לפני שהיא מדווחת על קטיעה. זאת מדיניות מפורשת שמונעת שימוש מאוחר ב־PGID ממוחזר, אבל אינה מנהלת צאצאים עוינים או daemon שהתנתק. הרחבת הבידוד בפרק העשירי תוסיף גבול ניהול מתאים באמצעות cgroups.

הקוד אינו מעביר שינויי גודל מסוף, ולכן תוכנית מסך מלאה היא תרחיש הרחבה מוגדר: יש לקבל שינוי גודל בהקשר בטוח, לקרוא TIOCGWINSZ ולעדכן את צד ה־PTY באמצעות TIOCSWINSZ. תנאי הבסיס כאן הוא תוכנית אינטראקטיבית שורתית, כגון shell או כלי שמבקש קלט, ולא אמולטור מסוף מלא.

בדקו פלט גדול עם תוכנית שמדפיסה הרבה שורות, ולא באמצעות טקסט שמכיל רצפי שליטה לא מוכרים ממקור חיצוני. בדקו timeout עם תוכנית עזר שמתעלמת מ־SIGTERM, וודאו שהדוח מסמן את נתיב הזמן. שימו לב ש־SIGKILL אינו מבטיח זמן קיר עליון לסיום פעולה במצב קרנל שאינו ניתן להפרעה, ולכן הדוח אינו טוען להבטחת זמן אמת.