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    #[error("port queue full")]
51    WouldBlock,
52    #[error("interrupted by signal")]
53    Interrupted,
54}
55
56/// An IPC port: a bounded message queue with blocking semantics.
57pub struct Port {
58    /// Unique port identifier.
59    pub id: PortId,
60    /// TaskId of the port's creator/owner.
61    pub owner: TaskId,
62    /// The message queue (bounded to `PORT_QUEUE_CAPACITY`).
63    queue: ArrayQueue<IpcMessage>,
64    /// Set to true when the port is destroyed; blocked tasks wake with error.
65    destroyed: AtomicBool,
66    /// Tasks blocked because the queue is full (waiting to send).
67    send_waitq: WaitQueue,
68    /// Tasks blocked because the queue is empty (waiting to receive).
69    recv_waitq: WaitQueue,
70}
71
72impl Port {
73    /// Create a new port owned by the given task.
74    fn new(id: PortId, owner: TaskId) -> Self {
75        Port {
76            id,
77            owner,
78            queue: ArrayQueue::new(PORT_QUEUE_CAPACITY),
79            destroyed: AtomicBool::new(false),
80            send_waitq: WaitQueue::new(),
81            recv_waitq: WaitQueue::new(),
82        }
83    }
84
85    /// Send a message to this port.
86    ///
87    /// If the queue is full, the calling task blocks until space is available.
88    /// Returns `Err(IpcError::PortDestroyed)` if the port is destroyed while
89    /// the sender is blocked.
90    pub fn send(&self, msg: IpcMessage) -> Result<(), IpcError> {
91        let result = self.send_waitq.wait_until(|| {
92            // P2 fix: check for pending signals to avoid livelock.
93            if crate::process::signal::has_pending_signals() {
94                return Some(Err(IpcError::Interrupted));
95            }
96            if self.destroyed.load(Ordering::Acquire) {
97                return Some(Err(IpcError::PortDestroyed));
98            }
99            match self.queue.push(msg) {
100                Ok(()) => Some(Ok(())),
101                Err(_) => None,
102            }
103        });
104        if result.is_ok() {
105            self.recv_waitq.wake_one();
106        }
107        result
108    }
109
110    /// Receive a message from this port.
111    ///
112    /// If the queue is empty, the calling task blocks until a message arrives.
113    /// Returns `Err(IpcError::PortDestroyed)` if the port is destroyed while
114    /// the receiver is blocked.
115    pub fn recv(&self) -> Result<IpcMessage, IpcError> {
116        let result = self.recv_waitq.wait_until(|| {
117            if let Some(msg) = self.queue.pop() {
118                return Some(Ok(msg));
119            }
120            // P2 fix: check for pending signals to avoid livelock.
121            if crate::process::signal::has_pending_signals() {
122                return Some(Err(IpcError::Interrupted));
123            }
124            if self.destroyed.load(Ordering::Acquire) {
125                return Some(Err(IpcError::PortDestroyed));
126            }
127            None
128        });
129        if result.is_ok() {
130            self.send_waitq.wake_one();
131        }
132        result
133    }
134
135    /// Try to send a message to this port without blocking.
136    ///
137    /// Returns `Ok(())` if the message was queued,
138    /// or `Err(WouldBlock)` if the queue is full.
139    pub fn try_send(&self, msg: IpcMessage) -> Result<(), IpcError> {
140        match self.queue.push(msg) {
141            Ok(()) => {
142                self.recv_waitq.wake_one();
143                Ok(())
144            }
145            Err(_) => Err(IpcError::WouldBlock),
146        }
147    }
148
149    /// Try to receive a message from this port without blocking.
150    ///
151    /// Returns `Ok(Some(msg))` if a message was available, `Ok(None)` if empty,
152    /// or `Err(IpcError::PortDestroyed)` if the port is destroyed.
153    pub fn try_recv(&self) -> Result<Option<IpcMessage>, IpcError> {
154        if let Some(msg) = self.queue.pop() {
155            self.send_waitq.wake_one();
156            return Ok(Some(msg));
157        }
158        if self.destroyed.load(Ordering::Acquire) {
159            return Err(IpcError::PortDestroyed);
160        }
161        Ok(None)
162    }
163
164    /// Mark the port as destroyed and wake all blocked tasks.
165    fn destroy(&self) {
166        self.destroyed.store(true, Ordering::Release);
167        self.send_waitq.wake_all();
168        self.recv_waitq.wake_all();
169    }
170
171    /// Returns whether messages is available.
172    pub fn has_messages(&self) -> bool {
173        !self.queue.is_empty()
174    }
175
176    /// Returns whether this can send.
177    pub fn can_send(&self) -> bool {
178        !self.destroyed.load(Ordering::Acquire) && !self.queue.is_full()
179    }
180
181    /// Returns whether destroyed.
182    pub fn is_destroyed(&self) -> bool {
183        self.destroyed.load(Ordering::Acquire)
184    }
185}
186
187// ===========================================================================
188// Global port registry
189// ===========================================================================
190
191/// Next port ID to assign.
192static NEXT_PORT_ID: AtomicU64 = AtomicU64::new(1);
193
194/// Global registry of all live ports.
195static PORTS: SpinLock<Option<BTreeMap<PortId, Arc<Port>>>> = SpinLock::new(None);
196
197/// Ensure the registry is initialized (called lazily).
198fn ensure_registry(guard: &mut Option<BTreeMap<PortId, Arc<Port>>>) {
199    if guard.is_none() {
200        *guard = Some(BTreeMap::new());
201    }
202}
203
204/// Create a new port owned by `owner`. Returns the new port's ID.
205pub fn create_port(owner: TaskId) -> PortId {
206    let id = PortId(NEXT_PORT_ID.fetch_add(1, Ordering::Relaxed));
207    let port = Arc::new(Port::new(id, owner));
208
209    let mut registry = PORTS.lock();
210    ensure_registry(&mut *registry);
211    registry.as_mut().unwrap().insert(id, port);
212
213    log::debug!("IPC: created port {} (owner={})", id, owner);
214    id
215}
216
217/// Look up a port by ID. Returns a cloned `Arc<Port>` if found.
218pub fn get_port(id: PortId) -> Option<Arc<Port>> {
219    let registry = PORTS.lock();
220    registry.as_ref().and_then(|map| map.get(&id).cloned())
221}
222
223/// Destroy a port, removing it from the registry and waking all waiters.
224///
225/// Returns `Ok(())` if destroyed, `Err` if not found or not owned by caller.
226pub fn destroy_port(id: PortId, caller: TaskId) -> Result<(), IpcError> {
227    let port = {
228        let mut registry = PORTS.lock();
229        let map = registry.as_mut().ok_or(IpcError::PortNotFound)?;
230        let port = map.get(&id).ok_or(IpcError::PortNotFound)?;
231        if port.owner != caller {
232            return Err(IpcError::NotOwner);
233        }
234        let port = port.clone();
235        map.remove(&id);
236        port
237    };
238
239    port.destroy();
240    log::debug!("IPC: destroyed port {} (by task {})", id, caller);
241    Ok(())
242}
243
244/// Clean up all ports owned by a dying task.
245///
246/// For each port: drain queued messages and deliver error replies to any
247/// callers blocked in `wait_for_reply`, then destroy the port.
248///
249/// NOTE: Only covers messages still in the queue. If the owner already
250/// recv'd a message but died before calling reply(), the caller task
251/// remains stuck. Servers should handle their own graceful shutdown.
252pub fn cleanup_ports_for_task(owner: TaskId) {
253    super::reply::cancel_replies_waiting_on(owner);
254
255    let owned: alloc::vec::Vec<Arc<Port>> = {
256        let mut registry = PORTS.lock();
257        let Some(map) = registry.as_mut() else { return };
258        let ids: alloc::vec::Vec<PortId> = map
259            .iter()
260            .filter(|(_, p)| p.owner == owner)
261            .map(|(id, _)| *id)
262            .collect();
263        let mut ports = alloc::vec::Vec::with_capacity(ids.len());
264        for id in ids {
265            if let Some(p) = map.remove(&id) {
266                ports.push(p);
267            }
268        }
269        ports
270    };
271
272    for port in owned {
273        port.destroy();
274        while let Some(msg) = port.queue.pop() {
275            let sender = TaskId::from_u64(msg.sender);
276            if sender == owner {
277                continue;
278            }
279            let mut err_reply = IpcMessage::new(0x80);
280            let epipe: u32 = 32;
281            err_reply.payload[0..4].copy_from_slice(&epipe.to_le_bytes());
282            // The dead owner is the awaited responder: authorization passes.
283            let _ = super::reply::deliver_reply(owner, sender, err_reply);
284        }
285    }
286}
strat9_kernel/ipc/port.rs

strat9_kernel/ipc/
port.rs
