yash_env/system/
shared.rs

1// This file is part of yash, an extended POSIX shell.
2// Copyright (C) 2021 WATANABE Yuki
3//
4// This program is free software: you can redistribute it and/or modify
5// it under the terms of the GNU General Public License as published by
6// the Free Software Foundation, either version 3 of the License, or
7// (at your option) any later version.
8//
9// This program is distributed in the hope that it will be useful,
10// but WITHOUT ANY WARRANTY; without even the implied warranty of
11// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12// GNU General Public License for more details.
13//
14// You should have received a copy of the GNU General Public License
15// along with this program.  If not, see <https://www.gnu.org/licenses/>.
16
17//! [`SharedSystem`] and related items
18
19use super::CaughtSignals;
20use super::Chdir;
21use super::ChildProcessStarter;
22use super::Clock;
23use super::CpuTimes;
24use super::Dir;
25use super::Disposition;
26use super::Dup;
27use super::Errno;
28use super::Exec;
29use super::Exit;
30use super::Fcntl;
31use super::FdFlag;
32use super::Fork;
33use super::Fstat;
34use super::GetCwd;
35use super::GetPid;
36use super::GetPw;
37use super::GetRlimit;
38use super::GetSigaction;
39use super::GetUid;
40use super::Gid;
41use super::IsExecutableFile;
42use super::Isatty;
43use super::LimitPair;
44use super::Mode;
45use super::OfdAccess;
46use super::Open;
47use super::OpenFlag;
48use super::Path;
49use super::PathBuf;
50use super::Pipe;
51use super::Read;
52use super::Resource;
53use super::Result;
54use super::Seek;
55use super::Select;
56use super::SelectSystem;
57use super::SendSignal;
58use super::SetPgid;
59use super::SetRlimit;
60use super::ShellPath;
61use super::Sigaction;
62use super::Sigmask;
63use super::SigmaskOp;
64use super::SignalStatus;
65use super::SignalSystem;
66use super::Signals;
67use super::Stat;
68use super::Sysconf;
69use super::TcGetPgrp;
70use super::TcSetPgrp;
71use super::Times;
72use super::Uid;
73use super::Umask;
74use super::UnixString;
75use super::Wait;
76use super::Write;
77use super::signal;
78#[cfg(doc)]
79use crate::Env;
80use crate::io::Fd;
81use crate::job::Pid;
82use crate::job::ProcessState;
83use crate::semantics::ExitStatus;
84use crate::system::Close;
85use enumset::EnumSet;
86use std::cell::RefCell;
87use std::convert::Infallible;
88use std::ffi::CStr;
89use std::ffi::CString;
90use std::ffi::c_int;
91use std::future::poll_fn;
92use std::io::SeekFrom;
93use std::ops::RangeInclusive;
94use std::rc::Rc;
95use std::task::Poll;
96use std::time::Duration;
97use std::time::Instant;
98
99/// System shared by a reference counter.
100///
101/// A `SharedSystem` is a reference-counted container of a [`System`] instance
102/// accompanied with an internal state for supporting asynchronous interactions
103/// with the system. As it is reference-counted, cloning a `SharedSystem`
104/// instance only increments the reference count without cloning the backing
105/// system instance. This behavior allows calling `SharedSystem`'s methods
106/// concurrently from different `async` tasks that each have a `SharedSystem`
107/// instance sharing the same state.
108///
109/// `SharedSystem` implements [`System`] by delegating to the contained system
110/// instance of type `S`. You should avoid calling some of the `System` methods,
111/// however. Prefer `async` functions provided by `SharedSystem` (e.g.,
112/// [`read_async`](Self::read_async)) over raw system functions (e.g.,
113/// [`read`](Read::read)).
114///
115/// The following example illustrates how multiple concurrent tasks are run in a
116/// single-threaded pool:
117///
118/// ```
119/// # use yash_env::{SharedSystem, VirtualSystem};
120/// # use yash_env::system::Pipe as _;
121/// # use futures_util::task::LocalSpawnExt;
122/// let system = SharedSystem::new(VirtualSystem::new());
123/// let system2 = system.clone();
124/// let system3 = system.clone();
125/// let (reader, writer) = system.pipe().unwrap();
126/// let mut executor = futures_executor::LocalPool::new();
127///
128/// // We add a task that tries to read from the pipe, but nothing has been
129/// // written to it, so the task is stalled.
130/// let read_task = executor.spawner().spawn_local_with_handle(async move {
131///     let mut buffer = [0; 1];
132///     system.read_async(reader, &mut buffer).await.unwrap();
133///     buffer[0]
134/// });
135/// executor.run_until_stalled();
136///
137/// // Let's add a task that writes to the pipe.
138/// executor.spawner().spawn_local(async move {
139///     system2.write_all(writer, &[123]).await.unwrap();
140/// });
141/// executor.run_until_stalled();
142///
143/// // The write task has written a byte to the pipe, but the read task is still
144/// // stalled. We need to wake it up by calling `select`.
145/// system3.select(false).unwrap();
146///
147/// // Now the read task can proceed to the end.
148/// let number = executor.run_until(read_task.unwrap());
149/// assert_eq!(number, 123);
150/// ```
151///
152/// If there is a child process in the [`VirtualSystem`], you should call
153/// [`SystemState::select_all`](super::virtual::SystemState::select_all) in
154/// addition to [`SharedSystem::select`] so that the child process task is woken
155/// up when needed.
156/// (TBD code example)
157///
158/// [`System`]: crate::system::System
159/// [`VirtualSystem`]: crate::system::virtual::VirtualSystem
160#[derive(Debug)]
161pub struct SharedSystem<S>(pub(super) Rc<RefCell<SelectSystem<S>>>);
162
163impl<S> SharedSystem<S> {
164    /// Creates a new shared system.
165    pub fn new(system: S) -> Self {
166        SharedSystem(Rc::new(RefCell::new(SelectSystem::new(system))))
167    }
168
169    /// Reads from the file descriptor.
170    ///
171    /// This function waits for one or more bytes to be available for reading.
172    /// If successful, returns the number of bytes read.
173    pub async fn read_async(&self, fd: Fd, buffer: &mut [u8]) -> Result<usize>
174    where
175        S: Fcntl + Read,
176    {
177        let was_nonblocking = self.get_and_set_nonblocking(fd, true)?;
178
179        // We need to retain a strong reference to the waker outside the poll_fn
180        // function because SelectSystem only retains a weak reference to it.
181        // This allows SelectSystem to discard defunct wakers if this async task
182        // is aborted.
183        let waker = Rc::new(RefCell::new(None));
184
185        let result = poll_fn(|context| {
186            let mut inner = self.0.borrow_mut();
187            match inner.read(fd, buffer) {
188                Err(Errno::EAGAIN) => {
189                    *waker.borrow_mut() = Some(context.waker().clone());
190                    inner.add_reader(fd, Rc::downgrade(&waker));
191                    Poll::Pending
192                }
193                result => Poll::Ready(result),
194            }
195        })
196        .await;
197
198        self.get_and_set_nonblocking(fd, was_nonblocking).ok();
199
200        result
201    }
202
203    /// Writes to the file descriptor.
204    ///
205    /// This function calls [`write`](Write::write) repeatedly until the whole
206    /// `buffer` is written to the FD. If the `buffer` is empty, `write` is not
207    /// called at all, so any error that would be returned from `write` is not
208    /// returned.
209    ///
210    /// This function silently ignores signals that may interrupt writes.
211    pub async fn write_all(&self, fd: Fd, mut buffer: &[u8]) -> Result<usize>
212    where
213        S: Fcntl + Write,
214    {
215        if buffer.is_empty() {
216            return Ok(0);
217        }
218
219        let was_nonblocking = self.get_and_set_nonblocking(fd, true)?;
220        let mut written = 0;
221
222        // We need to retain a strong reference to the waker outside the poll_fn
223        // function because SelectSystem only retains a weak reference to it.
224        // This allows SelectSystem to discard defunct wakers if this async task
225        // is aborted.
226        let waker = Rc::new(RefCell::new(None));
227
228        let result = poll_fn(|context| {
229            let mut inner = self.0.borrow_mut();
230            match inner.write(fd, buffer) {
231                Ok(count) => {
232                    written += count;
233                    buffer = &buffer[count..];
234                    if buffer.is_empty() {
235                        return Poll::Ready(Ok(written));
236                    }
237                }
238                Err(Errno::EAGAIN | Errno::EINTR) => (),
239                Err(error) => return Poll::Ready(Err(error)),
240            }
241
242            *waker.borrow_mut() = Some(context.waker().clone());
243            inner.add_writer(fd, Rc::downgrade(&waker));
244            Poll::Pending
245        })
246        .await;
247
248        self.get_and_set_nonblocking(fd, was_nonblocking).ok();
249
250        result
251    }
252
253    /// Convenience function for printing a message to the standard error
254    pub async fn print_error(&self, message: &str)
255    where
256        S: Fcntl + Write,
257    {
258        _ = self.write_all(Fd::STDERR, message.as_bytes()).await;
259    }
260
261    /// Waits until the specified time point.
262    pub async fn wait_until(&self, target: Instant)
263    where
264        S: Clock,
265    {
266        // We need to retain a strong reference to the waker outside the poll_fn
267        // function because SelectSystem only retains a weak reference to it.
268        // This allows SelectSystem to discard defunct wakers if this async task
269        // is aborted.
270        let waker = Rc::new(RefCell::new(None));
271
272        poll_fn(|context| {
273            let mut system = self.0.borrow_mut();
274            let now = system.now();
275            if now >= target {
276                return Poll::Ready(());
277            }
278            *waker.borrow_mut() = Some(context.waker().clone());
279            system.add_timeout(target, Rc::downgrade(&waker));
280            Poll::Pending
281        })
282        .await
283    }
284
285    /// Waits for some signals to be delivered to this process.
286    ///
287    /// Before calling this function, you need to [set the signal
288    /// disposition](Self::set_disposition) to `Catch`. Without doing so, this
289    /// function cannot detect the receipt of the signals.
290    ///
291    /// Returns an array of signals that were caught.
292    ///
293    /// If this `SharedSystem` is part of an [`Env`], you should call
294    /// [`Env::wait_for_signals`] rather than calling this function directly
295    /// so that the trap set can remember the caught signal.
296    pub async fn wait_for_signals(&self) -> Rc<[signal::Number]> {
297        let status = self.0.borrow_mut().add_signal_waker();
298        poll_fn(|context| {
299            let mut status = status.borrow_mut();
300            let dummy_status = SignalStatus::Expected(None);
301            let old_status = std::mem::replace(&mut *status, dummy_status);
302            match old_status {
303                SignalStatus::Caught(signals) => Poll::Ready(signals),
304                SignalStatus::Expected(_) => {
305                    *status = SignalStatus::Expected(Some(context.waker().clone()));
306                    Poll::Pending
307                }
308            }
309        })
310        .await
311    }
312
313    /// Waits for a signal to be delivered to this process.
314    ///
315    /// Before calling this function, you need to [set the signal
316    /// disposition](Self::set_disposition) to `Catch`.
317    /// Without doing so, this function cannot detect the receipt of the signal.
318    ///
319    /// If this `SharedSystem` is part of an [`Env`], you should call
320    /// [`Env::wait_for_signal`] rather than calling this function directly
321    /// so that the trap set can remember the caught signal.
322    pub async fn wait_for_signal(&self, signal: signal::Number) {
323        while !self.wait_for_signals().await.contains(&signal) {}
324    }
325
326    /// Waits for a next event to occur.
327    ///
328    /// This function calls [`Select::select`] with arguments computed from the
329    /// current internal state of the `SharedSystem`. It will wake up tasks
330    /// waiting for the file descriptor to be ready in
331    /// [`read_async`](Self::read_async) and [`write_all`](Self::write_all) or
332    /// for a signal to be caught in [`wait_for_signal`](Self::wait_for_signal).
333    /// If no tasks are woken for FDs or signals and `poll` is false, this
334    /// function will block until the first task waiting for a specific time
335    /// point is woken.
336    ///
337    /// If poll is true, this function does not block, so it may not wake up any
338    /// tasks.
339    ///
340    /// This function may wake up a task even if the condition it is expecting
341    /// has not yet been met.
342    pub fn select(&self, poll: bool) -> Result<()>
343    where
344        S: Select + CaughtSignals + Clock,
345    {
346        self.0.borrow_mut().select(poll)
347    }
348
349    /// Creates a new child process.
350    ///
351    /// See [`Fork::new_child_process`] for details.
352    pub fn new_child_process(&self) -> Result<ChildProcessStarter<S>>
353    where
354        S: Fork,
355    {
356        self.0.borrow().new_child_process()
357    }
358}
359
360impl<S> Clone for SharedSystem<S> {
361    fn clone(&self) -> Self {
362        SharedSystem(self.0.clone())
363    }
364}
365
366/// Delegates `Fstat` methods to the contained implementor.
367impl<T: Fstat> Fstat for SharedSystem<T> {
368    fn fstat(&self, fd: Fd) -> Result<Stat> {
369        self.0.borrow().fstat(fd)
370    }
371    fn fstatat(&self, dir_fd: Fd, path: &CStr, follow_symlinks: bool) -> Result<Stat> {
372        self.0.borrow().fstatat(dir_fd, path, follow_symlinks)
373    }
374}
375
376/// Delegates `IsExecutableFile` methods to the contained implementor.
377impl<T: IsExecutableFile> IsExecutableFile for SharedSystem<T> {
378    fn is_executable_file(&self, path: &CStr) -> bool {
379        self.0.borrow().is_executable_file(path)
380    }
381}
382
383/// Delegates `Pipe` methods to the contained implementor.
384impl<T: Pipe> Pipe for SharedSystem<T> {
385    fn pipe(&self) -> Result<(Fd, Fd)> {
386        self.0.borrow().pipe()
387    }
388}
389
390/// Delegates `Dup` methods to the contained implementor.
391impl<T: Dup> Dup for SharedSystem<T> {
392    fn dup(&self, from: Fd, to_min: Fd, flags: EnumSet<FdFlag>) -> Result<Fd> {
393        self.0.borrow().dup(from, to_min, flags)
394    }
395    fn dup2(&self, from: Fd, to: Fd) -> Result<Fd> {
396        self.0.borrow().dup2(from, to)
397    }
398}
399
400impl<T: Open> Open for SharedSystem<T> {
401    fn open(
402        &self,
403        path: &CStr,
404        access: OfdAccess,
405        flags: EnumSet<OpenFlag>,
406        mode: Mode,
407    ) -> Result<Fd> {
408        self.0.borrow().open(path, access, flags, mode)
409    }
410    fn open_tmpfile(&self, parent_dir: &Path) -> Result<Fd> {
411        self.0.borrow().open_tmpfile(parent_dir)
412    }
413    fn fdopendir(&self, fd: Fd) -> Result<impl Dir + use<T>> {
414        self.0.borrow().fdopendir(fd)
415    }
416    fn opendir(&self, path: &CStr) -> Result<impl Dir + use<T>> {
417        self.0.borrow().opendir(path)
418    }
419}
420
421impl<T: Close> Close for SharedSystem<T> {
422    fn close(&self, fd: Fd) -> Result<()> {
423        self.0.borrow().close(fd)
424    }
425}
426
427/// Delegates `Fcntl` methods to the contained implementor.
428impl<T: Fcntl> Fcntl for SharedSystem<T> {
429    fn ofd_access(&self, fd: Fd) -> Result<OfdAccess> {
430        self.0.borrow().ofd_access(fd)
431    }
432    fn get_and_set_nonblocking(&self, fd: Fd, nonblocking: bool) -> Result<bool> {
433        self.0.borrow().get_and_set_nonblocking(fd, nonblocking)
434    }
435    fn fcntl_getfd(&self, fd: Fd) -> Result<EnumSet<FdFlag>> {
436        self.0.borrow().fcntl_getfd(fd)
437    }
438    fn fcntl_setfd(&self, fd: Fd, flags: EnumSet<FdFlag>) -> Result<()> {
439        self.0.borrow().fcntl_setfd(fd, flags)
440    }
441}
442
443/// Delegates `Read` methods to the contained implementor.
444impl<T: Read> Read for SharedSystem<T> {
445    fn read(&self, fd: Fd, buffer: &mut [u8]) -> Result<usize> {
446        self.0.borrow().read(fd, buffer)
447    }
448}
449
450/// Delegates `Write` methods to the contained implementor.
451impl<T: Write> Write for SharedSystem<T> {
452    fn write(&self, fd: Fd, buffer: &[u8]) -> Result<usize> {
453        self.0.borrow().write(fd, buffer)
454    }
455}
456
457/// Delegates `Seek` methods to the contained implementor.
458impl<T: Seek> Seek for SharedSystem<T> {
459    fn lseek(&self, fd: Fd, position: SeekFrom) -> Result<u64> {
460        self.0.borrow().lseek(fd, position)
461    }
462}
463
464/// Delegates `Umask` methods to the contained implementor.
465impl<T: Umask> Umask for SharedSystem<T> {
466    fn umask(&self, new_mask: Mode) -> Mode {
467        self.0.borrow().umask(new_mask)
468    }
469}
470
471/// Delegates `GetCwd` methods to the contained implementor.
472impl<T: GetCwd> GetCwd for SharedSystem<T> {
473    fn getcwd(&self) -> Result<PathBuf> {
474        self.0.borrow().getcwd()
475    }
476}
477
478/// Delegates `Chdir` methods to the contained implementor.
479impl<T: Chdir> Chdir for SharedSystem<T> {
480    fn chdir(&self, path: &CStr) -> Result<()> {
481        self.0.borrow().chdir(path)
482    }
483}
484
485/// Delegates `Time` methods to the contained implementor.
486impl<T: Clock> Clock for SharedSystem<T> {
487    fn now(&self) -> Instant {
488        self.0.borrow().now()
489    }
490}
491
492/// Delegates `Times` methods to the contained implementor.
493impl<T: Times> Times for SharedSystem<T> {
494    fn times(&self) -> Result<CpuTimes> {
495        self.0.borrow().times()
496    }
497}
498
499/// Delegates `GetPid` methods to the contained implementor.
500impl<T: GetPid> GetPid for SharedSystem<T> {
501    fn getsid(&self, pid: Pid) -> Result<Pid> {
502        self.0.borrow().getsid(pid)
503    }
504
505    fn getpid(&self) -> Pid {
506        self.0.borrow().getpid()
507    }
508
509    fn getppid(&self) -> Pid {
510        self.0.borrow().getppid()
511    }
512
513    fn getpgrp(&self) -> Pid {
514        self.0.borrow().getpgrp()
515    }
516}
517
518/// Delegates `SetPgid` methods to the contained implementor.
519impl<T: SetPgid> SetPgid for SharedSystem<T> {
520    fn setpgid(&self, pid: Pid, pgid: Pid) -> Result<()> {
521        self.0.borrow().setpgid(pid, pgid)
522    }
523}
524
525/// Delegates `Signals` methods to the contained implementor.
526impl<T: Signals> Signals for SharedSystem<T> {
527    const SIGABRT: signal::Number = T::SIGABRT;
528    const SIGALRM: signal::Number = T::SIGALRM;
529    const SIGBUS: signal::Number = T::SIGBUS;
530    const SIGCHLD: signal::Number = T::SIGCHLD;
531    const SIGCLD: Option<signal::Number> = T::SIGCLD;
532    const SIGCONT: signal::Number = T::SIGCONT;
533    const SIGEMT: Option<signal::Number> = T::SIGEMT;
534    const SIGFPE: signal::Number = T::SIGFPE;
535    const SIGHUP: signal::Number = T::SIGHUP;
536    const SIGILL: signal::Number = T::SIGILL;
537    const SIGINFO: Option<signal::Number> = T::SIGINFO;
538    const SIGINT: signal::Number = T::SIGINT;
539    const SIGIO: Option<signal::Number> = T::SIGIO;
540    const SIGIOT: signal::Number = T::SIGIOT;
541    const SIGKILL: signal::Number = T::SIGKILL;
542    const SIGLOST: Option<signal::Number> = T::SIGLOST;
543    const SIGPIPE: signal::Number = T::SIGPIPE;
544    const SIGPOLL: Option<signal::Number> = T::SIGPOLL;
545    const SIGPROF: signal::Number = T::SIGPROF;
546    const SIGPWR: Option<signal::Number> = T::SIGPWR;
547    const SIGQUIT: signal::Number = T::SIGQUIT;
548    const SIGSEGV: signal::Number = T::SIGSEGV;
549    const SIGSTKFLT: Option<signal::Number> = T::SIGSTKFLT;
550    const SIGSTOP: signal::Number = T::SIGSTOP;
551    const SIGSYS: signal::Number = T::SIGSYS;
552    const SIGTERM: signal::Number = T::SIGTERM;
553    const SIGTHR: Option<signal::Number> = T::SIGTHR;
554    const SIGTRAP: signal::Number = T::SIGTRAP;
555    const SIGTSTP: signal::Number = T::SIGTSTP;
556    const SIGTTIN: signal::Number = T::SIGTTIN;
557    const SIGTTOU: signal::Number = T::SIGTTOU;
558    const SIGURG: signal::Number = T::SIGURG;
559    const SIGUSR1: signal::Number = T::SIGUSR1;
560    const SIGUSR2: signal::Number = T::SIGUSR2;
561    const SIGVTALRM: signal::Number = T::SIGVTALRM;
562    const SIGWINCH: signal::Number = T::SIGWINCH;
563    const SIGXCPU: signal::Number = T::SIGXCPU;
564    const SIGXFSZ: signal::Number = T::SIGXFSZ;
565
566    fn sigrt_range(&self) -> Option<RangeInclusive<signal::Number>> {
567        self.0.borrow().sigrt_range()
568    }
569
570    fn iter_sigrt(&self) -> impl DoubleEndedIterator<Item = signal::Number> + use<T> {
571        self.0.borrow().iter_sigrt()
572    }
573
574    fn validate_signal(&self, number: signal::RawNumber) -> Option<(signal::Name, signal::Number)> {
575        self.0.borrow().validate_signal(number)
576    }
577}
578
579/// Delegates `Sigmask` methods to the contained implementor.
580impl<T: Sigmask> Sigmask for SharedSystem<T> {
581    fn sigmask(
582        &self,
583        op: Option<(SigmaskOp, &[signal::Number])>,
584        old_mask: Option<&mut Vec<signal::Number>>,
585    ) -> Result<()> {
586        (**self.0.borrow()).sigmask(op, old_mask)
587    }
588}
589
590/// Delegates `GetSigaction` methods to the contained implementor.
591impl<T: Sigaction> GetSigaction for SharedSystem<T> {
592    fn get_sigaction(&self, signal: signal::Number) -> Result<Disposition> {
593        self.0.borrow().get_sigaction(signal)
594    }
595}
596
597/// Delegates `Sigaction` methods to the contained implementor.
598impl<T: Sigaction> Sigaction for SharedSystem<T> {
599    fn sigaction(&self, signal: signal::Number, action: Disposition) -> Result<Disposition> {
600        self.0.borrow().sigaction(signal, action)
601    }
602}
603
604/// Delegates `CaughtSignals` methods to the contained implementor.
605impl<T: CaughtSignals> CaughtSignals for SharedSystem<T> {
606    fn caught_signals(&self) -> Vec<signal::Number> {
607        self.0.borrow().caught_signals()
608    }
609}
610
611/// Delegates `SendSignal` methods to the contained implementor.
612impl<T: SendSignal> SendSignal for SharedSystem<T> {
613    fn kill(
614        &self,
615        target: Pid,
616        signal: Option<signal::Number>,
617    ) -> impl Future<Output = Result<()>> + use<T> {
618        self.0.borrow().kill(target, signal)
619    }
620    fn raise(&self, signal: signal::Number) -> impl Future<Output = Result<()>> + use<T> {
621        self.0.borrow().raise(signal)
622    }
623}
624
625/// Delegates `Select` methods to the contained implementor.
626impl<T: Select> Select for SharedSystem<T> {
627    fn select(
628        &self,
629        readers: &mut Vec<Fd>,
630        writers: &mut Vec<Fd>,
631        timeout: Option<Duration>,
632        signal_mask: Option<&[signal::Number]>,
633    ) -> Result<c_int> {
634        (**self.0.borrow()).select(readers, writers, timeout, signal_mask)
635    }
636}
637
638/// Delegates `Isatty` methods to the contained implementor.
639impl<T: Isatty> Isatty for SharedSystem<T> {
640    fn isatty(&self, fd: Fd) -> bool {
641        self.0.borrow().isatty(fd)
642    }
643}
644
645/// Delegates `TcGetPgrp` methods to the contained implementor.
646impl<T: TcGetPgrp> TcGetPgrp for SharedSystem<T> {
647    fn tcgetpgrp(&self, fd: Fd) -> Result<Pid> {
648        self.0.borrow().tcgetpgrp(fd)
649    }
650}
651
652/// Delegates `TcSetPgrp` methods to the contained implementor.
653impl<T: TcSetPgrp> TcSetPgrp for SharedSystem<T> {
654    fn tcsetpgrp(&self, fd: Fd, pgid: Pid) -> impl Future<Output = Result<()>> + use<T> {
655        self.0.borrow().tcsetpgrp(fd, pgid)
656    }
657}
658
659/// Delegates `Wait` methods to the contained implementor.
660impl<T: Wait> Wait for SharedSystem<T> {
661    fn wait(&self, target: Pid) -> Result<Option<(Pid, ProcessState)>> {
662        self.0.borrow().wait(target)
663    }
664}
665
666/// Delegates `Exec` methods to the contained implementor.
667impl<T: Exec> Exec for SharedSystem<T> {
668    fn execve(
669        &self,
670        path: &CStr,
671        args: &[CString],
672        envs: &[CString],
673    ) -> impl Future<Output = Result<Infallible>> + use<T> {
674        self.0.borrow().execve(path, args, envs)
675    }
676}
677
678/// Delegates `Exit` methods to the contained implementor.
679impl<T: Exit> Exit for SharedSystem<T> {
680    fn exit(&self, exit_status: ExitStatus) -> impl Future<Output = Infallible> + use<T> {
681        self.0.borrow().exit(exit_status)
682    }
683}
684
685/// Delegates `GetUid` methods to the contained implementor.
686impl<T: GetUid> GetUid for SharedSystem<T> {
687    fn getuid(&self) -> Uid {
688        self.0.borrow().getuid()
689    }
690    fn geteuid(&self) -> Uid {
691        self.0.borrow().geteuid()
692    }
693    fn getgid(&self) -> Gid {
694        self.0.borrow().getgid()
695    }
696    fn getegid(&self) -> Gid {
697        self.0.borrow().getegid()
698    }
699}
700
701/// Delegates `GetPw` methods to the contained implementor.
702impl<T: GetPw> GetPw for SharedSystem<T> {
703    fn getpwnam_dir(&self, name: &CStr) -> Result<Option<PathBuf>> {
704        self.0.borrow().getpwnam_dir(name)
705    }
706}
707
708/// Delegates `Sysconf` methods to the contained implementor.
709impl<T: Sysconf> Sysconf for SharedSystem<T> {
710    fn confstr_path(&self) -> Result<UnixString> {
711        self.0.borrow().confstr_path()
712    }
713}
714
715/// Delegates `ShellPath` methods to the contained implementor.
716impl<T: ShellPath> ShellPath for SharedSystem<T> {
717    fn shell_path(&self) -> CString {
718        self.0.borrow().shell_path()
719    }
720}
721
722/// Delegates `GetRlimit` methods to the contained implementor.
723impl<T: GetRlimit> GetRlimit for SharedSystem<T> {
724    fn getrlimit(&self, resource: Resource) -> Result<LimitPair> {
725        self.0.borrow().getrlimit(resource)
726    }
727}
728
729/// Delegates `SetRlimit` methods to the contained implementor.
730impl<T: SetRlimit> SetRlimit for SharedSystem<T> {
731    fn setrlimit(&self, resource: Resource, limits: LimitPair) -> Result<()> {
732        self.0.borrow().setrlimit(resource, limits)
733    }
734}
735
736impl<S: Signals + Sigmask + Sigaction> SignalSystem for SharedSystem<S> {
737    #[inline]
738    fn get_disposition(&self, signal: signal::Number) -> Result<Disposition> {
739        self.0.borrow().get_disposition(signal)
740    }
741
742    #[inline]
743    fn set_disposition(
744        &mut self,
745        signal: signal::Number,
746        disposition: Disposition,
747    ) -> Result<Disposition> {
748        self.0.borrow_mut().set_disposition(signal, disposition)
749    }
750}
751
752#[cfg(test)]
753mod tests {
754    use super::super::r#virtual::PIPE_SIZE;
755    use super::super::r#virtual::VirtualSystem;
756    use super::super::r#virtual::{SIGCHLD, SIGINT, SIGTERM, SIGUSR1};
757    use super::*;
758    use assert_matches::assert_matches;
759    use futures_util::FutureExt as _;
760    use std::task::Context;
761    use std::task::Poll;
762    use std::task::Waker;
763    use std::time::Duration;
764
765    #[test]
766    fn shared_system_read_async_ready() {
767        let system = SharedSystem::new(VirtualSystem::new());
768        let (reader, writer) = system.pipe().unwrap();
769        system.write(writer, &[42]).unwrap();
770
771        let mut buffer = [0; 2];
772        let result = system.read_async(reader, &mut buffer).now_or_never();
773        assert_eq!(result, Some(Ok(1)));
774        assert_eq!(buffer[..1], [42]);
775    }
776
777    #[test]
778    fn shared_system_read_async_not_ready_at_first() {
779        let system = VirtualSystem::new();
780        let process_id = system.process_id;
781        let state = Rc::clone(&system.state);
782        let system = SharedSystem::new(system);
783        let system2 = system.clone();
784        let (reader, writer) = system.pipe().unwrap();
785
786        let mut context = Context::from_waker(Waker::noop());
787        let mut buffer = [0; 2];
788        let mut future = Box::pin(system.read_async(reader, &mut buffer));
789        let result = future.as_mut().poll(&mut context);
790        assert_eq!(result, Poll::Pending);
791
792        let result = system2.select(false);
793        assert_eq!(result, Ok(()));
794        let result = future.as_mut().poll(&mut context);
795        assert_eq!(result, Poll::Pending);
796
797        state.borrow_mut().processes[&process_id].fds[&writer]
798            .open_file_description
799            .borrow_mut()
800            .write(&[56])
801            .unwrap();
802
803        let result = future.as_mut().poll(&mut context);
804        drop(future);
805        assert_eq!(result, Poll::Ready(Ok(1)));
806        assert_eq!(buffer[..1], [56]);
807    }
808
809    #[test]
810    fn shared_system_write_all_ready() {
811        let system = SharedSystem::new(VirtualSystem::new());
812        let (reader, writer) = system.pipe().unwrap();
813        let result = system.write_all(writer, &[17]).now_or_never().unwrap();
814        assert_eq!(result, Ok(1));
815
816        let mut buffer = [0; 2];
817        system.read(reader, &mut buffer).unwrap();
818        assert_eq!(buffer[..1], [17]);
819    }
820
821    #[test]
822    fn shared_system_write_all_not_ready_at_first() {
823        let system = VirtualSystem::new();
824        let process_id = system.process_id;
825        let state = Rc::clone(&system.state);
826        let system = SharedSystem::new(system);
827        let (reader, writer) = system.pipe().unwrap();
828
829        state.borrow_mut().processes[&process_id].fds[&writer]
830            .open_file_description
831            .borrow_mut()
832            .write(&[42; PIPE_SIZE])
833            .unwrap();
834
835        let mut context = Context::from_waker(Waker::noop());
836        let mut out_buffer = [87; PIPE_SIZE];
837        out_buffer[0] = 0;
838        out_buffer[1] = 1;
839        out_buffer[PIPE_SIZE - 2] = 0xFE;
840        out_buffer[PIPE_SIZE - 1] = 0xFF;
841        let mut future = Box::pin(system.write_all(writer, &out_buffer));
842        let result = future.as_mut().poll(&mut context);
843        assert_eq!(result, Poll::Pending);
844
845        let mut in_buffer = [0; PIPE_SIZE - 1];
846        state.borrow_mut().processes[&process_id].fds[&reader]
847            .open_file_description
848            .borrow_mut()
849            .read(&mut in_buffer)
850            .unwrap();
851        assert_eq!(in_buffer, [42; PIPE_SIZE - 1]);
852
853        let result = future.as_mut().poll(&mut context);
854        assert_eq!(result, Poll::Pending);
855
856        in_buffer[0] = 0;
857        state.borrow_mut().processes[&process_id].fds[&reader]
858            .open_file_description
859            .borrow_mut()
860            .read(&mut in_buffer[..1])
861            .unwrap();
862        assert_eq!(in_buffer[..1], [42; 1]);
863
864        let result = future.as_mut().poll(&mut context);
865        assert_eq!(result, Poll::Ready(Ok(out_buffer.len())));
866
867        state.borrow_mut().processes[&process_id].fds[&reader]
868            .open_file_description
869            .borrow_mut()
870            .read(&mut in_buffer)
871            .unwrap();
872        assert_eq!(in_buffer, out_buffer[..PIPE_SIZE - 1]);
873        state.borrow_mut().processes[&process_id].fds[&reader]
874            .open_file_description
875            .borrow_mut()
876            .read(&mut in_buffer)
877            .unwrap();
878        assert_eq!(in_buffer[..1], out_buffer[PIPE_SIZE - 1..]);
879    }
880
881    #[test]
882    fn shared_system_write_all_empty() {
883        let system = VirtualSystem::new();
884        let process_id = system.process_id;
885        let state = Rc::clone(&system.state);
886        let system = SharedSystem::new(system);
887        let (_reader, writer) = system.pipe().unwrap();
888
889        state.borrow_mut().processes[&process_id].fds[&writer]
890            .open_file_description
891            .borrow_mut()
892            .write(&[0; PIPE_SIZE])
893            .unwrap();
894
895        // Even if the pipe is full, empty write succeeds.
896        let mut context = Context::from_waker(Waker::noop());
897        let mut future = Box::pin(system.write_all(writer, &[]));
898        let result = future.as_mut().poll(&mut context);
899        assert_eq!(result, Poll::Ready(Ok(0)));
900        // TODO Make sure `write` is not called at all
901    }
902
903    // TODO Test SharedSystem::write_all where second write returns EINTR
904
905    #[test]
906    fn shared_system_wait_until() {
907        let system = VirtualSystem::new();
908        let state = Rc::clone(&system.state);
909        let system = SharedSystem::new(system);
910        let start = Instant::now();
911        state.borrow_mut().now = Some(start);
912        let target = start + Duration::from_millis(1_125);
913
914        let mut future = Box::pin(system.wait_until(target));
915        let mut context = Context::from_waker(Waker::noop());
916        let poll = future.as_mut().poll(&mut context);
917        assert_eq!(poll, Poll::Pending);
918
919        system.select(false).unwrap();
920        let poll = future.as_mut().poll(&mut context);
921        assert_eq!(poll, Poll::Ready(()));
922        assert_eq!(state.borrow().now, Some(target));
923    }
924
925    #[test]
926    fn shared_system_wait_for_signals() {
927        let system = VirtualSystem::new();
928        let process_id = system.process_id;
929        let state = Rc::clone(&system.state);
930        let mut system = SharedSystem::new(system);
931        system.set_disposition(SIGCHLD, Disposition::Catch).unwrap();
932        system.set_disposition(SIGINT, Disposition::Catch).unwrap();
933        system.set_disposition(SIGUSR1, Disposition::Catch).unwrap();
934
935        let mut context = Context::from_waker(Waker::noop());
936        let mut future = Box::pin(system.wait_for_signals());
937        let result = future.as_mut().poll(&mut context);
938        assert_eq!(result, Poll::Pending);
939
940        {
941            let mut state = state.borrow_mut();
942            let process = state.processes.get_mut(&process_id).unwrap();
943            assert!(process.blocked_signals().contains(&SIGCHLD));
944            assert!(process.blocked_signals().contains(&SIGINT));
945            assert!(process.blocked_signals().contains(&SIGUSR1));
946            let _ = process.raise_signal(SIGCHLD);
947            let _ = process.raise_signal(SIGINT);
948        }
949        let result = future.as_mut().poll(&mut context);
950        assert_eq!(result, Poll::Pending);
951
952        system.select(false).unwrap();
953        let result = future.as_mut().poll(&mut context);
954        assert_matches!(result, Poll::Ready(signals) => {
955            assert_eq!(signals.len(), 2);
956            assert!(signals.contains(&SIGCHLD));
957            assert!(signals.contains(&SIGINT));
958        });
959    }
960
961    #[test]
962    fn shared_system_wait_for_signal_returns_on_caught() {
963        let system = VirtualSystem::new();
964        let process_id = system.process_id;
965        let state = Rc::clone(&system.state);
966        let mut system = SharedSystem::new(system);
967        system.set_disposition(SIGCHLD, Disposition::Catch).unwrap();
968
969        let mut context = Context::from_waker(Waker::noop());
970        let mut future = Box::pin(system.wait_for_signal(SIGCHLD));
971        let result = future.as_mut().poll(&mut context);
972        assert_eq!(result, Poll::Pending);
973
974        {
975            let mut state = state.borrow_mut();
976            let process = state.processes.get_mut(&process_id).unwrap();
977            assert!(process.blocked_signals().contains(&SIGCHLD));
978            let _ = process.raise_signal(SIGCHLD);
979        }
980        let result = future.as_mut().poll(&mut context);
981        assert_eq!(result, Poll::Pending);
982
983        system.select(false).unwrap();
984        let result = future.as_mut().poll(&mut context);
985        assert_eq!(result, Poll::Ready(()));
986    }
987
988    #[test]
989    fn shared_system_wait_for_signal_ignores_irrelevant_signals() {
990        let system = VirtualSystem::new();
991        let process_id = system.process_id;
992        let state = Rc::clone(&system.state);
993        let mut system = SharedSystem::new(system);
994        system.set_disposition(SIGINT, Disposition::Catch).unwrap();
995        system.set_disposition(SIGTERM, Disposition::Catch).unwrap();
996
997        let mut context = Context::from_waker(Waker::noop());
998        let mut future = Box::pin(system.wait_for_signal(SIGINT));
999        let result = future.as_mut().poll(&mut context);
1000        assert_eq!(result, Poll::Pending);
1001
1002        {
1003            let mut state = state.borrow_mut();
1004            let process = state.processes.get_mut(&process_id).unwrap();
1005            let _ = process.raise_signal(SIGCHLD);
1006            let _ = process.raise_signal(SIGTERM);
1007        }
1008        system.select(false).unwrap();
1009
1010        let result = future.as_mut().poll(&mut context);
1011        assert_eq!(result, Poll::Pending);
1012    }
1013
1014    #[test]
1015    fn shared_system_select_consumes_all_pending_signals() {
1016        let system = VirtualSystem::new();
1017        let process_id = system.process_id;
1018        let state = Rc::clone(&system.state);
1019        let mut system = SharedSystem::new(system);
1020        system.set_disposition(SIGINT, Disposition::Catch).unwrap();
1021        system.set_disposition(SIGTERM, Disposition::Catch).unwrap();
1022
1023        {
1024            let mut state = state.borrow_mut();
1025            let process = state.processes.get_mut(&process_id).unwrap();
1026            let _ = process.raise_signal(SIGINT);
1027            let _ = process.raise_signal(SIGTERM);
1028        }
1029        system.select(false).unwrap();
1030
1031        let state = state.borrow();
1032        let process = state.processes.get(&process_id).unwrap();
1033        let blocked = process.blocked_signals();
1034        assert!(blocked.contains(&SIGINT));
1035        assert!(blocked.contains(&SIGTERM));
1036        let pending = process.pending_signals();
1037        assert!(!pending.contains(&SIGINT));
1038        assert!(!pending.contains(&SIGTERM));
1039    }
1040
1041    #[test]
1042    fn shared_system_select_does_not_wake_signal_waiters_on_io() {
1043        let system = VirtualSystem::new();
1044        let system_1 = SharedSystem::new(system);
1045        let mut system_2 = system_1.clone();
1046        let system_3 = system_1.clone();
1047        let (reader, writer) = system_1.pipe().unwrap();
1048        system_2
1049            .set_disposition(SIGCHLD, Disposition::Catch)
1050            .unwrap();
1051
1052        let mut buffer = [0];
1053        let mut read_future = Box::pin(system_1.read_async(reader, &mut buffer));
1054        let mut signal_future = Box::pin(system_2.wait_for_signals());
1055        let mut context = Context::from_waker(Waker::noop());
1056        let result = read_future.as_mut().poll(&mut context);
1057        assert_eq!(result, Poll::Pending);
1058        let result = signal_future.as_mut().poll(&mut context);
1059        assert_eq!(result, Poll::Pending);
1060        system_3.write(writer, &[42]).unwrap();
1061        system_3.select(false).unwrap();
1062
1063        let result = read_future.as_mut().poll(&mut context);
1064        assert_eq!(result, Poll::Ready(Ok(1)));
1065        let result = signal_future.as_mut().poll(&mut context);
1066        assert_eq!(result, Poll::Pending);
1067    }
1068
1069    #[test]
1070    fn shared_system_select_poll() {
1071        let system = VirtualSystem::new();
1072        let state = Rc::clone(&system.state);
1073        let system = SharedSystem::new(system);
1074        let start = Instant::now();
1075        state.borrow_mut().now = Some(start);
1076        let target = start + Duration::from_millis(1_125);
1077
1078        let mut future = Box::pin(system.wait_until(target));
1079        let mut context = Context::from_waker(Waker::noop());
1080        let poll = future.as_mut().poll(&mut context);
1081        assert_eq!(poll, Poll::Pending);
1082
1083        system.select(true).unwrap();
1084        let poll = future.as_mut().poll(&mut context);
1085        assert_eq!(poll, Poll::Pending);
1086        assert_eq!(state.borrow().now, Some(start));
1087    }
1088}
yash_env/system/shared.rs

yash_env/system/
shared.rs
