diff --git a/src/fable-library-py/fable_library/async_.py b/src/fable-library-py/fable_library/async_.py index 1b72efd6ee..7a5f46a02d 100644 --- a/src/fable-library-py/fable_library/async_.py +++ b/src/fable-library-py/fable_library/async_.py @@ -32,6 +32,7 @@ ) from .event import IEvent_2 from .protocols import IEnumerable_1 +from .system import TimeoutException__ctor from .task import TaskCompletionSource from .time_span import TimeSpan, to_milliseconds from .types import UNIT, Unit @@ -291,23 +292,29 @@ def cancel(_: OperationCanceledError) -> None: return tcs.get_task() -def start_child[T](computation: Async[T], ms: int | TimeSpan | None = None) -> Async[Async[T]]: - if ms is not None: - # 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. - task = start_as_task(computation) +def _start_child_timeout[T](computation: Async[T], ms: int | TimeSpan) -> Async[Async[T]]: + # Return the child result when ready, or raise TimeoutException when the deadline expires. + task = asyncio.ensure_future(start_as_task(computation)) - async def with_timeout() -> T: - try: - return await asyncio.wait_for(task, timeout=to_milliseconds(ms) / 1000.0) - except TimeoutError: - raise TimeoutError() + async def with_timeout() -> T: + try: + return await asyncio.wait_for(task, timeout=to_milliseconds(ms) / 1000.0) + except TimeoutError: + # invariant: a completed child retains its exception; only deadline cancellation is translated + if not task.cancelled(): + raise + raise TimeoutException__ctor() - def cont(ctx: IAsyncContext[Async[T]]) -> None: - protected_return(await_task(with_timeout()))(ctx) + def cont(ctx: IAsyncContext[Async[T]]) -> None: + protected_return(await_task(with_timeout()))(ctx) - return protected_cont(cont) + return protected_cont(cont) + + +def start_child[T](computation: Async[T], ms: int | TimeSpan | None = None) -> Async[Async[T]]: + # decision: share TimeoutException across callers; its native base preserves Python TimeoutError handlers + if ms is not None: + return _start_child_timeout(computation, ms) task = start_as_task(computation) diff --git a/src/fable-library-py/fable_library/util.py b/src/fable-library-py/fable_library/util.py index c2abf502e5..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() -def lock[T](lock_obj: Any, fn: Callable[[], T]) -> T: - @functools.lru_cache(maxsize=MAX_LOCKS) - def get_lock(n: int) -> RLock: - return RLock() +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 - lock = get_lock(id(lock_obj)) - with lock: + # invariant: each holder and waiter keeps a strong lock reference outside the registry guard + with object_lock: return fn() diff --git a/tests/Python/TestAsync.fs b/tests/Python/TestAsync.fs index 6db0fdf8c3..779d7f8653 100644 --- a/tests/Python/TestAsync.fs +++ b/tests/Python/TestAsync.fs @@ -467,7 +467,7 @@ let ``test Async.StartChild applies timeout`` () = let mutable x = "" let task = async { - x <- x + "A" + // A busy .NET runner can time out before this child is scheduled. do! Async.Sleep 1_000 x <- x + "X" // Never hit } @@ -482,8 +482,8 @@ let ``test Async.StartChild applies timeout`` () = x <- x + "C" - equal x "ABC" - } |> Async.StartImmediate + equal "BC" x + } |> Async.RunSynchronously [] let ``test Async.StartChild with timeout completes when computation finishes before timeout`` () = // See #4481 @@ -496,7 +496,74 @@ let ``test Async.StartChild with timeout completes when computation finishes bef with | :? TimeoutException -> failwith "should not time out" - } |> Async.StartImmediate + } |> Async.RunSynchronously + +let private assertChildFailure startChild (error: exn) = + async { + let failing: Async = async { + do! Async.Sleep 10 + return raise error + } + let! child = startChild(failing, 60_000) + let! result = Async.Catch child + match result with + | Choice2Of2 caught -> obj.ReferenceEquals(error, caught) |> equal true + | _ -> failwith "Expected child failure" + } |> Async.RunSynchronously + +[] +let ``test Async.StartChild preserves child TimeoutException`` () = + assertChildFailure + (fun (computation, timeout) -> Async.StartChild(computation, timeout)) + (TimeoutException("child timeout")) + +let private assertChildTimeout startChild assertError = + async { + let pending: Async = Async.FromContinuations(ignore) + let! child = startChild(pending, 0) + let! result = Async.Catch child + match result with + | Choice2Of2 ex -> assertError ex + | _ -> failwith "Expected child timeout" + } |> Async.RunSynchronously + +[] +let ``test Async.StartChild zero timeout raises TimeoutException`` () = + assertChildTimeout + (fun (computation, timeout) -> Async.StartChild(computation, timeout)) + (fun ex -> ex :? TimeoutException |> equal true) + +#if FABLE_COMPILER +module private NativeTimeout = + open Fable.Core + open Fable.Core.PyInterop + + let startChild: Func, int, Async>> = import "start_child" "fable_library.async_" + let timeoutError: obj = import "TimeoutError" "builtins" + + [] + let createTimeoutError (message: string) : exn = nativeOnly + +[] +let ``test Python start_child deadline matches both timeout types`` () = + assertChildTimeout + (fun (computation, timeout) -> NativeTimeout.startChild.Invoke(computation, timeout)) + (fun ex -> + ex :? TimeoutException |> equal true + Fable.Core.PyInterop.pyInstanceof ex NativeTimeout.timeoutError |> equal true) + +[] +let ``test Async.StartChild preserves child Python TimeoutError`` () = + assertChildFailure + (fun (computation, timeout) -> Async.StartChild(computation, timeout)) + (NativeTimeout.createTimeoutError "child timeout") + +[] +let ``test Python start_child preserves child TimeoutError`` () = + assertChildFailure + (fun (computation, timeout) -> NativeTimeout.startChild.Invoke(computation, timeout)) + (NativeTimeout.createTimeoutError "child timeout") +#endif [] let ``test Unit arguments are erased`` () = // See #1832 diff --git a/tests/Python/TestMisc.fs b/tests/Python/TestMisc.fs index 77534d182e..44eca8a4cc 100644 --- a/tests/Python/TestMisc.fs +++ b/tests/Python/TestMisc.fs @@ -1677,3 +1677,111 @@ let ``test exception variable captured in deferred single-arg closure`` () = with ex -> (fun (_: int) -> ex.Message) // lambda-with-arg captures ex sub 42 |> equal "boom" + +[] +let ``test lock is reentrant and returns the body value`` () = + let key = obj() + lock key (fun () -> lock key (fun () -> 42)) |> equal 42 + +[] +let ``test lock releases after a body exception`` () = + let key = obj() + let mutable error = "" + try + lock key (fun () -> failwith "body") + with ex -> + error <- ex.Message + equal "body" error + lock key (fun () -> 42) |> equal 42 + +#if FABLE_COMPILER +module private LockRuntime = + open Fable.Core.PyInterop + + type Threads = + abstract trace: string array + abstract errors: string array + abstract first_entered: bool + abstract second_attempted: bool + abstract second_blocked: bool + abstract threads_finished: bool + abstract record: string -> unit + abstract wait_release: unit -> bool + abstract run: Action * Action -> unit + + type Storage = + 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`` () = + let key = LockRuntime.makeKey.Invoke() + let threads = LockRuntime.makeThreads.Invoke() + let mutable error = "" + let first = Action(fun () -> + try + lock key (fun () -> + threads.record "first-enter" + threads.wait_release() |> equal true + threads.record "first-exit" + failwith "body") + with ex -> + error <- ex.Message) + let second = Action(fun () -> + lock key (fun () -> threads.record "second-enter") + threads.record "second-exit") + threads.run(first, second) + equal true threads.first_entered + equal true threads.second_attempted + equal true threads.second_blocked + equal true threads.threads_finished + equal [||] threads.errors + equal [| "first-enter"; "first-exit"; "second-enter"; "second-exit" |] threads.trace + equal "body" error + lock key (fun () -> 42) |> equal 42 + +[] +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() + 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 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 new file mode 100644 index 0000000000..22f29ffd71 --- /dev/null +++ b/tests/Python/py/lock_runtime.py @@ -0,0 +1,141 @@ +"""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() + self.blocked = False + + +class ObservedLock: + def __init__(self, acquisition: Acquisition) -> None: + self.lock = RLock() + self.acquisition = acquisition + + def __enter__(self) -> "ObservedLock": + acquired = self.lock.acquire(blocking=False) + self.acquisition.blocked = not acquired + # invariant: the contender's observation follows an actual acquisition attempt + self.acquisition.attempted.set() + if not acquired: + self.lock.acquire() + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self.lock.release() + + +class Threads: + def __init__(self) -> None: + self.entered = Event() + self.release = Event() + self._trace: list[str] = [] + self._errors: list[str] = [] + self.first_entered = False + self.second_attempted = False + self.second_blocked = False + self.threads_finished = False + + @property + def trace(self) -> Array[str]: + return Array(self._trace) + + @property + def errors(self) -> Array[str]: + return Array(self._errors) + + def record(self, entry: str) -> None: + self._trace.append(entry) + if entry == "first-enter": + self.entered.set() + + def wait_release(self) -> bool: + return self.release.wait(5) + + def run(self, first: Callable[[], None], second: Callable[[], None]) -> None: + acquisition = Acquisition() + + def invoke(action: Callable[[], None]) -> None: + try: + action() + except Exception as error: + self._errors.append(str(error)) + + first_thread = Thread(target=invoke, args=(first,), daemon=True) + second_thread = Thread(target=invoke, args=(second,), daemon=True) + with ( + patch.object(util, "RLock", lambda: ObservedLock(acquisition)), + patch.object(util, "_locks", WeakValueDictionary()), + ): + first_thread.start() + try: + self.first_entered = self.entered.wait(5) + if self.first_entered: + acquisition.attempted.clear() + second_thread.start() + self.second_attempted = acquisition.attempted.wait(5) + self.second_blocked = acquisition.blocked + finally: + self.release.set() + first_thread.join(5) + if second_thread.ident is not None: + second_thread.join(5) + self.threads_finished = not first_thread.is_alive() and not second_thread.is_alive() + + +class Storage: + def __init__(self) -> None: + self.before = getattr(util, "_locks", ()) + + @property + def stable(self) -> bool: + return getattr(util, "_locks", None) is self.before + + @property + def count(self) -> int: + gc.collect() + return len(getattr(util, "_locks", ())) + + +def make_key() -> object: + # decision: uses an unhashable, non-weak-referenceable key to cover arbitrary lock objects + 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() + + +def make_storage() -> Storage: + return Storage()