6574804013

Committed 19 Oct 2023 12:47PM UTC coverage: 51.937% (+1.2%) from 50.777%

Build # 6574804013

Build Type

Pull #327

github

Committed by

Christian-B

Commit Message

typing changes

Pull Request Pull Request #327: Type Annotations and Checking

Run Details

105 of 1288 branches covered (0.0%)

Branch coverage included in aggregate %.

2375 of 2375 new or added lines in 180 files covered. (100.0%)

4775 of 8108 relevant lines covered (58.89%)

0.59 hits per line

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

20.55

/spinnman/processes/application_copy_run_process.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 collections import defaultdict
from typing import cast, Iterable, List, Mapping, Tuple
from spinn_machine import Chip, CoreSubsets, Link, Machine
from spinnman.data import SpiNNManDataView
from spinnman.messages.scp.impl import AppCopyRun
from spinnman.processes import ConnectionSelector
from .abstract_multi_connection_process import AbstractMultiConnectionProcess

APP_COPY_RUN_TIMEOUT = 6.0


def _on_same_board(chip_1: Chip, chip_2: Chip):
    return (chip_1.nearest_ethernet_x == chip_2.nearest_ethernet_x and
            chip_1.nearest_ethernet_y == chip_2.nearest_ethernet_y)


def _get_next_chips(
        chips_done: Mapping[Tuple[int, int], List[Tuple[int, int]]],
        parent_chips: Mapping[Tuple[int, int], List[Chip]],
        machine: Machine) -> Iterable[Chip]:
    """
    Get the chips that are adjacent to the last set of chips, which
    haven't yet been loaded.  Also returned are the links for each chip,
    which gives the link which should be read from to get the data.

    :param dict((int,int),list(int,int)) chips_done:
        The coordinates of chips that have already been done by Ethernet
    :param dict((int, int),~spinn_machine.Chip) parent_chips:
        A dictionary of chip coordinates to chips that use that chip as a
        parent
    :return: A list of next chips to use
    :rtype: list(chip)
    """
    next_chips: List[Chip] = list()
    for eth_chip in chips_done:
        off_board_copy_done = False
        for c_x, c_y in chips_done[eth_chip]:
            chip_xy = machine[c_x, c_y]
            for chip in parent_chips[c_x, c_y]:
                on_same_board = _on_same_board(chip, chip_xy)
                eth = (chip.nearest_ethernet_x, chip.nearest_ethernet_y)
                if (eth not in chips_done or
                        (chip.x, chip.y) not in chips_done[eth]):
                    if on_same_board or not off_board_copy_done:
                        next_chips.append(chip)
                        if not on_same_board:
                            off_board_copy_done = True
                        # Only do one copy from each chip at a time
                        break

    return next_chips


def _compute_parent_chips(
        machine: Machine) -> Mapping[Tuple[int, int], List[Chip]]:
    """
    Compute a dictionary of chip coordinates to list of chips who use that chip
    as a parent in the tree.

    :param ~spinn_machine.Machine machine: The machine to compute the map for
    :rtype: dict((int, int), ~spinn_machine.Chip)
    """
    chip_links: Mapping[Tuple[int, int], List[Chip]] = defaultdict(list)
    for chip in machine.chips:
        if chip.parent_link is not None:
            link = cast(Link, chip.router.get_link(chip.parent_link))
            chip_links[link.destination_x, link.destination_y].append(chip)
    return chip_links


class ApplicationCopyRunProcess(AbstractMultiConnectionProcess):
    """
    Process to start a binary on a subset of cores on a subset of chips
    of a machine, performed by, on each chip, copying the data from
    an adjacent chip and then starting the binary.  This goes to each
    chip in turn, and so detects failures early on, as well as ensuring
    that the copy and execution is done in the case of success i.e. this
    ensures that if all commands are successful, the full binary has been
    copied and started.

    .. note::
        The binary must have been loaded to the boot chip before this is
        called!
    """
    __slots__ = ()

    def __init__(self, next_connection_selector: ConnectionSelector,
                 timeout: float = APP_COPY_RUN_TIMEOUT):
        AbstractMultiConnectionProcess.__init__(
            self, next_connection_selector, timeout=timeout)

    def run(self, size: int, app_id: int, core_subsets: CoreSubsets,
            checksum: int, wait: bool):
        """
        Run the process.

        :param int size: The size of the binary to copy
        :param int app_id: The application id to assign to the running binary
        :param CoreSubsets core_subsets: The cores to load the binary on to
        :param int checksum: The checksum of the data to test against
        :param bool wait:
            Whether to put the binary in "wait" mode or run it straight away
        """
        machine = SpiNNManDataView.get_machine()
        boot_chip = machine.boot_chip
        chips_done: Mapping[Tuple[int, int], List[Tuple[int, int]]] = \
            defaultdict(list)
        chips_done[boot_chip.x, boot_chip.y].append((boot_chip.x, boot_chip.y))
        parent_chips = _compute_parent_chips(machine)
        next_chips = _get_next_chips(chips_done, parent_chips, machine)

        while next_chips:
            # Do all the chips at the current level
            for chip in next_chips:
                subset = core_subsets.get_core_subset_for_chip(chip.x, chip.y)
                self._send_request(AppCopyRun(
                    chip.x, chip.y, cast(int, chip.parent_link), size, app_id,
                    subset.processor_ids, checksum, wait))
                eth = (chip.nearest_ethernet_x, chip.nearest_ethernet_y)
                chips_done[eth].append((chip.x, chip.y))
            self._finish()
            self.check_for_error()
            next_chips = _get_next_chips(chips_done, parent_chips, machine)

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
15	from collections import defaultdict	1✔
16	from typing import cast, Iterable, List, Mapping, Tuple	1✔
17	from spinn_machine import Chip, CoreSubsets, Link, Machine	1✔
18	from spinnman.data import SpiNNManDataView	1✔
19	from spinnman.messages.scp.impl import AppCopyRun	1✔
20	from spinnman.processes import ConnectionSelector	1✔
21	from .abstract_multi_connection_process import AbstractMultiConnectionProcess	1✔
22
23	APP_COPY_RUN_TIMEOUT = 6.0	1✔
24
25
26	def _on_same_board(chip_1: Chip, chip_2: Chip):	1✔
27	return (chip_1.nearest_ethernet_x == chip_2.nearest_ethernet_x and	×
28	chip_1.nearest_ethernet_y == chip_2.nearest_ethernet_y)
29
30
31	def _get_next_chips(	1✔
32	chips_done: Mapping[Tuple[int, int], List[Tuple[int, int]]],
33	parent_chips: Mapping[Tuple[int, int], List[Chip]],
34	machine: Machine) -> Iterable[Chip]:
35	"""
36	Get the chips that are adjacent to the last set of chips, which
37	haven't yet been loaded. Also returned are the links for each chip,
38	which gives the link which should be read from to get the data.
39
40	:param dict((int,int),list(int,int)) chips_done:
41	The coordinates of chips that have already been done by Ethernet
42	:param dict((int, int),~spinn_machine.Chip) parent_chips:
43	A dictionary of chip coordinates to chips that use that chip as a
44	parent
45	:return: A list of next chips to use
46	:rtype: list(chip)
47	"""
48	next_chips: List[Chip] = list()	×
49	for eth_chip in chips_done:	×
50	off_board_copy_done = False	×
51	for c_x, c_y in chips_done[eth_chip]:	×
52	chip_xy = machine[c_x, c_y]	×
53	for chip in parent_chips[c_x, c_y]:	×
54	on_same_board = _on_same_board(chip, chip_xy)	×
55	eth = (chip.nearest_ethernet_x, chip.nearest_ethernet_y)	×
56	if (eth not in chips_done or	×
57	(chip.x, chip.y) not in chips_done[eth]):
58	if on_same_board or not off_board_copy_done:	×
59	next_chips.append(chip)	×
60	if not on_same_board:	×
61	off_board_copy_done = True	×
62	# Only do one copy from each chip at a time
63	break	×
64
65	return next_chips	×
66
67
68	def _compute_parent_chips(	1✔
69	machine: Machine) -> Mapping[Tuple[int, int], List[Chip]]:
70	"""
71	Compute a dictionary of chip coordinates to list of chips who use that chip
72	as a parent in the tree.
73
74	:param ~spinn_machine.Machine machine: The machine to compute the map for
75	:rtype: dict((int, int), ~spinn_machine.Chip)
76	"""
77	chip_links: Mapping[Tuple[int, int], List[Chip]] = defaultdict(list)	×
78	for chip in machine.chips:	×
79	if chip.parent_link is not None:	×
80	link = cast(Link, chip.router.get_link(chip.parent_link))	×
81	chip_links[link.destination_x, link.destination_y].append(chip)	×
82	return chip_links	×
83
84
85	class ApplicationCopyRunProcess(AbstractMultiConnectionProcess):	1✔
86	"""
87	Process to start a binary on a subset of cores on a subset of chips
88	of a machine, performed by, on each chip, copying the data from
89	an adjacent chip and then starting the binary. This goes to each
90	chip in turn, and so detects failures early on, as well as ensuring
91	that the copy and execution is done in the case of success i.e. this
92	ensures that if all commands are successful, the full binary has been
93	copied and started.
94
95	.. note::
96	The binary must have been loaded to the boot chip before this is
97	called!
98	"""
99	__slots__ = ()	1✔
100
101	def __init__(self, next_connection_selector: ConnectionSelector,	1✔
102	timeout: float = APP_COPY_RUN_TIMEOUT):
103	AbstractMultiConnectionProcess.__init__(	×
104	self, next_connection_selector, timeout=timeout)
105
106	def run(self, size: int, app_id: int, core_subsets: CoreSubsets,	1✔
107	checksum: int, wait: bool):
108	"""
109	Run the process.
110
111	:param int size: The size of the binary to copy
112	:param int app_id: The application id to assign to the running binary
113	:param CoreSubsets core_subsets: The cores to load the binary on to
114	:param int checksum: The checksum of the data to test against
115	:param bool wait:
116	Whether to put the binary in "wait" mode or run it straight away
117	"""
118	machine = SpiNNManDataView.get_machine()	×
119	boot_chip = machine.boot_chip	×
120	chips_done: Mapping[Tuple[int, int], List[Tuple[int, int]]] = \	×
121	defaultdict(list)
122	chips_done[boot_chip.x, boot_chip.y].append((boot_chip.x, boot_chip.y))	×
123	parent_chips = _compute_parent_chips(machine)	×
124	next_chips = _get_next_chips(chips_done, parent_chips, machine)	×
125
126	while next_chips:	×
127	# Do all the chips at the current level
128	for chip in next_chips:	×
129	subset = core_subsets.get_core_subset_for_chip(chip.x, chip.y)	×
130	self._send_request(AppCopyRun(	×
131	chip.x, chip.y, cast(int, chip.parent_link), size, app_id,
132	subset.processor_ids, checksum, wait))
133	eth = (chip.nearest_ethernet_x, chip.nearest_ethernet_y)	×
134	chips_done[eth].append((chip.x, chip.y))	×
135	self._finish()	×
136	self.check_for_error()	×
137	next_chips = _get_next_chips(chips_done, parent_chips, machine)	×

SpiNNakerManchester / SpiNNMan / 6574804013

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