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

yash_env/system/
shared.rs
