feat(accuracy): restore codegen grade concurrency (AIP-1094) - #1237
feat(accuracy): restore codegen grade concurrency (AIP-1094)#1237debermudez wants to merge 18 commits into
Conversation
Try out this PRQuick install: pip install --upgrade --force-reinstall git+https://github.com/ai-dynamo/aiperf.git@98786ac5838e50877ce047d5890591bb5e709fffRecommended with virtual environment (using uv): uv venv --python 3.12 && source .venv/bin/activate
uv pip install --upgrade --force-reinstall git+https://github.com/ai-dynamo/aiperf.git@98786ac5838e50877ce047d5890591bb5e709fffLast updated for commit: |
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughThe codegen worker now batches queued requests and computes ordered per-request metrics. The client supports concurrent grading through request IDs, response demultiplexing, timeout handling, worker recovery, and safe shutdown. Unit and integration tests cover these behaviors. ChangesCodegen concurrency
Estimated code review effort: 4 (Complex) | ~60 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
docs/superpowers/plans/2026-07-29-codegen-grade-concurrency.md (1)
1-931: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winRemove this plan document or move it out of version control.
This file is a step-by-step implementation plan documenting code changes and decisions for AIP-1094. The coding guidelines explicitly prohibit creating Markdown files for this purpose.
Do not commit this plan file to the repository. If the plan is useful during development, keep it outside of
docs/(e.g., in a local scratch file, a PR description, or an issue tracker) instead of committing it. If it must stay underdocs/, confirm it's registered indocs/index.ymlper thedocs/**/*.mdguideline.Based on learnings, "
**/*.md: Use Mermaid diagrams instead of ASCII art in Markdown files. Do not create Markdown files to document code changes or decisions."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@docs/superpowers/plans/2026-07-29-codegen-grade-concurrency.md` around lines 1 - 931, Remove the implementation plan document from version control rather than committing it under docs. If the plan is needed during development, move it to an approved external location or ensure any retained docs entry is registered in docs/index.yml, while avoiding Markdown files used solely to document code changes or decisions.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/aiperf/accuracy/graders/_codegen_worker_client.py`:
- Around line 295-300: Update aclose to acquire _spawn_lock before marking
shutdown, set _closing while holding that lock, and keep the lock through the
worker termination sequence so startup cannot race with shutdown. In
_ensure_worker, check _closing after acquiring _spawn_lock and return without
spawning when shutdown has begun, including when subprocess creation is still
pending.
In `@src/aiperf/accuracy/graders/_codegen_worker.py`:
- Around line 154-213: The _drain_buffered function currently uses
stdin.peek(0), which can block after the first request is consumed. Replace this
with a non-blocking availability check on the underlying descriptor, treating
b"" and BlockingIOError as no additional data; restore blocking mode before
run_worker_loop performs the next blocking readline().
---
Outside diff comments:
In `@docs/superpowers/plans/2026-07-29-codegen-grade-concurrency.md`:
- Around line 1-931: Remove the implementation plan document from version
control rather than committing it under docs. If the plan is needed during
development, move it to an approved external location or ensure any retained
docs entry is registered in docs/index.yml, while avoiding Markdown files used
solely to document code changes or decisions.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 635bfcff-db3a-45e5-a51b-e35879968cb4
📒 Files selected for processing (6)
docs/superpowers/plans/2026-07-29-codegen-grade-concurrency.mdsrc/aiperf/accuracy/graders/_codegen_worker.pysrc/aiperf/accuracy/graders/_codegen_worker_client.pytests/component_integration/test_lcb_codegen_worker_e2e.pytests/unit/accuracy/test_codegen_worker.pytests/unit/accuracy/test_codegen_worker_client.py
8a03966 to
54185dc
Compare
|
Addressing the outside-diff comment on |
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (2)
tests/unit/accuracy/test_codegen_worker_client.py (1)
82-95: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
TestSerializationno longer tests serialization; consider moving this test intoTestConcurrency.The comment at Line 83-84 states this class previously verified serialized execution and now verifies the opposite (concurrent execution without a lock). Keeping the old class name is confusing since a
TestConcurrencyclass already exists immediately below with the same concurrency scope. Movetest_concurrent_grades_return_correct_resultsintoTestConcurrencyand drop the now-emptyTestSerializationclass, or rename the class to reflect its new purpose.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/accuracy/test_codegen_worker_client.py` around lines 82 - 95, Move test_concurrent_grades_return_correct_results from TestSerialization into the existing TestConcurrency class, preserving its assertions and cleanup. Remove the now-empty TestSerialization class and update any obsolete serialization wording associated with the test.src/aiperf/accuracy/graders/_codegen_worker.py (1)
80-158: 🚀 Performance & Scalability | 🔵 Trivial
handle_batchcorrectly isolates per-request failures and preserves order.The malformed-JSON, non-dict, missing-field, batch-exception, and per-item-metric-exception paths all populate
responsesat the original index and are covered by the referenced unit tests (test_malformed_request_in_batch_does_not_affect_others,test_batch_exception_returns_error_for_all,test_response_order_matches_request_order). The blindexcept Exceptionat Line 134 and Line 151 flagged by Ruff is intentional per the docstring's "Never raises" guarantee, so isolated per-request/per-batch failures do not take down the worker loop.One scalability note:
_drain_buffered(called fromrun_worker_loop) can growreqsunboundedly while queued input exists, and no response is written until the entire batch'scodegen_fncall completes. Under a burst of many concurrentgrade_codegencalls, this can inflate per-request latency and peak memory for largeevaluation_sample/generated_codepayloads. Consider capping batch size if this becomes an issue in practice.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/aiperf/accuracy/graders/_codegen_worker.py` around lines 80 - 158, Consider adding a maximum batch-size limit to the _drain_buffered flow used by run_worker_loop so reqs cannot grow without bound while input remains queued. Process buffered requests in capped batches and preserve response ordering and existing handle_batch error isolation.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/aiperf/accuracy/graders/_codegen_worker.py`:
- Around line 161-204: The non-blocking loop in _drain_buffered must not submit
incomplete JSONL fragments. Before calling stdin.readline(), verify the bytes
returned by stdin.peek(0) contain b"\n"; if not, stop draining so blocking mode
is restored and the next run_worker_loop cycle can complete the line.
In `@tests/unit/accuracy/test_codegen_worker_client.py`:
- Around line 155-169: Update test_stale_id_after_timeout_does_not_crash to
issue the documented second grade_codegen call after the expected timeout, using
a real timeout and the existing worker setup. Preserve the initial
near-zero-timeout assertion, and ensure the follow-up request verifies the
worker remains usable without hanging or crashing.
---
Nitpick comments:
In `@src/aiperf/accuracy/graders/_codegen_worker.py`:
- Around line 80-158: Consider adding a maximum batch-size limit to the
_drain_buffered flow used by run_worker_loop so reqs cannot grow without bound
while input remains queued. Process buffered requests in capped batches and
preserve response ordering and existing handle_batch error isolation.
In `@tests/unit/accuracy/test_codegen_worker_client.py`:
- Around line 82-95: Move test_concurrent_grades_return_correct_results from
TestSerialization into the existing TestConcurrency class, preserving its
assertions and cleanup. Remove the now-empty TestSerialization class and update
any obsolete serialization wording associated with the test.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 3fe99340-813b-4bf1-b636-3f7ca8643ed7
📒 Files selected for processing (5)
src/aiperf/accuracy/graders/_codegen_worker.pysrc/aiperf/accuracy/graders/_codegen_worker_client.pytests/component_integration/test_lcb_codegen_worker_e2e.pytests/unit/accuracy/test_codegen_worker.pytests/unit/accuracy/test_codegen_worker_client.py
🚧 Files skipped from review as they are similar to previous changes (3)
- tests/component_integration/test_lcb_codegen_worker_e2e.py
- src/aiperf/accuracy/graders/_codegen_worker_client.py
- tests/unit/accuracy/test_codegen_worker.py
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/aiperf/accuracy/graders/_codegen_worker.py (1)
191-196: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a pipe-backed partial-line test.
Use
io.BufferedReaderoveros.pipe(). Write one JSONL request in two writes and assert that the request completes after the newline arrives. ExistingBytesIOtests do not exercisepeek(0)or the partial-line guard.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/aiperf/accuracy/graders/_codegen_worker.py` around lines 191 - 196, Add a pipe-backed partial-line test for the worker’s JSONL request-reading path, using io.BufferedReader over os.pipe(). Split one request across two writes, verify the first partial write does not complete processing, then write the newline and assert the request completes successfully; keep the test focused on exercising peek(0) and the partial-line guard around the worker read loop.Source: Learnings
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@src/aiperf/accuracy/graders/_codegen_worker.py`:
- Around line 191-196: Add a pipe-backed partial-line test for the worker’s
JSONL request-reading path, using io.BufferedReader over os.pipe(). Split one
request across two writes, verify the first partial write does not complete
processing, then write the newline and assert the request completes
successfully; keep the test focused on exercising peek(0) and the partial-line
guard around the worker read loop.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 2b8b50ac-1a2f-42d9-b2fa-8d3f1d97c797
📒 Files selected for processing (2)
src/aiperf/accuracy/graders/_codegen_worker.pytests/unit/accuracy/test_codegen_worker_client.py
🚧 Files skipped from review as they are similar to previous changes (1)
- tests/unit/accuracy/test_codegen_worker_client.py
2e9f6ca to
b7f4edc
Compare
|
Fixed in b7f4edc — added |
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (7)
tests/unit/accuracy/test_codegen_worker.py (4)
135-147: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReuse the existing batch fixture instead of redefining it.
_nan_infis identical to_fake_codegen_batch_okat lines 32-38. Pass_fake_codegen_batch_okand keep only the custom_nan_compute.♻️ Proposed simplification
def test_non_finite_metric_values_are_dropped(self) -> None: # NaN/Inf must not cross the JSONL boundary (repo NaN/Inf discipline). - def _nan_inf( - samples: list, generations: list, **_kwargs: Any - ) -> tuple[dict[str, Any], dict[int, list]]: - n = len(samples) - return {"pass@1": 1.0}, {i: [[True]] for i in range(n)} - def _nan_compute(results: dict, **_kwargs: Any) -> dict[str, Any]: return {"pass@1": float("nan"), "extra": float("inf"), "ok": 1.0} req = {"id": 9, "evaluation_sample": [{}], "generated_code": [["x"]]} - resps = worker.handle_batch([req], _nan_inf, _nan_compute) + resps = worker.handle_batch([req], _fake_codegen_batch_ok, _nan_compute)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/accuracy/test_codegen_worker.py` around lines 135 - 147, Update test_non_finite_metric_values_are_dropped to remove the duplicate _nan_inf helper and pass the existing _fake_codegen_batch_ok fixture to worker.handle_batch, retaining only the custom _nan_compute behavior.
202-203: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMove
threadingandtimeimports to the file top.The repository test guidelines require imports at the file top.
As per coding guidelines: "keep imports at the top".
♻️ Proposed change
def test_partial_jsonl_line_is_deferred_to_next_cycle(self) -> None: # Exercises the O_NONBLOCK peek(0) + partial-line guard in _drain_buffered. # A partial write (no trailing newline) must not be submitted as a request; # only after the newline arrives should the line be processed. - import threading - import time - req = self._req(42)Add at the file top:
import threading import time🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/accuracy/test_codegen_worker.py` around lines 202 - 203, Move the threading and time imports from their local position near the affected test into the module-level import section at the top of tests/unit/accuracy/test_codegen_worker.py, leaving their usage unchanged.Source: Coding guidelines
241-244: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAnnotate the
codegen_fnparameter.
codegen_fnhas no type hint.TestRunWorkerLoopBatch._runalready usesCallable[..., tuple[dict[str, Any], dict[int, list]]]. Use the same annotation here.As per coding guidelines: "Add type hints to every function parameter and return value".
♻️ Proposed change
- def _run(self, requests: list[bytes], codegen_fn) -> list[dict]: + def _run( + self, + requests: list[bytes], + codegen_fn: Callable[..., tuple[dict[str, Any], dict[int, list]]], + ) -> list[dict[str, Any]]:🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/accuracy/test_codegen_worker.py` around lines 241 - 244, Update the _run method’s codegen_fn parameter annotation to match TestRunWorkerLoopBatch._run: Callable[..., tuple[dict[str, Any], dict[int, list]]]. Preserve the existing requests annotation and list[dict] return annotation.Source: Coding guidelines
210-237: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueGuard the shared
BytesIOagainst cross-thread access.The worker thread writes to
outwhile the main thread callsout.seek(0)andout.read()at lines 224-225.io.BytesIOkeeps one shared file position, so a concurrent write would land at the position set by the main thread. The current ordering is safe because the worker writes only after line 228, but the pattern breaks silently if the timing changes. Consider a small lock-protected writer wrapper, or snapshot without.getvalue()instead ofseek/read.♻️ Proposed change
- assert t.is_alive() - out.seek(0) - assert out.read() == b"" # nothing written yet + assert t.is_alive() + assert out.getvalue() == b"" # nothing written yet @@ assert not t.is_alive() - out.seek(0) - resps = [orjson.loads(ln) for ln in out if ln.strip()] + resps = [ + orjson.loads(ln) for ln in out.getvalue().splitlines() if ln.strip() + ]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/accuracy/test_codegen_worker.py` around lines 210 - 237, Protect the shared BytesIO access in the worker-thread test around the output inspection after starting worker.run_worker_loop. Replace the main thread’s seek/read pattern with a position-independent snapshot such as getvalue(), and use that snapshot for both the empty-output assertion and final response parsing so concurrent writes cannot alter the shared file position..gitignore (1)
60-60: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the duplicate ignore rule.
Line 56 already ignores
.worktrees/.♻️ Proposed change
tests/scripts/.chaos_runs/ -.worktrees/🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In @.gitignore at line 60, Remove the duplicate `.worktrees/` entry from the changed section, keeping the existing ignore rule at line 56 as the sole rule.src/aiperf/accuracy/graders/_codegen_worker_client.py (2)
296-300: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueExtract the pending-future failure loop.
The same loop appears in
_handle_faultat lines 251-254. Extract a small helper so both paths stay consistent.♻️ Proposed change
+ def _fail_pending(self, message: str) -> None: + for fut in list(self._pending.values()): + if not fut.done(): + fut.set_exception(CodegenWorkerError(message)) + self._pending.clear() + async def aclose(self) -> None: - for fut in list(self._pending.values()): - if not fut.done(): - fut.set_exception(CodegenWorkerError("grading worker closed")) - self._pending.clear() + self._fail_pending("grading worker closed") await self._kill()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/aiperf/accuracy/graders/_codegen_worker_client.py` around lines 296 - 300, Extract the pending-future failure loop from the current close path into a small helper near the relevant methods, then call that helper from both the close flow and `_handle_fault`. Preserve the existing behavior of setting CodegenWorkerError("grading worker closed") on unfinished futures and clearing `_pending` exactly once.
218-240: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winFault the worker if the reader raises an unexpected exception.
The
tryblock only handlesValueError,ConnectionError,BrokenPipeError,orjson.JSONDecodeError, andCancelledError. Any other exception ends the reader task silently. Every pending caller then waits for its full timeout, and the task exception is only reported when the task is garbage collected.Add a final
except Exceptionthat calls_handle_fault(). Note thatBrokenPipeErroris a subclass ofConnectionError, so it is redundant in the tuple at line 222.🛠️ Proposed change
except asyncio.CancelledError: pass + except Exception: + # A reader crash would leave every pending caller waiting for its + # timeout, so convert it into a worker fault. + await self._handle_fault()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/aiperf/accuracy/graders/_codegen_worker_client.py` around lines 218 - 240, Update the reader loop in the worker client to catch unexpected exceptions with a final except Exception branch and await _handle_fault() before returning or completing. Remove the redundant BrokenPipeError entry from the existing ConnectionError tuple, while preserving the current CancelledError handling and fault behavior for known read, parse, and dispatch failures.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/aiperf/accuracy/graders/_codegen_worker_client.py`:
- Around line 120-124: Update _ensure_worker to cancel and await the existing
_reader_task and _stderr_task before spawning a replacement whenever the current
process has exited, then clear the old worker task/process state as appropriate
before resetting _worker_proven. Ensure the respawn path cannot let stale
readers invoke _handle_fault against the new worker.
- Around line 193-200: Update the response handling around _pending.pop in the
worker client to detect resp values with id equal to None before treating the
response as stale, and return the protocol-fault result so callers fail fast.
Preserve existing handling for valid pending IDs, cancelled futures, and
unhashable IDs.
- Around line 115-118: Update the CancelledError handling in the request flow to
remove the cancelled request from _pending and re-raise the cancellation without
calling _handle_fault. Preserve worker usability so concurrent requests
continue, relying on _dispatch_response to discard any late response for the
stale request ID.
- Around line 106-108: Update the request-writing flow around the worker
client’s stdin write to explicitly validate that the process and stdin are
available instead of relying on assert, and catch BrokenPipeError and
ConnectionResetError from write or drain. Route these failures through
_handle_fault so the caller receives CodegenWorkerError and the pending
entry/future is cleaned up.
In `@src/aiperf/accuracy/graders/_codegen_worker.py`:
- Around line 113-122: Update the request-processing block around all_samples
and all_generations to read both evaluation_sample and generated_code into local
values before modifying any batch lists. Append to all_samples, all_generations,
and id_map only after both lookups succeed, preserving aligned entries when
malformed requests are caught by the existing exception handler.
- Around line 205-210: Update the stdin-reading fallback around the fd/`fcntl`
selection to run the read-all `stdin.read()` loop only when `fd < 0`, matching
the intended in-memory `BytesIO` path. When a real file descriptor exists but
`fcntl` is unavailable, skip the drain instead of blocking on the pipe, while
preserving the existing nonblocking read behavior when `fcntl` is available.
---
Nitpick comments:
In @.gitignore:
- Line 60: Remove the duplicate `.worktrees/` entry from the changed section,
keeping the existing ignore rule at line 56 as the sole rule.
In `@src/aiperf/accuracy/graders/_codegen_worker_client.py`:
- Around line 296-300: Extract the pending-future failure loop from the current
close path into a small helper near the relevant methods, then call that helper
from both the close flow and `_handle_fault`. Preserve the existing behavior of
setting CodegenWorkerError("grading worker closed") on unfinished futures and
clearing `_pending` exactly once.
- Around line 218-240: Update the reader loop in the worker client to catch
unexpected exceptions with a final except Exception branch and await
_handle_fault() before returning or completing. Remove the redundant
BrokenPipeError entry from the existing ConnectionError tuple, while preserving
the current CancelledError handling and fault behavior for known read, parse,
and dispatch failures.
In `@tests/unit/accuracy/test_codegen_worker.py`:
- Around line 135-147: Update test_non_finite_metric_values_are_dropped to
remove the duplicate _nan_inf helper and pass the existing
_fake_codegen_batch_ok fixture to worker.handle_batch, retaining only the custom
_nan_compute behavior.
- Around line 202-203: Move the threading and time imports from their local
position near the affected test into the module-level import section at the top
of tests/unit/accuracy/test_codegen_worker.py, leaving their usage unchanged.
- Around line 241-244: Update the _run method’s codegen_fn parameter annotation
to match TestRunWorkerLoopBatch._run: Callable[..., tuple[dict[str, Any],
dict[int, list]]]. Preserve the existing requests annotation and list[dict]
return annotation.
- Around line 210-237: Protect the shared BytesIO access in the worker-thread
test around the output inspection after starting worker.run_worker_loop. Replace
the main thread’s seek/read pattern with a position-independent snapshot such as
getvalue(), and use that snapshot for both the empty-output assertion and final
response parsing so concurrent writes cannot alter the shared file position.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 740b1553-a479-40e9-a264-8debc208ec7e
📒 Files selected for processing (6)
.gitignoresrc/aiperf/accuracy/graders/_codegen_worker.pysrc/aiperf/accuracy/graders/_codegen_worker_client.pytests/component_integration/test_lcb_codegen_worker_e2e.pytests/unit/accuracy/test_codegen_worker.pytests/unit/accuracy/test_codegen_worker_client.py
🚧 Files skipped from review as they are similar to previous changes (2)
- tests/component_integration/test_lcb_codegen_worker_e2e.py
- tests/unit/accuracy/test_codegen_worker_client.py
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/aiperf/accuracy/graders/_codegen_worker_client.py (1)
293-296: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winKill the process group after the worker leader exits.
Line 293 skips
_kill_process_group()whenproc.returncodeis already set. The worker can exit before a lighteval sandbox child exits. That child remains in the dedicated process group while_ensure_worker()spawns a replacement.Invoke process-group termination whenever
procexists. Treat an already-gone process group as successful cleanup.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/aiperf/accuracy/graders/_codegen_worker_client.py` around lines 293 - 296, Update the cleanup block around _kill_process_group in _ensure_worker so process-group termination runs whenever proc exists, regardless of proc.returncode. Preserve suppressing ProcessLookupError and awaiting the process as needed, treating an already-gone group as successful cleanup before spawning a replacement.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/aiperf/accuracy/graders/_codegen_worker_client.py`:
- Around line 110-129: Update the request flow around the stdin write and
response wait so the single timeout covers both `proc.stdin.drain()` and
`asyncio.wait_for(fut, ...)`. Establish one deadline before draining, use the
remaining time for the response await, and ensure `asyncio.CancelledError`
removes `req_id` from `_pending` whether cancellation happens during drain or
response handling; preserve timeout fault handling and error propagation.
---
Outside diff comments:
In `@src/aiperf/accuracy/graders/_codegen_worker_client.py`:
- Around line 293-296: Update the cleanup block around _kill_process_group in
_ensure_worker so process-group termination runs whenever proc exists,
regardless of proc.returncode. Preserve suppressing ProcessLookupError and
awaiting the process as needed, treating an already-gone group as successful
cleanup before spawning a replacement.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: fcc96d53-7355-4faa-86c7-d3d3ecdb1235
📒 Files selected for processing (3)
src/aiperf/accuracy/graders/_codegen_worker.pysrc/aiperf/accuracy/graders/_codegen_worker_client.pytests/unit/accuracy/test_codegen_worker_client.py
🚧 Files skipped from review as they are similar to previous changes (2)
- src/aiperf/accuracy/graders/_codegen_worker.py
- tests/unit/accuracy/test_codegen_worker_client.py
|
Addressing the two comments from the latest review: Kill process group after worker exits (lines 293-296): Fixed in 49ab2f5 — Single deadline covering drain + response wait (lines 110-129): Not addressing in this PR. The current split (drain is unbudgeted, |
Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
…ustness - Replace _drain_fd (broken: select() unreliable after BufferedReader pulls kernel data into userspace) and _drain_seekable with _drain_buffered that uses peek() for BufferedReader and a seekable fallback for BytesIO in tests. Remove unused _parse_batch and import select. - aclose() now sets CodegenWorkerError on pending futures instead of calling cancel(), so shutdown does not propagate CancelledError to grader callers that only catch CodegenWorkerError. - Extract _dispatch_response helper so _run_reader complexity stays within the C901 limit; unhashable req_id from a desynced worker now triggers a fault instead of killing the reader task with TypeError. - grade_codegen calls await stdin.drain() after write() to respect backpressure. - Add test_list_shaped_pass_at_1_is_preserved to TestHandleBatch (guards the silent-0.000 bug for list-shaped pass@1 from lighteval). - Remove unused _ECHO_ID_IN_METRICS constant from client tests. Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
…remove plan doc Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
…test Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
…e-backed partial-line drain test Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
…, and batch alignment Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
…h in drain Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
… exited Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
…g-vs-exception race Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
…oid cancel/timeout interaction Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
Python 3.11 asyncio.wait_for silently absorbs CancelledError when the inner future is already resolved at the point cancellation is delivered. _ECHO_OK responds fast enough that this race is hit consistently on 3.11. Replace the proc-up poll with a gated worker that blocks until a "go" file appears, ensuring fut is unresolved when grade.cancel() fires. Release the gate after asserting CancelledError so the subsequent grade succeeds without changing any production behaviour. Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
…nse wait timeout was applied independently to drain() and wait_for(fut), allowing the total to reach 2*timeout. Compute a deadline before drain() and pass the remaining duration (max(0.0, deadline - loop.time())) to each wait_for call so the combined operation stays within timeout. Signed-off-by: Elias Bermudez <dbermudez@nvidia.com>
d197aab to
98786ac
Compare
ajcasagrande
left a comment
ajcasagrande
left a comment
There was a problem hiding this comment.
Review summary
The design here is right — lighteval builds a fresh ProcessPoolExecutor per evaluate_generations call, so collapsing N requests into one codegen_metrics call is exactly the correct lever, and the id→future demux is clean. But there is one release-blocking defect that makes the whole LiveCodeBench benchmark report zeros, plus a dead except branch from the last commit.
Blocking
handle_batch nests the payload one level too deep, so every LCB grade silently returns pass@1 = 0.0 with ok: true. CodeExecutionGrader already sends lighteval's list form ([{...}] / [[code]]); origin/main forwarded that straight through, the PR wraps it again with .append(). lighteval's evaluate_generations then hands check_correctness a list where it expects a dict, run_test dies on sample["input_output"] with TypeError: list indices must be integers or slices, not str, and lighteval's own except Exception: pass swallows it into the [-2] "compile error" sentinel. No error surfaces anywhere.
Verified against the real worker subprocess and real lighteval 0.13.0, same request, same spawn parameters:
[PR #1237] {'id': 1, 'ok': True, 'metrics': {'pass@1': 0.0}}
[origin/main] {'id': 1, 'ok': True, 'metrics': {'pass@1': 1.0}}
Note the shape of the failure: wrong answers still "pass" because their expected value is already 0.0. Only a known-correct solution exposes it — which no concurrent test in this PR uses.
Why CI is green anyway
Three independent gaps, all worth closing regardless of the fix:
_fake_codegen_batch_okonly readslen(samples)and never dereferences an element, soTestHandleBatchpasses identically with and without the bug. I confirmed the unit suite is green on this HEAD and with the fix applied.test_worker_grades_multiple_problems_concurrentlygrades 4 identical, all-correct problems, so every expected value is equal and misalignment is invisible.- Both e2e tests are
@pytest.mark.slow, andrun-unit-tests.ymlruns component_integration with-m 'not ... and not slow', so CI never executes them.
Suggested fix order
- Flatten the batch payload (blocking) — a ~6-line change; I verified it restores 6/6 correct verdicts on distinct mixed-outcome problems.
- Add a distinct-problem concurrent test that would have caught this, and get one real-lighteval batch test into CI.
- Reorder the
exceptclauses so the drain-timeout branch is reachable. - Decide on retrying faulted siblings, or document the measured blast radius.
- Cleanups (assert escaping
aclose(), the no-op respawn test, stale line refs).
What's working well
- Per-problem demux is correct once the nesting is fixed — verified 6/6 on distinct problems with mixed pass/fail.
_next_idstays monotonic across respawns, so a late response from a killed worker can never collide with a new id.- The
_kill()self-task guard correctly avoids the self-await deadlock when the reader detects its own fault. - Draining stderr before unblocking callers is a genuinely subtle ordering fix and the comment earns its place.
- Protocol-fd isolation from forked sandbox children is careful and well justified.
Separate, pre-existing (not this PR)
_ensure_worker always sets AIPERF_CODEGEN_DEATH_FD, which starts the worker's watcher thread — despite the module docstring promising "a fresh, single-threaded interpreter" and the whole point of #1145 being to keep lighteval's fork away from a multithreaded parent. With that thread present, lighteval grades a known-correct solution 0.0 on origin/main too. Identical on both branches, so not a regression from this PR, but it means these e2e tests are red on main as well. Probably deserves its own issue.
Reproduction scripts and logs for every claim above are available on request; each finding was validated against the real worker + lighteval 0.13.0 on CPython 3.12.10.
| "error": f"malformed request: {exc!r}", | ||
| } | ||
| continue | ||
| all_samples.append(sample) |
There was a problem hiding this comment.
Blocking — this makes every LCB grade return pass@1 = 0.0.
sample here is already [{"input_output": ...}] and generation is already [[code]] — CodeExecutionGrader builds them in lighteval's list form (code_execution.py, _build_evaluation_sample returns a 1-element list, generated_code = [[snippet]]). origin/main passed them straight through as samples_list/generations_list.
.append() adds a second level, so all_samples becomes [[{...}]]. lighteval then does:
# lighteval/tasks/tasks/lcb/codegen_metrics.py:572
inputs = [[(generations_list[index], samples_list[index], timeout), index] ...]which hands check_correctness a list where it expects the sample dict, and run_test dies at codegen_metrics.py:365:
TypeError: list indices must be integers or slices, not str
lighteval swallows that in evaluate_generations_by_problem (except Exception: pass), leaving the [-2] compile-error sentinel — so the worker returns ok: true with pass@1: 0.0 and nothing is logged. Same symptom #1145 existed to fix.
Verified with the real worker subprocess, identical request and spawn parameters:
[PR #1237] {'id': 1, 'ok': True, 'metrics': {'pass@1': 0.0}}
[origin/main] {'id': 1, 'ok': True, 'metrics': {'pass@1': 1.0}}
Suggested fix — flatten, and track each request's span so a multi-sample request still demuxes (pairs with the change on the raw_results line below):
# id_map: list[tuple[int, Any, int, int]] # (req_idx, req_id, start, count)
start = len(all_samples)
all_samples.extend(sample)
all_generations.extend(generation)
id_map.append((i, req_id, start, len(all_samples) - start))With this applied I get 6/6 correct verdicts on 6 distinct problems (3 solvable, 3 deliberately wrong) graded in a single batch through the real worker, and tests/unit/accuracy/ stays green.
| else: | ||
| try: | ||
| metrics = compute_metrics_fn( | ||
| {0: raw_results[pos]}, |
There was a problem hiding this comment.
Pairs with the flattening fix above. Once all_samples is flat, pos (the index into id_map) is no longer the index into raw_results if any request ever carries more than one sample — index by the recorded span instead:
for req_idx, req_id, start, count in id_map:
...
metrics = compute_metrics_fn(
{j: raw_results[start + j] for j in range(count)},
k_list=list(_LCB_PASS_AT_K),
)For today's one-sample-per-request grader this reduces to {0: raw_results[start]}, so it costs nothing and stops the indexing from silently desyncing if the grader ever batches problems itself.
Worth saying explicitly: the reconstruction logic here is otherwise correct. evaluate_generations keys its results by input index (codegen_metrics.py:572-588), and rebuilding a single-entry {0: ...} dict per request does yield the right per-problem pass@1 — I verified alignment across a shared batch with mixed pass/fail once the nesting was fixed. The nesting is the only defect in this function.
| try: | ||
| proc.stdin.write(orjson.dumps(req) + b"\n") | ||
| await asyncio.wait_for(proc.stdin.drain(), max(0.0, deadline - loop.time())) | ||
| except (OSError, ConnectionError) as exc: |
There was a problem hiding this comment.
This clause swallows the TimeoutError below it, so the except TimeoutError branch on line 122 is dead code and _handle_fault() never runs on a drain timeout.
TimeoutError is an OSError subclass (PEP 3151), and since 3.11 asyncio.TimeoutError is TimeoutError:
TimeoutError.__mro__ = (TimeoutError, OSError, Exception, BaseException, object)
That matters because the whole point of commit 98786ac58 ("enforce single-request deadline across drain and response wait") was to kill a worker that has stopped consuming stdin. Reproduced with a worker that never reads stdin and a 4 MB payload (past the 64 KiB pipe buffer + 64 KiB high-water mark), timeout=0.5:
raised: CodegenWorkerError('failed to submit grading request: ')
__cause__ type: TimeoutError
worker._proc after the timeout: <Process 4069256> <-- still alive
worker._start_failures: 0
The wedged worker survives and is reused by every subsequent grade, each paying its full timeout before the response-wait branch finally kills it. The message is misleading too — wait_for raises a bare TimeoutError() with no args, so {exc} renders empty and the log reads failed to submit grading request: with nothing after the colon.
except TimeoutError as exc:
...
except OSError as exc: # ConnectionError is already an OSError
...(The second try block at line 129 orders these correctly — it's only this one.)
| f"start_failures={self._start_failures}); killed + respawning next grade" | ||
| + (f"; stderr tail:\n{chr(10).join(tail)}" if tail else "") | ||
| ) | ||
| for fut in list(self._pending.values()): |
There was a problem hiding this comment.
Worth quantifying the trade-off the description mentions, because it's larger than "a timeout faults concurrent siblings" suggests, and it fires on any fault — worker death, EOF, a malformed frame, a BrokenProcessPool inside handle_batch — not just timeouts.
Nothing retries the faulted siblings against the healthy respawn, so each one lands as a _grading_failure (correct=False, unparsed=True) — silent score depression rather than a visible error.
A/B on one identical workload — a worker that misbehaves exactly once (first spawn eats one line and exits, every respawn is healthy), 12 grades on a 5 ms stagger:
origin/main (serialized) : 11/12 graded
PR #1237 : 6/12 graded
To be fair to the PR: all six losses are consistent with the coupling you already documented — this measures its blast radius, it doesn't show an additional undocumented fault. (I specifically tried to show that a grade arriving after the old worker died could be faulted by the previous _handle_fault, and could not: the window between _kill() nulling _proc and the _pending sweep measures ~54 µs, while a replacement spawn costs ~705 µs, so an arriving grade reliably loses that race. Not a finding.)
Suggestion: resubmit faulted-but-unanswered requests once against the respawned worker, or call out the ~6× amplification in the PR/issue so it's a conscious accuracy trade-off.
|
|
||
| async def _run_reader(self) -> None: | ||
| """Read worker responses and resolve the corresponding pending futures by id.""" | ||
| assert self._proc is not None and self._proc.stdout |
There was a problem hiding this comment.
Minor, but this assert can escape aclose().
_kill() sets self._proc = None before awaiting proc.wait(). If the reader task runs for the first time inside that await, the assert fires — and _kill() suppresses only CancelledError, so the AssertionError propagates out of _kill() → aclose() → CodeExecutionGrader.aclose() on the shutdown path.
Measured:
AssertionError escaped 5/5 via _ensure_worker() + aclose() with no intervening yield
AssertionError escaped 0/5 via the public grade path
So it isn't reachable through grade_codegen() in normal operation (drain()/wait_for yield, letting the reader start first) — hence low severity. Two things make it worth fixing anyway: under python -O the assert is stripped and you get AttributeError: 'NoneType' object has no attribute 'stdout' instead, and _kill() suppressing only CancelledError means any reader-task exception escapes teardown.
Capturing the stream at task-creation time removes the assumption entirely:
self._reader_task = asyncio.create_task(self._run_reader(self._proc.stdout))and widening the suppress in _kill() to also cover Exception keeps teardown total.
|
|
||
| def _fake_codegen_boom(*_args: Any, **_kwargs: Any) -> tuple[dict[str, Any], Any]: | ||
| raise RuntimeError("sandbox exploded") | ||
| def _fake_codegen_batch_ok( |
There was a problem hiding this comment.
This fake is why the payload-nesting bug lands green.
It only reads len(samples) and never dereferences an element, so it behaves identically whether samples is [{...}] or [[{...}]] — every TestHandleBatch case passes with and without the defect. I confirmed both directions: tests/unit/accuracy/ is green on this HEAD (516 passed) and green with the flattening fix applied (575 passed).
Cheapest way to make this fake load-bearing is to assert the contract it's standing in for:
def _fake_codegen_batch_ok(samples, generations, **_kwargs):
# lighteval indexes samples_list[i] as the sample dict; catch nesting errors
assert all(isinstance(s, dict) and "input_output" in s for s in samples), samples
assert all(isinstance(g, list) and all(isinstance(c, str) for c in g)
for g in generations), generations
...That one assertion turns the entire existing TestHandleBatch class into a regression test for this bug at zero structural cost.
| *[worker.grade_codegen(sample, code, timeout=240) for _ in range(n)] | ||
| ) | ||
| assert len(results) == n | ||
| assert all(float(r["pass@1"]) == 1.0 for r in results), results |
There was a problem hiding this comment.
This assertion can't detect a per-problem misalignment, which is the main risk the batching change introduces.
All four grades use the same sample and the same correct solution, so every expected value is 1.0. Results can be swapped, duplicated, or misattributed between callers and the assertion still holds — it can only detect "all wrong", never "wrong per problem".
Using distinct problems with different expected verdicts closes that:
PROBLEMS = [ # (io pairs, solution, expected pass@1)
([("1 2\n", "3\n")], "a, b = map(int, input().split())\nprint(a + b)", 1.0),
([("3 4\n", "12\n")], "a, b = map(int, input().split())\nprint(a + b)", 0.0),
([("9 4\n", "5\n")], "a, b = map(int, input().split())\nprint(a - b)", 1.0),
([("5\n", "25\n")], "n = int(input())\nprint(n * n)", 1.0),
]
results = await asyncio.gather(*[w.grade_codegen(s, [[sol]], timeout=240)
for s, sol, _ in problems])
for (_, _, expected), got in zip(problems, results, strict=True):
assert float(got["pass@1"]) == expectedThat exact shape fails on this HEAD (the three correct solutions all score 0.0) and passes with the flattening fix, so it pins the bug directly.
Also worth noting: both tests here are @pytest.mark.slow, and run-unit-tests.yml runs component_integration with -m 'not ... and not slow' — so CI never executes either of them. This batch runs in ~0.2 s locally; dropping the slow marker on at least one real-lighteval batch test (or adding a CI lane for -m slow) would give the change actual coverage.
Minor: the PR description's test plan uses --run-slow, which this repo doesn't define (pytest: error: unrecognized arguments: --run-slow). The working invocation is -m "component_integration and slow".
| finally: | ||
| await w.aclose() | ||
|
|
||
| async def test_ensure_worker_respawns_after_worker_exits( |
There was a problem hiding this comment.
This test has no assertions, and it doesn't exercise the code it names.
The docstring says "We verify by checking that _proc changes between grades" — there's no such check, or any check at all. And rather than calling _ensure_worker(), the body re-implements it inline (async with w._spawn_lock: if ...: await w._kill()), so the respawn branch it claims to cover never runs. Delete _ensure_worker's respawn path and this test still passes.
Driving it through the public API gives it teeth:
await w.grade_codegen(...) # worker #1 serves, then exits
first_pid = w._proc.pid
await w._proc.wait()
metrics = await w.grade_codegen(...) # must transparently respawn
assert w._proc.pid != first_pid
assert metrics == {"pass@1": 1.0}While you're in this class: the Covers line NNN comments throughout TestCoverageGaps are already stale (e.g. "line 107-109" is now 108-110, "lines 316-317" is 336-338, "line 194" is 201-205, "line 254" is 262-265). Naming the behaviour instead of the line number keeps them from rotting. There's also some redundancy worth trimming — test_concurrent_grades_return_correct_results duplicates test_concurrent_grades_all_complete, aclose-with-pending is covered three times, and class _FakeStdin is unused.
Definitive proof + root cause for the blocking findingFollowing up on my review with an end-to-end reproduction through the real Cause: one
|
| level | origin/main |
this PR |
|---|---|---|
| one request, real worker subprocess, identical spawn args | pass@1 = 1.0 |
pass@1 = 0.0 |
| 6 distinct problems in one drained batch | correct mixed verdicts | all 0.0 |
full aiperf profile CLI (below) |
4/6 | 0/6 |
tests/unit/accuracy/ |
pass | pass — catches nothing |
The CLI run. Real lcb-codegeneration benchmark against the mock server's accuracy-oracle mode. The oracle is 6 real LCB problems from the v4_v5 subset, each answered by a lookup-table program mapping every test case's stdin to its expected stdout — so a served-correct row genuinely executes to pass@1 = 1.0. All 6 rows are correct solutions; the mock decides which come back wrong, via --random-seed 42 --accuracy-correct-rate 0.5. That decision is seeded per prompt, so it is order-independent and identical across arms — which makes the mock's own tally an independent oracle.
Only _codegen_worker.py and _codegen_worker_client.py differ between arms. Fresh mock per arm.
================ origin/main ================
mock /accuracy : matched 6, correct 4, incorrect 2, unmatched 0
aiperf CSV : OVERALL,4,6,0,0.6667
CROSS-CHECK : mock served correct 4/6 | aiperf graded correct 4/6 -> MATCH
================ this PR ====================
mock /accuracy : matched 6, correct 4, incorrect 2, unmatched 0
aiperf CSV : OVERALL,0,6,0,0.0000
CROSS-CHECK : mock served correct 4/6 | aiperf graded correct 0/6 -> MISMATCH
(4 known-correct answers graded wrong, unparsed=0)
origin/main agrees with the oracle exactly. This branch disagrees by precisely the 4 correct answers, silently.
Fix
Flatten instead of nesting, and track each request's span so the per-problem demux still holds:
id_map: list[tuple[int, Any, int, int]] = [] # (req_idx, req_id, start, count)
...
start = len(all_samples)
all_samples.extend(sample)
all_generations.extend(generation)
id_map.append((i, req_id, start, len(all_samples) - start))
...
for req_idx, req_id, start, count in id_map:
metrics = compute_metrics_fn(
{j: raw_results[start + j] for j in range(count)},
k_list=list(_LCB_PASS_AT_K),
)With this applied I get 6/6 correct verdicts on 6 distinct mixed-outcome problems in a shared batch, and tests/unit/accuracy/ stays green. The rest of the design is sound — batching into one codegen_metrics call is the right lever (lighteval builds a fresh ProcessPoolExecutor per call), and the raw_results indexing is otherwise correct.
Correction to my earlier comment
In the review summary I said the pre-existing AIPERF_CODEGEN_DEATH_FD problem was caused by the watcher thread making the worker multithreaded. That was wrong — a daemon thread blocked in os.read is harmless here. A 2×2×2 bisect over {stdout-guard, at-fork close, thread} shows only guard + atfork fails:
CASE=guard pass@1=1.0
CASE=atfork pass@1=1.0
CASE=thread pass@1=1.0
CASE=guard+atfork pass@1=0.0 raw={0: [[-1]]} <-- culprit
CASE=guard+atfork+thread pass@1=0.0 raw={0: [[-1]]}
_install_stdout_guard and _start_death_watcher each register an os.register_at_fork(after_in_child=close(fd)). Both re-fire at every lighteval fork (ProcessPoolExecutor worker → multiprocessing.Manager() → mp.Process), closing fd numbers that multiprocessing may since have recycled for its own pipes. [[-1]] is lighteval's "result list empty" path — the forked child died.
This reproduces identically on origin/main, so it is pre-existing and out of scope for this PR (probably deserves its own issue). It does mean both e2e tests in test_lcb_codegen_worker_e2e.py are red on main too — they're slow-marked, so CI never runs them. I disabled that path identically in both arms above, which is why origin/main scores 66.67% rather than 0%.
Pre-existing: the worker grades every LCB problem 0.0 on
|
2 participants
Summary
asyncio.LockinCodegenGradingWorkerwith anid → asyncio.Futuredemux table and a persistent reader task, allowing N concurrentgrade_codegen()calls to run without blocking each othercodegen_metricsonce with the full batch, so lighteval'sProcessPoolExecutor(max_workers=8)processes multiple problems in parallel — N concurrent grades complete in ~max(individual) instead of ~sumChanges
_codegen_worker.pyhandle_batch(reqs, codegen_fn, compute_metrics_fn)— grades N requests in onecodegen_metricscall, per-problem metrics demuxed fromevaluate_generations'results: dict[int, list]run_worker_loopwith batch-drain version: blocking read of first request, thenpeek(0)-based drain of buffered requests, then onecodegen_metricscall per cyclehandle_request(dead code)_codegen_worker_client.pyasyncio.Lockon the grade path; added_spawn_lock(spawn-only),_pending: dict[int, Future],_reader_task_run_reader()— persistent task that reads stdout and resolves futures by id; stale ids (already timed out) are silently skipped_mark_proven(),_dispatch_response()helpers_handle_faultis idempotent (_proc is Noneguard); cancels all pending futures before killing_kill()cancels and awaits_reader_taskwith self-task guardaclose()usesset_exception(notcancel) so callers getCodegenWorkerError, notCancelledError_request()Known tradeoff
A timeout on any in-flight grade kills the worker and faults all concurrent sibling futures with
CodegenWorkerError. This is the accepted batch-coupling tradeoff documented in the issue.Test plan
uv run pytest tests/unit/accuracy/ -v— newTestHandleBatch,TestRunWorkerLoopBatch,TestConcurrencyclasses all greenuv run pytest tests/component_integration/test_lcb_codegen_worker_e2e.py -v -s --run-slow— both single and concurrent e2e tests pass withpass@1 == 1.0🤖 Generated with Claude Code
Summary by CodeRabbit
Performance
Reliability
Validation