Distributed SWE bench by sharding SWE tasks across cpus for faster runtime - #458
Open
leopck wants to merge 9 commits into
Open
Distributed SWE bench by sharding SWE tasks across cpus for faster runtime#458leopck wants to merge 9 commits into
leopck wants to merge 9 commits into
Conversation
|
MLCommons CLA bot All contributors have signed the MLCommons CLA ✍️ ✅ |
github-actions
Bot
added
the
size/very-large
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #458 +/- ##
=======================================
Coverage ? 81.14%
=======================================
Files ? 160
Lines ? 21582
Branches ? 0
=======================================
Hits ? 17512
Misses ? 4070
Partials ? 0 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Collaborator
|
I was using a system with 64GiB host memory for accuracy testing running 30 workers and didn't run into host memory issue. Also didn't recall seeing multi-hundred GiB memory usage with 200-worker runs. Could you double-check if this is really needed? |
Collaborator
Author
@tianmu-li I think the maximum that I'm able to run this on is around 20 workers and I have around 1TB of host RAM, and the containers I did manually observe the host RAM increasing to 90GB at some points. Following this, this PR would help us to increase to 200 workers so that we can run SWE bench faster, which is the whole purpose of this PR, because even in your case, you are only limited to 30 workers am I right? With 200 workers we should be able to complete this SWE bench on a faster runtime. Open to discuss more if this is the right direction. |
…ork queue Adds the two foundations of the distributed SWE-bench harness: - units.py: shards an instance-id list into immutable, content-addressed units. The sha256 digest covers the ordered id list, so a plan cannot be silently reused across a different run, instance list, or ordering. - queue.py: a filesystem work queue whose claim is a bare os.mkdir (never makedirs(exist_ok=True), which hands a unit to every caller). available() is plan - claims - results, so deleting a result alone does NOT requeue a unit; requeue() is the only supported path and removes the result, the claim and the attempt records together. Env faults are ledgered separately from counted attempts, and abandoning a unit publishes a terminal result AND releases the claim so claims/ and results/ never disagree.
merge_run(wq, run_id) refuses to emit an accuracy number unless every planned unit has a terminal result, none is abandoned, every unit accounts for exactly its planned instance IDS (a set comparison, never a count), the union equals the plan with no cross-shard duplicates, every plan_digest matches, and no unit carries an infra error. Refusal is a structured MergeRefusal naming the offending units and ids; there is no force flag and no partial-credit path. There is deliberately no --all: merge_run takes a required run id and treats a foreign run id or digest as a hard error, not a skip. verify_inventory() cross-checks claims, results and the id-union as independent producers, so a blind spot shared by one instrument cannot certify itself.
reaper.py releases a stale claim only when it has no result, its heartbeat is past stale_after, AND its owner is provably gone. Liveness is a pluggable protocol: LocalProcessLiveness pairs pid with boot id so a recycled pid on a rebooted host is not read as a live owner, and SlurmStepLiveness treats a step missing from scontrol inside a live job as dead, because the job-level rule alone deadlocks the queue forever. An indeterminate probe releases NOTHING - a false reap creates two owners, duplicate results and a wrong denominator. guards.py kills a runaway graded test only under a full conjunction (RSS over threshold AND cwd inside the testbed AND a container-supervisor ancestor). Kills are by PID and refuse self and any ancestor of self; there is no pattern-kill path in the module at all, and a test greps the source to keep it that way. Each term reports its evidence count, and HealthVerdict.combine returns INDETERMINATE rather than UNHEALTHY when a term has zero evidence, so a conjunctive guard cannot collapse into its weakest clause.
The Pyxis sentinel only covers the agent phase; eval-phase error_ids were counted as real outcomes and never retried, which is what produced 24 of 25 permanently-bad runs on the source cluster. classify.py reads the SWE-bench report's error_ids and each instance's run_instance.log and classifies them through an ORDERED rule list, first match wins. The order is load-bearing: BuildImageError is checked before everything because its message embeds the other rules' needles, CONMON_EAGAIN and TEST_TIMEOUT precede WEDGE_EVAL, and PATCH_APPLY_FAILED is last. Anything unclassifiable is UNKNOWN and UNKNOWN is GENUINE, asserted by a membership test: a false bad-run costs one redo, a false retry biases the measurement toward optimism. Memory-kill markers are consumed by phase - an eval-phase kill is a genuine failure (an unbounded allocation is a failing patch), an agent-phase kill is recorded for audit only, since the agent merely gets an error observation and the instance still reaches a real outcome.
run_gates() calls assert_scale() before check() and treats GateScaleError as a gate FAILURE, never a skip. This is the code-level form of the most expensive lesson available: a tool-call gate that exercised the right operation at a 278-token prompt passed, while prompts over 2k tokens silently returned empty, and the run scored 0/80. - CheckpointIdentityGate probes /get_model_info then /v1/models and compares the served model path with == , never startswith or in: the bf16 path is a strict prefix of the fp8 path, so any substring test passes an FP8 engine as bf16. Unidentifiable or ambiguous endpoints fail closed. - ToolCallGate requires a well-formed bash tool call at a prompt of at least min_prompt_tokens measured with the server's own /tokenize, not estimated from characters. No tokenizer means the gate cannot prove its scale, so it fails. - EndpointFingerprintGate records a per-endpoint identity the dispatcher re-checks at publish time, so an engine restarted under a live client cannot yield a 0%-accuracy run that still exits rc=0.
EndpointFingerprintGate hashed the whole /v1/models payload. vLLM stamps that response with a request-time `created` field and mints a fresh `permission[].id` on every call, so two reads of one healthy, untouched engine produce two different fingerprints -- four calls, four values. The dispatcher records a fingerprint when a unit is claimed and re-reads it when the unit is published, and treats any difference as `endpoint_changed`: an infrastructure fault, which requeues the unit. With an unstable fingerprint that comparison is always true, so every unit is retried until it exhausts max_attempts, is published as abandoned, and the merge gate refuses the run. The failure costs the full agent and evaluation time of every attempt first, and reports itself as infrastructure damage rather than as a bug here. Hash only the identity-bearing fields by dropping the per-request ones. The gate still fails closed on an endpoint whose identity cannot be read at all, which is the property it exists to provide.
… fleet
Wires the pieces into a scorer registered as eval_method: swe_bench_fleet.
It is a scheduler in front of the existing SWE-bench service protocol, not a
new runtime: a unit is one RunRequest over a shard, so exact instance
binding, per-instance containers, artifact allow-listing and cancellation
are reused rather than reimplemented.
preflight() runs the gates against the inference endpoints and /health
against every service, raising SetupError before a single instance is
dispatched. score() plans, dispatches with one in-flight run per service,
classifies every unit, requeues any unit with infra_error_count > 0 even
when the service reported succeeded, and takes the accuracy number only
past the merge gate - self.complete comes from the gate, never a count
heuristic.
Stall quarantine verifies effect rather than status: a service that is
/health-OK but has completed no unit within stall_timeout_s is quarantined
and its in-flight unit requeued.
Also adds scripts/swe_bench_wq.py {status,merge,requeue,reap} for
operators. reap is dry-run by default and requeue prints exactly which
result, claim and attempt records it removed, because the failure mode in
the field was an operator believing a delete had requeued something.
score() reads the run's settings back from the report directory's config.yaml,
which yaml.safe_load() returns as plain dictionaries. It then handed that
mapping to SWEBenchScorer._generation_params(), which calls .model_dump() on
it, so the fleet scorer raised
AttributeError: 'dict' object has no attribute 'model_dump'
on every run, after the plan and the work queue had been written but before a
single unit was dispatched.
Re-validate the mapping into ModelParams instead of re-implementing the field
selection here, so the fleet path and the single-service path stay in
agreement about which generation settings are forwarded to the service.
Every unit was submitted with endpoint_urls[:1], so a fleet configured with N engines sent all of its work to the first one and left the other N-1 idle. The comment justified this by noting that the service accepts exactly one endpoint per run and that the fleet's parallelism comes from running many units -- true, but it does not follow that every unit must pick the same one. Two consequences. The obvious one is a throughput ceiling: concurrency is bounded by one engine no matter how much hardware the run was given. The serious one is a measurement hazard -- a single engine's behaviour decides the whole run's accuracy, so one degraded engine is indistinguishable from a degraded model, which is precisely the confusion the endpoint fingerprint exists to prevent. Bind unit -> endpoint by shard index instead. The mapping is deterministic, so a retried unit lands on the endpoint it was originally measured against and stays comparable to its first attempt, and a run with one endpoint behaves exactly as before.
leopck
force-pushed
the
swe-dist-6-fleet-scorer
branch
from
August 20, 2026 19:04
7a8e9a7 to
854f854
Compare
Collaborator
|
@leopck From my observation, 1TiB ram can run 200 concurrent workers with plenty of headroom. I'm not sure why you face RAM pressure unless you are using ram-backed storage, though even then 1TiB is sufficient for all images + running state +ram requirement. |
Collaborator
Author
|
@tianmu-li which runtime engine are you using? I'm assuming Docker? we are using root-less podman and pyxis runtime, we might be seeing different behaviours on our end |
Collaborator
|
@leopck I'm using docker. Currently redoing a run with 30 workers and barely seeing memory usage over 10GiB. Even when adding storage used it's less than 100GiB. I'll be surprised if podman/pyxis vs docker makes such a large difference. |
Collaborator
Author
@tianmu-li let me do a thorough check on the rootless podman where I am observing this memory pressure issue. Putting the memory issue aside, we still need this for pyxis environment as we are limited to 16 workers due to slurm limitation on this. And on memory limited machines, this distributed SWE is still helpful to increase the worker count. |
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
3 participants
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.
What does this PR do?
Depends on #452, #453, #454, #456 to be merged first
Distributed SWE shards the tasks across multiple CPUs into one server, this is to tackle the limitation on SWE bench itself. Namely:
This PR tackles these problems by distributing the containers across multiple CPU nodes, this reduces the memory pressure applied to the host memory and enabling the containers to spawn more reliably. As this depends on the number of CPU nodes you have available, this can greatly increase your worker count enabling extremely fast runtime (3x ~ 6x time saving).
This PR has been tested on both x86 (Podman/Docker) and ARM (Pyxis):
Type of change