Pipeline concurrent requests over a shared GPU with a narrowed lock

The single asyncio lock serialized the whole request (chroma -> seg ->
trimap -> ViTMatte -> cross-check -> foreground -> despill), so the GPU
sat idle ~70-80% of each request while the CPU stages ran. Replace it
with a thread pool that runs several requests concurrently, and move the
mutex down to just the GPU forwards: when one request steps off the GPU
to do CPU work, another steps on, keeping the device fed.

Two opt-in env knobs, both defaulting to the old strictly-serial
behaviour:
  * BGFILTER_CPU_WORKERS (pool size) attacks the CPU-bound regime -- with
    cross-check off, throughput went 0.91 -> 1.67 req/s (1.84x) at pool=8
    on a 4080S, vs a mere +6% from a second worker process, at half the
    VRAM (one model copy).
  * BGFILTER_GPU_CONCURRENCY caps how many forwards may co-schedule on the
    SMs. batch-1 underfills the GPU, so =2 recovers the SM overlap a lock
    forgoes: cross-check on went 0.83 -> 1.01 req/s, beating two worker
    processes (0.92) with one model copy (~19 GB vs 24 GB).

The CLI/single-thread path passes no lock and uses a nullcontext, so it
stays allocation- and contention-free. Concurrent output verified
identical (10x same-frame -> one hash). A per-worker admission semaphore
bounds in-flight decoded images to the pool size.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app.py

@@ -13,6 +13,7 @@ import logging
 import os
 import re
 import time
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import asynccontextmanager
 from pathlib import Path
 
@@ -67,7 +68,18 @@ async def require_auth(x_genarrative_image_token: str | None = Header(default=No
 
 
 _manager: PipelineManager | None = None
-_lock = asyncio.Lock()
+
+# Request pipelining. Each request's process() runs on a small thread pool so the
+# CPU stages (chroma / trimap / alpha_post / foreground) of concurrent requests
+# overlap, while a GPU mutex inside the pipeline keeps the model forwards mutually
+# exclusive on the single device -- i.e. when one request steps off the GPU to do
+# CPU work, another steps on, keeping the GPU fed instead of idle.
+# BGFILTER_CPU_WORKERS=1 (default) gives a 1-thread pool == the old strictly serial
+# behaviour. A semaphore bounds in-flight requests (hence decoded-image RAM) to the
+# pool size so a burst cannot pile unbounded decoded images before the pool drains.
+CPU_WORKERS = max(1, _int_env("BGFILTER_CPU_WORKERS", 1))
+_executor = ThreadPoolExecutor(max_workers=CPU_WORKERS, thread_name_prefix="bgf")
+_sem = asyncio.Semaphore(CPU_WORKERS)
 
 
 def _build_settings():
@@ -240,14 +252,16 @@ async def remove_background(
     # Honors the lane lock (a locked cpu-fast box reports/runs "off" even on cross_check=on).
     effective_cc = _manager.effective_cross_check(normalized_cc)
 
-    rgb = _decode_image(await file.read())
-
-    start = time.perf_counter()
-    async with _lock:  # one inference at a time per process
+    # Bound in-flight requests to the pool size, then run on the pool. The GPU
+    # sections are serialized inside the pipeline (a shared GPU mutex); the CPU
+    # sections of concurrent requests overlap across the pool threads.
+    async with _sem:
+        rgb = _decode_image(await file.read())
+        start = time.perf_counter()
         loop = asyncio.get_running_loop()
         try:
             result = await loop.run_in_executor(
-                None,
+                _executor,
                 lambda: _manager.process(
                     rgb,
                     screen_color=normalized_color,
@@ -268,7 +282,7 @@ async def remove_background(
         except Exception:
             logger.exception("inference failed")
             raise HTTPException(status_code=500, detail="inference failed")
-    elapsed_ms = int((time.perf_counter() - start) * 1000)
+        elapsed_ms = int((time.perf_counter() - start) * 1000)
 
     png = _encode_png(result.rgb, result.alpha)
     headers = {