Skip to content

Commit

Permalink
RFC 36 (async/await testing) (kuznia-rdzeni/coreblocks#742)
Browse files Browse the repository at this point in the history
  • Loading branch information
@tilk
tilk authored Nov 19, 2024
1 parent 06b79aa commit 5c44ab8
Show file tree
Hide file tree
Showing 26 changed files with 987 additions and 936 deletions.
120 changes: 63 additions & 57 deletions test/core/test_transactions.py
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from abc import abstractmethod
from unittest.case import TestCase
from amaranth_types import HasElaborate
import pytest
from amaranth import *
from amaranth.sim import *
Expand Down Expand Up @@ -56,52 +57,52 @@ def count_test(self, sched, cnt):
assert len(sched.grant) == cnt
assert len(sched.valid) == 1

def sim_step(self, sched, request, expected_grant):
yield sched.requests.eq(request)
yield Tick()
async def sim_step(self, sim, sched: Scheduler, request: int, expected_grant: int):
sim.set(sched.requests, request)
_, _, valid, grant = await sim.tick().sample(sched.valid, sched.grant)

if request == 0:
assert not (yield sched.valid)
assert not valid
else:
assert (yield sched.grant) == expected_grant
assert (yield sched.valid)
assert grant == expected_grant
assert valid

def test_single(self):
sched = Scheduler(1)
self.count_test(sched, 1)

def process():
yield from self.sim_step(sched, 0, 0)
yield from self.sim_step(sched, 1, 1)
yield from self.sim_step(sched, 1, 1)
yield from self.sim_step(sched, 0, 0)
async def process(sim):
await self.sim_step(sim, sched, 0, 0)
await self.sim_step(sim, sched, 1, 1)
await self.sim_step(sim, sched, 1, 1)
await self.sim_step(sim, sched, 0, 0)

with self.run_simulation(sched) as sim:
sim.add_process(process)
sim.add_testbench(process)

def test_multi(self):
sched = Scheduler(4)
self.count_test(sched, 4)

def process():
yield from self.sim_step(sched, 0b0000, 0b0000)
yield from self.sim_step(sched, 0b1010, 0b0010)
yield from self.sim_step(sched, 0b1010, 0b1000)
yield from self.sim_step(sched, 0b1010, 0b0010)
yield from self.sim_step(sched, 0b1001, 0b1000)
yield from self.sim_step(sched, 0b1001, 0b0001)
async def process(sim):
await self.sim_step(sim, sched, 0b0000, 0b0000)
await self.sim_step(sim, sched, 0b1010, 0b0010)
await self.sim_step(sim, sched, 0b1010, 0b1000)
await self.sim_step(sim, sched, 0b1010, 0b0010)
await self.sim_step(sim, sched, 0b1001, 0b1000)
await self.sim_step(sim, sched, 0b1001, 0b0001)

yield from self.sim_step(sched, 0b1111, 0b0010)
yield from self.sim_step(sched, 0b1111, 0b0100)
yield from self.sim_step(sched, 0b1111, 0b1000)
yield from self.sim_step(sched, 0b1111, 0b0001)
await self.sim_step(sim, sched, 0b1111, 0b0010)
await self.sim_step(sim, sched, 0b1111, 0b0100)
await self.sim_step(sim, sched, 0b1111, 0b1000)
await self.sim_step(sim, sched, 0b1111, 0b0001)

yield from self.sim_step(sched, 0b0000, 0b0000)
yield from self.sim_step(sched, 0b0010, 0b0010)
yield from self.sim_step(sched, 0b0010, 0b0010)
await self.sim_step(sim, sched, 0b0000, 0b0000)
await self.sim_step(sim, sched, 0b0010, 0b0010)
await self.sim_step(sim, sched, 0b0010, 0b0010)

with self.run_simulation(sched) as sim:
sim.add_process(process)
sim.add_testbench(process)


class TransactionConflictTestCircuit(Elaboratable):
Expand Down Expand Up @@ -132,14 +133,19 @@ def setup_method(self):
random.seed(42)

def make_process(
self, io: TestbenchIO, prob: float, src: Iterable[int], tgt: Callable[[int], None], chk: Callable[[int], None]
self,
io: TestbenchIO,
prob: float,
src: Iterable[int],
tgt: Callable[[int], None],
chk: Callable[[int], None],
):
def process():
async def process(sim):
for i in src:
while random.random() >= prob:
yield Tick()
await sim.tick()
tgt(i)
r = yield from io.call(data=i)
r = await io.call(sim, data=i)
chk(r["data"])

return process
Expand Down Expand Up @@ -193,9 +199,9 @@ def test_calls(self, name, prob1, prob2, probout):
self.m = TransactionConflictTestCircuit(self.__class__.scheduler)

with self.run_simulation(self.m, add_transaction_module=False) as sim:
sim.add_process(self.make_in1_process(prob1))
sim.add_process(self.make_in2_process(prob2))
sim.add_process(self.make_out_process(probout))
sim.add_testbench(self.make_in1_process(prob1))
sim.add_testbench(self.make_in2_process(prob2))
sim.add_testbench(self.make_out_process(probout))

assert not self.in_expected
assert not self.out1_expected
Expand All @@ -210,7 +216,7 @@ def __init__(self):
self.t2 = Signal()

@abstractmethod
def elaborate(self, platform) -> TModule:
def elaborate(self, platform) -> HasElaborate:
raise NotImplementedError


Expand Down Expand Up @@ -283,22 +289,22 @@ def setup_method(self):
def test_priorities(self, priority: Priority):
m = self.circuit(priority)

def process():
async def process(sim):
to_do = 5 * [(0, 1), (1, 0), (1, 1)]
random.shuffle(to_do)
for r1, r2 in to_do:
yield m.r1.eq(r1)
yield m.r2.eq(r2)
yield Settle()
assert (yield m.t1) != (yield m.t2)
sim.set(m.r1, r1)
sim.set(m.r2, r2)
_, t1, t2 = await sim.delay(1e-9).sample(m.t1, m.t2)
assert t1 != t2
if r1 == 1 and r2 == 1:
if priority == Priority.LEFT:
assert (yield m.t1)
assert t1
if priority == Priority.RIGHT:
assert (yield m.t2)
assert t2

with self.run_simulation(m) as sim:
sim.add_process(process)
sim.add_testbench(process)

@parameterized.expand([(Priority.UNDEFINED,), (Priority.LEFT,), (Priority.RIGHT,)])
def test_unsatisfiable(self, priority: Priority):
Expand Down Expand Up @@ -368,18 +374,18 @@ def setup_method(self):
def test_scheduling(self):
m = self.circuit()

def process():
async def process(sim):
to_do = 5 * [(0, 1), (1, 0), (1, 1)]
random.shuffle(to_do)
for r1, r2 in to_do:
yield m.r1.eq(r1)
yield m.r2.eq(r2)
yield Tick()
assert (yield m.t1) == r1
assert (yield m.t2) == r1 * r2
sim.set(m.r1, r1)
sim.set(m.r2, r2)
*_, t1, t2 = await sim.tick().sample(m.t1, m.t2)
assert t1 == r1
assert t2 == r1 * r2

with self.run_simulation(m) as sim:
sim.add_process(process)
sim.add_testbench(process)


class ScheduleBeforeTestCircuit(SchedulingTestCircuit):
Expand Down Expand Up @@ -414,18 +420,18 @@ def setup_method(self):
def test_schedule_before(self):
m = ScheduleBeforeTestCircuit()

def process():
async def process(sim):
to_do = 5 * [(0, 1), (1, 0), (1, 1)]
random.shuffle(to_do)
for r1, r2 in to_do:
yield m.r1.eq(r1)
yield m.r2.eq(r2)
yield Tick()
assert (yield m.t1) == r1
assert not (yield m.t2)
sim.set(m.r1, r1)
sim.set(m.r2, r2)
*_, t1, t2 = await sim.tick().sample(m.t1, m.t2)
assert t1 == r1
assert not t2

with self.run_simulation(m) as sim:
sim.add_process(process)
sim.add_testbench(process)


class SingleCallerTestCircuit(Elaboratable):
Expand Down
31 changes: 12 additions & 19 deletions test/lib/test_fifo.py
View file
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
from amaranth import *
from amaranth.sim import Settle, Tick

from transactron.lib import AdapterTrans, BasicFifo

from transactron.testing import TestCaseWithSimulator, TestbenchIO, data_layout
from transactron.testing import TestCaseWithSimulator, TestbenchIO, data_layout, TestbenchContext
from collections import deque
from parameterized import parameterized_class
import random
Expand Down Expand Up @@ -44,36 +43,30 @@ def test_randomized(self):

self.done = False

def source():
async def source(sim: TestbenchContext):
for _ in range(cycles):
if random.randint(0, 1):
yield # random delay
await self.random_wait_geom(sim, 0.5)

v = random.randint(0, (2**fifoc.fifo.width) - 1)
yield from fifoc.fifo_write.call(data=v)
expq.appendleft(v)
await fifoc.fifo_write.call(sim, data=v)

if random.random() < 0.005:
yield from fifoc.fifo_clear.call()
yield Settle()
await fifoc.fifo_clear.call(sim)
await sim.delay(1e-9)
expq.clear()

self.done = True

def target():
async def target(sim: TestbenchContext):
while not self.done or expq:
if random.randint(0, 1):
yield Tick()
await self.random_wait_geom(sim, 0.5)

yield from fifoc.fifo_read.call_init()
yield Tick()
v = await fifoc.fifo_read.call_try(sim)

v = yield from fifoc.fifo_read.call_result()
if v is not None:
assert v["data"] == expq.pop()

yield from fifoc.fifo_read.disable()
assert v.data == expq.pop()

with self.run_simulation(fifoc) as sim:
sim.add_process(source)
sim.add_process(target)
sim.add_testbench(source)
sim.add_testbench(target)
Loading

0 comments on commit 5c44ab8

Please sign in to comment.