fix(python): use weak per-object locks instead of stripes

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:

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()
 
 

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<Threads> = import "make_threads" "./py/lock_runtime.py"
     let makeStorage: Func<Storage> = import "make_storage" "./py/lock_runtime.py"
     let makeKey: Func<obj> = import "make_key" "./py/lock_runtime.py"
+    let makeCollidingKeys: Func<obj array> = import "make_colliding_keys" "./py/lock_runtime.py"
 
 [<Fact>]
 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
 
 [<Fact>]
-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
+
+[<Fact>]
+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
+
+[<Fact>]
+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

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,14 +104,14 @@ 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:
         return getattr(util, "_locks", None) is self.before
 
     @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()