From a75b47a6ba01609dc8ef64a7d1485c8a74f5a13d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 19 Sep 2026 02:07:05 +0000 Subject: [PATCH 1/5] Detect Gunicorn config thread-pool worker mutations at import time Extend the gunicorn.conf.py AST scanner to treat ThreadPool.map and ThreadPoolExecutor submit().result() callbacks like threading.Thread.start() when they mutate workers during config import. Co-authored-by: Alexander Wagner --- config.py | 97 +++++++++++++++++++ .../test_gunicorn_indirect_workers_bypass.py | 2 + 2 files changed, 99 insertions(+) diff --git a/config.py b/config.py index 5794612..e08fe87 100644 --- a/config.py +++ b/config.py @@ -3786,6 +3786,72 @@ def _thread_constructor_has_mutating_target( ) +_THREAD_POOL_CLASS_NAMES = frozenset({"ThreadPool", "ThreadPoolExecutor"}) + + +def _call_is_thread_pool_class_constructor(call, operator_bindings): + """Return True when a call constructs ThreadPool or ThreadPoolExecutor.""" + if not isinstance(call, ast.Call): + return False + func = call.func + reference_line = getattr(call, "lineno", 0) + if isinstance(func, ast.Name) and func.id in _THREAD_POOL_CLASS_NAMES: + alias_events = operator_bindings[43] if len(operator_bindings) > 43 else {} + if alias_events: + return _imported_alias_is_active(alias_events, func.id, reference_line) + return True + if isinstance(func, ast.Attribute) and func.attr in _THREAD_POOL_CLASS_NAMES: + module_events = operator_bindings[44] if len(operator_bindings) > 44 else {} + return _module_alias_active_at_line( + func.value, + set(), + module_events, + reference_line, + ) + return False + + +def _call_is_thread_pool_map_mutation(call, operator_bindings): + """Return True when ThreadPool.map executes a workers-mutating callback.""" + if not isinstance(call, ast.Call): + return False + func = call.func + if not (isinstance(func, ast.Attribute) and func.attr == "map"): + return False + receiver = func.value + if not ( + isinstance(receiver, ast.Call) + and _call_is_thread_pool_class_constructor(receiver, operator_bindings) + ): + return False + if not call.args: + return False + if isinstance(call.args[0], ast.Lambda): + return _lambda_mutates_workers(call.args[0], operator_bindings) + return _expression_is_mutating_lazy_iterator(call.args[0], operator_bindings) + + +def _call_is_executor_submit_result_mutation(call, operator_bindings): + """Return True when submit(...).result() runs a workers-mutating callback.""" + if not isinstance(call, ast.Call): + return False + func = call.func + if not (isinstance(func, ast.Attribute) and func.attr == "result"): + return False + submit_call = func.value + if not isinstance(submit_call, ast.Call): + return False + submit_func = submit_call.func + if not (isinstance(submit_func, ast.Attribute) and submit_func.attr == "submit"): + return False + if not submit_call.args: + return False + target = submit_call.args[0] + if isinstance(target, ast.Lambda): + return _lambda_mutates_workers(target, operator_bindings) + return False + + def _expression_starts_mutating_thread( expr, active_thread_names, @@ -3946,6 +4012,10 @@ def _call_consumes_mutating_lazy_iterator(call, operator_bindings): """Detect eager builtin consumers of direct or saved risky map/filter iterators.""" if not isinstance(call, ast.Call): return False + if _call_is_thread_pool_map_mutation(call, operator_bindings): + return True + if _call_is_executor_submit_result_mutation(call, operator_bindings): + return True if _attribute_call_consumes_mutating_lazy_iterator(call, operator_bindings): return True if _call_is_collections_lazy_consumer(call, operator_bindings): @@ -5845,6 +5915,33 @@ def _scan_gunicorn_config_worker_details(tree): *operator_bindings, mutating_generator_alias_events, ) + thread_pool_class_alias_events = _collect_imported_name_alias_events( + tree, + "multiprocessing.pool", + {"ThreadPool"}, + ) + thread_pool_class_alias_events.update( + _collect_imported_name_alias_events( + tree, + "concurrent.futures", + {"ThreadPoolExecutor"}, + ) + ) + thread_pool_module_alias_events = _collect_imported_module_alias_events( + tree, + "multiprocessing.pool", + ) + thread_pool_module_alias_events.update( + _collect_imported_module_alias_events( + tree, + "concurrent.futures", + ) + ) + operator_bindings = ( + *operator_bindings, + thread_pool_class_alias_events, + thread_pool_module_alias_events, + ) dict_shadow_line = operator_bindings[21] if len(operator_bindings) > 21 else None dict_subclass_names = _collect_dict_subclass_names(tree.body, dict_shadow_line) class_targets = _collect_class_side_effect_targets(tree, operator_bindings) diff --git a/tests/test_gunicorn_indirect_workers_bypass.py b/tests/test_gunicorn_indirect_workers_bypass.py index 08c455c..c6f1261 100644 --- a/tests/test_gunicorn_indirect_workers_bypass.py +++ b/tests/test_gunicorn_indirect_workers_bypass.py @@ -491,6 +491,8 @@ def test_cursor_unconsumed_or_rebound_generator_stays_static(tmp_path, config_co "workers = 1\nif True:\n def g():\n yield from map(lambda _: globals().update({'workers': 4}), [1])\nlist(g())\n", "workers = 1\nimport threading\ndef set_workers():\n global workers\n workers = 4\nt = threading.Thread(target=set_workers)\nt.start(); t.join()\n", "workers = 1\nfrom threading import Thread as T\nt = T(target=lambda: globals().update({'workers': 4}))\nt.start(); t.join()\n", + "workers = 1\nfrom concurrent.futures import ThreadPoolExecutor\nwith ThreadPoolExecutor(1) as ex:\n ex.submit(lambda: globals().update({'workers': 4})).result()\n", + "workers = 1\nfrom multiprocessing.pool import ThreadPool\nThreadPool(1).map(lambda _: globals().update({'workers': 4}), [1])\n", ], ) def test_codex_review_alias_and_execution_gaps_are_dynamic(tmp_path, config_content): From 2105e33a8e5dd8e1bc5f26450cca2f5a1f5b02ff Mon Sep 17 00:00:00 2001 From: Alexander Wagner Date: Sat, 19 Sep 2026 21:19:17 +0200 Subject: [PATCH 2/5] Fix Codex thread-pool scanner findings --- config.py | 447 +++++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 388 insertions(+), 59 deletions(-) diff --git a/config.py b/config.py index e08fe87..21d79ee 100644 --- a/config.py +++ b/config.py @@ -3787,71 +3787,399 @@ def _thread_constructor_has_mutating_target( _THREAD_POOL_CLASS_NAMES = frozenset({"ThreadPool", "ThreadPoolExecutor"}) +_THREAD_POOL_DIRECT_IMPORT_KINDS = { + ("multiprocessing.pool", "ThreadPool"): "pool", + ("concurrent.futures", "ThreadPoolExecutor"): "executor", +} +_THREAD_POOL_MODULE_IMPORT_KINDS = { + "multiprocessing.pool": "pool", + "concurrent.futures": "executor", +} -def _call_is_thread_pool_class_constructor(call, operator_bindings): - """Return True when a call constructs ThreadPool or ThreadPoolExecutor.""" +def _record_thread_pool_alias_event(events, name, state, line): + """Record a source-ordered thread-pool alias binding or rebinding.""" + if state is not None: + events.setdefault(name, []).append((line, state)) + elif name in events: + events[name].append((line, False)) + + +def _collect_thread_pool_alias_events(tree): + """Track supported thread-pool imports without losing colliding aliases.""" + class_events = {} + module_events = {} + for node in tree.body: + line = getattr(node, "lineno", 0) + if isinstance(node, ast.ImportFrom): + for imported in node.names: + name = imported.asname or imported.name + kind = _THREAD_POOL_DIRECT_IMPORT_KINDS.get( + (node.module, imported.name) + ) + _record_thread_pool_alias_event( + class_events, + name, + kind, + line, + ) + _record_thread_pool_alias_event( + module_events, + name, + None, + line, + ) + continue + if isinstance(node, ast.Import): + for imported in node.names: + bound_name = imported.asname or imported.name.split(".", 1)[0] + kind = _THREAD_POOL_MODULE_IMPORT_KINDS.get(imported.name) + module_name = imported.asname or imported.name + _record_thread_pool_alias_event( + module_events, + module_name if kind is not None else bound_name, + kind, + line, + ) + if module_name != bound_name: + _record_thread_pool_alias_event( + module_events, + bound_name, + None, + line, + ) + _record_thread_pool_alias_event( + class_events, + bound_name, + None, + line, + ) + continue + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + names = (node.name,) + else: + names = tuple(name for name, _value in _namespace_assignment_values(node)) + for name in names: + _record_thread_pool_alias_event(class_events, name, None, line) + _record_thread_pool_alias_event(module_events, name, None, line) + return class_events, module_events + + +def _thread_pool_reference_name(node): + """Return a dotted name for a simple module reference expression.""" + if isinstance(node, ast.Name): + return node.id + if isinstance(node, ast.Attribute): + parent = _thread_pool_reference_name(node.value) + if parent is not None: + return f"{parent}.{node.attr}" + return None + + +def _thread_pool_constructor_kind(call, operator_bindings): + """Return pool kind for a proven ThreadPool/ThreadPoolExecutor constructor.""" if not isinstance(call, ast.Call): - return False + return None func = call.func reference_line = getattr(call, "lineno", 0) - if isinstance(func, ast.Name) and func.id in _THREAD_POOL_CLASS_NAMES: - alias_events = operator_bindings[43] if len(operator_bindings) > 43 else {} - if alias_events: - return _imported_alias_is_active(alias_events, func.id, reference_line) - return True - if isinstance(func, ast.Attribute) and func.attr in _THREAD_POOL_CLASS_NAMES: - module_events = operator_bindings[44] if len(operator_bindings) > 44 else {} - return _module_alias_active_at_line( - func.value, - set(), - module_events, + class_events = operator_bindings[43] if len(operator_bindings) > 43 else {} + module_events = operator_bindings[44] if len(operator_bindings) > 44 else {} + if isinstance(func, ast.Name): + state = _binding_state_at_line( + class_events, + func.id, reference_line, ) - return False + return state if state in {"pool", "executor"} else None + if not isinstance(func, ast.Attribute): + return None + module_name = _thread_pool_reference_name(func.value) + if module_name is None: + return None + state = _binding_state_at_line( + module_events, + module_name, + reference_line, + ) + expected_name = { + "pool": "ThreadPool", + "executor": "ThreadPoolExecutor", + }.get(state) + return state if expected_name == func.attr else None + + +def _call_is_thread_pool_class_constructor(call, operator_bindings): + """Return True when a call constructs ThreadPool or ThreadPoolExecutor.""" + return _thread_pool_constructor_kind(call, operator_bindings) is not None + + +def _thread_pool_instance_constructor_from_value( + value, + operator_bindings, + events, +): + """Resolve an expression to the constructor that produced a tracked pool.""" + if isinstance(value, ast.Call) and _call_is_thread_pool_class_constructor( + value, + operator_bindings, + ): + return value + if isinstance(value, ast.Name): + state = _binding_state_at_line( + events, + value.id, + getattr(value, "lineno", 0), + ) + if isinstance(state, ast.Call): + return state + return None + + +def _record_thread_pool_target_binding(target, constructor, events, line): + """Bind a with-target or assignment target to one proven pool instance.""" + if isinstance(target, ast.Name): + events.setdefault(target.id, []).append((line, constructor)) + return + if isinstance(target, (ast.Tuple, ast.List)): + for element in target.elts: + _record_thread_pool_target_binding( + element, + constructor, + events, + line, + ) + + +def _scan_thread_pool_instance_alias_events( + statements, + operator_bindings, + events, + *, + conditional=False, +): + """Track saved and context-managed pool instances in source order.""" + for node in statements: + line = getattr(node, "lineno", 0) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + if not conditional: + _record_thread_pool_alias_event(events, node.name, None, line) + continue + if isinstance(node, ast.With): + for item in node.items: + if item.optional_vars is None: + continue + constructor = _thread_pool_instance_constructor_from_value( + item.context_expr, + operator_bindings, + events, + ) + if constructor is not None: + _record_thread_pool_target_binding( + item.optional_vars, + constructor, + events, + line, + ) + elif not conditional and isinstance(item.optional_vars, ast.Name): + _record_thread_pool_alias_event( + events, + item.optional_vars.id, + None, + line, + ) + for name, value in _namespace_assignment_values(node): + constructor = _thread_pool_instance_constructor_from_value( + value, + operator_bindings, + events, + ) + if constructor is not None: + events.setdefault(name, []).append((line, constructor)) + elif not conditional: + _record_thread_pool_alias_event(events, name, None, line) + for nested in _compound_statement_blocks(node): + _scan_thread_pool_instance_alias_events( + nested, + operator_bindings, + events, + conditional=True, + ) + + +def _collect_thread_pool_instance_alias_events(tree, operator_bindings): + """Collect names that hold proven thread-pool instances.""" + events = {} + _scan_thread_pool_instance_alias_events( + tree.body, + operator_bindings, + events, + ) + return events + + +def _thread_pool_receiver_constructor(receiver, operator_bindings): + """Resolve a map/submit receiver to its proven pool constructor.""" + if isinstance(receiver, ast.Call) and _call_is_thread_pool_class_constructor( + receiver, + operator_bindings, + ): + return receiver + if not isinstance(receiver, ast.Name): + return None + events = operator_bindings[45] if len(operator_bindings) > 45 else {} + state = _binding_state_at_line( + events, + receiver.id, + getattr(receiver, "lineno", 0), + ) + return state if isinstance(state, ast.Call) else None + + +def _thread_pool_callback_mutates_workers(callback, operator_bindings): + """Return whether one pool callback can mutate module workers.""" + if isinstance(callback, ast.Lambda): + return _lambda_mutates_workers(callback, operator_bindings) + mutator_names = operator_bindings[46] if len(operator_bindings) > 46 else set() + return isinstance(callback, ast.Name) and callback.id in mutator_names + + +def _thread_pool_constructor_initializer(constructor, operator_bindings): + """Return a constructor initializer expression, if one is present.""" + if not isinstance(constructor, ast.Call): + return None + for keyword in constructor.keywords: + if keyword.arg == "initializer": + return keyword.value + kind = _thread_pool_constructor_kind(constructor, operator_bindings) + index = {"pool": 1, "executor": 2}.get(kind) + if index is not None and len(constructor.args) > index: + return constructor.args[index] + return None + + +def _thread_pool_constructor_has_mutating_initializer( + constructor, + operator_bindings, +): + """Return True when a proven pool initializer mutates workers.""" + initializer = _thread_pool_constructor_initializer( + constructor, + operator_bindings, + ) + return ( + initializer is not None + and _thread_pool_callback_mutates_workers( + initializer, + operator_bindings, + ) + ) + + +def _thread_pool_map_arguments(call): + """Return callback and iterable expressions supplied to pool.map.""" + callback = call.args[0] if call.args else None + if callback is None: + callback = next( + ( + keyword.value + for keyword in call.keywords + if keyword.arg in {"func", "fn"} + ), + None, + ) + iterables = list(call.args[1:]) + iterables.extend( + keyword.value + for keyword in call.keywords + if keyword.arg == "iterable" + ) + return callback, iterables def _call_is_thread_pool_map_mutation(call, operator_bindings): - """Return True when ThreadPool.map executes a workers-mutating callback.""" + """Return True when a proven pool.map performs a workers mutation.""" if not isinstance(call, ast.Call): return False func = call.func if not (isinstance(func, ast.Attribute) and func.attr == "map"): return False - receiver = func.value - if not ( - isinstance(receiver, ast.Call) - and _call_is_thread_pool_class_constructor(receiver, operator_bindings) - ): + constructor = _thread_pool_receiver_constructor( + func.value, + operator_bindings, + ) + if constructor is None: return False - if not call.args: + callback, iterables = _thread_pool_map_arguments(call) + if callback is None or not iterables: return False - if isinstance(call.args[0], ast.Lambda): - return _lambda_mutates_workers(call.args[0], operator_bindings) - return _expression_is_mutating_lazy_iterator(call.args[0], operator_bindings) + if _thread_pool_constructor_has_mutating_initializer( + constructor, + operator_bindings, + ): + return True + if _thread_pool_callback_mutates_workers(callback, operator_bindings): + return True + return any( + _expression_is_mutating_lazy_iterator( + iterable, + operator_bindings, + ) + for iterable in iterables + ) -def _call_is_executor_submit_result_mutation(call, operator_bindings): - """Return True when submit(...).result() runs a workers-mutating callback.""" +def _call_is_thread_pool_submit_mutation(call, operator_bindings): + """Return True when ThreadPoolExecutor.submit schedules a risky callback.""" if not isinstance(call, ast.Call): return False func = call.func - if not (isinstance(func, ast.Attribute) and func.attr == "result"): - return False - submit_call = func.value - if not isinstance(submit_call, ast.Call): + if not (isinstance(func, ast.Attribute) and func.attr == "submit"): return False - submit_func = submit_call.func - if not (isinstance(submit_func, ast.Attribute) and submit_func.attr == "submit"): + constructor = _thread_pool_receiver_constructor( + func.value, + operator_bindings, + ) + if ( + constructor is None + or _thread_pool_constructor_kind(constructor, operator_bindings) + != "executor" + ): return False - if not submit_call.args: + target = call.args[0] if call.args else next( + ( + keyword.value + for keyword in call.keywords + if keyword.arg == "fn" + ), + None, + ) + if target is None: return False - target = submit_call.args[0] - if isinstance(target, ast.Lambda): - return _lambda_mutates_workers(target, operator_bindings) - return False + return ( + _thread_pool_constructor_has_mutating_initializer( + constructor, + operator_bindings, + ) + or _thread_pool_callback_mutates_workers( + target, + operator_bindings, + ) + ) +def _call_is_thread_pool_constructor_initializer_mutation( + call, + operator_bindings, +): + """Detect eager multiprocessing ThreadPool initializer side effects.""" + return ( + _thread_pool_constructor_kind(call, operator_bindings) == "pool" + and _thread_pool_constructor_has_mutating_initializer( + call, + operator_bindings, + ) + ) + def _expression_starts_mutating_thread( expr, active_thread_names, @@ -4012,9 +4340,14 @@ def _call_consumes_mutating_lazy_iterator(call, operator_bindings): """Detect eager builtin consumers of direct or saved risky map/filter iterators.""" if not isinstance(call, ast.Call): return False + if _call_is_thread_pool_constructor_initializer_mutation( + call, + operator_bindings, + ): + return True if _call_is_thread_pool_map_mutation(call, operator_bindings): return True - if _call_is_executor_submit_result_mutation(call, operator_bindings): + if _call_is_thread_pool_submit_mutation(call, operator_bindings): return True if _attribute_call_consumes_mutating_lazy_iterator(call, operator_bindings): return True @@ -5915,32 +6248,24 @@ def _scan_gunicorn_config_worker_details(tree): *operator_bindings, mutating_generator_alias_events, ) - thread_pool_class_alias_events = _collect_imported_name_alias_events( - tree, - "multiprocessing.pool", - {"ThreadPool"}, - ) - thread_pool_class_alias_events.update( - _collect_imported_name_alias_events( - tree, - "concurrent.futures", - {"ThreadPoolExecutor"}, - ) - ) - thread_pool_module_alias_events = _collect_imported_module_alias_events( - tree, - "multiprocessing.pool", + ( + thread_pool_class_alias_events, + thread_pool_module_alias_events, + ) = _collect_thread_pool_alias_events(tree) + operator_bindings = ( + *operator_bindings, + thread_pool_class_alias_events, + thread_pool_module_alias_events, ) - thread_pool_module_alias_events.update( - _collect_imported_module_alias_events( + thread_pool_instance_alias_events = ( + _collect_thread_pool_instance_alias_events( tree, - "concurrent.futures", + operator_bindings, ) ) operator_bindings = ( *operator_bindings, - thread_pool_class_alias_events, - thread_pool_module_alias_events, + thread_pool_instance_alias_events, ) dict_shadow_line = operator_bindings[21] if len(operator_bindings) > 21 else None dict_subclass_names = _collect_dict_subclass_names(tree.body, dict_shadow_line) @@ -5951,6 +6276,10 @@ def _scan_gunicorn_config_worker_details(tree): class_targets, dict_subclass_names, ) + operator_bindings = ( + *operator_bindings, + import_time_workers_mutators, + ) if _statements_start_mutating_thread( tree.body, operator_bindings, From 5795c6c5612136579653e31ba71605a98889d98e Mon Sep 17 00:00:00 2001 From: Alexander Wagner Date: Sat, 19 Sep 2026 21:19:21 +0200 Subject: [PATCH 3/5] Add regression tests for Codex PR 253 findings --- .../test_gunicorn_indirect_workers_bypass.py | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/tests/test_gunicorn_indirect_workers_bypass.py b/tests/test_gunicorn_indirect_workers_bypass.py index c6f1261..5008593 100644 --- a/tests/test_gunicorn_indirect_workers_bypass.py +++ b/tests/test_gunicorn_indirect_workers_bypass.py @@ -522,3 +522,41 @@ def test_codex_review_safe_alias_and_thread_patterns_stay_static( config_file.write_text(config_content, encoding="utf-8") assert config._workers_from_gunicorn_config_path(str(config_file)) == (1, False) +# Codex PR #253 follow-up: thread-pool alias, receiver, callback, and initializer gaps +@pytest.mark.parametrize( + "config_content", + [ + "workers = 1\nfrom multiprocessing.pool import ThreadPool as Pool\nPool(1).map(lambda _: globals().update({'workers': 4}), [1])\n", + "workers = 1\nfrom multiprocessing.pool import ThreadPool\npool = ThreadPool(1)\npool.map(lambda _: globals().update({'workers': 4}), [1])\n", + "workers = 1\nfrom multiprocessing.pool import ThreadPool\nwith ThreadPool(1) as pool:\n pool.map(lambda _: globals().update({'workers': 4}), [1])\n", + "workers = 1\nfrom multiprocessing.pool import ThreadPool\ndef set_workers(_):\n global workers\n workers = 4\nThreadPool(1).map(set_workers, [1])\n", + "workers = 1\nfrom concurrent.futures import ThreadPoolExecutor\nwith ThreadPoolExecutor(1) as ex:\n ex.submit(lambda: globals().update({'workers': 4}))\n", + "workers = 1\nfrom concurrent.futures import ThreadPoolExecutor\nwith ThreadPoolExecutor(1) as ex:\n future = ex.submit(lambda: globals().update({'workers': 4}))\n future.result()\n", + "workers = 1\nfrom multiprocessing.pool import ThreadPool\nThreadPool(1).map(lambda value: value, map(lambda _: globals().update({'workers': 4}), [1]))\n", + "workers = 1\nfrom multiprocessing.pool import ThreadPool\nThreadPool(1, initializer=lambda: globals().update({'workers': 4})).map(lambda value: value, [1])\n", + "workers = 1\nfrom concurrent.futures import ThreadPoolExecutor\nwith ThreadPoolExecutor(1, initializer=lambda: globals().update({'workers': 4})) as ex:\n ex.submit(lambda: None)\n", + "workers = 1\nfrom multiprocessing.pool import ThreadPool\nThreadPool(1).map(func=lambda _: globals().update({'workers': 4}), iterable=[1])\n", + "workers = 1\nimport multiprocessing.pool as p\np.ThreadPool(1).map(lambda _: globals().update({'workers': 4}), [1])\nimport concurrent.futures as p\n", + ], +) +def test_codex_pr253_thread_pool_followups_are_dynamic(tmp_path, config_content): + config_file = tmp_path / "gunicorn.conf.py" + config_file.write_text(config_content, encoding="utf-8") + assert config._workers_from_gunicorn_config_path(str(config_file)) == (1, True) + + +def test_codex_pr253_custom_submitter_stays_static(tmp_path): + config_file = tmp_path / "gunicorn.conf.py" + config_file.write_text( + "workers = 1\n" + "class Submitter:\n" + " def submit(self, fn):\n" + " class Future:\n" + " def result(self):\n" + " return None\n" + " return Future()\n" + "Submitter().submit(lambda: globals().update({'workers': 4})).result()\n", + encoding="utf-8", + ) + assert config._workers_from_gunicorn_config_path(str(config_file)) == (1, False) + From 29e3e722ba8fa6bf50f69d2ab523a54623bf5a40 Mon Sep 17 00:00:00 2001 From: Alexander Wagner Date: Sat, 19 Sep 2026 21:20:40 +0200 Subject: [PATCH 4/5] Remove unused thread-pool class constant --- config.py | 1 - 1 file changed, 1 deletion(-) diff --git a/config.py b/config.py index 21d79ee..58c6a85 100644 --- a/config.py +++ b/config.py @@ -3786,7 +3786,6 @@ def _thread_constructor_has_mutating_target( ) -_THREAD_POOL_CLASS_NAMES = frozenset({"ThreadPool", "ThreadPoolExecutor"}) _THREAD_POOL_DIRECT_IMPORT_KINDS = { ("multiprocessing.pool", "ThreadPool"): "pool", ("concurrent.futures", "ThreadPoolExecutor"): "executor", From a6fb45c44be40c99f223c646c591b49647fd511e Mon Sep 17 00:00:00 2001 From: Alexander Wagner Date: Sat, 19 Sep 2026 21:23:14 +0200 Subject: [PATCH 5/5] Reduce thread-pool scanner cognitive complexity --- config.py | 394 ++++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 277 insertions(+), 117 deletions(-) diff --git a/config.py b/config.py index 58c6a85..163309c 100644 --- a/config.py +++ b/config.py @@ -3804,6 +3804,69 @@ def _record_thread_pool_alias_event(events, name, state, line): events[name].append((line, False)) +def _record_thread_pool_direct_import_aliases( + node, + class_events, + module_events, + line, +): + """Record direct ThreadPool and ThreadPoolExecutor imports.""" + for imported in node.names: + name = imported.asname or imported.name + kind = _THREAD_POOL_DIRECT_IMPORT_KINDS.get( + (node.module, imported.name) + ) + _record_thread_pool_alias_event(class_events, name, kind, line) + _record_thread_pool_alias_event(module_events, name, None, line) + + +def _record_thread_pool_module_import_aliases( + node, + class_events, + module_events, + line, +): + """Record supported thread-pool module imports and alias collisions.""" + for imported in node.names: + bound_name = imported.asname or imported.name.split(".", 1)[0] + kind = _THREAD_POOL_MODULE_IMPORT_KINDS.get(imported.name) + module_name = imported.asname or imported.name + tracked_name = module_name if kind is not None else bound_name + _record_thread_pool_alias_event( + module_events, + tracked_name, + kind, + line, + ) + if module_name != bound_name: + _record_thread_pool_alias_event( + module_events, + bound_name, + None, + line, + ) + _record_thread_pool_alias_event( + class_events, + bound_name, + None, + line, + ) + + +def _thread_pool_rebound_names(node): + """Return names definitely rebound by one non-import statement.""" + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + return (node.name,) + return tuple(name for name, _value in _namespace_assignment_values(node)) + + +def _record_thread_pool_rebindings(node, class_events, module_events, line): + """Deactivate tracked thread-pool aliases rebound by one statement.""" + for name in _thread_pool_rebound_names(node): + _record_thread_pool_alias_event(class_events, name, None, line) + _record_thread_pool_alias_event(module_events, name, None, line) + + def _collect_thread_pool_alias_events(tree): """Track supported thread-pool imports without losing colliding aliases.""" class_events = {} @@ -3811,56 +3874,26 @@ def _collect_thread_pool_alias_events(tree): for node in tree.body: line = getattr(node, "lineno", 0) if isinstance(node, ast.ImportFrom): - for imported in node.names: - name = imported.asname or imported.name - kind = _THREAD_POOL_DIRECT_IMPORT_KINDS.get( - (node.module, imported.name) - ) - _record_thread_pool_alias_event( - class_events, - name, - kind, - line, - ) - _record_thread_pool_alias_event( - module_events, - name, - None, - line, - ) - continue - if isinstance(node, ast.Import): - for imported in node.names: - bound_name = imported.asname or imported.name.split(".", 1)[0] - kind = _THREAD_POOL_MODULE_IMPORT_KINDS.get(imported.name) - module_name = imported.asname or imported.name - _record_thread_pool_alias_event( - module_events, - module_name if kind is not None else bound_name, - kind, - line, - ) - if module_name != bound_name: - _record_thread_pool_alias_event( - module_events, - bound_name, - None, - line, - ) - _record_thread_pool_alias_event( - class_events, - bound_name, - None, - line, - ) - continue - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): - names = (node.name,) + _record_thread_pool_direct_import_aliases( + node, + class_events, + module_events, + line, + ) + elif isinstance(node, ast.Import): + _record_thread_pool_module_import_aliases( + node, + class_events, + module_events, + line, + ) else: - names = tuple(name for name, _value in _namespace_assignment_values(node)) - for name in names: - _record_thread_pool_alias_event(class_events, name, None, line) - _record_thread_pool_alias_event(module_events, name, None, line) + _record_thread_pool_rebindings( + node, + class_events, + module_events, + line, + ) return class_events, module_events @@ -3949,6 +3982,116 @@ def _record_thread_pool_target_binding(target, constructor, events, line): ) +def _record_thread_pool_definition_rebinding( + node, + events, + line, + *, + conditional, +): + """Handle a definition that may rebind a tracked pool instance.""" + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + return False + if not conditional: + _record_thread_pool_alias_event(events, node.name, None, line) + return True + + +def _record_thread_pool_with_alias( + item, + operator_bindings, + events, + line, + *, + conditional, +): + """Record one context-managed pool instance alias.""" + if item.optional_vars is None: + return + constructor = _thread_pool_instance_constructor_from_value( + item.context_expr, + operator_bindings, + events, + ) + if constructor is not None: + _record_thread_pool_target_binding( + item.optional_vars, + constructor, + events, + line, + ) + return + if not conditional and isinstance(item.optional_vars, ast.Name): + _record_thread_pool_alias_event( + events, + item.optional_vars.id, + None, + line, + ) + + +def _record_thread_pool_with_aliases( + node, + operator_bindings, + events, + line, + *, + conditional, +): + """Record context-managed thread-pool aliases from one statement.""" + if not isinstance(node, ast.With): + return + for item in node.items: + _record_thread_pool_with_alias( + item, + operator_bindings, + events, + line, + conditional=conditional, + ) + + +def _record_thread_pool_assignment_alias( + name, + value, + operator_bindings, + events, + line, + *, + conditional, +): + """Record one assignment to a possible thread-pool instance.""" + constructor = _thread_pool_instance_constructor_from_value( + value, + operator_bindings, + events, + ) + if constructor is not None: + events.setdefault(name, []).append((line, constructor)) + elif not conditional: + _record_thread_pool_alias_event(events, name, None, line) + + +def _record_thread_pool_assignment_aliases( + node, + operator_bindings, + events, + line, + *, + conditional, +): + """Record thread-pool aliases introduced by simple assignments.""" + for name, value in _namespace_assignment_values(node): + _record_thread_pool_assignment_alias( + name, + value, + operator_bindings, + events, + line, + conditional=conditional, + ) + + def _scan_thread_pool_instance_alias_events( statements, operator_bindings, @@ -3959,43 +4102,27 @@ def _scan_thread_pool_instance_alias_events( """Track saved and context-managed pool instances in source order.""" for node in statements: line = getattr(node, "lineno", 0) - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): - if not conditional: - _record_thread_pool_alias_event(events, node.name, None, line) + if _record_thread_pool_definition_rebinding( + node, + events, + line, + conditional=conditional, + ): continue - if isinstance(node, ast.With): - for item in node.items: - if item.optional_vars is None: - continue - constructor = _thread_pool_instance_constructor_from_value( - item.context_expr, - operator_bindings, - events, - ) - if constructor is not None: - _record_thread_pool_target_binding( - item.optional_vars, - constructor, - events, - line, - ) - elif not conditional and isinstance(item.optional_vars, ast.Name): - _record_thread_pool_alias_event( - events, - item.optional_vars.id, - None, - line, - ) - for name, value in _namespace_assignment_values(node): - constructor = _thread_pool_instance_constructor_from_value( - value, - operator_bindings, - events, - ) - if constructor is not None: - events.setdefault(name, []).append((line, constructor)) - elif not conditional: - _record_thread_pool_alias_event(events, name, None, line) + _record_thread_pool_with_aliases( + node, + operator_bindings, + events, + line, + conditional=conditional, + ) + _record_thread_pool_assignment_aliases( + node, + operator_bindings, + events, + line, + conditional=conditional, + ) for nested in _compound_statement_blocks(node): _scan_thread_pool_instance_alias_events( nested, @@ -4335,32 +4462,8 @@ def _call_is_collections_lazy_consumer(call, operator_bindings): ) -def _call_consumes_mutating_lazy_iterator(call, operator_bindings): - """Detect eager builtin consumers of direct or saved risky map/filter iterators.""" - if not isinstance(call, ast.Call): - return False - if _call_is_thread_pool_constructor_initializer_mutation( - call, - operator_bindings, - ): - return True - if _call_is_thread_pool_map_mutation(call, operator_bindings): - return True - if _call_is_thread_pool_submit_mutation(call, operator_bindings): - return True - if _attribute_call_consumes_mutating_lazy_iterator(call, operator_bindings): - return True - if _call_is_collections_lazy_consumer(call, operator_bindings): - if not call.args: - return False - return _expression_is_mutating_lazy_iterator( - call.args[0], - operator_bindings, - ) - if not isinstance(call.func, ast.Name): - return False - name = call.func.id - consumers = { +_EAGER_LAZY_ITERATOR_CONSUMERS = frozenset( + { "list", "tuple", "set", @@ -4372,15 +4475,55 @@ def _call_consumes_mutating_lazy_iterator(call, operator_bindings): "next", "sorted", } - if name not in consumers or not _builtin_consumer_is_active( - name, - call, +) + + +def _call_is_special_lazy_iterator_consumer(call, operator_bindings): + """Return True for non-builtin eager consumers handled specially.""" + return ( + _call_is_thread_pool_constructor_initializer_mutation( + call, + operator_bindings, + ) + or _call_is_thread_pool_map_mutation(call, operator_bindings) + or _call_is_thread_pool_submit_mutation(call, operator_bindings) + or _attribute_call_consumes_mutating_lazy_iterator( + call, + operator_bindings, + ) + ) + + +def _collections_call_consumes_mutating_lazy_iterator( + call, + operator_bindings, +): + """Return whether a proven collections consumer eagerly drains a risky source.""" + if not _call_is_collections_lazy_consumer(call, operator_bindings): + return False + return bool(call.args) and _expression_is_mutating_lazy_iterator( + call.args[0], operator_bindings, + ) + + +def _named_builtin_consumes_mutating_lazy_iterator(call, operator_bindings): + """Detect eager builtin consumers of a risky lazy iterator.""" + if not isinstance(call.func, ast.Name): + return False + name = call.func.id + if ( + name not in _EAGER_LAZY_ITERATOR_CONSUMERS + or not _builtin_consumer_is_active( + name, + call, + operator_bindings, + ) ): return False - if name in {"sorted", "max", "min"} and _key_lambda_mutates_workers( - call, - operator_bindings, + if ( + name in {"sorted", "max", "min"} + and _key_lambda_mutates_workers(call, operator_bindings) ): return True if not call.args: @@ -4393,6 +4536,23 @@ def _call_consumes_mutating_lazy_iterator(call, operator_bindings): ) +def _call_consumes_mutating_lazy_iterator(call, operator_bindings): + """Detect eager consumers of direct or saved risky map/filter iterators.""" + if not isinstance(call, ast.Call): + return False + return ( + _call_is_special_lazy_iterator_consumer(call, operator_bindings) + or _collections_call_consumes_mutating_lazy_iterator( + call, + operator_bindings, + ) + or _named_builtin_consumes_mutating_lazy_iterator( + call, + operator_bindings, + ) + ) + + def _lazy_iterator_assignment_is_mutating(value, active_names, operator_bindings): """Return whether an assignment stores a risky lazy iterator.""" return _expression_is_mutating_lazy_iterator(