From 089778773c3b95caac42334f6ffc5376696caef2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marco=20Macr=C3=AC?= <62335226+Mark2Mac@users.noreply.github.com> Date: Wed, 19 Aug 2026 02:01:58 +0200 Subject: [PATCH 1/2] fix(llm): bound total in-flight LLM requests, not one analyzer's fan-out MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `SKILLSPECTOR_MAX_LLM_CONCURRENCY` creates its semaphore inside a single analyzer's batch fan-out. The analyzers are separate graph nodes and the workflow fans out to them in parallel, so each one gets its own semaphore and the process puts N x limit requests on the wire. It is not merely ineffective, it multiplies. Peak in-flight requests measured with two analyzers running concurrently: limit 1 -> 2 in flight limit 2 -> 4 in flight The docstring says users on rate-limited endpoints "can set it to 1 to serialize requests"; they cannot. On a free-tier endpoint this is the difference between one analyzer completing and all four: the extra requests arrive together and come back 429. This shares one semaphore per event loop, per resolved limit. Keying by loop keeps unrelated loops independent (tests, repeated CLI invocations) and the weak key drops the entry with the loop; keying by limit as well means a caller that resolves a different value gets its own semaphore instead of replacing one other coroutines are currently holding. An explicit `max_concurrency=` argument stays local to its call. Callers that pass a number are asking for a fan-out width, not for a share of the process-wide budget, and the existing tests rely on that isolation. Defaults are unchanged: with the variable unset the ceiling is the same as before, now applied once instead of once per analyzer. Closes #387 Signed-off-by: Marco Macrì <62335226+Mark2Mac@users.noreply.github.com> --- src/skillspector/llm_analyzer_base.py | 38 ++++++++++++- tests/nodes/test_llm_analyzer_base.py | 82 +++++++++++++++++++++++++++ 2 files changed, 117 insertions(+), 3 deletions(-) diff --git a/src/skillspector/llm_analyzer_base.py b/src/skillspector/llm_analyzer_base.py index 0fbf8d23..0c03a52b 100644 --- a/src/skillspector/llm_analyzer_base.py +++ b/src/skillspector/llm_analyzer_base.py @@ -30,6 +30,7 @@ import asyncio import os import time +import weakref from collections import defaultdict from dataclasses import dataclass, field from typing import Any, Literal, cast @@ -93,12 +94,38 @@ def _uses_native_connection_retries(chat_model: object) -> bool: return False +# One semaphore per event loop, per resolved limit. The analyzers are separate graph nodes that +# the workflow fans out to in parallel, so a semaphore created inside one analyzer's fan-out +# bounds that analyzer alone: with N analyzers the process puts N x limit requests on the wire. +# Keying by loop keeps unrelated loops (tests, repeated CLI invocations) independent, and the +# weak key lets the entry go when the loop does. Keying by limit as well means a caller that +# resolves a different value gets its own semaphore instead of replacing one that other +# coroutines are currently holding. +_shared_semaphores: weakref.WeakKeyDictionary[ + asyncio.AbstractEventLoop, dict[int, asyncio.Semaphore] +] = weakref.WeakKeyDictionary() + + +def _shared_semaphore(limit: int) -> asyncio.Semaphore: + """Return the process-wide semaphore for *limit* on the running loop.""" + loop = asyncio.get_running_loop() + per_limit = _shared_semaphores.get(loop) + if per_limit is None: + per_limit = {} + _shared_semaphores[loop] = per_limit + semaphore = per_limit.get(limit) + if semaphore is None: + semaphore = asyncio.Semaphore(limit) + per_limit[limit] = semaphore + return semaphore + + def resolve_max_concurrency() -> int: """Resolve the LLM fan-out concurrency from ``SKILLSPECTOR_MAX_LLM_CONCURRENCY``. Defaults to :data:`DEFAULT_MAX_LLM_CONCURRENCY`. Users on rate-limited providers (free tiers with a low RPM) can set it to ``1`` to serialize - requests instead of bursting up to 10 in parallel — a burst that otherwise + requests across every analyzer instead of bursting up to 10 in parallel — a burst that otherwise guarantees 429s, and 429'd batches are dropped from the result (see the analyzer fan-out below). Invalid values fall back to the default; values below 1 are clamped to 1. @@ -815,9 +842,14 @@ async def arun_batches_detailed( **kwargs: object, ) -> BatchExecutionResult: """Execute batches concurrently and retain sanitized per-batch failures.""" + # Resolved from the environment: share one semaphore across every analyzer on this + # loop, because that is what the variable promises. An explicit argument keeps its + # documented meaning and stays local to this call — callers that pass a number are + # asking for a fan-out width, not for a share of the process-wide budget. if max_concurrency is None: - max_concurrency = resolve_max_concurrency() - sem = asyncio.Semaphore(max_concurrency) + sem = _shared_semaphore(resolve_max_concurrency()) + else: + sem = asyncio.Semaphore(max_concurrency) async def _process(batch: Batch) -> tuple[Batch, list]: async with sem: diff --git a/tests/nodes/test_llm_analyzer_base.py b/tests/nodes/test_llm_analyzer_base.py index a6fb0dae..71e5e0fc 100644 --- a/tests/nodes/test_llm_analyzer_base.py +++ b/tests/nodes/test_llm_analyzer_base.py @@ -17,6 +17,7 @@ from __future__ import annotations +import asyncio import json from unittest.mock import AsyncMock, MagicMock, patch @@ -2495,3 +2496,84 @@ def test_unknown_model_uses_default(self) -> None: out = get_max_output_tokens("unknown/model") assert inp == int(mocked_ctx * 0.75) assert out == int(mocked_ctx * 0.25) + + +class TestConcurrencyIsGlobal: + """`SKILLSPECTOR_MAX_LLM_CONCURRENCY` must bound the whole process, not one analyzer. + + The analyzers are separate LangGraph nodes and the graph fans out to them in parallel + (``workflow.add_edge("build_context", analyzer_id)`` for each). A semaphore created inside + one analyzer's fan-out therefore bounds only that analyzer: setting the variable to 1 still + puts N requests on the wire at once, which is exactly what a user on a rate-limited endpoint + set it to 1 to avoid. + """ + + MODEL = "gpt-4o-mini" + + @staticmethod + def _counting_analyzer(peak: list[int], live: list[int]) -> LLMAnalyzerBase: + analyzer = LLMAnalyzerBase(base_prompt="test", model=TestConcurrencyIsGlobal.MODEL) + + async def _invoke(*_args, **_kwargs): + live[0] += 1 + peak[0] = max(peak[0], live[0]) + # Yield control so a second in-flight request has the chance to be observed. Without + # this the coroutine could complete before any other one starts, and the assertion + # would pass on a serialization the code does not actually provide. + await asyncio.sleep(0.02) + live[0] -= 1 + return LLMAnalysisResult(findings=[]) + + analyzer._structured_llm.ainvoke = AsyncMock(side_effect=_invoke) + return analyzer + + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + async def test_two_analyzers_respect_one_global_slot(self, monkeypatch) -> None: + monkeypatch.setenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", "1") + peak, live = [0], [0] + first = self._counting_analyzer(peak, live) + second = self._counting_analyzer(peak, live) + batches = [Batch(file_path="a.py", content="a"), Batch(file_path="b.py", content="b")] + + await asyncio.gather( + first.arun_batches_detailed(list(batches)), + second.arun_batches_detailed(list(batches)), + ) + + assert peak[0] == 1, ( + f"{peak[0]} requests were in flight with the limit set to 1: the bound is per " + "analyzer, so N analyzers issue N simultaneous requests" + ) + + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + async def test_limit_of_two_allows_two_across_analyzers(self, monkeypatch) -> None: + """The bound must be the ceiling, not a serialization: a knob that only ever means 1 is + as wrong as one that means nothing.""" + monkeypatch.setenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", "2") + peak, live = [0], [0] + first = self._counting_analyzer(peak, live) + second = self._counting_analyzer(peak, live) + batches = [Batch(file_path="a.py", content="a"), Batch(file_path="b.py", content="b")] + + await asyncio.gather( + first.arun_batches_detailed(list(batches)), + second.arun_batches_detailed(list(batches)), + ) + + assert peak[0] == 2 + + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + async def test_explicit_argument_still_bounds_only_its_own_call(self, monkeypatch) -> None: + """An explicit ``max_concurrency`` keeps its documented meaning: it wins for that call. + + Callers that pass a number are asking for a local fan-out width, not for a share of the + process-wide budget, and tests rely on that isolation. + """ + monkeypatch.setenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", "1") + peak, live = [0], [0] + analyzer = self._counting_analyzer(peak, live) + batches = [Batch(file_path="a.py", content="a"), Batch(file_path="b.py", content="b")] + + await analyzer.arun_batches_detailed(batches, max_concurrency=2) + + assert peak[0] == 2 From cabbd0808be188e1f9481150a246d785ecc0995a Mon Sep 17 00:00:00 2001 From: Mark2Mac Date: Mon, 24 Aug 2026 21:19:44 +0200 Subject: [PATCH 2/2] fix(llm): bound the process, not one event loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The review is right and the previous commit was not. Keying the semaphore by the running event loop keys it by the analyzer: every analyzer node is a synchronous LangGraph node that reaches run_async(), which calls asyncio.run() -- a new loop each time, on a new thread when a loop is already running. So each node got its own semaphore and the N x limit burst survived. The tests could not see it, because they gathered two analyzers on one artificial loop, which is the one shape production never takes. The permits now live outside asyncio, under a plain lock, and each waiter parks on a future belonging to its own loop; releasing hands the permit to the next waiter through that loop's call_soon_threadsafe. Nothing blocks a worker thread while it waits, which rules out the obvious alternative of wrapping a threading.Semaphore in run_in_executor -- that pins one thread per queued request, and the queue is exactly as long as the fan-out this exists to bound. A cancelled waiter is handled explicitly, because that failure mode is silent: it is removed from the queue, and if the permit reached it in the same instant the wait was abandoned, the permit is passed on rather than dropped. A leaked permit does not raise -- the count simply never recovers, and every later request waits for a slot that no longer exists, hours after the cancellation. The new bench runs three analyzers the way the graph does: three threads, three loops, one lock-protected counter, through run_async and not through asyncio .gather. It was seen red against the code this review rejected -- "6 requests were in flight with the limit set to 2" -- which is exactly N x limit with three nodes. It also asserts the bound is a ceiling and not a queue (three nodes at a limit of 3 must reach 3) and that no permit is stranded by a cancellation. 162 tests in this file pass, ruff check and format clean. Three failures in tests/nodes/test_security_*.py are pre-existing on this branch: they fail identically with this change stashed. Signed-off-by: Marco Macrì --- src/skillspector/llm_analyzer_base.py | 134 ++++++++++++++++++++------ tests/nodes/test_llm_analyzer_base.py | 111 ++++++++++++++++++++- 2 files changed, 213 insertions(+), 32 deletions(-) diff --git a/src/skillspector/llm_analyzer_base.py b/src/skillspector/llm_analyzer_base.py index f4b9f8bf..3a682107 100644 --- a/src/skillspector/llm_analyzer_base.py +++ b/src/skillspector/llm_analyzer_base.py @@ -29,9 +29,9 @@ import asyncio import os +import threading import time -import weakref -from collections import defaultdict +from collections import defaultdict, deque from collections.abc import Callable from dataclasses import dataclass, field from typing import Any, Literal, cast @@ -104,30 +104,101 @@ def _uses_native_connection_retries( return False -# One semaphore per event loop, per resolved limit. The analyzers are separate graph nodes that -# the workflow fans out to in parallel, so a semaphore created inside one analyzer's fan-out -# bounds that analyzer alone: with N analyzers the process puts N x limit requests on the wire. -# Keying by loop keeps unrelated loops (tests, repeated CLI invocations) independent, and the -# weak key lets the entry go when the loop does. Keying by limit as well means a caller that -# resolves a different value gets its own semaphore instead of replacing one that other -# coroutines are currently holding. -_shared_semaphores: weakref.WeakKeyDictionary[ - asyncio.AbstractEventLoop, dict[int, asyncio.Semaphore] -] = weakref.WeakKeyDictionary() - - -def _shared_semaphore(limit: int) -> asyncio.Semaphore: - """Return the process-wide semaphore for *limit* on the running loop.""" - loop = asyncio.get_running_loop() - per_limit = _shared_semaphores.get(loop) - if per_limit is None: - per_limit = {} - _shared_semaphores[loop] = per_limit - semaphore = per_limit.get(limit) - if semaphore is None: - semaphore = asyncio.Semaphore(limit) - per_limit[limit] = semaphore - return semaphore +# ONE BUDGET FOR THE PROCESS, AND IT CANNOT BE AN asyncio.Semaphore. +# +# The analyzers are separate graph nodes and each node is a *synchronous* function: it reaches +# ``run_async()``, which calls ``asyncio.run()`` — and, when a loop is already running, does so on +# a fresh thread. Every analyzer therefore gets its own event loop. An ``asyncio.Semaphore`` is +# bound to the loop that created it, so one semaphore per loop is one semaphore per analyzer, and +# the process still puts N x limit requests on the wire. That is the burst +# ``SKILLSPECTOR_MAX_LLM_CONCURRENCY`` exists to prevent: users set it to 1 on a free tier +# precisely because a burst guarantees 429s, and 429'd batches are dropped from the result. +# +# So the permits live outside asyncio, under a plain lock, and each waiter parks on a future +# belonging to *its own* loop. Releasing hands the permit to the next waiter through that loop's +# ``call_soon_threadsafe``, which is the one asyncio primitive that is safe to call from another +# thread. Nothing blocks a worker thread while it waits, which rules out the obvious alternative +# of wrapping a ``threading.Semaphore`` in ``run_in_executor``: that pins one thread per queued +# request, and the queue is exactly as long as the fan-out this is meant to bound. +class _GlobalLLMLimiter: + """A permit counter shared by every event loop and thread in the process.""" + + def __init__(self, limit: int) -> None: + self._limit = max(1, limit) + self._lock = threading.Lock() + self._in_flight = 0 + # (loop, future) in arrival order. FIFO, so a late analyzer is not starved by an early + # one that keeps re-acquiring. + self._waiters: deque[tuple[asyncio.AbstractEventLoop, asyncio.Future]] = deque() + + async def acquire(self) -> None: + loop = asyncio.get_running_loop() + with self._lock: + if self._in_flight < self._limit: + self._in_flight += 1 + return + waiter: asyncio.Future = loop.create_future() + self._waiters.append((loop, waiter)) + try: + await waiter + except BaseException: + # Cancelled or timed out while queued. Leaving the future in the deque would let a + # later release() hand a permit to a coroutine that is already gone, and the permit + # would never come back. + with self._lock: + try: + self._waiters.remove((loop, waiter)) + except ValueError: + pass # already taken out of the queue: the handover was in flight + if waiter.done() and not waiter.cancelled() and waiter.exception() is None: + # The permit arrived in the same instant the wait was abandoned. It is ours and + # nobody will use it: pass it on rather than leak it. + self.release() + raise + + def release(self) -> None: + with self._lock: + while self._waiters: + loop, waiter = self._waiters.popleft() + if waiter.cancelled(): + continue + try: + # The permit is transferred, not returned: _in_flight stays as it is. + loop.call_soon_threadsafe(_settle, waiter) + except RuntimeError: + # That loop is closed — its waiter can never run. Try the next one. + continue + return + self._in_flight = max(0, self._in_flight - 1) + + async def __aenter__(self) -> _GlobalLLMLimiter: + await self.acquire() + return self + + async def __aexit__(self, *_exc: object) -> None: + self.release() + + +def _settle(waiter: asyncio.Future) -> None: + """Resolve a queued waiter, unless it went away between the handover and the callback.""" + if not waiter.done(): + waiter.set_result(None) + + +# Keyed by the resolved limit, not shared blindly: a caller that resolves a different value gets +# its own budget instead of silently resizing one that other coroutines are holding permits from. +_limiters: dict[int, _GlobalLLMLimiter] = {} +_limiters_lock = threading.Lock() + + +def _shared_limiter(limit: int) -> _GlobalLLMLimiter: + """Return the process-wide limiter for *limit*.""" + with _limiters_lock: + limiter = _limiters.get(limit) + if limiter is None: + limiter = _GlobalLLMLimiter(limit) + _limiters[limit] = limiter + return limiter def resolve_max_concurrency() -> int: @@ -924,12 +995,13 @@ async def arun_batches_detailed( **kwargs: object, ) -> BatchExecutionResult: """Execute batches concurrently and retain sanitized per-batch failures.""" - # Resolved from the environment: share one semaphore across every analyzer on this - # loop, because that is what the variable promises. An explicit argument keeps its - # documented meaning and stays local to this call — callers that pass a number are - # asking for a fan-out width, not for a share of the process-wide budget. + # Resolved from the environment: one budget for the whole process, because that is what + # the variable promises — "requests in parallel", not "requests in parallel per analyzer". + # An explicit argument keeps its documented meaning and stays local to this call: a caller + # that passes a number is asking for a fan-out width, not for a share of the budget. + sem: _GlobalLLMLimiter | asyncio.Semaphore if max_concurrency is None: - sem = _shared_semaphore(resolve_max_concurrency()) + sem = _shared_limiter(resolve_max_concurrency()) else: sem = asyncio.Semaphore(max_concurrency) diff --git a/tests/nodes/test_llm_analyzer_base.py b/tests/nodes/test_llm_analyzer_base.py index b36db626..bea0b743 100644 --- a/tests/nodes/test_llm_analyzer_base.py +++ b/tests/nodes/test_llm_analyzer_base.py @@ -18,7 +18,10 @@ from __future__ import annotations import asyncio +import concurrent.futures +import contextlib import json +import threading from unittest.mock import AsyncMock, MagicMock, patch import httpx @@ -39,6 +42,7 @@ LLMAnalyzerBase, LLMFinding, LLMRuntimeLimitError, + _shared_limiter, chunk_file_by_lines, estimate_tokens, findings_in_range, @@ -46,7 +50,7 @@ number_lines, resolve_max_concurrency, ) -from skillspector.llm_utils import AgentCLIChatModel, StructuredOutputParseError +from skillspector.llm_utils import AgentCLIChatModel, StructuredOutputParseError, run_async from skillspector.models import Finding from skillspector.nodes.meta_analyzer import ( LLMMetaAnalyzer, @@ -2751,3 +2755,108 @@ async def test_explicit_argument_still_bounds_only_its_own_call(self, monkeypatc await analyzer.arun_batches_detailed(batches, max_concurrency=2) assert peak[0] == 2 + + +class TestConcurrencyIsGlobalAcrossLoops: + """The same bound, exercised the way the graph actually runs. + + The tests above gather two analyzers on one artificial event loop, and that is not the shape + of production: every analyzer node is a *synchronous* function that reaches ``run_async()``, + which calls ``asyncio.run()`` — a brand-new loop each time, on a brand-new thread when a loop + is already running. Anything keyed by the running loop is therefore keyed by the analyzer, + and a same-loop test cannot tell the two apart. This class reproduces the real execution + model: N analyzers, N threads, N loops, one process-wide counter. + """ + + MODEL = "gpt-4o-mini" + + @staticmethod + def _counting_analyzer(state: dict, lock: threading.Lock) -> LLMAnalyzerBase: + analyzer = LLMAnalyzerBase( + base_prompt="test", model=TestConcurrencyIsGlobalAcrossLoops.MODEL + ) + + async def _invoke(*_args, **_kwargs): + # The counter is shared across threads now, so it needs a real lock: reading a peak + # through a data race would make this test lie in whichever direction was convenient. + with lock: + state["live"] += 1 + state["peak"] = max(state["peak"], state["live"]) + # Long enough that every thread is inside a request at the same time if the bound + # does not hold. Without the sleep a serial execution and a bounded one look alike. + await asyncio.sleep(0.05) + with lock: + state["live"] -= 1 + return LLMAnalysisResult(findings=[]) + + analyzer._structured_llm.ainvoke = AsyncMock(side_effect=_invoke) + return analyzer + + @staticmethod + def _run_like_a_node(analyzer: LLMAnalyzerBase, batches: list[Batch]) -> None: + """Exactly what an analyzer node does: a sync call into ``run_async``.""" + run_async(analyzer.arun_batches_detailed(batches)) + + @pytest.mark.parametrize("limit", [1, 2]) + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + def test_three_nodes_on_three_loops_share_one_budget(self, limit, monkeypatch) -> None: + monkeypatch.setenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", str(limit)) + state = {"live": 0, "peak": 0} + lock = threading.Lock() + analyzers = [self._counting_analyzer(state, lock) for _ in range(3)] + batches = [Batch(file_path="a.py", content="a"), Batch(file_path="b.py", content="b")] + + with concurrent.futures.ThreadPoolExecutor(max_workers=len(analyzers)) as pool: + futures = [pool.submit(self._run_like_a_node, a, list(batches)) for a in analyzers] + for f in futures: + f.result() + + assert state["peak"] <= limit, ( + f"{state['peak']} requests were in flight with the limit set to {limit}: each node " + "runs on its own event loop, so a per-loop bound is a per-analyzer bound" + ) + assert state["live"] == 0, "a permit was never released" + + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + def test_the_bound_is_a_ceiling_not_a_serialization(self, monkeypatch) -> None: + """A knob that always means 1 is as wrong as one that means nothing.""" + monkeypatch.setenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", "3") + state = {"live": 0, "peak": 0} + lock = threading.Lock() + analyzers = [self._counting_analyzer(state, lock) for _ in range(3)] + batches = [Batch(file_path="a.py", content="a")] + + with concurrent.futures.ThreadPoolExecutor(max_workers=len(analyzers)) as pool: + for f in [pool.submit(self._run_like_a_node, a, list(batches)) for a in analyzers]: + f.result() + + assert state["peak"] == 3, ( + f"peak was {state['peak']} with a limit of 3 across three nodes: the bound has " + "become a queue" + ) + + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + def test_a_cancelled_waiter_does_not_strand_its_permit(self, monkeypatch) -> None: + """A permit handed to a coroutine that has gone away must come back. + + This is the failure mode that does not announce itself: the count never recovers, and + every later request waits for a slot that no longer exists. It looks like a hang, hours + after the cancellation that caused it. + """ + monkeypatch.setenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", "1") + + async def _exercise() -> int: + limiter = _shared_limiter(1) + await limiter.acquire() + queued = asyncio.ensure_future(limiter.acquire()) + await asyncio.sleep(0) + queued.cancel() + with contextlib.suppress(asyncio.CancelledError): + await queued + limiter.release() + # The permit must be free again: a second acquire has to succeed immediately. + await asyncio.wait_for(limiter.acquire(), timeout=1.0) + limiter.release() + return limiter._in_flight + + assert asyncio.run(_exercise()) == 0, "the limiter leaked a permit"