strat9_kernel/ipc/
port.rs

1//! IPC Port : a kernel-managed message queue with blocking send/recv.
2//!
3//! Each port has a bounded FIFO queue of `IpcMessage`s. Senders block if
4//! the queue is full; receivers block if it's empty. The scheduler's
5//! block/wake API (via `WaitQueue`) provides the blocking mechanism.
6
7use super::message::IpcMessage;
8use crate::{
9    process::TaskId,
10    sync::{SpinLock, WaitQueue},
11};
12use alloc::{collections::BTreeMap, sync::Arc};
13use core::sync::atomic::{AtomicBool, AtomicU64, Ordering};
14use crossbeam_queue::ArrayQueue;
15
16/// Maximum number of messages buffered in a single port.
17const PORT_QUEUE_CAPACITY: usize = 16;
18
19/// Unique identifier for an IPC port.
20#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
21pub struct PortId(pub u64);
22
23impl PortId {
24    /// Get the raw u64 value.
25    pub fn as_u64(self) -> u64 {
26        self.0
27    }
28
29    /// Create a PortId from a raw u64.
30    pub fn from_u64(raw: u64) -> Self {
31        PortId(raw)
32    }
33}
34
35impl core::fmt::Display for PortId {
36    /// Performs the fmt operation.
37    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
38        write!(f, "{}", self.0)
39    }
40}
41
42#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
43pub enum IpcError {
44    #[error("port not found")]
45    PortNotFound,
46    #[error("not owner of port")]
47    NotOwner,
48    #[error("port destroyed")]
49    PortDestroyed,
50}
51
52/// An IPC port: a bounded message queue with blocking semantics.
53pub struct Port {
54    /// Unique port identifier.
55    pub id: PortId,
56    /// TaskId of the port's creator/owner.
57    pub owner: TaskId,
58    /// The message queue (bounded to `PORT_QUEUE_CAPACITY`).
59    queue: ArrayQueue<IpcMessage>,
60    /// Set to true when the port is destroyed; blocked tasks wake with error.
61    destroyed: AtomicBool,
62    /// Tasks blocked because the queue is full (waiting to send).
63    send_waitq: WaitQueue,
64    /// Tasks blocked because the queue is empty (waiting to receive).
65    recv_waitq: WaitQueue,
66}
67
68impl Port {
69    /// Create a new port owned by the given task.
70    fn new(id: PortId, owner: TaskId) -> Self {
71        Port {
72            id,
73            owner,
74            queue: ArrayQueue::new(PORT_QUEUE_CAPACITY),
75            destroyed: AtomicBool::new(false),
76            send_waitq: WaitQueue::new(),
77            recv_waitq: WaitQueue::new(),
78        }
79    }
80
81    /// Send a message to this port.
82    ///
83    /// If the queue is full, the calling task blocks until space is available.
84    /// Returns `Err(IpcError::PortDestroyed)` if the port is destroyed while
85    /// the sender is blocked.
86    pub fn send(&self, msg: IpcMessage) -> Result<(), IpcError> {
87        let result = self.send_waitq.wait_until(|| {
88            if self.destroyed.load(Ordering::Acquire) {
89                return Some(Err(IpcError::PortDestroyed));
90            }
91            match self.queue.push(msg) {
92                Ok(()) => Some(Ok(())),
93                Err(_) => None,
94            }
95        });
96        if result.is_ok() {
97            self.recv_waitq.wake_one();
98        }
99        result
100    }
101
102    /// Receive a message from this port.
103    ///
104    /// If the queue is empty, the calling task blocks until a message arrives.
105    /// Returns `Err(IpcError::PortDestroyed)` if the port is destroyed while
106    /// the receiver is blocked.
107    pub fn recv(&self) -> Result<IpcMessage, IpcError> {
108        let result = self.recv_waitq.wait_until(|| {
109            if let Some(msg) = self.queue.pop() {
110                return Some(Ok(msg));
111            }
112            if self.destroyed.load(Ordering::Acquire) {
113                return Some(Err(IpcError::PortDestroyed));
114            }
115            None
116        });
117        if result.is_ok() {
118            self.send_waitq.wake_one();
119        }
120        result
121    }
122
123    /// Try to send a message to this port without blocking.
124    ///
125    /// Returns `Ok(())` if the message was queued,
126    /// or `Err(WouldBlock)` if the queue is full.
127    pub fn try_send(&self, msg: IpcMessage) -> Result<(), IpcError> {
128        match self.queue.push(msg) {
129            Ok(()) => {
130                self.recv_waitq.wake_one();
131                Ok(())
132            }
133            Err(_) => Err(IpcError::PortNotFound), // WouldBlock
134        }
135    }
136
137    /// Try to receive a message from this port without blocking.
138    ///
139    /// Returns `Ok(Some(msg))` if a message was available, `Ok(None)` if empty,
140    /// or `Err(IpcError::PortDestroyed)` if the port is destroyed.
141    pub fn try_recv(&self) -> Result<Option<IpcMessage>, IpcError> {
142        if let Some(msg) = self.queue.pop() {
143            self.send_waitq.wake_one();
144            return Ok(Some(msg));
145        }
146        if self.destroyed.load(Ordering::Acquire) {
147            return Err(IpcError::PortDestroyed);
148        }
149        Ok(None)
150    }
151
152    /// Mark the port as destroyed and wake all blocked tasks.
153    fn destroy(&self) {
154        self.destroyed.store(true, Ordering::Release);
155        self.send_waitq.wake_all();
156        self.recv_waitq.wake_all();
157    }
158
159    /// Returns whether messages is available.
160    pub fn has_messages(&self) -> bool {
161        !self.queue.is_empty()
162    }
163
164    /// Returns whether this can send.
165    pub fn can_send(&self) -> bool {
166        !self.destroyed.load(Ordering::Acquire) && !self.queue.is_full()
167    }
168
169    /// Returns whether destroyed.
170    pub fn is_destroyed(&self) -> bool {
171        self.destroyed.load(Ordering::Acquire)
172    }
173}
174
175// ===========================================================================
176// Global port registry
177// ===========================================================================
178
179/// Next port ID to assign.
180static NEXT_PORT_ID: AtomicU64 = AtomicU64::new(1);
181
182/// Global registry of all live ports.
183static PORTS: SpinLock<Option<BTreeMap<PortId, Arc<Port>>>> = SpinLock::new(None);
184
185/// Ensure the registry is initialized (called lazily).
186fn ensure_registry(guard: &mut Option<BTreeMap<PortId, Arc<Port>>>) {
187    if guard.is_none() {
188        *guard = Some(BTreeMap::new());
189    }
190}
191
192/// Create a new port owned by `owner`. Returns the new port's ID.
193pub fn create_port(owner: TaskId) -> PortId {
194    let id = PortId(NEXT_PORT_ID.fetch_add(1, Ordering::Relaxed));
195    let port = Arc::new(Port::new(id, owner));
196
197    let mut registry = PORTS.lock();
198    ensure_registry(&mut *registry);
199    registry.as_mut().unwrap().insert(id, port);
200
201    log::debug!("IPC: created port {} (owner={})", id, owner);
202    id
203}
204
205/// Look up a port by ID. Returns a cloned `Arc<Port>` if found.
206pub fn get_port(id: PortId) -> Option<Arc<Port>> {
207    let registry = PORTS.lock();
208    registry.as_ref().and_then(|map| map.get(&id).cloned())
209}
210
211/// Destroy a port, removing it from the registry and waking all waiters.
212///
213/// Returns `Ok(())` if destroyed, `Err` if not found or not owned by caller.
214pub fn destroy_port(id: PortId, caller: TaskId) -> Result<(), IpcError> {
215    let port = {
216        let mut registry = PORTS.lock();
217        let map = registry.as_mut().ok_or(IpcError::PortNotFound)?;
218        let port = map.get(&id).ok_or(IpcError::PortNotFound)?;
219        if port.owner != caller {
220            return Err(IpcError::NotOwner);
221        }
222        let port = port.clone();
223        map.remove(&id);
224        port
225    };
226
227    port.destroy();
228    log::debug!("IPC: destroyed port {} (by task {})", id, caller);
229    Ok(())
230}
231
232/// Clean up all ports owned by a dying task.
233///
234/// For each port: drain queued messages and deliver error replies to any
235/// callers blocked in `wait_for_reply`, then destroy the port.
236///
237/// NOTE: Only covers messages still in the queue. If the owner already
238/// recv'd a message but died before calling reply(), the caller task
239/// remains stuck. Servers should handle their own graceful shutdown.
240pub fn cleanup_ports_for_task(owner: TaskId) {
241    super::reply::cancel_replies_waiting_on(owner);
242
243    let owned: alloc::vec::Vec<Arc<Port>> = {
244        let mut registry = PORTS.lock();
245        let Some(map) = registry.as_mut() else { return };
246        let ids: alloc::vec::Vec<PortId> = map
247            .iter()
248            .filter(|(_, p)| p.owner == owner)
249            .map(|(id, _)| *id)
250            .collect();
251        let mut ports = alloc::vec::Vec::with_capacity(ids.len());
252        for id in ids {
253            if let Some(p) = map.remove(&id) {
254                ports.push(p);
255            }
256        }
257        ports
258    };
259
260    for port in owned {
261        port.destroy();
262        while let Some(msg) = port.queue.pop() {
263            let sender = TaskId::from_u64(msg.sender);
264            if sender == owner {
265                continue;
266            }
267            let mut err_reply = IpcMessage::new(0x80);
268            let epipe: u32 = 32;
269            err_reply.payload[0..4].copy_from_slice(&epipe.to_le_bytes());
270            let _ = super::reply::deliver_reply(sender, err_reply);
271        }
272    }
273}
strat9_kernel/ipc/port.rs

strat9_kernel/ipc/
port.rs
