8906184429

Committed 01 May 2024 06:19AM UTC coverage: 47.241% (-0.06%) from 47.305%

Build # 8906184429

Build Type

Pull #1181

github

Committed by

Christian-B

Commit Message

add comment

Pull Request Pull Request #1181: Compressor error

Run Details

1761 of 4471 branches covered (39.39%)

Branch coverage included in aggregate %.

9 of 32 new or added lines in 4 files covered. (28.13%)

1 existing line in 1 file now uncovered.

5507 of 10914 relevant lines covered (50.46%)

0.5 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.0

/spinn_front_end_common/interface/provenance/fec_timer.py

# Copyright (c) 2017 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 __future__ import annotations
from collections.abc import Sized
import logging
import os
import time
from datetime import timedelta
from typing import List, Optional, Tuple, Union, TYPE_CHECKING
from typing_extensions import Literal, Self
from spinn_utilities.config_holder import (get_config_bool)
from spinn_utilities.log import FormatAdapter
from spinn_front_end_common.data import FecDataView
from .global_provenance import GlobalProvenance
from .timer_category import TimerCategory
if TYPE_CHECKING:
    from spinn_front_end_common.interface.abstract_spinnaker_base import (
        AbstractSpinnakerBase)
    from spinn_front_end_common.interface.provenance import TimerWork

logger = FormatAdapter(logging.getLogger(__name__))

# conversion factor
_NANO_TO_MICRO = 1000.0


class FecTimer(object):
    """
    Timer.
    """

    _simulator: Optional[AbstractSpinnakerBase] = None
    _provenance_path: Optional[str] = None
    _print_timings: bool = False
    _category_id: Optional[int] = None
    _category: Optional[TimerCategory] = None
    _category_time: int = 0
    _machine_on: bool = False
    _previous: List[TimerCategory] = []
    __slots__ = (
        # The start time when the timer was set off
        "_start_time",
        # Name of algorithm what is being timed
        "_algorithm",
        # Type of work being done
        "_work")

    # Algorithm Names used elsewhere
    APPLICATION_RUNNER = "Application runner"

    @classmethod
    def setup(cls, simulator: AbstractSpinnakerBase):
        """
        Checks and saves cfg values so they don't have to be read each time

        :param AbstractSpinnakerBase simulator: Not actually used
        """
        # pylint: disable=global-statement, protected-access
        cls._simulator = simulator
        if get_config_bool("Reports", "write_algorithm_timings"):
            cls._provenance_path = os.path.join(
                FecDataView.get_run_dir_path(),
                "algorithm_timings.rpt")
        else:
            cls._provenance_path = None
        cls._print_timings = get_config_bool(
            "Reports", "display_algorithm_timings") or False

    def __init__(self, algorithm: str, work: TimerWork):
        self._start_time: Optional[int] = None
        self._algorithm = algorithm
        self._work = work

    def __enter__(self) -> Self:
        self._start_time = time.perf_counter_ns()
        return self

    def _report(self, message: str):
        if self._provenance_path is not None:
            with open(self._provenance_path, "a", encoding="utf-8") as p_file:
                p_file.write(f"{message}\n")
        if self._print_timings:
            logger.info(message)

    def _insert_timing(
            self, time_taken: timedelta, skip_reason: Optional[str]):
        if self._category_id is not None:
            with GlobalProvenance() as db:
                db.insert_timing(
                    self._category_id, self._algorithm, self._work,
                    time_taken, skip_reason)

    def skip(self, reason: str):
        """
        Records that the algorithms is being skipped and ends the timer.

        :param str reason: Why the algorithm is being skipped
        """
        message = f"{self._algorithm} skipped as {reason}"
        time_taken = self._stop_timer()
        self._insert_timing(time_taken, reason)
        self._report(message)

    def skip_if_has_not_run(self) -> bool:
        """
        Skips if the simulation has not run.

        If the simulation has run used this methods
        keep the timer running and returns False (did not skip).

        If there was no run this method records the reason,
        ends the timing and returns True (it skipped).

        Currently not used as a better check is skip_if_empty on the data
        needed for the algorithm.

        :rtype: bool
        """
        if FecDataView.is_ran_ever():
            return False
        else:
            self.skip("simulator.has_run")
            return True

    def skip_if_virtual_board(self) -> bool:
        """
        Skips if a virtual board is being used.

        If a real board is being used this methods
        keep the timer running and returns False (did not skip).

        If a virtual board is being used this method records the reason,
        ends the timing and returns True (it skipped).

        Typically called for algorithms that require a real board to run.

        :rtype: bool
        """
        if get_config_bool("Machine", "virtual_board"):
            self.skip("virtual_board")
            return True
        else:
            return False

    def skip_if_empty(self, value: Optional[
            Union[bool, int, str, Sized]], name: str) -> bool:
        """
        Skips if the value is one that evaluates to False.

        If the value is considered True (if value) this methods
        keep the timer running and returns False (did not skip).

        If the value is False this method records the reason,
        ends the timing and returns True (it skipped).

        :param value: Value to check if True
        :param str name: Name to record for that value if skipping
        :rtype: bool
        """
        if value:
            return False
        if value is None:
            self.skip(f"{name} is None")
        elif isinstance(value, int) or len(value) == 0:
            self.skip(f"{name} is empty")
        else:
            self.skip(f"{name} is False for an unknown reason")
        return True

    def skip_if_cfg_false(self, section: str, option: str) -> bool:
        """
        Skips if a Boolean cfg values is False.

        If this cfg value is True this methods keep the timer running and
        returns False (did not skip).

        If the cfg value is False this method records the reason,
        ends the timing and returns True (it skipped).

        Typically called if the algorithm should run if the cfg value
        is set True.

        :param str section: Section level to be applied to both options
        :param str option1: One of the options to check
        :param str option2: The other option to check
        :rtype: bool
        """
        if get_config_bool(section, option):
            return False
        else:
            self.skip(f"cfg {section}:{option} is False")
            return True

    def skip_if_cfgs_false(
            self, section: str, option1: str, option2: str) -> bool:
        """
        Skips if two Boolean cfg values are both False.

        If either cfg value is True this methods keep the timer running and
        returns False (did not skip).

        If both cfg values are False this method records the reason,
        ends the timing and returns True (it skipped).

        Typically called if the algorithm should run if either cfg values
        is set True.

        :param str section: Section level to be applied to both options
        :param str option1: One of the options to check
        :param str option2: The other option to check
        :rtype: bool
        """
        if get_config_bool(section, option1):
            return False
        elif get_config_bool(section, option2):
            return False
        else:
            self.skip(f"cfg {section}:{option1} and {option2} are False")
            return True

    def skip_all_cfgs_false(
            self, pairs: List[Tuple[str, str]], reason: str) -> bool:
        """
        Skips if two Boolean cfg values are both False.

        If either cfg value is True this methods keep the timer running and
        returns False (did not skip).

        If both cfg values are False this method records the reason,
        ends the timing and returns True (it skipped).

        Typically called if the algorithm should run if either cfg values
        is set True.

        :param str section: Section level to be applied to both options
        :param list((str, str) pairs: section, options pairs to check
        :param str reason: Reason to record for the skip
        :rtype: bool
        """
        for section, option in pairs:
            if get_config_bool(section, option):
                return False
        self.skip(reason)
        return True

    def error(self, reason: str):
        """
         Ends an algorithm timing and records that it failed.

        :param str reason: What caused the error
        """
        time_taken = self._stop_timer()
        message = f"{self._algorithm} failed after {time_taken} as {reason}"
        self._insert_timing(time_taken, reason)
        self._report(message)

    def _stop_timer(self) -> timedelta:
        """
        Describes how long has elapsed since the instance that the
        :py:meth:`start_timing` method was last called.

        :rtype: datetime.timedelta
        """
        time_now = time.perf_counter_ns()
        assert self._start_time is not None
        diff = time_now - self._start_time
        self._start_time = None
        return self.__convert_to_timedelta(diff)

    @staticmethod
    def __convert_to_timedelta(time_diff: int) -> timedelta:
        """
        Have to convert to a timedelta for rest of code to read.

        As perf_counter_ns is nanoseconds, and time delta lowest is micro,
        need to convert.
        """
        return timedelta(microseconds=time_diff / _NANO_TO_MICRO)

    def __exit__(self, exc_type, exc_value, traceback) -> Literal[False]:
        if self._start_time is None:
            return False
        time_taken = self._stop_timer()
        if exc_type is None:
            message = f"{self._algorithm} took {time_taken} "
            skip = None
        else:
            try:
                message = (f"{self._algorithm} exited with "
                           f"{exc_type.__name__} after {time_taken}")
                skip = exc_type.__name__
            except Exception as ex:  # pylint: disable=broad-except
                message = (f"{self._algorithm} exited with an exception"
                           f"after {time_taken}")
                skip = f"Exception {ex}"

        self._insert_timing(time_taken, skip)
        self._report(message)
        return False

    @classmethod
    def __stop_category(cls) -> int:
        """
        Stops the current category and logs how long it took

        :return: Time the stop happened
        """
        time_now = time.perf_counter_ns()
        if cls._category_id:
            with GlobalProvenance() as db:
                diff = cls.__convert_to_timedelta(
                    time_now - cls._category_time)
                db.insert_category_timing(cls._category_id, diff)
        return time_now

    @classmethod
    def _change_category(cls, category: TimerCategory):
        """
        This method should only be called via the View!

        :param TimerCategory category: Category to switch to
        """
        time_now = cls.__stop_category()
        with GlobalProvenance() as db:
            cls._category_id = db.insert_category(category, cls._machine_on)
        cls._category = category
        cls._category_time = time_now

    @classmethod
    def start_category(cls, category: TimerCategory, machine_on=None):
        """
        This method should only be called via the View!

        :param TimerCategory category: category to switch to
        :param machine_on: What to change machine on too.
            Or `None` to leave as is
        :type machine_on: None or bool
        """
        if cls._category is not None:
            cls._previous.append(cls._category)
        if cls._category != category:
            cls._change_category(category)
        if machine_on is not None:
            cls._machine_on = machine_on

    @classmethod
    def end_category(cls, category: TimerCategory):
        """
        This method should only be
        called via the View!

        :param TimerCategory category: Stage to end
        """
        if cls._category != category:
            raise ValueError(
                f"Current category is {cls._category} not {category}")
        previous = cls._previous.pop() if cls._previous else None
        if previous is None:
            raise NotImplementedError(
                "Use stop_category_timing to end the last category")
        if category != previous:
            cls._change_category(previous)

    @classmethod
    def stop_category_timing(cls) -> None:
        """
        Stops all the timing.

        Typically only called during simulator shutdown
        """
        cls.__stop_category()
        cls._previous = []
        cls._category = None
        cls._category_id = None

1	# Copyright (c) 2017 The University of Manchester
2	#
3	# Licensed under the Apache License, Version 2.0 (the "License");
4	# you may not use this file except in compliance with the License.
5	# You may obtain a copy of the License at
6	#
7	# https://www.apache.org/licenses/LICENSE-2.0
8	#
9	# Unless required by applicable law or agreed to in writing, software
10	# distributed under the License is distributed on an "AS IS" BASIS,
11	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	# See the License for the specific language governing permissions and
13	# limitations under the License.
14	from __future__ import annotations	1✔
15	from collections.abc import Sized	1✔
16	import logging	1✔
17	import os	1✔
18	import time	1✔
19	from datetime import timedelta	1✔
20	from typing import List, Optional, Tuple, Union, TYPE_CHECKING	1✔
21	from typing_extensions import Literal, Self	1✔
22	from spinn_utilities.config_holder import (get_config_bool)	1✔
23	from spinn_utilities.log import FormatAdapter	1✔
24	from spinn_front_end_common.data import FecDataView	1✔
25	from .global_provenance import GlobalProvenance	1✔
26	from .timer_category import TimerCategory	1✔
27	if TYPE_CHECKING:
28	from spinn_front_end_common.interface.abstract_spinnaker_base import (
29	AbstractSpinnakerBase)
30	from spinn_front_end_common.interface.provenance import TimerWork
31
32	logger = FormatAdapter(logging.getLogger(__name__))	1✔
33
34	# conversion factor
35	_NANO_TO_MICRO = 1000.0	1✔
36
37
38	class FecTimer(object):	1✔
39	"""
40	Timer.
41	"""
42
43	_simulator: Optional[AbstractSpinnakerBase] = None	1✔
44	_provenance_path: Optional[str] = None	1✔
45	_print_timings: bool = False	1✔
46	_category_id: Optional[int] = None	1✔
47	_category: Optional[TimerCategory] = None	1✔
48	_category_time: int = 0	1✔
49	_machine_on: bool = False	1✔
50	_previous: List[TimerCategory] = []	1✔
51	__slots__ = (	1✔
52	# The start time when the timer was set off
53	"_start_time",
54	# Name of algorithm what is being timed
55	"_algorithm",
56	# Type of work being done
57	"_work")
58
59	# Algorithm Names used elsewhere
60	APPLICATION_RUNNER = "Application runner"	1✔
61
62	@classmethod	1✔
63	def setup(cls, simulator: AbstractSpinnakerBase):	1✔
64	"""
65	Checks and saves cfg values so they don't have to be read each time
66
67	:param AbstractSpinnakerBase simulator: Not actually used
68	"""
69	# pylint: disable=global-statement, protected-access
70	cls._simulator = simulator	1✔
71	if get_config_bool("Reports", "write_algorithm_timings"):	1!
72	cls._provenance_path = os.path.join(	1✔
73	FecDataView.get_run_dir_path(),
74	"algorithm_timings.rpt")
75	else:
76	cls._provenance_path = None	×
77	cls._print_timings = get_config_bool(	1✔
78	"Reports", "display_algorithm_timings") or False
79
80	def __init__(self, algorithm: str, work: TimerWork):	1✔
81	self._start_time: Optional[int] = None	1✔
82	self._algorithm = algorithm	1✔
83	self._work = work	1✔
84
85	def __enter__(self) -> Self:	1✔
86	self._start_time = time.perf_counter_ns()	1✔
87	return self	1✔
88
89	def _report(self, message: str):	1✔
90	if self._provenance_path is not None:	1!
91	with open(self._provenance_path, "a", encoding="utf-8") as p_file:	1✔
92	p_file.write(f"{message}\n")	1✔
93	if self._print_timings:	1!
94	logger.info(message)	1✔
95
96	def _insert_timing(	1✔
97	self, time_taken: timedelta, skip_reason: Optional[str]):
98	if self._category_id is not None:	1!
99	with GlobalProvenance() as db:	1✔
100	db.insert_timing(	1✔
101	self._category_id, self._algorithm, self._work,
102	time_taken, skip_reason)
103
104	def skip(self, reason: str):	1✔
105	"""
106	Records that the algorithms is being skipped and ends the timer.
107
108	:param str reason: Why the algorithm is being skipped
109	"""
110	message = f"{self._algorithm} skipped as {reason}"	1✔
111	time_taken = self._stop_timer()	1✔
112	self._insert_timing(time_taken, reason)	1✔
113	self._report(message)	1✔
114
115	def skip_if_has_not_run(self) -> bool:	1✔
116	"""
117	Skips if the simulation has not run.
118
119	If the simulation has run used this methods
120	keep the timer running and returns False (did not skip).
121
122	If there was no run this method records the reason,
123	ends the timing and returns True (it skipped).
124
125	Currently not used as a better check is skip_if_empty on the data
126	needed for the algorithm.
127
128	:rtype: bool
129	"""
130	if FecDataView.is_ran_ever():	×
131	return False	×
132	else:
133	self.skip("simulator.has_run")	×
134	return True	×
135
136	def skip_if_virtual_board(self) -> bool:	1✔
137	"""
138	Skips if a virtual board is being used.
139
140	If a real board is being used this methods
141	keep the timer running and returns False (did not skip).
142
143	If a virtual board is being used this method records the reason,
144	ends the timing and returns True (it skipped).
145
146	Typically called for algorithms that require a real board to run.
147
148	:rtype: bool
149	"""
150	if get_config_bool("Machine", "virtual_board"):	×
151	self.skip("virtual_board")	×
152	return True	×
153	else:
154	return False	×
155
156	def skip_if_empty(self, value: Optional[	1✔
157	Union[bool, int, str, Sized]], name: str) -> bool:
158	"""
159	Skips if the value is one that evaluates to False.
160
161	If the value is considered True (if value) this methods
162	keep the timer running and returns False (did not skip).
163
164	If the value is False this method records the reason,
165	ends the timing and returns True (it skipped).
166
167	:param value: Value to check if True
168	:param str name: Name to record for that value if skipping
169	:rtype: bool
170	"""
171	if value:	×
172	return False	×
173	if value is None:	×
174	self.skip(f"{name} is None")	×
175	elif isinstance(value, int) or len(value) == 0:	×
176	self.skip(f"{name} is empty")	×
177	else:
178	self.skip(f"{name} is False for an unknown reason")	×
179	return True	×
180
181	def skip_if_cfg_false(self, section: str, option: str) -> bool:	1✔
182	"""
183	Skips if a Boolean cfg values is False.
184
185	If this cfg value is True this methods keep the timer running and
186	returns False (did not skip).
187
188	If the cfg value is False this method records the reason,
189	ends the timing and returns True (it skipped).
190
191	Typically called if the algorithm should run if the cfg value
192	is set True.
193
194	:param str section: Section level to be applied to both options
195	:param str option1: One of the options to check
196	:param str option2: The other option to check
197	:rtype: bool
198	"""
199	if get_config_bool(section, option):	×
200	return False	×
201	else:
202	self.skip(f"cfg {section}:{option} is False")	×
203	return True	×
204
205	def skip_if_cfgs_false(	1✔
206	self, section: str, option1: str, option2: str) -> bool:
207	"""
208	Skips if two Boolean cfg values are both False.
209
210	If either cfg value is True this methods keep the timer running and
211	returns False (did not skip).
212
213	If both cfg values are False this method records the reason,
214	ends the timing and returns True (it skipped).
215
216	Typically called if the algorithm should run if either cfg values
217	is set True.
218
219	:param str section: Section level to be applied to both options
220	:param str option1: One of the options to check
221	:param str option2: The other option to check
222	:rtype: bool
223	"""
224	if get_config_bool(section, option1):	×
225	return False	×
226	elif get_config_bool(section, option2):	×
227	return False	×
228	else:
229	self.skip(f"cfg {section}:{option1} and {option2} are False")	×
230	return True	×
231
232	def skip_all_cfgs_false(	1✔
233	self, pairs: List[Tuple[str, str]], reason: str) -> bool:
234	"""
235	Skips if two Boolean cfg values are both False.
236
237	If either cfg value is True this methods keep the timer running and
238	returns False (did not skip).
239
240	If both cfg values are False this method records the reason,
241	ends the timing and returns True (it skipped).
242
243	Typically called if the algorithm should run if either cfg values
244	is set True.
245
246	:param str section: Section level to be applied to both options
247	:param list((str, str) pairs: section, options pairs to check
248	:param str reason: Reason to record for the skip
249	:rtype: bool
250	"""
NEW 251	for section, option in pairs:	×
NEW 252	if get_config_bool(section, option):	×
NEW 253	return False	×
NEW 254	self.skip(reason)	×
NEW 255	return True	×
256
257	def error(self, reason: str):	1✔
258	"""
259	Ends an algorithm timing and records that it failed.
260
261	:param str reason: What caused the error
262	"""
263	time_taken = self._stop_timer()	×
264	message = f"{self._algorithm} failed after {time_taken} as {reason}"	×
265	self._insert_timing(time_taken, reason)	×
266	self._report(message)	×
267
268	def _stop_timer(self) -> timedelta:	1✔
269	"""
270	Describes how long has elapsed since the instance that the
271	:py:meth:`start_timing` method was last called.
272
273	:rtype: datetime.timedelta
274	"""
275	time_now = time.perf_counter_ns()	1✔
276	assert self._start_time is not None	1✔
277	diff = time_now - self._start_time	1✔
278	self._start_time = None	1✔
279	return self.__convert_to_timedelta(diff)	1✔
280
281	@staticmethod	1✔
282	def __convert_to_timedelta(time_diff: int) -> timedelta:	1✔
283	"""
284	Have to convert to a timedelta for rest of code to read.
285
286	As perf_counter_ns is nanoseconds, and time delta lowest is micro,
287	need to convert.
288	"""
289	return timedelta(microseconds=time_diff / _NANO_TO_MICRO)	1✔
290
291	def __exit__(self, exc_type, exc_value, traceback) -> Literal[False]:	1✔
292	if self._start_time is None:	1✔
293	return False	1✔
294	time_taken = self._stop_timer()	1✔
295	if exc_type is None:	1✔
296	message = f"{self._algorithm} took {time_taken} "	1✔
297	skip = None	1✔
298	else:
299	try:	1✔
300	message = (f"{self._algorithm} exited with "	1✔
301	f"{exc_type.__name__} after {time_taken}")
302	skip = exc_type.__name__	1✔
303	except Exception as ex: # pylint: disable=broad-except	×
304	message = (f"{self._algorithm} exited with an exception"	×
305	f"after {time_taken}")
306	skip = f"Exception {ex}"	×
307
308	self._insert_timing(time_taken, skip)	1✔
309	self._report(message)	1✔
310	return False	1✔
311
312	@classmethod	1✔
313	def __stop_category(cls) -> int:	1✔
314	"""
315	Stops the current category and logs how long it took
316
317	:return: Time the stop happened
318	"""
319	time_now = time.perf_counter_ns()	1✔
320	if cls._category_id:	1✔
321	with GlobalProvenance() as db:	1✔
322	diff = cls.__convert_to_timedelta(	1✔
323	time_now - cls._category_time)
324	db.insert_category_timing(cls._category_id, diff)	1✔
325	return time_now	1✔
326
327	@classmethod	1✔
328	def _change_category(cls, category: TimerCategory):	1✔
329	"""
330	This method should only be called via the View!
331
332	:param TimerCategory category: Category to switch to
333	"""
334	time_now = cls.__stop_category()	1✔
335	with GlobalProvenance() as db:	1✔
336	cls._category_id = db.insert_category(category, cls._machine_on)	1✔
337	cls._category = category	1✔
338	cls._category_time = time_now	1✔
339
340	@classmethod	1✔
341	def start_category(cls, category: TimerCategory, machine_on=None):	1✔
342	"""
343	This method should only be called via the View!
344
345	:param TimerCategory category: category to switch to
346	:param machine_on: What to change machine on too.
347	Or `None` to leave as is
348	:type machine_on: None or bool
349	"""
350	if cls._category is not None:	1✔
351	cls._previous.append(cls._category)	1✔
352	if cls._category != category:	1✔
353	cls._change_category(category)	1✔
354	if machine_on is not None:	1✔
355	cls._machine_on = machine_on	1✔
356
357	@classmethod	1✔
358	def end_category(cls, category: TimerCategory):	1✔
359	"""
360	This method should only be
361	called via the View!
362
363	:param TimerCategory category: Stage to end
364	"""
365	if cls._category != category:	1✔
366	raise ValueError(	1✔
367	f"Current category is {cls._category} not {category}")
368	previous = cls._previous.pop() if cls._previous else None	1✔
369	if previous is None:	1✔
370	raise NotImplementedError(
371	"Use stop_category_timing to end the last category")
372	if category != previous:	1✔
373	cls._change_category(previous)	1✔
374
375	@classmethod	1✔
376	def stop_category_timing(cls) -> None:	1✔
377	"""
378	Stops all the timing.
379
380	Typically only called during simulator shutdown
381	"""
382	cls.__stop_category()	1✔
383	cls._previous = []	1✔
384	cls._category = None	1✔
385	cls._category_id = None	1✔

SpiNNakerManchester / SpiNNFrontEndCommon / 8906184429

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous