//! Pipes: a byte queue with a read end and a write end, each a `File`. //! `pipe()` puts both in the caller's fd table; the shell will hand the //! write end to one program as stdout and the read end to the next as stdin. //! //! Semantics like Unix: reading an empty pipe *blocks* until data arrives or //! every write end is closed (then read returns 0 = EOF); writing when every //! read end is closed fails with EPIPE. Closing = dropping the last `Arc` //! of an end (fd closed, process exited), which our `Drop` impls notice. //! //! The buffer holds at most PIPE_CAPACITY bytes (64 KiB, like Linux): a //! writer that runs ahead of its reader blocks until there's room. With //! O_NONBLOCK, instead of blocking, a read of an empty pipe or a write to //! a full one fails with EAGAIN ("try again later"); event loops use //! poll/epoll (poll.rs) to learn when that is. use super::file::{File, KIND_PIPE, Stat}; use crate::errno::Errno; use crate::poll::{self, POLLERR, POLLHUP, POLLIN, POLLOUT}; use crate::thread::WaitQueue; use alloc::{collections::VecDeque, sync::Arc}; use core::sync::atomic::{AtomicBool, Ordering}; use spin::Mutex; use x86_64::instructions::interrupts::without_interrupts; /// How many bytes a pipe buffers before writers have to wait. pub const PIPE_CAPACITY: usize = 64 * 1024; struct Pipe { /// Its number, for fstat's st_ino and /proc/self/fd ("pipe:[N]"). id: u64, // Locked with interrupts off only (it's touched in `wait_until`'s // condition, see the locking rule in thread.rs). state: Mutex, readers_wait: WaitQueue, writers_wait: WaitQueue, } struct PipeState { buffer: VecDeque, write_closed: bool, read_closed: bool, } pub struct PipeReader(Arc, AtomicBool); pub struct PipeWriter(Arc, AtomicBool); /// A new pipe: (read end, write end). pub fn new() -> (PipeReader, PipeWriter) { static NEXT_ID: core::sync::atomic::AtomicU64 = core::sync::atomic::AtomicU64::new(4000); let pipe = Arc::new(Pipe { id: NEXT_ID.fetch_add(1, Ordering::Relaxed), state: Mutex::new(PipeState { buffer: VecDeque::new(), write_closed: false, read_closed: false }), readers_wait: WaitQueue::new(), writers_wait: WaitQueue::new(), }); (PipeReader(pipe.clone(), AtomicBool::new(false)), PipeWriter(pipe, AtomicBool::new(false))) } impl File for PipeReader { fn anon_ino(&self) -> Option { Some(self.0.id) // both ends: one pipe, one number } fn status_flags(&self) -> u64 { 0 // O_RDONLY: the read end } fn read(&self, buf: &mut [u8]) -> Result { if buf.is_empty() { return Ok(0); } let nonblocking = self.nonblocking(); // Block until there's data or no writer is left (or a signal). let n = self.0.readers_wait.wait_until_intr(|| { let mut state = self.0.state.lock(); if state.buffer.is_empty() { return match () { _ if state.write_closed => Some(Ok(0)), _ if nonblocking => Some(Err(Errno::EAGAIN)), _ => None, }; } let n = buf.len().min(state.buffer.len()); for (dst, src) in buf.iter_mut().zip(state.buffer.drain(..n)) { *dst = src; } Some(Ok(n)) })??; if n > 0 { self.0.writers_wait.wake_all(); // there's room now poll::notify(); } Ok(n) } fn write(&self, _buf: &[u8]) -> Result { Err(Errno::EBADF) // wrong end } fn stat(&self) -> Result { let size = without_interrupts(|| self.0.state.lock().buffer.len()); Ok(Stat { kind: KIND_PIPE, size: size as u64, mode: 0 }) } fn poll(&self) -> u16 { without_interrupts(|| { let state = self.0.state.lock(); let mut events = 0; if !state.buffer.is_empty() { events |= POLLIN; } if state.write_closed { events |= POLLHUP; // (a read returns 0 = EOF) } events }) } fn set_nonblocking(&self, on: bool) { self.1.store(on, Ordering::Relaxed); } fn nonblocking(&self) -> bool { self.1.load(Ordering::Relaxed) } } impl File for PipeWriter { fn anon_ino(&self) -> Option { Some(self.0.id) // both ends: one pipe, one number } fn status_flags(&self) -> u64 { 1 // O_WRONLY: the write end } fn read(&self, _buf: &mut [u8]) -> Result { Err(Errno::EBADF) } /// Write all of `buf`, waiting for room as needed. Interrupted by a /// signal (or non-blocking) after some bytes: return that count. fn write(&self, buf: &[u8]) -> Result { let nonblocking = self.nonblocking(); let mut written = 0; let result = self.0.writers_wait.wait_until_intr(|| { let mut state = self.0.state.lock(); if state.read_closed { return Some(Err(Errno::EPIPE)); } let room = PIPE_CAPACITY - state.buffer.len(); let n = room.min(buf.len() - written); state.buffer.extend(&buf[written..written + n]); written += n; if n > 0 { // Wake readers now, not after the loop: if the pipe is full // we're about to sleep until one of them makes room. self.0.readers_wait.wake_all(); poll::notify(); } match () { _ if written == buf.len() => Some(Ok(written)), _ if nonblocking && written > 0 => Some(Ok(written)), _ if nonblocking => Some(Err(Errno::EAGAIN)), _ => None, // full: wait for a reader to make room } }); match result { Ok(r) => r, Err(_) if written > 0 => Ok(written), // a signal, after some bytes Err(e) => Err(e), } } fn stat(&self) -> Result { Ok(Stat { kind: KIND_PIPE, size: 0, mode: 0 }) } fn poll(&self) -> u16 { without_interrupts(|| { let state = self.0.state.lock(); if state.read_closed { POLLERR // (a write fails with EPIPE) } else if state.buffer.len() < PIPE_CAPACITY { POLLOUT } else { 0 } }) } fn set_nonblocking(&self, on: bool) { self.1.store(on, Ordering::Relaxed); } fn nonblocking(&self) -> bool { self.1.load(Ordering::Relaxed) } } impl Drop for PipeWriter { fn drop(&mut self) { without_interrupts(|| self.0.state.lock().write_closed = true); self.0.readers_wait.wake_all(); // readers see EOF poll::notify(); } } impl Drop for PipeReader { fn drop(&mut self) { without_interrupts(|| self.0.state.lock().read_closed = true); self.0.writers_wait.wake_all(); // writers get EPIPE poll::notify(); } }