7002704121

Committed 27 Nov 2023 08:57AM UTC coverage: 84.232% (+0.6%) from 83.65%

Build # 7002704121

Build Type

Pull #516

github

Committed by

rowleya

Commit Message

Mypy makes things convoluted!

Pull Request Pull Request #516: Interdimensional compatibility

Run Details

1574 of 2048 branches covered (0.0%)

Branch coverage included in aggregate %.

135 of 164 new or added lines in 12 files covered. (82.32%)

12 existing lines in 4 files now uncovered.

5141 of 5924 relevant lines covered (86.78%)

0.87 hits per line

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

87.67

/pacman/operations/routing_table_generators/merged_routing_table_generator.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 typing import (
    Dict, Iterable, List, Optional, Tuple, TypeVar, Generic)
from spinn_utilities.progress_bar import ProgressBar
from spinn_machine import MulticastRoutingEntry
from pacman.data import PacmanDataView
from pacman.exceptions import PacmanRoutingException
from pacman.model.routing_tables import (
    UnCompressedMulticastRoutingTable, MulticastRoutingTables)
from pacman.model.graphs.application import ApplicationVertex
from pacman.model.routing_info import (
    RoutingInfo, AppVertexRoutingInfo, MachineVertexRoutingInfo)
from pacman.model.graphs import AbstractVertex
from pacman.model.routing_table_by_partition import (
    MulticastRoutingTableByPartitionEntry)
from pacman.model.graphs.machine.machine_vertex import MachineVertex


def merged_routing_table_generator() -> MulticastRoutingTables:
    """
    Creates routing entries by merging adjacent entries from the same
    application vertex when possible.

    :rtype: MulticastRoutingTables
    """
    routing_table_by_partitions = (
        PacmanDataView.get_routing_table_by_partition())
    routing_infos = PacmanDataView.get_routing_infos()
    progress = ProgressBar(
        routing_table_by_partitions.n_routers, "Generating routing tables")
    routing_tables = MulticastRoutingTables()
    for x, y in progress.over(routing_table_by_partitions.get_routers()):
        parts = routing_table_by_partitions.get_entries_for_router(x, y)
        if parts is None:
            continue
        routing_tables.add_routing_table(__create_routing_table(
            x, y, parts, routing_infos))

    return routing_tables


def __create_routing_table(
        x: int, y: int,
        partitions_in_table: Dict[Tuple[AbstractVertex, str],
                                  MulticastRoutingTableByPartitionEntry],
        routing_info: RoutingInfo) -> UnCompressedMulticastRoutingTable:
    """
    :param int x:
    :param int y:
    :param partitions_in_table:
    :type partitions_in_table:
        dict(((ApplicationVertex or MachineVertex), str),
        MulticastRoutingTableByPartitionEntry)
    :param RoutingInfo routing_infos:
    :rtype: MulticastRoutingTable
    """
    table = UnCompressedMulticastRoutingTable(x, y)
    iterator = _IteratorWithNext(partitions_in_table.items())
    while iterator.has_next:
        (vertex, part_id), entry = iterator.pop()
        r_info = routing_info.get_routing_info_from_pre_vertex(vertex, part_id)
        if r_info is None:
            raise PacmanRoutingException(
                f"Missing Routing information for {vertex}, {part_id}")

        # If we have an application vertex, just put the entry in and move on
        if isinstance(vertex, ApplicationVertex):
            table.add_multicast_routing_entry(
                MulticastRoutingEntry(
                    r_info.key, r_info.mask, defaultable=entry.defaultable,
                    spinnaker_route=entry.spinnaker_route))
            continue
        # Otherwise it has to be a machine vertex...
        assert isinstance(vertex, MachineVertex)
        assert isinstance(r_info, MachineVertexRoutingInfo)

        # If there is no application vertex, just add the entry
        if vertex.app_vertex is None:
            table.add_multicast_routing_entry(
                MulticastRoutingEntry(
                    r_info.key, r_info.mask, defaultable=entry.defaultable,
                    spinnaker_route=entry.spinnaker_route))
            continue

        # This has to be AppVertexRoutingInfo!
        app_r_info = routing_info.get_routing_info_from_pre_vertex(
            vertex.app_vertex, part_id)
        assert isinstance(app_r_info, AppVertexRoutingInfo)

        # Get the entries to merge
        entries: List[Tuple[
            MulticastRoutingTableByPartitionEntry,
            MachineVertexRoutingInfo]] = [(entry, r_info)]
        while __match(iterator, vertex, part_id, r_info, entry, routing_info,
                      app_r_info):
            (vertex, part_id), entry = iterator.pop()
            r_info = routing_info.get_routing_info_from_pre_vertex(
                vertex, part_id)
            if r_info is not None:
                assert isinstance(r_info, MachineVertexRoutingInfo)
                entries.append((entry, r_info))

        # Now attempt to merge sources together as much as possible
        for mrentry in __merged_keys_and_masks(app_r_info, entries):
            table.add_multicast_routing_entry(mrentry)

    return table


def __match(
        iterator, vertex, part_id, r_info, entry, routing_info,
        app_r_info) -> bool:
    if not iterator.has_next:
        return False
    if r_info.mask != app_r_info.machine_mask:
        return False
    (next_vertex, next_part_id), next_entry = iterator.peek()
    if isinstance(next_vertex, ApplicationVertex):
        return False
    if part_id != next_part_id:
        return False
    if __mask_has_holes(r_info.mask):
        return False
    next_r_info = routing_info.get_routing_info_from_pre_vertex(
        next_vertex, next_part_id)
    if next_r_info is None:
        raise KeyError(
            f"No routing info found for {next_vertex}, {next_part_id}")
    if next_r_info.index != r_info.index + 1:
        return False
    app_src = vertex.app_vertex
    next_app_src = next_vertex.app_vertex
    return next_app_src == app_src and entry.has_same_route(next_entry)


def __mask_has_holes(mask: int) -> bool:
    """
    Detect if the mask has a "hole" somewhere other than at the bottom.

    :param int mask: The mask to check
    :rtype: bool
    """
    # The mask is inverted and then add 1.  If this number is a power of 2,
    # the mask doesn't have holes
    inv_mask = (~mask & 0xFFFFFFFF) + 1
    # a & ~a == 0 if a is a power of 2
    return (inv_mask & (inv_mask - 1)) != 0


def __merged_keys_and_masks(
        app_r_info: AppVertexRoutingInfo,
        entries: List[Tuple[
            MulticastRoutingTableByPartitionEntry,
            MachineVertexRoutingInfo]]) -> Iterable[MulticastRoutingEntry]:
    if not entries:
        return
    (entry, r_info) = entries[0]
    if len(entries) == 1:
        yield MulticastRoutingEntry(
            r_info.key, r_info.mask, defaultable=entry.defaultable,
            spinnaker_route=entry.spinnaker_route)
    else:
        yield from app_r_info.merge_machine_entries(entries)


#: :meta private:
E = TypeVar("E")


class _IteratorWithNext(Generic[E]):
    def __init__(self, iterable: Iterable[E]):
        self.__iterator = iter(iterable)
        try:
            self.__next: Optional[E] = next(self.__iterator)
            self.__has_next = True
        except StopIteration:
            self.__next = None
            self.__has_next = False

    def peek(self) -> Optional[E]:
        return self.__next

    @property
    def has_next(self) -> bool:
        return self.__has_next

    def pop(self) -> E:
        if not self.__has_next:
            raise StopIteration
        nxt = self.__next
        assert nxt is not None
        try:
            self.__next = next(self.__iterator)
            self.__has_next = True
        except StopIteration:
            self.__next = None
            self.__has_next = False
        return nxt

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 typing import (	1✔
15	Dict, Iterable, List, Optional, Tuple, TypeVar, Generic)
16	from spinn_utilities.progress_bar import ProgressBar	1✔
17	from spinn_machine import MulticastRoutingEntry	1✔
18	from pacman.data import PacmanDataView	1✔
19	from pacman.exceptions import PacmanRoutingException	1✔
20	from pacman.model.routing_tables import (	1✔
21	UnCompressedMulticastRoutingTable, MulticastRoutingTables)
22	from pacman.model.graphs.application import ApplicationVertex	1✔
23	from pacman.model.routing_info import (	1✔
24	RoutingInfo, AppVertexRoutingInfo, MachineVertexRoutingInfo)
25	from pacman.model.graphs import AbstractVertex	1✔
26	from pacman.model.routing_table_by_partition import (	1✔
27	MulticastRoutingTableByPartitionEntry)
28	from pacman.model.graphs.machine.machine_vertex import MachineVertex	1✔
29
30
31	def merged_routing_table_generator() -> MulticastRoutingTables:	1✔
32	"""
33	Creates routing entries by merging adjacent entries from the same
34	application vertex when possible.
35
36	:rtype: MulticastRoutingTables
37	"""
38	routing_table_by_partitions = (	1✔
39	PacmanDataView.get_routing_table_by_partition())
40	routing_infos = PacmanDataView.get_routing_infos()	1✔
41	progress = ProgressBar(	1✔
42	routing_table_by_partitions.n_routers, "Generating routing tables")
43	routing_tables = MulticastRoutingTables()	1✔
44	for x, y in progress.over(routing_table_by_partitions.get_routers()):	1✔
45	parts = routing_table_by_partitions.get_entries_for_router(x, y)	1✔
46	if parts is None:	1!
47	continue	×
48	routing_tables.add_routing_table(__create_routing_table(	1✔
49	x, y, parts, routing_infos))
50
51	return routing_tables	1✔
52
53
54	def __create_routing_table(	1✔
55	x: int, y: int,
56	partitions_in_table: Dict[Tuple[AbstractVertex, str],
57	MulticastRoutingTableByPartitionEntry],
58	routing_info: RoutingInfo) -> UnCompressedMulticastRoutingTable:
59	"""
60	:param int x:
61	:param int y:
62	:param partitions_in_table:
63	:type partitions_in_table:
64	dict(((ApplicationVertex or MachineVertex), str),
65	MulticastRoutingTableByPartitionEntry)
66	:param RoutingInfo routing_infos:
67	:rtype: MulticastRoutingTable
68	"""
69	table = UnCompressedMulticastRoutingTable(x, y)	1✔
70	iterator = _IteratorWithNext(partitions_in_table.items())	1✔
71	while iterator.has_next:	1✔
72	(vertex, part_id), entry = iterator.pop()	1✔
73	r_info = routing_info.get_routing_info_from_pre_vertex(vertex, part_id)	1✔
74	if r_info is None:	1✔
75	raise PacmanRoutingException(	1✔
76	f"Missing Routing information for {vertex}, {part_id}")
77
78	# If we have an application vertex, just put the entry in and move on
79	if isinstance(vertex, ApplicationVertex):	1✔
80	table.add_multicast_routing_entry(	1✔
81	MulticastRoutingEntry(
82	r_info.key, r_info.mask, defaultable=entry.defaultable,
83	spinnaker_route=entry.spinnaker_route))
84	continue	1✔
85	# Otherwise it has to be a machine vertex...
86	assert isinstance(vertex, MachineVertex)	1✔
87	assert isinstance(r_info, MachineVertexRoutingInfo)	1✔
88
89	# If there is no application vertex, just add the entry
90	if vertex.app_vertex is None:	1!
NEW 91	table.add_multicast_routing_entry(	×
92	MulticastRoutingEntry(
93	r_info.key, r_info.mask, defaultable=entry.defaultable,
94	spinnaker_route=entry.spinnaker_route))
NEW 95	continue	×
96
97	# This has to be AppVertexRoutingInfo!
98	app_r_info = routing_info.get_routing_info_from_pre_vertex(	1✔
99	vertex.app_vertex, part_id)
100	assert isinstance(app_r_info, AppVertexRoutingInfo)	1✔
101
102	# Get the entries to merge
103	entries: List[Tuple[	1✔
104	MulticastRoutingTableByPartitionEntry,
105	MachineVertexRoutingInfo]] = [(entry, r_info)]
106	while __match(iterator, vertex, part_id, r_info, entry, routing_info,	1✔
107	app_r_info):
108	(vertex, part_id), entry = iterator.pop()	1✔
109	r_info = routing_info.get_routing_info_from_pre_vertex(	1✔
110	vertex, part_id)
111	if r_info is not None:	1!
112	assert isinstance(r_info, MachineVertexRoutingInfo)	1✔
113	entries.append((entry, r_info))	1✔
114
115	# Now attempt to merge sources together as much as possible
116	for mrentry in __merged_keys_and_masks(app_r_info, entries):	1✔
117	table.add_multicast_routing_entry(mrentry)	1✔
118
119	return table	1✔
120
121
122	def __match(	1✔
123	iterator, vertex, part_id, r_info, entry, routing_info,
124	app_r_info) -> bool:
125	if not iterator.has_next:	1✔
126	return False	1✔
127	if r_info.mask != app_r_info.machine_mask:	1!
UNCOV 128	return False	×
129	(next_vertex, next_part_id), next_entry = iterator.peek()	1✔
130	if isinstance(next_vertex, ApplicationVertex):	1!
131	return False	×
132	if part_id != next_part_id:	1!
133	return False	×
134	if __mask_has_holes(r_info.mask):	1!
135	return False	×
136	next_r_info = routing_info.get_routing_info_from_pre_vertex(	1✔
137	next_vertex, next_part_id)
138	if next_r_info is None:	1!
139	raise KeyError(	×
140	f"No routing info found for {next_vertex}, {next_part_id}")
141	if next_r_info.index != r_info.index + 1:	1✔
142	return False	1✔
143	app_src = vertex.app_vertex	1✔
144	next_app_src = next_vertex.app_vertex	1✔
145	return next_app_src == app_src and entry.has_same_route(next_entry)	1✔
146
147
148	def __mask_has_holes(mask: int) -> bool:	1✔
149	"""
150	Detect if the mask has a "hole" somewhere other than at the bottom.
151
152	:param int mask: The mask to check
153	:rtype: bool
154	"""
155	# The mask is inverted and then add 1. If this number is a power of 2,
156	# the mask doesn't have holes
157	inv_mask = (~mask & 0xFFFFFFFF) + 1	1✔
158	# a & ~a == 0 if a is a power of 2
159	return (inv_mask & (inv_mask - 1)) != 0	1✔
160
161
162	def __merged_keys_and_masks(	1✔
163	app_r_info: AppVertexRoutingInfo,
164	entries: List[Tuple[
165	MulticastRoutingTableByPartitionEntry,
166	MachineVertexRoutingInfo]]) -> Iterable[MulticastRoutingEntry]:
167	if not entries:	1!
168	return	×
169	(entry, r_info) = entries[0]	1✔
170	if len(entries) == 1:	1✔
171	yield MulticastRoutingEntry(	1✔
172	r_info.key, r_info.mask, defaultable=entry.defaultable,
173	spinnaker_route=entry.spinnaker_route)
174	else:
175	yield from app_r_info.merge_machine_entries(entries)	1✔
176
177
178	#: :meta private:
179	E = TypeVar("E")	1✔
180
181
182	class _IteratorWithNext(Generic[E]):	1✔
183	def __init__(self, iterable: Iterable[E]):	1✔
184	self.__iterator = iter(iterable)	1✔
185	try:	1✔
186	self.__next: Optional[E] = next(self.__iterator)	1✔
187	self.__has_next = True	1✔
188	except StopIteration:	1✔
189	self.__next = None	1✔
190	self.__has_next = False	1✔
191
192	def peek(self) -> Optional[E]:	1✔
193	return self.__next	1✔
194
195	@property	1✔
196	def has_next(self) -> bool:	1✔
197	return self.__has_next	1✔
198
199	def pop(self) -> E:	1✔
200	if not self.__has_next:	1✔
201	raise StopIteration	1✔
202	nxt = self.__next	1✔
203	assert nxt is not None	1✔
204	try:	1✔
205	self.__next = next(self.__iterator)	1✔
206	self.__has_next = True	1✔
207	except StopIteration:	1✔
208	self.__next = None	1✔
209	self.__has_next = False	1✔
210	return nxt	1✔

SpiNNakerManchester / PACMAN / 7002704121

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