diff --git a/scripts/cpu-perf-check b/scripts/cpu-perf-check new file mode 100755 index 00000000..2a15a6d0 --- /dev/null +++ b/scripts/cpu-perf-check @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +# tractor: distributed structured concurrency. +# Copyright 2018-eternity Tyler Goodlet. +# +# SPDX-License-Identifier: AGPL-3.0-or-later +''' +`cpu-perf-check` — sustained-load CPU throttle detector. + +Idle freq snapshots LIE. A laptop can read +`governor=performance`, `EPP=performance`, +`platform_profile=performance`, `scaling_max_freq=` +and momentarily clock a P-core at 5GHz — while a +firmware/EC power cap (AMD PPT/STAPM and friends) clamps +the whole package to ~1.5GHz the instant a sustained +multi-core load lands. That throttle masquerades as a +`trio`-backend test *regression*: a wave of `fail_after` / +`TooSlowError` / `Cancelled(source='deadline')` deadline +misses on spawn-heavy tests, on byte-identical code that +was green yesterday. + +The existing `tests/conftest.py:cpu_scaling_factor()` only +reads STATIC `scaling_max_freq` vs `*_pstate_max_freq`, so +it returns `1.0` (no throttle) during exactly this failure +— it can't see the cap. This script complements it by +BURNING every core for a few seconds and sampling the +ACHIEVED `scaling_cur_freq`, which is the only thing that +exposes the clamp. + +Exit code: `0` if sustained perf looks restored, `1` if +throttled — so it gates a test run: + + py313/bin/python scripts/cpu-perf-check && pytest tests/ ... + +Tunables (env-overridable): + CPU_PERF_SECS load duration (default 4.0) + CPU_PERF_HEALTHY_FRAC sustained/max floor (default 0.45) + +''' +from __future__ import annotations +import glob +import os +import time +import multiprocessing as mp + + +def _read(path: str) -> str | None: + try: + with open(path) as f: + return f.read().strip() + except OSError: + return None + + +def _cur_freqs_mhz() -> list[int]: + out: list[int] = [] + for f in glob.glob( + '/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_cur_freq' + ): + if (v := _read(f)): + out.append(int(v) // 1000) + return out + + +def _pkg_max_mhz() -> int: + ''' + Highest per-core ceiling across the package — the + P-core max on hybrid parts. + + ''' + mxs: list[int] = [] + for f in glob.glob( + '/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_max_freq' + ): + if (v := _read(f)): + mxs.append(int(v) // 1000) + return max(mxs) if mxs else 0 + + +def _burn(stop: float) -> None: + # fixed-width (64-bit-masked) arithmetic keeps this a steady + # ALU load; an unmasked `x` grows ~x**2/iter into a huge bigint, + # degenerating into noisy alloc/mul-bound work (and needless + # memory) across N procs. + x: int = 1 + while time.perf_counter() < stop: + x = (x + (x * x ^ 0x5)) & 0xFFFF_FFFF_FFFF_FFFF + + +def main( + secs: float = float(os.environ.get('CPU_PERF_SECS', 4.0)), + # sustained aggregate must clear this fraction of the + # package max-freq ceiling. Throttled (~1.5GHz of 5.1GHz) + # ~= 0.29; a healthy all-core load easily clears 0.5. + healthy_frac: float = float( + os.environ.get('CPU_PERF_HEALTHY_FRAC', 0.45) + ), +) -> int: + if not glob.glob('/sys/devices/system/cpu/cpu0/cpufreq'): + print('no cpufreq sysfs (non-linux?) — skipping, assume OK') + return 0 + + b: str = '/sys/devices/system/cpu/cpu0/cpufreq/' + pkg_max: int = _pkg_max_mhz() + print('=== static knobs (ALL can read fine while throttled) ===') + print(f' governor : {_read(b + "scaling_governor")}') + print(f' EPP : {_read(b + "energy_performance_preference")}') + print(f' platform_profile : ' + f'{_read("/sys/firmware/acpi/platform_profile")}') + print(f' pkg max freq : {pkg_max} MHz') + + ncpu: int = os.cpu_count() or 1 + print(f'\n=== sustained {ncpu}-core load ({secs:.0f}s) — the real test ===') + stop: float = time.perf_counter() + secs + procs = [ + mp.Process(target=_burn, args=(stop,), daemon=True) + for _ in range(ncpu) + ] + for p in procs: + p.start() + + # skip the initial ~0.6s ramp, then sample steady-state. + # `try/finally` so a Ctrl-C / sampling error still reaps the + # burners instead of leaving stray CPU hogs (daemon=True is the + # backstop should we exit abnormally before the join). + samples: list[int] = [] + try: + time.sleep(0.6) + while time.perf_counter() < stop - 0.2: + if (fr := _cur_freqs_mhz()): + samples.append(sum(fr) // len(fr)) + time.sleep(0.3) + finally: + for p in procs: + p.terminate() + for p in procs: + p.join() + + if not (samples and pkg_max): + print(' could not sample cur freq — assume OK') + return 0 + + sustained: int = sum(samples) // len(samples) + frac: float = sustained / pkg_max + print(f' aggregate cur-freq samples: {samples}') + print(f' sustained avg : {sustained} MHz ' + f'({frac * 100:.0f}% of {pkg_max} MHz max)') + + if frac < healthy_frac: + print( + f'\n ❌ THROTTLED — sustained {sustained}MHz is only ' + f'{frac * 100:.0f}% of max (< {healthy_frac * 100:.0f}%).\n' + f' Power cap (PPT/STAPM) still engaged. Fixes:\n' + f' - bounce /sys/firmware/acpi/platform_profile\n' + f' (balanced -> performance)\n' + f' - unplug/replug USB-C to re-negotiate PD\n' + f' - ryzenadj to lift STAPM/PPT\n' + f' - else reboot\n' + f' Do NOT bump test budgets — the box is slow, not the code.' + ) + return 1 + + print( + f'\n ✅ PERF OK — sustained {sustained}MHz holds ' + f'{frac * 100:.0f}% of max. Cap looks lifted; safe to run tests.' + ) + return 0 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/tests/conftest.py b/tests/conftest.py index 5d5ce803..db2528f6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -134,6 +134,150 @@ def cpu_scaling_factor() -> float: return factor +# session-cached sustained-load throttle multiplier — measured +# once (lazily) on the first `cpu_perf_headroom()` call. `None` +# = not-yet-measured. +_sustained_headroom: float|None = None + + +def _measure_sustained_headroom( + secs: float = 0.9, + # a healthy all-core sustained clock holds AT/ABOVE this + # fraction of the package single-core max ceiling (boost sags + # under full multi-core load even un-throttled, but not far); + # at/above it we assume no throttle and return 1.0. + throttle_gate: float = 0.6, + max_headroom: float = 4., +) -> float: + ''' + One-shot all-core burn returning a latency multiplier + (>= 1.0) that reflects *sustained-load* CPU throttle. + + Catches the firmware/EC power-cap clamp (AMD PPT/STAPM & + friends) that pins achieved `scaling_cur_freq` to a fraction + of the ceiling under multi-core load while EVERY static knob + (`governor`, `scaling_max_freq`, `EPP`, `platform_profile`) + still reads "full performance". That cap is INVISIBLE to + `cpu_scaling_factor()` and is the gremlin behind mass `trio` + deadline-miss failures on byte-identical code — see + `scripts/cpu-perf-check`. + + Best-effort: returns 1.0 on non-linux / missing sysfs / any + error so it can never break a test run. + + ''' + import glob + import multiprocessing as mp + + def _read_mhz(path: str) -> int|None: + try: + with open(path) as f: + return int(f.read()) // 1000 + except OSError: + return None + + try: + maxs: list[int] = [ + v for f in glob.glob( + '/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_max_freq' + ) + if (v := _read_mhz(f)) is not None + ] + pkg_max: int = max(maxs) if maxs else 0 + if not pkg_max: + return 1. + + def _burn(stop: float) -> None: + # fixed-width (64-bit-masked) arithmetic keeps this a + # steady ALU load; an unmasked `x` grows ~x**2/iter into + # a huge bigint, degenerating into noisy alloc/mul-bound + # work (and needless memory) across N procs. + x: int = 1 + while time.perf_counter() < stop: + x = (x + (x * x ^ 0x5)) & 0xFFFF_FFFF_FFFF_FFFF + + # explicit `fork` ctx so we're immune to whatever global + # mp start-method tractor/the suite may have set (`spawn` + # would re-exec + re-import 24x — slow and pointless here). + ctx = mp.get_context('fork') + ncpu: int = os.cpu_count() or 1 + stop: float = time.perf_counter() + secs + procs = [ + ctx.Process(target=_burn, args=(stop,), daemon=True) + for _ in range(ncpu) + ] + for p in procs: + p.start() + + # skip the ~0.4s boost window so we sample the steady + # state AFTER any power-cap has engaged. + samples: list[int] = [] + time.sleep(0.4) + while time.perf_counter() < stop - 0.1: + curs: list[int] = [ + v for f in glob.glob( + '/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_cur_freq' + ) + if (v := _read_mhz(f)) is not None + ] + if curs: + samples.append(sum(curs) // len(curs)) + time.sleep(0.15) + for p in procs: + p.join() + + if not samples: + return 1. + frac: float = (sum(samples) // len(samples)) / pkg_max + # below the gate we read it as a power-cap throttle. The + # spawn/IPC/fork-bound work these budgets guard slows ~1:1 + # with the achieved-vs-max freq ratio, so compensate by the + # FULL inverse fraction (a boost-discounted factor + # under-shoots and still trips the marginal cases). + if frac >= throttle_gate: + return 1. + # a 0/parked-core freq read would `ZeroDivisionError` the + # inverse below — silently swallowed by the outer `except` + # into a 1.0 (no-throttle), defeating the probe on exactly + # the broken box it should flag; read 0 as max throttle. + if frac <= 0: + return max_headroom + return min(max_headroom, 1. / frac) + + except Exception: + return 1. + + +def cpu_perf_headroom() -> float: + ''' + Latency-headroom multiplier (>= 1.0) covering BOTH cpu-perf + throttle classes — multiply a test's deadline by it, e.g. + `timeout *= cpu_perf_headroom()`: + + - static cpu-freq scaling — via `cpu_scaling_factor()` + (governor/policy lowered the `scaling_max_freq` ceiling). + + - sustained-load power-cap throttle — via + `_measure_sustained_headroom()` (firmware/EC PPT/STAPM + clamps achieved freq under load while every static knob + reads "performance"; INVISIBLE to the static check). This + is the gremlin behind mass `trio` deadline-miss failures + on unchanged code — see + `ai/conc-anal/trio_033_cancel_cascade_slowdown_depth3_issue.md`. + + The sustained probe runs ONCE per session (cached); the cost + is a ~0.9s all-core burn on first call only. + + ''' + global _sustained_headroom + static: float = cpu_scaling_factor() + if _non_linux: + return static + if _sustained_headroom is None: + _sustained_headroom = _measure_sustained_headroom() + return max(static, _sustained_headroom) + + # NOTE, the `--ll`/`--tl` CLI flags + the `loglevel`, `test_log` # and `testing_pkg_name` fixtures have been factored into the # `tractor._testing.pytest` plugin (loaded via the `-p` entry in diff --git a/tests/devx/test_debugger.py b/tests/devx/test_debugger.py index 1c3338c5..dfcf36d8 100644 --- a/tests/devx/test_debugger.py +++ b/tests/devx/test_debugger.py @@ -794,6 +794,14 @@ def test_multi_nested_subactors_error_through_nurseries( loglevel='pdb', ) last_send_char: str|None = None + + # inflate pexpect waits under CPU throttle — incl. the + # sustained-load power-cap invisible to static freq reads — so + # a slow-to-boot child REPL doesn't trip a false `TIMEOUT`. + # See `scripts/cpu-perf-check`. + from ..conftest import cpu_perf_headroom + headroom: float = cpu_perf_headroom() + for ( i, send_char, @@ -817,6 +825,9 @@ def test_multi_nested_subactors_error_through_nurseries( if is_forking_spawner: timeout += 4 + if headroom != 1.: + timeout *= headroom + try: child.expect( PROMPT, diff --git a/tests/test_advanced_streaming.py b/tests/test_advanced_streaming.py index c04bf130..3e9fa3f5 100644 --- a/tests/test_advanced_streaming.py +++ b/tests/test_advanced_streaming.py @@ -189,11 +189,18 @@ def test_dynamic_pub_sub( # sits forever until external SIGINT. The `afk_alarm_w_trace` # outer guard below is the AFK-safety counterpart (SIGALRM # raises in the main thread regardless of trio scope state). - fail_after_s: int = ( + fail_after_s: float = ( 8 if is_forking_spawner else 20 ) + # inflate under CPU throttle — incl. the sustained-load + # power-cap invisible to static freq reads — so a slow box + # doesn't trip the deadline. See `scripts/cpu-perf-check`. + from .conftest import cpu_perf_headroom + headroom: float = cpu_perf_headroom() + if headroom != 1.: + fail_after_s *= headroom async def main(): # bug-class-3 breadcrumb: tag each level of the cancel path diff --git a/tests/test_cancellation.py b/tests/test_cancellation.py index ab26c03d..bff4c0c2 100644 --- a/tests/test_cancellation.py +++ b/tests/test_cancellation.py @@ -596,6 +596,15 @@ async def test_nested_multierrors( # depth=3, BOTH variants will reliably `xpass` and # pytest will yell — our signal to drop the marker. See # `ai/conc-anal/cancel_cascade_too_slow_under_main_thread_forkserver_issue.md`. + # + # Probe CPU throttle ONCE up-front (folds in the sustained-load + # power-cap that static freq reads miss): used BOTH to inflate + # the deadline budget below AND to xfail depth=3, whose failure + # mode under throttle is a runtime-internal reap deadline — not + # a test-budget miss. See `scripts/cpu-perf-check`. + from .conftest import cpu_perf_headroom + headroom: float = cpu_perf_headroom() + if start_method == 'main_thread_forkserver': request.node.add_marker( pytest.mark.xfail( @@ -609,6 +618,34 @@ async def test_nested_multierrors( ) ) + # Under CPU throttle (incl. the sustained-load power-cap that + # static freq reads miss) the DEEP depth=3 tree trips tractor's + # INTERNAL reap deadlines (`soft_kill`/`hard_kill` + # `move_on_after`/`terminate_after=1.6`) before slow subprocs + # exit, injecting a `Cancelled(source='deadline')` into the BEG + # — the SAME shape-mismatch class as the MTF xfail above, and + # NOT fixable by inflating the test-level budget (the Cancelled + # is minted inside the runtime, not by our `fail_after`). + # xfail(strict=False) so it auto-clears the moment the box is + # un-throttled (`headroom == 1.`); depth=1's shallow tree stays + # under those internal deadlines so it just rides the budget + # inflation below. See `scripts/cpu-perf-check`. + elif ( + depth == 3 + and + headroom != 1. + ): + request.node.add_marker( + pytest.mark.xfail( + strict=False, + reason=( + 'CPU throttled — tractor reap deadline injects ' + 'Cancelled into BEG; see conc-anal/' + 'trio_033_cancel_cascade_slowdown_depth3_issue.md' + ), + ) + ) + # Per-backend/-depth budgets: in the non-hang case the # whole spawn + cancel-cascade should complete in well # under these. On the borderline hang case the @@ -636,11 +673,14 @@ async def test_nested_multierrors( case ('main_thread_forkserver', 3): timeout = 30 - # headroom for CPU-freq scaling AND/OR slow CI so the inner - # snapshot-capturing budget doesn't fire spuriously on a - # sluggish runner; see `cpu_scaling_factor()`. - from .conftest import cpu_scaling_factor - timeout *= cpu_scaling_factor() + # inflate the budget by the throttle headroom probed above so + # a slow box doesn't masquerade as a deadline regression. + # NOTE, `headroom = cpu_perf_headroom()` (set above) is the + # SUPERSET of `cpu_scaling_factor()` — it folds in the static + # cpu-freq scaling + slow-CI bump AND the sustained-load + # throttle probe this depth-3 cascade was the poster child for. + if headroom != 1.: + timeout *= headroom async with fail_after_w_trace(timeout): try: @@ -748,7 +788,7 @@ def test_cancel_via_SIGINT_other_task( started from a seperate ``trio`` child task. ''' - from .conftest import cpu_scaling_factor + from .conftest import cpu_perf_headroom pid: int = os.getpid() timeout: float = ( @@ -758,8 +798,9 @@ def test_cancel_via_SIGINT_other_task( if _friggin_windows: # smh timeout += 1 - # add latency headroom for CPU freq scaling (auto-cpufreq et al.) - headroom: float = cpu_scaling_factor() + # latency headroom for static cpu-freq scaling + sustained-load + # throttle + CI (auto-cpufreq et al.); see `cpu_perf_headroom()`. + headroom: float = cpu_perf_headroom() if headroom != 1.: timeout *= headroom @@ -962,11 +1003,11 @@ def test_fast_graceful_cancel_when_spawn_task_in_soft_proc_wait_for_daemon( if _friggin_windows: # smh timeout += 1 - # CPU-scaling / CI latency headroom — macOS CI especially is - # slow for this graceful-vs-hard-reap timing race; see - # `cpu_scaling_factor()`. - from .conftest import cpu_scaling_factor - timeout *= cpu_scaling_factor() + # CPU-scaling / sustained-throttle / CI latency headroom — macOS + # CI especially is slow for this graceful-vs-hard-reap timing + # race; see `cpu_perf_headroom()`. + from .conftest import cpu_perf_headroom + timeout *= cpu_perf_headroom() async def main(): start = time.time() diff --git a/tests/test_clustering.py b/tests/test_clustering.py index efb47d19..c426c746 100644 --- a/tests/test_clustering.py +++ b/tests/test_clustering.py @@ -24,8 +24,14 @@ def test_empty_mngrs_input_raises( 'actor-cluster teardown hangs intermittently on UDS' ) + # inflate under CPU throttle — incl. the sustained-load + # power-cap invisible to static freq reads. See + # `scripts/cpu-perf-check`. + from .conftest import cpu_perf_headroom + fail_after_s: float = 3 * cpu_perf_headroom() + async def main(): - with trio.fail_after(3): + with trio.fail_after(fail_after_s): async with ( open_actor_cluster( modules=[__name__], @@ -93,6 +99,13 @@ async def test_streaming_to_actor_cluster( 10 if is_forking_spawner else 6 ) + # inflate under CPU throttle — incl. the sustained-load + # power-cap invisible to static freq reads. See + # `scripts/cpu-perf-check`. + from .conftest import cpu_perf_headroom + headroom: float = cpu_perf_headroom() + if headroom != 1.: + delay *= headroom with trio.fail_after(delay): async with ( open_actor_cluster(modules=[__name__]) as portals, diff --git a/tests/test_docs_examples.py b/tests/test_docs_examples.py index 4f791727..cce85292 100644 --- a/tests/test_docs_examples.py +++ b/tests/test_docs_examples.py @@ -160,7 +160,7 @@ def test_example( 'root-causing. Passes on Linux.' ) - from .conftest import cpu_scaling_factor + from .conftest import cpu_perf_headroom timeout: float = ( 60 @@ -168,8 +168,9 @@ def test_example( else 16 ) - # add latency headroom for CPU freq scaling (auto-cpufreq et al.) - headroom: float = cpu_scaling_factor() + # add latency headroom for CPU freq scaling/throttle + # (auto-cpufreq et al.) + headroom: float = cpu_perf_headroom() if headroom != 1.: timeout *= headroom diff --git a/tests/test_inter_peer_cancellation.py b/tests/test_inter_peer_cancellation.py index 6bf74880..e0bf5dab 100644 --- a/tests/test_inter_peer_cancellation.py +++ b/tests/test_inter_peer_cancellation.py @@ -24,7 +24,7 @@ from tractor._testing import ( expect_ctxc, ) -from .conftest import cpu_scaling_factor +from .conftest import cpu_perf_headroom pytestmark = [ pytest.mark.skipon_spawn_backend( @@ -1280,12 +1280,12 @@ def test_peer_spawns_and_cancels_service_subactor( async def _main(): - headroom: float = cpu_scaling_factor() + headroom: float = cpu_perf_headroom() this_fast_on_linux: float = 3 this_fast = this_fast_on_linux * headroom if headroom != 1.: test_log.warning( - f'Adding latency headroom on linux bc CPU scaling,\n' + f'Adding latency headroom on linux bc CPU perf scaling/throttle,\n' f'headroom: {headroom}\n' f'this_fast_on_linux: {this_fast_on_linux} -> {this_fast}\n' ) diff --git a/tests/test_legacy_one_way_streaming.py b/tests/test_legacy_one_way_streaming.py index 648ba1b8..ade76763 100644 --- a/tests/test_legacy_one_way_streaming.py +++ b/tests/test_legacy_one_way_streaming.py @@ -326,11 +326,11 @@ def time_quad_ex( ): timeout += 1 - # inflate the cancel-deadline for CPU-freq scaling AND/OR CI - # latency (see `cpu_scaling_factor()`) so the example isn't - # cancelled mid-stream on a throttled/CI runner. - from .conftest import cpu_scaling_factor - timeout *= cpu_scaling_factor() + # inflate the cancel-deadline for static cpu-freq scaling + + # sustained-load throttle + CI latency (see `cpu_perf_headroom()`) + # so the example isn't cancelled mid-stream on a throttled/CI box. + from .conftest import cpu_perf_headroom + timeout *= cpu_perf_headroom() start: float = time.time() results: list[int] = trio.run(partial( @@ -379,8 +379,8 @@ def test_a_quadruple_example( # https://github.com/AdnanHodzic/auto-cpufreq?tab=readme-ov-file#example-config-file-contents # # HENCE this below latency-headroom compensation logic.. - from .conftest import cpu_scaling_factor - headroom: float = cpu_scaling_factor() + from .conftest import cpu_perf_headroom + headroom: float = cpu_perf_headroom() if headroom != 1.: this_fast = this_fast_on_linux * headroom test_log.warning( diff --git a/tests/test_resource_cache.py b/tests/test_resource_cache.py index 4902fb7a..049c0a6b 100644 --- a/tests/test_resource_cache.py +++ b/tests/test_resource_cache.py @@ -213,12 +213,12 @@ def test_open_local_sub_to_stream( N local tasks using `trionics.maybe_open_context()`. ''' - from .conftest import cpu_scaling_factor + from .conftest import cpu_perf_headroom timeout: float = ( 4 if not platform.system() == "Windows" else 10 - ) * cpu_scaling_factor() + ) * cpu_perf_headroom() if debug_mode: timeout = 999 diff --git a/tests/test_spawning.py b/tests/test_spawning.py index b6128b96..6237282e 100644 --- a/tests/test_spawning.py +++ b/tests/test_spawning.py @@ -153,9 +153,9 @@ async def test_most_beautiful_word( # actor spawn + IPC round-trip is comfortably sub-second on a # warm box, but slow/noisy CI runners (esp. macOS) blow a flat # 1s deadline. Scale for CI/CPU-throttle headroom — `== 1s` - # locally where `cpu_scaling_factor()` is `1.0`. - from .conftest import cpu_scaling_factor - with trio.fail_after(1 * cpu_scaling_factor()): + # locally where `cpu_perf_headroom()` is `1.0`. + from .conftest import cpu_perf_headroom + with trio.fail_after(1 * cpu_perf_headroom()): async with tractor.open_nursery( debug_mode=debug_mode, ) as an: diff --git a/tractor/_testing/addr.py b/tractor/_testing/addr.py index db95889d..58acfd1a 100644 --- a/tractor/_testing/addr.py +++ b/tractor/_testing/addr.py @@ -50,9 +50,14 @@ def get_rando_addr( fight over the bind, cascading into a chain of "Address already in use" failures. - For UDS this concern doesn't apply: `UDSAddress.get_random()` - already builds socket paths from `os.getpid()` so each - pytest process gets its own socket-path namespace. + For UDS the *cross*-process concern is handled by keying + socket paths off `os.getpid()` (each pytest process gets its + own namespace). *Within* a process — where `get_random()` has + no live runtime to name from — it also appends a per-call + `uuid4` token, so two calls (e.g. a `reg_addr` + a distinct + bind addr in one test body) yield distinct sockpaths rather + than aliasing to the same `name@pid.sock` and tripping + `EADDRINUSE`. ''' addr_type: Type[_addr.Addres] = _addr._address_types[tpt_proto] diff --git a/tractor/ipc/_uds.py b/tractor/ipc/_uds.py index 8c57664d..d1e9d2f6 100644 --- a/tractor/ipc/_uds.py +++ b/tractor/ipc/_uds.py @@ -36,6 +36,7 @@ from typing import ( TYPE_CHECKING, ClassVar, ) +from uuid import uuid4 import msgspec import trio @@ -204,7 +205,18 @@ class UDSAddress( else: prefix: str = 'no_runtime_actor' - sockname: str = f'{prefix}@{pid}' + # XXX, no live actor -> no `Aid` to key off, so mix a + # per-CALL token into the NAME part for uniqueness: + # w/o a runtime the sockname is otherwise a pure fn of + # `(prefix, pid)`, so two `get_random()` calls in one + # proc alias to the SAME sockpath and the 2nd `.bind()` + # trips `EADDRINUSE`. Token goes BEFORE `@{pid}` so the + # canonical `...@{pid}.sock` suffix stays intact for the + # reapers keyed off it: `._testing._reap`'s + # `(?P.+)@(?P\d+)\.sock` regex, and the + # `spawn._reap` `{name}@{pid}.sock` reconstruction. + token: str = uuid4().hex[:8] + sockname: str = f'{prefix}.{token}@{pid}' sockpath: Path = Path(f'{sockname}.sock') return UDSAddress(