B7: an execution client reaches only a server it may use

benchmark/matched_cellprofiler_batch.py

@@ -30,7 +30,6 @@
     ensure_global_config_context,
     rebuild_lazy_config_with_new_global_reference,
 )
-from zmqruntime import DataControlPortPairAuthority
 from zmqruntime.messages import ExecutionStatusSnapshot
 
 from benchmark.adapters.cellprofiler import (
@@ -114,7 +113,6 @@
 from openhcs.interop.cellprofiler.plate_workspace import (
     prepare_cellprofiler_input_workspace,
 )
-from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG
 from openhcs.runtime.zmq_execution_client import (
     OpenHCSExecutionSubmission,
     ZMQExecutionClient,
@@ -1206,11 +1204,7 @@ def main(argv: list[str] | None = None) -> int:
     if args.preflight_native_references:
         execution_scope = nullcontext(None)
     else:
-        port = DataControlPortPairAuthority.acquire(
-            OPENHCS_ZMQ_CONFIG,
-            transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode,
-        ).data_port
-        execution_scope = ZMQExecutionClient(port=port, persistent=False)
+        execution_scope = ZMQExecutionClient.with_private_server()
     receipts = {}
     with execution_scope as client:
         for case in selected_cases:

benchmark/openhcs_measured_run.py

@@ -8,11 +8,14 @@
 import sys
 import time
 from collections.abc import Callable, Mapping
-from dataclasses import dataclass
+from contextlib import AbstractContextManager
+from dataclasses import (
+    dataclass,
+    field,
+)
 from pathlib import Path
 from typing import Any
 
-from zmqruntime import DataControlPortPairAuthority
 from zmqruntime.messages import PongResponse
 
 from benchmark.contracts.measured_run_receipt import (
@@ -31,7 +34,6 @@
     RuntimeArtifactExecutionObservation,
 )
 from openhcs.runtime.zmq_application import OPENHCS_ENDPOINT_APPLICATION
-from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG
 from openhcs.runtime.zmq_execution_client import (
     ZMQExecutionRequestBuilder,
     OpenHCSExecutionSubmission,
@@ -251,6 +253,29 @@ def _require_measured_submission(
         raise ValueError("Measured OpenHCS observation export path must be absolute.")
 
 
+def measured_execution_client(
+    *,
+    execution_port: int | None,
+    progress_callback=None,
+) -> AbstractContextManager[ZMQExecutionClient]:
+    """The client of one measured run.
+
+    With no port the run starts a server at an address of its own. A port is
+    an address the caller named; the client still refuses a server there that
+    another process started for its own use or that runs other code.
+    """
+
+    if execution_port is None:
+        return ZMQExecutionClient.with_private_server(
+            progress_callback=progress_callback
+        )
+    return ZMQExecutionClient(
+        port=execution_port,
+        persistent=False,
+        progress_callback=progress_callback,
+    )
+
+
 def execute_measured_openhcs_pipeline(
     *,
     submission: OpenHCSExecutionSubmission,
@@ -263,18 +288,9 @@ def execute_measured_openhcs_pipeline(
     """Measure an ordinary pipeline submission with its requested observation."""
 
     _require_measured_submission(submission, expected_axis_count)
-    client_port = execution_port
-    if client_port is None and require_owned_server:
-        client_port = DataControlPortPairAuthority.acquire(
-            OPENHCS_ZMQ_CONFIG,
-            transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode,
-        ).data_port
-    client = ZMQExecutionClient(
-        port=client_port,
-        persistent=False,
-        progress_callback=timing_observer,
-    )
-    with client:
+    with measured_execution_client(
+        execution_port=execution_port, progress_callback=timing_observer
+    ) as client:
         return execute_measured_openhcs_pipeline_on_client(
             client=client,
             submission=submission,

benchmark/well_throughput_scaling.py

@@ -17,7 +17,6 @@
 
 import psutil
 from objectstate.lazy_factory import rebuild_lazy_config_with_new_global_reference
-from zmqruntime import DataControlPortPairAuthority
 
 if TYPE_CHECKING:
     from benchmark.reports.cppipe_figures import BenchmarkMetricRow
@@ -34,6 +33,7 @@
     _ZMQProgressTimingObserver,
     execute_measured_openhcs_pipeline,
     execute_measured_openhcs_pipeline_on_client,
+    measured_execution_client,
 )
 from benchmark.timing import BenchmarkPhase, PhaseTimingTrace
 from openhcs.core.components.parser_metaprogramming import FilenameParseResult
@@ -68,7 +68,6 @@
     prepare_cellprofiler_input_workspace,
 )
 from openhcs.core.dataset_sources.source_schema import SourceSchemaFilenameParser
-from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG
 from openhcs.runtime.zmq_execution_client import (
     OpenHCSExecutionSubmission,
     ZMQExecutionClient,
@@ -2001,16 +2000,9 @@ def run_well_throughput_suite(
                 if observation_key in completed or observation_key in skipped:
                     continue
                 if reuse_execution_server and shared_client is None:
-                    client_port = execution_port
-                    if client_port is None:
-                        client_port = DataControlPortPairAuthority.acquire(
-                            OPENHCS_ZMQ_CONFIG,
-                            transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode,
-                        ).data_port
                     shared_client = client_stack.enter_context(
-                        ZMQExecutionClient(
-                            port=client_port,
-                            persistent=False,
+                        measured_execution_client(
+                            execution_port=execution_port,
                             progress_callback=shared_observer,
                         )
                     )