Struct Concurrent 

pub struct Concurrent<S: Sigmask> { /* private fields */ }

Decorator for systems that makes blocking I/O operations concurrency-friendly

This struct is used as a wrapper for systems for enabling concurrent execution of multiple possibly blocking I/O tasks on a single thread. The inner system is expected to implement the Read, Write, and super::Select traits with synchronous (blocking) behavior. This struct leaves Futures returned by I/O methods pending until the I/O operation is ready to avoid blocking the entire process. This allows you to start multiple I/O tasks and wait for them to complete concurrently on a single thread. This struct also provides methods for waiting for signals and waiting for a specified duration, which are represented as Futures as well. The select method of this struct consolidates blocking behavior into a single system call so that the process can resume execution as soon as any of the specified events occurs.

For system calls that do not block, such as Pipe, the wrapper directly forwards the call to the inner system without any modification.

This struct is designed to be used in an Rc to allow multiple tasks to share the same concurrent system. Some traits, such as Read and Write, are implemented for Rc<Concurrent<S>> instead of Concurrent<S> to allow the methods to return futures that capture a clone of the Rc and keep it alive until the operation is finished. This is necessary because the futures need to access the internal state of the Concurrent system without capturing a reference to the original Concurrent struct, which may not live long enough.

The following example illustrates how multiple concurrent tasks are run in a single-threaded pool:

let system = Rc::new(Concurrent::new(VirtualSystem::new()));
let system2 = system.clone();
let system3 = system.clone();
let (reader, writer) = system.pipe().unwrap();
let mut executor = futures_executor::LocalPool::new();

// We add a task that tries to read from the pipe, but nothing has been
// written to it, so the task is stalled.
let read_task = executor.spawner().spawn_local_with_handle(async move {
    let mut buffer = [0; 1];
    system.read(reader, &mut buffer).await.unwrap();
    buffer[0]
});
executor.run_until_stalled();

// Let's add a task that writes to the pipe.
executor.spawner().spawn_local(async move {
    system2.write_all(writer, &[123]).await.unwrap();
});
executor.run_until_stalled();

// The write task has written a byte to the pipe, but the read task is still
// stalled. We need to wake it up by calling `select` or `peek`.
system3.peek();

// Now the read task can proceed to the end.
let number = executor.run_until(read_task.unwrap());
assert_eq!(number, 123);

Implementations

impl<S> Concurrent<S>

where S: Fork + Sigmask,

pub fn new_child_process(&self) -> Result<ChildProcessStarter<S>>

👎Deprecated since 0.14.0:

use the Fork::run_in_child_process method instead

Creates a new child process.

Returns the ChildProcessStarter<S> returned by the inner system’s Fork::new_child_process method. This method is an inherent method of Concurrent<S> instead of an implementation of the Fork trait because the return type does not match with that of the inner system S.

impl Concurrent<RealSystem>

pub fn run_real<F, T>(&self, task: F) -> T

where F: Future<Output = T>,

Runs the given task with concurrency support.

This function implements the main loop of the shell process. It runs the given task while also calling select to handle signals and other events. The task is expected to perform I/O operations using the methods of this Concurrent instance, so that it can yield when the operations would block. The function returns the output of the task when it completes.

This method supports concurrency only inside the task. Other tasks created outside the task will not be run concurrently. This method blocks the current thread until the task completes, so it should only be called in the main function of the shell process. See the run_virtual method for the VirtualSystem counterpart.

impl Concurrent<VirtualSystem>

pub async fn run_virtual<F>(&self, task: F)

where F: Future<Output = ()>,

Runs the given task with concurrency support.

This function implements the main loop of the shell process. It runs the given task while also calling select to handle signals and other events. The task is expected to perform I/O operations using the methods of this Concurrent instance, so that it can yield when the operations would block. The function returns when the task completes or the process is terminated.

This is the VirtualSystem counterpart for the run_real method. To allow VirtualSystem to run multiple tasks concurrently, this method is asynchronous and returns a future that completes when the task finishes or the process is terminated.

impl<S: Sigmask> Concurrent<S>

pub fn new(inner: S) -> Self

Creates a new Concurrent system that wraps the given inner system.

Trait Implementations

impl<S> Chdir for Concurrent<S>

where S: Chdir + Sigmask,

fn chdir(&self, path: &CStr) -> Result<()>

Changes the working directory.

impl<S> Clock for Concurrent<S>

where S: Clock + Sigmask,

fn now(&self) -> Instant

Returns the current time.

impl<S: Clone + Sigmask> Clone for Concurrent<S>

where S::Sigset: Clone,

fn clone(&self) -> Concurrent<S>

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<S> Close for Concurrent<S>

where S: Close + Sigmask,

fn close(&self, fd: Fd) -> Result<()>

Closes a file descriptor.

impl<S: Debug + Sigmask> Debug for Concurrent<S>

where S::Sigset: Debug,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<S: Default + Sigmask> Default for Concurrent<S>

where S::Sigset: Default,

fn default() -> Concurrent<S>

Returns the “default value” for a type.

impl<S> Dup for Concurrent<S>

where S: Dup + Sigmask,

fn dup(&self, from: Fd, to_min: Fd, flags: EnumSet<FdFlag>) -> Result<Fd>

Duplicates a file descriptor.

fn dup2(&self, from: Fd, to: Fd) -> Result<Fd>

Duplicates a file descriptor.

impl<S> Exec for Concurrent<S>

where S: Exec + Sigmask,

fn execve( &self, path: &CStr, args: &[CString], envs: &[CString], ) -> impl Future<Output = Result<Infallible>> + use<S>

Replaces the current process with an external utility.

impl<S> Exit for Concurrent<S>

where S: Exit + Sigmask,

fn exit( &self, exit_status: ExitStatus, ) -> impl Future<Output = Infallible> + use<S>

Terminates the current process.

impl<S> Fcntl for Concurrent<S>

where S: Fcntl + Sigmask,

fn ofd_access(&self, fd: Fd) -> Result<OfdAccess>

Returns the open file description access mode.

fn get_and_set_nonblocking(&self, fd: Fd, nonblocking: bool) -> Result<bool>

Gets and sets the non-blocking mode for the open file description.

fn fcntl_getfd(&self, fd: Fd) -> Result<EnumSet<FdFlag>>

Returns the attributes for the file descriptor.

fn fcntl_setfd(&self, fd: Fd, flags: EnumSet<FdFlag>) -> Result<()>

Sets attributes for the file descriptor.

impl<S> Fstat for Concurrent<S>

where S: Fstat + Sigmask,

type Stat = <S as Fstat>::Stat

Metadata type returned by fstat and fstatat

fn fstat(&self, fd: Fd) -> Result<Self::Stat>

Retrieves metadata of a file.

fn fstatat( &self, dir_fd: Fd, path: &CStr, follow_symlinks: bool, ) -> Result<Self::Stat>

Retrieves metadata of a file.

fn is_directory(&self, path: &CStr) -> bool

Whether there is a directory at the specified path.

fn fd_is_pipe(&self, fd: Fd) -> bool

Tests if a file descriptor is a pipe.

impl<S> GetCwd for Concurrent<S>

where S: GetCwd + Sigmask,

fn getcwd(&self) -> Result<PathBuf>

Returns the current working directory path.

impl<S> GetPid for Concurrent<S>

where S: GetPid + Sigmask,

fn getpid(&self) -> Pid

Returns the process ID of the current process.

fn getppid(&self) -> Pid

Returns the process ID of the parent process.

fn getpgrp(&self) -> Pid

Returns the process group ID of the current process.

fn getsid(&self, pid: Pid) -> Result<Pid>

Returns the session ID of the specified process.

impl<S> GetPw for Concurrent<S>

where S: GetPw + Sigmask,

fn getpwnam_dir(&self, name: &CStr) -> Result<Option<PathBuf>>

Returns the home directory path of the given user.

impl<S> GetRlimit for Concurrent<S>

where S: GetRlimit + Sigmask,

fn getrlimit(&self, resource: Resource) -> Result<LimitPair>

Returns the limits for the specified resource.

impl<S> GetSigaction for Concurrent<S>

where S: GetSigaction + Sigmask,

fn get_sigaction(&self, signal: Number) -> Result<Disposition>

Gets the disposition for a signal.

impl<S> GetUid for Concurrent<S>

where S: GetUid + Sigmask,

fn getuid(&self) -> Uid

Returns the real user ID of the current process.

fn geteuid(&self) -> Uid

Returns the effective user ID of the current process.

fn getgid(&self) -> Gid

Returns the real group ID of the current process.

fn getegid(&self) -> Gid

Returns the effective group ID of the current process.

impl<S> IsExecutableFile for Concurrent<S>

where S: IsExecutableFile + Sigmask,

fn is_executable_file(&self, path: &CStr) -> bool

Whether there is an executable regular file at the specified path.

impl<S> Isatty for Concurrent<S>

where S: Isatty + Sigmask,

fn isatty(&self, fd: Fd) -> bool

Tests if a file descriptor is associated with a terminal device.

impl<S> Open for Concurrent<S>

where S: Open + Sigmask,

This implementation does not (yet) support non-blocking open operations.

fn open( &self, path: &CStr, access: OfdAccess, flags: EnumSet<OpenFlag>, mode: Mode, ) -> impl Future<Output = Result<Fd>> + use<S>

Opens a file descriptor.

fn open_tmpfile(&self, parent_dir: &Path) -> Result<Fd>

Opens a file descriptor associated with an anonymous temporary file.

fn fdopendir(&self, fd: Fd) -> Result<impl Dir + use<S>>

Opens a directory for enumerating entries.

fn opendir(&self, path: &CStr) -> Result<impl Dir + use<S>>

Opens a directory for enumerating entries.

impl<S> Pipe for Concurrent<S>

where S: Pipe + Sigmask,

fn pipe(&self) -> Result<(Fd, Fd)>

Creates an unnamed pipe.

impl<S> ReadAll for Concurrent<S>

where S: Fcntl + Read + Sigmask,

async fn read_all_to(&self, fd: Fd, buffer: &mut Vec<u8>) -> Result<(), Errno>

Reads from the file descriptor until EOF is reached, appending the data to the provided buffer.

fn read_all(&self, fd: Fd) -> impl Future<Output = Result<Vec<u8>, Errno>>

Reads from the file descriptor until EOF is reached, returning the collected data as a Vec<u8>.

impl<S> Seek for Concurrent<S>

where S: Seek + Sigmask,

fn lseek(&self, fd: Fd, position: SeekFrom) -> Result<u64>

Moves the position of the open file description.

impl<S> Select for Concurrent<S>

where S: CaughtSignals + Clock + Select + Sigmask,

fn peek(&self)

Peeks for any ready events without blocking.

fn select(&self) -> impl Future<Output = ()> + use<'_, S>

Waits for any of the pending tasks to become ready.

impl<S> SendSignal for Concurrent<S>

where S: SendSignal + Sigmask,

fn kill( &self, pid: Pid, signal: Option<Number>, ) -> impl Future<Output = Result<()>> + use<S>

Sends a signal.

fn raise(&self, signal: Number) -> impl Future<Output = Result<()>> + use<S>

Sends a signal to the current process.

impl<S> SetPgid for Concurrent<S>

where S: SetPgid + Sigmask,

fn setpgid(&self, pid: Pid, pgid: Pid) -> Result<()>

Modifies the process group ID of a process.

impl<S> SetRlimit for Concurrent<S>

where S: SetRlimit + Sigmask,

fn setrlimit(&self, resource: Resource, limits: LimitPair) -> Result<()>

Sets the limits for the specified resource.

impl<S> ShellPath for Concurrent<S>

where S: ShellPath + Sigmask,

fn shell_path(&self) -> CString

Returns the path to the shell executable.

impl<S> Sigaction for Concurrent<S>

where S: Sigaction + Sigmask,

Exposes the inner system’s sigaction method.

This implementation of Sigaction simply delegates to the inner system’s sigaction method, which bypasses the internal state of Concurrent and may prevent the peek and select methods from responding to received signals without race conditions. To ensure that signal dispositions are configured in a way that allows Concurrent to respond to signals correctly, direct calls to sigaction should be avoided, and, if necessary, only used to temporarily change the signal disposition for specific operations while ensuring that the original disposition is restored afterward before a next call to peek, select, or set_disposition.

The standard way to set a signal disposition to Concurrent is to use the set_disposition method provided by the SignalSystem trait, which ensures that the signal disposition and the signal mask are updated consistently.

fn sigaction( &self, signal: Number, disposition: Disposition, ) -> Result<Disposition>

Gets and sets the disposition for a signal.

impl<S> Sigmask for Concurrent<S>

where S: Sigmask,

Exposes the inner system’s sigmask method.

This implementation of Sigmask simply delegates to the inner system’s sigmask method, which bypasses the internal state of Concurrent and may prevent the peek and select methods from responding to received signals without race conditions. To ensure that the signal mask is configured in a way that allows Concurrent to respond to signals correctly, direct calls to sigmask should be avoided, and, if necessary, only used to temporarily change the signal mask for specific operations while ensuring that the original mask is restored afterward before a next call to peek, select, or set_disposition.

type Sigset = <S as Sigmask>::Sigset

The type representing a set of signals for this system

fn sigmask( &self, op_and_signals: Option<(SigmaskOp, &Self::Sigset)>, old_mask: Option<&mut Self::Sigset>, ) -> impl Future<Output = Result<()>> + use<S>

Gets and/or sets the signal blocking mask.

impl<S> Signals for Concurrent<S>

where S: Sigmask,

const SIGABRT: Number = S::SIGABRT

The signal number for SIGABRT

const SIGALRM: Number = S::SIGALRM

The signal number for SIGALRM

const SIGBUS: Number = S::SIGBUS

The signal number for SIGBUS

const SIGCHLD: Number = S::SIGCHLD

The signal number for SIGCHLD

const SIGCLD: Option<Number> = S::SIGCLD

The signal number for SIGCLD, if available on the system

const SIGCONT: Number = S::SIGCONT

The signal number for SIGCONT

const SIGEMT: Option<Number> = S::SIGEMT

The signal number for SIGEMT, if available on the system

const SIGFPE: Number = S::SIGFPE

The signal number for SIGFPE

const SIGHUP: Number = S::SIGHUP

The signal number for SIGHUP

const SIGILL: Number = S::SIGILL

The signal number for SIGILL

const SIGINFO: Option<Number> = S::SIGINFO

The signal number for SIGINFO, if available on the system

const SIGINT: Number = S::SIGINT

The signal number for SIGINT

const SIGIO: Option<Number> = S::SIGIO

The signal number for SIGIO, if available on the system

const SIGIOT: Number = S::SIGIOT

The signal number for SIGIOT

const SIGKILL: Number = S::SIGKILL

The signal number for SIGKILL

const SIGLOST: Option<Number> = S::SIGLOST

The signal number for SIGLOST, if available on the system

const SIGPIPE: Number = S::SIGPIPE

The signal number for SIGPIPE

const SIGPOLL: Option<Number> = S::SIGPOLL

The signal number for SIGPOLL, if available on the system

const SIGPROF: Number = S::SIGPROF

The signal number for SIGPROF

const SIGPWR: Option<Number> = S::SIGPWR

The signal number for SIGPWR, if available on the system

const SIGQUIT: Number = S::SIGQUIT

The signal number for SIGQUIT

const SIGSEGV: Number = S::SIGSEGV

The signal number for SIGSEGV

const SIGSTKFLT: Option<Number> = S::SIGSTKFLT

The signal number for SIGSTKFLT, if available on the system

const SIGSTOP: Number = S::SIGSTOP

The signal number for SIGSTOP

const SIGSYS: Number = S::SIGSYS

The signal number for SIGSYS

const SIGTERM: Number = S::SIGTERM

The signal number for SIGTERM

const SIGTHR: Option<Number> = S::SIGTHR

The signal number for SIGTHR, if available on the system

const SIGTRAP: Number = S::SIGTRAP

The signal number for SIGTRAP

const SIGTSTP: Number = S::SIGTSTP

The signal number for SIGTSTP

const SIGTTIN: Number = S::SIGTTIN

The signal number for SIGTTIN

const SIGTTOU: Number = S::SIGTTOU

The signal number for SIGTTOU

const SIGURG: Number = S::SIGURG

The signal number for SIGURG

const SIGUSR1: Number = S::SIGUSR1

The signal number for SIGUSR1

const SIGUSR2: Number = S::SIGUSR2

The signal number for SIGUSR2

const SIGVTALRM: Number = S::SIGVTALRM

The signal number for SIGVTALRM

const SIGWINCH: Number = S::SIGWINCH

The signal number for SIGWINCH

const SIGXCPU: Number = S::SIGXCPU

The signal number for SIGXCPU

const SIGXFSZ: Number = S::SIGXFSZ

The signal number for SIGXFSZ

const NAMED_SIGNALS: &'static [(&'static str, Option<Number>)] = S::NAMED_SIGNALS

List of all signal names and their numbers, excluding real-time signals

fn sigrt_range(&self) -> Option<RangeInclusive<Number>>

Returns the range of real-time signals supported by the system.

fn iter_sigrt(&self) -> impl DoubleEndedIterator<Item = Number> + use<S>

Returns an iterator over all real-time signals supported by the system.

fn to_signal_number<N: Into<RawNumber>>(&self, number: N) -> Option<Number>

Tests if a signal number is valid and returns its signal number.

fn sig2str<N: Into<RawNumber>>(&self, signal: N) -> Option<Cow<'static, str>>

Converts a signal number to its string representation.

fn str2sig(&self, name: &str) -> Option<Number>

Converts a string representation of a signal to its signal number.

fn validate_signal(&self, number: RawNumber) -> Option<(Name, Number)>

Tests if a signal number is valid and returns its name and number.

fn signal_name_from_number(&self, number: Number) -> Name

Returns the signal name for the signal number.

fn signal_number_from_name(&self, name: Name) -> Option<Number>

Gets the signal number from the signal name.

impl<S> Sleep for Concurrent<S>

where S: Clock + Sigmask,

async fn sleep_until(&self, deadline: Instant)

Waits until the specified deadline.

async fn sleep(&self, duration: Duration)

Waits for the specified duration to elapse.

impl<S> Sysconf for Concurrent<S>

where S: Sysconf + Sigmask,

fn confstr_path(&self) -> Result<UnixString>

Returns the standard $PATH value where all standard utilities are expected to be found.

impl<S> TcGetPgrp for Concurrent<S>

where S: Sigmask + TcGetPgrp,

fn tcgetpgrp(&self, fd: Fd) -> Result<Pid>

Returns the current foreground process group ID.

impl<S> TcSetPgrp for Concurrent<S>

where S: Sigmask + TcSetPgrp,

fn tcsetpgrp( &self, fd: Fd, pgid: Pid, ) -> impl Future<Output = Result<()>> + use<S>

Switches the foreground process group.

impl<S> Times for Concurrent<S>

where S: Sigmask + Times,

fn times(&self) -> Result<CpuTimes>

Returns the consumed CPU time statistics.

impl<S> Umask for Concurrent<S>

where S: Sigmask + Umask,

fn umask(&self, new_mask: Mode) -> Mode

Gets and sets the file creation mode mask.

impl<S> Wait for Concurrent<S>

where S: Sigmask + Wait,

fn wait(&self, target: Pid) -> Result<Option<(Pid, ProcessState)>>

Reports updated status of a child process.

impl<S: Sigmask> WaitForSignals for Concurrent<S>

async fn wait_for_signals(&self) -> Rc<SignalList>

Waits for signals to be caught.

impl<S> WriteAll for Concurrent<S>

where S: Fcntl + Sigmask + Write,

async fn write_all(&self, fd: Fd, data: &[u8]) -> Result<(), Errno>

Writes all data from the provided buffer to the file descriptor.

fn print_error<T: AsRef<[u8]>>(&self, message: T) -> impl Future<Output = ()>

Writes the given message to standard error.