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
34 changes: 32 additions & 2 deletions src/zmqruntime/viewer_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
import logging
import threading
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple
Expand Down Expand Up @@ -61,6 +62,21 @@ def is_healthy(self) -> bool:
VisualizerFactory = Callable[[], "VisualizerProcessManager"]


class ViewerReuseAdmissionABC(ABC):
"""Caller-owned constraints admitted inside atomic viewer acquisition."""

@abstractmethod
def require_reusable(self, visualizer: "VisualizerProcessManager") -> None:
"""Raise without changing the existing instance when reuse is forbidden."""


class HealthyViewerReuseAdmission(ViewerReuseAdmissionABC):
"""Generic callers require health but declare no application launch settings."""

def require_reusable(self, visualizer: "VisualizerProcessManager") -> None:
pass


class ViewerStateManager:
"""
Singleton manager for viewer lifecycle state.
Expand Down Expand Up @@ -123,6 +139,8 @@ def get_or_create_viewer(
factory: VisualizerFactory,
wait_for_ready: bool = True,
ready_timeout: float = 30.0,
*,
reuse_admission: ViewerReuseAdmissionABC = HealthyViewerReuseAdmission(),
) -> Tuple["VisualizerProcessManager", bool]:
"""
Atomically get existing viewer or create a new one.
Expand All @@ -136,6 +154,9 @@ def get_or_create_viewer(
factory: Callable that creates a VisualizerProcessManager
wait_for_ready: If True, block until viewer is ready (or timeout)
ready_timeout: Maximum seconds to wait for ready state
reuse_admission: Application constraints checked under the acquisition
lock before returning a healthy existing instance. Rejection does
not stop, replace or forget the existing viewer.

Returns:
Tuple of (visualizer_instance, was_created)
Expand All @@ -152,6 +173,7 @@ def get_or_create_viewer(

if existing is not None:
if existing.is_healthy:
reuse_admission.require_reusable(existing.visualizer)
existing.last_used = time.time()
logger.debug(
"ViewerStateManager: Reusing existing %s viewer on port %d (state=%s)",
Expand Down Expand Up @@ -215,7 +237,10 @@ def _start_and_wait(
"""Start the viewer process and optionally wait for ready state."""
try:
# Start the process
instance.visualizer.start()
# A factory may attach to an already-live, externally-owned viewer.
# Registering that lifecycle must not restart its foreign process.
if not instance.visualizer.is_running:
instance.visualizer.start()
logger.info(
"ViewerStateManager: Started %s viewer on port %d",
instance.viewer_type,
Expand Down Expand Up @@ -482,6 +507,8 @@ def get_or_create_viewer(
factory: VisualizerFactory,
wait_for_ready: bool = True,
ready_timeout: float = 30.0,
*,
reuse_admission: ViewerReuseAdmissionABC = HealthyViewerReuseAdmission(),
) -> Tuple["VisualizerProcessManager", bool]:
"""
Convenience function for atomic viewer get-or-create.
Expand All @@ -494,4 +521,7 @@ def get_or_create_viewer(
)
"""
manager = ViewerStateManager.get_instance()
return manager.get_or_create_viewer(viewer_type, port, factory, wait_for_ready, ready_timeout)
return manager.get_or_create_viewer(
viewer_type, port, factory, wait_for_ready, ready_timeout,
reuse_admission=reuse_admission,
)
54 changes: 52 additions & 2 deletions tests/test_viewer_state.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import pytest

from zmqruntime.viewer_state import ViewerStateManager
from zmqruntime.viewer_state import ViewerReuseAdmissionABC, ViewerStateManager


class RecordingVisualizer:
Expand All @@ -9,11 +9,14 @@ def __init__(self, *, fail_start: bool = False) -> None:
self.stop_calls = 0
self.force_stop_calls = 0
self.ready_timeouts = []
self.running = True
self.running = False
self.start_calls = 0

def start(self):
self.start_calls += 1
if self.fail_start:
raise RuntimeError("start failed")
self.running = True
return None

def wait_for_ready(self, timeout: float) -> bool:
Expand Down Expand Up @@ -96,3 +99,50 @@ def test_viewer_state_subscription_owns_idempotent_release(viewer_manager):
viewer_manager.release_viewer("napari", 5700)

assert len(observed) == 2


def test_reuse_admission_runs_atomically_and_preserves_rejected_viewer(viewer_manager):
import threading

existing = RecordingVisualizer()
viewer_manager.get_or_create_viewer("napari", 5700, lambda: existing)
observations = []

class RejectReuse(ViewerReuseAdmissionABC):
def require_reusable(self, visualizer):
# A second thread cannot acquire the owner's lock during admission.
def probe():
acquired = viewer_manager._lock.acquire(blocking=False)
observations.append(acquired)
if acquired:
viewer_manager._lock.release()

thread = threading.Thread(target=probe)
thread.start()
thread.join(timeout=1)
assert not thread.is_alive()
assert visualizer is existing
raise ValueError("incompatible launch request")

def forbidden_factory():
raise AssertionError("rejected reuse must not construct or restart")

with pytest.raises(ValueError, match="incompatible launch request"):
viewer_manager.get_or_create_viewer(
"napari", 5700, forbidden_factory, reuse_admission=RejectReuse()
)
assert observations == [False]
assert viewer_manager.get_viewer("napari", 5700) is existing
assert existing.force_stop_calls == 0
assert existing.start_calls == 1


def test_live_external_factory_is_registered_without_start(viewer_manager):
existing = RecordingVisualizer(fail_start=True)
existing.running = True
acquired, created = viewer_manager.get_or_create_viewer(
"fiji", 5701, lambda: existing
)
assert acquired is existing and created is True
assert viewer_manager.get_viewer("fiji", 5701) is existing
assert existing.start_calls == 0
Loading