Hardware
/
litex
mirror of https://github.com/enjoy-digital/litex.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

soc/interconnect/packet: Remove last_be support in LiteX, specialized Packetizer/Depacketizer have been moved to LiteEth to simplify development and avoid eventual regresion on others cores. 

As seen during the last LiteEth developments, last_be data qualifier is not easy to handle correctly and should be replaced by a simpler data qualifier (similar to AXI's tkeep/tstrb). It will
be easier to do so by having a local copy of Packetizer/Depacketizer directly in LiteEth (still with last_be support) and work on the simpler data qualifier in LiteX (and test it on LitePCIe).
This commit is contained in:
Florent Kermarrec 2021-10-25 11:16:01 +02:00
parent 96101521be
commit 47b3c9bc08
3 changed files with 0 additions and 702 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

175
litex/soc/interconnect/packet.py
View File

@ -254,85 +254,6 @@ class Packetizer(Module):
) )
) )
# Last BE.
# FIXME: Switch to a simpler Data Qualifier and remove Last BE handling/section.
if hasattr(sink, "last_be") and hasattr(source, "last_be"):
# For an 8-bit data path, last_be really should be 1 when last is
# asserted, other values do not make sense. However, legacy code
# might not set last_be at all, and thus it will be set to 0. To
# remain compatible with this code, this "corrects" last_be for
# 8-bit paths by setting it to the value of last.
if len(sink.last_be) == 1:
sink_last_be = Signal.like(sink.last_be)
self.comb += [ sink_last_be.eq(sink.last) ]
else:
sink_last_be = sink.last_be
# last_be needs to be right-rotated by the number of bytes which
# would be required to have a properly aligned header.
right_rot_by = header_leftover
# Calculate a rotated last_be
new_last_be = Signal.like(sink_last_be)
self.comb += [
new_last_be.eq(Cat([
sink_last_be[(i - right_rot_by) % bytes_per_clk]
for i in range(bytes_per_clk)
]))
]
# Conditionally delay the calculated last_be for one clock cycle, if
# it now applies to the next bus word OR if the source is not ready.
delayed_last_be = Signal.like(sink_last_be)
# FSM used to conveniently assign combinational and synchronous
# signals in the same context.
self.submodules.last_be_fsm = last_be_fsm = FSM(reset_state="DEFAULT")
# Whether the main FSM is in one of the DATA-COPY states. This is
# important as we overwrite sink.ready below and need to have
# different behavior depending on the Packetizer's state
in_data_copy = Signal()
self.comb += [
in_data_copy.eq(self.fsm.ongoing("ALIGNED-DATA-COPY") | self.fsm.ongoing("UNALIGNED-DATA-COPY"))
]
self.last_be_fsm.act("DEFAULT",
# Test whether our right-shift causes a wrap-around. In that
# case apply the last value to the current bus word. Otherwise
# delay it to the next.
If(in_data_copy & sink.last & (sink_last_be > new_last_be),
# Right shift did not wrap around, need to delay the
# calculated last_be value and last by one cycle.
source.last.eq(0),
source.last_be.eq(0),
If(source.ready & source.valid,
NextValue(delayed_last_be, new_last_be),
NextState("DELAYED"),
),
).Elif(in_data_copy,
# Output the calculated last_be value on the current packet
# already. For the next sink packet, ignore any last_be
source.last.eq(sink.last),
source.last_be.eq(new_last_be),
),
If(in_data_copy,
sink.ready.eq(source.ready),
).Elif(self.fsm.ongoing("IDLE"),
sink.ready.eq(~sink.valid),
)
)
self.last_be_fsm.act("DELAYED",
# Output the delayed last and last_be signals
source.last.eq(1),
source.last_be.eq(delayed_last_be),
sink.ready.eq(0),
If(source.ready,
NextState("DEFAULT"),
),
)
# Error. # Error.
if hasattr(sink, "error") and hasattr(source, "error"): if hasattr(sink, "error") and hasattr(source, "error"):
self.comb += source.error.eq(sink.error) self.comb += source.error.eq(sink.error)
@ -438,102 +359,6 @@ class Depacketizer(Module):
if hasattr(sink, "error") and hasattr(source, "error"): if hasattr(sink, "error") and hasattr(source, "error"):
self.comb += source.error.eq(sink.error) self.comb += source.error.eq(sink.error)
# Last BE.
# FIXME: Switch to a simpler Data Qualifier and remove Last BE handling/section.
if hasattr(sink, "last_be") and hasattr(source, "last_be"):
# For an 8-bit data path, last_be really should be 1 when last is
# asserted, other values do not make sense. However, legacy code
# might not set last_be at all, and thus it will be set to 0. To
# remain compatible with this code, this "corrects" last_be for
# 8-bit paths by setting it to the value of last.
if len(sink.last_be) == 1:
sink_last_be = Signal.like(sink.last_be)
self.comb += [ sink_last_be.eq(sink.last) ]
else:
sink_last_be = sink.last_be
# last_be needs to be left-rotated by the number of bytes which
# would be required to have a properly aligned header.
left_rot_by = (bytes_per_clk - header_leftover) % bytes_per_clk
# Calculate a rotated last_be
new_last_be = Signal.like(sink_last_be)
self.comb += [
new_last_be.eq(Cat([
sink_last_be[(i - left_rot_by) % bytes_per_clk]
for i in range(bytes_per_clk)
]))
]
# Conditionally delay the calculated last_be for one clock cycle, if
# it now applies to the next bus word.
delayed_last_be = Signal.like(sink_last_be)
# FSM used to conveniently assign combinational and synchronous
# signals in the same context.
self.submodules.last_be_fsm = last_be_fsm = FSM(reset_state="DEFAULT")
# Whether the main FSM is / was in one of the DATA-COPY states. This
# is important as we must handle a special case when last is
# asserted while the Depacketizer has just transitioned out of
# writing the header, which we can then detect by checking
# (~was_in_copy & is_in_copy).
is_in_copy = Signal()
was_in_copy = Signal()
self.comb += [
is_in_copy.eq(
self.fsm.ongoing("ALIGNED-DATA-COPY") | self.fsm.ongoing("UNALIGNED-DATA-COPY")
)
]
self.sync += [
was_in_copy.eq(is_in_copy)
]
self.last_be_fsm.act("DEFAULT",
# Test whether our left-shift has caused an wrap-around, in that
# case delay last and last_be and apply to the next bus word.
If(sink.valid & sink.last & (sink_last_be > new_last_be),
# last_be did wrap around. Need to delay the calculated
# last_be value and last by one cycle.
source.last.eq(0),
source.last_be.eq(0),
# Normally just wait until a source bus transaction occurred
# (ready + valid) until the delayed last_be can be
# output. However, if the first word is also the last, this
# will mean that the source isn't valid currently as the
# header is still being sent. Thus, if sink.valid and
# sink.last and we've just transitioned into the data copy
# phase, we can immediately jump into the DELAYED state (as
# the very first bus word containing proper data is also the
# last one, and we've just transitioned to putting that on
# the bus).
If((source.ready & source.valid) | (~was_in_copy & is_in_copy),
NextValue(delayed_last_be, new_last_be),
NextState("DELAYED"),
),
).Elif(sink.last,
# Simply forward the calculated last_be value.
source.last.eq(1),
source.last_be.eq(new_last_be),
),
If(self.fsm.ongoing("ALIGNED-DATA-COPY") \
| (self.fsm.ongoing("UNALIGNED-DATA-COPY") & ~fsm_from_idle),
sink.ready.eq(source.ready),
).Else(
sink.ready.eq(1),
),
)
self.last_be_fsm.act("DELAYED",
# Output the delayed last and last_be signals
source.last.eq(1),
source.last_be.eq(delayed_last_be),
sink.ready.eq(0),
If(source.ready & source.valid,
NextState("DEFAULT"),
),
)
# PacketFIFO --------------------------------------------------------------------------------------- # PacketFIFO ---------------------------------------------------------------------------------------
class PacketFIFO(Module): class PacketFIFO(Module):

188
test/test_packet2.py
View File

@ -1,188 +0,0 @@
#
# This file is part of LiteX.
#
# Copyright (c) 2021 Leon Schuermann <leon@is.currently.online>
# SPDX-License-Identifier: BSD-2-Clause
import unittest
import random
from migen import *
from litex.soc.interconnect.stream import *
from litex.soc.interconnect.packet import *
from .test_stream2 import StreamPacket, stream_inserter, stream_collector, compare_packets
def mask_last_be(dw, data, last_be):
masked_data = 0
for byte in range(dw // 8):
if 2**byte > last_be:
break
masked_data |= data & (0xFF << (byte * 8))
return masked_data
class TestPacket(unittest.TestCase):
def loopback_test(self, dw, seed=42, with_last_be=False, debug_print=False):
# Independent random number generator to ensure we're the
# stream_inserter and stream_collectors still have
# reproducible behavior independent of the headers
prng = random.Random(seed + 5)
# Generate a random number of differently sized header fields
nheader_fields = prng.randrange(16)
i = 0
packet_header_length = 0
packet_header_fields = {}
while packet_header_length < dw // 8 or i < nheader_fields:
# Header field size can be 1, 2, 4, 8, 16 bytes
field_length = 2**prng.randrange(5)
packet_header_fields["field{}_{}b".format(i, field_length * 8)] = \
HeaderField(packet_header_length, 0, field_length * 8)
packet_header_length += field_length
i += 1
packet_header = Header(
fields = packet_header_fields,
length = packet_header_length,
swap_field_bytes = bool(prng.getrandbits(1)))
def packet_description(dw):
param_layout = packet_header.get_layout()
payload_layout = [("data", dw)]
if with_last_be:
payload_layout += [("last_be", dw // 8)]
return EndpointDescription(payload_layout, param_layout)
def raw_description(dw):
payload_layout = [("data", dw)]
if with_last_be:
payload_layout += [("last_be", dw // 8)]
return EndpointDescription(payload_layout)
# Prepare packets
npackets = 32
packets = []
for n in range(npackets):
header = {}
for name, headerfield in packet_header_fields.items():
header[name] = prng.randrange(2**headerfield.width)
datas = [prng.randrange(2**8) for _ in range(prng.randrange(dw - 1) + 1)]
packets.append(StreamPacket(datas, header))
class DUT(Module):
def __init__(self):
self.submodules.packetizer = Packetizer(
packet_description(dw),
raw_description(dw),
packet_header,
)
self.submodules.depacketizer = Depacketizer(
raw_description(dw),
packet_description(dw),
packet_header,
)
self.comb += self.packetizer.source.connect(self.depacketizer.sink)
self.sink, self.source = self.packetizer.sink, self.depacketizer.source
dut = DUT()
recvd_packets = []
run_simulation(
dut,
[
stream_inserter(
dut.sink,
src=packets,
seed=seed,
debug_print=debug_print,
valid_rand=50,
),
stream_collector(
dut.source,
dest=recvd_packets,
expect_npackets=npackets,
seed=seed,
debug_print=debug_print,
ready_rand=50,
),
],
)
# When we don't have a last_be signal, the Packetizer will simply throw
# away the partial bus word. The Depacketizer will then fill up these
# values with garbage again. Thus we also have to remove the proper
# amount of bytes from the sent packets so the comparson will work.
if not with_last_be and dw != 8:
# Modulo operation which returns the divisor instead of zero.
def upmod(a, b):
return b if a % b == 0 else a % b
for (packet, recvd_packet) in zip(packets, recvd_packets):
# How many bytes of the header have to be interleaved with the
# first data word on the bus.
header_leftover = packet_header_length % (dw // 8)
# If the last word of our data would fit together with the
# header_leftover bytes in a single bus word, all data (plus
# some trailing garbage) will arrive. Otherwise, some data bytes
# will be missing.
if header_leftover != 0 and \
header_leftover + upmod(len(packet.data), dw // 8) <= (dw // 8):
# The entire data will arrive, plus some trailing
# garbage. Remove that.
garbage_bytes = -len(packet.data) % (dw // 8)
recvd_packet.data = recvd_packet.data[:-garbage_bytes]
else:
# header_leftover bytes in received data have been replaced
# with garbage. Remove these bytes from the received and
# sent data.
recvd_packet.data = recvd_packet.data[:-header_leftover]
packet.data = packet.data[:len(recvd_packet.data)]
self.assertTrue(compare_packets(packets, recvd_packets))
# def test_8bit_loopback(self):
# for seed in range(42, 48):
# with self.subTest(seed=seed):
# self.loopback_test(dw=8, seed=seed)
#
# def test_8bit_loopback_last_be(self):
# for seed in range(42, 48):
# with self.subTest(seed=seed):
# self.loopback_test(dw=8, seed=seed, with_last_be=True)
#
# def test_32bit_loopback(self):
# for seed in range(42, 48):
# with self.subTest(seed=seed):
# self.loopback_test(dw=32, seed=seed)
#
# def test_32bit_loopback_last_be(self):
# for seed in range(42, 48):
# with self.subTest(seed=seed):
# self.loopback_test(dw=32, seed=seed, with_last_be=True)
#
# def test_64bit_loopback(self):
# for seed in range(42, 48):
# with self.subTest(seed=seed):
# self.loopback_test(dw=64, seed=seed)
#
# def test_64bit_loopback_last_be(self):
# for seed in range(42, 48):
# with self.subTest(seed=seed):
# self.loopback_test(dw=64, seed=seed, with_last_be=True)
#
# def test_128bit_loopback(self):
# for seed in range(42, 48):
# with self.subTest(seed=seed):
# self.loopback_test(dw=128, seed=seed)
#
# def test_128bit_loopback_last_be(self):
# for seed in range(42, 48):
# with self.subTest(seed=seed):
# self.loopback_test(dw=128, seed=seed, with_last_be=True)

339
test/test_stream2.py
View File

@ -1,339 +0,0 @@
#
# This file is part of LiteX.
#
# Copyright (c) 2021 Leon Schuermann <leon@is.currently.online>
# SPDX-License-Identifier: BSD-2-Clause
import unittest
import unittest
import random
import itertools
import sys
from migen import *
from litex.soc.interconnect.stream import *
# Function to iterate over chunks of data, from
# https://docs.python.org/3/library/itertools.html#itertools-recipes
def grouper(iterable, n, fillvalue=None):
"Collect data into fixed-length chunks or blocks"
# grouper('ABCDEFG', 3, 'x') --> ABC DEF Gxx"
args = [iter(iterable)] * n
return itertools.zip_longest(*args, fillvalue=fillvalue)
class StreamPacket:
def __init__(self, data, params={}):
# Data must be a list of bytes
assert type(data) == list
for b in data:
assert type(b) == int and b >= 0 and b < 256
# Params must be a dictionary of strings mapping to integers
assert type(params) == dict
for param_key, param_value in params.items():
assert type(param_key) == str
assert type(param_value) == int
self.data = data
self.params = params
def compare(self, other, quiet=True, output_target=sys.stdout):
if len(self.data) != len(other.data):
if not quiet:
print("Length mismatch in number of received bytes of packet:" \
" {} {}".format(len(self.data), len(other.data)),
file=sys.stdout)
return False
for nbyte, (byte_a, byte_b) in enumerate(zip(self.data, other.data)):
if byte_a != byte_b:
if not quiet:
print("Mismatch between sent and received bytes {}: " \
"0x{:02x} 0x{:02x}".format(nbyte, byte_a, byte_b),
file=sys.stdout)
return False
if set(self.params.keys()) != set(other.params.keys()):
if not quiet:
print("Sent and received packets have different param fields:" \
" {} {}".format(self.params.keys(), other.params.keys()),
file=sys.stdout)
return False
for param_name, self_param_value in self.params.items():
other_param_value = other.params[param_name]
if self_param_value != other_param_value:
if not quiet:
print("Sent and received packets have different value for" \
" param signal \"{}\": 0x{:x} 0x{:x}".format(
param_name,
self_param_value,
other_param_value),
file=sys.stdout)
return False
return True
def stream_inserter(
sink,
src,
seed=42,
valid_rand=50,
debug_print=False,
broken_8bit_last_be=True):
"""Insert a list of packets of bytes on to the stream interface `sink`. If
`sink` has a `last_be` signal, that is set accordingly.
"""
prng = random.Random(seed)
# Extract the data width from the provided sink Endpoint
dw = len(sink.data)
# Make sure dw is evenly divisible by 8 as the logic below relies on
# that. Also, last_be wouldn't make much sense otherwise.
assert dw % 8 == 0
# If a last_be signal is provided, it must contain one bit per byte of data,
# i.e. be dw // 8 long.
if hasattr(sink, "last_be"):
assert dw // 8 == len(sink.last_be)
# src is a list of lists. Each list represents a packet of bytes. Send each
# packet over the bus.
for pi, packet in enumerate(src):
assert len(packet.data) > 0, "Packets of length 0 are not compatible " \
"with the ready/valid stream interface"
# Each packet is a list. We must send dw // 8 bytes at a time. Use the
# grouper method to get a chunked iterator over the packet bytes and
# shift them to their correct position. Use a random filler byte to
# complete a bus word.
words = []
for chunk in grouper(packet.data, dw // 8, prng.randrange(256)):
word = 0
for i, b in enumerate(chunk):
assert b >= 0 and b < 256
word |= b << (i * 8)
words += [word]
if hasattr(sink, "last_be"):
encoded_last_be = Constant(
1 << ((len(packet.data) - 1) % (dw // 8)),
bits_sign=len(sink.last_be)
)
# In legacy code for 8bit data paths last_be might not be set
# properly: while last_be should always be equal to last for 8bit
# data paths, if new code interacts with old code which is not yet
# last_be aware, it might always be deasserted. If
# broken_8bit_last_be is set and we have an 8bit data path, randomly
# set last_be to either one or zero to check whether the DUT handles
# these cases properly.
if broken_8bit_last_be and dw == 8:
encoded_last_be = Constant(prng.randrange(2), bits_sign=1)
# At the very beginning of the packet transmission, set the param
# signals
for param_signal, param_value in packet.params.items():
yield getattr(sink, param_signal).eq(param_value)
for i, word in enumerate(words):
last = i == len(words) - 1
# Place the word on the bus, if its the last word set last and
# last_be accordingly and finally set sink to valid
yield sink.data.eq(word)
yield sink.last.eq(last)
if hasattr(sink, "last_be"):
if last:
yield sink.last_be.eq(encoded_last_be)
else:
yield sink.last_be.eq(0)
yield sink.valid.eq(1)
yield
# Wait until the sink has become ready for one clock cycle
while not (yield sink.ready):
yield
# Set sink to not valid for a random amount of time
yield sink.valid.eq(0)
while prng.randrange(100) < valid_rand:
yield
# Okay, we've transmitted a packet. We must set sink.valid to false, for
# good measure clear all other signals as well. We don't explicitly
# yield, given a there might be a new packet waiting already.
yield sink.data.eq(0)
yield sink.last.eq(0)
if hasattr(sink, "last_be"):
yield sink.last_be.eq(0)
for param_signal in packet.params.keys():
yield getattr(sink, param_signal).eq(0)
yield sink.valid.eq(0)
if debug_print:
print("Sent packet {}.".format(pi), file=sys.stderr)
# All packets have been transmitted. sink.valid has already been
# deasserted, yield once to properly apply that value.
yield
def stream_collector(
source,
dest=[],
expect_npackets=None,
seed=42,
ready_rand=50,
debug_print=False):
"""Consume some packets of bytes from the stream interface
`source`. If `source` has a `last_be` signal, that is respected
properly.
"""
prng = random.Random(seed)
# Extract the data width from the provided source endpoint
dw = len(source.data)
# Make sure dw is evenly divisible by 8 as the logic below relies on
# that. Also, last_be wouldn't make much sense otherwise.
assert dw % 8 == 0
# If a last_be signal is provided, it must contain one bit per byte of data,
# i.e. be dw // 8 long.
if hasattr(source, "last_be"):
assert dw // 8 == len(source.last_be)
# Extract "param_signals" from the source Endpoint. They are extracted on
# the first valid word of a packet. If dest will be a list of tuples with
# data and param signals if there are any, otherwise just a list of lists.
param_signals = [
signal_name for signal_name, _, _ in source.param.layout
] if hasattr(source, "param") else []
# Loop for collecting individual packets, separated by source.last
while expect_npackets == None or len(dest) < expect_npackets:
# Buffer for the current packet
collected_bytes = []
param_signal_states = {}
# Iterate until "last" has been seen. That concludes the end of a bus
# transaction / packet.
read_last = False
first_word = True
while not read_last:
# We are ready to accept another bus word
yield source.ready.eq(1)
yield
# Wait for data to become valid
while (yield source.valid) == 0:
yield
# Data is now valid, read it byte by byte
data = yield source.data
for byte in range(dw // 8):
if (yield source.last) == 1:
read_last = True
if hasattr(source, "last_be") and \
2**byte > (yield source.last_be):
break
collected_bytes += [((data >> (byte * 8)) & 0xFF)]
# Also, if this is the first loop iteration, latch all param signals
for param_signal in param_signals:
param_signal_states[param_signal] = \
yield getattr(source, param_signal)
# Set source to not valid for a random amount of time
yield source.ready.eq(0)
while prng.randrange(100) < ready_rand:
yield
# This is no longer the first loop iteration
first_word = False
# A full packet has been read. Append it to dest.
dest += [StreamPacket(collected_bytes, param_signal_states)]
if debug_print:
print("Received packet {}.".format(len(dest) - 1), file=sys.stderr)
def generate_test_packets(npackets, seed=42):
# Generate a number of last-terminated bus transaction byte contents (dubbed
# packets)
prng = random.Random(42)
packets = []
for _ in range(npackets):
# With a random number of bytes from [1, 1024)
values = []
for _ in range(prng.randrange(1023) + 1):
# With random values from [0, 256).
values += [prng.randrange(256)]
packets += [StreamPacket(values)]
return packets
def compare_packets(packets_a, packets_b):
if len(packets_a) != len(packets_b):
print("Length mismatch in number of received packets: {} {}"
.format(len(packets_a), len(packets_b)), file=sys.stderr)
return False
for npacket, (packet_a, packet_b) in enumerate(zip(packets_a, packets_b)):
if not packet_a.compare(packet_b):
print("Error in packet", npacket)
packet_a.compare(packet_b, quiet=False)
return False
return True
class TestStream(unittest.TestCase):
def pipe_test(self, dut, seed=42, npackets=64, debug_print=False):
# Get some data to test with
packets = generate_test_packets(npackets, seed=seed)
# Buffer for received packets (filled by collector)
recvd_packets = []
run_simulation(
dut,
[
stream_inserter(
dut.sink,
src=packets,
debug_print=debug_print,
seed=seed,
),
stream_collector(
dut.source,
dest=recvd_packets,
expect_npackets=npackets,
debug_print=debug_print,
seed=seed,
),
],
)
self.assertTrue(compare_packets(packets, recvd_packets))
def test_pipe_valid(self):
# PipeValid either connects the entire payload or not. Thus we don't
# need to test for 8bit support or a missing last_be signal
# specifically. This test does however ensure that last_be will continue
# to be respected in the future.
dut = PipeValid([("data", 32), ("last_be", 4)])
self.pipe_test(dut)
def test_pipe_ready(self):
# PipeReady either connects the entire stream Endpoint or not. Thus we
# don't need to test for 8bit support or a missing last_be signal
# specifically. This test does however ensure that last_be will continue
# to be respected in the future.
dut = PipeReady([("data", 64), ("last_be", 8)])
self.pipe_test(dut)