train_iter: enable parallel run

[rbx]

python train_iter.py <...> --parallel 8

[coderabbitai]

Warning

Rate limit exceeded

@rbx has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 3 minutes and 51 seconds before requesting another review.

⌛ How to resolve this issue?

After the wait time has elapsed, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout.

Please see our FAQ for further information.

📝 Walkthrough

Walkthrough

Reintroduces torchinfo and constrains setuptools to >=78.1.1,<82 in requirements.txt. Refactors train_iter.py to separate command construction (build_command) and add run_all_parallel plus a --parallel CLI option to orchestrate concurrent training subprocesses.

Changes

Cohort / File(s)	Summary
Dependency Management requirements.txt	Readded torchinfo and added a setuptools constraint >=78.1.1,<82 to avoid incompatibilities with newer setuptools releases.
Training Execution Refactoring train_iter.py	Replaced run() with build_command() (constructs train.py command). Added run_all_parallel() to schedule and manage concurrent subprocesses with a max-parallel limit and polling wait loop. Added --parallel CLI option and updated main flow to use the orchestrator.

Sequence Diagram(s)

sequenceDiagram
    participant Main
    participant Orchestrator as run_all_parallel()
    participant Builder as build_command()
    participant Worker as Subprocess
    participant Poller as SlotMonitor

    Main->>Orchestrator: combinations + max_parallel
    loop schedule until all combos launched
        Orchestrator->>Builder: build command for combo
        Builder-->>Orchestrator: command string
        Orchestrator->>Worker: spawn subprocess (Popen)
    end
    loop poll for slot availability
        Poller->>Worker: check process status (poll)
        Worker-->>Poller: running/finished
        Poller-->>Orchestrator: free slot / report completion
        Orchestrator->>Orchestrator: launch next job if slots free
    end
    Orchestrator-->>Main: all jobs completed

Loading

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 0.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title 'train_iter: enable parallel run' directly and clearly summarizes the main change: adding parallel execution capability to train_iter.py.
Description check	✅ Passed	The description provides a concrete usage example of the new --parallel feature, demonstrating how to use the added functionality.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 2

🧹 Nitpick comments (1)

train_iter.py (1)

134-172: No subprocess cleanup on KeyboardInterrupt — child processes will be orphaned.

If the user interrupts with Ctrl+C, the spawned Popen children keep running. Wrapping the orchestration in a try/except KeyboardInterrupt (or finally) block that terminates or kills active processes would prevent orphaned training runs consuming resources.

♻️ Suggested approach

 def run_all_parallel(combinations, max_parallel, iter_limit_per_step, session, prices,
                      job_durations, jobs, hourly_jobs, plot_dashboard, dashboard_hours,
                      carry_over_state, seed):
     active = []  # list of (proc, label)
     current_env = os.environ.copy()
 
-    for combo in combinations:
-        efficiency_weight, price_weight, idle_weight, job_age_weight, drop_weight = combo
-        label = f"efficiency={efficiency_weight}, price={price_weight}, idle={idle_weight}, job_age={job_age_weight}, drop={drop_weight}"
-
-        # Wait until a slot is free
-        while len(active) >= max_parallel:
-            still_running = []
-            for proc, lbl in active:
-                if proc.poll() is None:
-                    still_running.append((proc, lbl))
-                else:
-                    rc = proc.returncode
-                    status = "done" if rc == 0 else f"error (rc={rc})"
-                    print(f"[run] {status}: {lbl}")
-            active = still_running
-            if len(active) >= max_parallel:
-                time.sleep(1)
-
-        command = build_command(
-            efficiency_weight, price_weight, idle_weight, job_age_weight, drop_weight,
-            iter_limit_per_step, session, prices, job_durations, jobs, hourly_jobs,
-            plot_dashboard, dashboard_hours, carry_over_state, seed,
-        )
-        print(f"[run] starting: {label}")
-        proc = subprocess.Popen(command, env=current_env)
-        active.append((proc, label))
-
-    # Wait for all remaining processes
-    for proc, label in active:
-        proc.wait()
-        rc = proc.returncode
-        status = "done" if rc == 0 else f"error (rc={rc})"
-        print(f"[run] {status}: {label}")
+    try:
+        for combo in combinations:
+            efficiency_weight, price_weight, idle_weight, job_age_weight, drop_weight = combo
+            label = f"efficiency={efficiency_weight}, price={price_weight}, idle={idle_weight}, job_age={job_age_weight}, drop={drop_weight}"
+
+            # Wait until a slot is free
+            while len(active) >= max_parallel:
+                still_running = []
+                for proc, lbl in active:
+                    if proc.poll() is None:
+                        still_running.append((proc, lbl))
+                    else:
+                        rc = proc.returncode
+                        status = "done" if rc == 0 else f"error (rc={rc})"
+                        print(f"[run] {status}: {lbl}")
+                active = still_running
+                if len(active) >= max_parallel:
+                    time.sleep(1)
+
+            command = build_command(
+                efficiency_weight, price_weight, idle_weight, job_age_weight, drop_weight,
+                iter_limit_per_step, session, prices, job_durations, jobs, hourly_jobs,
+                plot_dashboard, dashboard_hours, carry_over_state, seed,
+            )
+            print(f"[run] starting: {label}")
+            proc = subprocess.Popen(command, env=current_env)
+            active.append((proc, label))
+
+        # Wait for all remaining processes
+        for proc, label in active:
+            proc.wait()
+            rc = proc.returncode
+            status = "done" if rc == 0 else f"error (rc={rc})"
+            print(f"[run] {status}: {label}")
+    except KeyboardInterrupt:
+        print("\n[run] interrupted — terminating active processes...")
+        for proc, label in active:
+            proc.terminate()
+        for proc, label in active:
+            proc.wait()
+        sys.exit(1)

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@train_iter.py` around lines 134 - 172, Wrap the orchestration in
run_all_parallel with a try/finally (or try/except KeyboardInterrupt then
re-raise) so that on interruption you iterate the active list and clean up
subprocesses: for each (proc, label) call proc.terminate(), wait with a short
timeout, and if still running call proc.kill(), then proc.wait() to reap them;
log termination attempts using the existing label variable and ensure the final
wait loop still runs to capture return codes. Use the existing active list and
proc variables and perform cleanup before exiting the function so no child Popen
processes are orphaned.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@requirements.txt`:
- Around line 8-9: The requirements currently pin setuptools as "setuptools<70",
which prevents security fixes and will break once pkg_resources is removed;
update this constraint: either upgrade upstream deps (e.g., TensorBoard) to
versions that use importlib.metadata, or as an interim change replace the
"setuptools<70" constraint with a safe range such as "setuptools>=78.1.1,<82" to
include fixes for GHSA-cx63-2mw6-8hw5 and GHSA-5rjg-fvgr-3xxf while preserving
pkg_resources until dependencies are migrated.

In `@train_iter.py`:
- Line 206: The parser currently accepts non-positive values for --parallel
which makes max_parallel (used in the loop that checks len(active) >=
max_parallel) be ≤ 0 and cause an infinite spin; add a post-parse validation
that enforces args.parallel >= 1 (e.g. after parser.parse_args()) and call
parser.error("--parallel must be at least 1") if violated, or coerce
max_parallel = max(1, args.parallel) before the loop that uses max_parallel
(referencing parser.add_argument, args.parallel, max_parallel and the active
list check len(active) >= max_parallel).

---

Nitpick comments:
In `@train_iter.py`:
- Around line 134-172: Wrap the orchestration in run_all_parallel with a
try/finally (or try/except KeyboardInterrupt then re-raise) so that on
interruption you iterate the active list and clean up subprocesses: for each
(proc, label) call proc.terminate(), wait with a short timeout, and if still
running call proc.kill(), then proc.wait() to reap them; log termination
attempts using the existing label variable and ensure the final wait loop still
runs to capture return codes. Use the existing active list and proc variables
and perform cleanup before exiting the function so no child Popen processes are
orphaned.

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (1)

train_iter.py (1)

134-172: Add KeyboardInterrupt cleanup to avoid orphaned training processes.

When Ctrl+C interrupts train_iter.py (or when SIGINT is delivered only to the parent, e.g., in CI or nohup contexts), the time.sleep(1) on line 156 immediately raises KeyboardInterrupt, the function unwinds, and every train.py process in active is left running. When using the subprocess module, the caller is responsible for handling SIGINT and SIGQUIT signals separately. A try/finally block cleanly terminates all active subprocesses regardless of how the parent exits.

♻️ Proposed fix — wrap body in `try/finally`

 def run_all_parallel(combinations, max_parallel, iter_limit_per_step, session, prices,
                      job_durations, jobs, hourly_jobs, plot_dashboard, dashboard_hours,
                      carry_over_state, seed):
     active = []  # list of (proc, label)
     current_env = os.environ.copy()
 
-    for combo in combinations:
-        ...
-
-    # Wait for all remaining processes
-    for proc, label in active:
-        proc.wait()
-        rc = proc.returncode
-        status = "done" if rc == 0 else f"error (rc={rc})"
-        print(f"[run] {status}: {label}")
+    try:
+        for combo in combinations:
+            efficiency_weight, price_weight, idle_weight, job_age_weight, drop_weight = combo
+            label = f"efficiency={efficiency_weight}, price={price_weight}, idle={idle_weight}, job_age={job_age_weight}, drop={drop_weight}"
+
+            # Wait until a slot is free
+            while len(active) >= max_parallel:
+                still_running = []
+                for proc, lbl in active:
+                    if proc.poll() is None:
+                        still_running.append((proc, lbl))
+                    else:
+                        rc = proc.returncode
+                        status = "done" if rc == 0 else f"error (rc={rc})"
+                        print(f"[run] {status}: {lbl}")
+                active = still_running
+                if len(active) >= max_parallel:
+                    time.sleep(1)
+
+            command = build_command(
+                efficiency_weight, price_weight, idle_weight, job_age_weight, drop_weight,
+                iter_limit_per_step, session, prices, job_durations, jobs, hourly_jobs,
+                plot_dashboard, dashboard_hours, carry_over_state, seed,
+            )
+            print(f"[run] starting: {label}")
+            proc = subprocess.Popen(command, env=current_env)
+            active.append((proc, label))
+
+        # Wait for all remaining processes
+        for proc, label in active:
+            proc.wait()
+            rc = proc.returncode
+            status = "done" if rc == 0 else f"error (rc={rc})"
+            print(f"[run] {status}: {label}")
+    except KeyboardInterrupt:
+        print("\n[run] Interrupted — terminating active processes...")
+        for proc, lbl in active:
+            proc.terminate()
+        for proc, lbl in active:
+            proc.wait()
+        raise

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@train_iter.py` around lines 134 - 172, The run_all_parallel function can
leave child train.py processes orphaned on KeyboardInterrupt; wrap the main loop
that starts and monitors subprocesses in a try/finally so that cleanup always
runs, and in the finally iterate the active list to terminate and, if necessary
after a short wait, kill any remaining subprocesses (call proc.terminate(), wait
with timeout, then proc.kill() if still exited), also call proc.wait() to reap
them; reference the active list and subprocess.Popen instances created in
run_all_parallel to perform this cleanup so no children remain if the parent is
interrupted.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@train_iter.py`:
- Around line 167-172: run_all_parallel currently waits on subprocesses (loop
over active) and only prints "error (rc=N)" but returns None, so main() never
sees failures; update run_all_parallel to track failures by counting non-zero
return codes (increment a failure_count when proc.returncode != 0) and return
that count (or a boolean/rc) instead of None, and ensure the polling section
that checks running subprocesses also increments the same failure_count when a
process exits with non-zero; then modify main() to capture the value returned by
run_all_parallel and call sys.exit(failure_count) (or sys.exit(1) if any
failures) so the script exits non-zero on any subprocess failure.

---

Duplicate comments:
In `@train_iter.py`:
- Around line 206-213: The earlier infinite-loop risk from an invalid parallel
value is resolved by the new validation: ensure the
parser.add_argument("--parallel", ...) and the guard if args.parallel < 1:
parser.error("--parallel must be at least 1") are kept (or restored) so that
invalid values never reach the while len(active) >= max_parallel loop; no
further code change is required other than keeping these two lines intact.

---

Nitpick comments:
In `@train_iter.py`:
- Around line 134-172: The run_all_parallel function can leave child train.py
processes orphaned on KeyboardInterrupt; wrap the main loop that starts and
monitors subprocesses in a try/finally so that cleanup always runs, and in the
finally iterate the active list to terminate and, if necessary after a short
wait, kill any remaining subprocesses (call proc.terminate(), wait with timeout,
then proc.kill() if still exited), also call proc.wait() to reap them; reference
the active list and subprocess.Popen instances created in run_all_parallel to
perform this cleanup so no children remain if the parent is interrupted.