From 7aa16420bb653b3fbf23a8ef04863736c2ef9e59 Mon Sep 17 00:00:00 2001
From: nataliekung <4054699+nataliekung@users.noreply.github.com>
Date: Thu, 29 Oct 2026 06:10:37 -0700
Subject: [PATCH 1/3] [fix] Let concurrent jobs keep their TransferQueue actors
apart with TQ_RAY_NAMESPACE
The controller, the Ray storage actor and the lock managers are all named actors in the fixed Ray namespace transfer_queue. Two jobs on one Ray cluster therefore attach to the same controller: the second job reads the first job's partitions, and the first job's close() kills the controller the second job is using.
Read the namespace from TQ_RAY_NAMESPACE and keep transfer_queue as the default, so cross-job sharing (#115) is unchanged.
Signed-off-by: nataliekung <4054699+nataliekung@users.noreply.github.com>
---
README.md | 4 ++
tests/test_ray_namespace.py | 51 +++++++++++++++++++
transfer_queue/interface.py | 5 +-
transfer_queue/kv_lock.py | 6 +--
.../storage/clients/ray_storage_client.py | 5 +-
transfer_queue/utils/common.py | 9 ++++
6 files changed, 73 insertions(+), 7 deletions(-)
create mode 100644 tests/test_ray_namespace.py
diff --git a/README.md b/README.md
index 2b7cd65f..13f7b32a 100644
--- a/README.md
+++ b/README.md
@@ -205,6 +205,10 @@ Developers can leverage `TransferQueueClient` directly to implement advanced fea
- [tutorial/05_custom_sampler.py](https://github.com/Ascend/TransferQueue/blob/main/tutorial/05_custom_sampler.py)
+#### Concurrent jobs on one Ray cluster
+
+The controller and `RayStorageClient`'s storage actor are named actors in the Ray namespace `transfer_queue`, so every job on the cluster that calls `tq.init()` attaches to the same controller. This is what lets a trainer job connect to TransferQueue created by another job. Jobs that must not share data set `TQ_RAY_NAMESPACE` to a value unique per job (for example the job id) in every process of the job, for instance through the job-level `runtime_env` `env_vars`. Without it, a second job reads the first job's partitions, and the first job's `tq.close()` kills the controller the second one is using.
+
🔥 Showcases
### Collocated Example
diff --git a/tests/test_ray_namespace.py b/tests/test_ray_namespace.py
new file mode 100644
index 00000000..3f1c5e63
--- /dev/null
+++ b/tests/test_ray_namespace.py
@@ -0,0 +1,51 @@
+# Copyright 2025 The TransferQueue Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""TQ_RAY_NAMESPACE moves the named actors out of the cluster-wide ``transfer_queue`` namespace."""
+
+import pytest
+import ray
+from omegaconf import OmegaConf
+
+import transfer_queue as tq
+from transfer_queue.utils.common import ray_namespace
+
+
+@pytest.mark.parametrize("value", [None, ""])
+def test_default_namespace_is_shared(monkeypatch, value):
+ if value is None:
+ monkeypatch.delenv("TQ_RAY_NAMESPACE", raising=False)
+ else:
+ monkeypatch.setenv("TQ_RAY_NAMESPACE", value)
+ assert ray_namespace() == "transfer_queue"
+
+
+def test_env_overrides_namespace(monkeypatch):
+ monkeypatch.setenv("TQ_RAY_NAMESPACE", "job-a")
+ assert ray_namespace() == "job-a"
+
+
+def test_controller_lives_in_configured_namespace(monkeypatch):
+ monkeypatch.setenv("TQ_RAY_NAMESPACE", "job-a")
+ ray.init(ignore_reinit_error=True, num_cpus=4)
+ try:
+ tq.init(OmegaConf.create({"backend": {"SimpleStorage": {"num_data_storage_units": 1}}}))
+ try:
+ ray.get_actor("TransferQueueController", namespace="job-a")
+ with pytest.raises(ValueError):
+ ray.get_actor("TransferQueueController", namespace="transfer_queue")
+ finally:
+ tq.close()
+ finally:
+ ray.shutdown()
diff --git a/transfer_queue/interface.py b/transfer_queue/interface.py
index 21e5a116..5aeadc42 100644
--- a/transfer_queue/interface.py
+++ b/transfer_queue/interface.py
@@ -54,6 +54,7 @@
from transfer_queue.sampler import BaseSampler
from transfer_queue.storage.bootstrap import StorageBootstrapProvider
from transfer_queue.storage.managers.simple_storage_manager import AsyncSimpleStorageManager
+from transfer_queue.utils.common import ray_namespace
from transfer_queue.utils.logging_utils import get_logger
from transfer_queue.utils.yuanrong_utils import cleanup_yuanrong_resources
from transfer_queue.utils.zmq_utils import process_zmq_server_info
@@ -123,7 +124,7 @@ def _init_from_existing() -> bool:
global _TQ_CONTROLLER
try:
if _TQ_CONTROLLER is None:
- _TQ_CONTROLLER = ray.get_actor("TransferQueueController", namespace="transfer_queue")
+ _TQ_CONTROLLER = ray.get_actor("TransferQueueController", namespace=ray_namespace())
except ValueError:
logger.info("Called _init_from_existing() but TransferQueueController has not been initialized yet.")
@@ -207,7 +208,7 @@ def init(conf: DictConfig | None = None) -> DictConfig | None:
try:
global _TQ_CONTROLLER, _TQ_IS_OWNER
_TQ_CONTROLLER = TransferQueueController.options( # type: ignore[attr-defined]
- name="TransferQueueController", namespace="transfer_queue"
+ name="TransferQueueController", namespace=ray_namespace()
).remote(sampler=sampler, polling_mode=final_conf.controller.polling_mode)
_TQ_IS_OWNER = True
logger.info("TransferQueueController has been created.")
diff --git a/transfer_queue/kv_lock.py b/transfer_queue/kv_lock.py
index 3586d2af..fc68839f 100644
--- a/transfer_queue/kv_lock.py
+++ b/transfer_queue/kv_lock.py
@@ -42,7 +42,7 @@
from ray.exceptions import GetTimeoutError, RayActorError
from transfer_queue.lock_manager import _RESERVED_SLOTS, TransferQueueLockManager
-from transfer_queue.utils.common import get_env_bool, get_node_round_robin_scheduling_strategies
+from transfer_queue.utils.common import get_env_bool, get_node_round_robin_scheduling_strategies, ray_namespace
# Not asyncio.Lock: it is bound to one event loop and is not thread-safe, while a process
# may run several loops plus plain threads. A key is held exactly while it is in _table,
@@ -292,7 +292,7 @@ def _manager_name(shard: int) -> str:
def _lock_manager(shard: int):
if shard not in _managers:
try:
- _managers[shard] = ray.get_actor(_manager_name(shard), namespace="transfer_queue")
+ _managers[shard] = ray.get_actor(_manager_name(shard), namespace=ray_namespace())
except ValueError:
raise RuntimeError(_NOT_INITIALIZED) from None
return _managers[shard]
@@ -556,7 +556,7 @@ def _start_lock_managers(settings: tuple[int, int] | None) -> None:
_owned = [
TransferQueueLockManager.options( # type: ignore[attr-defined]
name=_manager_name(shard),
- namespace="transfer_queue",
+ namespace=ray_namespace(),
get_if_exists=True,
max_concurrency=max_waiting_acquires + _RESERVED_SLOTS,
scheduling_strategy=strategy,
diff --git a/transfer_queue/storage/clients/ray_storage_client.py b/transfer_queue/storage/clients/ray_storage_client.py
index add04974..81bad177 100644
--- a/transfer_queue/storage/clients/ray_storage_client.py
+++ b/transfer_queue/storage/clients/ray_storage_client.py
@@ -20,6 +20,7 @@
import torch
from transfer_queue.storage.clients.base import StorageClientFactory, StorageKVClient
+from transfer_queue.utils.common import ray_namespace
@ray.remote(max_concurrency=8)
@@ -57,10 +58,10 @@ def __init__(self, config=None):
# initialize actor
try:
- self.storage_actor = ray.get_actor("RayObjectRefStorage", namespace="transfer_queue")
+ self.storage_actor = ray.get_actor("RayObjectRefStorage", namespace=ray_namespace())
except ValueError:
self.storage_actor = RayObjectRefStorage.options(
- name="RayObjectRefStorage", namespace="transfer_queue", get_if_exists=False
+ name="RayObjectRefStorage", namespace=ray_namespace(), get_if_exists=False
).remote()
def put(self, keys: list[str], values: list[Any]) -> list[Any] | None:
diff --git a/transfer_queue/utils/common.py b/transfer_queue/utils/common.py
index 1278dfd4..40574413 100644
--- a/transfer_queue/utils/common.py
+++ b/transfer_queue/utils/common.py
@@ -122,6 +122,15 @@ def limit_pytorch_auto_parallel_threads(target_num_threads: int | None = None, i
)
+def ray_namespace() -> str:
+ """Ray namespace of the TransferQueue named actors.
+
+ Defaults to ``transfer_queue``, shared by every job on the Ray cluster (see #115). Set
+ ``TQ_RAY_NAMESPACE`` in every process of a job to keep concurrent jobs apart.
+ """
+ return os.environ.get("TQ_RAY_NAMESPACE") or "transfer_queue"
+
+
def get_env_bool(env_key: str, default: bool = False) -> bool:
"""Robustly get a boolean from an environment variable."""
env_value = os.getenv(env_key)
From fd8a05fda130dec97684895e77a9c2ec0cede8e8 Mon Sep 17 00:00:00 2001
From: nataliekung <4054699+nataliekung@users.noreply.github.com>
Date: Thu, 29 Oct 2026 16:16:04 -0700
Subject: [PATCH 2/3] [fix] Use the configured Ray namespace in the streaming
tutorial and relax demo
Both look up the controller by name with the hard-coded transfer_queue namespace, so they would miss the controller once TQ_RAY_NAMESPACE is set.
Signed-off-by: nataliekung <4054699+nataliekung@users.noreply.github.com>
---
recipe/simple_use_case/relax_demo.py | 3 ++-
tutorial/06_streaming_dataloader.py | 3 ++-
2 files changed, 4 insertions(+), 2 deletions(-)
diff --git a/recipe/simple_use_case/relax_demo.py b/recipe/simple_use_case/relax_demo.py
index 7e3efa83..6cb1766e 100644
--- a/recipe/simple_use_case/relax_demo.py
+++ b/recipe/simple_use_case/relax_demo.py
@@ -40,6 +40,7 @@
import transfer_queue as tq
from transfer_queue import RankAwareSampler, StreamingDataLoader, StreamingDataset
from transfer_queue.metadata import BatchMeta
+from transfer_queue.utils.common import ray_namespace
from transfer_queue.utils.logging_utils import get_logger
logger = get_logger("MAIN", default_level="INFO")
@@ -131,7 +132,7 @@ class BaseStageWorker:
def __init__(self, tq_config, tracker, worker_id: int, config: "DemoConfig"):
tq.init(tq_config)
self.tq_client = tq.get_client()
- controller = ray.get_actor("TransferQueueController", namespace="transfer_queue")
+ controller = ray.get_actor("TransferQueueController", namespace=ray_namespace())
self.cfg = ray.get(controller.get_config.remote())
self.tracker = tracker
self.worker_id = worker_id
diff --git a/tutorial/06_streaming_dataloader.py b/tutorial/06_streaming_dataloader.py
index 0571e108..5b5caee5 100644
--- a/tutorial/06_streaming_dataloader.py
+++ b/tutorial/06_streaming_dataloader.py
@@ -70,6 +70,7 @@
StreamingDataLoader,
StreamingDataset,
)
+from transfer_queue.utils.common import ray_namespace # noqa: E402
def setup_transfer_queue():
@@ -184,7 +185,7 @@ def update_worker(
# Step 1: Create StreamingDataset
# This dataset integrates with TransferQueue and handles batch retrieval
- controller = ray.get_actor("TransferQueueController", namespace="transfer_queue")
+ controller = ray.get_actor("TransferQueueController", namespace=ray_namespace())
config = ray.get(controller.get_config.remote())
dataset = StreamingDataset(
From 412a1c3e10e11c76a2cb191aea41c9f7e6f0b552 Mon Sep 17 00:00:00 2001
From: nataliekung <4054699+nataliekung@users.noreply.github.com>
Date: Fri, 30 Oct 2026 05:35:36 -0700
Subject: [PATCH 3/3] [fix] Drop the README section and the namespace test
Per review: keep README unchanged and add no new test for this option. ray_namespace() documents TQ_RAY_NAMESPACE.
Signed-off-by: nataliekung <4054699+nataliekung@users.noreply.github.com>
---
README.md | 4 ---
tests/test_ray_namespace.py | 51 -------------------------------------
2 files changed, 55 deletions(-)
delete mode 100644 tests/test_ray_namespace.py
diff --git a/README.md b/README.md
index 13f7b32a..2b7cd65f 100644
--- a/README.md
+++ b/README.md
@@ -205,10 +205,6 @@ Developers can leverage `TransferQueueClient` directly to implement advanced fea
- [tutorial/05_custom_sampler.py](https://github.com/Ascend/TransferQueue/blob/main/tutorial/05_custom_sampler.py)
-#### Concurrent jobs on one Ray cluster
-
-The controller and `RayStorageClient`'s storage actor are named actors in the Ray namespace `transfer_queue`, so every job on the cluster that calls `tq.init()` attaches to the same controller. This is what lets a trainer job connect to TransferQueue created by another job. Jobs that must not share data set `TQ_RAY_NAMESPACE` to a value unique per job (for example the job id) in every process of the job, for instance through the job-level `runtime_env` `env_vars`. Without it, a second job reads the first job's partitions, and the first job's `tq.close()` kills the controller the second one is using.
-
🔥 Showcases
### Collocated Example
diff --git a/tests/test_ray_namespace.py b/tests/test_ray_namespace.py
deleted file mode 100644
index 3f1c5e63..00000000
--- a/tests/test_ray_namespace.py
+++ /dev/null
@@ -1,51 +0,0 @@
-# Copyright 2025 The TransferQueue Team
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-"""TQ_RAY_NAMESPACE moves the named actors out of the cluster-wide ``transfer_queue`` namespace."""
-
-import pytest
-import ray
-from omegaconf import OmegaConf
-
-import transfer_queue as tq
-from transfer_queue.utils.common import ray_namespace
-
-
-@pytest.mark.parametrize("value", [None, ""])
-def test_default_namespace_is_shared(monkeypatch, value):
- if value is None:
- monkeypatch.delenv("TQ_RAY_NAMESPACE", raising=False)
- else:
- monkeypatch.setenv("TQ_RAY_NAMESPACE", value)
- assert ray_namespace() == "transfer_queue"
-
-
-def test_env_overrides_namespace(monkeypatch):
- monkeypatch.setenv("TQ_RAY_NAMESPACE", "job-a")
- assert ray_namespace() == "job-a"
-
-
-def test_controller_lives_in_configured_namespace(monkeypatch):
- monkeypatch.setenv("TQ_RAY_NAMESPACE", "job-a")
- ray.init(ignore_reinit_error=True, num_cpus=4)
- try:
- tq.init(OmegaConf.create({"backend": {"SimpleStorage": {"num_data_storage_units": 1}}}))
- try:
- ray.get_actor("TransferQueueController", namespace="job-a")
- with pytest.raises(ValueError):
- ray.get_actor("TransferQueueController", namespace="transfer_queue")
- finally:
- tq.close()
- finally:
- ray.shutdown()