forked from M-Labs/artiq
1
0
Fork 0

drtio: CrossDomainNotification

This commit is contained in:
Sebastien Bourdeauducq 2016-10-21 22:45:45 +08:00
parent c71c4c89e0
commit 1e313afe64
2 changed files with 65 additions and 2 deletions

View File

@ -345,3 +345,23 @@ class _CrossDomainRequest(Module):
self.comb += reply.i.eq(srv_stb & srv_ack) self.comb += reply.i.eq(srv_stb & srv_ack)
class _CrossDomainNotification(Module):
def __init__(self, domain,
emi_stb, emi_data,
rec_stb, rec_ack, rec_data):
emi_data_r = Signal.like(emi_data)
self.specials += NoRetiming(emi_data_r)
dsync = getattr(self.sync, domain)
dsync += If(emi_stb, emi_data_r.eq(emi_data))
ps = PulseSynchronizer(domain, "sys")
self.submodules += ps
self.comb += ps.i.eq(emi_stb)
self.sync += [
If(rec_ack, rec_stb.eq(0)),
If(ps.o,
rec_data.eq(emi_data_r),
rec_stb.eq(1)
)
]

View File

@ -5,7 +5,8 @@ import random
from migen import * from migen import *
from artiq.gateware.drtio.rt_packets import * from artiq.gateware.drtio.rt_packets import *
from artiq.gateware.drtio.rt_packets import _CrossDomainRequest from artiq.gateware.drtio.rt_packets import (_CrossDomainRequest,
_CrossDomainNotification)
class PacketInterface: class PacketInterface:
@ -123,7 +124,7 @@ class TestSatellite(unittest.TestCase):
self.assertEqual(tx_times, rx_times) self.assertEqual(tx_times, rx_times)
class TestCrossDomainRequest(unittest.TestCase): class TestCDC(unittest.TestCase):
def test_cross_domain_request(self): def test_cross_domain_request(self):
prng = random.Random(1) prng = random.Random(1)
for sys_freq in 3, 6, 11: for sys_freq in 3, 6, 11:
@ -167,3 +168,45 @@ class TestCrossDomainRequest(unittest.TestCase):
{"sys": requester(), "srv": server()}, {"sys": requester(), "srv": server()},
{"sys": sys_freq, "srv": srv_freq}) {"sys": sys_freq, "srv": srv_freq})
self.assertEqual(test_seq, received_seq) self.assertEqual(test_seq, received_seq)
def test_cross_domain_notification(self):
prng = random.Random(1)
emi_stb = Signal()
emi_data = Signal(8)
rec_stb = Signal()
rec_ack = Signal()
rec_data = Signal(8)
test_seq = [23, 12, 8, 3, 28]
received_seq = []
def emitter():
for data in test_seq:
yield emi_stb.eq(1)
yield emi_data.eq(data)
yield
yield emi_stb.eq(0)
yield
for j in range(prng.randrange(0, 3)):
yield
def receiver():
for i in range(len(test_seq)):
while not (yield rec_stb):
yield
received_seq.append((yield rec_data))
yield rec_ack.eq(1)
yield
yield rec_ack.eq(0)
yield
for j in range(prng.randrange(0, 3)):
yield
dut = _CrossDomainNotification("emi",
emi_stb, emi_data,
rec_stb, rec_ack, rec_data)
run_simulation(dut,
{"emi": emitter(), "sys": receiver()},
{"emi": 13, "sys": 3})
self.assertEqual(test_seq, received_seq)