执行摘要

本 PR 为 SGLang 的 Rust 服务器新增了完整的请求消息层，从 HTTP 请求体解析到调度器 wire 编码，并改进了请求标识符设计以消除并发碰撞。这是 Rust 服务器线路的关键基础，但部分采样参数验证仍为 todo!()，且缺少端到端集成测试。

功能与动机

作为从 #29799 拆分的独立 PR，目标是将请求消息定义从 Python 迁移到 Rust，为后续完全替换 Python TokenizerManager 铺路。需要强类型、高性能的消息定义，同时保持与现有 Python 调度器的 wire 兼容性。关键需求包括：支持批处理扇出、处理重复 rid、规范化采样参数等。

实现拆解

1. 共享 wire 类型（types.rs）：定义 OneOrMany<T> 枚举（通过密封 trait 限制安全类型）、TokenIds 别名和 wire_struct! 宏。
2. 请求体与扇出（request.rs）：声明 GenerateBody 结构体，使用 #[serde(deny_unknown_fields)] 配合 unknown fields allowed 实现后向兼容；into_requests 函数完成批处理展开与强校验。
3. 调度器 wire struct（io_struct.rs）：通过 wire_struct! 和 control_messages! 宏声明调度器通信结构，确保字段顺序与 Python 端一致；内联单元测试验证 msgpack 形状。
4. 请求标识符重构（ids.rs）：以 Rid 结构体替代 RidHash，新增 from_client（唯一化客户端 ID）和 client_facing（截取原始 ID）方法，消除重复Ｒid 碰撞风险。
5. 采样参数类型（sampling.rs）：初步定义 SamplingParams，接口占位（todo!()），设置 deny_unknown_fields。

rust/sglang-server/src/message/request.rs

核心文件：定义 HTTP 请求体（GenerateBody）和到调度器的编码逻辑，替换 Python 端的分批处理。

/// The `/generate` wire body before batch splitting.
/// Unknown keys are IGNORED, matching Python's `GenerateReqInput`.
#[derive(Debug, Clone, Default, Deserialize)]
pub struct GenerateBody {
    /// Optional client-supplied request id(s): a single string (which fans out
    /// as `{rid}_{i}`) or one per item.
    #[serde(default)]
    pub rid: Option<OneOrMany<String>>,
    #[serde(default)]
    pub text: Option<OneOrMany<String>>,
    #[serde(default)]
    pub input_ids: Option<OneOrMany<TokenIds>>,
    #[serde(default)]
    pub stream: bool,
    /// One params object (broadcast) or a list of them (per item).
    #[serde(default)]
    pub sampling_params: Option<SamplingParamsInput>,
    // ... other fields omitted for brevity
}

rust/sglang-server/src/message/types.rs

提供共享 wire 类型（OneOrMany、TokenIds、wire_struct! 宏），确保序列化安全与 Python 端对齐。

/// A field taking a bare `T` **or** `[T, …]` (e.g. `text: "hi"` or `text: ["a","b"]`).
/// `untagged` takes the first variant that matches, so a `T` that itself accepts
/// a sequence would make `Many` unreachable — hence the [`OneOrManyItem`] gate.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(untagged)]
pub enum OneOrMany<T: OneOrManyItem> {
    One(T),
    Many(Vec<T>),
}
​
/// Types vetted for [`OneOrMany`]. Sealed, so adding one is a deliberate act.
pub trait OneOrManyItem: sealed::SealedItem {}
impl<T: sealed::SealedItem> OneOrManyItem for T {}
​
mod sealed {
    pub trait SealedItem {}
    impl SealedItem for bool {}
    impl SealedItem for i64 {}
    impl SealedItem for String {}
    impl SealedItem for super::TokenIds {}
}

rust/sglang-server/src/ids.rs

重写请求标识符：引入 Rid 结构体替代旧的 RidHash，支持客户端 ID 唯一化以防止重复 rid 碰撞。

/// Separates a client-supplied rid from the uniquifier appended to it.
/// `Rid::new()` is uuid hex and `Rid::new_health_check()` adds only
/// `HEALTH_CHECK_`, so its presence at the fixed offset below is what lets
/// `Rid::client_facing()` recognize a suffix without carrying a flag.
/// This matters because `Rid` rides on every `ChunkEvent`.
const UNIQ_SEP: u8 = b'#';
const UNIQ_DIGITS: usize = 16;
const UNIQ_SUFFIX_LEN: usize = 1 + UNIQ_DIGITS;
​
#[derive(Clone, Debug)]
pub struct Rid {
    id: String,
    /// Partition key, derived from `id`. Never part of identity.
    hash: u64,
}
​
// Identity is the ID, not the digest.
impl PartialEq for Rid {
    fn eq(&self, other: &Self) -> bool { self.id == other.id }
}
impl Eq for Rid {}
impl Hash for Rid {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) { self.id.hash(state); }
}
​
impl Rid {
    pub fn new() -> Self {
        let id = Uuid::new_v4().simple().to_string();
        Rid::from(id)
    }
​
    pub fn new_health_check() -> Self {
        let id = format!("{}_" , crate::HEALTH_CHECK_RID_PREFIX)
            + &Uuid::new_v4().simple().to_string();
        Rid::from(id)
    }
​
    /// A CLIENT-SUPPLIED rid, made unique for internal use by appending a
    /// uniquifier. Prevents hash collisions (duplicate client rid) from
    /// causing two concurrent requests to share the same downstream sink.
    pub fn from_client(id: &str) -> Self {
        use std::sync::atomic::{AtomicU32, Ordering};
        static BASE: OnceLock<u32> = OnceLock::new();
        static NEXT: AtomicU32 = AtomicU32::new(0);
        let base = *BASE.get_or_init(|| Uuid::new_v4().as_u128() as u32);
        let n = NEXT.fetch_add(1, Ordering::Relaxed);
        Rid::from(format!("{id}{sep}{base:08x}{n:08x}" , sep = UNIQ_SEP as char))
    }
​
    /// The rid as written by the client — what `meta_info.id` must echo.
    pub fn client_facing(&self) -> &str {
        let b = self.id.as_bytes();
        let Some(cut) = b.len().checked_sub(UNIQ_SUFFIX_LEN) else {
            return &self.id;
        };
        if b[cut] == UNIQ_SEP && b[cut + 1..].iter().all(u8::is_ascii_hexdigit) {
            &self.id[..cut]
        } else {
            &self.id
        }
    }
​
    #[inline]
    pub fn shard(&self, n: usize) -> usize {
        debug_assert!(n > 0);
        (self.hash as usize) % n
    }
}
​
impl From<String> for Rid {
    fn from(id: String) -> Self {
        let hash = {
            let state = std::collections::hash_map::RandomState::new();
            state.hash_one(&id)
        };
        Rid { id, hash }
    }
}

评论区精华

• 设计权衡：OneOrMany 的 untagged 语义与类型安全性被仔细讨论，最终采用密封 trait 限制允许的类型，使设计更健壮。
• 正确性：merrymercy 在重复 rid 和空 input_ids 校验上坚持与 Python 对齐，作者通过唯一化和验证增强解决了问题。
• 可维护性：错误类型统一和 wire struct 定义顺序依赖的讨论反映了对代码质量的追求。

风险与影响

• 兼容性：Rust 与 Python 端 wire format 必须严格同步，未来 Python 新增字段需同步更新 Rust。
• 完整性：SamplingParams::normalize() 和 verify() 尚未实现，过渡期需确保 Python 端仍然执行；此外缺失 priority、session_id 等字段支持。
• 测试覆盖：仅有单元测试验证 msgpack 形状，缺少集成测试确保整个请求路径正确。

关联脉络

本 PR 是 Rust 服务器迁移的第二步，依赖 #32240，并从 #29799 拆分。后续 PR 将实现 egress 和 detok 集成，届时需要与本消息层配合完成完整请求生命周期。