Something went wrong. Try again.
OR-1 dataflow CPU sketch
Something went wrong. Try again.
56 kB · 1705 lines
Python
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706"""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 pytestimport 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_instructionfrom emu.events import ( Emitted, Executed, FrameAllocated, FrameFreed, FrameSlotWritten, Matched, TokenReceived, TokenRejected,)from emu.pe import ProcessingElementfrom emu.types import PEConfigfrom 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"