rtio: fix DMA data MSB and stop signaling, self-checking unittest

This commit is contained in:
Sebastien Bourdeauducq 2016-12-05 17:26:52 +08:00
parent 43a5455058
commit 75ea13748a
2 changed files with 54 additions and 25 deletions

View File

@ -153,13 +153,16 @@ class RecordConverter(Module):
self.end_marker_found = Signal()
self.flush = Signal()
hdrlen = (layout_len(record_layout) - 512)//8
record_raw = Record(record_layout)
self.comb += [
record_raw.raw_bits().eq(stream_slicer.source),
self.source.channel.eq(record_raw.channel),
self.source.timestamp.eq(record_raw.timestamp),
self.source.address.eq(record_raw.address),
self.source.data.eq(record_raw.data)
Case(record_raw.length,
{hdrlen+i: self.source.data.eq(record_raw.data[:i*8])
for i in range(1, 512//8+1)}),
]
fsm = FSM(reset_state="FLOWING")
@ -328,9 +331,10 @@ class CRIMaster(Module, AutoCSR):
class DMA(Module):
def __init__(self, membus):
self.enable = CSRStorage(write_from_dev=True)
self.enable = CSR()
self.submodules.dma = DMAReader(membus, self.enable.storage)
flow_enable = Signal()
self.submodules.dma = DMAReader(membus, flow_enable)
self.submodules.slicer = RecordSlicer(len(membus.dat_w))
self.submodules.time_offset = TimeOffset()
self.submodules.cri_master = CRIMaster()
@ -345,24 +349,29 @@ class DMA(Module):
fsm = FSM(reset_state="IDLE")
self.submodules += fsm
self.comb += self.enable.dat_w.eq(0)
fsm.act("IDLE",
If(self.enable.storage, NextState("FLOWING"))
If(self.enable.re & self.enable.r, NextState("FLOWING"))
)
fsm.act("FLOWING",
If(self.slicer.end_marker_found, self.enable.we.eq(1)),
If(~self.enable.storage,
self.slicer.flush.eq(1),
NextState("WAIT_EOP")
self.enable.w.eq(1),
flow_enable.eq(1),
If(self.slicer.end_marker_found | (self.enable.re & ~self.enable.r),
NextState("FLUSH")
)
)
fsm.act("FLUSH",
self.enable.w.eq(1),
self.slicer.flush.eq(1),
NextState("WAIT_EOP")
)
fsm.act("WAIT_EOP",
self.enable.w.eq(1),
If(self.cri_master.sink.stb & self.cri_master.sink.ack & self.cri_master.sink.eop,
NextState("WAIT_CRI_MASTER")
)
)
fsm.act("WAIT_CRI_MASTER",
self.enable.w.eq(1),
If(~self.cri_master.busy, NextState("IDLE"))
)

View File

@ -3,7 +3,7 @@ import unittest
from migen import *
from misoc.interconnect import wishbone
from artiq.gateware.rtio import dma
from artiq.gateware.rtio import dma, cri
def encode_n(n, min_length, max_length):
@ -36,34 +36,54 @@ def pack(x, size):
return r
test_writes = [
(0x01, 0x23, 0x12, 0x33),
(0x901, 0x902, 0x911, 0xeeeeeeeeeeeeeefffffffffffffffffffffffffffffff28888177772736646717738388488),
(0x81, 0x288, 0x88, 0x8888)
]
class TB(Module):
def __init__(self, ws):
sequence = []
sequence += encode_record(0x01, 0x23, 0x12, 0x33)
sequence += encode_record(0x901, 0x902, 0x911, 0xeeeeeeeeeeeeeefffffffffffffffffffffffffffffff28888177772736646717738388488)
sequence += encode_record(0x81, 0x288, 0x88, 0x8888)
sequence = [b for write in test_writes for b in encode_record(*write)]
sequence.append(0)
self.sequence = pack(sequence, ws)
sequence = pack(sequence, ws)
bus = wishbone.Interface(ws*8)
self.submodules.memory = wishbone.SRAM(
1024, init=self.sequence, bus=bus)
1024, init=sequence, bus=bus)
self.submodules.dut = dma.DMA(bus)
# TODO: remove this hack when misoc supports csr write_from_dev simulation
self.sync += If(self.dut.enable.we, self.dut.enable.storage.eq(self.dut.enable.dat_w))
class TestDMA(unittest.TestCase):
def test_dma(self):
def test_dma_noerror(self):
ws = 64
tb = TB(ws)
def gen():
def do_dma():
for i in range(2):
yield from tb.dut.enable.write(1)
for i in range(30):
print((yield from tb.dut.enable.read()))
yield
while ((yield from tb.dut.enable.read())):
yield
run_simulation(tb, gen(), vcd_name="foo.vcd")
received = []
@passive
def rtio_sim():
dut_cri = tb.dut.cri
while True:
cmd = yield dut_cri.cmd
if cmd == cri.commands["nop"]:
pass
elif cmd == cri.commands["write"]:
channel = yield dut_cri.chan_sel
timestamp = yield dut_cri.o_timestamp
address = yield dut_cri.o_address
data = yield dut_cri.o_data
received.append((channel, timestamp, address, data))
else:
self.fail("unexpected RTIO command")
yield
run_simulation(tb, [do_dma(), rtio_sim()])
self.assertEqual(received, test_writes*2)