stream/Buffer: Integrate PipeValid/PipeReady (both configurable) and add tests.

Allow selecting pipelining of valid/data or/and ready and creating a full Skid Buffer
(Pipeline of both valid/data and ready).
This commit is contained in:
Florent Kermarrec 2022-09-07 08:58:51 +02:00
parent ce90181046
commit a6acfb9a37
2 changed files with 38 additions and 1 deletions

View File

@ -766,7 +766,32 @@ class PipeReady(Module):
# Buffer ------------------------------------------------------------------------------------------- # Buffer -------------------------------------------------------------------------------------------
class Buffer(PipeValid): pass # FIXME: Replace Buffer with PipeValid in codebase? class Buffer(Module):
"""Pipe valid/payload and/or ready to cut timing path"""
def __init__(self, layout, pipe_valid=True, pipe_ready=False):
self.sink = sink = Endpoint(layout)
self.source = source = Endpoint(layout)
# # #
pipeline = []
# Pipe Valid (Optional).
if pipe_valid:
self.submodules.pipe_valid = PipeValid(layout)
pipeline.append(self.pipe_valid)
# Pipe Ready (Optional).
if pipe_ready:
self.submodules.pipe_ready = PipeReady(layout)
pipeline.append(self.pipe_ready)
# Buffer Pipeline.
self.submodules.pipeline = Pipeline(
sink,
*pipeline,
source
)
# Cast --------------------------------------------------------------------------------------------- # Cast ---------------------------------------------------------------------------------------------

View File

@ -50,3 +50,15 @@ class TestStream(unittest.TestCase):
def test_pipe_ready(self): def test_pipe_ready(self):
dut = PipeReady([("data", 8)]) dut = PipeReady([("data", 8)])
self.pipe_test(dut) self.pipe_test(dut)
def test_buffer_valid(self):
dut = Buffer([("data", 8)], pipe_valid=True, pipe_ready=False)
self.pipe_test(dut)
def test_buffer_ready(self):
dut = Buffer([("data", 8)], pipe_valid=False, pipe_ready=True)
self.pipe_test(dut)
def test_buffer_valid_ready(self):
dut = Buffer([("data", 8)], pipe_valid=True, pipe_ready=True)
self.pipe_test(dut)