os/ipc/
message.rs

1//! 进程间消息模块
2
3use core::sync::atomic::Ordering;
4
5use alloc::{collections::vec_deque::VecDeque, vec::Vec};
6
7use crate::{
8    kernel::{WaitQueue, current_cpu, current_task, yield_task},
9    sync::{Mutex, SpinLock},
10};
11
12/// 最大队列容量（字节）
13const DEFAULT_QUEUE_BYTES: usize = 1024;
14
15/// 进程间消息
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub struct Message {
18    /// 消息类型
19    pub mtype: i32,
20    /// 消息大小
21    pub msize: usize,
22    /// 消息内容
23    pub mtext: Vec<u8>,
24}
25
26impl Message {
27    /// 创建一个新的进程间消息
28    pub fn new(mtype: i32, mtext: Vec<u8>) -> Self {
29        let msize = mtext.len();
30        Message {
31            mtype,
32            msize,
33            mtext,
34        }
35    }
36}
37
38/// 内部状态：受 Mutex 保护
39struct QueueState {
40    messages: VecDeque<Message>,
41    bytes: usize,
42}
43
44impl QueueState {
45    fn new() -> Self {
46        Self {
47            messages: VecDeque::new(),
48            bytes: 0,
49        }
50    }
51}
52
53/// 进程间消息队列（支持阻塞的发送/接收）
54pub struct MessageQueue {
55    /// 队列状态：消息与当前字节数
56    state: Mutex<QueueState>,
57    /// 等待发送（队列满）的进程
58    send_waiters: SpinLock<WaitQueue>,
59    /// 等待接收（队列空）的进程
60    recv_waiters: SpinLock<WaitQueue>,
61    /// 队列的最大容量（字节）
62    max_bytes: usize,
63}
64
65impl MessageQueue {
66    /// 创建一个新的进程间消息队列
67    pub fn new() -> Self {
68        MessageQueue {
69            state: Mutex::new(QueueState::new()),
70            send_waiters: SpinLock::new(WaitQueue::new()),
71            recv_waiters: SpinLock::new(WaitQueue::new()),
72            max_bytes: DEFAULT_QUEUE_BYTES,
73        }
74    }
75
76    /// 发送消息（阻塞直到有空间）
77    pub fn send(&self, msg: Message) {
78        let size = msg.msize;
79        let mut pending = Some(msg); // 保存消息所有权，直到真正 push 时再取走
80        loop {
81            let mut st = self.state.lock();
82            if st.bytes + size <= self.max_bytes {
83                let m = pending.take().expect("message already taken");
84                st.bytes += size;
85                st.messages.push_back(m);
86                self.recv_waiters.lock().wake_up_all();
87                return;
88            }
89
90            // XXX: 是否有丢失唤醒风险
91            drop(st);
92            self.send_waiters.lock().sleep(current_task());
93            // 被唤醒后重试
94        }
95    }
96
97    /// 接收任意类型的消息（阻塞直到有消息）
98    pub fn recv(&self) -> Message {
99        loop {
100            if let Some(msg) = {
101                let mut st = self.state.lock();
102                match st.messages.pop_front() {
103                    Some(m) => {
104                        st.bytes -= m.msize;
105                        Some(m)
106                    }
107                    None => None,
108                }
109            } {
110                self.send_waiters.lock().wake_up_all();
111                return msg;
112            }
113
114            self.recv_waiters.lock().sleep(current_task());
115        }
116    }
117
118    /// 按类型接收（阻塞直到有匹配类型）
119    pub fn recv_by_type(&self, mtype: i32) -> Message {
120        loop {
121            if let Some(msg) = {
122                let mut st = self.state.lock();
123                if let Some(i) = st.messages.iter().position(|m| m.mtype == mtype) {
124                    let m = st.messages.remove(i).unwrap();
125                    st.bytes -= m.msize;
126                    Some(m)
127                } else {
128                    None
129                }
130            } {
131                self.send_waiters.lock().wake_up_all();
132                return msg;
133            }
134
135            self.recv_waiters.lock().sleep(current_task());
136        }
137    }
138
139    /// 非阻塞尝试：返回是否成功发送
140    pub fn try_send(&self, msg: Message) -> bool {
141        let mut st = self.state.lock();
142        if st.bytes + msg.msize <= self.max_bytes {
143            st.bytes += msg.msize;
144            st.messages.push_back(msg);
145            drop(st);
146            self.recv_waiters.lock().wake_up_all();
147            true
148        } else {
149            false
150        }
151    }
152
153    /// 非阻塞尝试接收
154    pub fn try_recv(&self) -> Option<Message> {
155        let mut st = self.state.lock();
156        let msg = st.messages.pop_front()?;
157        st.bytes -= msg.msize;
158        drop(st);
159        self.send_waiters.lock().wake_up_all();
160        Some(msg)
161    }
162
163    /// 非阻塞按类型接收
164    pub fn try_recv_by_type(&self, mtype: i32) -> Option<Message> {
165        let mut st = self.state.lock();
166        let ix = st.messages.iter().position(|m| m.mtype == mtype)?;
167        let msg = st.messages.remove(ix).unwrap();
168        st.bytes -= msg.msize;
169        drop(st);
170        self.send_waiters.lock().wake_up_all();
171        Some(msg)
172    }
173
174    /// 查询当前已用字节数
175    pub fn used_bytes(&self) -> usize {
176        self.state.lock().bytes
177    }
178
179    /// 设置最大容量
180    pub fn set_max_bytes(&mut self, max: usize) {
181        let mut st = self.state.lock();
182        st.bytes = st.bytes.min(max);
183        self.max_bytes = max;
184        // 容量增大后唤醒可能阻塞的发送者
185        drop(st);
186        self.send_waiters.lock().wake_up_all();
187    }
188}
os/ipc/message.rs

os/ipc/
message.rs
