os/ipc/
pipe.rs

1//! 管道模块
2
3use alloc::{
4    sync::{Arc, Weak},
5    vec::{self, Vec},
6};
7
8use crate::{
9    sync::Mutex,
10    util::{ring_buffer::RingBuffer, user_buffer::UserBuffer},
11};
12
13/// 创建一个管道，返回读端和写端
14pub fn make_pipe() -> (Arc<Pipe>, Arc<Pipe>) {
15    let buffer = Arc::new(Mutex::new(PipeRingBuffer::new()));
16    let read_end = Arc::new(Pipe::init_read_end(buffer.clone()));
17    let write_end = Arc::new(Pipe::init_write_end(buffer.clone()));
18    buffer.lock().set_write_end(&write_end);
19    (read_end, write_end)
20}
21
22/// 管道结构体
23pub struct Pipe {
24    readable: bool,
25    writable: bool,
26    buffer: Arc<Mutex<PipeRingBuffer>>,
27}
28
29impl Pipe {
30    /// 创建一个读端
31    pub fn init_read_end(buffer: Arc<Mutex<PipeRingBuffer>>) -> Self {
32        Self {
33            readable: true,
34            writable: false,
35            buffer,
36        }
37    }
38
39    /// 创建一个写端
40    pub fn init_write_end(buffer: Arc<Mutex<PipeRingBuffer>>) -> Self {
41        Self {
42            readable: false,
43            writable: true,
44            buffer,
45        }
46    }
47
48    /// 从管道中读取数据到用户缓冲区
49    /// 返回值: 实际读取的字节数
50    pub fn read(&self, buf: UserBuffer) -> usize {
51        if buf.is_empty() {
52            return 0;
53        }
54        let cap = buf.len();
55        let mut data = Vec::with_capacity(cap);
56        {
57            let mut rb = self.buffer.lock();
58            while data.len() < cap {
59                match rb.read_byte() {
60                    Some(b) => data.push(b),
61                    None => break,
62                }
63            }
64        }
65        let n = data.len();
66        // Safety:
67        // - sys_read 调用处构造的 UserBuffer 已验证其指针在用户空间且长度为 n；
68        // - data 是内核侧 Vec，与用户缓冲不重叠；
69        // - 拷贝内部会临时开启 SUM。
70        unsafe { buf.copy_to_user(&data) };
71        n
72    }
73
74    /// 向管道中写入数据
75    /// 返回值: 实际写入的字节数
76    pub fn write(&self, buf: UserBuffer) -> usize {
77        if buf.is_empty() {
78            return 0;
79        }
80        // Safety:
81        // - sys_write 调用处构造的 UserBuffer 已验证其指针在用户空间且长度 buf.len()；
82        // - 读取源是用户空间只读区域；内部会临时开启 SUM。
83        let data = unsafe { buf.copy_from_user() };
84        let mut wrote = 0;
85        {
86            let mut rb = self.buffer.lock();
87            for byte in data {
88                if rb.write_byte(byte).is_ok() {
89                    wrote += 1;
90                } else {
91                    break;
92                }
93            }
94        }
95        wrote
96    }
97}
98
99/// 管道环形缓冲区
100pub(crate) struct PipeRingBuffer {
101    buffer: RingBuffer,
102    write_end: Option<Weak<Pipe>>,
103}
104
105impl PipeRingBuffer {
106    /// 创建一个新的 PipeRingBuffer
107    pub fn new() -> Self {
108        Self {
109            buffer: RingBuffer::new(),
110            write_end: None,
111        }
112    }
113
114    /// 读取一个字节
115    pub fn read_byte(&mut self) -> Option<u8> {
116        self.buffer.read_byte()
117    }
118
119    /// 写入一个字节
120    pub fn write_byte(&mut self, byte: u8) -> Result<(), ()> {
121        self.buffer.write_byte(byte)
122    }
123
124    /// 可用空间
125    pub fn available_space(&self) -> usize {
126        self.buffer.available_space()
127    }
128
129    /// 设置写端
130    pub fn set_write_end(&mut self, write_end: &Arc<Pipe>) {
131        self.write_end = Some(Arc::downgrade(write_end));
132    }
133
134    /// 检查是否所有写端都已被丢弃
135    pub fn all_write_ends_dropped(&self) -> bool {
136        self.write_end
137            .as_ref()
138            .map_or(true, |weak| weak.upgrade().is_none())
139    }
140}
os/ipc/pipe.rs

os/ipc/
pipe.rs
