diff --git a/src/fable-library-py/fable_library/async_.py b/src/fable-library-py/fable_library/async_.py index 36c814ac79..3035b42909 100644 --- a/src/fable-library-py/fable_library/async_.py +++ b/src/fable-library-py/fable_library/async_.py @@ -295,9 +295,7 @@ def cancel(_: OperationCanceledError) -> None: def _start_child_timeout[T]( computation: Async[T], ms: int | TimeSpan, timeout_error: Callable[[], Exception] ) -> Async[Async[T]]: - # Race the computation against a timeout: whichever settles first wins. - # asyncio.gather (the previous implementation via parallel2) waited for BOTH to settle, - # which meant the timeout always fired even when the computation finished first. + # Return the child result when ready, or raise the selected timeout error if it takes too long. task = start_as_task(computation) async def with_timeout() -> T: diff --git a/src/fable-library-py/fable_library/util.py b/src/fable-library-py/fable_library/util.py index d6797f14f7..1d3f5648ed 100644 --- a/src/fable-library-py/fable_library/util.py +++ b/src/fable-library-py/fable_library/util.py @@ -29,6 +29,7 @@ overload, ) from urllib.parse import quote, unquote +from weakref import WeakValueDictionary from .array_ import Array from .bases import ( @@ -351,16 +352,21 @@ def assert_not_equal[T](actual: T, expected: T, msg: str | None = None) -> None: raise Exception(msg or f"Expected not equal to: {expected} - Actual: {actual}") -MAX_LOCKS = 1024 +# decision: weak values give each active object its own lock without retaining idle locks or their objects +_locks: WeakValueDictionary[int, RLock] = WeakValueDictionary() +_lock_registry_guard = RLock() -# decision: fixed stripes bound synchronization storage without evicting a lock held by another thread -# tradeoff: unrelated objects can share a stripe and serialize, but object identity always selects the same lock -_locks = tuple(RLock() for _ in range(MAX_LOCKS)) +def lock[T](lock_obj: object, fn: Callable[[], T]) -> T: + key = id(lock_obj) + with _lock_registry_guard: + object_lock = _locks.get(key) + if object_lock is None: + object_lock = RLock() + _locks[key] = object_lock - -def lock[T](lock_obj: Any, fn: Callable[[], T]) -> T: - with _locks[(id(lock_obj) >> 4) % MAX_LOCKS]: + # invariant: each holder and waiter keeps a strong lock reference outside the registry guard + with object_lock: return fn() diff --git a/tests/Python/TestMisc.fs b/tests/Python/TestMisc.fs index ef027191cd..769d89fc7a 100644 --- a/tests/Python/TestMisc.fs +++ b/tests/Python/TestMisc.fs @@ -1622,13 +1622,13 @@ module private LockRuntime = abstract run: Action * Action -> unit type Storage = - abstract capacity: int abstract stable: bool abstract count: int let makeThreads: Func = import "make_threads" "./py/lock_runtime.py" let makeStorage: Func = import "make_storage" "./py/lock_runtime.py" let makeKey: Func = import "make_key" "./py/lock_runtime.py" + let makeCollidingKeys: Func = import "make_colliding_keys" "./py/lock_runtime.py" [] let ``test real threads share a lock and release after an exception`` () = @@ -1658,10 +1658,42 @@ let ``test real threads share a lock and release after an exception`` () = lock key (fun () -> 42) |> equal 42 [] -let ``test lock storage stays bounded across new objects`` () = +let ``test independent objects do not share a lock`` () = + let keys = LockRuntime.makeCollidingKeys.Invoke() + let threads = LockRuntime.makeThreads.Invoke() + let first = Action(fun () -> + lock keys.[0] (fun () -> + threads.record "first-enter" + threads.wait_release() |> equal true + threads.record "first-exit")) + let second = Action(fun () -> + lock keys.[1] (fun () -> threads.record "second-enter") + threads.record "second-exit") + threads.run(first, second) + equal true threads.first_entered + equal true threads.second_attempted + equal false threads.second_blocked + equal true threads.threads_finished + equal [||] threads.errors + equal 4 threads.trace.Length + +[] +let ``test lock storage releases idle entries across new objects`` () = + let iterations = 2048 let storage = LockRuntime.makeStorage.Invoke() - for _ in 1 .. storage.capacity * 2 do + let before = storage.count + lock (LockRuntime.makeKey.Invoke()) (fun () -> equal (before + 1) storage.count) + for _ in 1 .. iterations do lock (LockRuntime.makeKey.Invoke()) (fun () -> 1) |> equal 1 equal true storage.stable - equal storage.capacity storage.count + equal before storage.count + +[] +let ``test lock storage releases idle entries after exceptions`` () = + let storage = LockRuntime.makeStorage.Invoke() + let before = storage.count + try + lock (LockRuntime.makeKey.Invoke()) (fun () -> failwith "body") + with _ -> () + equal before storage.count #endif diff --git a/tests/Python/py/lock_runtime.py b/tests/Python/py/lock_runtime.py index e0d0dbf86e..22f29ffd71 100644 --- a/tests/Python/py/lock_runtime.py +++ b/tests/Python/py/lock_runtime.py @@ -1,14 +1,19 @@ """Thread and acquisition controls for the generated F# lock tests.""" +import gc from collections.abc import Callable from threading import Event, RLock, Thread from types import TracebackType from unittest.mock import patch +from weakref import WeakValueDictionary from fable_library import util from fable_library.array_ import Array +LEGACY_STRIPE_COUNT = 1024 + + class Acquisition: def __init__(self) -> None: self.attempted = Event() @@ -67,7 +72,6 @@ def wait_release(self) -> bool: def run(self, first: Callable[[], None], second: Callable[[], None]) -> None: acquisition = Acquisition() - stripes = tuple(ObservedLock(acquisition) for _ in range(util.MAX_LOCKS)) def invoke(action: Callable[[], None]) -> None: try: @@ -77,10 +81,9 @@ def invoke(action: Callable[[], None]) -> None: first_thread = Thread(target=invoke, args=(first,), daemon=True) second_thread = Thread(target=invoke, args=(second,), daemon=True) - # decision: observes both the original per-call factory and the fixed shared stripes with ( patch.object(util, "RLock", lambda: ObservedLock(acquisition)), - patch.object(util, "_locks", stripes, create=True), + patch.object(util, "_locks", WeakValueDictionary()), ): first_thread.start() try: @@ -101,7 +104,6 @@ def invoke(action: Callable[[], None]) -> None: class Storage: def __init__(self) -> None: self.before = getattr(util, "_locks", ()) - self.capacity = util.MAX_LOCKS @property def stable(self) -> bool: @@ -109,6 +111,7 @@ def stable(self) -> bool: @property def count(self) -> int: + gc.collect() return len(getattr(util, "_locks", ())) @@ -117,6 +120,19 @@ def make_key() -> object: return [] +def make_colliding_keys() -> Array[object]: + # decision: reproduces the former stripe collision using distinct, unhashable objects + keys: dict[int, object] = {} + for _ in range(LEGACY_STRIPE_COUNT + 1): + key: object = [] + stripe = (id(key) >> 4) % LEGACY_STRIPE_COUNT + previous = keys.get(stripe) + if previous is not None: + return Array([previous, key]) + keys[stripe] = key + raise AssertionError("Expected a collision in the former fixed stripes") + + def make_threads() -> Threads: return Threads()