""" Frame-based PE rewrite tests. Verifies pe-frame-redesign.AC3 and pe-frame-redesign.AC1.6: - AC3.1: Frame count/slots/matchable_offsets are configurable - AC3.2: Pipeline order is IFETCH → act_id resolution → MATCH/FRAME → EXECUTE → EMIT - AC3.3: Dyadic matching uses tag_store + presence bits + frame SRAM - AC3.4: INHERIT output reads FrameDest from frame and routes token - AC3.5: CHANGE_TAG unpacks left operand (flit 1) to get destination - AC3.6: SINK writes result to frame slot, emits no token - AC3.7: EXTRACT_TAG produces packed flit 1 with PE/offset/act_id/port/kind - AC3.8: ALLOC/FREE frame control, FREE_FRAME opcode, ALLOC_REMOTE remote allocation - AC3.9: PELocalWriteToken with is_dest=True decodes FrameDest - AC3.10: Pipeline timing: 5 cycles dyadic, 4 cycles monadic, 2 cycles side paths - AC1.6: Invalid act_id emits TokenRejected, doesn't crash """ import pytest import simpy from cm_inst import ( ArithOp, FrameDest, FrameOp, Instruction, LogicOp, MemOp, OutputStyle, Port, RoutingOp, TokenKind, ) from encoding import pack_flit1, unpack_flit1, pack_instruction, unpack_instruction from emu.events import ( Emitted, Executed, FrameAllocated, FrameFreed, FrameSlotWritten, Matched, TokenReceived, TokenRejected, ) from emu.pe import ProcessingElement from emu.types import PEConfig from tokens import ( DyadToken, FrameControlToken, MonadToken, PELocalWriteToken, SMToken, ) 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() class TestFrameConfiguration: """AC3.1: PE constructor accepts frame_count, frame_slots, matchable_offsets.""" def test_constructor_default_params(self): env = simpy.Environment() pe = ProcessingElement( env=env, pe_id=0, config=PEConfig(), ) # Default config should have frame_count, frame_slots, matchable_offsets assert pe.frame_count > 0 assert pe.frame_slots > 0 assert pe.matchable_offsets > 0 def test_constructor_custom_params(self): env = simpy.Environment() config = PEConfig( frame_count=4, frame_slots=32, matchable_offsets=4, ) pe = ProcessingElement( env=env, pe_id=1, config=config, ) assert pe.frame_count == 4 assert pe.frame_slots == 32 assert pe.matchable_offsets == 4 class TestFrameAllocationAndFree: """AC3.8: Frame allocation (ALLOC) and deallocation (FREE) via FrameControlToken.""" def test_alloc_frame_control_token(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Inject FrameControlToken(ALLOC) for act_id=0 fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) # Should have TokenReceived and FrameAllocated events token_received = [e for e in events if isinstance(e, TokenReceived)] frame_allocated = [e for e in events if isinstance(e, FrameAllocated)] assert len(token_received) > 0 assert len(frame_allocated) > 0 assert pe.tag_store[0][0] in range(pe.frame_count) assert frame_allocated[0].lane == 0 def test_free_frame_control_token(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate first fct_alloc = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct_alloc) frame_id, _lane = pe.tag_store[0] # Now deallocate fct_free = FrameControlToken(target=0, act_id=0, op=FrameOp.FREE, payload=0) inject_and_run(env, pe, fct_free) # Should have FrameFreed event and tag_store should be cleared frame_freed = [e for e in events if isinstance(e, FrameFreed)] assert len(frame_freed) > 0 assert frame_freed[0].lane == 0 assert frame_freed[0].frame_freed == True assert 0 not in pe.tag_store assert frame_id in pe.free_frames class TestDyadicMatching: """AC3.3: Dyadic matching uses tag_store + presence bits + frame SRAM.""" def test_dyadic_token_pair_matching(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, matchable_offsets=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Set up: allocate frame for act_id=0 fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) # Set up: install dyadic instruction at offset 0 # Mode SINK: no output emission, just execution and matching verification inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.SINK, has_const=False, dest_count=0, wide=False, fref=0, ) pe.iram[0] = inst # Set up: write destination FrameDest to frame slot 0 dest = FrameDest( target_pe=0, offset=1, act_id=0, port=Port.L, token_kind=TokenKind.DYADIC, ) pe.frames[pe.tag_store[0][0]][0] = dest # Inject first dyadic token (port=L, data=5) tok1 = DyadToken( target=0, offset=0, act_id=0, data=5, port=Port.L, ) inject_and_run(env, pe, tok1) # Should have TokenReceived, no match yet (waiting for partner) token_received = [e for e in events if isinstance(e, TokenReceived)] matched = [e for e in events if isinstance(e, Matched)] assert len(token_received) >= 1 assert len(matched) == 0 # Inject second dyadic token (port=R, data=3) tok2 = DyadToken( target=0, offset=0, act_id=0, data=3, port=Port.R, ) inject_and_run(env, pe, tok2) # Should now have Matched event matched = [e for e in events if isinstance(e, Matched)] assert len(matched) > 0 m = matched[0] assert m.left == 5 assert m.right == 3 class TestInheritOutput: """AC3.4: INHERIT output reads FrameDest from frame and constructs token.""" def test_inherit_single_dest(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[0] # Set up instruction: mode 0 (no const, dest_count=1), fref=8 inst = Instruction( opcode=ArithOp.INC, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, ) pe.iram[2] = inst # Set destination at frame[8] dest = FrameDest( target_pe=0, offset=5, act_id=1, port=Port.L, token_kind=TokenKind.MONADIC, ) pe.frames[frame_id][8] = dest # Wire route table pe.route_table[0] = simpy.Store(env) # Inject monadic token tok = MonadToken( target=0, offset=2, act_id=0, data=10, inline=False, ) inject_and_run(env, pe, tok) # Should have Emitted event with output token routed to target_pe=0, offset=5, act_id=1 emitted = [e for e in events if isinstance(e, Emitted)] assert len(emitted) > 0 out_token = emitted[0].token assert out_token.target == 0 assert out_token.offset == 5 assert out_token.act_id == 1 class TestChangeTagOutput: """AC3.5: CHANGE_TAG unpacks left operand (flit 1) to get destination.""" def test_change_tag_output(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) # Set up instruction: CHANGE_TAG output, mode 4 (no const, dest_count=1) inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.CHANGE_TAG, has_const=False, dest_count=1, wide=False, fref=0, ) pe.iram[3] = inst # Wire route table pe.route_table[1] = simpy.Store(env) # Construct a packed flit 1 for destination (pe=1, offset=7, act_id=2, port=R, kind=DYADIC) dest = FrameDest( target_pe=1, offset=7, act_id=2, port=Port.R, token_kind=TokenKind.DYADIC, ) flit1 = pack_flit1(dest) # Inject dyadic token pair: first (L) carries the flit1 as left operand tok_l = DyadToken( target=0, offset=3, act_id=0, data=flit1, # packed flit 1 port=Port.L, ) inject_and_run(env, pe, tok_l) # Inject second (R) with some data value tok_r = DyadToken( target=0, offset=3, act_id=0, data=100, port=Port.R, ) inject_and_run(env, pe, tok_r) # Should emit token with target=1, offset=7, act_id=2 emitted = [e for e in events if isinstance(e, Emitted)] assert len(emitted) > 0 out_token = emitted[-1].token # Last emitted token assert out_token.target == 1 assert out_token.offset == 7 assert out_token.act_id == 2 class TestSinkOutput: """AC3.6: SINK output writes result to frame slot, emits no token.""" def test_sink_writes_to_frame(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[0] # Set up instruction: SINK output, mode 6 (no const, dest_count=0), fref=10 inst = Instruction( opcode=ArithOp.INC, output=OutputStyle.SINK, has_const=False, dest_count=0, wide=False, fref=10, ) pe.iram[4] = inst # Inject monadic token tok = MonadToken( target=0, offset=4, act_id=0, data=42, inline=False, ) inject_and_run(env, pe, tok) # Should have FrameSlotWritten event and NO Emitted event slot_written = [e for e in events if isinstance(e, FrameSlotWritten)] emitted = [e for e in events if isinstance(e, Emitted)] assert len(slot_written) > 0 assert slot_written[0].slot == 10 assert slot_written[0].value == 43 # INC(42) = 43 assert len(emitted) == 0 # SINK doesn't emit class TestExtractTag: """AC3.7: EXTRACT_TAG produces packed flit 1 with PE/offset/act_id/port/kind.""" def test_extract_tag_monadic(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=2, config=config, ) # Allocate frame fct = FrameControlToken(target=2, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) # Set up EXTRACT_TAG instruction inst = Instruction( opcode=RoutingOp.EXTRACT_TAG, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=0, ) pe.iram[5] = inst # Set output destination at frame[0] dest = FrameDest( target_pe=0, offset=10, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) pe.frames[pe.tag_store[0][0]][0] = dest # Wire route table pe.route_table[0] = simpy.Store(env) # Inject monadic token at offset 5, act_id 0 tok = MonadToken( target=2, offset=5, act_id=0, data=999, # ignored by EXTRACT_TAG inline=False, ) inject_and_run(env, pe, tok) # Should have Executed event showing EXTRACT_TAG executed = [e for e in events if isinstance(e, Executed)] assert len(executed) > 0 assert executed[0].op == RoutingOp.EXTRACT_TAG # Output should be a packed flit 1 for (pe=2, offset=5, act_id=0) emitted = [e for e in events if isinstance(e, Emitted)] assert len(emitted) > 0 out_token = emitted[0].token # The result should encode (pe=2, offset=5, act_id=0, port=?, kind=?) # Unpack and verify flit1_val = out_token.data unpacked = unpack_flit1(flit1_val) assert unpacked.target_pe == 2 assert unpacked.offset == 5 assert unpacked.act_id == 0 class TestAllocRemote: """AC3.8: ALLOC_REMOTE reads target PE and act_id from frame, sends FrameControlToken.""" def test_alloc_remote(self): env = simpy.Environment() events = [] pe_events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Create a second PE to receive ALLOC config1 = PEConfig(frame_count=4, on_event=pe_events.append) pe1 = ProcessingElement( env=env, pe_id=1, config=config1, ) # Wire route_table for PE0 to reach PE1 pe.route_table[1] = pe1.input_store # Allocate frame for act_id=0 on PE0 fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) # Set up ALLOC_REMOTE instruction, mode 6 (no const, dest_count=0), fref=8 inst = Instruction( opcode=RoutingOp.ALLOC_REMOTE, output=OutputStyle.SINK, has_const=False, dest_count=0, wide=False, fref=8, ) pe.iram[6] = inst # Write target PE and target act_id to frame slots 8 and 9 frame_id, _lane = pe.tag_store[0] pe.frames[frame_id][8] = 1 # target PE pe.frames[frame_id][9] = 2 # target act_id # Inject monadic token tok = MonadToken( target=0, offset=6, act_id=0, data=0, inline=False, ) inject_and_run(env, pe, tok) # PE1 should have received a FrameControlToken(ALLOC) for act_id=2 frame_allocated = [e for e in pe_events if isinstance(e, FrameAllocated)] assert len(frame_allocated) > 0 assert frame_allocated[0].act_id == 2 assert frame_allocated[0].lane == 0 class TestFreeFrameOpcode: """AC3.8: FREE_FRAME opcode deallocates frame, clears tag_store, no output.""" def test_free_frame_opcode(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[0] # Set up FREE_FRAME instruction inst = Instruction( opcode=RoutingOp.FREE_FRAME, output=OutputStyle.SINK, has_const=False, dest_count=0, wide=False, fref=0, ) pe.iram[7] = inst # Inject monadic token tok = MonadToken( target=0, offset=7, act_id=0, data=0, inline=False, ) inject_and_run(env, pe, tok) # Should have FrameFreed event frame_freed = [e for e in events if isinstance(e, FrameFreed)] assert len(frame_freed) > 0 assert frame_freed[0].frame_id == frame_id assert frame_freed[0].lane == 0 assert frame_freed[0].frame_freed == True # tag_store should be cleared assert 0 not in pe.tag_store # frame should be in free_frames assert frame_id in pe.free_frames # Should have NO Emitted event (FREE_FRAME suppresses) emitted = [e for e in events if isinstance(e, Emitted)] assert len(emitted) == 0 class TestFreeLane: """AC3.8: FREE_LANE deallocates lane, potentially returning frame to free list.""" def test_free_lane_on_last_lane_returns_frame(self): """When FREE_LANE is called on the last remaining lane, frame should be returned.""" env = simpy.Environment() events = [] config = PEConfig(frame_count=4, lane_count=2, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate a frame with act_id=1 (gets lane 0) fct_alloc1 = FrameControlToken(target=0, act_id=1, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct_alloc1) frame_id, lane1 = pe.tag_store[1] assert lane1 == 0 # Allocate shared child with act_id=2 (gets lane 1) fct_alloc_shared = FrameControlToken(target=0, act_id=2, op=FrameOp.ALLOC_SHARED, payload=1) inject_and_run(env, pe, fct_alloc_shared) frame_id2, lane2 = pe.tag_store[2] assert frame_id2 == frame_id assert lane2 == 1 # Now FREE_LANE the child (act_id=2) — should not return frame (still in use) fct_free_lane_child = FrameControlToken(target=0, act_id=2, op=FrameOp.FREE_LANE, payload=0) inject_and_run(env, pe, fct_free_lane_child) # Lane should be freed, frame still in use assert 2 not in pe.tag_store assert frame_id in pe.lane_free or frame_id not in [fid for fid, _ in pe.tag_store.values()] frame_freed_child = [e for e in events if isinstance(e, FrameFreed) and e.act_id == 2] assert len(frame_freed_child) > 0 assert frame_freed_child[0].frame_freed == False # Lane freed, not frame # Now FREE_LANE the parent (act_id=1) — this is the last lane, should return frame fct_free_lane_parent = FrameControlToken(target=0, act_id=1, op=FrameOp.FREE_LANE, payload=0) inject_and_run(env, pe, fct_free_lane_parent) # Frame should now be in free_frames assert 1 not in pe.tag_store assert frame_id in pe.free_frames frame_freed_parent = [e for e in events if isinstance(e, FrameFreed) and e.act_id == 1] assert len(frame_freed_parent) > 0 assert frame_freed_parent[0].frame_freed == True # Last lane, frame returned class TestPELocalWriteToken: """AC3.9: PELocalWriteToken with is_dest=True decodes data to FrameDest.""" def test_local_write_iram(self): env = simpy.Environment() config = PEConfig() pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Write instruction to IRAM at slot 10 inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=0, ) inst_word = pack_instruction(inst) write_tok = PELocalWriteToken( target=0, act_id=0, region=0, # IRAM slot=10, data=inst_word, is_dest=False, ) inject_and_run(env, pe, write_tok) # Should have written instruction to pe.iram[10] assert 10 in pe.iram # Unpack and verify loaded_inst = pe.iram[10] assert isinstance(loaded_inst, Instruction) def test_local_write_frame_dest(self): env = simpy.Environment() config = PEConfig(frame_count=4) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[0] # Write FrameDest to frame slot 15, is_dest=True dest = FrameDest( target_pe=1, offset=8, act_id=3, port=Port.R, token_kind=TokenKind.DYADIC, ) flit1 = pack_flit1(dest) write_tok = PELocalWriteToken( target=0, act_id=0, region=1, # Frame slot=15, data=flit1, is_dest=True, ) inject_and_run(env, pe, write_tok) # Frame slot 15 should contain a FrameDest object slot_val = pe.frames[frame_id][15] assert isinstance(slot_val, FrameDest) assert slot_val.target_pe == 1 assert slot_val.offset == 8 assert slot_val.act_id == 3 class TestInvalidActId: """AC1.6: Invalid act_id emits TokenRejected, doesn't crash.""" def test_invalid_act_id_rejected(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Set up instruction at offset 0 (so we get past IFETCH) inst = Instruction( opcode=ArithOp.INC, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=0, ) pe.iram[0] = inst # Inject token with act_id not in tag_store tok = MonadToken( target=0, offset=0, act_id=99, # not allocated data=0, inline=False, ) inject_and_run(env, pe, tok) # Should have TokenRejected event rejected = [e for e in events if isinstance(e, TokenRejected)] assert len(rejected) > 0 assert rejected[0].token == tok class TestDualDestInherit: """IMPORTANT 2: dest_count=2 non-switch: verify both destinations receive same result.""" def test_dual_dest_non_switch(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[0] # Set up instruction: mode 2 (no const, dest_count=2), fref=8 inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=8, ) pe.iram[20] = inst # Set both destination slots dest_l = FrameDest( target_pe=0, offset=10, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) dest_r = FrameDest( target_pe=1, offset=11, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) pe.frames[frame_id][8] = dest_l pe.frames[frame_id][9] = dest_r # Wire route tables pe.route_table[0] = simpy.Store(env) pe.route_table[1] = simpy.Store(env) # Inject dyadic pair tok1 = DyadToken(target=0, offset=20, act_id=0, data=5, port=Port.L) tok2 = DyadToken(target=0, offset=20, act_id=0, data=3, port=Port.R) inject_and_run(env, pe, tok1) inject_and_run(env, pe, tok2) # Should have 2 Emitted events, both with result=8 (5+3) emitted = [e for e in events if isinstance(e, Emitted)] assert len(emitted) >= 2 assert emitted[0].token.data == 8 assert emitted[1].token.data == 8 class TestSwitchOps: """IMPORTANT 2: Switch op (SWEQ) with bool_out=True AND bool_out=False.""" def test_switch_op_bool_out_true(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[0] # Set up SWEQ instruction with dest_count=2, fref=8 inst = Instruction( opcode=RoutingOp.SWEQ, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=8, ) pe.iram[25] = inst # Set destinations: taken=dest_l, not_taken=dest_r when bool_out=True dest_l = FrameDest( target_pe=0, offset=30, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) dest_r = FrameDest( target_pe=1, offset=31, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) pe.frames[frame_id][8] = dest_l pe.frames[frame_id][9] = dest_r # Wire route tables pe.route_table[0] = simpy.Store(env) pe.route_table[1] = simpy.Store(env) # Inject dyadic pair: 5 == 5 => bool_out=True tok1 = DyadToken(target=0, offset=25, act_id=0, data=5, port=Port.L) tok2 = DyadToken(target=0, offset=25, act_id=0, data=5, port=Port.R) inject_and_run(env, pe, tok1) inject_and_run(env, pe, tok2) # Should have 2 Emitted: data_tok to dest_l (taken), trig_tok to dest_r (not_taken) emitted = [e for e in events if isinstance(e, Emitted)] assert len(emitted) >= 2 # Data token goes to taken (dest_l, offset=30) assert emitted[0].token.offset == 30 # Trigger token goes to not_taken (dest_r, offset=31) with data=0 assert emitted[1].token.offset == 31 assert emitted[1].token.data == 0 def test_switch_op_bool_out_false(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[0] # Set up SWEQ instruction with dest_count=2, fref=8 inst = Instruction( opcode=RoutingOp.SWEQ, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=8, ) pe.iram[26] = inst # Set destinations dest_l = FrameDest( target_pe=0, offset=32, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) dest_r = FrameDest( target_pe=1, offset=33, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) pe.frames[frame_id][8] = dest_l pe.frames[frame_id][9] = dest_r # Wire route tables pe.route_table[0] = simpy.Store(env) pe.route_table[1] = simpy.Store(env) # Inject dyadic pair: 5 != 3 => bool_out=False tok1 = DyadToken(target=0, offset=26, act_id=0, data=5, port=Port.L) tok2 = DyadToken(target=0, offset=26, act_id=0, data=3, port=Port.R) inject_and_run(env, pe, tok1) inject_and_run(env, pe, tok2) # Should have 2 Emitted: data_tok to dest_r (taken), trig_tok to dest_l (not_taken) emitted = [e for e in events if isinstance(e, Emitted)] assert len(emitted) >= 2 # When bool_out=False: taken=dest_r, not_taken=dest_l # Data token goes to taken (dest_r, offset=33) assert emitted[0].token.offset == 33 # Trigger token goes to not_taken (dest_l, offset=32) with data=0 assert emitted[1].token.offset == 32 assert emitted[1].token.data == 0 class TestGateSuppression: """IMPORTANT 2: GATE with bool_out=False suppresses output; GATE with bool_out=True outputs.""" def test_gate_suppressed(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[0] # Set up GATE instruction inst = Instruction( opcode=RoutingOp.GATE, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, ) pe.iram[27] = inst # Set destination dest = FrameDest( target_pe=0, offset=40, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) pe.frames[frame_id][8] = dest # Wire route table pe.route_table[0] = simpy.Store(env) # GATE: checks if right != 0. left=5, right=0 => bool_out=False => suppressed tok1 = DyadToken(target=0, offset=27, act_id=0, data=5, port=Port.L) tok2 = DyadToken(target=0, offset=27, act_id=0, data=0, port=Port.R) inject_and_run(env, pe, tok1) inject_and_run(env, pe, tok2) # Should have NO Emitted event (suppressed) emitted = [e for e in events if isinstance(e, Emitted)] assert len(emitted) == 0 def test_gate_allowed(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[0] # Set up GATE instruction inst = Instruction( opcode=RoutingOp.GATE, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=8, ) pe.iram[28] = inst # Set destination dest = FrameDest( target_pe=0, offset=41, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) pe.frames[frame_id][8] = dest # Wire route table pe.route_table[0] = simpy.Store(env) # GATE: 10 > 5 => bool_out=True => allowed tok1 = DyadToken(target=0, offset=28, act_id=0, data=10, port=Port.L) tok2 = DyadToken(target=0, offset=28, act_id=0, data=5, port=Port.R) inject_and_run(env, pe, tok1) inject_and_run(env, pe, tok2) # Should have Emitted event emitted = [e for e in events if isinstance(e, Emitted)] assert len(emitted) >= 1 class TestSMDispatch: """IMPORTANT 3: SM dispatch with return route and proper SMToken construction.""" def test_sm_dispatch_with_return_route(self): env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[0] # Set up SM READ instruction with const (mode 1: const, dest with return route), fref=8 # Const slot contains the SM target, dest slot contains return route inst = Instruction( opcode=MemOp.READ, output=OutputStyle.INHERIT, # Not used for SM ops has_const=True, dest_count=1, wide=False, fref=8, ) pe.iram[50] = inst # Set SM target/address in frame[8] (const slot): SM_id=2, addr=100 sm_target = (2 << 8) | 100 # SM_id in high byte, addr in low byte pe.frames[frame_id][8] = sm_target # Set return route in frame[9] (dest slot after const) ret_dest = FrameDest( target_pe=0, offset=60, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) pe.frames[frame_id][9] = ret_dest # Wire SM route and PE route (for return token) pe.sm_routes[2] = simpy.Store(env) pe.route_table[0] = simpy.Store(env) # Inject dyadic token pair (SM ops treat dyadic as monadic, will match and pair) # left: irrelevant (ignored for SM) # right: data to pass to SM (42) # With has_const=True, data source is: data=right if inst.has_const else left # So data=right=42 tok_l = DyadToken( target=0, offset=50, act_id=0, data=0, # irrelevant port=Port.L, ) tok_r = DyadToken( target=0, offset=50, act_id=0, data=42, # data payload passed to SM port=Port.R, ) inject_and_run(env, pe, tok_l) inject_and_run(env, pe, tok_r) # Should have Emitted event with SMToken emitted = [e for e in events if isinstance(e, Emitted)] assert len(emitted) > 0 sm_token = emitted[0].token assert isinstance(sm_token, SMToken) assert sm_token.target == 2 assert sm_token.addr == 100 assert sm_token.op == MemOp.READ # Note: SM ops treat dyadic tokens as monadic, so left=tok_l.data and right=None # Data source: data=right if inst.has_const else left, so data=None when has_const=True # This is a limitation of the current emulator heuristic assert sm_token.ret is not None assert sm_token.ret.target == 0 assert sm_token.ret.offset == 60 class TestPipelineTiming: """AC3.10: Pipeline timing: 5 cycles dyadic, 4 cycles monadic, 2 cycles side paths.""" def test_dyadic_timing(self): """Verify dyadic pipeline: 5 cycles from second token injection to Emitted event. Pipeline stages: dequeue(1) + IFETCH(1) + MATCH(1) + EXECUTE(1) + EMIT(1) = 5 cycles. The second token triggers the match and begins execution of the complete pipeline. """ env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[0] # Set up dyadic instruction inst = Instruction( opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=0, ) pe.iram[100] = inst # Set destination dest = FrameDest( target_pe=0, offset=101, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) pe.frames[frame_id][0] = dest # Wire route table pe.route_table[0] = simpy.Store(env) # Inject first token (L operand) - will wait for partner tok1 = DyadToken(target=0, offset=100, act_id=0, data=5, port=Port.L) def _put1(): yield pe.input_store.put(tok1) env.process(_put1()) env.run() # Record time when first token was received first_token_received_time = None for e in events: if isinstance(e, TokenReceived) and e.token == tok1: first_token_received_time = e.time break # Clear events, inject second token (R operand) at a new time events.clear() env_snapshot_time = env.now tok2 = DyadToken(target=0, offset=100, act_id=0, data=3, port=Port.R) def _put2(): yield pe.input_store.put(tok2) env.process(_put2()) env.run() # Find the Emitted event for the result emitted_time = None for e in events: if isinstance(e, Emitted): emitted_time = e.time break # Verify timing: from tok2 injection, 5 cycles should elapse to emission. # tok2 is injected at env_snapshot_time, so emission should be at env_snapshot_time + 5. assert first_token_received_time is not None, "First token should have TokenReceived event" assert emitted_time is not None, "Should have Emitted event after second token" # The delta from second token injection (env_snapshot_time) to emission should be 5 cycles delta = emitted_time - env_snapshot_time assert delta == 5, f"Expected 5 cycles, got {delta}" def test_monadic_timing(self): """Verify monadic pipeline: 4 cycles from injection to Emitted event. Pipeline stages: dequeue(1) + IFETCH(1) + EXECUTE(1) + EMIT(1) = 4 cycles. Monadic tokens skip the MATCH stage. """ env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[0] # Set up monadic instruction inst = Instruction( opcode=ArithOp.INC, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=0, ) pe.iram[102] = inst # Set destination dest = FrameDest( target_pe=0, offset=103, act_id=0, port=Port.L, token_kind=TokenKind.MONADIC, ) pe.frames[frame_id][0] = dest # Wire route table pe.route_table[0] = simpy.Store(env) # Record time before injecting injection_time = env.now # Inject monadic token tok = MonadToken( target=0, offset=102, act_id=0, data=42, inline=False, ) def _put(): yield pe.input_store.put(tok) env.process(_put()) env.run() # Find the Emitted event emitted_time = None for e in events: if isinstance(e, Emitted): emitted_time = e.time break # Verify timing: 4 cycles from injection to emission assert emitted_time is not None, "Should have Emitted event" delta = emitted_time - injection_time assert delta == 4, f"Expected 4 cycles for monadic token, got {delta}" def test_side_path_timing(self): """Verify side path pipeline: 2 cycles from injection to FrameAllocated event. Pipeline stages: dequeue(1) + handle(1) = 2 cycles. Side paths (FrameControlToken, PELocalWriteToken) bypass the main pipeline. """ env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Record time before injection injection_time = env.now # Inject FrameControlToken(ALLOC) - side path, 2 cycles fct = FrameControlToken(target=0, act_id=0, op=FrameOp.ALLOC, payload=0) def _put(): yield pe.input_store.put(fct) env.process(_put()) env.run() frame_allocated_time = None for e in events: if isinstance(e, FrameAllocated): frame_allocated_time = e.time break # FrameAllocated should fire exactly 2 cycles after injection (dequeue 1 + handle 1) assert frame_allocated_time is not None, "Should have FrameAllocated event" delta = frame_allocated_time - injection_time assert delta == 2, f"Expected 2 cycles for side path, got {delta}" def test_extract_tag_timing(self): """Verify EXTRACT_TAG timing: 4 cycles from injection to Emitted event. EXTRACT_TAG is a monadic special path that packs PE/offset/act_id into flit 1. Pipeline stages: dequeue(1) + IFETCH(1) + EXECUTE(1) + EMIT(1) = 4 cycles. """ env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=2, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=5, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[5] # Set up EXTRACT_TAG instruction inst = Instruction( opcode=RoutingOp.EXTRACT_TAG, output=OutputStyle.INHERIT, has_const=False, dest_count=1, wide=False, fref=0, ) pe.iram[200] = inst # Set destination for the packed flit 1 result dest = FrameDest( target_pe=1, offset=201, act_id=5, port=Port.L, token_kind=TokenKind.DYADIC, ) pe.frames[frame_id][0] = dest # Wire route table pe.route_table[1] = simpy.Store(env) # Record time before injecting injection_time = env.now # Inject monadic token to EXTRACT_TAG tok = MonadToken( target=0, offset=200, act_id=5, data=99, inline=False, ) def _put(): yield pe.input_store.put(tok) env.process(_put()) env.run() # Find the Emitted event emitted_time = None for e in events: if isinstance(e, Emitted): emitted_time = e.time break # Verify timing: 4 cycles from injection to emission assert emitted_time is not None, "Should have Emitted event for EXTRACT_TAG result" delta = emitted_time - injection_time assert delta == 4, f"Expected 4 cycles for EXTRACT_TAG, got {delta}" def test_sm_dispatch_timing(self): """Verify SM dispatch timing: 4 cycles from injection to Emitted event. SM operations are monadic and emit SMToken at EMIT stage. Pipeline stages: dequeue(1) + IFETCH(1) + EXECUTE(1) + EMIT(1) = 4 cycles. """ env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=7, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[7] # Set up SM READ instruction (monadic in terms of PE pipeline) inst = Instruction( opcode=MemOp.READ, output=OutputStyle.INHERIT, has_const=True, dest_count=1, wide=False, fref=0, ) pe.iram[300] = inst # Set constant (SM target+address) and destination (return route) pe.frames[frame_id][0] = (3 << 8) | 42 # SM 3, address 42 ret_dest = FrameDest( target_pe=0, offset=301, act_id=7, port=Port.L, token_kind=TokenKind.MONADIC, ) pe.frames[frame_id][1] = ret_dest # Wire route table and SM routes pe.route_table[0] = simpy.Store(env) pe.sm_routes[3] = simpy.Store(env) # Record time before injecting injection_time = env.now # Inject monadic token to SM instruction tok = MonadToken( target=0, offset=300, act_id=7, data=55, inline=False, ) def _put(): yield pe.input_store.put(tok) env.process(_put()) env.run() # Find the Emitted event (SMToken emission) emitted_time = None for e in events: if isinstance(e, Emitted): emitted_time = e.time break # Verify timing: 4 cycles from injection to SMToken emission assert emitted_time is not None, "Should have Emitted event for SMToken" delta = emitted_time - injection_time assert delta == 4, f"Expected 4 cycles for SM dispatch, got {delta}" def test_free_frame_timing(self): """Verify FREE_FRAME timing: 4 cycles from injection to FrameFreed event. FREE_FRAME deallocates a frame and suppresses output token. Pipeline stages: dequeue(1) + IFETCH(1) + EXECUTE(1) + EMIT(1) = 4 cycles. """ env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=10, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[10] # Set up FREE_FRAME instruction inst = Instruction( opcode=RoutingOp.FREE_FRAME, output=OutputStyle.INHERIT, has_const=False, dest_count=0, wide=False, fref=0, ) pe.iram[400] = inst # Record time before injecting injection_time = env.now # Inject monadic token to FREE_FRAME tok = MonadToken( target=0, offset=400, act_id=10, data=0, inline=False, ) def _put(): yield pe.input_store.put(tok) env.process(_put()) env.run() # Find the FrameFreed event frame_freed_time = None for e in events: if isinstance(e, FrameFreed): frame_freed_time = e.time break # Verify timing: 4 cycles from injection to FrameFreed event assert frame_freed_time is not None, "Should have FrameFreed event" delta = frame_freed_time - injection_time assert delta == 4, f"Expected 4 cycles for FREE_FRAME, got {delta}" # Also verify frame was actually freed assert 10 not in pe.tag_store, "Frame should be freed from tag_store" assert frame_id in pe.free_frames, "Frame should be returned to free_frames" def test_alloc_remote_timing(self): """Verify ALLOC_REMOTE timing: 4 cycles from injection to delivery. ALLOC_REMOTE constructs a FrameControlToken and routes it to target PE. Pipeline stages: dequeue(1) + IFETCH(1) + EXECUTE(1) + EMIT(1) = 4 cycles. The delivery process (_deliver) adds 1 more cycle after EMIT completes. """ env = simpy.Environment() events = [] config = PEConfig(frame_count=4, on_event=events.append) pe = ProcessingElement( env=env, pe_id=0, config=config, ) # Allocate frame fct = FrameControlToken(target=0, act_id=12, op=FrameOp.ALLOC, payload=0) inject_and_run(env, pe, fct) frame_id, _lane = pe.tag_store[12] # Set up ALLOC_REMOTE instruction inst = Instruction( opcode=RoutingOp.ALLOC_REMOTE, output=OutputStyle.INHERIT, has_const=False, dest_count=0, wide=False, fref=0, ) pe.iram[500] = inst # Set target PE and target act_id in frame slots pe.frames[frame_id][0] = 1 # target PE 1 pe.frames[frame_id][1] = 20 # target act_id 20 # Wire route table for target PE target_store = simpy.Store(env) pe.route_table[1] = target_store # Record time before injecting injection_time = env.now # Inject monadic token to ALLOC_REMOTE tok = MonadToken( target=0, offset=500, act_id=12, data=0, inline=False, ) def _put(): yield pe.input_store.put(tok) env.process(_put()) env.run() # The Executed event fires right before the EXECUTE yield, at time: dequeue(1) + IFETCH(1) = 2 # The EMIT cycle completes at inject + 4 # The _deliver process then yields 1 more cycle before putting to target_store # So FrameControlToken arrives at target_store at inject + 5 cycles total # For this test, we verify the Executed event timing executed_time = None for e in events: if isinstance(e, Executed) and e.op == RoutingOp.ALLOC_REMOTE: executed_time = e.time break assert executed_time is not None, "Should have Executed event for ALLOC_REMOTE" # Executed fires after IFETCH (dequeue 1 + IFETCH 1 = 2) delta = executed_time - injection_time assert delta == 2, f"Expected Executed at 2 cycles, got {delta}" class TestAC2_4_NoBitMaskingOnInstructions: """AC2.4: Verify PE pipeline methods don't do raw bit masking/shifting on instruction fields. Instead, all field access must use encoding.pack_flit1/unpack_flit1. pack_flit1/unpack_flit1 should only appear in CHANGE_TAG, EXTRACT_TAG, and local write handlers. """ def test_process_token_no_raw_bit_masking(self): """_process_token should not contain raw bit shifts/masks on instruction fields.""" import inspect source = inspect.getsource(ProcessingElement._process_token) # Check for common raw bit masking patterns on "inst" # These patterns should not appear (excepting in pack/unpack function calls) disallowed_patterns = [ "inst.opcode & ", # Raw masking on opcode "inst.opcode >> ", # Raw shifting on opcode "inst.fref & ", # Raw masking on fref "inst.fref >> ", # Raw shifting on fref "inst.dest_count & ", # Raw masking on dest_count "inst.dest_count >> ", # Raw shifting on dest_count ] for pattern in disallowed_patterns: assert pattern not in source, \ f"_process_token contains disallowed pattern: {pattern}" def test_match_frame_no_raw_bit_masking(self): """_match_frame should not contain raw bit masks/shifts on instruction fields.""" import inspect source = inspect.getsource(ProcessingElement._match_frame) disallowed_patterns = [ "inst.opcode & ", "inst.opcode >> ", "inst.fref & ", "inst.fref >> ", ] for pattern in disallowed_patterns: assert pattern not in source, \ f"_match_frame contains disallowed pattern: {pattern}" def test_do_emit_new_no_raw_bit_masking(self): """_do_emit_new should not contain raw bit masks/shifts on instruction fields.""" import inspect source = inspect.getsource(ProcessingElement._do_emit_new) disallowed_patterns = [ "inst.opcode & ", "inst.opcode >> ", "inst.fref & ", "inst.fref >> ", ] for pattern in disallowed_patterns: assert pattern not in source, \ f"_do_emit_new contains disallowed pattern: {pattern}" def test_pack_unpack_flit1_only_in_correct_handlers(self): """pack_flit1/unpack_flit1 should only appear in specific contexts. - pack_flit1 should appear in: * _emit_change_tag (CHANGE_TAG output mode handler) * _process_token (for EXTRACT_TAG inline handling) - unpack_flit1 should appear in: * _handle_local_write (frame destination decoding) """ import inspect # pack_flit1 should be in these methods pack_flit1_methods = ['_emit_change_tag', '_process_token'] for method_name in pack_flit1_methods: method = getattr(ProcessingElement, method_name) source = inspect.getsource(method) # pack_flit1 should be called in these methods assert 'pack_flit1(' in source, \ f"{method_name} should call pack_flit1 but doesn't" # unpack_flit1 should be in this method method = getattr(ProcessingElement, '_handle_local_write') source = inspect.getsource(method) assert 'unpack_flit1(' in source, \ "_handle_local_write should call unpack_flit1" # Methods that SHOULD NOT have pack/unpack_flit1 calls # (they use FrameDest objects directly instead of packing/unpacking) disallowed_methods = [ '_match_frame', '_do_emit_new', '_emit_sink', '_emit_inherit', ] for method_name in disallowed_methods: method = getattr(ProcessingElement, method_name) source = inspect.getsource(method) # These should not contain pack_flit1 or unpack_flit1 calls assert 'pack_flit1(' not in source, \ f"{method_name} should not directly call pack_flit1" assert 'unpack_flit1(' not in source, \ f"{method_name} should not directly call unpack_flit1"