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::FlexFuture;
33use super::Fork;
34use super::Fstat;
35use super::GetCwd;
36use super::GetPid;
37use super::GetPw;
38use super::GetRlimit;
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::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/// If there is a child process in the [`VirtualSystem`], you should call
152/// [`SystemState::select_all`](super::virtual::SystemState::select_all) in
153/// addition to [`SharedSystem::select`] so that the child process task is woken
154/// up when needed.
155/// (TBD code example)
156///
157/// [`System`]: crate::system::System
158/// [`VirtualSystem`]: crate::system::virtual::VirtualSystem
159#[derive(Debug)]
160pub struct SharedSystem<S>(pub(super) Rc<RefCell<SelectSystem<S>>>);
161
162impl<S> SharedSystem<S> {
163    /// Creates a new shared system.
164    pub fn new(system: S) -> Self {
165        SharedSystem(Rc::new(RefCell::new(SelectSystem::new(system))))
166    }
167
168    /// Reads from the file descriptor.
169    ///
170    /// This function waits for one or more bytes to be available for reading.
171    /// If successful, returns the number of bytes read.
172    pub async fn read_async(&self, fd: Fd, buffer: &mut [u8]) -> Result<usize>
173    where
174        S: Fcntl + Read,
175    {
176        let was_nonblocking = self.get_and_set_nonblocking(fd, true)?;
177
178        // We need to retain a strong reference to the waker outside the poll_fn
179        // function because SelectSystem only retains a weak reference to it.
180        // This allows SelectSystem to discard defunct wakers if this async task
181        // is aborted.
182        let waker = Rc::new(RefCell::new(None));
183
184        let result = poll_fn(|context| {
185            let mut inner = self.0.borrow_mut();
186            match inner.read(fd, buffer) {
187                Err(Errno::EAGAIN) => {
188                    *waker.borrow_mut() = Some(context.waker().clone());
189                    inner.add_reader(fd, Rc::downgrade(&waker));
190                    Poll::Pending
191                }
192                result => Poll::Ready(result),
193            }
194        })
195        .await;
196
197        self.get_and_set_nonblocking(fd, was_nonblocking).ok();
198
199        result
200    }
201
202    /// Writes to the file descriptor.
203    ///
204    /// This function calls [`write`](Write::write) repeatedly until the whole
205    /// `buffer` is written to the FD. If the `buffer` is empty, `write` is not
206    /// called at all, so any error that would be returned from `write` is not
207    /// returned.
208    ///
209    /// This function silently ignores signals that may interrupt writes.
210    pub async fn write_all(&self, fd: Fd, mut buffer: &[u8]) -> Result<usize>
211    where
212        S: Fcntl + Write,
213    {
214        if buffer.is_empty() {
215            return Ok(0);
216        }
217
218        let was_nonblocking = self.get_and_set_nonblocking(fd, true)?;
219        let mut written = 0;
220
221        // We need to retain a strong reference to the waker outside the poll_fn
222        // function because SelectSystem only retains a weak reference to it.
223        // This allows SelectSystem to discard defunct wakers if this async task
224        // is aborted.
225        let waker = Rc::new(RefCell::new(None));
226
227        let result = poll_fn(|context| {
228            let mut inner = self.0.borrow_mut();
229            match inner.write(fd, buffer) {
230                Ok(count) => {
231                    written += count;
232                    buffer = &buffer[count..];
233                    if buffer.is_empty() {
234                        return Poll::Ready(Ok(written));
235                    }
236                }
237                Err(Errno::EAGAIN | Errno::EINTR) => (),
238                Err(error) => return Poll::Ready(Err(error)),
239            }
240
241            *waker.borrow_mut() = Some(context.waker().clone());
242            inner.add_writer(fd, Rc::downgrade(&waker));
243            Poll::Pending
244        })
245        .await;
246
247        self.get_and_set_nonblocking(fd, was_nonblocking).ok();
248
249        result
250    }
251
252    /// Convenience function for printing a message to the standard error
253    pub async fn print_error(&self, message: &str)
254    where
255        S: Fcntl + Write,
256    {
257        _ = self.write_all(Fd::STDERR, message.as_bytes()).await;
258    }
259
260    /// Waits until the specified time point.
261    pub async fn wait_until(&self, target: Instant)
262    where
263        S: Clock,
264    {
265        // We need to retain a strong reference to the waker outside the poll_fn
266        // function because SelectSystem only retains a weak reference to it.
267        // This allows SelectSystem to discard defunct wakers if this async task
268        // is aborted.
269        let waker = Rc::new(RefCell::new(None));
270
271        poll_fn(|context| {
272            let mut system = self.0.borrow_mut();
273            let now = system.now();
274            if now >= target {
275                return Poll::Ready(());
276            }
277            *waker.borrow_mut() = Some(context.waker().clone());
278            system.add_timeout(target, Rc::downgrade(&waker));
279            Poll::Pending
280        })
281        .await
282    }
283
284    /// Waits for some signals to be delivered to this process.
285    ///
286    /// Before calling this function, you need to [set the signal
287    /// disposition](Self::set_disposition) to `Catch`. Without doing so, this
288    /// function cannot detect the receipt of the signals.
289    ///
290    /// Returns an array of signals that were caught.
291    ///
292    /// If this `SharedSystem` is part of an [`Env`], you should call
293    /// [`Env::wait_for_signals`] rather than calling this function directly
294    /// so that the trap set can remember the caught signal.
295    pub async fn wait_for_signals(&self) -> Rc<[signal::Number]> {
296        let status = self.0.borrow_mut().add_signal_waker();
297        poll_fn(|context| {
298            let mut status = status.borrow_mut();
299            let dummy_status = SignalStatus::Expected(None);
300            let old_status = std::mem::replace(&mut *status, dummy_status);
301            match old_status {
302                SignalStatus::Caught(signals) => Poll::Ready(signals),
303                SignalStatus::Expected(_) => {
304                    *status = SignalStatus::Expected(Some(context.waker().clone()));
305                    Poll::Pending
306                }
307            }
308        })
309        .await
310    }
311
312    /// Waits for a signal to be delivered to this process.
313    ///
314    /// Before calling this function, you need to [set the signal
315    /// disposition](Self::set_disposition) to `Catch`.
316    /// Without doing so, this function cannot detect the receipt of the signal.
317    ///
318    /// If this `SharedSystem` is part of an [`Env`], you should call
319    /// [`Env::wait_for_signal`] rather than calling this function directly
320    /// so that the trap set can remember the caught signal.
321    pub async fn wait_for_signal(&self, signal: signal::Number) {
322        while !self.wait_for_signals().await.contains(&signal) {}
323    }
324
325    /// Waits for a next event to occur.
326    ///
327    /// This function calls [`Select::select`] with arguments computed from the
328    /// current internal state of the `SharedSystem`. It will wake up tasks
329    /// waiting for the file descriptor to be ready in
330    /// [`read_async`](Self::read_async) and [`write_all`](Self::write_all) or
331    /// for a signal to be caught in [`wait_for_signal`](Self::wait_for_signal).
332    /// If no tasks are woken for FDs or signals and `poll` is false, this
333    /// function will block until the first task waiting for a specific time
334    /// point is woken.
335    ///
336    /// If poll is true, this function does not block, so it may not wake up any
337    /// tasks.
338    ///
339    /// This function may wake up a task even if the condition it is expecting
340    /// has not yet been met.
341    pub fn select(&self, poll: bool) -> Result<()>
342    where
343        S: Select + CaughtSignals + Clock,
344    {
345        self.0.borrow_mut().select(poll)
346    }
347
348    /// Creates a new child process.
349    ///
350    /// See [`Fork::new_child_process`] for details.
351    pub fn new_child_process(&self) -> Result<ChildProcessStarter<S>>
352    where
353        S: Fork,
354    {
355        self.0.borrow().new_child_process()
356    }
357}
358
359impl<S> Clone for SharedSystem<S> {
360    fn clone(&self) -> Self {
361        SharedSystem(self.0.clone())
362    }
363}
364
365/// Delegates `Fstat` methods to the contained implementor.
366impl<T: Fstat> Fstat for SharedSystem<T> {
367    fn fstat(&self, fd: Fd) -> Result<Stat> {
368        self.0.borrow().fstat(fd)
369    }
370    fn fstatat(&self, dir_fd: Fd, path: &CStr, follow_symlinks: bool) -> Result<Stat> {
371        self.0.borrow().fstatat(dir_fd, path, follow_symlinks)
372    }
373}
374
375/// Delegates `IsExecutableFile` methods to the contained implementor.
376impl<T: IsExecutableFile> IsExecutableFile for SharedSystem<T> {
377    fn is_executable_file(&self, path: &CStr) -> bool {
378        self.0.borrow().is_executable_file(path)
379    }
380}
381
382/// Delegates `Pipe` methods to the contained implementor.
383impl<T: Pipe> Pipe for SharedSystem<T> {
384    fn pipe(&self) -> Result<(Fd, Fd)> {
385        self.0.borrow().pipe()
386    }
387}
388
389/// Delegates `Dup` methods to the contained implementor.
390impl<T: Dup> Dup for SharedSystem<T> {
391    fn dup(&self, from: Fd, to_min: Fd, flags: EnumSet<FdFlag>) -> Result<Fd> {
392        self.0.borrow().dup(from, to_min, flags)
393    }
394    fn dup2(&self, from: Fd, to: Fd) -> Result<Fd> {
395        self.0.borrow().dup2(from, to)
396    }
397}
398
399impl<T: Open> Open for SharedSystem<T> {
400    fn open(
401        &self,
402        path: &CStr,
403        access: OfdAccess,
404        flags: EnumSet<OpenFlag>,
405        mode: Mode,
406    ) -> Result<Fd> {
407        self.0.borrow().open(path, access, flags, mode)
408    }
409    fn open_tmpfile(&self, parent_dir: &Path) -> Result<Fd> {
410        self.0.borrow().open_tmpfile(parent_dir)
411    }
412    #[allow(refining_impl_trait)]
413    fn fdopendir(&self, fd: Fd) -> Result<impl Dir + use<T>> {
414        self.0.borrow().fdopendir(fd)
415    }
416    #[allow(refining_impl_trait)]
417    fn opendir(&self, path: &CStr) -> Result<impl Dir + use<T>> {
418        self.0.borrow().opendir(path)
419    }
420}
421
422impl<T: Close> Close for SharedSystem<T> {
423    fn close(&self, fd: Fd) -> Result<()> {
424        self.0.borrow().close(fd)
425    }
426}
427
428/// Delegates `Fcntl` methods to the contained implementor.
429impl<T: Fcntl> Fcntl for SharedSystem<T> {
430    fn ofd_access(&self, fd: Fd) -> Result<OfdAccess> {
431        self.0.borrow().ofd_access(fd)
432    }
433    fn get_and_set_nonblocking(&self, fd: Fd, nonblocking: bool) -> Result<bool> {
434        self.0.borrow().get_and_set_nonblocking(fd, nonblocking)
435    }
436    fn fcntl_getfd(&self, fd: Fd) -> Result<EnumSet<FdFlag>> {
437        self.0.borrow().fcntl_getfd(fd)
438    }
439    fn fcntl_setfd(&self, fd: Fd, flags: EnumSet<FdFlag>) -> Result<()> {
440        self.0.borrow().fcntl_setfd(fd, flags)
441    }
442}
443
444/// Delegates `Read` methods to the contained implementor.
445impl<T: Read> Read for SharedSystem<T> {
446    fn read(&self, fd: Fd, buffer: &mut [u8]) -> Result<usize> {
447        self.0.borrow().read(fd, buffer)
448    }
449}
450
451/// Delegates `Write` methods to the contained implementor.
452impl<T: Write> Write for SharedSystem<T> {
453    fn write(&self, fd: Fd, buffer: &[u8]) -> Result<usize> {
454        self.0.borrow().write(fd, buffer)
455    }
456}
457
458/// Delegates `Seek` methods to the contained implementor.
459impl<T: Seek> Seek for SharedSystem<T> {
460    fn lseek(&self, fd: Fd, position: SeekFrom) -> Result<u64> {
461        self.0.borrow().lseek(fd, position)
462    }
463}
464
465/// Delegates `Umask` methods to the contained implementor.
466impl<T: Umask> Umask for SharedSystem<T> {
467    fn umask(&self, new_mask: Mode) -> Mode {
468        self.0.borrow().umask(new_mask)
469    }
470}
471
472/// Delegates `GetCwd` methods to the contained implementor.
473impl<T: GetCwd> GetCwd for SharedSystem<T> {
474    fn getcwd(&self) -> Result<PathBuf> {
475        self.0.borrow().getcwd()
476    }
477}
478
479/// Delegates `Chdir` methods to the contained implementor.
480impl<T: Chdir> Chdir for SharedSystem<T> {
481    fn chdir(&self, path: &CStr) -> Result<()> {
482        self.0.borrow().chdir(path)
483    }
484}
485
486/// Delegates `Time` methods to the contained implementor.
487impl<T: Clock> Clock for SharedSystem<T> {
488    fn now(&self) -> Instant {
489        self.0.borrow().now()
490    }
491}
492
493/// Delegates `Times` methods to the contained implementor.
494impl<T: Times> Times for SharedSystem<T> {
495    fn times(&self) -> Result<CpuTimes> {
496        self.0.borrow().times()
497    }
498}
499
500/// Delegates `GetPid` methods to the contained implementor.
501impl<T: GetPid> GetPid for SharedSystem<T> {
502    fn getsid(&self, pid: Pid) -> Result<Pid> {
503        self.0.borrow().getsid(pid)
504    }
505
506    fn getpid(&self) -> Pid {
507        self.0.borrow().getpid()
508    }
509
510    fn getppid(&self) -> Pid {
511        self.0.borrow().getppid()
512    }
513
514    fn getpgrp(&self) -> Pid {
515        self.0.borrow().getpgrp()
516    }
517}
518
519/// Delegates `SetPgid` methods to the contained implementor.
520impl<T: SetPgid> SetPgid for SharedSystem<T> {
521    fn setpgid(&self, pid: Pid, pgid: Pid) -> Result<()> {
522        self.0.borrow().setpgid(pid, pgid)
523    }
524}
525
526/// Delegates `Signals` methods to the contained implementor.
527impl<T: Signals> Signals for SharedSystem<T> {
528    fn validate_signal(&self, number: signal::RawNumber) -> Option<(signal::Name, signal::Number)> {
529        self.0.borrow().validate_signal(number)
530    }
531    fn signal_number_from_name(&self, name: signal::Name) -> Option<signal::Number> {
532        self.0.borrow().signal_number_from_name(name)
533    }
534}
535
536/// Delegates `Sigmask` methods to the contained implementor.
537impl<T: Sigmask> Sigmask for SharedSystem<T> {
538    fn sigmask(
539        &self,
540        op: Option<(SigmaskOp, &[signal::Number])>,
541        old_mask: Option<&mut Vec<signal::Number>>,
542    ) -> Result<()> {
543        (**self.0.borrow()).sigmask(op, old_mask)
544    }
545}
546
547/// Delegates `Sigaction` methods to the contained implementor.
548impl<T: Sigaction> Sigaction for SharedSystem<T> {
549    fn get_sigaction(&self, signal: signal::Number) -> Result<Disposition> {
550        self.0.borrow().get_sigaction(signal)
551    }
552    fn sigaction(&self, signal: signal::Number, action: Disposition) -> Result<Disposition> {
553        self.0.borrow().sigaction(signal, action)
554    }
555}
556
557/// Delegates `CaughtSignals` methods to the contained implementor.
558impl<T: CaughtSignals> CaughtSignals for SharedSystem<T> {
559    fn caught_signals(&self) -> Vec<signal::Number> {
560        self.0.borrow().caught_signals()
561    }
562}
563
564/// Delegates `SendSignal` methods to the contained implementor.
565impl<T: SendSignal> SendSignal for SharedSystem<T> {
566    fn kill(&self, target: Pid, signal: Option<signal::Number>) -> FlexFuture<Result<()>> {
567        self.0.borrow().kill(target, signal)
568    }
569    fn raise(&self, signal: signal::Number) -> FlexFuture<Result<()>> {
570        self.0.borrow().raise(signal)
571    }
572}
573
574/// Delegates `Select` methods to the contained implementor.
575impl<T: Select> Select for SharedSystem<T> {
576    fn select(
577        &self,
578        readers: &mut Vec<Fd>,
579        writers: &mut Vec<Fd>,
580        timeout: Option<Duration>,
581        signal_mask: Option<&[signal::Number]>,
582    ) -> Result<c_int> {
583        (**self.0.borrow()).select(readers, writers, timeout, signal_mask)
584    }
585}
586
587/// Delegates `Isatty` methods to the contained implementor.
588impl<T: Isatty> Isatty for SharedSystem<T> {
589    fn isatty(&self, fd: Fd) -> bool {
590        self.0.borrow().isatty(fd)
591    }
592}
593
594/// Delegates `TcGetPgrp` methods to the contained implementor.
595impl<T: TcGetPgrp> TcGetPgrp for SharedSystem<T> {
596    fn tcgetpgrp(&self, fd: Fd) -> Result<Pid> {
597        self.0.borrow().tcgetpgrp(fd)
598    }
599}
600
601/// Delegates `TcSetPgrp` methods to the contained implementor.
602impl<T: TcSetPgrp> TcSetPgrp for SharedSystem<T> {
603    fn tcsetpgrp(&self, fd: Fd, pgid: Pid) -> FlexFuture<Result<()>> {
604        self.0.borrow().tcsetpgrp(fd, pgid)
605    }
606}
607
608// TODO: This implementation should be removed after refactoring Fork API (#662).
609/// Delegates `Fork` methods to the contained implementor.
610impl<T: Fork> Fork for SharedSystem<T> {
611    /// This method is not supported for `SharedSystem` because types do not match.
612    ///
613    /// You should call the inherent method [`SharedSystem::new_child_process`] instead.
614    /// If you call this trait method, it will panic.
615    fn new_child_process(&self) -> Result<ChildProcessStarter<Self>> {
616        // self.0.borrow().new_child_process()
617        unimplemented!(
618            "new_child_process is not supported for SharedSystem because types do not match"
619        )
620    }
621}
622
623/// Delegates `Wait` methods to the contained implementor.
624impl<T: Wait> Wait for SharedSystem<T> {
625    fn wait(&self, target: Pid) -> Result<Option<(Pid, ProcessState)>> {
626        self.0.borrow().wait(target)
627    }
628}
629
630/// Delegates `Exec` methods to the contained implementor.
631impl<T: Exec> Exec for SharedSystem<T> {
632    fn execve(
633        &self,
634        path: &CStr,
635        args: &[CString],
636        envs: &[CString],
637    ) -> FlexFuture<Result<Infallible>> {
638        self.0.borrow().execve(path, args, envs)
639    }
640}
641
642/// Delegates `Exit` methods to the contained implementor.
643impl<T: Exit> Exit for SharedSystem<T> {
644    fn exit(&self, exit_status: ExitStatus) -> FlexFuture<Infallible> {
645        self.0.borrow().exit(exit_status)
646    }
647}
648
649/// Delegates `GetUid` methods to the contained implementor.
650impl<T: GetUid> GetUid for SharedSystem<T> {
651    fn getuid(&self) -> Uid {
652        self.0.borrow().getuid()
653    }
654    fn geteuid(&self) -> Uid {
655        self.0.borrow().geteuid()
656    }
657    fn getgid(&self) -> Gid {
658        self.0.borrow().getgid()
659    }
660    fn getegid(&self) -> Gid {
661        self.0.borrow().getegid()
662    }
663}
664
665/// Delegates `GetPw` methods to the contained implementor.
666impl<T: GetPw> GetPw for SharedSystem<T> {
667    fn getpwnam_dir(&self, name: &CStr) -> Result<Option<PathBuf>> {
668        self.0.borrow().getpwnam_dir(name)
669    }
670}
671
672/// Delegates `Sysconf` methods to the contained implementor.
673impl<T: Sysconf> Sysconf for SharedSystem<T> {
674    fn confstr_path(&self) -> Result<UnixString> {
675        self.0.borrow().confstr_path()
676    }
677}
678
679/// Delegates `ShellPath` methods to the contained implementor.
680impl<T: ShellPath> ShellPath for SharedSystem<T> {
681    fn shell_path(&self) -> CString {
682        self.0.borrow().shell_path()
683    }
684}
685
686/// Delegates `GetRlimit` methods to the contained implementor.
687impl<T: GetRlimit> GetRlimit for SharedSystem<T> {
688    fn getrlimit(&self, resource: Resource) -> Result<LimitPair> {
689        self.0.borrow().getrlimit(resource)
690    }
691}
692
693/// Delegates `SetRlimit` methods to the contained implementor.
694impl<T: SetRlimit> SetRlimit for SharedSystem<T> {
695    fn setrlimit(&self, resource: Resource, limits: LimitPair) -> Result<()> {
696        self.0.borrow().setrlimit(resource, limits)
697    }
698}
699
700impl<S: Signals + Sigmask + Sigaction> SignalSystem for SharedSystem<S> {
701    #[inline]
702    fn signal_name_from_number(&self, number: signal::Number) -> signal::Name {
703        Signals::signal_name_from_number(self, number)
704    }
705
706    #[inline]
707    fn signal_number_from_name(&self, name: signal::Name) -> Option<signal::Number> {
708        Signals::signal_number_from_name(self, name)
709    }
710
711    #[inline]
712    fn get_disposition(&self, signal: signal::Number) -> Result<Disposition> {
713        self.0.borrow().get_disposition(signal)
714    }
715
716    #[inline]
717    fn set_disposition(
718        &mut self,
719        signal: signal::Number,
720        disposition: Disposition,
721    ) -> Result<Disposition> {
722        self.0.borrow_mut().set_disposition(signal, disposition)
723    }
724}
725
726#[cfg(test)]
727mod tests {
728    use super::super::r#virtual::PIPE_SIZE;
729    use super::super::r#virtual::VirtualSystem;
730    use super::super::r#virtual::{SIGCHLD, SIGINT, SIGTERM, SIGUSR1};
731    use super::*;
732    use assert_matches::assert_matches;
733    use futures_util::FutureExt as _;
734    use std::task::Context;
735    use std::task::Poll;
736    use std::task::Waker;
737    use std::time::Duration;
738
739    #[test]
740    fn shared_system_read_async_ready() {
741        let system = SharedSystem::new(VirtualSystem::new());
742        let (reader, writer) = system.pipe().unwrap();
743        system.write(writer, &[42]).unwrap();
744
745        let mut buffer = [0; 2];
746        let result = system.read_async(reader, &mut buffer).now_or_never();
747        assert_eq!(result, Some(Ok(1)));
748        assert_eq!(buffer[..1], [42]);
749    }
750
751    #[test]
752    fn shared_system_read_async_not_ready_at_first() {
753        let system = VirtualSystem::new();
754        let process_id = system.process_id;
755        let state = Rc::clone(&system.state);
756        let system = SharedSystem::new(system);
757        let system2 = system.clone();
758        let (reader, writer) = system.pipe().unwrap();
759
760        let mut context = Context::from_waker(Waker::noop());
761        let mut buffer = [0; 2];
762        let mut future = Box::pin(system.read_async(reader, &mut buffer));
763        let result = future.as_mut().poll(&mut context);
764        assert_eq!(result, Poll::Pending);
765
766        let result = system2.select(false);
767        assert_eq!(result, Ok(()));
768        let result = future.as_mut().poll(&mut context);
769        assert_eq!(result, Poll::Pending);
770
771        state.borrow_mut().processes[&process_id].fds[&writer]
772            .open_file_description
773            .borrow_mut()
774            .write(&[56])
775            .unwrap();
776
777        let result = future.as_mut().poll(&mut context);
778        drop(future);
779        assert_eq!(result, Poll::Ready(Ok(1)));
780        assert_eq!(buffer[..1], [56]);
781    }
782
783    #[test]
784    fn shared_system_write_all_ready() {
785        let system = SharedSystem::new(VirtualSystem::new());
786        let (reader, writer) = system.pipe().unwrap();
787        let result = system.write_all(writer, &[17]).now_or_never().unwrap();
788        assert_eq!(result, Ok(1));
789
790        let mut buffer = [0; 2];
791        system.read(reader, &mut buffer).unwrap();
792        assert_eq!(buffer[..1], [17]);
793    }
794
795    #[test]
796    fn shared_system_write_all_not_ready_at_first() {
797        let system = VirtualSystem::new();
798        let process_id = system.process_id;
799        let state = Rc::clone(&system.state);
800        let system = SharedSystem::new(system);
801        let (reader, writer) = system.pipe().unwrap();
802
803        state.borrow_mut().processes[&process_id].fds[&writer]
804            .open_file_description
805            .borrow_mut()
806            .write(&[42; PIPE_SIZE])
807            .unwrap();
808
809        let mut context = Context::from_waker(Waker::noop());
810        let mut out_buffer = [87; PIPE_SIZE];
811        out_buffer[0] = 0;
812        out_buffer[1] = 1;
813        out_buffer[PIPE_SIZE - 2] = 0xFE;
814        out_buffer[PIPE_SIZE - 1] = 0xFF;
815        let mut future = Box::pin(system.write_all(writer, &out_buffer));
816        let result = future.as_mut().poll(&mut context);
817        assert_eq!(result, Poll::Pending);
818
819        let mut in_buffer = [0; PIPE_SIZE - 1];
820        state.borrow_mut().processes[&process_id].fds[&reader]
821            .open_file_description
822            .borrow_mut()
823            .read(&mut in_buffer)
824            .unwrap();
825        assert_eq!(in_buffer, [42; PIPE_SIZE - 1]);
826
827        let result = future.as_mut().poll(&mut context);
828        assert_eq!(result, Poll::Pending);
829
830        in_buffer[0] = 0;
831        state.borrow_mut().processes[&process_id].fds[&reader]
832            .open_file_description
833            .borrow_mut()
834            .read(&mut in_buffer[..1])
835            .unwrap();
836        assert_eq!(in_buffer[..1], [42; 1]);
837
838        let result = future.as_mut().poll(&mut context);
839        assert_eq!(result, Poll::Ready(Ok(out_buffer.len())));
840
841        state.borrow_mut().processes[&process_id].fds[&reader]
842            .open_file_description
843            .borrow_mut()
844            .read(&mut in_buffer)
845            .unwrap();
846        assert_eq!(in_buffer, out_buffer[..PIPE_SIZE - 1]);
847        state.borrow_mut().processes[&process_id].fds[&reader]
848            .open_file_description
849            .borrow_mut()
850            .read(&mut in_buffer)
851            .unwrap();
852        assert_eq!(in_buffer[..1], out_buffer[PIPE_SIZE - 1..]);
853    }
854
855    #[test]
856    fn shared_system_write_all_empty() {
857        let system = VirtualSystem::new();
858        let process_id = system.process_id;
859        let state = Rc::clone(&system.state);
860        let system = SharedSystem::new(system);
861        let (_reader, writer) = system.pipe().unwrap();
862
863        state.borrow_mut().processes[&process_id].fds[&writer]
864            .open_file_description
865            .borrow_mut()
866            .write(&[0; PIPE_SIZE])
867            .unwrap();
868
869        // Even if the pipe is full, empty write succeeds.
870        let mut context = Context::from_waker(Waker::noop());
871        let mut future = Box::pin(system.write_all(writer, &[]));
872        let result = future.as_mut().poll(&mut context);
873        assert_eq!(result, Poll::Ready(Ok(0)));
874        // TODO Make sure `write` is not called at all
875    }
876
877    // TODO Test SharedSystem::write_all where second write returns EINTR
878
879    #[test]
880    fn shared_system_wait_until() {
881        let system = VirtualSystem::new();
882        let state = Rc::clone(&system.state);
883        let system = SharedSystem::new(system);
884        let start = Instant::now();
885        state.borrow_mut().now = Some(start);
886        let target = start + Duration::from_millis(1_125);
887
888        let mut future = Box::pin(system.wait_until(target));
889        let mut context = Context::from_waker(Waker::noop());
890        let poll = future.as_mut().poll(&mut context);
891        assert_eq!(poll, Poll::Pending);
892
893        system.select(false).unwrap();
894        let poll = future.as_mut().poll(&mut context);
895        assert_eq!(poll, Poll::Ready(()));
896        assert_eq!(state.borrow().now, Some(target));
897    }
898
899    #[test]
900    fn shared_system_wait_for_signals() {
901        let system = VirtualSystem::new();
902        let process_id = system.process_id;
903        let state = Rc::clone(&system.state);
904        let mut system = SharedSystem::new(system);
905        system.set_disposition(SIGCHLD, Disposition::Catch).unwrap();
906        system.set_disposition(SIGINT, Disposition::Catch).unwrap();
907        system.set_disposition(SIGUSR1, Disposition::Catch).unwrap();
908
909        let mut context = Context::from_waker(Waker::noop());
910        let mut future = Box::pin(system.wait_for_signals());
911        let result = future.as_mut().poll(&mut context);
912        assert_eq!(result, Poll::Pending);
913
914        {
915            let mut state = state.borrow_mut();
916            let process = state.processes.get_mut(&process_id).unwrap();
917            assert!(process.blocked_signals().contains(&SIGCHLD));
918            assert!(process.blocked_signals().contains(&SIGINT));
919            assert!(process.blocked_signals().contains(&SIGUSR1));
920            let _ = process.raise_signal(SIGCHLD);
921            let _ = process.raise_signal(SIGINT);
922        }
923        let result = future.as_mut().poll(&mut context);
924        assert_eq!(result, Poll::Pending);
925
926        system.select(false).unwrap();
927        let result = future.as_mut().poll(&mut context);
928        assert_matches!(result, Poll::Ready(signals) => {
929            assert_eq!(signals.len(), 2);
930            assert!(signals.contains(&SIGCHLD));
931            assert!(signals.contains(&SIGINT));
932        });
933    }
934
935    #[test]
936    fn shared_system_wait_for_signal_returns_on_caught() {
937        let system = VirtualSystem::new();
938        let process_id = system.process_id;
939        let state = Rc::clone(&system.state);
940        let mut system = SharedSystem::new(system);
941        system.set_disposition(SIGCHLD, Disposition::Catch).unwrap();
942
943        let mut context = Context::from_waker(Waker::noop());
944        let mut future = Box::pin(system.wait_for_signal(SIGCHLD));
945        let result = future.as_mut().poll(&mut context);
946        assert_eq!(result, Poll::Pending);
947
948        {
949            let mut state = state.borrow_mut();
950            let process = state.processes.get_mut(&process_id).unwrap();
951            assert!(process.blocked_signals().contains(&SIGCHLD));
952            let _ = process.raise_signal(SIGCHLD);
953        }
954        let result = future.as_mut().poll(&mut context);
955        assert_eq!(result, Poll::Pending);
956
957        system.select(false).unwrap();
958        let result = future.as_mut().poll(&mut context);
959        assert_eq!(result, Poll::Ready(()));
960    }
961
962    #[test]
963    fn shared_system_wait_for_signal_ignores_irrelevant_signals() {
964        let system = VirtualSystem::new();
965        let process_id = system.process_id;
966        let state = Rc::clone(&system.state);
967        let mut system = SharedSystem::new(system);
968        system.set_disposition(SIGINT, Disposition::Catch).unwrap();
969        system.set_disposition(SIGTERM, Disposition::Catch).unwrap();
970
971        let mut context = Context::from_waker(Waker::noop());
972        let mut future = Box::pin(system.wait_for_signal(SIGINT));
973        let result = future.as_mut().poll(&mut context);
974        assert_eq!(result, Poll::Pending);
975
976        {
977            let mut state = state.borrow_mut();
978            let process = state.processes.get_mut(&process_id).unwrap();
979            let _ = process.raise_signal(SIGCHLD);
980            let _ = process.raise_signal(SIGTERM);
981        }
982        system.select(false).unwrap();
983
984        let result = future.as_mut().poll(&mut context);
985        assert_eq!(result, Poll::Pending);
986    }
987
988    #[test]
989    fn shared_system_select_consumes_all_pending_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        {
998            let mut state = state.borrow_mut();
999            let process = state.processes.get_mut(&process_id).unwrap();
1000            let _ = process.raise_signal(SIGINT);
1001            let _ = process.raise_signal(SIGTERM);
1002        }
1003        system.select(false).unwrap();
1004
1005        let state = state.borrow();
1006        let process = state.processes.get(&process_id).unwrap();
1007        let blocked = process.blocked_signals();
1008        assert!(blocked.contains(&SIGINT));
1009        assert!(blocked.contains(&SIGTERM));
1010        let pending = process.pending_signals();
1011        assert!(!pending.contains(&SIGINT));
1012        assert!(!pending.contains(&SIGTERM));
1013    }
1014
1015    #[test]
1016    fn shared_system_select_does_not_wake_signal_waiters_on_io() {
1017        let system = VirtualSystem::new();
1018        let system_1 = SharedSystem::new(system);
1019        let mut system_2 = system_1.clone();
1020        let system_3 = system_1.clone();
1021        let (reader, writer) = system_1.pipe().unwrap();
1022        system_2
1023            .set_disposition(SIGCHLD, Disposition::Catch)
1024            .unwrap();
1025
1026        let mut buffer = [0];
1027        let mut read_future = Box::pin(system_1.read_async(reader, &mut buffer));
1028        let mut signal_future = Box::pin(system_2.wait_for_signals());
1029        let mut context = Context::from_waker(Waker::noop());
1030        let result = read_future.as_mut().poll(&mut context);
1031        assert_eq!(result, Poll::Pending);
1032        let result = signal_future.as_mut().poll(&mut context);
1033        assert_eq!(result, Poll::Pending);
1034        system_3.write(writer, &[42]).unwrap();
1035        system_3.select(false).unwrap();
1036
1037        let result = read_future.as_mut().poll(&mut context);
1038        assert_eq!(result, Poll::Ready(Ok(1)));
1039        let result = signal_future.as_mut().poll(&mut context);
1040        assert_eq!(result, Poll::Pending);
1041    }
1042
1043    #[test]
1044    fn shared_system_select_poll() {
1045        let system = VirtualSystem::new();
1046        let state = Rc::clone(&system.state);
1047        let system = SharedSystem::new(system);
1048        let start = Instant::now();
1049        state.borrow_mut().now = Some(start);
1050        let target = start + Duration::from_millis(1_125);
1051
1052        let mut future = Box::pin(system.wait_until(target));
1053        let mut context = Context::from_waker(Waker::noop());
1054        let poll = future.as_mut().poll(&mut context);
1055        assert_eq!(poll, Poll::Pending);
1056
1057        system.select(true).unwrap();
1058        let poll = future.as_mut().poll(&mut context);
1059        assert_eq!(poll, Poll::Pending);
1060        assert_eq!(state.borrow().now, Some(start));
1061    }
1062}
yash_env/system/shared.rs

yash_env/system/
shared.rs
