from nmigen import Module, Signal, Cat
from nmigen.compat.sim import run_simulation
from nmigen.cli import verilog, rtlil
-from nmigen.lib.coding import PriorityEncoder
-from example_buf_pipe import UnbufferedPipeline
+from singlepipe import PassThroughStage
+from multipipe import (CombMultiInPipeline, PriorityCombMuxInPipe)
-class InputPriorityArbiter:
- def __init__(self, pipe, num_rows):
- self.pipe = pipe
- self.num_rows = num_rows
- self.mmax = int(log(self.num_rows) / log(2))
- self.m_id = Signal(self.mmax, reset_less=True) # multiplex id
- self.active = Signal(reset_less=True)
-
- def elaborate(self, platform):
- m = Module()
-
- assert len(self.pipe.p) == self.num_rows, \
- "must declare input to be same size"
- pe = PriorityEncoder(self.num_rows)
- m.submodules.selector = pe
-
- # connect priority encoder
- in_ready = []
- for i in range(self.num_rows):
- p_i_valid = Signal(reset_less=True)
- m.d.comb += p_i_valid.eq(self.pipe.p[i].i_valid_logic())
- in_ready.append(p_i_valid)
- m.d.comb += pe.i.eq(Cat(*in_ready)) # array of input "valids"
- m.d.comb += self.active.eq(~pe.n) # encoder active (one input valid)
- m.d.comb += self.m_id.eq(pe.o) # output one active input
-
- return m
-
- def ports(self):
- return [self.m_id, self.active]
-
-
-class PriorityUnbufferedPipeline(UnbufferedPipeline):
- def __init__(self, stage, p_len=4):
- p_mux = InputPriorityArbiter(self, p_len)
- UnbufferedPipeline.__init__(self, stage, p_len=p_len, p_mux=p_mux)
-
- def ports(self):
- return self.p_mux.ports()
- #return UnbufferedPipeline.ports(self) + self.p_mux.ports()
-
class PassData:
def __init__(self):
- self.mid = Signal(2)
- self.idx = Signal(6)
- self.data = Signal(16)
+ self.mid = Signal(2, reset_less=True)
+ self.idx = Signal(6, reset_less=True)
+ self.data = Signal(16, reset_less=True)
def eq(self, i):
return [self.mid.eq(i.mid), self.idx.eq(i.idx), self.data.eq(i.data)]
def ports(self):
return [self.mid, self.idx, self.data]
-class PassThroughStage:
- def ispec(self):
- return PassData()
- def ospec(self):
- return self.ispec() # same as ospec
-
- def process(self, i):
- return i # pass-through
-
-
def testbench(dut):
stb = yield dut.out_op.stb
self.dut = dut
self.di = {}
self.do = {}
- self.tlen = 4
+ self.tlen = 10
for mid in range(dut.num_rows):
self.di[mid] = {}
self.do[mid] = {}
for i in range(self.tlen):
- self.di[mid][i] = randint(0, 100)
+ self.di[mid][i] = randint(0, 100) + (mid<<8)
self.do[mid][i] = self.di[mid][i]
def send(self, mid):
for i in range(self.tlen):
op2 = self.di[mid][i]
rs = dut.p[mid]
- yield rs.i_valid.eq(0)
- o_p_ready = yield rs.o_ready
+ yield rs.valid_i.eq(1)
+ yield rs.data_i.data.eq(op2)
+ yield rs.data_i.idx.eq(i)
+ yield rs.data_i.mid.eq(mid)
+ yield
+ o_p_ready = yield rs.ready_o
while not o_p_ready:
yield
- o_p_ready = yield rs.o_ready
+ o_p_ready = yield rs.ready_o
- print ("send", mid, i, op2)
+ print ("send", mid, i, hex(op2))
- yield rs.i_valid.eq(1)
- yield rs.i_data.data.eq(op2)
- yield rs.i_data.idx.eq(i)
- yield rs.i_data.mid.eq(mid)
- #for v in self.dut.set_input((op2, i, mid)):
- # yield v
- yield
+ yield rs.valid_i.eq(0)
+ # wait random period of time before queueing another value
+ for i in range(randint(0, 3)):
+ yield
- yield rs.i_valid.eq(0)
+ yield rs.valid_i.eq(0)
## wait random period of time before queueing another value
#for i in range(randint(0, 3)):
# yield
#stall_range = randint(0, 3)
#for j in range(randint(1,10)):
# stall = randint(0, stall_range) != 0
- # yield self.dut.n[0].i_ready.eq(stall)
+ # yield self.dut.n[0].ready_i.eq(stall)
# yield
- n = self.dut.n[0]
- yield n.i_ready.eq(1)
+ n = self.dut.n
+ yield n.ready_i.eq(1)
yield
- o_n_valid = yield n.o_valid
- i_n_ready = yield n.i_ready
+ o_n_valid = yield n.valid_o
+ i_n_ready = yield n.ready_i
if not o_n_valid or not i_n_ready:
continue
- mid = yield n.o_data.mid
- out_i = yield n.o_data.idx
- out_v = yield n.o_data.data
+ mid = yield n.data_o.mid
+ out_i = yield n.data_o.idx
+ out_v = yield n.data_o.data
- print ("recv", mid, out_i, out_v)
+ print ("recv", mid, out_i, hex(out_v))
# see if this output has occurred already, delete it if it has
assert out_i in self.do[mid], "out_i %d not in array %s" % \
break
-class TestPriorityMuxPipe(PriorityUnbufferedPipeline):
+class TestPriorityMuxPipe(PriorityCombMuxInPipe):
def __init__(self):
- self.num_rows = 2
- stage = PassThroughStage()
- PriorityUnbufferedPipeline.__init__(self, stage, p_len=self.num_rows)
-
- def ports(self):
- res = []
- for i in range(len(self.p)):
- res += [self.p[i].i_valid, self.p[i].o_ready] + \
- self.p[i].i_data.ports()
- for i in range(len(self.n)):
- res += [self.n[i].i_ready, self.n[i].o_valid] + \
- self.n[i].o_data.ports()
- return res
+ self.num_rows = 4
+ def iospecfn(): return PassData()
+ stage = PassThroughStage(iospecfn)
+ PriorityCombMuxInPipe.__init__(self, stage, p_len=self.num_rows)
if __name__ == '__main__':
dut = TestPriorityMuxPipe()
vl = rtlil.convert(dut, ports=dut.ports())
- with open("test_inputgroup.il", "w") as f:
+ with open("test_inputgroup_multi.il", "w") as f:
f.write(vl)
#run_simulation(dut, testbench(dut), vcd_name="test_inputgroup.vcd")
test = InputTest(dut)
run_simulation(dut, [test.send(1), test.send(0),
- #test.send(3), test.send(2),
+ test.send(3), test.send(2),
test.rcv()],
- vcd_name="test_inputgroup_parallel.vcd")
+ vcd_name="test_inputgroup_multi.vcd")
projects / ieee754fpu.git / blobdiff