Something went wrong. Try again.
OR-1 dataflow CPU sketch
Something went wrong. Try again.
28 kB · 917 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918"""Tests for ProcessingElement with frame-based redesign.
Verifies:- AC3.1: Frame configuration (frame_count, frame_slots, matchable_offsets)- AC3.2: Pipeline order for dyadic and monadic tokens- AC3.3: Dyadic matching using tag_store and presence bits- AC3.4: INHERIT output mode reads FrameDest from frame- AC3.5: CHANGE_TAG mode extracts destination from left operand- AC3.6: SINK mode writes result to frame slot- AC3.8: Frame allocation/deallocation via FrameControlToken- AC1.1-AC1.9: Original AC tests adapted to new model"""
import pytestimport simpyfrom hypothesis import given
from cm_inst import ( ArithOp, FrameDest, FrameOp, Instruction, LogicOp, MemOp, OutputStyle, Port, RoutingOp, TokenKind,)from emu.pe import ProcessingElementfrom emu.types import PEConfigfrom tests.conftest import dyad_tokenfrom tokens import DyadToken, MonadToken
def inject_and_run(env, pe, token): """Helper: inject token and run simulation.""" def _put(): yield pe.input_store.put(token) env.process(_put()) env.run()
def inject_two_and_run(env, pe, token1, token2): """Helper: inject two tokens and run simulation.""" def _put(): yield pe.input_store.put(token1) yield pe.input_store.put(token2) env.process(_put()) env.run()
class TestFrameConfiguration: """AC3.1: Frame configuration is applied correctly."""
def test_default_frame_config(self): """PE accepts default frame configuration.""" env = simpy.Environment() config = PEConfig() pe = ProcessingElement(env=env, pe_id=0, config=config)
assert pe.frame_count > 0 assert pe.frame_slots > 0 assert pe.matchable_offsets > 0
def test_custom_frame_config(self): """PE accepts custom frame configuration.""" env = simpy.Environment() config = PEConfig( frame_count=4, frame_slots=32, matchable_offsets=4, ) pe = ProcessingElement(env=env, pe_id=0, config=config)
assert pe.frame_count == 4 assert pe.frame_slots == 32 assert pe.matchable_offsets == 4
class TestMonadicBypass: """AC1.1: Monadic token bypasses matching store."""
def test_monad_immediate_execution(self): """AC1.1: Monadic token executes immediately without matching.""" env = simpy.Environment()
# PASS instruction (monadic safe, no operands needed) pass_inst = Instruction( opcode=RoutingOp.PASS, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, )
# Set up destination in frame slot dest = FrameDest( target_pe=1, offset=0, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, )
config = PEConfig( pe_id=0, iram={0: pass_inst}, initial_frames={0: {8: dest}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
# Inject monadic token token = MonadToken(target=0, offset=0, act_id=0, data=0x1234, inline=False) inject_and_run(env, pe, token)
# Should produce one output assert len(output_store.items) == 1 out = output_store.items[0] assert isinstance(out, MonadToken) assert out.data == 0x1234
class TestDyadicMatching: """AC1.2, AC1.3: Dyadic token matching store behavior."""
def test_first_dyadic_no_fire(self): """AC1.2: First dyadic token stores in matching, no output.""" env = simpy.Environment()
add_inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, )
dest = FrameDest( target_pe=1, offset=0, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, )
config = PEConfig( pe_id=0, iram={0: add_inst}, initial_frames={0: {8: dest}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
# First dyadic token token_l = DyadToken(target=0, offset=0, act_id=0, data=0x1111, port=Port.L)
inject_and_run(env, pe, token_l)
# No output from first token assert len(output_store.items) == 0 # Matching store should have the operand frame_id, _lane = pe.tag_store[0] assert pe.presence[frame_id][0][0] is True
def test_second_dyadic_fires_left_first(self): """AC1.3: Second dyadic token fires when partner found (L then R).""" env = simpy.Environment()
add_inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, )
dest = FrameDest( target_pe=1, offset=0, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, )
config = PEConfig( pe_id=0, iram={0: add_inst}, initial_frames={0: {8: dest}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
token_l = DyadToken(target=0, offset=0, act_id=0, data=0x1111, port=Port.L) token_r = DyadToken(target=0, offset=0, act_id=0, data=0x2222, port=Port.R)
inject_two_and_run(env, pe, token_l, token_r)
# Should produce one output: 0x1111 + 0x2222 = 0x3333 assert len(output_store.items) == 1 assert output_store.items[0].data == 0x3333
def test_second_dyadic_fires_right_first(self): """AC1.3: Second dyadic fires, operands ordered by port (R then L).""" env = simpy.Environment()
add_inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, )
dest = FrameDest( target_pe=1, offset=0, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, )
config = PEConfig( pe_id=0, iram={0: add_inst}, initial_frames={0: {8: dest}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
# Inject R then L (reversed order) token_r = DyadToken(target=0, offset=0, act_id=0, data=0x2222, port=Port.R) token_l = DyadToken(target=0, offset=0, act_id=0, data=0x1111, port=Port.L)
inject_two_and_run(env, pe, token_r, token_l)
# Should still compute correctly: ADD(0x1111, 0x2222) = 0x3333 assert len(output_store.items) == 1 assert output_store.items[0].data == 0x3333
class TestOutputFormatterSingleMode: """AC1.5: SINGLE mode emits one token to dest_l."""
def test_single_mode_one_output(self): """AC1.5: SINGLE mode emits exactly one token.""" env = simpy.Environment()
# ADD with only dest_l (SINGLE mode) add_inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, )
dest = FrameDest( target_pe=2, offset=1, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, )
config = PEConfig( pe_id=0, iram={0: add_inst}, initial_frames={0: {8: dest}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[2] = output_store
token_l = DyadToken(target=0, offset=0, act_id=0, data=0x0005, port=Port.L) token_r = DyadToken(target=0, offset=0, act_id=0, data=0x0003, port=Port.R)
inject_two_and_run(env, pe, token_l, token_r)
# Exactly one output assert len(output_store.items) == 1 assert output_store.items[0].data == 0x0008 # 5 + 3
class TestOutputFormatterDualMode: """AC1.6: DUAL mode emits two tokens with same data."""
def test_dual_mode_two_outputs(self): """AC1.6: DUAL mode emits two tokens with same data.""" env = simpy.Environment()
# ADD with both dest_l and dest_r (DUAL mode) add_inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=8, )
dest_l = FrameDest( target_pe=2, offset=1, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, ) dest_r = FrameDest( target_pe=3, offset=2, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, )
config = PEConfig( pe_id=0, iram={0: add_inst}, initial_frames={0: {8: dest_l, 9: dest_r}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_l = simpy.Store(env, capacity=10) output_r = simpy.Store(env, capacity=10) pe.route_table[2] = output_l pe.route_table[3] = output_r
token_l = DyadToken(target=0, offset=0, act_id=0, data=0x0010, port=Port.L) token_r = DyadToken(target=0, offset=0, act_id=0, data=0x0020, port=Port.R)
inject_two_and_run(env, pe, token_l, token_r)
# Two outputs with same data assert len(output_l.items) == 1 assert len(output_r.items) == 1 assert output_l.items[0].data == 0x0030 assert output_r.items[0].data == 0x0030
class TestOutputFormatterSwitchMode: """AC1.7: SWITCH mode routes data and trigger separately."""
def test_switch_mode_true_condition(self): """AC1.7: SWITCH with true condition sends data to dest_l, trigger to dest_r.""" env = simpy.Environment()
# SWEQ with both dests sweq_inst = Instruction( opcode=RoutingOp.SWEQ, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=8, )
dest_l = FrameDest( target_pe=2, offset=1, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) dest_r = FrameDest( target_pe=3, offset=2, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, )
config = PEConfig( pe_id=0, iram={0: sweq_inst}, initial_frames={0: {8: dest_l, 9: dest_r}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_l = simpy.Store(env, capacity=10) output_r = simpy.Store(env, capacity=10) pe.route_table[2] = output_l pe.route_table[3] = output_r
# Equal tokens: bool_out = True token_l = DyadToken(target=0, offset=0, act_id=0, data=0x1234, port=Port.L) token_r = DyadToken(target=0, offset=0, act_id=0, data=0x1234, port=Port.R)
inject_two_and_run(env, pe, token_l, token_r)
# Data to dest_l, trigger to dest_r assert len(output_l.items) == 1 assert len(output_r.items) == 1
data_token = output_l.items[0] assert isinstance(data_token, MonadToken) assert data_token.data == 0x1234
trigger = output_r.items[0] assert isinstance(trigger, MonadToken) assert trigger.data == 0 # Trigger has zero data
def test_switch_mode_false_condition(self): """AC1.7: SWITCH with false condition sends data to dest_r, trigger to dest_l.""" env = simpy.Environment()
sweq_inst = Instruction( opcode=RoutingOp.SWEQ, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=8, )
dest_l = FrameDest( target_pe=2, offset=1, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) dest_r = FrameDest( target_pe=3, offset=2, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, )
config = PEConfig( pe_id=0, iram={0: sweq_inst}, initial_frames={0: {8: dest_l, 9: dest_r}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_l = simpy.Store(env, capacity=10) output_r = simpy.Store(env, capacity=10) pe.route_table[2] = output_l pe.route_table[3] = output_r
# Unequal tokens: bool_out = False token_l = DyadToken(target=0, offset=0, act_id=0, data=0x1234, port=Port.L) token_r = DyadToken(target=0, offset=0, act_id=0, data=0x5678, port=Port.R)
inject_two_and_run(env, pe, token_l, token_r)
# Data to dest_r, trigger to dest_l assert len(output_l.items) == 1 assert len(output_r.items) == 1
trigger = output_l.items[0] assert isinstance(trigger, MonadToken) assert trigger.data == 0 # Trigger has zero data
data_token = output_r.items[0] assert isinstance(data_token, MonadToken) assert data_token.data == 0x1234
class TestOutputFormatterSuppressMode: """AC1.8: SUPPRESS mode emits no tokens."""
def test_suppress_free_frame_instruction(self): """AC1.8: FREE_FRAME instruction suppresses output.""" env = simpy.Environment()
# FREE_FRAME (monadic, SUPPRESS) free_inst = Instruction( opcode=RoutingOp.FREE_FRAME, output=OutputStyle.SINK, has_const=False, dest_count=0, wide=False, fref=8, )
config = PEConfig( pe_id=0, iram={0: free_inst}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
token = MonadToken(target=0, offset=0, act_id=0, data=0x4567, inline=False)
inject_and_run(env, pe, token)
# No output assert len(output_store.items) == 0
def test_suppress_gate_false(self): """AC1.8: GATE with false condition suppresses output.""" env = simpy.Environment()
gate_inst = Instruction( opcode=RoutingOp.GATE, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, )
dest = FrameDest( target_pe=1, offset=0, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, )
config = PEConfig( pe_id=0, iram={0: gate_inst}, initial_frames={0: {8: dest}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
# L=42, R=0 (false) token_l = DyadToken(target=0, offset=0, act_id=0, data=0x002A, port=Port.L) token_r = DyadToken(target=0, offset=0, act_id=0, data=0x0000, port=Port.R)
inject_two_and_run(env, pe, token_l, token_r)
# No output (suppressed by false condition) assert len(output_store.items) == 0
def test_gate_true_passes(self): """AC1.8: GATE with true condition passes output.""" env = simpy.Environment()
gate_inst = Instruction( opcode=RoutingOp.GATE, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, )
dest = FrameDest( target_pe=1, offset=0, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, )
config = PEConfig( pe_id=0, iram={0: gate_inst}, initial_frames={0: {8: dest}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
# L=42, R=1 (true) token_l = DyadToken(target=0, offset=0, act_id=0, data=0x002A, port=Port.L) token_r = DyadToken(target=0, offset=0, act_id=0, data=0x0001, port=Port.R)
inject_two_and_run(env, pe, token_l, token_r)
# One output (gate opened) assert len(output_store.items) == 1 assert output_store.items[0].data == 0x002A
class TestNonExistentOffset: """AC1.9: Non-existent IRAM offset doesn't crash."""
def test_missing_iram_offset_no_crash(self): """AC1.9: Token targeting non-existent offset doesn't crash.""" env = simpy.Environment()
config = PEConfig(pe_id=0, iram={})
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
token = MonadToken(target=0, offset=99, act_id=0, data=0xDEAD, inline=False)
inject_and_run(env, pe, token)
# Should not crash, no output (instruction doesn't exist) assert len(output_store.items) == 0
class TestMatchingStoreCleared: """Matching store is cleared after firing."""
@given(dyad_token(target=0, offset=5, act_id=1)) def test_matching_store_cleared_after_firing(self, token_l: DyadToken): """After token pair fires, matching store slot is reset.""" env = simpy.Environment()
add_inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, )
dest = FrameDest( target_pe=1, offset=5, act_id=1, port=Port.L, token_kind=TokenKind.DYADIC, )
config = PEConfig( pe_id=0, iram={5: add_inst}, initial_frames={0: {8: dest}}, initial_tag_store={1: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
# Create matching right token token_r = DyadToken( target=0, offset=token_l.offset, act_id=token_l.act_id, data=0x5555, port=Port.R, )
inject_two_and_run(env, pe, token_l, token_r)
# After firing, matching store should be clear frame_id, _lane = pe.tag_store[token_l.act_id] assert pe.presence[frame_id][token_l.offset % pe.matchable_offsets][0] is False
class TestOutputTokenCountMatchesMode: """Output token count matches output mode."""
@given(dyad_token(target=0, offset=0, act_id=0)) def test_suppress_mode_produces_zero_tokens(self, token_l: DyadToken): """SUPPRESS mode produces zero output tokens.""" env = simpy.Environment()
free_inst = Instruction( opcode=RoutingOp.FREE_FRAME, output=OutputStyle.SINK, has_const=False, dest_count=0, wide=False, fref=8, )
config = PEConfig( pe_id=0, iram={0: free_inst}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
token_r = DyadToken( target=0, offset=token_l.offset, act_id=token_l.act_id, data=0x2222, port=Port.R, )
inject_two_and_run(env, pe, token_l, token_r)
# SUPPRESS mode produces zero outputs assert len(output_store.items) == 0
@given(dyad_token(target=0, offset=0, act_id=0)) def test_single_mode_produces_one_token(self, token_l: DyadToken): """SINGLE mode produces one output token.""" env = simpy.Environment()
add_inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, )
dest = FrameDest( target_pe=1, offset=0, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, )
config = PEConfig( pe_id=0, iram={0: add_inst}, initial_frames={0: {8: dest}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
token_r = DyadToken( target=0, offset=token_l.offset, act_id=token_l.act_id, data=0x2222, port=Port.R, )
inject_two_and_run(env, pe, token_l, token_r)
# SINGLE mode produces exactly one output assert len(output_store.items) == 1
@given(dyad_token(target=0, offset=0, act_id=0)) def test_dual_mode_produces_two_tokens(self, token_l: DyadToken): """DUAL mode produces two output tokens (one per destination).""" env = simpy.Environment()
add_inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=8, )
dest_l = FrameDest( target_pe=1, offset=0, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, ) dest_r = FrameDest( target_pe=2, offset=1, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, )
config = PEConfig( pe_id=0, iram={0: add_inst}, initial_frames={0: {8: dest_l, 9: dest_r}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store_l = simpy.Store(env, capacity=10) output_store_r = simpy.Store(env, capacity=10) pe.route_table[1] = output_store_l pe.route_table[2] = output_store_r
token_r = DyadToken( target=0, offset=token_l.offset, act_id=token_l.act_id, data=0x2222, port=Port.R, )
inject_two_and_run(env, pe, token_l, token_r)
# DUAL mode produces two outputs assert len(output_store_l.items) == 1 assert len(output_store_r.items) == 1
@given(dyad_token(target=0, offset=0, act_id=0)) def test_switch_mode_produces_two_tokens(self, token_l: DyadToken): """SWITCH mode produces two output tokens (data + trigger).""" env = simpy.Environment()
sweq_inst = Instruction( opcode=RoutingOp.SWEQ, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=8, )
dest_l = FrameDest( target_pe=1, offset=0, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, ) dest_r = FrameDest( target_pe=2, offset=1, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, )
config = PEConfig( pe_id=0, iram={0: sweq_inst}, initial_frames={0: {8: dest_l, 9: dest_r}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store_l = simpy.Store(env, capacity=10) output_store_r = simpy.Store(env, capacity=10) pe.route_table[1] = output_store_l pe.route_table[2] = output_store_r
# Use same value to trigger bool_out=True token_r = DyadToken( target=0, offset=token_l.offset, act_id=token_l.act_id, data=token_l.data, port=Port.R, )
inject_two_and_run(env, pe, token_l, token_r)
# SWITCH mode produces two outputs assert len(output_store_l.items) == 1 assert len(output_store_r.items) == 1
class TestStaleTokensProduceNoOutput: """Stale tokens (act_id mismatch) produce no output."""
@given(dyad_token(target=0, offset=0, act_id=0)) def test_stale_token_no_output(self, token_l: DyadToken): """Token with invalid act_id is rejected, produces no output.""" env = simpy.Environment()
add_inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, )
dest = FrameDest( target_pe=1, offset=0, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, )
config = PEConfig( pe_id=0, iram={0: add_inst}, initial_frames={0: {8: dest}}, initial_tag_store={0: (0, 0)}, # Note: only act_id 0 is allocated; other act_ids will be invalid )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
# Use act_id that was NOT allocated invalid_act_id = 3 token = DyadToken( target=0, offset=0, act_id=invalid_act_id, data=0x1234, port=Port.L, )
inject_and_run(env, pe, token)
# Invalid act_id produces no output assert len(output_store.items) == 0
class TestBoundaryEdgeCases: """Boundary edge cases."""
def test_iram_write_multiple_instructions(self): """Multiple instructions can be loaded and executed.""" env = simpy.Environment()
add_inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, )
inc_inst = Instruction( opcode=ArithOp.INC, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, )
dest = FrameDest( target_pe=1, offset=0, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, )
config = PEConfig( pe_id=0, iram={0: add_inst, 1: inc_inst}, initial_frames={0: {8: dest}}, initial_tag_store={0: (0, 0)}, )
pe = ProcessingElement(env=env, pe_id=0, config=config) output_store = simpy.Store(env, capacity=10) pe.route_table[1] = output_store
# Execute ADD at offset 0 token_l = DyadToken(target=0, offset=0, act_id=0, data=0x10, port=Port.L) token_r = DyadToken(target=0, offset=0, act_id=0, data=0x20, port=Port.R)
inject_two_and_run(env, pe, token_l, token_r)
# Execute INC at offset 1 token_inc = MonadToken(target=0, offset=1, act_id=0, data=0x10, inline=False)
inject_and_run(env, pe, token_inc)
# Both instructions executed assert len(output_store.items) == 2 assert output_store.items[0].data == 0x30 # ADD result assert output_store.items[1].data == 0x11 # INC result