rust-kernel

//! 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<PipeState>,
    readers_wait: WaitQueue,
    writers_wait: WaitQueue,
}

struct PipeState {
    buffer: VecDeque<u8>,
    write_closed: bool,
    read_closed: bool,
}

pub struct PipeReader(Arc<Pipe>, AtomicBool);
pub struct PipeWriter(Arc<Pipe>, 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<u64> {
        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<usize, Errno> {
        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<usize, Errno> {
        Err(Errno::EBADF) // wrong end
    }
    fn stat(&self) -> Result<Stat, Errno> {
        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<u64> {
        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<usize, Errno> {
        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<usize, Errno> {
        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<Stat, Errno> {
        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();
    }
}