Struct catalyzer::internals::crates::tokio::net::unix::pipe::Sender

pub struct Sender { /* private fields */ }

Writing end of a Unix pipe.

It can be constructed from a FIFO file with OpenOptions::open_sender.

Opening a named pipe for writing involves a few steps. Call to OpenOptions::open_sender might fail with an error indicating different things:

• io::ErrorKind::NotFound - There is no file at the specified path.
• io::ErrorKind::InvalidInput - The file exists, but it is not a FIFO.
• ENXIO - The file is a FIFO, but no process has it open for reading. Sleep for a while and try again.
• Other OS errors not specific to opening FIFO files.

Opening a Sender from a FIFO file should look like this:

use tokio::net::unix::pipe;
use tokio::time::{self, Duration};

const FIFO_NAME: &str = "path/to/a/fifo";

// Wait for a reader to open the file.
let tx = loop {
    match pipe::OpenOptions::new().open_sender(FIFO_NAME) {
        Ok(tx) => break tx,
        Err(e) if e.raw_os_error() == Some(libc::ENXIO) => {},
        Err(e) => return Err(e.into()),
    }

    time::sleep(Duration::from_millis(50)).await;
};

On Linux, it is possible to create a Sender without waiting in a sleeping loop. This is done by opening a named pipe in read-write access mode with OpenOptions::read_write. This way, a Sender can at the same time hold both a writing end and a reading end, and the latter allows to open a FIFO without ENXIO error since the pipe is open for reading as well.

Sender cannot be used to read from a pipe, so in practice the read access is only used when a FIFO is opened. However, using a Sender in read-write mode may lead to lost data, because written data will be dropped by the system as soon as all pipe ends are closed. To avoid lost data you have to make sure that a reading end has been opened before dropping a Sender.

Note that using read-write access mode with FIFO files is not defined by the POSIX standard and it is only guaranteed to work on Linux.

use tokio::io::AsyncWriteExt;
use tokio::net::unix::pipe;

const FIFO_NAME: &str = "path/to/a/fifo";

let mut tx = pipe::OpenOptions::new()
    .read_write(true)
    .open_sender(FIFO_NAME)?;

// Asynchronously write to the pipe before a reader.
tx.write_all(b"hello world").await?;

Implementations

impl Sender

pub fn from_file(file: File) -> Result<Sender, Error>

Creates a new Sender from a File.

This function is intended to construct a pipe from a File representing a special FIFO file. It will check if the file is a pipe and has write access, set it in non-blocking mode and perform the conversion.

Errors

Fails with io::ErrorKind::InvalidInput if the file is not a pipe or it does not have write access. Also fails with any standard OS error if it occurs.

Panics

This function panics if it is not called from within a runtime with IO enabled.

The runtime is usually set implicitly when this function is called from a future driven by a tokio runtime, otherwise runtime can be set explicitly with Runtime::enter function.

pub fn from_owned_fd(owned_fd: OwnedFd) -> Result<Sender, Error>

Creates a new Sender from an OwnedFd.

This function is intended to construct a pipe from an OwnedFd representing an anonymous pipe or a special FIFO file. It will check if the file descriptor is a pipe and has write access, set it in non-blocking mode and perform the conversion.

Errors

Fails with io::ErrorKind::InvalidInput if the file descriptor is not a pipe or it does not have write access. Also fails with any standard OS error if it occurs.

Panics

This function panics if it is not called from within a runtime with IO enabled.

The runtime is usually set implicitly when this function is called from a future driven by a tokio runtime, otherwise runtime can be set explicitly with Runtime::enter function.

pub fn from_file_unchecked(file: File) -> Result<Sender, Error>

Creates a new Sender from a File without checking pipe properties.

This function is intended to construct a pipe from a File representing a special FIFO file. The conversion assumes nothing about the underlying file; it is left up to the user to make sure it is opened with write access, represents a pipe and is set in non-blocking mode.

Examples

use tokio::net::unix::pipe;
use std::fs::OpenOptions;
use std::os::unix::fs::{FileTypeExt, OpenOptionsExt};

const FIFO_NAME: &str = "path/to/a/fifo";

let file = OpenOptions::new()
    .write(true)
    .custom_flags(libc::O_NONBLOCK)
    .open(FIFO_NAME)?;
if file.metadata()?.file_type().is_fifo() {
    let tx = pipe::Sender::from_file_unchecked(file)?;
    /* use the Sender */
}

Panics

This function panics if it is not called from within a runtime with IO enabled.

The runtime is usually set implicitly when this function is called from a future driven by a tokio runtime, otherwise runtime can be set explicitly with Runtime::enter function.

pub fn from_owned_fd_unchecked(owned_fd: OwnedFd) -> Result<Sender, Error>

Creates a new Sender from an OwnedFd without checking pipe properties.

This function is intended to construct a pipe from an OwnedFd representing an anonymous pipe or a special FIFO file. The conversion assumes nothing about the underlying pipe; it is left up to the user to make sure that the file descriptor represents the writing end of a pipe and the pipe is set in non-blocking mode.

Panics

This function panics if it is not called from within a runtime with IO enabled.

The runtime is usually set implicitly when this function is called from a future driven by a tokio runtime, otherwise runtime can be set explicitly with Runtime::enter function.

pub async fn ready(&self, interest: Interest) -> Result<Ready, Error>

Waits for any of the requested ready states.

This function can be used instead of writable() to check the returned ready set for Ready::WRITABLE and Ready::WRITE_CLOSED events.

The function may complete without the pipe being ready. This is a false-positive and attempting an operation will return with io::ErrorKind::WouldBlock. The function can also return with an empty Ready set, so you should always check the returned value and possibly wait again if the requested states are not set.

Cancel safety

This method is cancel safe. Once a readiness event occurs, the method will continue to return immediately until the readiness event is consumed by an attempt to write that fails with WouldBlock or Poll::Pending.

pub async fn writable(&self) -> Result<(), Error>

Waits for the pipe to become writable.

This function is equivalent to ready(Interest::WRITABLE) and is usually paired with try_write().

Examples

use tokio::net::unix::pipe;
use std::io;

#[tokio::main]
async fn main() -> io::Result<()> {
    // Open a writing end of a fifo
    let tx = pipe::OpenOptions::new().open_sender("path/to/a/fifo")?;

    loop {
        // Wait for the pipe to be writable
        tx.writable().await?;

        // Try to write data, this may still fail with `WouldBlock`
        // if the readiness event is a false positive.
        match tx.try_write(b"hello world") {
            Ok(n) => {
                break;
            }
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                continue;
            }
            Err(e) => {
                return Err(e.into());
            }
        }
    }

    Ok(())
}

pub fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Error>>

Polls for write readiness.

If the pipe is not currently ready for writing, this method will store a clone of the Waker from the provided Context. When the pipe becomes ready for writing, Waker::wake will be called on the waker.

Note that on multiple calls to poll_write_ready or poll_write, only the Waker from the Context passed to the most recent call is scheduled to receive a wakeup.

This function is intended for cases where creating and pinning a future via writable is not feasible. Where possible, using writable is preferred, as this supports polling from multiple tasks at once.

Return value

The function returns:

• Poll::Pending if the pipe is not ready for writing.
• Poll::Ready(Ok(())) if the pipe is ready for writing.
• Poll::Ready(Err(e)) if an error is encountered.

Errors

This function may encounter any standard I/O error except WouldBlock.

pub fn try_write(&self, buf: &[u8]) -> Result<usize, Error>

Tries to write a buffer to the pipe, returning how many bytes were written.

The function will attempt to write the entire contents of buf, but only part of the buffer may be written. If the length of buf is not greater than PIPE_BUF (an OS constant, 4096 under Linux), then the write is guaranteed to be atomic, i.e. either the entire content of buf will be written or this method will fail with WouldBlock. There is no such guarantee if buf is larger than PIPE_BUF.

This function is usually paired with writable.

Return

If data is successfully written, Ok(n) is returned, where n is the number of bytes written. If the pipe is not ready to write data, Err(io::ErrorKind::WouldBlock) is returned.

Examples

use tokio::net::unix::pipe;
use std::io;

#[tokio::main]
async fn main() -> io::Result<()> {
    // Open a writing end of a fifo
    let tx = pipe::OpenOptions::new().open_sender("path/to/a/fifo")?;

    loop {
        // Wait for the pipe to be writable
        tx.writable().await?;

        // Try to write data, this may still fail with `WouldBlock`
        // if the readiness event is a false positive.
        match tx.try_write(b"hello world") {
            Ok(n) => {
                break;
            }
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                continue;
            }
            Err(e) => {
                return Err(e.into());
            }
        }
    }

    Ok(())
}

pub fn try_write_vectored(&self, buf: &[IoSlice<'_>]) -> Result<usize, Error>

Tries to write several buffers to the pipe, returning how many bytes were written.

Data is written from each buffer in order, with the final buffer read from possible being only partially consumed. This method behaves equivalently to a single call to try_write() with concatenated buffers.

If the total length of buffers is not greater than PIPE_BUF (an OS constant, 4096 under Linux), then the write is guaranteed to be atomic, i.e. either the entire contents of buffers will be written or this method will fail with WouldBlock. There is no such guarantee if the total length of buffers is greater than PIPE_BUF.

This function is usually paired with writable.

Return

If data is successfully written, Ok(n) is returned, where n is the number of bytes written. If the pipe is not ready to write data, Err(io::ErrorKind::WouldBlock) is returned.

Examples

use tokio::net::unix::pipe;
use std::io;

#[tokio::main]
async fn main() -> io::Result<()> {
    // Open a writing end of a fifo
    let tx = pipe::OpenOptions::new().open_sender("path/to/a/fifo")?;

    let bufs = [io::IoSlice::new(b"hello "), io::IoSlice::new(b"world")];

    loop {
        // Wait for the pipe to be writable
        tx.writable().await?;

        // Try to write data, this may still fail with `WouldBlock`
        // if the readiness event is a false positive.
        match tx.try_write_vectored(&bufs) {
            Ok(n) => {
                break;
            }
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                continue;
            }
            Err(e) => {
                return Err(e.into());
            }
        }
    }

    Ok(())
}

pub fn into_blocking_fd(self) -> Result<OwnedFd, Error>

Converts the pipe into an OwnedFd in blocking mode.

This function will deregister this pipe end from the event loop, set it in blocking mode and perform the conversion.

pub fn into_nonblocking_fd(self) -> Result<OwnedFd, Error>

Converts the pipe into an OwnedFd in nonblocking mode.

This function will deregister this pipe end from the event loop and perform the conversion. The returned file descriptor will be in nonblocking mode.

Trait Implementations

impl AsFd for Sender

fn as_fd(&self) -> BorrowedFd<'_>

Borrows the file descriptor.

impl AsRawFd for Sender

fn as_raw_fd(&self) -> i32

Extracts the raw file descriptor.

impl AsyncWrite for Sender

fn poll_write( self: Pin<&mut Sender>, cx: &mut Context<'_>, buf: &[u8] ) -> Poll<Result<usize, Error>>

Attempt to write bytes from buf into the object.

fn poll_write_vectored( self: Pin<&mut Sender>, cx: &mut Context<'_>, bufs: &[IoSlice<'_>] ) -> Poll<Result<usize, Error>>

Like poll_write, except that it writes from a slice of buffers.

fn is_write_vectored(&self) -> bool

Determines if this writer has an efficient poll_write_vectored implementation.

fn poll_flush( self: Pin<&mut Sender>, _: &mut Context<'_> ) -> Poll<Result<(), Error>>

Attempts to flush the object, ensuring that any buffered data reach their destination.

fn poll_shutdown( self: Pin<&mut Sender>, _: &mut Context<'_> ) -> Poll<Result<(), Error>>

Initiates or attempts to shut down this writer, returning success when the I/O connection has completely shut down.

impl Debug for Sender

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

Formats the value using the given formatter.