128 lines
7.1 KiB
Python
128 lines
7.1 KiB
Python
from nmigen import *
|
|
from nmigen.asserts import *
|
|
from nmigen.test.utils import *
|
|
from ...sed.output_network import *
|
|
from datetime import datetime
|
|
|
|
"""
|
|
Verification tasks for OutputNetwork
|
|
"""
|
|
|
|
class OutputNetworkSpec(Elaboratable):
|
|
def __init__(self, lane_count, seqn_width, layout_payload):
|
|
self.lane_count = lane_count
|
|
self.seqn_width = seqn_width
|
|
self.layout_payload = layout_payload
|
|
def elaborate(self, platform):
|
|
m = Module()
|
|
m.submodules.output_network = output_network = OutputNetwork(self.lane_count, self.seqn_width, self.layout_payload)
|
|
|
|
# Model arbitrary inputs for network nodes
|
|
for i in range(output_network.lane_count):
|
|
m.d.comb += output_network.input[i].eq(AnySeq(1))
|
|
m.d.comb += output_network.input[i].seqn.eq(AnySeq(output_network.seqn_width))
|
|
m.d.comb += output_network.input[i].replace_occured.eq(0)
|
|
m.d.comb += output_network.input[i].nondata_replace_occured.eq(0)
|
|
for field, width in output_network.layout_payload:
|
|
m.d.comb += getattr(output_network.input[i].payload, field).eq(AnySeq(width))
|
|
# Indicator of when inputs from the first clock cycle make it
|
|
# through the sorting network
|
|
network_latency = latency(output_network.lane_count)
|
|
counter = Signal(range(network_latency + 1))
|
|
m.d.sync += counter.eq(counter + 1)
|
|
with m.If(counter == network_latency):
|
|
m.d.sync += counter.eq(counter)
|
|
f_output_valid = Signal()
|
|
m.d.comb += f_output_valid.eq(counter == network_latency)
|
|
|
|
with m.If(f_output_valid):
|
|
valid_replacement_occurred = Signal()
|
|
for node in output_network.output:
|
|
with m.If(node.valid & node.replace_occured):
|
|
m.d.comb += valid_replacement_occurred.eq(1)
|
|
valid_channels_unique = Signal(reset=1)
|
|
for node1 in range(len(output_network.input)):
|
|
with m.If(Past(output_network.input[node1].valid, clocks=network_latency)):
|
|
for node2 in range(node1):
|
|
with m.If(Past(output_network.input[node2].valid, clocks=network_latency)):
|
|
k1 = Past(output_network.input[node1].payload.channel, clocks=network_latency)
|
|
k2 = Past(output_network.input[node2].payload.channel, clocks=network_latency)
|
|
with m.If(k1 == k2):
|
|
m.d.comb += valid_channels_unique.eq(0)
|
|
# Among the valid outputs, if there are no replacements then:
|
|
# - Channel numbers are unique among valid inputs
|
|
# - All valid inputs make it through the sorting network unmodified
|
|
with m.If(~valid_replacement_occurred):
|
|
m.d.comb += Assert(valid_channels_unique)
|
|
for input_node in output_network.input:
|
|
appeared = Signal()
|
|
for output_node in output_network.output:
|
|
match = Signal(reset=1)
|
|
with m.If(Past(input_node.valid, clocks=network_latency) != output_node.valid):
|
|
m.d.comb += match.eq(0)
|
|
with m.If(Past(input_node.seqn, clocks=network_latency) != output_node.seqn):
|
|
m.d.comb += match.eq(0)
|
|
with m.If(Past(input_node.replace_occured, clocks=network_latency) != output_node.replace_occured):
|
|
m.d.comb += match.eq(0)
|
|
with m.If(Past(input_node.nondata_replace_occured, clocks=network_latency) != output_node.nondata_replace_occured):
|
|
m.d.comb += match.eq(0)
|
|
for field, _ in output_network.layout_payload:
|
|
with m.If(Past(getattr(input_node.payload, field), clocks=network_latency) != getattr(output_node.payload, field)):
|
|
m.d.comb += match.eq(0)
|
|
with m.If(match):
|
|
m.d.comb += appeared.eq(1)
|
|
with m.If(Past(input_node.valid, clocks=network_latency)):
|
|
m.d.comb += Assert(appeared)
|
|
# Otherwise:
|
|
# - There is a channel number collision among the valid inputs
|
|
# - All channel numbers in valid inputs appear at least once as a
|
|
# valid output
|
|
# - All valid outputs correspond to a valid input modulo accounting
|
|
# information
|
|
with m.Else():
|
|
m.d.comb += Assert(~valid_channels_unique)
|
|
for input_node in output_network.input:
|
|
with m.If(Past(input_node.valid, clocks=network_latency)):
|
|
appeared = Signal()
|
|
for output_node in output_network.output:
|
|
with m.If(output_node.valid & (output_node.payload.channel == Past(input_node.payload.channel, clocks=network_latency))):
|
|
m.d.comb += appeared.eq(1)
|
|
m.d.comb += Assert(appeared)
|
|
for output_node in output_network.output:
|
|
with m.If(output_node.valid):
|
|
found_input = Signal()
|
|
for input_node in output_network.input:
|
|
match = Signal(reset=1)
|
|
with m.If(~Past(input_node.valid, clocks=network_latency)):
|
|
m.d.comb += match.eq(0)
|
|
with m.If(Past(input_node.seqn, clocks=network_latency) != output_node.seqn):
|
|
m.d.comb += match.eq(0)
|
|
for field, _ in output_network.layout_payload:
|
|
with m.If(Past(getattr(input_node.payload, field), clocks=network_latency) != getattr(output_node.payload, field)):
|
|
m.d.comb += match.eq(0)
|
|
with m.If(match):
|
|
m.d.comb += found_input.eq(1)
|
|
m.d.comb += Assert(found_input)
|
|
|
|
return m
|
|
|
|
class OutputNetworkTestCase(FHDLTestCase):
|
|
def verify(self):
|
|
startTime = datetime.now()
|
|
self.assertFormal(
|
|
OutputNetworkSpec(2, 2, [("data", 32), ("channel", 3)]),
|
|
mode="prove", depth=latency(2))
|
|
print("Verification of sorting network with 2 lanes completed in %f ms" % ((datetime.now() - startTime).total_seconds() * 1000))
|
|
startTime = datetime.now()
|
|
self.assertFormal(
|
|
OutputNetworkSpec(4, 2, [("data", 32), ("channel", 3)]),
|
|
mode="prove", depth=latency(4))
|
|
print("Verification of sorting network with 4 lanes completed in %f ms" % ((datetime.now() - startTime).total_seconds() * 1000))
|
|
startTime = datetime.now()
|
|
self.assertFormal(
|
|
OutputNetworkSpec(8, 2, [("data", 32), ("channel", 3)]),
|
|
mode="prove", depth=latency(8))
|
|
print("Verification of sorting network with 8 lanes completed in %f ms" % ((datetime.now() - startTime).total_seconds() * 1000))
|
|
|
|
OutputNetworkTestCase().verify()
|