Make the pipelining knobs config-file settable (server: section)

CPU_WORKERS / GPU_CONCURRENCY were env-only. Add a `server:` config
section (server.cpu_workers, server.gpu_concurrency) so a lane's
concurrency lives in its yaml alongside device/precision, matching how
model.device already works. Precedence mirrors BGFILTER_DEVICE: config is
the default, the env var overrides per box.

The pool + admission semaphore now build in lifespan (after the config
loads) instead of at import, and PipelineManager reads gpu_concurrency
from the settings. configs/gpu.yaml ships 8 / 2 (its target is the 32 GB
GPU boxes) with an inline VRAM caveat: gpu_concurrency>1 must fit
weights + N × forward, so keep it 1 on ≤16 GB cards.

Verified on a 4080 SUPER: startup logs cpu_workers=8 gpu_concurrency=2
from gpu.yaml with no env set, request returns a valid RGBA.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-14 12:58:15 +00:00
parent c748fc22a4
commit 07ba02fdde
7 changed files with 56 additions and 15 deletions
+16 -8
View File
@@ -74,12 +74,13 @@ _manager: PipelineManager | None = None
# 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)
# The pool size defaults to config (server.cpu_workers, default 1 == strictly serial)
# and is overridable by BGFILTER_CPU_WORKERS. Both are resolved in lifespan once the
# config is loaded; a same-sized semaphore bounds in-flight requests (hence
# decoded-image RAM) so a burst cannot pile unbounded decoded images before the pool
# drains.
_executor: ThreadPoolExecutor | None = None
_sem: asyncio.Semaphore | None = None
def _build_settings():
@@ -105,15 +106,22 @@ def _build_settings():
@asynccontextmanager
async def lifespan(app: FastAPI):
global _manager
global _manager, _executor, _sem
settings = _build_settings()
preload = os.environ.get("BGFILTER_PRELOAD", "1").strip().lower() not in ("0", "false", "no", "")
# Pool size: config default (server.cpu_workers), env override wins.
cpu_workers = max(1, _int_env("BGFILTER_CPU_WORKERS", settings.server.cpu_workers))
_executor = ThreadPoolExecutor(max_workers=cpu_workers, thread_name_prefix="bgf")
_sem = asyncio.Semaphore(cpu_workers)
logger.info(
"starting bgfilter service: device=%s default_seg=%s cross_check=%s preload=%s max_pixels=%d auth=%s",
"starting bgfilter service: device=%s default_seg=%s cross_check=%s preload=%s "
"cpu_workers=%d gpu_concurrency=%d max_pixels=%d auth=%s",
settings.model.device,
settings.segmentation.backend,
settings.cross_check.enabled,
preload,
cpu_workers,
max(1, _int_env("BGFILTER_GPU_CONCURRENCY", settings.server.gpu_concurrency)),
MAX_IMAGE_PIXELS,
"enabled" if AUTH_TOKEN else "DISABLED(open)",
)