Muscle manager: check timelines

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

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

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:

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)
 

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

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: >

src/cpp/libmuscle/mmp_client.cpp

@@ -196,6 +196,8 @@ void MMPClient::register_instance(
     if (response[0].as<int>() == static_cast<int>(ResponseType::error))
         throw std::runtime_error(
                 "Error registering instance: " + response[1].as<std::string>());
+
+    timeline_ = ymmsl::Timeline(response[1].as<std::string>());
 }
 
 ymmsl::Settings MMPClient::get_settings() {
@@ -363,6 +365,12 @@ bool MMPClient::is_deadlocked() {
     return response[1].as<bool>();
 }
 
+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<std::recursive_timed_mutex> lock(mutex_, std::defer_lock);
 

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<ymmsl::Timeline> timeline_;
+
         /* Helper function that encodes/decodes and calls the manager.
          */
         DataConstRef call_manager_(DataConstRef const & request, bool timid = false);

src/cpp/libmuscle/tests/mocks/mock_mmp_client.hpp

@@ -39,6 +39,7 @@ class MockMMPClient : public MockClass<MockMMPClient> {
             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<MockMMPClient> {
                             "simulation_time", Data::list()),
                     Optional<std::string>(),
                     Optional<std::string>());
+
+            get_timeline.return_value = ::ymmsl::Timeline(":");
         }
 
         MockMMPClient() {
@@ -102,6 +105,8 @@ class MockMMPClient : public MockClass<MockMMPClient> {
             > waiting_for_receive_done;
 
         MockFun<Val<bool>> is_deadlocked;
+
+        MockFun<Val<::ymmsl::Timeline>> get_timeline;
 };
 
 using MMPClient = MockMMPClient;

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:

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,

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

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,21 +189,21 @@ 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)
     decoded_result = msgpack.unpackb(result, raw=False)
 
     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):

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

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.