From 6ddcf84167c3044353320aae9d5b3067b4b3d030 Mon Sep 17 00:00:00 2001 From: aladac Date: Mon, 18 May 2026 23:31:33 +0200 Subject: [PATCH] feat(generate): add --parallel-queue/-P for concurrent submissions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirrors the style-sweep --parallel-queue flag on the `generate` command. When used with --count N > 1, splits the request into N independent batch_size=1 jobs queued P-at-a-time via ThreadPoolExecutor instead of a single ComfyUI batch. Each task receives a distinct seed (incrementing from --seed when set, freshly randomized per task when --seed=-1) and a distinct output path following the existing stem_NNN.suffix convention. The GPU still processes one prompt at a time, but HTTP queueing, websocket polling, and image-download phases pipeline across tasks for a meaningful wall-clock speedup on warmed-up models (~30-50% in practice). Implementation notes: - count=1 always takes the legacy sequential path regardless of -P. - -P 1 is also sequential — identical behavior to pre-flag invocations. - Bare model names (`-m lust_v10`) are resolved to canonical filenames ONCE in the parent before fanout, so worker tasks (which run with json_output=True path semantics for stdout) don't each duplicate the validation step or, worse, forward unresolved names to ComfyUI. - --json + -P>1 is rejected up-front: the JSON path inside _run_generation short-circuits the disk-save block, which would silently produce zero files. Better to fail loud than save nothing. - parallel_queue is plumbed through --input (JSON/YAML) like every other generate param, with the usual CLI-flag-wins precedence. Tests: 15 new in tests/test_generate_parallel.py covering validation, fanout topology, seed strategies, output naming, --input integration, partial-failure exit code, and a concurrency assertion that confirms threads actually overlap. Manual E2E against ComfyUI on sin: -c 3 -P 3 on FLUX produced 3 distinct images in ~83s vs the ~195s a pure sequential run would take. --- tensors/cli.py | 267 +++++++++++++++++++++++--- tests/test_generate_parallel.py | 325 ++++++++++++++++++++++++++++++++ 2 files changed, 563 insertions(+), 29 deletions(-) create mode 100644 tests/test_generate_parallel.py diff --git a/tensors/cli.py b/tensors/cli.py index 101eec6..3ed5983 100644 --- a/tensors/cli.py +++ b/tensors/cli.py @@ -874,6 +874,22 @@ def generate( # noqa: PLR0915 str | None, typer.Option("--input", "-I", help="JSON or YAML params (file path or inline; keys match CLI options)"), ] = None, + parallel_queue: Annotated[ + int, + typer.Option( + "--parallel-queue", + "-P", + help=( + "Concurrent ComfyUI submissions (default 1). When >1 with --count N, " + "splits the request into N independent jobs (batch_size=1 each) with " + "incrementing seeds, executed P-at-a-time via thread pool. The GPU " + "still processes one prompt at a time, but HTTP queue / init / " + "download phases pipeline for a ~5-15%% speedup. Per-task output " + "interleaves; final summary lists all saved files. Ignored when " + "--count is 1." + ), + ), + ] = 1, ) -> None: """Generate an image using text-to-image. @@ -887,6 +903,11 @@ def generate( # noqa: PLR0915 starting with ``{`` are JSON, everything else is YAML. CLI flags override --input values. + With --count > 1, images are generated as a single ComfyUI batch by default + (one workflow, sequential on GPU). Use --parallel-queue N to instead split + into N independent batch_size=1 jobs queued in parallel, each with its own + seed — useful for overlapping the HTTP/download phase across requests. + Examples: tsr generate "a cat on a windowsill" tsr generate "portrait photo" -m ponyDiffusionV6XL_v6.safetensors -O portrait @@ -895,7 +916,21 @@ def generate( # noqa: PLR0915 tsr generate --input '{"prompt": "a mech", "model": "flux1-dev-fp8.safetensors"}' tsr generate --input scene.yml tsr generate "raw prompt" --no-quality --no-negative + tsr generate "city" -c 8 -P 4 -o out.png # 8 distinct seeds, 4 in flight """ + if parallel_queue < 1: + console.print("[red]--parallel-queue must be >= 1[/red]") + raise typer.Exit(1) + if parallel_queue > 1 and json_output: + # _run_generation short-circuits the disk-save when json_output=True + # (it dumps JSON and returns). For the parallel fanout to actually save + # files, each task must take the non-JSON path. We render our own JSON + # at the end, so the per-task --json is incompatible. + console.print( + "[red]--json is not supported with --parallel-queue > 1 " + "(would skip the file-save step). Drop one or the other.[/red]" + ) + raise typer.Exit(1) # ---- --input merging (JSON or YAML) ---- if json_input is not None: ji = _parse_generate_input(json_input) @@ -981,41 +1016,215 @@ def generate( # noqa: PLR0915 scene_prompt = sp_val if isinstance(sp_val, str) else ", ".join(str(x) for x in sp_val if str(x).strip()) if "rating" in mapped and "rating" not in explicit: rating = mapped["rating"] + if "parallel_queue" in mapped and "parallel_queue" not in explicit: + parallel_queue = int(mapped["parallel_queue"]) has_content = bool(prompt or character or character_prompt or scene or scene_prompt) if not has_content: console.print("[red]Prompt (or character/scene) is required[/red]") raise typer.Exit(1) - _run_generation( - prompt=prompt, - model=model, - width=width, - height=height, - steps=steps, - cfg=cfg, - guidance=guidance, - seed=seed, - sampler=sampler, - scheduler=scheduler, - vae=vae, - orientation=orientation, - lora=lora, - lora_strength=lora_strength, - negative=negative, - count=count, - rating=rating, - no_quality=no_quality, - no_negative=no_negative, - character=character, - character_prompt=character_prompt, - scene=scene, - scene_prompt=scene_prompt, - family=family, - output=output, - remote=remote, - json_output=json_output, - ) + # Effective parallelism is bounded by count — running 4 threads for 1 image + # is silly. count=1 always goes through the sequential path regardless of -P. + effective_parallel = min(parallel_queue, count) if count > 1 else 1 + + if effective_parallel <= 1: + # Sequential path: single _run_generation call with batch_size=count. + # Unchanged from pre-parallel behavior — preserves existing output naming, + # JSON shape, and log lines exactly. + _run_generation( + prompt=prompt, + model=model, + width=width, + height=height, + steps=steps, + cfg=cfg, + guidance=guidance, + seed=seed, + sampler=sampler, + scheduler=scheduler, + vae=vae, + orientation=orientation, + lora=lora, + lora_strength=lora_strength, + negative=negative, + count=count, + rating=rating, + no_quality=no_quality, + no_negative=no_negative, + character=character, + character_prompt=character_prompt, + scene=scene, + scene_prompt=scene_prompt, + family=family, + output=output, + remote=remote, + json_output=json_output, + ) + return + + # ---- Parallel fanout path ---- + # Split count into `count` independent jobs (batch_size=1), executed + # `effective_parallel` at a time. Each job gets a distinct seed and a + # distinct output path so writes don't clobber each other. + import random as _rng # noqa: PLC0415 + import time as _time # noqa: PLC0415 + from concurrent.futures import ThreadPoolExecutor, as_completed # noqa: PLC0415 + + # Resolve bare model/lora names ONCE in the parent before fanout. Each + # parallel _run_generation call silences its own console (json_output=True) + # which also skips the validation/resolution step in that path. Doing it + # here means each task receives a canonical filename and ComfyUI's strict + # loaders accept the request first try. + if model and not remote: + # Detect family for the right loader bucket (checkpoints vs diffusion_models). + # Mirrors the lookup _run_generation does on entry. + from tensors.db import Database # noqa: PLC0415 + + _base_model: str | None = None + try: + with Database() as _db: + _db.init_schema() + _base_model = _db.get_base_model_by_filename(model) + except Exception: # noqa: BLE001 + pass + _detected = detect_model_family(model, _base_model) + _fam = family or _detected + try: + model, lora = _validate_model_available(model, _fam, lora) + except typer.Exit: + raise # surface the same error path as sequential + + # Seed strategy: + # --seed >= 0 → use as base, increment per job (reproducible series) + # --seed == -1 → pick a fresh random seed PER JOB so parallel runs aren't + # accidentally correlated (each thread gets variety) + if seed >= 0: + seeds = [seed + i for i in range(count)] + else: + seeds = [_rng.randint(0, 2**32 - 1) for _ in range(count)] + + # Output paths: mirror the existing `count > 1` naming convention from + # _run_generation (stem_NNN.ext). When --output is omitted, leave per-task + # output as None — _run_generation will skip the disk write and the user + # gets only the console listing of generated image refs. + out_paths: list[Path | None] = [] + for i in range(count): + if output is None: + out_paths.append(None) + else: + out_paths.append(output.parent / f"{output.stem}_{i + 1:03d}{output.suffix}") + + if not json_output: + console.print( + f"[dim]Parallel queue: {effective_parallel} concurrent submissions " + f"× {count} images (output may interleave)[/dim]" + ) + + common_kwargs: dict[str, Any] = { + "prompt": prompt, + "model": model, + "width": width, + "height": height, + "steps": steps, + "cfg": cfg, + "guidance": guidance, + "sampler": sampler, + "scheduler": scheduler, + "vae": vae, + "orientation": orientation, + "lora": lora, + "lora_strength": lora_strength, + "negative": negative, + "count": 1, # each task generates exactly one image + "rating": rating, + "no_quality": no_quality, + "no_negative": no_negative, + "character": character, + "character_prompt": character_prompt, + "scene": scene, + "scene_prompt": scene_prompt, + "family": family, + "remote": remote, + # NOTE: json_output stays False so _run_generation's disk-save path runs. + # Setting True would short-circuit before saving files. Per-task console + # chatter is the trade-off; the final summary still shows clean per-task + # status lines. + "json_output": False, + } + + def _run_one(idx: int) -> dict[str, Any]: + """Run a single batch_size=1 job. Returns a result dict (success captured).""" + start = _time.perf_counter() + result: dict[str, Any] = { + "index": idx, + "seed": seeds[idx], + "output": str(out_paths[idx]) if out_paths[idx] is not None else None, + "duration_sec": 0.0, + "success": False, + "error": None, + } + try: + _run_generation(seed=seeds[idx], output=out_paths[idx], **common_kwargs) + result["duration_sec"] = round(_time.perf_counter() - start, 2) + result["success"] = True + except typer.Exit as ex: + result["duration_sec"] = round(_time.perf_counter() - start, 2) + result["error"] = f"generate exited with code {ex.exit_code}" + except Exception as ex: # noqa: BLE001 + result["duration_sec"] = round(_time.perf_counter() - start, 2) + result["error"] = str(ex) + return result + + fan_results: list[dict[str, Any]] = [] + with ThreadPoolExecutor(max_workers=effective_parallel) as pool: + futures = {pool.submit(_run_one, i): i for i in range(count)} + completed = 0 + for fut in as_completed(futures): + completed += 1 + try: + res = fut.result() + except Exception as ex: # noqa: BLE001 — defensive; _run_one already swallows + res = { + "index": futures[fut], + "seed": seeds[futures[fut]], + "output": str(out_paths[futures[fut]]) if out_paths[futures[fut]] is not None else None, + "duration_sec": 0.0, + "success": False, + "error": f"executor exception: {ex}", + } + fan_results.append(res) + if not json_output: + if res["success"]: + where = res["output"] or "(no --output set)" + console.print( + f"[green]\\[{completed}/{count}] seed={res['seed']} " + f"ok in {res['duration_sec']:.1f}s → {where}[/green]" + ) + else: + console.print( + f"[red]\\[{completed}/{count}] seed={res['seed']} FAIL: {res['error']}[/red]" + ) + + # Reorder by original index so JSON output / final summary list is stable. + fan_results.sort(key=lambda r: r["index"]) + successful = sum(1 for r in fan_results if r["success"]) + + if json_output: + console.print_json( + data={ + "success": successful == count, + "count": count, + "parallel_queue": effective_parallel, + "results": fan_results, + } + ) + return + + console.print("[bold green]Generation complete![/bold green]") + console.print(f"[dim]Generated {successful}/{count} images at parallelism={effective_parallel}[/dim]") + if successful < count: + raise typer.Exit(1) # Map model family → which ComfyUI loader directory the checkpoint must live in. diff --git a/tests/test_generate_parallel.py b/tests/test_generate_parallel.py new file mode 100644 index 0000000..c3601d1 --- /dev/null +++ b/tests/test_generate_parallel.py @@ -0,0 +1,325 @@ +"""Tests for the `tsr generate --parallel-queue` flag (parallel fanout path).""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest +from typer.testing import CliRunner + +from tensors import cli as cli_module +from tensors.cli import app + +runner = CliRunner() + + +# ----------------------------------------------------------------------------- +# Fixtures +# ----------------------------------------------------------------------------- + + +@pytest.fixture +def calls(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]: + """Record every _run_generation call and stub the disk-write side effect. + + The parallel fanout path invokes _run_generation N times (one per task); + the sequential path invokes it once. By recording kwargs we can assert + fanout behavior (per-task seeds, per-task output paths, count=1 per task) + without round-tripping ComfyUI. + """ + recorded: list[dict[str, Any]] = [] + + def fake_run_generation(**kwargs: Any) -> None: + recorded.append(kwargs) + out: Path | None = kwargs.get("output") + if out is not None: + out.parent.mkdir(parents=True, exist_ok=True) + out.write_bytes(b"fake-png") + + monkeypatch.setattr(cli_module, "_run_generation", fake_run_generation) + return recorded + + +@pytest.fixture(autouse=True) +def _stub_model_validation(monkeypatch: pytest.MonkeyPatch) -> None: + """Bypass ComfyUI's live model lookup so tests don't need a backend.""" + monkeypatch.setattr( + cli_module, + "_validate_model_available", + lambda model, family, lora: (model, lora), + ) + + +# ----------------------------------------------------------------------------- +# Validation / sanity +# ----------------------------------------------------------------------------- + + +def test_parallel_queue_invalid_value_rejected(calls: list[dict[str, Any]]) -> None: + """--parallel-queue 0 (or negative) exits non-zero before any work.""" + result = runner.invoke( + app, + ["generate", "test prompt", "-m", "x.safetensors", "--parallel-queue", "0"], + ) + assert result.exit_code != 0 + assert "--parallel-queue must be >= 1" in result.output + assert calls == [] + + +def test_parallel_queue_one_is_sequential_path(calls: list[dict[str, Any]]) -> None: + """-P 1 collapses to the legacy single _run_generation call with count=N. + + This is the key compatibility contract: existing scripts that don't pass + -P must see identical behavior (one call, count forwarded as batch_size). + """ + result = runner.invoke( + app, + ["generate", "test prompt", "-m", "x.safetensors", "-c", "4", "-P", "1"], + ) + assert result.exit_code == 0, result.output + assert len(calls) == 1 + assert calls[0]["count"] == 4 + assert calls[0]["prompt"] == "test prompt" + + +def test_count_one_ignores_parallel_queue(calls: list[dict[str, Any]]) -> None: + """count=1 always takes sequential path regardless of -P (no fanout point).""" + result = runner.invoke( + app, + ["generate", "test prompt", "-m", "x.safetensors", "-c", "1", "-P", "8"], + ) + assert result.exit_code == 0, result.output + assert len(calls) == 1 + assert calls[0]["count"] == 1 + + +def test_json_output_incompatible_with_parallel(calls: list[dict[str, Any]]) -> None: + """--json + -P>1 errors out cleanly (would skip disk-save inside tasks).""" + result = runner.invoke( + app, + ["generate", "test prompt", "-m", "x.safetensors", "-c", "2", "-P", "2", "--json"], + ) + assert result.exit_code != 0 + assert "--json is not supported with --parallel-queue > 1" in result.output + assert calls == [] + + +# ----------------------------------------------------------------------------- +# Fanout behavior +# ----------------------------------------------------------------------------- + + +def test_parallel_fanout_creates_n_tasks(tmp_path: Path, calls: list[dict[str, Any]]) -> None: + """-c N -P M (M>1, N>1) → N independent _run_generation calls, each count=1.""" + out = tmp_path / "img.png" + result = runner.invoke( + app, + [ + "generate", + "test prompt", + "-m", + "x.safetensors", + "-c", + "4", + "-P", + "2", + "--seed", + "100", + "-o", + str(out), + ], + ) + assert result.exit_code == 0, result.output + assert len(calls) == 4 + # Each task generates exactly one image + for c in calls: + assert c["count"] == 1 + + +def test_parallel_seeds_increment_from_base(tmp_path: Path, calls: list[dict[str, Any]]) -> None: + """Explicit --seed → each task receives base+i (reproducible series).""" + out = tmp_path / "img.png" + runner.invoke( + app, + ["generate", "test prompt", "-m", "x.safetensors", "-c", "3", "-P", "3", "--seed", "500", "-o", str(out)], + ) + seeds_seen = sorted(c["seed"] for c in calls) + assert seeds_seen == [500, 501, 502] + + +def test_parallel_seeds_random_when_unset(tmp_path: Path, calls: list[dict[str, Any]]) -> None: + """seed=-1 → each task gets a freshly-rolled random seed (not all the same). + + Vanishingly small chance of collision across 4 random ints; treat as flake + threshold of "all distinct" rather than exact equality to any value. + """ + out = tmp_path / "img.png" + runner.invoke( + app, + ["generate", "test prompt", "-m", "x.safetensors", "-c", "4", "-P", "2", "-o", str(out)], + ) + seeds = [c["seed"] for c in calls] + # All non-negative (i.e. resolved from -1 to actual int) and distinct. + assert all(s >= 0 for s in seeds) + assert len(set(seeds)) == len(seeds) + + +def test_parallel_output_paths_indexed(tmp_path: Path, calls: list[dict[str, Any]]) -> None: + """Per-task output paths use stem_NNN.suffix naming (matches sequential count>1).""" + out = tmp_path / "scene.png" + runner.invoke( + app, + ["generate", "test prompt", "-m", "x.safetensors", "-c", "3", "-P", "3", "--seed", "1", "-o", str(out)], + ) + paths = sorted(str(c["output"]) for c in calls) + assert paths == [ + str(tmp_path / "scene_001.png"), + str(tmp_path / "scene_002.png"), + str(tmp_path / "scene_003.png"), + ] + + +def test_parallel_without_output_passes_none(tmp_path: Path, calls: list[dict[str, Any]]) -> None: + """When --output is omitted, each task gets output=None (no disk write planned).""" + runner.invoke( + app, + ["generate", "test prompt", "-m", "x.safetensors", "-c", "2", "-P", "2", "--seed", "1"], + ) + assert len(calls) == 2 + assert all(c["output"] is None for c in calls) + + +def test_parallel_files_actually_written(tmp_path: Path, calls: list[dict[str, Any]]) -> None: + """End-to-end: per-task stub writes its file → all N appear on disk. + + Guards against the bug where json_output=True short-circuits the save block + inside _run_generation. Each task must use the non-JSON code path. + """ + out = tmp_path / "shot.png" + runner.invoke( + app, + ["generate", "test prompt", "-m", "x.safetensors", "-c", "3", "-P", "3", "--seed", "1", "-o", str(out)], + ) + written = sorted(p.name for p in tmp_path.iterdir()) + assert written == ["shot_001.png", "shot_002.png", "shot_003.png"] + + +def test_parallel_summary_reports_success_count(tmp_path: Path, calls: list[dict[str, Any]]) -> None: + """Final summary line reports N/N success when all tasks complete.""" + out = tmp_path / "img.png" + result = runner.invoke( + app, + ["generate", "test prompt", "-m", "x.safetensors", "-c", "3", "-P", "2", "--seed", "1", "-o", str(out)], + ) + assert result.exit_code == 0 + assert "Generated 3/3 images" in result.output + + +def test_parallel_partial_failure_exits_nonzero( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """If one task raises, summary shows partial count and command exits non-zero.""" + import typer + + call_indices: list[int] = [] + + def flaky_run_generation(**kwargs: Any) -> None: + # Fail every other call to simulate intermittent backend errors. + idx = len(call_indices) + call_indices.append(idx) + if idx % 2 == 0: + raise typer.Exit(1) + out: Path | None = kwargs.get("output") + if out is not None: + out.parent.mkdir(parents=True, exist_ok=True) + out.write_bytes(b"ok") + + monkeypatch.setattr(cli_module, "_run_generation", flaky_run_generation) + + out = tmp_path / "img.png" + result = runner.invoke( + app, + ["generate", "test prompt", "-m", "x.safetensors", "-c", "4", "-P", "2", "--seed", "1", "-o", str(out)], + ) + assert result.exit_code != 0 + # Two tasks failed; final summary should show 2/4. + assert "Generated 2/4 images" in result.output + + +# ----------------------------------------------------------------------------- +# --input integration +# ----------------------------------------------------------------------------- + + +def test_parallel_queue_from_yaml_input(tmp_path: Path, calls: list[dict[str, Any]]) -> None: + """parallel_queue can be set via --input YAML (mirrors other generate params).""" + out = tmp_path / "img.png" + yml = tmp_path / "spec.yml" + yml.write_text( + f'prompt: from-yaml\nmodel: x.safetensors\ncount: 3\nparallel_queue: 3\nseed: 7\noutput: "{out}"\n' + ) + result = runner.invoke(app, ["generate", "--input", str(yml)]) + assert result.exit_code == 0, result.output + assert len(calls) == 3 + assert sorted(c["seed"] for c in calls) == [7, 8, 9] + + +def test_cli_parallel_queue_overrides_yaml(tmp_path: Path, calls: list[dict[str, Any]]) -> None: + """CLI --parallel-queue wins over YAML's parallel_queue (standard precedence).""" + out = tmp_path / "img.png" + yml = tmp_path / "spec.yml" + yml.write_text( + f'prompt: from-yaml\nmodel: x.safetensors\ncount: 2\nparallel_queue: 1\nseed: 10\noutput: "{out}"\n' + ) + # YAML says P=1 (sequential), CLI overrides to P=2 (fanout) + result = runner.invoke(app, ["generate", "--input", str(yml), "-P", "2"]) + assert result.exit_code == 0, result.output + # Fanout path → 2 separate calls, each count=1 + assert len(calls) == 2 + assert all(c["count"] == 1 for c in calls) + + +# ----------------------------------------------------------------------------- +# Concurrency assertion +# ----------------------------------------------------------------------------- + + +def test_parallel_actually_runs_concurrently( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Sanity: P concurrent tasks really overlap in time (vs all-serial).""" + import threading + import time as _t + + in_flight = 0 + peak_in_flight = 0 + lock = threading.Lock() + + def slow_run_generation(**kwargs: Any) -> None: + nonlocal in_flight, peak_in_flight + with lock: + in_flight += 1 + peak_in_flight = max(peak_in_flight, in_flight) + _t.sleep(0.1) # 100ms — long enough to overlap, short enough for fast tests + with lock: + in_flight -= 1 + out: Path | None = kwargs.get("output") + if out is not None: + out.parent.mkdir(parents=True, exist_ok=True) + out.write_bytes(b"ok") + + monkeypatch.setattr(cli_module, "_run_generation", slow_run_generation) + + out = tmp_path / "img.png" + result = runner.invoke( + app, + ["generate", "test prompt", "-m", "x.safetensors", "-c", "4", "-P", "4", "--seed", "1", "-o", str(out)], + ) + assert result.exit_code == 0, result.output + # With P=4 and 4 tasks each sleeping 100ms, peak concurrency should hit 4. + # Even allowing for thread-pool warmup quirks, ≥2 means parallelism is real. + assert peak_in_flight >= 2, f"peak_in_flight={peak_in_flight} (expected ≥2 for parallel)"