arceos_posix_api/imp/
pipe.rs

1use alloc::sync::Arc;
2use core::ffi::c_int;
3
4use axerrno::{LinuxError, LinuxResult};
5use axio::PollState;
6use axsync::Mutex;
7
8use super::fd_ops::{FileLike, add_file_like, close_file_like};
9use crate::ctypes;
10
11#[derive(Copy, Clone, PartialEq)]
12enum RingBufferStatus {
13    Full,
14    Empty,
15    Normal,
16}
17
18const RING_BUFFER_SIZE: usize = 256;
19
20pub struct PipeRingBuffer {
21    arr: [u8; RING_BUFFER_SIZE],
22    head: usize,
23    tail: usize,
24    status: RingBufferStatus,
25}
26
27impl PipeRingBuffer {
28    pub const fn new() -> Self {
29        Self {
30            arr: [0; RING_BUFFER_SIZE],
31            head: 0,
32            tail: 0,
33            status: RingBufferStatus::Empty,
34        }
35    }
36
37    pub fn write_byte(&mut self, byte: u8) {
38        self.status = RingBufferStatus::Normal;
39        self.arr[self.tail] = byte;
40        self.tail = (self.tail + 1) % RING_BUFFER_SIZE;
41        if self.tail == self.head {
42            self.status = RingBufferStatus::Full;
43        }
44    }
45
46    pub fn read_byte(&mut self) -> u8 {
47        self.status = RingBufferStatus::Normal;
48        let c = self.arr[self.head];
49        self.head = (self.head + 1) % RING_BUFFER_SIZE;
50        if self.head == self.tail {
51            self.status = RingBufferStatus::Empty;
52        }
53        c
54    }
55
56    /// Get the length of remaining data in the buffer
57    pub const fn available_read(&self) -> usize {
58        if matches!(self.status, RingBufferStatus::Empty) {
59            0
60        } else if self.tail > self.head {
61            self.tail - self.head
62        } else {
63            self.tail + RING_BUFFER_SIZE - self.head
64        }
65    }
66
67    /// Get the length of remaining space in the buffer
68    pub const fn available_write(&self) -> usize {
69        if matches!(self.status, RingBufferStatus::Full) {
70            0
71        } else {
72            RING_BUFFER_SIZE - self.available_read()
73        }
74    }
75}
76
77pub struct Pipe {
78    readable: bool,
79    buffer: Arc<Mutex<PipeRingBuffer>>,
80}
81
82impl Pipe {
83    pub fn new() -> (Pipe, Pipe) {
84        let buffer = Arc::new(Mutex::new(PipeRingBuffer::new()));
85        let read_end = Pipe {
86            readable: true,
87            buffer: buffer.clone(),
88        };
89        let write_end = Pipe {
90            readable: false,
91            buffer,
92        };
93        (read_end, write_end)
94    }
95
96    pub const fn readable(&self) -> bool {
97        self.readable
98    }
99
100    pub const fn writable(&self) -> bool {
101        !self.readable
102    }
103
104    pub fn write_end_close(&self) -> bool {
105        Arc::strong_count(&self.buffer) == 1
106    }
107}
108
109impl FileLike for Pipe {
110    fn read(&self, buf: &mut [u8]) -> LinuxResult<usize> {
111        if !self.readable() {
112            return Err(LinuxError::EPERM);
113        }
114        let mut read_size = 0usize;
115        let max_len = buf.len();
116        loop {
117            let mut ring_buffer = self.buffer.lock();
118            let loop_read = ring_buffer.available_read();
119            if loop_read == 0 {
120                if self.write_end_close() {
121                    return Ok(read_size);
122                }
123                drop(ring_buffer);
124                // Data not ready, wait for write end
125                crate::sys_sched_yield(); // TODO: use synconize primitive
126                continue;
127            }
128            for _ in 0..loop_read {
129                if read_size == max_len {
130                    return Ok(read_size);
131                }
132                buf[read_size] = ring_buffer.read_byte();
133                read_size += 1;
134            }
135        }
136    }
137
138    fn write(&self, buf: &[u8]) -> LinuxResult<usize> {
139        if !self.writable() {
140            return Err(LinuxError::EPERM);
141        }
142        let mut write_size = 0usize;
143        let max_len = buf.len();
144        loop {
145            let mut ring_buffer = self.buffer.lock();
146            let loop_write = ring_buffer.available_write();
147            if loop_write == 0 {
148                drop(ring_buffer);
149                // Buffer is full, wait for read end to consume
150                crate::sys_sched_yield(); // TODO: use synconize primitive
151                continue;
152            }
153            for _ in 0..loop_write {
154                if write_size == max_len {
155                    return Ok(write_size);
156                }
157                ring_buffer.write_byte(buf[write_size]);
158                write_size += 1;
159            }
160        }
161    }
162
163    fn stat(&self) -> LinuxResult<ctypes::stat> {
164        let st_mode = 0o10000 | 0o600u32; // S_IFIFO | rw-------
165        Ok(ctypes::stat {
166            st_ino: 1,
167            st_nlink: 1,
168            st_mode,
169            st_uid: 1000,
170            st_gid: 1000,
171            st_blksize: 4096,
172            ..Default::default()
173        })
174    }
175
176    fn into_any(self: Arc<Self>) -> Arc<dyn core::any::Any + Send + Sync> {
177        self
178    }
179
180    fn poll(&self) -> LinuxResult<PollState> {
181        let buf = self.buffer.lock();
182        Ok(PollState {
183            readable: self.readable() && buf.available_read() > 0,
184            writable: self.writable() && buf.available_write() > 0,
185        })
186    }
187
188    fn set_nonblocking(&self, _nonblocking: bool) -> LinuxResult {
189        Ok(())
190    }
191}
192
193/// Create a pipe
194///
195/// Return 0 if succeed
196pub fn sys_pipe(fds: &mut [c_int]) -> c_int {
197    debug!("sys_pipe <= {:#x}", fds.as_ptr() as usize);
198    syscall_body!(sys_pipe, {
199        if fds.len() != 2 {
200            return Err(LinuxError::EFAULT);
201        }
202
203        let (read_end, write_end) = Pipe::new();
204        let read_fd = add_file_like(Arc::new(read_end))?;
205        let write_fd = add_file_like(Arc::new(write_end)).inspect_err(|_| {
206            close_file_like(read_fd).ok();
207        })?;
208
209        fds[0] = read_fd as c_int;
210        fds[1] = write_fd as c_int;
211
212        Ok(0)
213    })
214}
arceos_posix_api/imp/pipe.rs

arceos_posix_api/imp/
pipe.rs
