Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "hatchling.build"

[project]
name = "zmqruntime"
version = "0.6.3"
version = "0.6.4"
description = "Generic ZMQ-based distributed execution framework"
authors = [{name = "Tristan Simas", email = "tristan.simas@mail.mcgill.ca"}]
license = {text = "MIT"}
Expand All @@ -26,7 +26,7 @@ keywords = ["zmq", "zeromq", "distributed", "execution", "rpc", "messaging"]

dependencies = [
"annotated-types>=0.7.0",
"metaclass-registry>=0.1.6",
"metaclass-registry>=0.4.0",
"numpy>=1.24",
"portalocker>=2.8.2",
"pyzmq>=22.0",
Expand Down
11 changes: 4 additions & 7 deletions src/zmqruntime/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@
from collections.abc import Callable, Mapping

import zmq
from metaclass_registry import AutoRegisterMeta
from typing_extensions import Self

from zmqruntime.config import TransportMode, ZMQConfig
Expand All @@ -38,17 +37,15 @@
logger = logging.getLogger(__name__)


class ZMQServer(ABC, metaclass=AutoRegisterMeta):
class ZMQServer(ABC):
"""
ABC for ZMQ servers - dual-channel pattern with ping/pong handshake.

Registry auto-created and stored as ZMQServer.__registry__.
Subclasses auto-register by setting _server_type class attribute.
``_server_type`` is the runtime role a server reports in its pong. Several
server classes may serve one role, so servers are not registered by it.
"""

__registry_key__ = "_server_type"

_server_type: str | None = None # Override in subclasses for registration
_server_type: str | None = None # The runtime role reported in the pong
_server_role = ServerRole.GENERIC

@classmethod
Expand Down
11 changes: 9 additions & 2 deletions src/zmqruntime/transport_modes.py
Original file line number Diff line number Diff line change
Expand Up @@ -588,6 +588,13 @@ def data_control_pair_is_available(
@classmethod
@contextmanager
def private_endpoint(cls, config: ZMQConfig) -> Iterator[ZMQConfig]:
"""A new socket directory, removed when the caller is done with it."""
with tempfile.TemporaryDirectory(prefix=f"{config.app_name}-") as directory:
"""A new socket directory, removed when the caller is done with it.

The directory is created beneath ``config.socket_root()``, so the
length of its socket paths does not depend on ``TMPDIR``: a path
longer than ``zmq.IPC_PATH_MAX_LEN`` cannot be bound.
"""
root = config.socket_root()
root.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix="private-", dir=root) as directory:
yield replace(config, ipc_socket_dir=directory)
30 changes: 30 additions & 0 deletions tests/test_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,36 @@ def test_transport_mode_owns_ipc_socket_binding() -> None:
assert endpoint_socket.random_bind_requests == []


@pytest.mark.skipif(platform.system() == "Windows", reason="IPC is POSIX-only")
def test_private_ipc_endpoint_lives_under_the_socket_root_not_tmpdir(
monkeypatch, tmp_path
) -> None:
long_tmpdir = tmp_path / ("t" * 120)
long_tmpdir.mkdir()
monkeypatch.setenv("TMPDIR", str(long_tmpdir))
monkeypatch.setattr("tempfile.tempdir", None)
config = ZMQConfig(app_name=f"zmqruntime-test-{uuid.uuid4().hex[:8]}")
root = config.socket_root()
declaration = TransportMode.IPC.declaration

try:
with declaration.private_endpoint(config) as first:
with declaration.private_endpoint(config) as second:
first_dir = Path(first.ipc_socket_dir)
second_dir = Path(second.ipc_socket_dir)
assert first_dir != second_dir
assert first_dir.parent == second_dir.parent == root
assert first_dir.is_dir() and second_dir.is_dir()
socket_path = declaration.socket_path(65535, first)
assert socket_path.parent == first_dir
assert len(str(socket_path).encode()) <= zmq.IPC_PATH_MAX_LEN
assert not first_dir.exists() and not second_dir.exists()
assert list(long_tmpdir.iterdir()) == []
finally:
if root.exists():
root.rmdir()


@pytest.mark.skipif(platform.system() == "Windows", reason="IPC is POSIX-only")
def test_ipc_startup_lock_obeys_total_operation_deadline() -> None:
config = ZMQConfig(
Expand Down
Loading