Channel IPC

概述

Channel 是 OpenOS 的主要 IPC 机制。它是一个双向、同步的消息管道，支持 RPC 模式。

Channel 结构

pub struct Channel {
    end_a: EndState,
    end_b: EndState,
    /// 待传输的 Handle（序列化为 u64）
    pub pending_handles: Vec<u64>,
}

struct EndState {
    /// 等待接收的消息（由对端的 send 设置）
    pending_msg: Option<Vec<u8>>,
    /// 在此端阻塞的 receive 任务 ID
    blocked_receiver: Option<u64>,
    /// 在此端阻塞的 send 任务 ID（等待对端接收）
    blocked_sender: Option<u64>,
    /// 在此端阻塞的 call 任务 ID（等待对端回复）
    blocked_caller: Option<u64>,
    /// 等待被 caller 消费的回复消息
    pending_reply: Option<Vec<u8>>,
}

端点标识

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EndId {
    A,
    B,
}

Channel 操作

send

从一端发送消息到另一端：

pub fn send(&mut self, from: EndId, msg: Vec<u8>, sender_task_id: u64) -> SendResult

返回类型：

pub enum SendResult {
    /// 消息已传递给等待的接收方，包含接收方任务 ID
    Delivered(u64),
    /// 消息已存储，发送方应阻塞直到接收方调用 receive
    Pending,
}

行为：

如果对端已阻塞在 receive，消息立即传递（Delivered）
否则，消息存储在目标端，发送方阻塞（Pending）

receive

在一端接收消息：

pub fn receive(&mut self, on: EndId, task_id: u64) -> RecvResult

返回类型：

pub enum RecvResult {
    /// 消息可用，包含消息字节
    GotMessage(Vec<u8>),
    /// 无消息可用，调用方已注册为阻塞
    Blocked,
}

行为：

如果有消息，立即返回（GotMessage）
否则，阻塞接收方（Blocked）

call

原子性的发送 + 阻塞等待回复（RPC 原语）：

pub fn call(&mut self, from: EndId, msg: Vec<u8>, task_id: u64) -> CallResult

返回类型：

pub enum CallResult {
    /// 回复已可用（快速路径）
    GotReply(Vec<u8>),
    /// 无回复，调用方已注册为阻塞
    Blocked,
}

行为：

如果已有回复，立即返回（GotReply）
否则，发送消息并阻塞（Blocked）

reply

回复接收到的消息，解除调用方 call 的阻塞：

pub fn reply(&mut self, on: EndId, reply: Vec<u8>) -> ReplyResult

返回类型：

pub enum ReplyResult {
    /// 调用方已等待，包含调用方任务 ID
    Unblocked(u64),
    /// 无调用方等待，回复已存储
    Stored,
}

消息流向

任务 A (End A)                      任务 B (End B)
    │                                    │
    │  send(A, msg)                      │
    │  ─────────────────────────────→    │
    │                                    │  receive(B)
    │                                    │  GotMessage(msg)
    │                                    │
    │                                    │  process(msg)
    │                                    │
    │                                    │  reply(B, response)
    │  ←─────────────────────────────    │
    │  call() returns response           │

RPC 模式

基本 RPC

// 客户端
let result = channel.call(EndId::A, request_msg, client_task_id);
match result {
    CallResult::GotReply(reply) => handle_reply(reply),
    CallResult::Blocked => { /* 等待回复 */ }
}

// 服务器
let recv = channel.receive(EndId::B, server_task_id);
match recv {
    RecvResult::GotMessage(msg) => {
        let response = process(msg);
        channel.reply(EndId::B, response);
    }
    RecvResult::Blocked => { /* 等待消息 */ }
}

内联处理

内核内部可以直接触发处理：

// channel_send 触发内联处理
if matches!(result, SendResult::Pending) {
    // 内核检查是否有接收方在等待
    // 如果有，立即处理消息
}

Handle 传递

pending_handles

Channel 有一个 pending_handles Vec，用于存储待传输的 Handle：

pub pending_handles: Vec<u64>

流程：

handle_transfer 将 Handle 值存储到 Channel 的 pending_handles
接收方在下一次 receive 时获取 Handle

会合语义

Channel 实现同步会合（rendezvous）：

发送方阻塞 — 如果接收方未准备好
接收方阻塞 — 如果没有消息
原子性 — 消息传递是原子的

这与 POSIX 管道不同：

POSIX 管道有缓冲区，发送方不阻塞
Channel 无缓冲区，发送方必须等待接收方

测试用例

#[test]
fn test_channel_send_receive() {
    let mut ch = Channel::new();
    let msg = vec![b'H', b'i'];

    // 从 A 发送 — 应该 Pending（无接收方）
    let result = ch.send(EndId::A, msg.clone(), 1);
    assert!(matches!(result, SendResult::Pending));

    // 在 B 接收 — 应该 GotMessage
    let result = ch.receive(EndId::B, 2);
    match result {
        RecvResult::GotMessage(data) => assert_eq!(data, msg),
        _ => panic!("Expected GotMessage"),
    }
}

#[test]
fn test_channel_call_reply() {
    let mut ch = Channel::new();
    let msg = vec![b'q', b'u', b'e', b'r', b'y'];

    // 从 A call — 应该 Blocked
    let result = ch.call(EndId::A, msg.clone(), 1);
    assert!(matches!(result, CallResult::Blocked));

    // 在 B 接收
    let recv = ch.receive(EndId::B, 2);
    match recv {
        RecvResult::GotMessage(data) => assert_eq!(data, msg),
        _ => panic!("Expected GotMessage"),
    }

    // 从 B reply — 应该 Unblocked
    let reply = vec![b'o', b'k'];
    let result = ch.reply(EndId::B, reply.clone());
    match result {
        ReplyResult::Unblocked(task_id) => assert_eq!(task_id, 1),
        _ => panic!("Expected Unblocked"),
    }

    // A 的下一次 call 应该 GotReply
    let result = ch.call(EndId::A, vec![], 1);
    match result {
        CallResult::GotReply(data) => assert_eq!(data, reply),
        _ => panic!("Expected GotReply"),
    }
}

#[test]
fn test_channel_bidirectional() {
    let mut ch = Channel::new();

    // A 发送到 B
    ch.send(EndId::A, vec![1], 1);
    // B 发送到 A
    ch.send(EndId::B, vec![2], 2);

    // B 从 A 接收
    let r = ch.receive(EndId::B, 2);
    assert!(matches!(r, RecvResult::GotMessage(ref d) if d == &[1]));

    // A 从 B 接收
    let r = ch.receive(EndId::A, 1);
    assert!(matches!(r, RecvResult::GotMessage(ref d) if d == &[2]));
}

与旧实现对比

方面	旧实现（端口）	新实现（Channel）
IPC 原语	端口 + 消息	Channel + Handle
同步性	异步发送	同步会合
能力模型	无	Handle + Rights
Handle 传递	不支持	支持（pending_handles）
RPC	手动实现	call/reply 原语

Channel IPC

概述

Channel 结构

端点标识

Channel 操作

send

receive

call

reply

消息流向

RPC 模式

基本 RPC

内联处理

Handle 传递

pending_handles

会合语义

测试用例

与旧实现对比

相关文档