diff --git a/docs/source/examples/nested_rdmc.ymmsl b/docs/source/examples/nested_rdmc.ymmsl index 1a465835..ac563d70 100644 --- a/docs/source/examples/nested_rdmc.ymmsl +++ b/docs/source/examples/nested_rdmc.ymmsl @@ -31,8 +31,9 @@ models: qmc: ports: - o_i: settings_out # vector port, Settings - s: final_state_in # vector port, 1D array + timeline mc:rr: + o_i: settings_out # vector port, Settings + s: final_state_in # vector port, 1D array components: mc: @@ -65,8 +66,9 @@ models: components: uq: ports: - o_i: settings_out # vector port, Settings - s: final_state_in # vecotr port, 1D array + timeline mc:rr: + o_i: settings_out # vector port, Settings + s: final_state_in # vector port, 1D array description: | Implements a quasi-Monte Carlo uncertainty quantification method. implementation: qmc diff --git a/docs/source/examples/nested_rdmc_imports.ymmsl b/docs/source/examples/nested_rdmc_imports.ymmsl index 67ef1905..bbb642e0 100644 --- a/docs/source/examples/nested_rdmc_imports.ymmsl +++ b/docs/source/examples/nested_rdmc_imports.ymmsl @@ -9,8 +9,9 @@ models: components: uq: ports: - o_i: settings_out # vector port, Settings - s: final_state_in # vecotr port, 1D array + timeline mc:rr: + o_i: settings_out # vector port, Settings + s: final_state_in # vecotr port, 1D array description: | Implements a quasi-Monte Carlo uncertainty quantification method. implementation: qmc diff --git a/docs/source/examples/ymmsl/uncertainty.ymmsl b/docs/source/examples/ymmsl/uncertainty.ymmsl index 99ff6731..ddeea235 100644 --- a/docs/source/examples/ymmsl/uncertainty.ymmsl +++ b/docs/source/examples/ymmsl/uncertainty.ymmsl @@ -7,8 +7,9 @@ imports: models: qmc: ports: - o_i: settings_out # vector port, Settings - s: final_state_in # vector port, 1D array + timeline mc:rr: + o_i: settings_out # vector port, Settings + s: final_state_in # vector port, 1D array components: mc: diff --git a/integration_test/test_registration.py b/integration_test/test_registration.py index e4865508..2f38c2ff 100644 --- a/integration_test/test_registration.py +++ b/integration_test/test_registration.py @@ -7,7 +7,7 @@ def test_registration(log_file_in_tmpdir, mmp_server): - instance_name = Reference("test_instance") + instance_name = Reference("macro") client = MMPClient(instance_name, mmp_server.get_location()) port = Port(Reference("test_in"), Operator.S) diff --git a/integration_test/test_snapshot_complex_coupling.py b/integration_test/test_snapshot_complex_coupling.py index e685a520..040807d0 100644 --- a/integration_test/test_snapshot_complex_coupling.py +++ b/integration_test/test_snapshot_complex_coupling.py @@ -139,7 +139,7 @@ def config(): implementation: main_component cacheA: ports: - f_init: in1 + f_init: in1 in2 o_i: sub_out1 sub_out2 s: sub_in1 sub_in2 o_f: out1 out2 @@ -155,7 +155,7 @@ def config(): implementation: cache_component cacheC: ports: - f_init: in1 + f_init: in1 in2 o_i: sub_out1 sub_out2 s: sub_in1 o_f: out1 diff --git a/integration_test/test_snapshot_interact.py b/integration_test/test_snapshot_interact.py index 1dee5c54..c980adce 100644 --- a/integration_test/test_snapshot_interact.py +++ b/integration_test/test_snapshot_interact.py @@ -80,8 +80,9 @@ def test_snapshot_interact_lockstep(tmp_path): implementation: component comp2: ports: - o_i: o_i - s: s + timeline comp1: + o_i: o_i + s: s description: The second component implementation: component conduits: @@ -145,10 +146,10 @@ def test_snapshot_interact_varstep(tmp_path, scale): implementation: component coupler: ports: - timeline timeline_a: + timeline comp1: o_i: a_in s: a_out - timeline timeline_b: + timeline comp2: o_i: b_in s: b_out description: > diff --git a/src/cpp/libmuscle/mmp_client.cpp b/src/cpp/libmuscle/mmp_client.cpp index 86259b7d..ece5082d 100644 --- a/src/cpp/libmuscle/mmp_client.cpp +++ b/src/cpp/libmuscle/mmp_client.cpp @@ -196,6 +196,8 @@ void MMPClient::register_instance( if (response[0].as() == static_cast(ResponseType::error)) throw std::runtime_error( "Error registering instance: " + response[1].as()); + + timeline_ = ymmsl::Timeline(response[1].as()); } ymmsl::Settings MMPClient::get_settings() { @@ -363,6 +365,12 @@ bool MMPClient::is_deadlocked() { return response[1].as(); } +ymmsl::Timeline const & MMPClient::get_timeline() const { + if (!timeline_.is_set()) + throw std::logic_error("Cannot get timeline before instance is registered."); + return timeline_.get(); +} + DataConstRef MMPClient::call_manager_(DataConstRef const & request, bool timid) { std::unique_lock lock(mutex_, std::defer_lock); diff --git a/src/cpp/libmuscle/mmp_client.hpp b/src/cpp/libmuscle/mmp_client.hpp index 93b9b229..96f01ff7 100644 --- a/src/cpp/libmuscle/mmp_client.hpp +++ b/src/cpp/libmuscle/mmp_client.hpp @@ -133,12 +133,16 @@ class MMPClient { /** Ask the manager if we're part of a deadlock. */ bool is_deadlocked(); + ymmsl::Timeline const & get_timeline() const; + private: ymmsl::Reference instance_id_; mcp::TcpTransportClient transport_client_; mutable std::recursive_timed_mutex mutex_; std::thread::id cur_owner_; + Optional timeline_; + /* Helper function that encodes/decodes and calls the manager. */ DataConstRef call_manager_(DataConstRef const & request, bool timid = false); diff --git a/src/cpp/libmuscle/tests/mocks/mock_mmp_client.hpp b/src/cpp/libmuscle/tests/mocks/mock_mmp_client.hpp index 785001a4..b6a0153c 100644 --- a/src/cpp/libmuscle/tests/mocks/mock_mmp_client.hpp +++ b/src/cpp/libmuscle/tests/mocks/mock_mmp_client.hpp @@ -39,6 +39,7 @@ class MockMMPClient : public MockClass { NAME_MOCK_MEM_FUN(MockMMPClient, waiting_for_receive); NAME_MOCK_MEM_FUN(MockMMPClient, waiting_for_receive_done); NAME_MOCK_MEM_FUN(MockMMPClient, is_deadlocked); + NAME_MOCK_MEM_FUN(MockMMPClient, get_timeline); // Create some empty return objects for return values with a complex // structure, to make it easier to set them in the tests or fixtures. @@ -55,6 +56,8 @@ class MockMMPClient : public MockClass { "simulation_time", Data::list()), Optional(), Optional()); + + get_timeline.return_value = ::ymmsl::Timeline(":"); } MockMMPClient() { @@ -102,6 +105,8 @@ class MockMMPClient : public MockClass { > waiting_for_receive_done; MockFun> is_deadlocked; + + MockFun> get_timeline; }; using MMPClient = MockMMPClient; diff --git a/src/python/libmuscle/manager/manager.py b/src/python/libmuscle/manager/manager.py index 04fee5cb..db81bc7d 100644 --- a/src/python/libmuscle/manager/manager.py +++ b/src/python/libmuscle/manager/manager.py @@ -5,7 +5,7 @@ from typing import Optional from ymmsl import save as save_ymmsl -from ymmsl.v0_2 import Configuration +from ymmsl.v0_2 import Configuration, resolve_timelines import libmuscle from libmuscle.manager.deadlock_detector import DeadlockDetector @@ -56,6 +56,9 @@ def __init__( _logger.info("libmuscle version: %s", libmuscle.__version__) + # Ensure timelines are consistent + resolve_timelines(configuration.root_model()) + if run_dir is not None: snapshot_dir = run_dir.snapshot_dir() else: diff --git a/src/python/libmuscle/manager/mmp_server.py b/src/python/libmuscle/manager/mmp_server.py index b17d3b5f..9c4cb1d2 100644 --- a/src/python/libmuscle/manager/mmp_server.py +++ b/src/python/libmuscle/manager/mmp_server.py @@ -201,9 +201,10 @@ def _register_instance( instance = Reference(instance_id) try: self._instance_registry.add(instance, locations, port_objs) + timeline = self._topology_store.get_timeline(instance) _logger.info(f"Registered instance {instance_id}") - return [ResponseType.SUCCESS.value] + return [ResponseType.SUCCESS.value, str(timeline)] except AlreadyRegistered: return [ ResponseType.ERROR.value, diff --git a/src/python/libmuscle/manager/test/conftest.py b/src/python/libmuscle/manager/test/conftest.py index 9b500d8d..ad314866 100644 --- a/src/python/libmuscle/manager/test/conftest.py +++ b/src/python/libmuscle/manager/test/conftest.py @@ -1,5 +1,13 @@ import pytest -from ymmsl.v0_2 import Component, Conduit, Configuration, Model, Ports, Reference +from ymmsl.v0_2 import ( + Component, + Conduit, + Configuration, + Model, + Ports, + Reference, + resolve_timelines, +) from libmuscle.manager.deadlock_detector import DeadlockDetector from libmuscle.manager.instance_registry import InstanceRegistry @@ -20,7 +28,7 @@ def logger(tmp_path): @pytest.fixture def mmp_configuration(): - return Configuration( + config = Configuration( "mmp_configuration", [], [ @@ -49,6 +57,8 @@ def mmp_configuration(): ) ], ) + resolve_timelines(config.models["test_model"]) + return config @pytest.fixture diff --git a/src/python/libmuscle/manager/test/test_mmp_request_handler.py b/src/python/libmuscle/manager/test/test_mmp_request_handler.py index e452b75f..8d6b5968 100644 --- a/src/python/libmuscle/manager/test/test_mmp_request_handler.py +++ b/src/python/libmuscle/manager/test/test_mmp_request_handler.py @@ -100,7 +100,7 @@ def test_get_settings(mmp_configuration, mmp_request_handler): def test_register_instance(mmp_request_handler, instance_registry): request = [ RequestType.REGISTER_INSTANCE.value, - "test_instance", + "macro", ["tcp://localhost:10000"], [["test_in", "F_INIT"]], libmuscle.__version__, @@ -111,17 +111,17 @@ def test_register_instance(mmp_request_handler, instance_registry): decoded_result = msgpack.unpackb(result, raw=False) assert decoded_result[0] == ResponseType.SUCCESS.value - assert instance_registry._locations["test_instance"] == ["tcp://localhost:10000"] + assert instance_registry._locations["macro"] == ["tcp://localhost:10000"] registered_ports = instance_registry._ports - assert registered_ports["test_instance"][0].name == "test_in" - assert registered_ports["test_instance"][0].operator == Operator.F_INIT + assert registered_ports["macro"][0].name == "test_in" + assert registered_ports["macro"][0].operator == Operator.F_INIT def test_register_instance_no_version(mmp_request_handler): request = [ RequestType.REGISTER_INSTANCE.value, - "test_instance", + "macro", ["tcp://localhost:10000"], [["test_in", "F_INIT"]], ] @@ -137,7 +137,7 @@ def test_register_instance_no_version(mmp_request_handler): def test_register_instance_version_mismatch(mmp_request_handler): request = [ RequestType.REGISTER_INSTANCE.value, - "test_instance", + "macro", ["tcp://localhost:10000"], [["test_in", "F_INIT"]], libmuscle.__version__ + "dev", @@ -153,12 +153,12 @@ def test_register_instance_version_mismatch(mmp_request_handler): def test_get_checkpoint_info(mmp_configuration, mmp_request_handler): resume_path = Path("/path/to/resume.pack") - mmp_configuration.resume = {Reference("test_instance"): resume_path} + mmp_configuration.resume = {Reference("macro"): resume_path} mmp_configuration.checkpoints = Checkpoints( True, [CheckpointRangeRule(every=10), CheckpointAtRule([1, 2, 3.0])] ) - request = [RequestType.GET_CHECKPOINT_INFO.value, "test_instance"] + request = [RequestType.GET_CHECKPOINT_INFO.value, "macro"] encoded_request = msgpack.packb(request, use_bin_type=True) result = mmp_request_handler.handle_request(encoded_request) @@ -189,7 +189,7 @@ def test_get_checkpoint_info(mmp_configuration, mmp_request_handler): def test_get_checkpoint_info2(registered_mmp_request_handler2, tmp_path): - request = [RequestType.GET_CHECKPOINT_INFO.value, "test_instance"] + request = [RequestType.GET_CHECKPOINT_INFO.value, "macro"] encoded_request = msgpack.packb(request, use_bin_type=True) result = registered_mmp_request_handler2.handle_request(encoded_request) @@ -197,13 +197,13 @@ def test_get_checkpoint_info2(registered_mmp_request_handler2, tmp_path): assert decoded_result[0] == ResponseType.SUCCESS.value snapshot_directory = decoded_result[4] - assert snapshot_directory == (str(tmp_path) + "/instances/test_instance/snapshots") + assert snapshot_directory == (str(tmp_path) + "/instances/macro/snapshots") def test_double_register_instance(mmp_request_handler): request = [ RequestType.REGISTER_INSTANCE.value, - "test_instance", + "macro", ["tcp://localhost:10000"], [["test_in", "F_INIT"]], libmuscle.__version__, @@ -219,7 +219,7 @@ def test_double_register_instance(mmp_request_handler): decoded_result = msgpack.unpackb(result, raw=False) assert decoded_result[0] == ResponseType.ERROR.value - assert "test_instance" in decoded_result[1] + assert "macro" in decoded_result[1] def test_deregister_instance(registered_mmp_request_handler, instance_registry): @@ -266,7 +266,7 @@ def test_request_peers_fanout(registered_mmp_request_handler): assert name == f"micro[{i // 10}][{i % 10}]" assert locs == [f"direct:{name}"] - assert ports == [["out", "O_I", ""], ["in", "S", ""]] + assert ports == [["out", "O_I", "macro"], ["in", "S", "macro"]] def test_request_peers_fanin(registered_mmp_request_handler): diff --git a/src/python/libmuscle/manager/topology_store.py b/src/python/libmuscle/manager/topology_store.py index fb0cedf6..63f7118d 100644 --- a/src/python/libmuscle/manager/topology_store.py +++ b/src/python/libmuscle/manager/topology_store.py @@ -1,4 +1,4 @@ -from ymmsl.v0_2 import Conduit, Configuration, Ports, Reference +from ymmsl.v0_2 import Conduit, Configuration, Ports, Reference, Timeline from libmuscle.util import generate_indices, instance_indices @@ -108,3 +108,10 @@ def get_peer_instances(self, instance: Reference) -> list[Reference]: for peer_indices in generate_indices(peer_dims[len(dims) :]): peers.append(base + peer_indices) return peers + + def get_timeline(self, instance: Reference) -> Timeline: + """Get the timeline for the provided instance.""" + component = instance.without_trailing_ints() + timeline = self.model.components[component].timeline + assert timeline is not None + return timeline diff --git a/src/python/libmuscle/mmp_client.py b/src/python/libmuscle/mmp_client.py index 0fcd189f..e90f4321 100644 --- a/src/python/libmuscle/mmp_client.py +++ b/src/python/libmuscle/mmp_client.py @@ -262,6 +262,7 @@ def register_instance(self, locations: list[str], ports: list[Port]) -> None: response = self._call_manager(request) if response[0] == ResponseType.ERROR.value: raise RuntimeError(f"Error registering instance: {response[1]}") + self._timeline = Timeline(response[1]) def request_peers(self) -> PeerInfo: """Request connection information about peers. @@ -360,6 +361,10 @@ def is_deadlocked(self) -> bool: response = self._call_manager(request) return bool(response[1]) + def get_timeline(self) -> Timeline: + """Ask the manager what our (absolute) timeline is.""" + return self._timeline + def _call_manager(self, request: Any, timid: bool = False) -> Any: """Call the manager and do en/decoding. diff --git a/src/python/libmuscle/pytest/implementation_tester.py b/src/python/libmuscle/pytest/implementation_tester.py index 0e3710f8..621043a6 100644 --- a/src/python/libmuscle/pytest/implementation_tester.py +++ b/src/python/libmuscle/pytest/implementation_tester.py @@ -2,7 +2,7 @@ import os from typing import Optional -from ymmsl.v0_2 import Configuration, Operator, Reference +from ymmsl.v0_2 import Configuration, Operator, Reference, Settings from libmuscle import Instance, Message @@ -46,6 +46,12 @@ def __init__( self._instance._communicator.set_receive_timeout(default_timeout) self._instance.reuse_instance() + # We have a __settings_in__ port if the tested component doesn't have F_INIT + # ports: send an empty message on it to make the timeline logic work out: + if "__settings_in__" in test_model.ports: + msg = Message(float("-inf"), data=Settings()) + self._instance.send("__settings_in__", msg) + def send( self, port_name: str, message: Message, slot: Optional[int] = None ) -> None: @@ -53,10 +59,12 @@ def send( Send a message on the specified port. Args: - port_name: Name of the port to send on (without 'send_' prefix). + port_name: Name of the port to send on. message: The message to send. slot: Optional slot number for vector ports. """ + if port_name == "muscle_settings_in": + port_name = "__settings_in__" self._instance.send(port_name, message, slot) def receive( @@ -70,7 +78,7 @@ def receive( Receive a message from the specified port. Args: - port_name: Name of the port to receive from (without 'receive_' prefix). + port_name: Name of the port to receive from. slot: Optional slot number for vector ports. timeout: Timeout in seconds. If None, uses default_timeout. diff --git a/src/python/libmuscle/pytest/muscle_tester.py b/src/python/libmuscle/pytest/muscle_tester.py index f323e4a6..cee6afff 100644 --- a/src/python/libmuscle/pytest/muscle_tester.py +++ b/src/python/libmuscle/pytest/muscle_tester.py @@ -14,17 +14,21 @@ Conduit, Configuration, ExecutionModel, + Identifier, Implementation, Model, + Operator, + Port, Ports, Program, Reference, - ThreadedResReq, + Timeline, ) from libmuscle.manager.manager import Manager from libmuscle.manager.run_dir import RunDir from libmuscle.mcp.tcp_transport_client import RECONNECT_TIMEOUT +from libmuscle.mcp.tcp_transport_server import TcpTransportServer from libmuscle.mmp_client import PEER_TIMEOUT from libmuscle.pytest.implementation_tester import ImplementationTester from libmuscle.receive_timeout_handler import ReceiveTimeoutHandler @@ -82,33 +86,48 @@ def _add_tester_component( ) tester_name = "muscle3_implementation_tester" + tester_timeline = Timeline(tester_name) + tester_ports: list[Port] = [] test_model_name = "muscle3_test_model" - tester_o_i_ports = [] - tester_s_ports = [] tester_model = Model(name=test_model_name) - # Inputs of target → tester sends (O_I) - for port_name in implementation.ports.receiving_port_names(): - tester_o_i_ports.append(f"{port_name}") + # Generate ports and conduits + for port in implementation.ports.values(): + tester_port = f"{tester_name}.{port.name}" + implementation_port = f"{implementation_name}.{port.name}" + if port.operator.allows_receiving(): + conduit = Conduit(tester_port, implementation_port) + tester_operator = Operator.O_I + else: + conduit = Conduit(implementation_port, tester_port) + tester_operator = Operator.S + if port.operator in (Operator.O_I, Operator.S): + port_timeline = port.timeline or Timeline(implementation_name) + timeline = tester_timeline + port_timeline + else: + timeline = tester_timeline + + tester_model.conduits.append(conduit) + tester_ports.append(Port(port.name, tester_operator, timeline)) + + if not any( + p.operator is Operator.F_INIT for p in implementation.ports.values() + ): + # We'll connect muscle_settings_in to make the timeline logic work tester_model.conduits.append( Conduit( - f"{tester_name}.{port_name}", f"{implementation_name}.{port_name}" + f"{tester_name}.__settings_in__", + f"{implementation_name}.muscle_settings_in", ) ) - - # Outputs of target → tester receives (S) - for port_name in implementation.ports.sending_port_names(): - tester_s_ports.append(f"{port_name}") - tester_model.conduits.append( - Conduit( - f"{implementation_name}.{port_name}", f"{tester_name}.{port_name}" - ) + tester_ports.append( + Port(Identifier("__settings_in__"), Operator.O_I, tester_timeline) ) tester_model.components[Reference(tester_name)] = Component( name=tester_name, - ports=Ports(o_i=tester_o_i_ports, s=tester_s_ports), + ports=Ports(tester_ports), description="Tester component for implementation testing", implementation=tester_name, optional=False, @@ -124,15 +143,11 @@ def _add_tester_component( config.programs[Reference(tester_name)] = Program( name=tester_name, - ports=Ports(o_i=tester_o_i_ports, s=tester_s_ports), + ports=Ports(tester_ports), execution_model=ExecutionModel.MANUAL, description="Manual tester program for implementation testing", ) - config.resources[Reference(tester_name)] = ThreadedResReq( - name=Reference(tester_name), threads=1 - ) - config.models[Reference(test_model_name)] = tester_model return config @@ -193,6 +208,18 @@ def start_implementation( "libmuscle.mmp_client.PEER_TIMEOUT", min(PEER_TIMEOUT, default_timeout) ) ) + # Ensure we won't wait forever on our outboxes + self._exitstack.enter_context( + patch("libmuscle.post_office.PostOffice.wait_for_receivers") + ) + # And we close() our TCP Servers ungracefully + origclose = TcpTransportServer.close + self._exitstack.enter_context( + patch.multiple( + TcpTransportServer, + close=lambda self, _=True: origclose(self, False), + ) + ) self.implementation_tester = ImplementationTester( default_timeout, muscle_manager_address, test_ymmsl_config ) diff --git a/src/python/libmuscle/test/test_mmp_client.py b/src/python/libmuscle/test/test_mmp_client.py index d461362f..547d880b 100644 --- a/src/python/libmuscle/test/test_mmp_client.py +++ b/src/python/libmuscle/test/test_mmp_client.py @@ -82,7 +82,7 @@ def test_get_settings(mocked_mmp_client, profile_data) -> None: def test_register_instance(mocked_mmp_client, profile_data) -> None: client, stub = mocked_mmp_client - result = [ResponseType.SUCCESS.value] + result = [ResponseType.SUCCESS.value, ":"] stub.call.return_value = (msgpack.packb(result, use_bin_type=True), profile_data) client.register_instance( diff --git a/src/python/muscle3/muscle_manager.py b/src/python/muscle3/muscle_manager.py index bffbaff7..30abe587 100644 --- a/src/python/muscle3/muscle_manager.py +++ b/src/python/muscle3/muscle_manager.py @@ -153,6 +153,13 @@ def _manage_simulation( ) sys.exit(1) + for model_obj in configuration.models.values(): + try: + v0_2.check_timelines(model_obj) + except RuntimeError as exc: + print(exc, file=sys.stderr) + sys.exit(1) + # find root models, error if multiple model_ref: Optional[v0_2.Reference] = None if model: