Add iteration counts to MPP protocol

docs/source/examples/python/interact_coupling.py

@@ -241,8 +241,7 @@ def temporal_coupler() -> None:
     """
     instance = Instance({
         Operator.O_I: ['a_out', 'b_out'],
-        Operator.S: ['a_in', 'b_in']},
-        InstanceFlags.SKIP_MMSF_SEQUENCE_CHECKS)
+        Operator.S: ['a_in', 'b_in']})
 
     while instance.reuse_instance():
         # Receive initial messages and initialise state
@@ -277,7 +276,7 @@ def checkpointing_temporal_coupler() -> None:
     instance = Instance({
         Operator.O_I: ['a_out', 'b_out'],
         Operator.S: ['a_in', 'b_in']},
-        InstanceFlags.USES_CHECKPOINT_API | InstanceFlags.SKIP_MMSF_SEQUENCE_CHECKS)
+        InstanceFlags.USES_CHECKPOINT_API)
 
     while instance.reuse_instance():
         if instance.resuming():

integration_test/test_snapshot_complex_coupling.py

@@ -8,7 +8,6 @@
 
 from libmuscle import (
     KEEPS_NO_STATE_FOR_NEXT_USE,
-    SKIP_MMSF_SEQUENCE_CHECKS,
     USES_CHECKPOINT_API,
     Instance,
     Message,
@@ -71,7 +70,7 @@ def echo_component(max_channels=2):
         Operator.F_INIT: [f"in{i + 1}" for i in range(max_channels)],
         Operator.O_F: [f"out{i + 1}" for i in range(max_channels)],
     }
-    instance = Instance(ports, KEEPS_NO_STATE_FOR_NEXT_USE | SKIP_MMSF_SEQUENCE_CHECKS)
+    instance = Instance(ports, KEEPS_NO_STATE_FOR_NEXT_USE)
 
     while instance.reuse_instance():
         for p_in, p_out in zip(ports[Operator.F_INIT], ports[Operator.O_F]):

integration_test/test_snapshot_interact.py

@@ -145,10 +145,10 @@ def test_snapshot_interact_varstep(tmp_path, scale):
       implementation: component
     coupler:
       ports:
-        +timeline_a:
+        timeline timeline_a:
           o_i: a_in
           s: a_out
-        +timeline_b:
+        timeline timeline_b:
           o_i: b_in
           s: b_out
       description: >

libmuscle/python/libmuscle/communicator.py

@@ -14,6 +14,7 @@
 from libmuscle.profiler import Profiler
 from libmuscle.profiling import ProfileEvent, ProfileEventType, ProfileTimestamp
 from libmuscle.receive_timeout_handler import Deadlock, ReceiveTimeoutHandler
+from libmuscle.timeline_manager import TimelineManager
 
 _logger = logging.getLogger(__name__)
 
@@ -108,6 +109,8 @@ def __init__(
         # indexed by remote instance id
         self._clients: dict[Reference, MPPClient] = {}
 
+        self._timeline_manager = TimelineManager(port_manager)
+
     def get_locations(self) -> list[str]:
         """Returns a list of locations that we can be reached at.
 
@@ -124,12 +127,14 @@ def set_peer_info(self, peer_info: PeerInfo) -> None:
         """Inform this Communicator about its peers.
 
         This tells the Communicator about its peers, so that it can route
-        messages accordingly.
+        messages accordingly. This also completes the TimelineManager's
+        initialization.
 
         Args:
             peer_info: Information about the peers.
         """
         self._peer_info = peer_info
+        self._timeline_manager.connect_sub_timelines()
 
     def set_receive_timeout(self, receive_timeout: float) -> None:
         """Update the timeout after which the manager is notified that we are waiting
@@ -172,6 +177,11 @@ def send_message(
             # log sending on disconnected port
             return
 
+        if isinstance(message.data, ClosePort):
+            iteration = None
+        else:
+            iteration = self._timeline_manager.check_send_message(port_name, slot)
+
         port = self._port_manager.get_port(port_name)
         profile_event = ProfileEvent(
             ProfileEventType.SEND,
@@ -204,6 +214,7 @@ def send_message(
                 port.get_num_messages(slot),
                 checkpoints_considered_until,
                 message.data,
+                iteration,
             )
             encoded_message = mpp_message.encoded()
             self._server.deposit(recv_endpoint.ref(), encoded_message)
@@ -245,10 +256,7 @@ def receive_message(
             RuntimeError: If the network connection had an error, or the
                     message number was incorrect.
         """
-        if port_name == "muscle_settings_in":
-            port = self._port_manager._muscle_settings_in
-        else:
-            port = self._port_manager.get_port(port_name)
+        port = self._port_manager.get_port(port_name)
 
         if slot is None:
             port_and_slot = port_name
@@ -321,6 +329,8 @@ def receive_message(
 
         if isinstance(mpp_message.data, ClosePort):
             port.set_closed(slot)
+        else:
+            self._timeline_manager.check_receive(port_name, slot)
 
         message = Message(
             mpp_message.timestamp,
@@ -395,6 +405,11 @@ def receive_message(
         if isinstance(mpp_message.data, ClosePort):
             _logger.debug(f"Port {port_and_slot} is now closed")
 
+        if not isinstance(mpp_message.data, ClosePort):
+            self._timeline_manager.check_received_message(
+                port_name, mpp_message.iteration, slot
+            )
+
         return message, mpp_message.saved_until
 
     def shutdown(self) -> None:

libmuscle/python/libmuscle/instance.py

@@ -13,7 +13,6 @@
 from libmuscle.logging import LogLevel
 from libmuscle.logging_handler import MuscleManagerHandler
 from libmuscle.mmp_client import MMPClient
-from libmuscle.mmsf_validator import MMSFValidator
 from libmuscle.mpp_message import ClosePort
 from libmuscle.port_manager import PortManager
 from libmuscle.profiler import Profiler
@@ -84,14 +83,6 @@ class InstanceFlags(Flag):
     :external:py:attr:`ymmsl.KeepsStateForNextUse.NECESSARY`).
     """
 
-    SKIP_MMSF_SEQUENCE_CHECKS = auto()
-    """Disable the checks whether the MMSF is strictly followed when sending/receiving
-    messages.
-
-    See :class:`~libmuscle.mmsf_validator.MMSFValidator` for a detailed description of
-    the checks.
-    """
-
 
 _CHECKPOINT_SUPPORT_MASK = (
     InstanceFlags.USES_CHECKPOINT_API
@@ -205,12 +196,6 @@ def __init__(
         self._set_remote_log_level()
         self._setup_profiling()
         self._setup_receive_timeout()
-        # MMSFValidator needs a connected port manager, and does some logging
-        self._mmsf_validator = (
-            None
-            if InstanceFlags.SKIP_MMSF_SEQUENCE_CHECKS in self._flags
-            else MMSFValidator(self._port_manager)
-        )
 
     def reuse_instance(self) -> bool:
         """Decide whether to run this instance again.
@@ -247,8 +232,6 @@ def reuse_instance(self) -> bool:
                 :meth:`save_final_snapshot`, or the checkpointing tutorial.
         """
         self._api_guard.verify_reuse_instance()
-        if self._mmsf_validator:
-            self._mmsf_validator.reuse_instance()
 
         if self._do_reuse is not None:
             # thank you, should_save_final_snapshot, for running this already
@@ -257,8 +240,11 @@ def reuse_instance(self) -> bool:
         else:
             do_reuse = self._decide_reuse_instance()
 
-        if self._do_resume and not self._do_init and self._mmsf_validator:
-            self._mmsf_validator.skip_f_init()
+        restored_from_intermediate = self._do_resume and not self._do_init
+        if restored_from_intermediate:
+            self._communicator._timeline_manager.restore_state(
+                self._snapshot_manager.resume_state()
+            )
 
         # now _first_run, _do_resume and _do_init are also set correctly
 
@@ -278,6 +264,12 @@ def reuse_instance(self) -> bool:
                 # store a None instead of a Message
                 self._save_snapshot(None, True, self.__f_init_max_timestamp)
 
+        if (
+            not restored_from_intermediate
+            and self._communicator._timeline_manager.cycle_complete()
+        ):
+            self._communicator._timeline_manager.reset()
+
         if not do_reuse:
             self.__shutdown()
 
@@ -500,8 +492,6 @@ def send(
             slot: The slot to send the message on, if any.
         """
         self.__check_port(port_name, slot, True)
-        if self._mmsf_validator:
-            self._mmsf_validator.check_send(port_name, slot)
         if message.settings is None:
             message = copy(message)
             message.settings = self._settings_manager.overlay
@@ -961,13 +951,15 @@ def _save_snapshot(
         """
         triggers = self._trigger_manager.get_triggers()
         walltime = self._trigger_manager.elapsed_walltime()
+        timeline_state = self._communicator._timeline_manager.get_state()
         timestamp = self._snapshot_manager.save_snapshot(
             message,
             final,
             triggers,
             walltime,
             f_init_max_timestamp,
             self._settings_manager.overlay,
+            timeline_state,
         )
         self._trigger_manager.update_checkpoints(timestamp)
 
@@ -984,8 +976,6 @@ def __receive_message(
         description of those.
         """
         self.__check_port(port_name, slot, False, True)
-        if self._mmsf_validator:
-            self._mmsf_validator.check_receive(port_name, slot)
 
         port = self._port_manager.get_port(port_name)
         if port.operator == Operator.F_INIT:

libmuscle/python/libmuscle/manager/mmp_server.py

@@ -55,7 +55,10 @@ def encode_conduit(conduit: Conduit) -> list[str]:
 
 def encode_ports(ports: Ports) -> list[list[str]]:
     """Convert a Ports to a MsgPack-compatible value."""
-    return [[str(port_name), ports[port_name].operator.name] for port_name in ports]
+    return [
+        [str(port_name), ports[port_name].operator.name, str(ports[port_name].timeline)]
+        for port_name in ports
+    ]
 
 
 def encode_checkpoint_rule(rule: CheckpointRule) -> dict[str, Any]:

libmuscle/python/libmuscle/manager/test/test_mmp_request_handler.py

@@ -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", ""], ["in", "S", ""]]
 
 
 def test_request_peers_fanin(registered_mmp_request_handler):
@@ -288,7 +288,7 @@ def test_request_peers_fanin(registered_mmp_request_handler):
 
     assert locations["macro"] == ["direct:macro"]
 
-    assert ports == [["in", "F_INIT"], ["out", "O_F"]]
+    assert ports == [["in", "F_INIT", ""], ["out", "O_F", ""]]
 
 
 def test_request_peers_bidir(registered_mmp_request_handler2):
@@ -319,7 +319,12 @@ def test_request_peers_bidir(registered_mmp_request_handler2):
     for i in range(10):
         assert locations[f"micro[2][{i}]"] == [f"direct:micro[2][{i}]"]
 
-    assert ports == [["init", "F_INIT"], ["out", "O_I"], ["in", "S"], ["final", "O_F"]]
+    assert ports == [
+        ["init", "F_INIT", ""],
+        ["out", "O_I", ""],
+        ["in", "S", ""],
+        ["final", "O_F", ""],
+    ]
 
 
 def test_request_peers_own_conduits(registered_mmp_request_handler2):
@@ -337,7 +342,7 @@ def test_request_peers_own_conduits(registered_mmp_request_handler2):
     assert conduits[1][0] == "meso.final"
     assert conduits[1][1] == "macro.in"
 
-    assert ports == [["out", "O_I"], ["in", "S"]]
+    assert ports == [["out", "O_I", ""], ["in", "S", ""]]
 
 
 def test_request_peers_unknown(registered_mmp_request_handler2):

libmuscle/python/libmuscle/mmp_client.py

@@ -18,6 +18,7 @@
     Port,
     Reference,
     Settings,
+    Timeline,
 )
 
 import libmuscle
@@ -311,7 +312,10 @@ def request_peers(self) -> PeerInfo:
             Reference(instance): locs for instance, locs in response[3].items()
         }
 
-        ports = [Port(Identifier(name), Operator[op]) for name, op in response[4]]
+        ports = [
+            Port(Identifier(name), Operator[op], Timeline(tl))
+            for name, op, tl in response[4]
+        ]
         name = instance_to_kernel(self._instance_id)
         index = instance_indices(self._instance_id)
         return PeerInfo(name, index, conduits, peer_dimensions, peer_locations, ports)