Skip to content

sglang rust server tm ingress - #32361

Closed
rainj-me wants to merge 4 commits into
mainfrom
rainj-me/rust-server-tm-ingress
Closed

rainj-me wants to merge 4 commits into
mainfrom
rainj-me/rust-server-tm-ingress

Conversation

@rainj-me

@rainj-me rainj-me commented Jul 25, 2026 •

Copy link
Copy Markdown
Collaborator

Motivation

Split PR from #29799

Depends on #32358

Modifications

Accuracy Tests

Speed Tests and Profiling

Checklist

Review and Merge Process

  1. Ping Merge Oncalls to start the process. See the PR Merge Process.
  2. Get approvals from CODEOWNERS and other reviewers.
  3. Trigger CI tests with comments or contact authorized users to do so.
    • Common commands include /tag-and-rerun-ci, /tag-run-ci-label, /rerun-failed-ci
  4. After green CI and required approvals, ask Merge Oncalls or people with Write permission to merge the PR.

CI States

Latest PR Test (Base): ✅ Run #30491048579
Latest PR Test (Extra): ❌ Run #30491048349

@gemini-code-assist

Copy link
Copy Markdown
Contributor

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

@rainj-me
rainj-me force-pushed the rainj-me/rust-server-tm-ingress branch from cb1c421 to 8b9d769 Compare July 25, 2026 00:28
@rainj-me rainj-me mentioned this pull request Jul 25, 2026
5 tasks done
@alexnails
alexnails changed the base branch from main to rainj-me/rust-server-tokenizer-manager July 26, 2026 20:10
@alexnails
alexnails changed the base branch from rainj-me/rust-server-tokenizer-manager to rainj-me/rust-server-tm-sampling July 26, 2026 20:11
req: &mut Request,
skip_tokenizer_init: bool,
vocab_size: Option<u64>,
) -> Result<(), Error> {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do we need parity here with _validate_one_request?

}
}
}
if let Some(rmpv::Value::Array(ids)) = &g.token_ids_logprob {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

double check: do we want any validations for token_ids_logprob?

// here instead — mirroring the Python `TokenizerManager` validation.
if let (Some(vs), RequestKind::Generate(g)) = (vocab_size, &req.kind) {
if let Some(ids) = &g.input_ids {
for &id in ids {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we need to clean up per request overhead here

[agent verification]

~40 µs/request recoverable on the single ingress thread, at 5.7×.

This loop is O(prompt_len) on the one thread that admits every request. The id as u64 widening plus the two-predicate form keeps LLVM from vectorising it. Measured (128k-token prompt — the bench_serving client-input_ids shape — --release, 200 iters):

per request
id < 0 || id as u64 >= vs (as written) 48.4 µs
(id as u32) >= vs as u32, folded with | 8.5 µs
drive() end-to-end, 3-token request (reference) 2.2 µs

The trick is that for i32, id < 0 || id >= vs is exactly (id as u32) >= vs whenever vs <= u32::MAX — negatives wrap above every real vocab size. One unsigned compare per element, no branch, so it vectorises:

match u32::try_from(vs) {
    Ok(vs32) => {
        // fold rather than `any`, so there's no early-exit branch to block SIMD
        if ids.iter().fold(0u32, |a, &id| a | ((id as u32) >= vs32) as u32) != 0 {
            return Err(/* … */);
        }
    }
    // vocab beyond u32 can't be exceeded by any non-negative i32
    Err(_) => if ids.iter().any(|&id| id < 0) { return Err(/* … */); },
}

Note the trade-off I measured against: the fold drops the early break, so a request whose first id is bad now scans the whole prompt. That's the right way round — valid is the overwhelmingly common case, and if you care, checking the accumulator once per 4 KiB chunk keeps most of the vectorisation with a bounded worst case.

The error message needs the offending id, so it'd re-scan to find it on the rejection path. That's fine — it's the path that's about to 400 anyway.

Same shape, ~34 µs, at the input_ids_i64_le() call on line 278. That function lives in message/request.rs on the base branch so I can't anchor there, but it's this PR's caller and the same request:

per request
per-element extend_from_slice(&(id as i64).to_le_bytes()) (as written) 65.5 µs
Vec<i64> widened via extend, then one byte copy 32.0 µs

Byte-at-a-time extend_from_slice into a Vec<u8> can't vectorise the widening; separating the widen from the byte copy lets it. Together with the vocab loop that's ~74 µs of a ~116 µs long-request ingress path.

To be clear about the stakes, same as on #32249: prefill for a 128k prompt is seconds, so none of this is visible per-request. It matters because this is the single-threaded admission point, so these two loops set the request-rate ceiling — roughly 8.7k/s → 25k/s for this work alone.

/// with positional arguments.
pub struct Ingress {
rx: flume::Receiver<TmEvent>,
senders: Senders,

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same as my other comment on egress PR on what is actually used here

// Hand off to the tokenizer pool; it returns the request as a
// `Tokenized` event (Queued, or Failed on error). Doesn't loop.
RequestState::Tokenizing => {
if let Err(err) = self.senders.tok.send(req) {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we are for sure blocking here?

let (decode_logprob_text, no_stop_trim) = match &req.kind {
RequestKind::Generate(g) => (
g.return_text_in_logprobs.unwrap_or(false),
sampling_flag(&g.sampling_params, "no_stop_trim"),

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we read no_stop_trim out of the samplingparams before normalize_sampling_params has been called?

/// Client disconnected: deregister, then push an `AbortReq(rid)` so the
/// scheduler stops generating. Fire-and-forget (a full ring drops the abort;
/// the request then finishes at EOS).
fn on_abort(&self, rid: String) {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

on_abort acts on any rid string with no validations rust side? (iirc there is python side check? can we clean this behavior up in any way? (or todo for now?)

}
};

if !self.ingress.try_push(IngressMsg { header, ids }) {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we have serialization before necessary here?

[agent verification]

By the time try_push reports the ring is full, this request has already paid for its full serialization: to_header_msgpack (a fresh Vec plus a 16-element rmpv array) and input_ids_i64_le (a prompt_len * 8 allocation — 1 MB for a 128k prompt, measured at ~66 µs). All of it is dropped on the floor.

That's the wrong way round under sustained backpressure, which is exactly when the ring is full: the more overloaded the scheduler is, the more ingress CPU goes to requests that get a 503. Checking capacity first turns the overload path cheap:

if self.ingress.is_full() {           // thin wrapper over flume's len()/capacity()
    self.fail(&mut req, Error::QueueFull);
    return;
}

Racy, so keep the try_push rejection as the real gate — this is just an early-out that skips the serialization in the common overload case.

While here: serialized builds (header, ids) as a tuple so the immutable borrow of g ends before fail(&mut req), and the comment explains that clearly — good. Worth noting the tuple also means the 1 MB ids buffer stays alive across the try_push, which is unavoidable, but it's the reason an overloaded server's ingress thread holds one prompt-sized buffer per in-flight rejection.

}

fn generate_req(id: u64, sampling_params: rmpv::Value) -> Request {
let (tx, _rx) = mpsc::channel(8);

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

have tests that validate client side? (or does this fall too close to integration test? fine to say no)

}
if let Some(rmpv::Value::Array(ids)) = &g.token_ids_logprob {
for v in ids {
let id = v.as_i64().unwrap_or(-1);

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

split out a message that says what's actually wrong when we have bad id?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

amd apple-silicon blackwell SM100/SM120 deepseek dependencies Pull requests that update a dependency file diffusion SGLang Diffusion documentation Improvements or additions to documentation hicache Hierarchical Caching for SGLang jit-kernel lora Multi-modal multi-modal language model npu quant LLM Quantization speculative-decoding

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants