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.
//! Simplification: the buffer is unbounded, so writers never block.

use super::file::{File, KIND_PIPE, Stat};
use crate::errno::Errno;
use crate::thread::WaitQueue;
use alloc::{collections::VecDeque, sync::Arc};
use spin::Mutex;
use x86_64::instructions::interrupts::without_interrupts;

struct Pipe {
    // 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,
}

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

pub struct PipeReader(Arc<Pipe>);
pub struct PipeWriter(Arc<Pipe>);

/// A new pipe: (read end, write end).
pub fn new() -> (PipeReader, PipeWriter) {
    let pipe = Arc::new(Pipe {
        state: Mutex::new(PipeState { buffer: VecDeque::new(), write_closed: false, read_closed: false }),
        readers_wait: WaitQueue::new(),
    });
    (PipeReader(pipe.clone()), PipeWriter(pipe))
}

impl File for PipeReader {
    fn read(&self, buf: &mut [u8]) -> Result<usize, Errno> {
        if buf.is_empty() {
            return Ok(0);
        }
        // Block until there's data or no writer is left.
        Ok(self.0.readers_wait.wait_until(|| {
            let mut state = self.0.state.lock();
            if state.buffer.is_empty() {
                return if state.write_closed { Some(0) } else { None };
            }
            let n = buf.len().min(state.buffer.len());
            for (dst, src) in buf.iter_mut().zip(state.buffer.drain(..n)) {
                *dst = src;
            }
            Some(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 })
    }
}

impl File for PipeWriter {
    fn read(&self, _buf: &mut [u8]) -> Result<usize, Errno> {
        Err(Errno::EBADF)
    }
    fn write(&self, buf: &[u8]) -> Result<usize, Errno> {
        without_interrupts(|| {
            let mut state = self.0.state.lock();
            if state.read_closed {
                return Err(Errno::EPIPE);
            }
            state.buffer.extend(buf);
            Ok(buf.len())
        })?;
        self.0.readers_wait.wake_all();
        Ok(buf.len())
    }
    fn stat(&self) -> Result<Stat, Errno> {
        Ok(Stat { kind: KIND_PIPE, size: 0 })
    }
}

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
    }
}

impl Drop for PipeReader {
    fn drop(&mut self) {
        without_interrupts(|| self.0.state.lock().read_closed = true);
    }
}