Skip to content

perf(api-server): stop keeping finished sequences, keep prompt rendering and tokenizing off the event loop - #2912

Merged
valarLip merged 14 commits into
ROCm:mainfrom
yhl-amd:hyi/api-server-hot-path
Nov 5, 2026
Merged

valarLip merged 14 commits into
ROCm:mainfrom
yhl-amd:hyi/api-server-hot-path

Conversation

@yhl-amd

@yhl-amd yhl-amd commented Nov 2, 2026 •

Copy link
Copy Markdown
Collaborator

Summary

Two API-server costs that grow with agentic conversation length:

  1. Memory leak: finished Sequences are kept forever on a queue nothing reads online. Fixed by making that queue opt-in for its readers (Config.keep_finished_outputs, off by default).
  2. Event loop: chat-template rendering runs on the event loop, and for /v1/messages and /v1/responses the whole prompt was also tokenized there, up to three times. Fixed by rendering off the loop, and for those two endpoints by one executor step that renders, tokenizes the engine's way and cuts to the context.

(An earlier revision also memoized find_last_user_index in DeepSeek's encoding.py. Measured on the checkpoints' own files it gave nothing, so it was dropped; see Validation.)

No new environment variables or flags.

Problem

On DSV4.1-Flash DEP4 + DSpark, MLPerf Endpoints agentic c512 (MI455X, 1 h run), throughput fell from ~12k to ~5k tok/s/GPU. The engine was idle: ~0 queued, KV 2–4% used, prefill and decode step times flat. Requests were waiting unread in the API server's socket receive queues, because the API server could not keep up as conversations grew (later turns ~80k tokens, ~135 messages, whole history resent each turn).

  • Leak. CoreManager's output thread runs outputs_queue.put_nowait(seqs) for every finished batch. Only offline LLMEngine.step() / generate() and the asyncio path read that queue, so online every finished request stays reachable.
  • Event loop. apply_chat_template runs on the event loop in /v1/chat/completions, /v1/messages and /v1/responses. Its cost grows with the conversation.

Changes

  • Config.keep_finished_outputs (default False). Its readers turn it on: LLMEngine.add_request (for step / generate / rollout) and asyncio_mode. Online front ends pass nothing, and the output thread only queues finished sequences when it is on. _bp_fail_stalled_run ends pending seqs that wait on a stream callback through that callback, as abort does.
  • /v1/chat/completions renders with asyncio.to_thread.
  • /v1/messages and /v1/responses run one executor step, _prepare_prompt: render, tokenize through the engine's own path (InputOutputProcessor.encode_prompt: segment cache and prefix block-hash chain), and cut to the context. The reasoning channel (and feat(api): constrain JSON output and expose deployment capabilities #2862's structured channel) is seeded from the text the engine runs, and the engine gets the encoding without tokenizing again. A context with no room for a prompt, or an empty cut, is a 400. Token counts in usage come from the engine.

Validation

  • GPU numbers below were measured on the 10-30 container with env-gated equivalents of these fixes (drop finished sequences, render off the loop), before the review rework. The current head (construction-time default, one prepare step) has only been run on CPU.
  • f01-2, c512 25 min (with the three fixes on): API RSS flat at ~3.6–4.1 GB (the run without the fixes reached 16 GB by minute 32), and throughput the same as without.
    • In a front-end-saturation repro (+30 ms of event-loop CPU per request), keeping the template off the loop removed the decline: running requests 421 → 406 instead of 411 → 325, and unread-socket connections ≤ 9 instead of 58–165.
  • b09-2, c512 1 h (same config): memory flat at ~4.6 GB. Throughput still declines, but about 5 min later and 10–20% higher in the second half. One process remains front-end bound at this request rate; the multi-process front end is a separate PR.
  • Encoder memo, dropped: rendering synthetic agentic histories (18–302 messages, thinking and chat) with DeepSeek-V4-Flash/encoding/encoding_dsv4.py and DeepSeek-V4.1-Flash/encoding/encoding.py, memoized vs raw: identical output, 0.97–1.03x time (1.6 / 2.6 ms at 302 messages). merge_tool_messages folds tool results into user turns, so the backward scan stays short and the O(n²) never shows up.
  • Unit tests (CPU, on 9786106): tests/entrypoints/, test_engine_core_mgr_shared_state.py, test_dp_load_balance.py, test_dp_backpressure.py, test_gc_utils.py, test_distributed_store.py, test_required_channel.py, test_arg_utils_spec.py, test_llm_engine_add_request_fanout.py: 2938 passed, 99 skipped.

🤖 Generated with Claude Code

…tes off the event loop

Three costs in the OpenAI server that grow with the length of agentic
conversations, found on DSV4.1-Flash DEP4 c512 (MI455X) where throughput
fell from ~12k to ~5k tok/s/GPU over a 1 h run:

- CoreManager's output thread put every finished Sequence list on
  outputs_queue, which only offline LLMEngine.step()/generate() read. Online,
  each finished request (token arrays, block tables) stayed reachable for the
  life of the process: API RSS 3 -> 26 GB in 1 h, gen-2 GC pauses
  0.13 -> 0.92 s. The server now clears CoreManager.consumes_outputs_queue.
- apply_chat_template ran on the event loop for /v1/chat/completions,
  /v1/messages and /v1/responses. It now runs on the default executor
  (context variables carried), so the loop keeps reading requests.
- DeepSeek's encoding.py calls find_last_user_index once per rendered
  message, each a backward scan: O(n^2) per render. The loader memoizes it
  per (messages list, length), per thread.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@github-actions

github-actions Bot commented Nov 2, 2026

Copy link
Copy Markdown
Contributor

🏷️ CI Guide

Runs automatically on every eligible PR before approval:

  • ✅ Pre Checkin: Black, Ruff, catalog schema validation, non-GPU unit tests

Heavy model tests:

  • ✅ Run after the PR is approved and Pre Checkin passes
  • ✅ Run immediately when an approval review is submitted
  • ✅ Can be requested before approval with labels
Label Tests
ci:full Run all heavy PR model tests: native ATOM, vLLM, and SGLang
ci:atom Run native ATOM model accuracy tests
ci:vllm Run ATOM vLLM OOT model accuracy tests
ci:sglang Run ATOM SGLang model accuracy tests

Heavy jobs are skipped when the PR is not approved and no matching ci:* label is present.
Add labels via the sidebar or gh pr edit 2912 --add-label <label>

@valar-cop valar-cop left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

some changes needed..

measure the encoder change on the real deepseek encoding.py?

ci green.

Comment thread atom/entrypoints/openai/api_server.py Outdated
engine = engine_args.create_engine(tokenizer=tokenizer)
# Every output reaches its request through the per-sequence callbacks;
# nothing here reads the offline outputs queue.
engine.core_mgr.consumes_outputs_queue = False

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

atomesh front end has the same leak, nothing there calls get_output.. please clear it there too, or default False in CoreManager?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Right, atomesh's standalone service also delivers only through the callbacks. It now clears the flag too (bd367f7). Kept the class default True, since offline generate/step and the asyncio path do read the queue.

local = threading.local()

@functools.wraps(orig)
def find_last_user_index(messages):

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

signature mismatch will 500 every request silently, use *args/**kwargs and forward?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Dropped the memo entirely (bd367f7). On the checkpoints' own encoding.py it measured 0.97-1.03x, so there is nothing left to forward. Numbers are in the top-level reply.

if cached is not None and cached[0] is messages and cached[1] == len(messages):
return cached[2]
result = orig(messages)
local.cached = (messages, len(messages), result)

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

this pins a full history per thread.. weakref instead?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Gone with the memo (bd367f7).

Comment thread atom/model_engine/engine_core_mgr.py Outdated
exc_info=True,
)
self.outputs_queue.put_nowait(seqs)
if self.consumes_outputs_queue:

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

no test hits this line, delete it and suite is still green.. assert the behaviour instead?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Agreed. Added test_finished_sequences_reach_the_outputs_queue_only_if_consumed in tests/test_engine_core_mgr_shared_state.py. It drives the real output thread with an ADD frame and checks the queue size with the flag on and off, so removing the check fails it. The source-grep test now only checks that both front ends clear the flag.

…ur, drop the encoder memo

- atomesh's standalone front end also delivers through callbacks only, so it
  clears consumes_outputs_queue like the OpenAI server.
- The output-thread test drives a real ADD frame and checks the queue with the
  flag on and off, instead of grepping the source.
- Measured on the checkpoints' own encoding.py (V4-Flash, V4.1-Flash): the
  find_last_user_index memo is 0.97-1.03x up to 302 messages, because
  merge_tool_messages folds tool results into user turns and the backward scan
  stays short. Dropped.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@yhl-amd yhl-amd added the ci:atom label Nov 2, 2026
@yhl-amd

yhl-amd commented Nov 2, 2026

Copy link
Copy Markdown
Collaborator Author

Pushed bd367f7 for the review:

  • Leak: atomesh's standalone front end clears consumes_outputs_queue too. The output-thread behaviour now has a real test (ADD frame in, queue checked with the flag on and off).
  • Encoder change, measured and dropped: I rendered synthetic agentic histories with the checkpoints' own encoders, DeepSeek-V4-Flash/encoding/encoding_dsv4.py and DeepSeek-V4.1-Flash/encoding/encoding.py, memoized vs raw. Output was identical, but time was 0.97-1.03x from 18 to 302 messages (V4.1: 2.6 ms raw at 302 messages). merge_tool_messages folds tool results into user turns, so find_last_user_index stops early and the O(n²) never shows up. Thanks for asking for this.
  • Unit tests: 485 passed (api server / chat / anthropic / responses / dp load balance / core manager state).

The PR is now just two parts: the leak fix and rendering the template off the event loop. The leak fix overlaps with #2680 and #2786; once this lands, those can drop their copy on rebase.

@valar-cop re-review please. @valarLip PTAL when you have time.

@valar-cop

valar-cop commented Nov 2, 2026 •

Copy link
Copy Markdown

@yhl-amd

yhl-amd commented Nov 3, 2026

Copy link
Copy Markdown
Collaborator Author

@valarLip gentle ping: valar-cop's follow-up at bd367f7 came back clean, and CI is green. #2913 (multi-process front end) builds on this one; its GPU e2e at head is posted there. Could you take a look when you have a moment? Thanks.

@yhl-amd

yhl-amd commented Nov 3, 2026

Copy link
Copy Markdown
Collaborator Author

@valar-cop Can we merge?

@valar-cop

Copy link
Copy Markdown

@yhl-amd lgtm from my side. atomesh clears the flag, the output-thread test actually fails without the check, and the encoder memo is gone; ci green on bd367f7. only valarLip's approval left.

@valarLip valarLip left a comment

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.

Re-read at bd367f7. Thanks for dropping the memo and for the behavioural queue test. The direction of both fixes is right. I found nothing that returns wrong output. I ran the new tests (3 passed) and timed the tokenizer separately: DeepSeek-V4.1-Flash's encode takes about 117 ms for 100K tokens. What follows is about the depth of the event-loop fix and about how the leak fix is wired.

Must change

1. The larger cost is still on the event loop for /v1/messages and /v1/responses (api_server.py:2277, 2303, 2704, 2723). Right after the render moves to a thread, both endpoints call tokenizer.encode(prompt) on the full prompt, on the loop. When the prompt is over the limit, they encode it again to truncate. preprocess in the executor then encodes it a third time. Your render numbers are 1.6–2.6 ms at 302 messages; encoding an 80–100K-token agentic turn takes about 100 ms. So for these two endpoints, the symptom the description reports as fixed (other connections' sockets go unread at c512) is still there. Please do it the way chat_completions does: the length check and truncation happen inside the one preprocess call in the executor, using seq.num_prompt_tokens, with no encode on the loop. Alternatively, narrow the description's claim.

2. Decide who consumes the outputs at construction, not by assigning a flag from outside (engine_core_mgr.py:198). consumes_outputs_queue is a mutable class default. Each front end flips it after create_engine with engine.core_mgr.consumes_outputs_queue = False (api_server.py:3222, atomesh/server.py:107). That reaches past LLMEngine into CoreManager, and it is a second "who reads the outputs" switch next to the existing config.asyncio_mode. The next online front end (the multi-process front end in the description, for instance) that misses this line brings back the 3 GB → 26 GB/hour leak without any error. Keeping the default True for offline generate / step and the asyncio path is fine. Please make it a construction-time decision: an EngineArgs / Config field at the same level as asyncio_mode, or derived from whether a sequence has a callback. #2786's keep_finished_outputs has the same shape, and the same fix applies. With that change:

  • the field moves into _init_shared_state, next to self.outputs_queue. That is the place its docstring names. Right now it sits inside the class-level backpressure defaults (:194), which splits that group;
  • the source-text test goes away (tests/entrypoints/test_api_server_helpers.py:918-920, inspect.getsource(fn) searched for "consumes_outputs_queue = False"). It passes on dead text and fails on a harmless refactor.

3. _render_chat_template re-implements asyncio.to_thread (api_server.py:1746). The standard library's version is exactly run_in_executor(None, partial(copy_context().run, func, ...)), and this file already uses asyncio.to_thread at 1596 / 1630. Please call await asyncio.to_thread(apply_chat_template, tokenizer, custom_message_encoder, messages, tools=..., **kwargs) at the three call sites. That drops the wrapper, the contextvars / functools imports, and the contextvar part of test_chat_template_renders_off_the_event_loop.

The docstring's "Context variables (request id, tracing) follow the call" is not accurate either:

  • /v1/messages (2275) and /v1/responses (2703) set the request id only after rendering;
  • neither apply_chat_template nor the custom encoders read any ContextVar.

Worth addressing

  • Two serial hops into the default pool per request (api_server.py:1853). There is the render, then setup_*_request's run_in_executor(None, do_preprocess) (843 / 966 / 1289 / 1509). Both share the default pool with multimodal media fetches, which can block in urlopen for up to 30 s. Rendering is pure Python under the GIL, so the thread only yields the loop; it does not save CPU. Doing render and tokenize as one "prepare the prompt" step inside the existing preprocess call would cost one hop instead of two, and it is also where item 1 puts the encode.
  • The backpressure send-failure path, pre-existing (engine_core_mgr.py:1290). It still does outputs.put(RuntimeError(...)) unconditionally. Online, that is now a queue declared unread, so _bp_pending requests get no terminal output and their clients hang. This PR is what makes "online does not read this queue" explicit, so the error should reach those requests through their callbacks.

yhl-amd and others added 2 commits November 3, 2026 12:13
…utputs at construction, asyncio.to_thread

- /v1/messages and /v1/responses render with asyncio.to_thread and pass
  max_prompt_tokens down; the length check and truncation run in the
  preprocess call on the executor (_preprocess_prompt), from the engine's
  own num_prompt_tokens. Input/output token counts come from the engine, so
  neither endpoint encodes on the event loop any more.
- Config.keep_finished_outputs (next to asyncio_mode), set at construction
  through EngineArgs.create_engine(keep_finished_outputs=False) by both online
  front ends; CoreManager reads it in _init_shared_state. The class-level flag
  and the source-text test are gone.
- _render_chat_template is gone: asyncio.to_thread at the call sites.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@yhl-amd

yhl-amd commented Nov 3, 2026

Copy link
Copy Markdown
Collaborator Author

@valarLip thanks for the careful read. Pushed 3036898 (on top of a merge of current main). Item by item:

Must change

  1. No encode on the loop for /v1/messages and /v1/responses. Both now render with asyncio.to_thread and pass max_prompt_tokens to setup_streaming_request / generate_async. The length check and truncation happen in the existing preprocess call on the executor (_preprocess_prompt), from seq.num_prompt_tokens. The prompt is tokenized once; only an over-long one is rebuilt from its own first max_prompt_tokens ids, and the discarded build is popped from io_processor.requests. The headroom rule (_prompt_headroom, so both PD legs truncate alike) is unchanged. It sits in one helper, _max_prompt_tokens, instead of two copies. That also drops the per-request logger.warning("[anthropic] resolved max_ctx=..."). input_tokens now comes from the engine (num_prompt_tokens / num_tokens_input). The non-stream output_tokens = len(tokenizer.encode(raw_text)) on the loop is replaced by the engine's num_tokens_output, so neither endpoint calls tokenizer.encode on the loop any more. The PD truncation tests still check the same thing: a prefill leg with max_tokens=1 and a decode leg with 4096 get the same prompt with a handoff, and different ones without.
  2. Decided at construction. It is now Config.keep_finished_outputs (default True), next to asyncio_mode. Both online front ends build with engine_args.create_engine(tokenizer=..., keep_finished_outputs=False). CoreManager reads it in _init_shared_state, next to outputs_queue, using the same getattr(config, ..., default) as the other stand-in-tolerant fields there. The class attribute and the source-text test are gone. Tests: the output-thread test now builds the manager with the config value, and a new test checks that create_engine hands the choice to LLMEngine/Config with the default kept. feat(api-server): --api-server-count N, front-end worker processes for the OpenAI server #2913 will use the same keyword.
  3. asyncio.to_thread. _render_chat_template is gone, along with the contextvars / functools imports and the contextvar test. You're right that the docstring claim didn't hold.

Worth addressing

  • One hop instead of two. Agreed on the reasoning (GIL, a shared default pool with media fetches). I tried folding the render into the preprocess call. The catch is that the chat path decides the reasoning channel from the rendered text before it builds the stream/non-stream generators, which take reasoning= up front. So moving the render there means restructuring all four chat call shapes, plus the stubs the compat/PD tests use at that boundary. I'd rather do that as its own follow-up than grow this PR. For messages/responses, this change already makes it render plus one tokenize, both off the loop.
  • Backpressure send-failure path. Online requests never enter _bp_pending. add_request only takes the backpressure path for an offline batch (_bp_offline_batch, no rank hints, no sticky sessions); the comment there says online requests "must not wait in _bp_pending", and test_online_add_does_not_enter_pending_pool pins it. So the only seqs _bp_fail_stalled_run can strand are offline ones, and offline generate reads outputs_queue. If that invariant ever changes, the error should go through the callbacks as you say; I left a note rather than code for a path that can't occur today. Happy to add it anyway if you prefer.

Unit tests (CPU): tests/entrypoints/, core manager state, dp load balance / backpressure, gc utils, distributed store, required channel, arg_utils: 2817 passed, 98 skipped.

@valar-cop re-review please.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@yhl-amd yhl-amd changed the title perf(api-server): stop keeping finished sequences, render chat templates off the event loop perf(api-server): stop keeping finished sequences, keep prompt rendering and tokenizing off the event loop Nov 3, 2026

@valar-cop valar-cop left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

a few small things..

ci green.

Comment thread atom/entrypoints/openai/api_server.py Outdated
token_ids = tokenizer.encode(prompt)[:max_input]
prompt = tokenizer.decode(token_ids, skip_special_tokens=False)
input_tokens = max_input
max_input = _max_prompt_tokens(request.max_tokens, kv_transfer_params)

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

prompt is untruncated here, so the filter is seeded from text the engine never runs.. seed from the sent prefix?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Good catch, fixed in b7387b9. The cap now travels as _PromptCap. When the preprocess call cuts the prompt, it decodes the kept ids on the executor into cap.cut_text, and both endpoints seed the reasoning channel from that (stream and non-stream). That matches what the old decode-then-send did, without the loop work. The unit test checks cut_text.

Comment thread atom/config.py Outdated
# front ends deliver every output through per-sequence callbacks and build
# the engine with this off; left on there, every finished Sequence stays
# on a queue nothing reads, for the life of the process.
keep_finished_outputs: bool = True

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

pr description still describes the old design, please update it to match the diff..

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Updated after this review: Summary/Changes now describe Config.keep_finished_outputs via create_engine, asyncio.to_thread, and the check/truncation in the preprocess call. Validation lists the current unit test run.

unittest.main()


def test_online_front_ends_build_the_engine_without_kept_outputs(monkeypatch):

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

this never touches a front end, nothing pins the flag. drive the stubs and assert kwargs, or rename it?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Fair. Renamed it to test_create_engine_hands_keep_finished_outputs_to_the_config, which is what it checks. Added test_atomesh_front_end_builds_the_engine_without_kept_outputs, which drives atomesh's initialize_engine with stubbed EngineArgs / tokenizer and asserts it passes keep_finished_outputs=False. The OpenAI server's main() is too big to drive in a unit test; its call is the same one-line keyword.

… the front end

- _PromptCap carries the cap into the preprocess call and, when the prompt is
  cut, the decoded text of what the engine runs (decoded on the executor);
  /v1/messages and /v1/responses seed the reasoning channel from it.
- Tests: create_engine's test renamed to what it checks; a new test drives
  atomesh's initialize_engine and asserts keep_finished_outputs=False.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@yhl-amd

yhl-amd commented Nov 3, 2026

Copy link
Copy Markdown
Collaborator Author

Pushed b7387b9 for valar-cop's three follow-ups (replies inline): a cut prompt now seeds the reasoning channel from the text the engine runs, decoded on the executor; there's a test that drives a front end (atomesh) and asserts keep_finished_outputs=False; and the description matches the diff. CPU unit tests: 2818 passed.

@valarLip @valar-cop re-review please.

@yhl-amd

yhl-amd commented Nov 3, 2026

Copy link
Copy Markdown
Collaborator Author

@valar-cop can we merge

@valar-cop

Copy link
Copy Markdown

@yhl-amd not yet, the ruff job is still red on this head (log is empty, rerun or check the format locally): https://github.com/ROCm/ATOM/actions/runs/41581924513/job/125738494221

your three follow-ups look right, cut_text seeds the reasoning channel in both non-stream branches and the atomesh test drives the real front end. lgtm once ruff is green, and valarLip still owes a re-read after 3036898.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@yhl-amd

yhl-amd commented Nov 3, 2026

Copy link
Copy Markdown
Collaborator Author

Thanks @valar-cop. The ruff failure was RUF023 (_PromptCap.__slots__ not sorted), the only finding in the diff; fixed in 6ca4ec8.

@valar-cop

Copy link
Copy Markdown

@yhl-amd yeah, __slots__ is sorted at atom/entrypoints/openai/api_server.py:1259, and the leftover contextvars / functools imports are gone. the new ci is queued, will confirm once the ruff job reports; nothing else from me otherwise.

…sponses decides its channel before the engine call)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@yhl-amd

yhl-amd commented Nov 3, 2026

Copy link
Copy Markdown
Collaborator Author

Merged current main (80aea67) to resolve the conflict with #2862:

@valar-cop valar-cop left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

some changes needed..

ci green.

# Decided before the engine call, since structured output configures
# the sampling params from it; so for a prompt over the cap it reads
# the template's ending, not the cut text the engine runs.
_reasoning = reasoning_channel(

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

channel seeded from untruncated prompt here, /v1/messages uses cut_text.. recompute _reasoning after cut_text for schema None, and seed structured from the cut prompt

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Done in 991517c, both paths:

  • No schema: the preprocess call cuts the prompt as before, and the channel is then re-seeded from cut_text (_reseed_after_cut), stream and non-stream, the same as /v1/messages.
  • Structured: feat(api): constrain JSON output and expose deployment capabilities #2862 builds the channel into the sampling params before the engine call, so here the cut has to come first. _cut_prompt tokenizes off the loop (asyncio.to_thread) and, if the prompt is over the cap, returns the kept ids and their decoded text. The channel is seeded from that text, and the engine gets the ids, so preprocess doesn't cut again. This costs one extra tokenize for structured requests only, and none on the loop.
  • Unit test test_responses_channel_follows_the_text_the_engine_runs covers both helpers. CPU tests: 2855 passed.

@valar-cop re-review please.

…ngine runs

- With a schema, the channel goes into the sampling params before the engine
  call, so the prompt is cut first (_cut_prompt, off the loop) and the engine
  gets the kept ids.
- Without one, the preprocess call cuts it and the channel is re-seeded from
  cut_text afterwards (_reseed_after_cut), as /v1/messages does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

@valar-cop valar-cop left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

a few small things..

ci green.

Comment thread atom/entrypoints/openai/api_server.py Outdated
token_ids = tokenizer.encode(prompt)
if len(token_ids) <= prompt_cap.tokens:
return prompt, prompt
token_ids = token_ids[: prompt_cap.tokens]

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

no log on truncation here? preprocess used to warn, now silent for structured..

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Right. _cut_prompt now logs the same "Prompt too long (n > cap), truncating" warning as the preprocess path (bf3f38a).

assert list(requests) == [seq.id]


def test_responses_channel_follows_the_text_the_engine_runs(monkeypatch):

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

this doesn't drive responses_create, so deleting the call stays green.. can we add a structured case over the cap?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Added TestStructuredResponsesCutFirst in test_pd_messages_responses.py (bf3f38a). It drives responses_create with json_object and an over-cap prompt, and asserts three things: the engine is handed the kept ids (a list, already at the cap), the structured config ran, and the channel was seeded from the cut text before the engine call. Removing the _cut_prompt call fails it, because the engine would then get the uncut string. CPU tests: 2856 passed.

@valar-cop re-review please.

…er-cap case

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

@valar-cop valar-cop left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

a few small things..

ci green.

"prompt_starts_in_reasoning",
lambda text: seeds.append(text) or False,
)
monkeypatch.setattr(api_server, "validate_output", lambda *a, **kw: None)

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

this patch is dead for the non-stream path, it calls validate_responses_output. patch that and assert status 200?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Good catch. It now patches validate_responses_output (the one responses calls) and asserts 200 (25cb433). @valar-cop re-review please.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

@valarLip valarLip left a comment

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.

Re-read at 25cb433, by reading the code and running the affected tests (test_pd_messages_responses, test_api_server_helpers, test_engine_core_mgr_shared_state, test_api_compatibility, test_gc_utils: 281 passed); no GPU. Thanks for working through both rounds. The leak fix is right: nothing online reads outputs_queue (atomesh delivers by callback, asyncio_mode is never on). The render now runs off the loop via asyncio.to_thread. Non-streaming usage now uses the engine's counts, which makes it agree with streaming. That's a semantic change, not a bug.

What's left is mostly the over-length cut for /v1/messages and /v1/responses. Following my 11-03 item 1, valar-cop's seed-from-cut request and #2862's structured channel, it has grown into two mechanisms, so part of this is on me. Each item ends with the change I'm asking for.

A. One prompt preparation step instead of two cut mechanisms

1. Two cut paths, five helpers and a mutable out-parameter (api_server.py:1278-1347). With a schema, _cut_prompt cuts before the engine call. Without one, _preprocess_prompt cuts inside preprocess. To bridge them the PR adds _PromptCap (whose cut_text is filled in on the executor), _reseed_after_cut, and a prompt_cap parameter on both generate_async and setup_streaming_request. The write-back is spelled three ways (responses builds _reasoning then reseeds; anthropic does prompt = prompt_cap.cut_text or prompt), and the two paths must agree on every count. Three symptoms follow:

  • _cut_prompt (:1318) tokenizes with plain tokenizer.encode, bypassing io_processor.segment_tokenizer, the prefix cache built for agentic resends. When the prompt fits it returns the string, so the engine tokenizes it again. Every structured /v1/responses request, not only over-long ones, pays a full encode twice; an 80k-token agentic turn pays it every turn.
  • _preprocess_prompt (:1336) builds a whole Sequence first: full tokenize, ship_prompt_hash_chain, a "Request arrived" log line, a seq id. Then it reaches into engine.io_processor.requests.pop(seq.id) from the API layer and builds it again from the ids, with no prefix_encoding, so the hash chain is recomputed from scratch.
  • On /v1/responses, _reasoning is built from the uncut prompt and replaced after the engine call (:2890). With a schema, _reseed_after_cut still runs unconditionally (:2914 streaming, :3034 non-streaming), so if the two counts ever disagree, a structured response would lose literal_answer. That's not reachable while segment_tokenizer is exact, but it comes from the duplication.

→ Change: make one "prepare the prompt" step on the executor that renders, tokenizes through the engine's own path (io_processor, segment cache included), cuts to the cap, and returns (token_ids, cut_text_or_None). Run it once before building the sampling params, so #2862's structured channel and the plain reasoning channel are both seeded from what the engine runs, and pass the ids to the engine so preprocess doesn't tokenize again. That also merges the two executor hops from my 11-03 note into one. Then delete _PromptCap, _cut_prompt, _reseed_after_cut, _preprocess_prompt's rebuild and the prompt_cap parameters.

2. The cap can be ≤ 0 (api_server.py:1306-1311). _prompt_headroom reserves at least min(max_tokens, 1024). With --max-model-len 512 and max_tokens=1024, tokens is ≤ 0, so every prompt counts as over-long and is cut to empty or has its tail dropped ([: negative]), instead of being refused. If cut_text is "", anthropic's cut_text or prompt falls back to the full text. The old code had this too, but these are the lines this PR rewrites.
→ Change: in the item 1 step, refuse with a 400 when the cap is < 1, and treat an empty cut as "refuse", not "use the full prompt".

B. The leak switch

3. keep_finished_outputs defaults to True, so every online front end has to remember False (engine_core_mgr.py:237). This is the shape I asked for on 11-03 (a construction-time field next to asyncio_mode, default kept for offline). Seeing it wired, the default is on the wrong side. This PR itself first missed atomesh. The multi-process front end in #2913, and any later one, has to pass it too, or the 3 GB → 26 GB/hour leak returns silently. test_gc_utils already greps source for the same kind of "every front end must remember" rule (freeze_gc_heap).
→ Change: default it to False, and have the readers of outputs_queue turn it on: LLMEngine.generate / step, and asyncio_mode. Then the online front ends pass nothing, and the create_engine / atomesh plumbing and their tests go away. Also read config.keep_finished_outputs directly instead of getattr(config, "keep_finished_outputs", True), so a missing field fails rather than silently leaking.

4. The _bp_fail_stalled_run docstring states an invariant without enforcing it (engine_core_mgr.py:1308-1319). It says only offline seqs are ever in _bp_pending because add_request keeps online ones out, and "if online seqs could wait there, this error would have to reach them through their callbacks". That answers my 11-03 note in prose, but nothing breaks if it stops being true, and the hypothetical branch is just text.
→ Change: replace the conditional paragraph with an assertion where seqs enter _bp_pending (no callback on a pooled seq), plus a test for it.

C. Tests and comments

5. The e2e cut tests check the fake, not _preprocess_prompt (tests/entrypoints/test_pd_messages_responses.py:~601). The fake engine_prompt() implements truncation by prompt_cap.tokens itself and counts any id list as 2 tokens. So deleting the cut from _preprocess_prompt leaves TestBothLegsTruncateAlike / TestStructuredResponsesCutFirst green, and the ids path's input_tokens is never asserted.
→ Change: let the fake do only what the engine does (tokenize and report the real count), and keep the cut in production code under test. With item 1, test the single prepare step directly.

6. The new tests sit in the wrong file (tests/test_engine_core_mgr_shared_state.py:222). Two pytest functions are appended after if __name__ == "__main__": unittest.main(), and they test arg_utils.create_engine and atomesh's initialize_engine, not CoreManager's shared state. test_create_engine_hands_keep_finished_outputs_to_the_config checks the kwargs handed to a fake LLMEngine, not the config. The atomesh case leaves the module globals engine / tokenizer set to "engine" / "tok".
→ Change: with item 3 these tests go away. Otherwise, move them to test_arg_utils_spec.py and the atomesh tests, name them for what they check, and restore the globals (monkeypatch.setattr).

7. The same explanation is written in many places (config.py:2106, engine_core_mgr.py:234, arg_utils.create_engine, atomesh/server.py, api_server.main, and the three to_thread call sites around api_server.py:1996). The "front ends read by callback, nothing drains the queue" rationale appears five times. Each to_thread site carries a near-identical "Off the loop..." note, and generate_async / setup_streaming_request gained prompt_cap docstring lines that item 1 removes.
→ Change: keep one constraint line on the Config field and delete the rest.

yhl-amd and others added 2 commits November 3, 2026 23:12
…ault

- /v1/messages and /v1/responses: _prepare_prompt renders, tokenizes through
  the engine's own path (InputOutputProcessor.encode_prompt: segment cache
  and block-hash chain) and cuts to the context, once, on the executor,
  before the structured/reasoning channels are built. The engine gets the
  encoding and does not tokenize again. A context with no room for a prompt,
  or an empty cut, is a 400. _PromptCap, _cut_prompt, _reseed_after_cut, the
  preprocess rebuild and the prompt_cap parameters are gone.
- Config.keep_finished_outputs defaults to False; its readers turn it on
  (LLMEngine.add_request, asyncio_mode). Online front ends pass nothing;
  the create_engine / atomesh plumbing and their tests are gone. Read
  directly from the config.
- _bp_fail_stalled_run ends pending seqs that wait on a callback through it,
  as abort does, instead of a docstring promise.
- Tests: the fakes only count what the engine is handed; the prepare step is
  tested directly; trimmed duplicate comments.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@yhl-amd

yhl-amd commented Nov 3, 2026

Copy link
Copy Markdown
Collaborator Author

@valarLip thanks for the second read. Pushed 9786106 (on a merge of current main). Item by item:

A1. One prompt-preparation step. _prepare_prompt runs once, via asyncio.to_thread, after the sampling params are built and before the reasoning / structured channels. It renders, then tokenizes through the engine's own path: a new InputOutputProcessor.encode_prompt, which is the code preprocess_fanout used inline (segment cache, and the prefix block-hash chain when enabled), so preprocess now calls it too. It then cuts to the cap and returns (engine_input, text). Both endpoints seed _reasoning, and #2862's structured channel, from text. They hand engine_input to the engine, and preprocess_fanout accepts a PrefixEncoding / id list without tokenizing again. One hop, one tokenize, through the segment cache. _PromptCap, _cut_prompt, _reseed_after_cut, the preprocess rebuild and the prompt_cap parameters are gone. A cut prompt goes as plain ids, so its chain is hashed in preprocess; that only happens on the over-cap path.

A2. A cap < 1 raises ValueError (a 400 on both endpoints, naming max_tokens and the context). An empty cut is refused the same way, not replaced by the full prompt.

B3. Config.keep_finished_outputs now defaults to False, and its readers turn it on:

  • LLMEngine.add_request (the entry for step / generate / rollout) sets it before submitting;
  • asyncio_mode implies it.

Online front ends pass nothing. The create_engine parameter, the atomesh / api_server plumbing and their tests are gone. _init_shared_state reads config.keep_finished_outputs or config.asyncio_mode directly; the test stand-in configs gained the two fields.

B4. I replaced the conditional paragraph with code rather than an assertion. Main already treats a callback seq in the pool as possible: LLMEngine.add_request takes stream_callback inside offline_backpressure_batch, the abort path delivers through the callback for a pending seq, and test_abort_pending_seq_is_not_dispatched pins that. An assertion at the pool's entry would break that test. So _bp_fail_stalled_run now takes any pending seq that has a callback out of the pool and ends it through the callback, as abort does; the outputs_queue error still goes to generate. Test: test_a_stalled_run_ends_pending_seqs_that_wait_on_a_callback. If you'd rather have the pool callback-free, I can switch to the assertion and adjust the abort test instead.

C5. The fakes only count what they are handed (num_tokens_input = len(ids)); the cut lives in _prepare_prompt, which is tested directly (TestPreparePrompt):

  • fits, with the encoding and chain passed through;
  • over-long, cut to the cap;
  • no room, 400;
  • empty cut, 400.

The PD truncation tests now check the length of the ids the engine got, and a new test asserts usage.input_tokens equals the engine's count of the cut prompt.

C6. Gone with B3. The remaining additions are in the files of what they test (test_llm_engine_add_request_fanout.py for the offline turn-on, test_dp_backpressure.py, test_api_server_helpers.py).

C7. One line on the Config field; the other repeats and the to_thread note are deleted.

CPU tests (tests/entrypoints/, core manager state, dp load balance / backpressure, gc utils, distributed store, required channel, arg_utils, llm_engine add_request): 2938 passed. ruff 0.17.3 and black are clean on the diff. No GPU this round.

@valar-cop re-review please.

@valar-cop valar-cop left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

a few small things..

ci failed: Pre Checkin, could not tell why yet..

Comment thread atom/config.py
asyncio_mode: bool = False
# Queue finished sequences on outputs_queue; only its readers turn it on
# (LLMEngine.add_request for step / generate, and asyncio_mode).
keep_finished_outputs: bool = False

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

the PR description still describes the old design, please update summary and validation to match this head..

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Updated. The Summary now names both fixes as they are at this head. Validation now says plainly that its GPU numbers were measured on the 10-30 container with env-gated equivalents, before the review rework; this head has only been run on CPU.

)
token_ids = encoded.token_ids if isinstance(encoded, PrefixEncoding) else encoded
if len(token_ids) <= cap:
return encoded, prompt

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

widen these annotations to include PrefixEncoding? now they hide the third input

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

Done (5cfc14e): generate_async / setup_streaming_request take str | list[int] | PrefixEncoding, and _prepare_prompt declares its tuple[PrefixEncoding | list[int], str] return. @valar-cop re-review please.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@valarLip
valarLip merged commit df61c5d into ROCm:main Nov 5, 2026
36 of 53 checks passed
yhl-amd added a commit to yhl-amd/ATOM that referenced this pull request Nov 5, 2026
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
ZLkanyo009 added a commit that referenced this pull request Nov 5, 2026
main's #2912 made `CoreManager._init_shared_state` read
`keep_finished_outputs` and `asyncio_mode`, which the hand-built
namespace lacked. `atom_config_double` derives its fields from
`dataclasses.fields(Config)`, so a field main adds next arrives with its
default.

Co-authored-by: Cursor <cursoragent@cursor.com>
xysheng-AMD added a commit to ZhangDanyang-AMD/Lumen-RL that referenced this pull request Nov 6, 2026
Since ROCm/ATOM#2912, CoreManager puts finished sequences on
outputs_queue only once a reader has asked for them, which
LLMEngine.add_request does. The pump adds requests to CoreManager itself,
so it never asked: every finished sequence was dropped, no group came
back, and a fully async run waited forever for its first batch.

The pump now asks when it starts. An older ATOM queues them anyway and
never reads the flag. The test engine drops finished sequences until
asked, as CoreManager does.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants