From 2f774d33eb51b8cf4eb3250c02b8b0a56b56876c Mon Sep 17 00:00:00 2001 From: Luke Kenneth Casson Leighton Date: Wed, 27 Mar 2019 14:08:04 +0000 Subject: [PATCH] add start of outputmux pipe test --- src/add/multipipe.py | 23 ++-- src/add/test_outmux_pipe.py | 234 ++++++++++++++++++++++++++++++++++++ 2 files changed, 246 insertions(+), 11 deletions(-) create mode 100644 src/add/test_outmux_pipe.py diff --git a/src/add/multipipe.py b/src/add/multipipe.py index 84f52f3e..af52ad9c 100644 --- a/src/add/multipipe.py +++ b/src/add/multipipe.py @@ -73,7 +73,7 @@ class MultiInControlBase: class MultiOutControlBase: """ Common functions for Pipeline API """ - def __init__(self, n_len=1): + def __init__(self, n_len=1, in_multi=None): """ Multi-output Control class. Conforms to same API as ControlBase... mostly. has additional indices to the multiple *output* stages [MultiInControlBase has multiple *input* stages] @@ -88,6 +88,7 @@ class MultiOutControlBase: # set up input and output IO ACK (prev/next ready/valid) self.p = PrevControl(in_multi) + n = [] for i in range(n_len): n.append(NextControl()) self.n = Array(n) @@ -145,17 +146,17 @@ class CombMultiOutPipeline(MultiOutControlBase): def __init__(self, stage, n_len, n_mux): MultiOutControlBase.__init__(self, n_len=n_len) self.stage = stage - self.p_mux = p_mux + self.n_mux = n_mux # set up the input and output data - for i in range(p_len): - self.p[i].i_data = stage.ispec() # input type - self.n.o_data = stage.ospec() + self.p.i_data = stage.ispec() # input type + for i in range(n_len): + self.n[i].o_data = stage.ospec() # output type def elaborate(self, platform): m = Module() - m.submodules += self.p_mux + #m.submodules += self.n_mux # need buffer register conforming to *input* spec r_data = self.stage.ispec() # input type @@ -174,17 +175,17 @@ class CombMultiOutPipeline(MultiOutControlBase): p_i_valid = Signal(reset_less=True) m.d.comb += p_i_valid.eq(self.p.i_valid_logic()) - mid = self.p_mux.m_id + mid = self.n_mux.m_id - for i in range(p_len): + for i in range(n_len): m.d.comb += data_valid[i].eq(0) m.d.comb += n_i_readyn[i].eq(1) - m.d.comb += self.n[i].o_valid.eq(data_valid[i]) - m.d.comb += self.p[mid].o_ready.eq(~data_valid[mid] | self.n.i_ready) + m.d.comb += self.n[i].o_valid.eq(0) + m.d.comb += self.n[mid].o_valid.eq(data_valid[mid]) m.d.comb += n_i_readyn[mid].eq(~self.n[mid].i_ready & data_valid[mid]) anyvalid = Signal(i, reset_less=True) av = [] - for i in range(p_len): + for i in range(n_len): av.append(~data_valid[i] | self.n[i].i_ready) anyvalid = Cat(*av) m.d.comb += self.p.o_ready.eq(anyvalid.bool()) diff --git a/src/add/test_outmux_pipe.py b/src/add/test_outmux_pipe.py new file mode 100644 index 00000000..a95d2d68 --- /dev/null +++ b/src/add/test_outmux_pipe.py @@ -0,0 +1,234 @@ +from random import randint +from math import log +from nmigen import Module, Signal, Cat +from nmigen.compat.sim import run_simulation +from nmigen.cli import verilog, rtlil + +from multipipe import CombMultiOutPipeline + + +class MuxUnbufferedPipeline(CombMultiOutPipeline): + def __init__(self, stage, n_len): + # HACK: stage is also the n-way multiplexer + CombMultiOutPipeline.__init__(self, stage, n_len=n_len, n_mux=stage) + + # HACK: n-mux is also the stage... so set the muxid equal to input mid + stage.m_id = self.p.i_data.mid + + def ports(self): + return self.p_mux.ports() + + +class PassInData: + def __init__(self): + self.mid = Signal(2, reset_less=True) + self.data = Signal(16, reset_less=True) + + def eq(self, i): + return [self.mid.eq(i.mid), self.data.eq(i.data)] + + def ports(self): + return [self.mid, self.data] + + +class PassThroughStage: + + def ispec(self): + return PassInData() + + def ospec(self): + return Signal(16, reset_less=True) + + def process(self, i): + return i.data + + + +def testbench(dut): + stb = yield dut.out_op.stb + assert stb == 0 + ack = yield dut.out_op.ack + assert ack == 0 + + # set row 1 input 0 + yield dut.rs[1].in_op[0].eq(5) + yield dut.rs[1].stb.eq(0b01) # strobe indicate 1st op ready + #yield dut.rs[1].ack.eq(1) + yield + + # check row 1 output (should be inactive) + decode = yield dut.rs[1].out_decode + assert decode == 0 + if False: + op0 = yield dut.rs[1].out_op[0] + op1 = yield dut.rs[1].out_op[1] + assert op0 == 0 and op1 == 0 + + # output should be inactive + out_stb = yield dut.out_op.stb + assert out_stb == 1 + + # set row 0 input 1 + yield dut.rs[1].in_op[1].eq(6) + yield dut.rs[1].stb.eq(0b11) # strobe indicate both ops ready + + # set acknowledgement of output... takes 1 cycle to respond + yield dut.out_op.ack.eq(1) + yield + yield dut.out_op.ack.eq(0) # clear ack on output + yield dut.rs[1].stb.eq(0) # clear row 1 strobe + + # output strobe should be active, MID should be 0 until "ack" is set... + out_stb = yield dut.out_op.stb + assert out_stb == 1 + out_mid = yield dut.mid + assert out_mid == 0 + + # ... and output should not yet be passed through either + op0 = yield dut.out_op.v[0] + op1 = yield dut.out_op.v[1] + assert op0 == 0 and op1 == 0 + + # wait for out_op.ack to activate... + yield dut.rs[1].stb.eq(0b00) # set row 1 strobes to zero + yield + + # *now* output should be passed through + op0 = yield dut.out_op.v[0] + op1 = yield dut.out_op.v[1] + assert op0 == 5 and op1 == 6 + + # set row 2 input + yield dut.rs[2].in_op[0].eq(3) + yield dut.rs[2].in_op[1].eq(4) + yield dut.rs[2].stb.eq(0b11) # strobe indicate 1st op ready + yield dut.out_op.ack.eq(1) # set output ack + yield + yield dut.rs[2].stb.eq(0) # clear row 2 strobe + yield dut.out_op.ack.eq(0) # set output ack + yield + op0 = yield dut.out_op.v[0] + op1 = yield dut.out_op.v[1] + assert op0 == 3 and op1 == 4, "op0 %d op1 %d" % (op0, op1) + out_mid = yield dut.mid + assert out_mid == 2 + + # set row 0 and 3 input + yield dut.rs[0].in_op[0].eq(9) + yield dut.rs[0].in_op[1].eq(8) + yield dut.rs[0].stb.eq(0b11) # strobe indicate 1st op ready + yield dut.rs[3].in_op[0].eq(1) + yield dut.rs[3].in_op[1].eq(2) + yield dut.rs[3].stb.eq(0b11) # strobe indicate 1st op ready + + # set acknowledgement of output... takes 1 cycle to respond + yield dut.out_op.ack.eq(1) + yield + yield dut.rs[0].stb.eq(0) # clear row 1 strobe + yield + out_mid = yield dut.mid + assert out_mid == 0, "out mid %d" % out_mid + + yield + yield dut.rs[3].stb.eq(0) # clear row 1 strobe + yield dut.out_op.ack.eq(0) # clear ack on output + yield + out_mid = yield dut.mid + assert out_mid == 3, "out mid %d" % out_mid + + +class OutputTest: + def __init__(self, dut): + self.dut = dut + self.di = [] + self.do = {} + self.tlen = 3 + for i in range(self.tlen * dut.num_rows): + mid = randint(0, dut.num_rows-1) + data = randint(0, 255) + (mid<<8) + if mid not in self.do: + self.do[mid] = [] + self.di.append((data, mid)) + self.do[mid].append(data) + + def send(self): + for i in range(self.tlen * dut.num_rows): + op2 = self.di[i][0] + mid = self.di[i][1] + rs = dut.p + yield rs.i_valid.eq(1) + yield rs.i_data.data.eq(op2) + yield rs.i_data.mid.eq(mid) + yield + o_p_ready = yield rs.o_ready + while not o_p_ready: + yield + o_p_ready = yield rs.o_ready + + print ("send", mid, i, hex(op2)) + + yield rs.i_valid.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) + + def rcv(self, mid): + out_i = 0 + count = 0 + while out_i != self.tlen: + count += 1 + if count == 1000: + break + #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 + n = self.dut.n[mid] + yield n.i_ready.eq(1) + yield + o_n_valid = yield n.o_valid + i_n_ready = yield n.i_ready + if not o_n_valid or not i_n_ready: + continue + + out_v = yield n.o_data + + print ("recv", mid, out_i, hex(out_v)) + + assert self.do[mid][out_i] == out_v # pass-through data + + out_i += 1 + + +class TestPriorityMuxPipe(MuxUnbufferedPipeline): + def __init__(self): + self.num_rows = 4 + stage = PassThroughStage() + MuxUnbufferedPipeline.__init__(self, stage, n_len=self.num_rows) + + def ports(self): + res = [self.p.i_valid, self.p.o_ready] + \ + self.p.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] + #self.n[i].o_data.ports() + return res + + +if __name__ == '__main__': + dut = TestPriorityMuxPipe() + vl = rtlil.convert(dut, ports=dut.ports()) + with open("test_outmux_pipe.il", "w") as f: + f.write(vl) + #run_simulation(dut, testbench(dut), vcd_name="test_inputgroup.vcd") + + test = OutputTest(dut) + run_simulation(dut, [test.rcv(1), test.rcv(0), + test.rcv(3), test.rcv(2), + test.send()], + vcd_name="test_outmux_pipe.vcd") + -- 2.30.2