Fix pause

spinn_front_end_common/interface/interface_functions/application_runner.py

@@ -18,17 +18,15 @@
 from threading import Condition
 from typing import Optional
 from spinn_utilities.log import FormatAdapter
-from spinn_utilities.progress_bar import ProgressBar
-from spinnman.messages.sdp import SDPMessage, SDPHeader, SDPFlag
 from spinnman.messages.scp.enums import Signal
-from spinnman.model.enums import (
-    ExecutableType, SDP_PORTS, SDP_RUNNING_MESSAGE_CODES, CPUState)
+from spinnman.model.enums import ExecutableType
 from spinn_front_end_common.data import FecDataView
 from spinn_front_end_common.utilities.exceptions import (
-    ConfigurationException, ExecutableFailedToStopException)
+    ConfigurationException)
 from spinn_front_end_common.utilities.constants import (
     MICRO_TO_MILLISECOND_CONVERSION)
-from spinn_front_end_common.utilities.scp import GetCurrentTimeProcess
+from spinn_front_end_common.utilities.scp import (
+    GetCurrentTimeProcess, SendPauseProcess)
 
 SAFETY_FINISH_TIME = 0.1
 
@@ -120,12 +118,16 @@ def run_app(
             with state_condition:
                 while FecDataView.is_no_stop_requested():
                     state_condition.wait()
-            self.__send_pause()
-            self._wait_for_end()
 
             core_subsets = FecDataView.get_cores_for_type(
                 ExecutableType.USES_SIMULATION_INTERFACE)
             n_cores = len(core_subsets)
+
+            SendPauseProcess(
+                FecDataView.get_scamp_connection_selector()).send_pause(
+                    core_subsets, n_cores)
+            self._wait_for_end()
+
             process = GetCurrentTimeProcess(
                 FecDataView.get_scamp_connection_selector())
             latest_runtime = process.get_latest_runtime(n_cores, core_subsets)
@@ -228,53 +230,3 @@ def _determine_simulation_sync_signals(self) -> Optional[Signal]:
             sync_signal = Signal.SYNC0
 
         return sync_signal
-
-    def __send_pause(self) -> None:
-        all_core_subsets = FecDataView.get_executable_types()[
-            ExecutableType.USES_SIMULATION_INTERFACE]
-        total_processors = len(all_core_subsets)
-        last_finished_count = 0
-        with ProgressBar(total_processors, "Pausing application") as progress:
-            # check that the right number of processors are finished
-            while (processors_finished := self.__txrx.get_core_state_count(
-                    self.__app_id, CPUState.READY)) != total_processors:
-                if processors_finished > last_finished_count:
-                    progress.update(processors_finished - last_finished_count)
-                    last_finished_count = processors_finished
-
-                processors_rte = self.__txrx.get_core_state_count(
-                    self.__app_id, CPUState.RUN_TIME_EXCEPTION)
-                processors_watchdogged = self.__txrx.get_core_state_count(
-                    self.__app_id, CPUState.WATCHDOG)
-
-                if processors_rte > 0 or processors_watchdogged > 0:
-                    cpu_infos = self.__txrx.get_cpu_infos(
-                        all_core_subsets,
-                        [CPUState.RUN_TIME_EXCEPTION, CPUState.WATCHDOG], True)
-                    logger.error(cpu_infos.get_status_string())
-
-                    raise ExecutableFailedToStopException(
-                        f"{processors_rte + processors_watchdogged} of "
-                        f"{total_processors} processors went into an error "
-                        "state when shutting down")
-
-                successful_cores_finished = self.__txrx.get_cpu_infos(
-                    all_core_subsets, CPUState.READY, include=True)
-
-                for core_subset in all_core_subsets:
-                    for processor in core_subset.processor_ids:
-                        if not successful_cores_finished.is_core(
-                                core_subset.x, core_subset.y, processor):
-                            self.__send_pause_message(
-                                core_subset.x, core_subset.y, processor)
-                sleep(0.5)
-
-    def __send_pause_message(self, x, y, p):
-        sdp_port = SDP_PORTS.RUNNING_COMMAND_SDP_PORT.value
-        code = _ONE_WORD.pack(
-            SDP_RUNNING_MESSAGE_CODES.SDP_PAUSE_ID_CODE.value)
-        self.__txrx.send_sdp_message(SDPMessage(
-            sdp_header=SDPHeader(
-                flags=SDPFlag.REPLY_NOT_EXPECTED, destination_port=sdp_port,
-                destination_chip_x=x, destination_chip_y=y, destination_cpu=p),
-            data=code))

spinn_front_end_common/utilities/scp/__init__.py

@@ -17,10 +17,12 @@
 from .reinjector_control_process import ReinjectorControlProcess
 from .update_runtime_process import UpdateRuntimeProcess
 from .get_current_time_process import GetCurrentTimeProcess
+from .send_pause_process import SendPauseProcess
 
 __all__ = (
     "ClearIOBUFProcess",
     "LoadMCRoutesProcess",
     "ReinjectorControlProcess",
     "UpdateRuntimeProcess",
-    "GetCurrentTimeProcess")
+    "GetCurrentTimeProcess",
+    "SendPauseProcess")

spinn_front_end_common/utilities/scp/send_pause_process.py

@@ -0,0 +1,77 @@
+# Copyright (c) 2016 The University of Manchester
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from functools import partial
+from spinn_utilities.overrides import overrides
+from spinn_utilities.progress_bar import ProgressBar
+from spinn_machine import CoreSubsets
+from spinnman.messages.sdp import SDPHeader, SDPFlag
+from spinnman.messages.scp.abstract_messages import AbstractSCPRequest
+from spinnman.messages.scp import SCPRequestHeader
+from spinnman.messages.scp.impl import CheckOKResponse
+from spinnman.processes import AbstractMultiConnectionProcess
+from spinnman.model.enums import SDP_PORTS, SDP_RUNNING_MESSAGE_CODES
+
+
+class _SendPauseRequest(AbstractSCPRequest[CheckOKResponse]):
+    def __init__(self, x: int, y: int, p: int):
+        """
+        :param int x:
+        :param int y:
+        :param int p:
+        """
+        # pylint: disable=too-many-arguments
+        sdp_flags = SDPFlag.REPLY_EXPECTED
+
+        super().__init__(
+            SDPHeader(
+                flags=sdp_flags,
+                destination_port=SDP_PORTS.RUNNING_COMMAND_SDP_PORT.value,
+                destination_cpu=p, destination_chip_x=x, destination_chip_y=y),
+            SCPRequestHeader(
+                command=SDP_RUNNING_MESSAGE_CODES.SDP_PAUSE_ID_CODE))
+
+    @overrides(AbstractSCPRequest.get_scp_response)
+    def get_scp_response(self) -> CheckOKResponse:
+        return CheckOKResponse(
+            "update runtime",
+            SDP_RUNNING_MESSAGE_CODES.SDP_PAUSE_ID_CODE.value)
+
+
+class SendPauseProcess(AbstractMultiConnectionProcess[CheckOKResponse]):
+    """
+    Send pause to a set of cores
+
+    .. note::
+        The cores must be using the simulation interface.
+    """
+    __slots__ = ()
+
+    def __receive_response(
+            self, progress: ProgressBar, _response: CheckOKResponse):
+        progress.update()
+
+    def send_pause(self, core_subsets: CoreSubsets, n_cores: int):
+        """
+        :param ~spinn_machine.CoreSubsets core_subsets:
+        :param int n_cores: Number of cores being updated
+        """
+        with ProgressBar(n_cores, "Sending pause request") as progress, \
+                self._collect_responses():
+            for core_subset in core_subsets:
+                for processor_id in core_subset.processor_ids:
+                    self._send_request(
+                        _SendPauseRequest(
+                            core_subset.x, core_subset.y, processor_id),
+                        callback=partial(self.__receive_response, progress))