1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143

use core::sync::atomic::{AtomicUsize, Ordering};

use alloc::{collections::VecDeque, string::String, sync::Arc};
use kernel_user_link::file::BlockingMode;

use crate::{
    fs::{self, FileAccess, FileAttributes, FileNode, FileSystemError},
    sync::spin::mutex::Mutex,
};

use super::Device;

/// Create a connected pipe pair.
/// The first returned file is the read side of the pipe.
/// The second returned file is the write side of the pipe.
pub fn create_pipe_pair() -> (fs::File, fs::File) {
    let pipe = Arc::new(Mutex::new(InnerPipe {
        buffer: VecDeque::new(),
        read_side_available: true,
        write_side_available: true,
    }));

    let read_device = Arc::new(PipeSide {
        inner: pipe.clone(),
        is_read_side: true,
        clones: AtomicUsize::new(1),
    });
    let write_device = Arc::new(PipeSide {
        inner: pipe.clone(),
        is_read_side: false,
        clones: AtomicUsize::new(1),
    });

    let read_inode = FileNode::new_device(
        String::from("read_pipe"),
        FileAttributes::EMPTY,
        read_device,
    );
    let write_inode = FileNode::new_device(
        String::from("write_pipe"),
        FileAttributes::EMPTY,
        write_device,
    );
    let read_file = fs::File::from_inode(
        read_inode,
        String::from("read_pipe"),
        fs::empty_filesystem(),
        0,
        BlockingMode::Block(1),
        FileAccess::READ,
    )
    .expect("This is a file, shouldn't fail");
    // no blocking for write
    let write_file = fs::File::from_inode(
        write_inode,
        String::from("write_pipe"),
        fs::empty_filesystem(),
        0,
        BlockingMode::None,
        FileAccess::WRITE,
    )
    .expect("This is a file, shouldn't fail");

    (read_file, write_file)
}

/// Pipe is a device that allows two processes to communicate with each other.
#[derive(Debug)]
struct InnerPipe {
    /// The buffer of the pipe.
    buffer: VecDeque<u8>,
    read_side_available: bool,
    write_side_available: bool,
}

/// Represent one side of a pipe.
/// Check [`create_pipe_pair`] for more details.
#[derive(Debug)]
pub struct PipeSide {
    inner: Arc<Mutex<InnerPipe>>,
    clones: AtomicUsize,
    is_read_side: bool,
}

impl Device for PipeSide {
    fn name(&self) -> &str {
        "pipe"
    }

    fn read(&self, _offset: u64, buf: &mut [u8]) -> Result<u64, FileSystemError> {
        if !self.is_read_side {
            return Err(FileSystemError::ReadNotSupported);
        }
        let mut pipe = self.inner.lock();
        if !pipe.write_side_available && pipe.buffer.is_empty() {
            return Err(FileSystemError::EndOfFile);
        }
        let mut bytes_read = 0;
        for byte in buf.iter_mut() {
            if let Some(b) = pipe.buffer.pop_back() {
                *byte = b;
                bytes_read += 1;
            } else {
                break;
            }
        }
        Ok(bytes_read)
    }

    fn write(&self, _offset: u64, buf: &[u8]) -> Result<u64, FileSystemError> {
        if self.is_read_side {
            return Err(FileSystemError::WriteNotSupported);
        }
        let mut pipe = self.inner.lock();
        if !pipe.read_side_available {
            return Err(FileSystemError::EndOfFile);
        }
        for byte in buf.iter() {
            pipe.buffer.push_front(*byte);
        }
        Ok(buf.len() as u64)
    }

    fn close(&self) -> Result<(), FileSystemError> {
        // only close the pipe when all clones are closed
        if self.clones.fetch_sub(1, Ordering::AcqRel) != 1 {
            return Ok(());
        }

        let mut pipe = self.inner.lock();
        if self.is_read_side {
            pipe.read_side_available = false;
        } else {
            pipe.write_side_available = false;
        }
        Ok(())
    }

    fn clone_device(&self) -> Result<(), FileSystemError> {
        self.clones.fetch_add(1, Ordering::AcqRel);
        Ok(())
    }
}