Skip to content

perf: run native sampling on managed loop - #2

Open
daanishkhazi wants to merge 4 commits into
mainfrom
codex/managed-sampling-loop
Open

perf: run native sampling on managed loop#2
daanishkhazi wants to merge 4 commits into
mainfrom
codex/managed-sampling-loop

Conversation

@daanishkhazi

@daanishkhazi daanishkhazi commented Aug 15, 2026

Copy link
Copy Markdown

Summary

  • expose native sample_with_prompt_tokens through FiretitanSamplingClient
  • run native deployment sampling on the client-owned event-loop thread
  • preserve ContextVars, completion objects, cancellation, and error propagation
  • make close atomic with submission and drain managed-loop work before shutdown

Why

The controlled APEX ablation identified the dedicated sampling loop as the load-bearing intervention. This keeps Fireworks request and response processing off the rollout coordinator loop without adding custom transport or parser behavior.

Scope

This PR is based directly on SDK v1.2.9. It does not change DeploymentSampler parsing, SSE transport, trainer or scheduler behavior, model parameters, loss, retries, routing, or raw-sampler behavior.

Verification

  • deployment and sampling suites: 146 passed
  • Ruff check passed
  • diff check passed
  • sampling.py and _sse.py are byte-identical to v1.2.9

@cubic-dev-ai cubic-dev-ai Bot 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.

No issues found across 3 files

Re-trigger cubic

@daanishkhazi
daanishkhazi force-pushed the codex/managed-sampling-loop branch from f4e2ea7 to 972da36 Compare August 17, 2026 22:12
@daanishkhazi
daanishkhazi changed the base branch from codex/cooperative-sse-processing to main August 17, 2026 22:12
await async_client.aclose()
self.deployment_sampler._sync_client.close()

async def _shutdown_loop(self) -> 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 is reasonable for general hardening, but perhaps a bit complicated for our use case given that the exit stack calls this after the run_training() function has unwound. The modal process will exit shortly after. The important cleanup for us service.close(), which cleans up remote Fireworks resources which is handled separately.

self._loop: asyncio.AbstractEventLoop | None = None
self._loop_thread: threading.Thread | None = None
self._loop_lock = threading.Lock()
self._loop_lock = threading.RLock()

@rohitrastogi rohitrastogi Aug 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

nit: arguably using an RLock here is a smell as it obscures the critical sections. I think the code can be refactored to keep a standard lock so each function has to explicitly acquire the lock to document that part of the function must be atomic.


try:
result = asyncio.run(_run())
assert captured["thread_name"] == "fireworks-sampling-client"

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

is this a redundant assertion to 1717?

"logprobs": True,
"request_marker": marker,
}
assert result == [completion]

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

is the assertion between 1730 and 1731 redundant?

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

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants