| from collections import defaultdict | |
| from data_structs import PinDirection, ConnectionType | |
| from utils import yield_muxes | |
| from rr_utils import add_node, connect | |
| # ============================================================================= | |
| class SwitchboxModel(object): | |
| """ | |
| Represents a model of connectivity of a concrete instance of a switchbox. | |
| """ | |
| def __init__(self, graph, loc, phy_loc, switchbox): | |
| self.graph = graph | |
| self.loc = loc | |
| self.phy_loc = phy_loc | |
| self.switchbox = switchbox | |
| self.fixed_muxsels = set() | |
| self.fixed_muxes = None | |
| self.mux_input_to_node = {} | |
| self.mux_output_to_node = {} | |
| self.input_to_node = {} | |
| @staticmethod | |
| def get_metadata_for_mux(loc, stage, switch_id, mux_id, pin_id): | |
| """ | |
| Formats fasm features for the given edge representin a switchbox mux. | |
| Returns a list of fasm features. | |
| """ | |
| metadata = [] | |
| # Format prefix | |
| prefix = "X{}Y{}.ROUTING".format(loc.x, loc.y) | |
| # A mux in the HIGHWAY stage | |
| if stage.type == "HIGHWAY": | |
| feature = "I_highway.IM{}.I_pg{}".format(switch_id, pin_id) | |
| # A mux in the STREET stage | |
| elif stage.type == "STREET": | |
| feature = "I_street.Isb{}{}.I_M{}.I_pg{}".format( | |
| stage.id + 1, switch_id + 1, mux_id, pin_id | |
| ) | |
| else: | |
| assert False, stage | |
| metadata.append(".".join([prefix, feature])) | |
| return metadata | |
| @staticmethod | |
| def get_chan_dirs_for_stage(stage): | |
| """ | |
| Returns channel directions for inputs and outputs of a stage. | |
| """ | |
| if stage.type == "HIGHWAY": | |
| return "Y", "X" | |
| elif stage.type == "STREET": | |
| dir_inp = "Y" if (stage.id % 2) else "X" | |
| dir_out = "X" if (stage.id % 2) else "Y" | |
| return dir_inp, dir_out | |
| else: | |
| assert False, stage.type | |
| @staticmethod | |
| def get_connection(switchbox, src, dst): | |
| """ | |
| Returns the SwitchboxConnection object that spans two muxes given their | |
| locations. Parameters src and dst should be tuples containing: | |
| (stage_id, switch_id, mux_id) | |
| """ | |
| for connection in switchbox.connections: | |
| c_src = ( | |
| connection.src.stage_id, connection.src.switch_id, | |
| connection.src.mux_id | |
| ) | |
| c_dst = ( | |
| connection.dst.stage_id, connection.dst.switch_id, | |
| connection.dst.mux_id | |
| ) | |
| if c_src == src and c_dst == dst: | |
| return connection | |
| return None | |
| @staticmethod | |
| def get_switchbox_routes(switchbox, out_name, inp_name): | |
| """ | |
| Returns a list of routes inside the switchbox that connect the given | |
| output pin with the given input pin. | |
| Returns a list of lists. Each inner list contain tuples with | |
| (stage_id, switch_id, mux_id, pin_id) | |
| """ | |
| # Route list | |
| routes = [] | |
| def walk(ctx, target_name, route=None): | |
| """ | |
| An inner recursive walk function. Walks from a location within | |
| the switchbox until the target input pin is reached. | |
| """ | |
| # Copy/create the route list | |
| if route is None: | |
| route = [] | |
| else: | |
| route = list(route) | |
| # Add this mux | |
| route.append(ctx) | |
| # Get the mux object | |
| stage_id, switch_id, mux_id = ctx | |
| stage = switchbox.stages[stage_id] | |
| switch = stage.switches[switch_id] | |
| mux = switch.muxes[mux_id] | |
| # Get its input connections | |
| connections = {} | |
| for connection in switchbox.connections: | |
| is_stage_id = connection.dst.stage_id == stage_id | |
| is_switch_id = connection.dst.switch_id == switch_id | |
| is_mux_id = connection.dst.mux_id == mux_id | |
| if is_stage_id and is_switch_id and is_mux_id: | |
| connections[connection.dst.pin_id] = connection | |
| # Expand all its inputs | |
| for pin_id, pin in mux.inputs.items(): | |
| # An input goes to another mux, expand it | |
| if pin.name is None and pin_id in connections: | |
| connection = connections[pin_id] | |
| next_ctx = ( | |
| connection.src.stage_id, | |
| connection.src.switch_id, | |
| connection.src.mux_id, | |
| ) | |
| walk(next_ctx, target_name, route) | |
| # This is a switchbox input | |
| elif pin.name is not None: | |
| # We've hit the target | |
| if pin.name == target_name: | |
| # Append the current mux and its selection | |
| final_route = list(route) | |
| final_route[-1] = tuple( | |
| list(final_route[-1]) + [pin_id] | |
| ) | |
| # Trace the route back, append mux selections | |
| for i in range(len(final_route) - 1): | |
| dst = final_route[i][:3] | |
| src = final_route[i + 1][:3] | |
| connection = SwitchboxModel.get_connection( | |
| switchbox, src, dst | |
| ) | |
| sel = connection.dst.pin_id | |
| final_route[i] = tuple( | |
| list(final_route[i]) + [sel] | |
| ) | |
| routes.append(final_route) | |
| # Should not happen | |
| else: | |
| assert False, pin | |
| # Get the output pin | |
| pin = switchbox.outputs[out_name] | |
| assert len(pin.locs) == 1 | |
| loc = pin.locs[0] | |
| # Walk from the output, collect routes | |
| ctx = ( | |
| loc.stage_id, | |
| loc.switch_id, | |
| loc.mux_id, | |
| ) | |
| walk(ctx, inp_name) | |
| return routes | |
| def _create_muxes(self): | |
| """ | |
| Creates nodes for muxes and internal edges within them. Annotates the | |
| internal edges with fasm data. | |
| Builds maps of muxs' inputs and outpus to VPR nodes. | |
| """ | |
| # Build mux driver timing map. Assign each mux output its timing data | |
| driver_timing = {} | |
| for connection in self.switchbox.connections: | |
| src = connection.src | |
| stage = self.switchbox.stages[src.stage_id] | |
| switch = stage.switches[src.switch_id] | |
| mux = switch.muxes[src.mux_id] | |
| pin = mux.inputs[src.pin_id] | |
| if pin.id not in mux.timing: | |
| continue | |
| timing = mux.timing[pin.id].driver | |
| key = (src.stage_id, src.switch_id, src.mux_id) | |
| if key in driver_timing: | |
| assert driver_timing[key] == timing, \ | |
| (self.loc, key, driver_timing[key], timing) | |
| else: | |
| driver_timing[key] = timing | |
| # Create muxes | |
| segment_id = self.graph.get_segment_id_from_name("sbox") | |
| for stage, switch, mux in yield_muxes(self.switchbox): | |
| dir_inp, dir_out = self.get_chan_dirs_for_stage(stage) | |
| # Output node | |
| key = (stage.id, switch.id, mux.id) | |
| assert key not in self.mux_output_to_node | |
| out_node = add_node(self.graph, self.loc, dir_out, segment_id) | |
| self.mux_output_to_node[key] = out_node | |
| # Intermediate output node | |
| int_node = add_node(self.graph, self.loc, dir_out, segment_id) | |
| # Get switch id for the switch assigned to the driver. If | |
| # there is none then use the delayless switch. Probably the | |
| # driver is connected to a const. | |
| if key in driver_timing: | |
| switch_id = self.graph.get_switch_id( | |
| driver_timing[key].vpr_switch | |
| ) | |
| else: | |
| switch_id = self.graph.get_delayless_switch_id() | |
| # Output driver edge | |
| connect( | |
| self.graph, | |
| int_node, | |
| out_node, | |
| switch_id=switch_id, | |
| segment_id=segment_id, | |
| ) | |
| # Input nodes + mux edges | |
| for pin in mux.inputs.values(): | |
| key = (stage.id, switch.id, mux.id, pin.id) | |
| assert key not in self.mux_input_to_node | |
| # Input node | |
| inp_node = add_node(self.graph, self.loc, dir_inp, segment_id) | |
| self.mux_input_to_node[key] = inp_node | |
| # Get mux metadata | |
| metadata = self.get_metadata_for_mux( | |
| self.phy_loc, stage, switch.id, mux.id, pin.id | |
| ) | |
| if len(metadata): | |
| meta_name = "fasm_features" | |
| meta_value = "\n".join(metadata) | |
| else: | |
| meta_name = None | |
| meta_value = "" | |
| # Get switch id for the switch assigned to the mux edge. If | |
| # there is none then use the delayless switch. Probably the | |
| # edge is connected to a const. | |
| if pin.id in mux.timing: | |
| switch_id = self.graph.get_switch_id( | |
| mux.timing[pin.id].sink.vpr_switch | |
| ) | |
| else: | |
| switch_id = self.graph.get_delayless_switch_id() | |
| # Mux switch with appropriate timing and fasm metadata | |
| connect( | |
| self.graph, | |
| inp_node, | |
| int_node, | |
| switch_id=switch_id, | |
| segment_id=segment_id, | |
| meta_name=meta_name, | |
| meta_value=meta_value, | |
| ) | |
| def _connect_muxes(self): | |
| """ | |
| Creates VPR edges that connects muxes within the switchbox. | |
| """ | |
| segment_id = self.graph.get_segment_id_from_name("sbox") | |
| switch_id = self.graph.get_switch_id("short") | |
| # Add internal connections between muxes. | |
| for connection in self.switchbox.connections: | |
| src = connection.src | |
| dst = connection.dst | |
| # Check | |
| assert src.pin_id == 0, src | |
| assert src.pin_direction == PinDirection.OUTPUT, src | |
| # Get the input node | |
| key = (dst.stage_id, dst.switch_id, dst.mux_id, dst.pin_id) | |
| dst_node = self.mux_input_to_node[key] | |
| # Get the output node | |
| key = (src.stage_id, src.switch_id, src.mux_id) | |
| src_node = self.mux_output_to_node[key] | |
| # Connect | |
| connect( | |
| self.graph, | |
| src_node, | |
| dst_node, | |
| switch_id=switch_id, | |
| segment_id=segment_id | |
| ) | |
| def _create_input_drivers(self): | |
| """ | |
| Creates VPR nodes and edges that model input connectivity of the | |
| switchbox. | |
| """ | |
| # Create a driver map containing all mux pin locations that are | |
| # connected to a driver. The map is indexed by (pin_name, vpr_switch) | |
| # and groups togeather inputs that should be driver by a specific | |
| # switch due to the timing model. | |
| driver_map = defaultdict(lambda: []) | |
| for pin in self.switchbox.inputs.values(): | |
| for loc in pin.locs: | |
| stage = self.switchbox.stages[loc.stage_id] | |
| switch = stage.switches[loc.switch_id] | |
| mux = switch.muxes[loc.mux_id] | |
| pin = mux.inputs[loc.pin_id] | |
| if pin.id not in mux.timing: | |
| vpr_switch = None | |
| else: | |
| vpr_switch = mux.timing[pin.id].driver.vpr_switch | |
| key = (pin.name, vpr_switch) | |
| driver_map[key].append(loc) | |
| # Create input nodes for each input pin | |
| segment_id = self.graph.get_segment_id_from_name("sbox") | |
| for pin in self.switchbox.inputs.values(): | |
| node = add_node(self.graph, self.loc, "Y", segment_id) | |
| assert pin.name not in self.input_to_node, pin.name | |
| self.input_to_node[pin.name] = node | |
| # Create driver nodes, connect everything | |
| for (pin_name, vpr_switch), locs in driver_map.items(): | |
| # Create the driver node | |
| drv_node = add_node(self.graph, self.loc, "X", segment_id) | |
| # Connect input node to the driver node. Use the switch with timing. | |
| inp_node = self.input_to_node[pin_name] | |
| # Get switch id for the switch assigned to the driver. If | |
| # there is none then use the delayless switch. Probably the | |
| # driver is connected to a const. | |
| if vpr_switch is not None: | |
| switch_id = self.graph.get_switch_id(vpr_switch) | |
| else: | |
| switch_id = self.graph.get_delayless_switch_id() | |
| # Connect | |
| connect( | |
| self.graph, | |
| inp_node, | |
| drv_node, | |
| switch_id=switch_id, | |
| segment_id=segment_id | |
| ) | |
| # Now connect the driver node with its loads | |
| switch_id = self.graph.get_switch_id("short") | |
| for loc in locs: | |
| key = (loc.stage_id, loc.switch_id, loc.mux_id, loc.pin_id) | |
| dst_node = self.mux_input_to_node[key] | |
| connect( | |
| self.graph, | |
| drv_node, | |
| dst_node, | |
| switch_id=switch_id, | |
| segment_id=segment_id | |
| ) | |
| def build(self): | |
| """ | |
| Build the switchbox model by creating and adding its nodes and edges | |
| to the RR graph. | |
| """ | |
| # TODO: FIXME: When a switchbox model contains fixed muxes only they | |
| # should be removed and the rest of the switchbox should be added | |
| # to the rr graph. For now if there is any fixed mux, remove the | |
| # whole switchbox. | |
| if len(self.fixed_muxsels): | |
| # A list of muxes to avoid | |
| self.fixed_muxes = set([f[:3] for f in self.fixed_muxsels]) | |
| print( | |
| "Switchbox model '{}' at '{}' contains '{}' fixed muxes.". | |
| format(self.switchbox.type, self.loc, len(self.fixed_muxes)) | |
| ) | |
| return | |
| # Create and connect muxes | |
| self._create_muxes() | |
| self._connect_muxes() | |
| # Create and connect input drivers models | |
| self._create_input_drivers() | |
| def get_input_node(self, pin_name): | |
| """ | |
| Returns a VPR node associated with the given input of the switchbox | |
| """ | |
| return self.input_to_node[pin_name] | |
| def get_output_node(self, pin_name): | |
| """ | |
| Returns a VPR node associated with the given output of the switchbox | |
| """ | |
| # Get the output pin | |
| pin = self.switchbox.outputs[pin_name] | |
| assert len(pin.locs) == 1 | |
| loc = pin.locs[0] | |
| # Return its node | |
| key = (loc.stage_id, loc.switch_id, loc.mux_id) | |
| return self.mux_output_to_node[key] | |
| # ============================================================================= | |
| class QmuxSwitchboxModel(SwitchboxModel): | |
| """ | |
| Represents a model of connectivity of a concrete instance of a switchbox | |
| located at a QMUX tile | |
| """ | |
| def __init__( | |
| self, graph, loc, phy_loc, switchbox, qmux_cells, connections | |
| ): | |
| super().__init__(graph, loc, phy_loc, switchbox) | |
| self.qmux_cells = qmux_cells | |
| self.connections = connections | |
| self.ctrl_routes = {} | |
| def _find_control_routes(self): | |
| """ | |
| """ | |
| PINS = ( | |
| "IS0", | |
| "IS1", | |
| ) | |
| for cell in self.qmux_cells.values(): | |
| # Get IS0 and IS1 connection endpoints | |
| eps = {} | |
| for connection in self.connections: | |
| if connection.dst.type == ConnectionType.CLOCK: | |
| dst_cell, dst_pin = connection.dst.pin.split(".") | |
| if dst_cell == cell.name and dst_pin in PINS: | |
| eps[dst_pin] = connection.src | |
| # Find all routes for IS0 and IS1 pins that go to GND and VCC | |
| routes = {} | |
| for pin in PINS: | |
| # Find the routes | |
| vcc_routes = self.get_switchbox_routes( | |
| self.switchbox, eps[pin].pin, "VCC" | |
| ) | |
| gnd_routes = self.get_switchbox_routes( | |
| self.switchbox, eps[pin].pin, "GND" | |
| ) | |
| routes[pin] = {"VCC": vcc_routes, "GND": gnd_routes} | |
| # Store | |
| self.ctrl_routes[cell.name] = routes | |
| def build(self): | |
| """ | |
| Builds the QMUX switchbox model | |
| """ | |
| # Find routes inside the switchbox for GMUX control pins | |
| self._find_control_routes() | |
| # Filter routes so GND routes go through stage 2, switch 0 and VCC | |
| # routes go through stage 2, switch 1. | |
| for cell_name, cell_routes in self.ctrl_routes.items(): | |
| for pin, pin_routes in cell_routes.items(): | |
| for net, net_routes in pin_routes.items(): | |
| routes = [] | |
| for route in net_routes: | |
| # Assume 3-stage switchbox | |
| assert len( | |
| route | |
| ) == 3, "FIXME: Assuming 3-stage switchbox!" | |
| if route[1][1] == 0 and net == "GND": | |
| routes.append(route) | |
| if route[1][1] == 1 and net == "VCC": | |
| routes.append(route) | |
| pin_routes[net] = routes | |
| def get_input_node(self, pin_name): | |
| return None | |
| def get_output_node(self, pin_name): | |
| return None |