Source code for spinn_front_end_common.utilities.system_control_logic
# Copyright (c) 2019 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 spinnman.exceptions import (
SpinnmanException, SpiNNManCoresNotInStateException)
from spinnman.messages.scp.enums import Signal
from spinnman.model import ExecutableTargets
from spinn_front_end_common.data import FecDataView
from spinn_front_end_common.utilities.utility_objs import ExecutableType
from spinn_front_end_common.utilities.iobuf_extractor import IOBufExtractor
[docs]def run_system_application(
executable_cores, app_id,
read_algorithm_iobuf, check_for_success_function,
cpu_end_states, needs_sync_barrier, filename_template,
binaries_to_track=None, progress_bar=None, logger=None, timeout=None):
"""
Executes the given _system_ application.
Used for on-chip expanders, compressors, etc.
:param ~spinnman.model.ExecutableTargets executable_cores:
the cores to run the executable on.
:param int app_id: the app-id for the executable
:param bool read_algorithm_iobuf: whether to report IOBUFs
:param callable check_for_success_function:
function used to check success;
expects `executable_cores`, `transceiver` as inputs
:param set(~spinnman.model.enums.CPUState) cpu_end_states:
the states that a successful run is expected to terminate in
:param bool needs_sync_barrier: whether a sync barrier is needed
:param str filename_template: the IOBUF filename template.
:param list(str) binaries_to_track:
A list of binary names to check for exit state.
Or `None` for all binaries
:param progress_bar: Possible progress bar to update.
end() will be called after state checked
:type progress_bar: ~spinn_utilities.progress_bar.ProgressBar or None
:param ~logging.Logger logger:
If provided and IOBUF is extracted, will be used to log errors and
warnings
:param timeout:
Number of seconds to wait before force stopping, or `None` to wait
forever
:type timeout: float or None
:raise SpinnmanException:
If one should arise from the underlying SpiNNMan calls
"""
transceiver = FecDataView.get_transceiver()
# load the executable
transceiver.execute_application(executable_cores, app_id)
if needs_sync_barrier:
# fire all signals as required
transceiver.send_signal(app_id, Signal.SYNC0)
error = None
binary_start_types = dict()
if binaries_to_track is None:
check_targets = executable_cores
else:
check_targets = ExecutableTargets()
for binary_name in binaries_to_track:
check_targets.add_subsets(
binary_name,
executable_cores.get_cores_for_binary(binary_name))
for binary_name in executable_cores.binaries:
binary_start_types[binary_name] = ExecutableType.SYSTEM
# Wait for the executable to finish
succeeded = False
core_state_string = None
try:
transceiver.wait_for_cores_to_be_in_state(
check_targets.all_core_subsets, app_id, cpu_end_states,
progress_bar=progress_bar, timeout=timeout)
if progress_bar is not None:
progress_bar.end()
succeeded = True
except SpiNNManCoresNotInStateException as ex:
error = ex
core_state_string = transceiver.get_core_status_string(
ex.failed_core_states())
except SpinnmanException as ex:
# Delay the exception until iobuf is ready
error = ex
if progress_bar is not None:
progress_bar.end()
# Check if any cores have not completed successfully
if succeeded and check_for_success_function:
succeeded = check_for_success_function(executable_cores)
# if doing iobuf or on failure (succeeded is None is not failure)
if read_algorithm_iobuf or not succeeded:
_report_iobuf_messages(executable_cores, logger, filename_template)
# stop anything that's associated with the compressor binary
transceiver.stop_application(app_id)
FecDataView.free_id(app_id)
if error is not None:
if core_state_string is not None:
print(core_state_string)
raise error # pylint: disable=raising-bad-type
def _report_iobuf_messages(cores, logger, filename_template):
"""
:param ~spinnman.model.ExecutableTargets cores:
:param ~logging.Logger logger:
:param str filename_template:
"""
# Import in this function to prevent circular import issue
iobuf_reader = IOBufExtractor(
cores,
filename_template=filename_template, suppress_progress=False)
error_entries, warn_entries = iobuf_reader.extract_iobuf()
if logger is not None:
for entry in warn_entries:
logger.warn(entry)
for entry in error_entries:
logger.error(entry)