from enum import Enum
from nmigen import *
from nmigen.hdl.rec import *
from nmigen.asserts import *
from nmigen.cli import main
# NEXT TASKS (IMMEDIATE)
# finish fleshing out code, attach buses from the Flow controller to the gearbox
# do the write-logic
# figure out how the Altera gearbox design works (number of pipeline stages and what each does)
# 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
# can make a testbench only operating on the indices and the flowcontrol signals that keeps a
# model of the valid/invalid bits and verifies that when a transaction happens, the right bits
# get read/written and the right indices get moved in the right ways
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()
internalfault = Signal()
# 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 bus signals, which are
# used to gate the read/write index updates:
# write_happens_this_cycle
# read_happens_this_cycle
# We replicate signals to avoid repeating "self.bus."
read_ptr = Signal(range(self.len_storage))
m.d.comb += read_ptr.eq(self.bus.read_ptr)
write_ptr = Signal(range(self.len_storage))
m.d.comb += write_ptr.eq(self.bus.write_ptr)
len_storage = self.len_storage
disambiguator = Signal(IndexDisambiguator, reset=IndexDisambiguator.LAST_OP_WAS_READ)
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.sync += internalfault.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 and invalid bits:
numvalid = Signal(range(len_storage))
numinvalid = Signal(range(len_storage))
m.d.comb += numvalid.eq(write_ptr - read_ptr)
m.d.comb += numinvalid.eq(len_storage - write_ptr + read_ptr)
with m.If(numvalid >= self.out_width):
m.d.comb += can_read_this_cycle.eq(1)
with m.If(numinvalid >= self.in_width):
m.d.comb += can_write_this_cycle.eq(1)
with m.Elif(read_ptr > 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
# We first calculate the number of valid and invalid bits:
numvalid = Signal(range(len_storage))
numinvalid = Signal(range(len_storage))
m.d.comb += numvalid.eq(len_storage - read_ptr + write_ptr)
m.d.comb += numinvalid.eq(read_ptr - write_ptr)
with m.If(numvalid >= self.out_width):
m.d.comb += can_read_this_cycle.eq(1)
with m.If(numinvalid >= self.in_width):
m.d.comb += can_write_this_cycle.eq(1)
with m.Else(): # should never happen
m.d.sync += internalfault.eq(1)
return m
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))
# 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:
# 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(self.bus.ready_in == 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
))
#m.d.sync += Assert(read_ptr == 0)
#m.d.comb += Assume((self.bus.data_in == 3) |( self.bus.data_in==5))
return m
class IdempotentGearbox(Elaboratable):
def __init__(self, *, in_width, out_width):
self.in_width = in_width
self.out_width = out_width
assert(in_width == out_width)
self.bus = GearboxBus(in_width=in_width, out_width=out_width)
def elaborate(self, platform):
m = Module()
skid_buffer = Signal(self.in_width)
skid_buffer_frozen = Signal(1)
# we signal ready to upstream if and only if the skid buffer is empty
m.d.comb += self.bus.ready_out.eq(~skid_buffer_frozen)
# should we capture a value in the skid buffer?
# we should if upstream has data for us *and* downstream cannot accept
# If we want to fill our skid buffer, we need:
# 1. valid input data (bus.valid_in == 1)
# 2. a buffer that is empty (skid_buffer_frozen == 0)
# 3. stalled downstream (bus.ready_in == 0) & (bus.valid_out == 1)
with m.If(self.bus.valid_in & (~skid_buffer_frozen) & (~self.bus.ready_in) & (self.bus.valid_out)):
m.d.sync += skid_buffer_frozen.eq(1)
# if downstream is accepting data again, we will flush our skid buffer
with m.If(self.bus.ready_in == 1):
m.d.sync += skid_buffer_frozen.eq(0)
# Stalled == (bus.ready_in == 0 & bus.valid_out == 1)
# so not stalled = !(bus.ready_in == 0 & bus.valid_out == 1)
# = bus.ready_in == 1 | bus.valid_out == 0
# by de Morgan
with m.If((self.bus.ready_in) | (~self.bus.valid_out)):
m.d.sync += self.bus.valid_out.eq(self.bus.valid_in | skid_buffer_frozen)
# data path
# always clock data into the skid buffer as long as the buffer isn't filled
with m.If(skid_buffer_frozen == 0):
m.d.sync += skid_buffer.eq(self.bus.data_in)
# not stalled condition
with m.If((self.bus.ready_in) | (~self.bus.valid_out)):
with m.If(skid_buffer_frozen == 1):
m.d.sync += self.bus.data_out.eq(skid_buffer)
with m.Elif(self.bus.valid_in == 1):
m.d.sync += self.bus.data_out.eq(self.bus.data_in)
with m.If(Past(ResetSignal()) == 0):
with m.If((self.bus.valid_in == 1) & (Past(self.bus.valid_in) == 1)):
m.d.comb += Assert(Rose(self.bus.valid_out))
return m
# This is non-synthesizable but is intended to provide a model for non-unitary ratio gearbox
# in order to do formal verification and model-checking against both the unitary-ratio gearbox
# and also the arbitrary ratio gearbox.
class GoldenGearboxModel(Elaboratable):
def __init__(self, *, in_width, out_width, sim_memory_size):
self.in_width = in_width
self.out_width = out_width
assert(in_width == out_width)
self.memory = Signal(sim_memory_size)
self.bus = GearboxBus(in_width=in_width, out_width=out_width)
def elaborate(self, platform):
write_ptr = Signal(range(sim_memory_size))
read_ptr = Signal(range(sim_memory_size))
m = Module()
return m
class DummyPlug(Elaboratable):
#def __init__(self):
def elaborate(self, platform):
m = Module()
m.submodules.gearbox = gearbox = IdempotentGearbox(in_width=(3), out_width=3)
#counter = Signal(8)
#m.d.sync += counter.eq(counter+1)
# with m.If(counter == 3):
# m.d.comb += gearbox.bus.valid_in.eq(1)
m.d.comb += gearbox.bus.data_in.eq(AnySeq(3))
m.d.comb += gearbox.bus.ready_in.eq(AnySeq(1))
m.d.comb += gearbox.bus.valid_in.eq(AnySeq(1))
return m
if __name__ == '__main__':
baka =DummyPlug()
main(baka)
#platform.build(DummyPlug())