Something went wrong. Try again.
OR-1 dataflow CPU sketch
Something went wrong. Try again.
30 kB · 872 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873"""Tests for Structure Memory operations.
Verifies all acceptance criteria:- or1-emu.AC3.1: READ on FULL cell returns data immediately via result token- or1-emu.AC3.2: READ on EMPTY cell with empty deferred register stashes return route, sets WAITING- or1-emu.AC3.3: WRITE on WAITING cell satisfies deferred read — emits result token to stashed return route- or1-emu.AC3.4: WRITE on EMPTY/RESERVED sets cell to FULL- or1-emu.AC3.5: CLEAR sets cell to EMPTY, cancels deferred read if targeting that cell- or1-emu.AC3.6: READ_INC/READ_DEC atomically modify and return value (lower 256 cells only)- or1-emu.AC3.7: Depth-1 constraint: second blocking READ on different empty cell stalls until first deferred read is satisfied- or1-emu.AC3.8: WRITE on FULL cell overwrites data (diagnostic flag set if modelled)- or1-emu.AC3.9: CAS on FULL cell: if current value == expected (SMToken.flags), writes new value (SMToken.data) and returns old value; if mismatch, cell unchanged and returns old value (lower 256 cells only)"""
import simpyfrom hypothesis import given, strategies as st
from cm_inst import MemOpfrom emu.sm import StructureMemoryfrom sm_mod import Presencefrom tests.conftest import sm_token, sm_return_route, uint16from tokens import CMToken, SMToken
def inject_token(env: simpy.Environment, store: simpy.Store, token): """Helper to inject token into store via a process.""" def _injector(): yield store.put(token)
env.process(_injector())
class TestAC3_1ReadOnFull: """AC3.1: READ on FULL cell returns data immediately via result token."""
def test_read_on_full_cell(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Pre-populate cell 10 to FULL with data 0xBEEF sm.cells[10].pres = Presence.FULL sm.cells[10].data_l = 0xBEEF
# Create collector store for results collector = simpy.Store(env) sm.route_table[0] = collector
# Inject READ token targeting cell 10 ret_route = CMToken(target=0, offset=5, act_id=1, data=0) read_token = SMToken(target=0, addr=10, op=MemOp.READ, flags=None, data=None, ret=ret_route) inject_token(env, sm.input_store, read_token)
# Run simulation env.run(until=100)
# Verify result token in collector with correct data assert len(collector.items) == 1 result = collector.items[0] assert result.data == 0xBEEF assert result.offset == 5 assert result.act_id == 1
class TestAC3_2ReadOnEmpty: """AC3.2: READ on EMPTY cell with empty deferred register stashes return route, sets WAITING."""
def test_read_on_empty_cell(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Cell 20 starts EMPTY assert sm.cells[20].pres == Presence.EMPTY
# Create collector store collector = simpy.Store(env) sm.route_table[0] = collector
# Inject READ token targeting cell 20 ret_route = CMToken(target=0, offset=7, act_id=2, data=0) read_token = SMToken(target=0, addr=20, op=MemOp.READ, flags=None, data=None, ret=ret_route) inject_token(env, sm.input_store, read_token)
# Run simulation env.run(until=100)
# Verify cell is now WAITING assert sm.cells[20].pres == Presence.WAITING
# Verify deferred_read is set assert sm.deferred_read is not None assert sm.deferred_read.cell_addr == 20 assert sm.deferred_read.return_route == ret_route
# Verify no result token emitted yet assert len(collector.items) == 0
class TestAC3_3DeferredReadSatisfaction: """AC3.3: WRITE on WAITING cell satisfies deferred read — emits result token to stashed return route."""
def test_write_satisfies_deferred_read(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Create collector store collector = simpy.Store(env) sm.route_table[0] = collector
# Set up deferred read on cell 30 ret_route = CMToken(target=0, offset=8, act_id=3, data=0) read_token = SMToken(target=0, addr=30, op=MemOp.READ, flags=None, data=None, ret=ret_route) inject_token(env, sm.input_store, read_token)
# Run to let deferred read be set up env.run(until=10)
# Now inject WRITE to cell 30 write_token = SMToken(target=0, addr=30, op=MemOp.WRITE, flags=None, data=0xDEAD, ret=None) inject_token(env, sm.input_store, write_token)
# Continue simulation env.run(until=100)
# Verify cell is now FULL assert sm.cells[30].pres == Presence.FULL assert sm.cells[30].data_l == 0xDEAD
# Verify result token was emitted with written data assert len(collector.items) == 1 result = collector.items[0] assert result.data == 0xDEAD assert result.offset == 8 assert result.act_id == 3
# Verify deferred_read is cleared assert sm.deferred_read is None
class TestAC3_4WriteOnEmptyOrReserved: """AC3.4: WRITE on EMPTY/RESERVED sets cell to FULL."""
def test_write_on_empty_cell(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Cell 40 starts EMPTY assert sm.cells[40].pres == Presence.EMPTY
# Inject WRITE write_token = SMToken(target=0, addr=40, op=MemOp.WRITE, flags=None, data=0xCAFE, ret=None) inject_token(env, sm.input_store, write_token)
env.run(until=100)
# Verify cell is FULL with correct data assert sm.cells[40].pres == Presence.FULL assert sm.cells[40].data_l == 0xCAFE
def test_write_on_reserved_cell(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Set cell 50 to RESERVED sm.cells[50].pres = Presence.RESERVED
# Inject WRITE write_token = SMToken(target=0, addr=50, op=MemOp.WRITE, flags=None, data=0xF00D, ret=None) inject_token(env, sm.input_store, write_token)
env.run(until=100)
# Verify cell is FULL with correct data assert sm.cells[50].pres == Presence.FULL assert sm.cells[50].data_l == 0xF00D
class TestAC3_5Clear: """AC3.5: CLEAR sets cell to EMPTY, cancels deferred read if targeting that cell."""
def test_clear_sets_empty(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Set cell 60 to FULL sm.cells[60].pres = Presence.FULL sm.cells[60].data_l = 0x1234
# Inject CLEAR clear_token = SMToken(target=0, addr=60, op=MemOp.CLEAR, flags=None, data=None, ret=None) inject_token(env, sm.input_store, clear_token)
env.run(until=100)
# Verify cell is EMPTY assert sm.cells[60].pres == Presence.EMPTY assert sm.cells[60].data_l is None
def test_clear_cancels_deferred_read(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Set up deferred read on cell 70 ret_route = CMToken(target=0, offset=10, act_id=0, data=0) read_token = SMToken(target=0, addr=70, op=MemOp.READ, flags=None, data=None, ret=ret_route) inject_token(env, sm.input_store, read_token)
env.run(until=10)
# Now inject CLEAR on cell 70 clear_token = SMToken(target=0, addr=70, op=MemOp.CLEAR, flags=None, data=None, ret=None) inject_token(env, sm.input_store, clear_token)
env.run(until=100)
# Verify cell is EMPTY assert sm.cells[70].pres == Presence.EMPTY
# Verify deferred_read is cleared assert sm.deferred_read is None
# Verify no result token was emitted (deferred read was cancelled) assert len(collector.items) == 0
class TestAC3_8WriteOnFull: """AC3.8: WRITE on FULL cell overwrites data."""
def test_write_overwrites_full_cell(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Set cell 80 to FULL with data X sm.cells[80].pres = Presence.FULL sm.cells[80].data_l = 0x5555
# Inject WRITE with data Y write_token = SMToken(target=0, addr=80, op=MemOp.WRITE, flags=None, data=0xAAAA, ret=None) inject_token(env, sm.input_store, write_token)
env.run(until=100)
# Verify cell still FULL but data is overwritten assert sm.cells[80].pres == Presence.FULL assert sm.cells[80].data_l == 0xAAAA
class TestAC3_6AtomicOps: """AC3.6: READ_INC/READ_DEC atomically modify and return value (lower 256 cells only)."""
def test_read_inc_returns_old_value(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Set cell 100 to FULL with value 42 sm.cells[100].pres = Presence.FULL sm.cells[100].data_l = 42
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Inject RD_INC ret_route = CMToken(target=0, offset=11, act_id=0, data=0) inc_token = SMToken(target=0, addr=100, op=MemOp.RD_INC, flags=None, data=None, ret=ret_route) inject_token(env, sm.input_store, inc_token)
env.run(until=100)
# Verify cell was incremented assert sm.cells[100].data_l == 43
# Verify result token has old value assert len(collector.items) == 1 assert collector.items[0].data == 42
def test_read_inc_wraps_at_0xFFFF(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Set cell 110 to FULL with max value sm.cells[110].pres = Presence.FULL sm.cells[110].data_l = 0xFFFF
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Inject RD_INC ret_route = CMToken(target=0, offset=12, act_id=0, data=0) inc_token = SMToken(target=0, addr=110, op=MemOp.RD_INC, flags=None, data=None, ret=ret_route) inject_token(env, sm.input_store, inc_token)
env.run(until=100)
# Verify cell wrapped to 0 assert sm.cells[110].data_l == 0
# Verify result token has old value assert len(collector.items) == 1 assert collector.items[0].data == 0xFFFF
def test_read_dec_returns_old_value(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Set cell 120 to FULL with value 100 sm.cells[120].pres = Presence.FULL sm.cells[120].data_l = 100
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Inject RD_DEC ret_route = CMToken(target=0, offset=13, act_id=0, data=0) dec_token = SMToken(target=0, addr=120, op=MemOp.RD_DEC, flags=None, data=None, ret=ret_route) inject_token(env, sm.input_store, dec_token)
env.run(until=100)
# Verify cell was decremented assert sm.cells[120].data_l == 99
# Verify result token has old value assert len(collector.items) == 1 assert collector.items[0].data == 100
def test_atomic_op_rejected_on_non_full_cell(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Cell 130 starts EMPTY assert sm.cells[130].pres == Presence.EMPTY
# Inject RD_INC ret_route = CMToken(target=0, offset=14, act_id=0, data=0) inc_token = SMToken(target=0, addr=130, op=MemOp.RD_INC, flags=None, data=None, ret=ret_route) inject_token(env, sm.input_store, inc_token)
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
env.run(until=100)
# Verify no result token emitted and cell still EMPTY assert sm.cells[130].pres == Presence.EMPTY assert len(collector.items) == 0
def test_atomic_op_rejected_on_addr_gte_256(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Set cell 256 to FULL sm.cells[256].pres = Presence.FULL sm.cells[256].data_l = 50
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Inject RD_INC on cell 256 (should be rejected) ret_route = CMToken(target=0, offset=15, act_id=0, data=0) inc_token = SMToken(target=0, addr=256, op=MemOp.RD_INC, flags=None, data=None, ret=ret_route) inject_token(env, sm.input_store, inc_token)
env.run(until=100)
# Verify no result token emitted and cell unchanged assert sm.cells[256].data_l == 50 assert len(collector.items) == 0
class TestAC3_7DepthOneConstraint: """AC3.7: Depth-1 constraint - SM enforces only one outstanding deferred read."""
def test_two_blocking_reads_stall_and_unblock(self): """ AC3.7: Inject READ A then READ B while deferred register is occupied. SM stalls on second READ. WRITE A satisfies first, SM unblocks and retries READ B. Then WRITE B satisfies second. Both results arrive. """ env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
collector = simpy.Store(env) sm.route_table[0] = collector
ret_a = CMToken(target=0, offset=20, act_id=0, data=0) ret_b = CMToken(target=0, offset=21, act_id=1, data=0) read_a = SMToken(target=0, addr=140, op=MemOp.READ, flags=None, data=None, ret=ret_a) read_b = SMToken(target=0, addr=150, op=MemOp.READ, flags=None, data=None, ret=ret_b)
# Inject both READs before any WRITE inject_token(env, sm.input_store, read_a) inject_token(env, sm.input_store, read_b) env.run(until=10)
# Cell A is WAITING, deferred register holds A's route assert sm.cells[140].pres == Presence.WAITING assert sm.deferred_read is not None assert sm.deferred_read.cell_addr == 140 # Cell B still EMPTY — SM stalled on second READ assert sm.cells[150].pres == Presence.EMPTY
# Satisfy first deferred by writing to A write_a = SMToken(target=0, addr=140, op=MemOp.WRITE, flags=None, data=0x1111, ret=None) inject_token(env, sm.input_store, write_a) env.run(until=50)
# SM unblocked, retried READ B → cell B now WAITING assert sm.cells[150].pres == Presence.WAITING assert sm.deferred_read is not None assert sm.deferred_read.cell_addr == 150
# Satisfy second deferred write_b = SMToken(target=0, addr=150, op=MemOp.WRITE, flags=None, data=0x2222, ret=None) inject_token(env, sm.input_store, write_b) env.run(until=200)
# Both results collected assert len(collector.items) == 2 assert collector.items[0].data == 0x1111 assert collector.items[1].data == 0x2222 assert sm.cells[150].pres == Presence.FULL assert sm.cells[150].data_l == 0x2222
class TestAC3_9CAS: """AC3.9: CAS on FULL cell with compare-and-swap semantics."""
def test_cas_match_swaps_value(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Set cell 160 to FULL with value 10 sm.cells[160].pres = Presence.FULL sm.cells[160].data_l = 10
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Inject CMP_SW with flags=10 (expected), data=99 (new) ret_route = CMToken(target=0, offset=22, act_id=0, data=0) cas_token = SMToken(target=0, addr=160, op=MemOp.CMP_SW, flags=10, data=99, ret=ret_route) inject_token(env, sm.input_store, cas_token)
env.run(until=100)
# Verify cell now has new value assert sm.cells[160].data_l == 99
# Verify result token has old value assert len(collector.items) == 1 assert collector.items[0].data == 10
def test_cas_mismatch_no_swap(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Set cell 170 to FULL with value 10 sm.cells[170].pres = Presence.FULL sm.cells[170].data_l = 10
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Inject CMP_SW with flags=20 (mismatch), data=99 (new) ret_route = CMToken(target=0, offset=23, act_id=0, data=0) cas_token = SMToken(target=0, addr=170, op=MemOp.CMP_SW, flags=20, data=99, ret=ret_route) inject_token(env, sm.input_store, cas_token)
env.run(until=100)
# Verify cell unchanged assert sm.cells[170].data_l == 10
# Verify result token has old value assert len(collector.items) == 1 assert collector.items[0].data == 10
def test_cas_rejected_on_addr_gte_256(self): env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Set cell 256 to FULL sm.cells[256].pres = Presence.FULL sm.cells[256].data_l = 10
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Inject CMP_SW on cell 256 (should be rejected) ret_route = CMToken(target=0, offset=24, act_id=0, data=0) cas_token = SMToken(target=0, addr=256, op=MemOp.CMP_SW, flags=10, data=99, ret=ret_route) inject_token(env, sm.input_store, cas_token)
env.run(until=100)
# Verify no result token and cell unchanged assert sm.cells[256].data_l == 10 assert len(collector.items) == 0
class TestPresenceStateMachineInvariant: """Property-based test: presence state machine invariant."""
@given(sm_token()) def test_valid_presence_state_after_operations(self, token): """Verify that after any valid operation, cell is in a valid Presence state.""" env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Create collector for results collector = simpy.Store(env) sm.route_table[0] = collector
# Ensure return route is valid if token.ret is None: token = SMToken( target=token.target, addr=token.addr, op=token.op, flags=token.flags, data=token.data, ret=CMToken(target=0, offset=0, act_id=0, data=0), )
# Pre-populate target cell as FULL if it's an atomic operation if token.op in (MemOp.RD_INC, MemOp.RD_DEC, MemOp.CMP_SW): sm.cells[token.addr].pres = Presence.FULL sm.cells[token.addr].data_l = 0x1234
inject_token(env, sm.input_store, token)
# Run simulation env.run(until=100)
# Verify cell is in valid state and specific transitions per op type cell = sm.cells[token.addr] assert cell.pres in (Presence.EMPTY, Presence.RESERVED, Presence.FULL, Presence.WAITING)
# Verify specific state transitions per operation type if token.op == MemOp.WRITE: # WRITE should set cell to FULL assert cell.pres == Presence.FULL assert cell.data_l == token.data elif token.op == MemOp.READ: # READ on empty cell should set to WAITING if cell.pres == Presence.WAITING: assert sm.deferred_read is not None else: assert cell.pres in (Presence.EMPTY, Presence.FULL) elif token.op == MemOp.CLEAR: # CLEAR should set to EMPTY assert cell.pres == Presence.EMPTY elif token.op == MemOp.ALLOC: # ALLOC on EMPTY should set to RESERVED assert cell.pres == Presence.RESERVED elif token.op in (MemOp.RD_INC, MemOp.RD_DEC): # Atomic ops on FULL should stay FULL with modified data assert cell.pres == Presence.FULL elif token.op == MemOp.CMP_SW: # CMP_SW on FULL should stay FULL (data may change) assert cell.pres == Presence.FULL
class TestWriteAlwaysSetsDataL: """Property-based test: WRITE always sets data_l to the written value."""
@given(sm_token(op=MemOp.WRITE)) def test_write_sets_data_l(self, token): """Verify WRITE operations always set data_l to the written value.""" env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Inject token inject_token(env, sm.input_store, token)
# Run simulation env.run(until=100)
# Verify data_l is set to token.data assert sm.cells[token.addr].data_l == token.data assert sm.cells[token.addr].pres == Presence.FULL
class TestWriteReadRoundtrip: """Property-based test: WRITE→READ roundtrip always returns written value."""
@given(sm_token(op=MemOp.WRITE), sm_return_route()) def test_write_read_roundtrip(self, write_token, read_route): """WRITE→READ roundtrip always returns written value.""" env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Create collector for results collector = simpy.Store(env) sm.route_table[0] = collector
# Inject WRITE inject_token(env, sm.input_store, write_token) env.run(until=10)
# Now READ from same cell with return route read_token = SMToken( target=write_token.target, addr=write_token.addr, op=MemOp.READ, flags=None, data=None, ret=read_route, ) inject_token(env, sm.input_store, read_token) env.run(until=100)
# Verify result token has written data assert len(collector.items) == 1 result = collector.items[0] assert result.data == write_token.data
class TestClearAlwaysEmptifies: """Property-based test: CLEAR on any state produces EMPTY."""
@given(sm_token(op=MemOp.CLEAR)) def test_clear_on_any_state(self, token): """CLEAR on any state produces EMPTY.""" env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Pre-set cell to a random state (FULL with data) sm.cells[token.addr].pres = Presence.FULL sm.cells[token.addr].data_l = 0xBEEF
# Inject CLEAR inject_token(env, sm.input_store, token) env.run(until=100)
# Verify cell is EMPTY assert sm.cells[token.addr].pres == Presence.EMPTY assert sm.cells[token.addr].data_l is None
class TestRDIncDecRestores: """Property-based test: RD_INC then RD_DEC restores original value (mod 16-bit)."""
@given(uint16, sm_return_route(), sm_return_route()) def test_rd_inc_dec_restores(self, original_value, return_route_inc, return_route_dec): """RD_INC then RD_DEC restores original value.""" env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Create collector for results collector = simpy.Store(env) sm.route_table[0] = collector
# Set cell to original value sm.cells[100].pres = Presence.FULL sm.cells[100].data_l = original_value
# Inject RD_INC inc_token = SMToken( target=0, addr=100, op=MemOp.RD_INC, flags=None, data=None, ret=return_route_inc, ) inject_token(env, sm.input_store, inc_token) env.run(until=10)
# Cell now has (original_value + 1) & 0xFFFF # Inject RD_DEC dec_token = SMToken( target=0, addr=100, op=MemOp.RD_DEC, flags=None, data=None, ret=return_route_dec, ) inject_token(env, sm.input_store, dec_token) env.run(until=100)
# Cell should be back to original_value assert sm.cells[100].data_l == original_value
class TestAtomicOpsOnNonFullRejected: """Property-based test: Atomic ops on non-FULL cells are rejected."""
@given(sm_token(op=MemOp.RD_INC), sm_return_route()) def test_rd_inc_on_non_full_rejected(self, token, return_route): """RD_INC on non-FULL cell is rejected.""" env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Ensure cell is EMPTY (not FULL) assert sm.cells[token.addr].pres == Presence.EMPTY
# Create RD_INC token with return route inc_token = SMToken( target=token.target, addr=token.addr, op=MemOp.RD_INC, flags=None, data=None, ret=return_route, )
inject_token(env, sm.input_store, inc_token) env.run(until=100)
# Verify no result token and cell still EMPTY assert len(collector.items) == 0 assert sm.cells[token.addr].pres == Presence.EMPTY
@given(sm_token(op=MemOp.RD_DEC), sm_return_route()) def test_rd_dec_on_non_full_rejected(self, token, return_route): """RD_DEC on non-FULL cell is rejected.""" env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Ensure cell is EMPTY assert sm.cells[token.addr].pres == Presence.EMPTY
# Create RD_DEC token dec_token = SMToken( target=token.target, addr=token.addr, op=MemOp.RD_DEC, flags=None, data=None, ret=return_route, )
inject_token(env, sm.input_store, dec_token) env.run(until=100)
# Verify no result token and cell still EMPTY assert len(collector.items) == 0 assert sm.cells[token.addr].pres == Presence.EMPTY
@given(sm_token(op=MemOp.CMP_SW), sm_return_route()) def test_cmp_sw_on_non_full_rejected(self, token, return_route): """CMP_SW on non-FULL cell is rejected.""" env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Ensure cell is EMPTY assert sm.cells[token.addr].pres == Presence.EMPTY
# Create CMP_SW token cas_token = SMToken( target=token.target, addr=token.addr, op=MemOp.CMP_SW, flags=0x1234, data=0x5678, ret=return_route, )
inject_token(env, sm.input_store, cas_token) env.run(until=100)
# Verify no result token and cell still EMPTY assert len(collector.items) == 0 assert sm.cells[token.addr].pres == Presence.EMPTY
class TestBoundaryEdgeCases: """Test boundary and edge cases for SM operations."""
def test_rd_dec_wrapping_0x0000_to_0xffff(self): """RD_DEC on 0x0000 wraps to 0xFFFF.""" env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Set cell 150 to FULL with value 0 sm.cells[150].pres = Presence.FULL sm.cells[150].data_l = 0x0000
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Inject RD_DEC ret_route = CMToken(target=0, offset=16, act_id=0, data=0) dec_token = SMToken(target=0, addr=150, op=MemOp.RD_DEC, flags=None, data=None, ret=ret_route) inject_token(env, sm.input_store, dec_token)
env.run(until=100)
# Verify cell wrapped to 0xFFFF assert sm.cells[150].data_l == 0xFFFF
# Verify result token has old value (0) assert len(collector.items) == 1 assert collector.items[0].data == 0x0000
def test_alloc_empty_to_reserved(self): """ALLOC changes cell from EMPTY to RESERVED.""" env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Cell 200 starts EMPTY assert sm.cells[200].pres == Presence.EMPTY
# Inject ALLOC alloc_token = SMToken(target=0, addr=200, op=MemOp.ALLOC, flags=None, data=None, ret=None) inject_token(env, sm.input_store, alloc_token)
env.run(until=100)
# Verify cell is now RESERVED assert sm.cells[200].pres == Presence.RESERVED
def test_cmp_sw_on_non_full_cell_rejected(self): """CMP_SW on non-FULL cell (EMPTY state) is rejected.""" env = simpy.Environment() sm = StructureMemory(env, 0, cell_count=512)
# Create collector collector = simpy.Store(env) sm.route_table[0] = collector
# Cell 210 is EMPTY assert sm.cells[210].pres == Presence.EMPTY
# Inject CMP_SW on EMPTY cell ret_route = CMToken(target=0, offset=17, act_id=0, data=0) cas_token = SMToken( target=0, addr=210, op=MemOp.CMP_SW, flags=0x1111, data=0x2222, ret=ret_route ) inject_token(env, sm.input_store, cas_token)
env.run(until=100)
# Verify no result token and cell still EMPTY assert len(collector.items) == 0 assert sm.cells[210].pres == Presence.EMPTY
# Task 3: T0/T1 Tier Split and EXEC Opcode Tests