feat(multimodal): ThreadedMicroBatcher + batcher-backed AsyncVisionEncoder (cross-request batching)#11037
Open
furionw wants to merge 21 commits into
Open
feat(multimodal): ThreadedMicroBatcher + batcher-backed AsyncVisionEncoder (cross-request batching)#11037furionw wants to merge 21 commits into
furionw wants to merge 21 commits into
Conversation
furionw
temporarily deployed
to
external_collaborator
GitHub Actions
Inactive
github-actions
Bot
added
feat
backend::vllm
This comment has been minimized.
This comment has been minimized.
This was referenced Jun 29, 2026
feat(multimodal): CUDA-graph bucket ladder (Dynamo-side target_bucket) + Qwen3-VL ViT example
#11039
Closed
furionw
force-pushed
the
qiwa/partial-embed-assembly
branch
from
768e5c8 to
90cc67e
Compare
furionw
force-pushed
the
qiwa/vision-encoder-batcher
branch
from
fa42f16 to
4be49c4
Compare
furionw
force-pushed
the
qiwa/partial-embed-assembly
branch
from
90cc67e to
1c5be47
Compare
furionw
force-pushed
the
qiwa/vision-encoder-batcher
branch
from
4be49c4 to
3c1f269
Compare
furionw
force-pushed
the
qiwa/partial-embed-assembly
branch
from
1c5be47 to
ada3d9a
Compare
furionw
force-pushed
the
qiwa/vision-encoder-batcher
branch
from
3c1f269 to
729398f
Compare
furionw
force-pushed
the
qiwa/partial-embed-assembly
branch
from
ada3d9a to
b3a7c13
Compare
furionw
force-pushed
the
qiwa/vision-encoder-batcher
branch
from
729398f to
4f361b7
Compare
furionw
force-pushed
the
qiwa/vision-encoder-batcher
branch
from
4f361b7 to
b86a07f
Compare
furionw
force-pushed
the
qiwa/partial-embed-assembly
branch
from
b3a7c13 to
5677c09
Compare
furionw
force-pushed
the
qiwa/vision-encoder-batcher
branch
from
b86a07f to
7d148e3
Compare
…IXL/SHM path Reword the confusing 'would otherwise silently win' aside: the point is simply that a configured encoder owns the image path, so route there instead of the normal NIXL/SHM pre-render + HF path.
furionw
force-pushed
the
qiwa/partial-embed-assembly
branch
from
d8118b4 to
1c6d658
Compare
…er path In _assemble_custom_encoder_prompt, non-image modalities are already rejected earlier, so when no usable image URLs remain there is nothing left to load — _extract_multimodal_data can only walk empty lists and return an empty/None result. Return (None, None, None) directly for the text-only case and drop the now-unused mm_processor_kwargs parameter. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
image_urls is image_items filtered to dicts with a 'Url' key, so the old `image_items and not image_urls` check only fired when ALL items were malformed — a mixed list (some 'Url', some 'Decoded'/malformed) slipped through and was encoded with the bad items silently dropped. Compare lengths instead: reject the request if any item lacks a usable 'Url'. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…coder (cross-request batching) PR3 of the custom vision-encoder series, stacked on PR2. Replaces the serial direct-call glue with cross-request batching to the author's max_batch_cost token budget (scalar cost, one-dimensional packing — no shape buckets, no graph ladder; that lands in PR4). handlers.py is unchanged — the AsyncVisionEncoder public surface is the stable boundary. - ThreadedMicroBatcher (generic, torch-free): one pinned non-daemon actor thread running on_start (build) / every fn(items) / on_stop (close); eager drain-on-completion (no timer; max_wait_ms is an opt-in knob); coalesce by scalar cost up to max_batch_cost (None = pass-through). Per-item idempotent finalizer; cancellation + repeat-cancel retirement; worker supervisor; optional max_outstanding_cost admission; cross-loop-safe shutdown. submit(items, costs). - AsyncVisionEncoder now routes preprocessed items through the batcher (was a single-worker serial executor); the A5 preprocess barrier is unchanged; build is build(model_id) and the hardcoded image_token_id is validated at load. Tests: batcher concurrency suite (eager-drain, cost budget, pass-through, cancellation/retirement, supervisor, lifecycle, on_stop) + batcher-backed glue. Note: based on the series' validated merge-base; rebase onto main before merge. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…Batcher shutdown() returned early when a slow fn was still in flight and relied on a later call to reap — but the worker already self-exits once its fn returns and reads the stop sentinel. Make the drain sentinel-safe (re-put it instead of consuming it) so shutdown can finalize in one linear pass: signal, bounded join, mark closed, drain; warn if a slow fn is still finishing (it exits on its own).
…cher (TODO) Cancellation added real surface — a per-request `cancelled` flag + `retired` future, a CancelledError/shield/repeat-cancel block in submit(), and cancelled checks in tombstoning/finalisation — for an optimization, not correctness. Drop it for v0 with a TODO: a cancelled await now abandons its result but its items still run through fn (admission frees when they finish). Safe (Python refcounting keeps the items alive; the abandoned future no-ops via InvalidStateError). Failure tombstoning (a sibling batch's error skips a request's remaining items) is kept — that's request atomicity, not cancellation.
…ckpressure - Merge CLOSING/CLOSED into one CLOSED state. Nothing branched on the difference (submit, the shutdown idempotency guard, and _run_batch all treated them the same); only a bookkeeping transition distinguished them. - Drop the unused max_outstanding_cost admission ceiling and its machinery (_outstanding, cost_total, BatcherOverloaded): AsyncVisionEncoder never sets it, so the check never fired. TODO left to add real backpressure if a producer can outrun the encoder. on_start/on_stop stay -- they pin build/close to the actor thread (CUDA-graph affinity).
furionw
force-pushed
the
qiwa/vision-encoder-batcher
branch
from
419f1c0 to
98078c9
Compare
Codex-review round 1 (behaviour-preserving simplifications): - Collapse the startup handshake — a ready Event, a start-error field, and a write-only _terminated Event — into one concurrent.futures.Future that start() blocks on and re-raises from. - Fix a start() thread leak: a retry after a failed thread spawn passed the old `_thread is None` guard and spawned a second, orphaned worker. Guard on state (reject any non-NEW) instead. - Drop the unreachable post-join queue re-drain in shutdown() (admission is already closed and the queue drained under the lock) and inline the now single-use _consume_all helper. - Drop a dead logging import/logger in async_vision_encoder and a redundant buckets=None override (matches the ABC default) in the fake test backend. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…path Codex-review round 2 (behaviour-preserving simplifications): - Remove the _NO_RESULT sentinel: bare _consume() only ever accounts an already-tombstoned item (error set → result write skipped), and result slots default to None, so a real None result is indistinguishable — default result=None and assign whenever no error is set. - Drop the always-true `if batch` guard on the final flush in _dispatch (live is non-empty and every work is appended after any mid-loop flush). - Stop _drain_queue_locked from re-queueing _SHUTDOWN: both callers own the stop signal (shutdown() re-enqueues one; the crash supervisor drains as the sole worker exits), so a seen sentinel is stale — drop it and keep draining. - Build the batcher before the preprocess pool in load(), so a config the batcher ctor rejects fails before any pool is spawned (nothing to reap); update the corresponding test. - Drop the write-only _Recorder.batches field in the batcher tests. Kept _Request.done (not folded into remaining==0): _abort() sets done with remaining>0, so it distinguishes a force-aborted request from a cleanly completed one — an explicit finalized-flag, not redundant state. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Codex-review round 3: - Fix a concurrent double-start race the round-1 guard change opened: `_state` stays NEW while the first start() waits on on_start, so a second start() could pass a state-only guard, spawn a second worker, and race on _started. Guard on both `_thread is not None` and non-NEW state (covers double-start AND failed-spawn retry). - shutdown() now snapshots `_thread` under the lock instead of reading it unlocked, so it can't race a concurrent start()'s publish. - encode(): drop the `errors` list comprehension and the `list(settled)` copy — scan settled in order, raise the first exception, and alias (not copy) the already-returned list. - Strengthen the passthrough test to assert the raws reach forward_batch unchanged and in order (count+shape alone missed substitution/reorder). - Simplify the module header: dedicated worker thread, eager batching by default, optional micro-batching when max_batch_cost is given. Kept _State.FAILED as an explicit terminal state (not folded into CLOSED + _terminal_error): "shut down" and "worker crashed" are distinct conditions and the enum keeps the state machine legible. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The existing test_double_start_raises only covers a sequential second start(). Add a concurrent one: two threads race into start(), asserting exactly one wins and on_start runs once (no second worker). Guards the fix for the race where _state stays NEW while the first start() blocks on on_start, so the guard must also key on _thread. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
Codex round 4: after the concurrent-start guard, only the winning start() ever sets RUNNING, so the post-start "already RUNNING → skip" arm is unreachable — reject non-NEW, then set RUNNING. Likewise submit()'s `if RUNNING: pass` is vacuous — check FAILED, then reject non-RUNNING, else proceed. Behavior unchanged (unit + concurrency stress green). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Codex round 5: the timing-based 40-run version could pass a buggy state-only guard if the first start() reached RUNNING before the second was scheduled. Park the winner inside on_start (via an event) so `_state` is provably still NEW when the loser races in, and assert the loser is rejected while the winner is still parked — forcing the exact double-start window every run. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
furionw
force-pushed
the
qiwa/partial-embed-assembly
branch
from
99e2a74 to
0509d8a
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Based on #10832.
Why
#10832 drives the custom encoder serially — one
encode()call's images at a time on the actor thread, with no cross-request coalescing — so under concurrent load the encoder GPU sits idle between requests. This change swapsAsyncVisionEncoder's internals for aThreadedMicroBatcherthat coalesces images from concurrentencode()calls into batchedforward_batchcalls, while keeping the public surface (load/encode/get_image_placeholder_token_id/shutdown) unchanged — so the worker integration from #10832 doesn't move. Batching is eager by default (drain whatever is queued whenever the worker is free, no timer); passingmax_batch_costinstead caps each batch by a summed per-item cost budget.What Change
ThreadedMicroBatcher(generic, torch-free): one dedicated worker thread runs build / everyfn/ close; eager batching by default, optional cost-bounded micro-batching; exactly-once per-request finalization; a crash supervisor never hangs an awaiter.AsyncVisionEncoderinternals to the batcher; its public surface and the preprocess-atomicity barrier are unchanged.Test Plan
agg_custome2e that splices to "42" under a 30-way burst.pytest components/src/dynamo/vllm/tests/multimodal_utils/🤖 Generated with Claude Code