from enum import Enum
from nmigen import *
from nmigen.hdl.rec import *
from nmigen.cli import main
# REMAINING WORK
#
# make a quick-and-dirty testbench that selects a random inwidth, random outwidth, and randomly
# starts/stops the input/output and verifies that the sequence of bits coming out is the same
# as that coming in (can do with PRBS) and that bus interface constraints are not violated
class IndexDisambiguator(Enum):
LAST_OP_UNKNOWN = 0
LAST_OP_WAS_WRITE = 1
LAST_OP_WAS_READ = 2
class GearboxBusLayout(Layout):
def __init__(self, *, in_width, out_width): # the * forces keyword args
super().__init__([
# DATA
("data_in", unsigned(in_width)), # FROM SOURCE
("data_out", unsigned(out_width)), # TO DEST
# CONTROL
("valid_in", 1), # FROM SOURCE
("ready_out", 1), # TO SOURCE
("ready_in", 1), # FROM DEST
("valid_out", 1), # TO DEST
("fault", 1)
])
class GearboxBus(Record):
def __init__(self, *, in_width, out_width):
super().__init__(GearboxBusLayout(in_width=in_width, out_width=out_width))
class GearboxFCLayout(Layout):
def __init__(self, *, len_storage):
super().__init__([
# DATA
("read_ptr", unsigned(len(range(len_storage)))), # FROM GEARBOX
("write_ptr", unsigned(len(range(len_storage)))), # FROM GEARBOX
# CONTROL
("write_happens_this_cycle", 1), # TO GEARBOX
("read_happens_this_cycle", 1), # TO GEARBOX
("ready_in", 1), # FROM GEARBOX (FROM DOWNSTREAM)
("valid_in", 1), # FROM GEARBOX (FROM UPSTREAM)
])
class GearboxFCBus(Record):
def __init__(self, *, len_storage):
super().__init__(GearboxFCLayout(len_storage=len_storage))
class GearboxFlowControl(Elaboratable):
def __init__(self, *, in_width, out_width, len_storage):
self.in_width = in_width
self.out_width = out_width
self.len_storage = len_storage
self.bus = GearboxFCBus(len_storage=len_storage)
print(len(self.bus.read_ptr))
def elaborate(self, platform):
m = Module()
# The top-level flow control logic works as follows.
# First, we determine which operations are *possible* based on the read/write indices
# and the index disambiguator bit:
# 1) we determine if we have enough invalid bits in the buffer to write-in
# If so, we set ready_out to signal to upstream we're ready to ingest
can_write_this_cycle = Signal(1)
# 2) we determine if we have enough valid bits in the buffer to read-out
# If so, we set valid_out to signal to downstream we're ready to produce
can_read_this_cycle = Signal(1)
# Then, we look at the flow-control signals from upstream/downstream to see which
# transactions will happen, and set the following two internal signals, which are
# used to gate the read/write index updates:
write_happens_this_cycle = Signal(1)
read_happens_this_cycle = Signal(1)
with m.If(read_ptr == write_ptr): # the special case first
with m.Switch(disambiguator):
with m.Case(IndexDisambiguator.LAST_OP_UNKNOWN): # fault
m.d.comb += can_read_this_cycle.eq(0)
m.d.comb += can_write_this_cycle.eq(0)
m.d.comb += self.bus.fault.eq(1)
with m.Case(IndexDisambiguator.LAST_OP_WAS_WRITE): # completely full
m.d.comb += can_read_this_cycle.eq(1)
m.d.comb += can_write_this_cycle.eq(0)
with m.Case(IndexDisambiguator.LAST_OP_WAS_READ): # completely empty
m.d.comb += can_read_this_cycle.eq(0)
m.d.comb += can_write_this_cycle.eq(1)
with m.Elif(read_ptr < write_ptr):
# read_ptr < write_ptr. Here, the valid bits do not wrap, the invalid bits wrap.
# The valid bits are: inclusive [read_ptr, write_ptr) exclusive
# the invalid bits are inclusive [write_ptr, K] inclusive, union with inclusive [0, read_ptr) exclusive
# We first calculate the number of valid bits:
numvalid = Signal(range(len_storage))
m.d.comb += numvalid.eq(write_ptr - read_ptr)
with m.If(numvalid >= self.out_width):
m.d.comb += can_read_this_cycle.eq(1)
# We calculate the number of invalid bits:
numinvalid = Signal(range(len_storage))
m.d.comb += numinvalid.eq(len_storage - write_ptr + read_ptr)
with m.If(numinvalid >= self.in_width):
m.d.comb += can_write_this_cycle.eq(1)
class ArbitraryGearbox(Elaboratable):
def __init__(self, *, in_width, out_width):
self.in_width = in_width
self.out_width = out_width
self.bus = GearboxBus(in_width=in_width, out_width=out_width)
def elaborate(self, platform):
m = Module()
loop = Signal(1)
len_storage = self.in_width + self.out_width
m.submodules.flow_controller = flow_controller = GearboxFlowControl(in_width=self.in_width, out_width=self.out_width, len_storage=len_storage)
#storage = Signal(len_storage, reset=0b001_010_011_100_101_110_111)
#storage = Signal(len_storage, reset= 0b111_110_101_100_011_010_001)
storage = Signal(len_storage)
write_ptr = Signal(range(len_storage))
read_ptr = Signal(range(len_storage))
disambiguator = Signal(IndexDisambiguator, reset=IndexDisambiguator.LAST_OP_WAS_READ)
# The buffer is composed of two different flavor of bits:
# Invalid bits CANNOT BE READ OUT CAN BE WRITTEN TO
# Valid bits CAN BE READ OUT CANNOT BE WRITTEN TO
# and the location of those bits are given by the read_ptr and the write_ptr
# We only allow a read operation (which is not idempotent as the read_ptr is advanced) if:
# 1) the downstream interface is signaling ready
# 2) there are out_width valid bits in front of the read_ptr
# Likewise, we only allow a write operation (which is heavily non-idempotent as the write_ptr is advanced
# and the buffer is modified) if:
# 1) the upstream interface is signaling valid (we cannot allow ourselves to ingest invalid data!)
# 2) there are in_width invalid bits in front of the write_ptr
# There are multiple cases for read_ptr and write_ptr:
# write_ptr < read_ptr. Here, the valid bits wrap, and the invalid bits do not wrap.
# The valid bits are: inclusive [read_ptr, K] inclusive, union with inclusive [0, write_ptr) exclusive
# the invalid bits are inclusive [write_ptr, read_ptr) exclusive
# Naturally, there is a tricky edge case which requires an extra bit (literally) of disambiguation.
# If write_ptr == read_ptr we don't know if the buffer is entirely full or entirely empty.
# If the last operation was a read, and we find write_ptr == read_ptr, we know it's empty
# If the last operation was a write, and we find write_ptr == read_ptr, we know it's full
# We keep a bit of state -- which is only relevant if write_ptr == read_ptr, to keep track of
# this: IndexDisambiguator, which can either be LAST_OP_WAS_WRITE or LAST_OP_WAS_READ
with m.If(self.bus.fault == 0):
# read index update:
with m.If(0 == 1):
with m.If(read_ptr + self.out_width >= len_storage):
m.d.sync += read_ptr.eq(read_ptr + self.out_width - len_storage)
with m.Else():
m.d.sync += read_ptr.eq(read_ptr + self.out_width)
# read-out bits case analysis:
with m.If(read_ptr + self.out_width <= len_storage):
m.d.comb += self.bus.data_out.eq(storage.bit_select(read_ptr, self.out_width))
with m.Else():
with m.Switch(read_ptr):
for cand_rptr in range(len_storage - self.out_width + 1, len_storage):
with m.Case(cand_rptr):
m.d.comb += loop.eq(1)
non_wrapping_bitwidth = len_storage - cand_rptr
wrapping_bitwidth = self.out_width + cand_rptr - len_storage
m.d.comb += self.bus.data_out.eq(Cat(
storage.bit_select(cand_rptr, non_wrapping_bitwidth), # the non-wrapping bits are less-significant
storage.bit_select(0, wrapping_bitwidth) # ...than the wrapping bit
))
# write index update:
with m.If(0 == 1):
with m.If(write_ptr + self.in_width <= len_storage):
m.d.sync += write_ptr.eq(write_ptr + self.in_width - len_storage)
with m.Else():
m.d.sync += write_ptr.eq(write_ptr + self.in_width)
# actually write-in the bits
with m.If(write_ptr + self.out_width <= len_storage):
m.d.comb += self.bus.data_out.eq(storage.bit_select(read_ptr, self.out_width))
with m.Else():
with m.Switch(write_ptr):
for cand_wptr in range(len_storage - self.out_width + 1, len_storage):
with m.Case(cand_wptr):
m.d.comb += loop.eq(1)
non_wrapping_bitwidth = len_storage - cand_wptr
wrapping_bitwidth = self.out_width + cand_wptr - len_storage
m.d.comb += self.bus.data_out.eq(Cat(
storage.bit_select(cand_wptr, non_wrapping_bitwidth), # the non-wrapping bits are less-significant
storage.bit_select(0, wrapping_bitwidth) # ...than the wrapping bit
))
return m
class DummyPlug(Elaboratable):
#def __init__(self):
def elaborate(self, platform):
m = Module()
m.submodules.gearbox = gearbox = ArbitraryGearbox(in_width=(9-3), out_width=3)
counter = Signal(8)
m.d.sync += counter.eq(counter+1)
with m.If(counter == 3):
m.d.comb += gearbox.bus.data_in.eq(1)
return m
if __name__ == '__main__':
baka =DummyPlug()
main(baka)
#platform.build(DummyPlug())