Struct Concurrent 

pub struct Concurrent<S> { /* 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 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,

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

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 the output of the task when it completes.

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> Concurrent<S>

where S: Fcntl + Read,

pub 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.

In case of an error, the buffer will contain all data read up to the point of failure.

Use read_all if you don’t have an existing buffer to append to.

pub async fn read_all(&self, fd: Fd) -> Result<Vec<u8>, Errno>

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

This is a convenience method that allocates a buffer and calls read_all_to.

impl<S> Concurrent<S>

where S: Fcntl + Write,

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

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

This method ensures that all data is written, even if multiple write operations are required due to partial writes.

If the data is empty, this method will return immediately without performing write operations.

pub async fn print_error<T: AsRef<[u8]>>(&self, message: T)

Writes the given message to standard error.

This is a convenience method that calls write_all with Fd::STDERR.

impl<S> Concurrent<S>

pub fn new(inner: S) -> Self

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

impl<S> Concurrent<S>

where S: Clock,

pub async fn sleep_until(&self, deadline: Instant)

Waits until the specified deadline.

The returned future will be pending until the specified deadline is reached, at which point it will complete.

pub async fn sleep(&self, duration: Duration)

Waits for the specified duration to elapse.

The returned future will be pending until the specified duration has elapsed, at which point it will complete.

impl<S> Concurrent<S>

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

Waits for signals to be caught.

The returned future will be pending until any signal is caught, at which point it will complete with a list of caught signals. The list is shared among all tasks waiting for signals, so that they can see the same list of caught signals when they are woken up.

Before calling this method, the caller needs to set_disposition for the signals it wants to catch.

If this Concurrent system is used in an Env, you should call Env::wait_for_signals instead of this method, so that the trap set can handle the signals properly.

impl<S> Concurrent<S>

where S: CaughtSignals + Clock + Select,

pub fn peek(&self)

Peeks for any ready events without blocking.

This method performs a select system call with the file descriptors and timeout of pending tasks, and wakes the tasks whose events are ready. This method is similar to select, but it does not block and returns immediately.

pub async fn select(&self)

Waits for any of pending tasks to become ready.

This method performs a select system call with the file descriptors and timeout of pending tasks, and wakes the tasks whose events are ready. This method should be called in the main loop of the process to ensure that tasks can make progress. In a typical use case, the main loop would look like this:

loop {
    // Run ready tasks until they yield again
    run_ready_tasks();
    // Wait for any pending task to become ready
    concurrent.select().await;
}

The run_real and run_virtual methods provide a convenient way to implement such a main loop.

The future returned by this method will be pending if and only if the future returned by the inner system’s select method is pending.

Trait Implementations

impl<S> Chdir for Concurrent<S>

where S: Chdir,

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

Changes the working directory.

impl<S> Clock for Concurrent<S>

where S: Clock,

fn now(&self) -> Instant

Returns the current time.

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

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,

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

Closes a file descriptor.

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

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

Formats the value using the given formatter.

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

fn default() -> Concurrent<S>

Returns the “default value” for a type.

impl<S> Dup for Concurrent<S>

where S: Dup,

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,

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,

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,

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,

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,

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

Returns the current working directory path.

impl<S> GetPid for Concurrent<S>

where S: GetPid,

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,

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,

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

Returns the limits for the specified resource.

impl<S> GetSigaction for Concurrent<S>

where S: GetSigaction,

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

Gets the disposition for a signal.

impl<S> GetUid for Concurrent<S>

where S: GetUid,

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,

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,

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,

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,

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

Creates an unnamed pipe.

impl<S> Seek for Concurrent<S>

where S: Seek,

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

Moves the position of the open file description.

impl<S> SendSignal for Concurrent<S>

where S: SendSignal,

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,

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,

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

Sets the limits for the specified resource.

impl<S> ShellPath for Concurrent<S>

where S: ShellPath,

fn shell_path(&self) -> CString

Returns the path to the shell executable.

impl<S> Sigaction for Concurrent<S>

where S: Sigaction,

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.

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

Gets and/or sets the signal blocking mask.

impl<S> Signals for Concurrent<S>

where S: Signals,

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> Sysconf for Concurrent<S>

where S: Sysconf,

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: TcGetPgrp,

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

Returns the current foreground process group ID.

impl<S> TcSetPgrp for Concurrent<S>

where S: 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: Times,

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

Returns the consumed CPU time statistics.

impl<S> Umask for Concurrent<S>

where S: Umask,

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

Gets and sets the file creation mode mask.

impl<S> Wait for Concurrent<S>

where S: Wait,

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

Reports updated status of a child process.