From fbd4236cafa3c9a00d60304466dc5dd441d53a68 Mon Sep 17 00:00:00 2001 From: OutstanderWang Date: Tue, 20 Oct 2026 21:25:43 +0800 Subject: [PATCH 1/3] [fix] a forked child keeps the lock handles it inherits instead of releasing them Releasing an inherited Ray actor handle in a fork child runs the parent's core worker there. The child crashes in _int_free, the parent's lock actors can be reported out of scope and die, and the parent's later GCS calls time out. A datasets map with num_proc after tq.init() hit all three: every TransferQueueLockManager died and ray.get_actor in the parent timed out. _reset_after_fork now parks the inherited _owned, _managers and _leases in _inherited_after_fork, which is never released, before resetting the state. Signed-off-by: OutstanderWang --- tests/test_kv_lock.py | 22 ++++++++++++++++++++++ transfer_queue/kv_lock.py | 4 ++++ 2 files changed, 26 insertions(+) diff --git a/tests/test_kv_lock.py b/tests/test_kv_lock.py index 1a239ed9..10d2fcdc 100644 --- a/tests/test_kv_lock.py +++ b/tests/test_kv_lock.py @@ -14,9 +14,11 @@ # limitations under the License. import asyncio +import gc import importlib import threading import time +import weakref from unittest.mock import MagicMock import pytest @@ -172,6 +174,26 @@ def test_invalid_lock_settings_raise_before_any_actor_starts(fake_lock_actors, m fake_lock_actors.options.assert_not_called() +def test_fork_child_keeps_inherited_lock_handles_referenced(monkeypatch): + class Handle: + pass + + owned, managed = Handle(), Handle() + for name in ("_state_lock", "_table", "_registry", "_leases", "_renewer_running"): + monkeypatch.setattr(kvl, name, getattr(kvl, name)) + monkeypatch.setattr(kvl, "_owned", [owned]) + monkeypatch.setattr(kvl, "_managers", {0: managed}) + monkeypatch.setattr(kvl, "_inherited_after_fork", []) + owned_ref, managed_ref = weakref.ref(owned), weakref.ref(managed) + del owned, managed + + kvl._reset_after_fork() + gc.collect() + + assert kvl._owned == [] and kvl._managers == {} + assert owned_ref() is not None and managed_ref() is not None + + def test_lock_settings_are_ignored_while_the_lock_is_off(monkeypatch): monkeypatch.delenv("TQ_ENABLE_KV_GLOBAL_LOCK", raising=False) monkeypatch.setenv("TQ_KV_GLOBAL_LOCK_NUM_SHARDS", "0") diff --git a/transfer_queue/kv_lock.py b/transfer_queue/kv_lock.py index 4ddff4f1..ba72d060 100644 --- a/transfer_queue/kv_lock.py +++ b/transfer_queue/kv_lock.py @@ -216,6 +216,7 @@ async def async_kv_local_lock(keys: str | list[str], partition_id: str, timeout: _managers: dict = {} # shard -> TransferQueueLockManager handle, looked up on first use _owned: list = [] # the lock actors this process's tq.init() created, killed by its tq.close() _closes = 0 # this process's tq.close() calls: a lease created before the latest one is given up +_inherited_after_fork: list = [] # a forked child's copies of the parent's handles, never released # Longest one call waits inside a lock actor; waiting longer takes another call that keeps # the queue position. The bound is long because every waiter spends one call per window, and # those calls share the actor's limited throughput with everyone else's, renewals included. @@ -585,7 +586,10 @@ def _close_lock_managers() -> None: def _reset_after_fork() -> None: # The child inherits keys, leases and Ray handles that belong to threads and connections it lacks. + # Releasing an inherited handle runs the parent's core worker in the child: that crashes the child, + # can report the parent's lock actors out of scope, and leaves the parent's GCS calls timing out. global _state_lock, _table, _registry, _leases, _renewer_running, _managers, _owned + _inherited_after_fork.append((_owned, _managers, _leases)) _state_lock, _table = threading.Lock(), {} _registry, _leases, _renewer_running, _managers, _owned = threading.Condition(), {}, False, {}, [] From d7d15c06f1c795bcb88e5e48810938ca81f76f3e Mon Sep 17 00:00:00 2001 From: OutstanderWang Date: Fri, 23 Oct 2026 14:25:48 +0800 Subject: [PATCH 2/3] [chore] Drop the duplicate comment on _inherited_after_fork Signed-off-by: OutstanderWang --- transfer_queue/kv_lock.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/transfer_queue/kv_lock.py b/transfer_queue/kv_lock.py index ba72d060..3586d2af 100644 --- a/transfer_queue/kv_lock.py +++ b/transfer_queue/kv_lock.py @@ -216,7 +216,7 @@ async def async_kv_local_lock(keys: str | list[str], partition_id: str, timeout: _managers: dict = {} # shard -> TransferQueueLockManager handle, looked up on first use _owned: list = [] # the lock actors this process's tq.init() created, killed by its tq.close() _closes = 0 # this process's tq.close() calls: a lease created before the latest one is given up -_inherited_after_fork: list = [] # a forked child's copies of the parent's handles, never released +_inherited_after_fork: list = [] # Longest one call waits inside a lock actor; waiting longer takes another call that keeps # the queue position. The bound is long because every waiter spends one call per window, and # those calls share the actor's limited throughput with everyone else's, renewals included. From 8eee484554dd13bea7b5fd0a75a06589006a58b9 Mon Sep 17 00:00:00 2001 From: OutstanderWang Date: Fri, 23 Oct 2026 14:25:48 +0800 Subject: [PATCH 3/3] [test] Cover lease handles kept after fork Signed-off-by: OutstanderWang --- tests/test_kv_lock.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/tests/test_kv_lock.py b/tests/test_kv_lock.py index 10d2fcdc..b1795434 100644 --- a/tests/test_kv_lock.py +++ b/tests/test_kv_lock.py @@ -178,20 +178,21 @@ def test_fork_child_keeps_inherited_lock_handles_referenced(monkeypatch): class Handle: pass - owned, managed = Handle(), Handle() - for name in ("_state_lock", "_table", "_registry", "_leases", "_renewer_running"): + owned, managed, lease = Handle(), Handle(), Handle() + for name in ("_state_lock", "_table", "_registry", "_renewer_running"): monkeypatch.setattr(kvl, name, getattr(kvl, name)) monkeypatch.setattr(kvl, "_owned", [owned]) monkeypatch.setattr(kvl, "_managers", {0: managed}) + monkeypatch.setattr(kvl, "_leases", {"token": lease}) monkeypatch.setattr(kvl, "_inherited_after_fork", []) - owned_ref, managed_ref = weakref.ref(owned), weakref.ref(managed) - del owned, managed + refs = [weakref.ref(obj) for obj in (owned, managed, lease)] + del owned, managed, lease kvl._reset_after_fork() gc.collect() - assert kvl._owned == [] and kvl._managers == {} - assert owned_ref() is not None and managed_ref() is not None + assert kvl._owned == [] and kvl._managers == {} and kvl._leases == {} + assert all(ref() is not None for ref in refs) def test_lock_settings_are_ignored_while_the_lock_is_off(monkeypatch):