Something went wrong. Try again.
OR-1 dataflow CPU sketch
Something went wrong. Try again.
22 kB · 726 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727"""Tests for monitor/graph_json.py.
Verifies:- or1-monitor.AC3.1 (partial): Graph structure for rendering- or1-monitor.AC3.7: PE state serialization- or1-monitor.AC3.8: SM state serialization"""
from cm_inst import ArithOp, MemOp, Port, FrameDest, Instruction, OutputStylefrom asm.ir import IRGraph, IRNode, IREdge, SourceLoc, SystemConfigfrom emu.events import TokenReceived, Matched, Executed, Emittedfrom monitor.graph_json import graph_to_monitor_json, graph_loaded_jsonfrom monitor.snapshot import StateSnapshot, PESnapshot, SMSnapshot, SMCellSnapshotfrom sm_mod import Presencefrom tokens import DyadToken
class TestGraphJsonStructure: """Test basic JSON structure output."""
def test_graph_loaded_json_structure(self): """Test graph_loaded_json returns correct structure.""" # Create minimal IR graph node = IRNode( name="&test", opcode=ArithOp.ADD, pe=0, iram_offset=0, act_id=0, loc=SourceLoc(1, 1), ) ir_graph = IRGraph( nodes={"&test": node}, edges=[], system=SystemConfig(pe_count=1, sm_count=0), )
# Create Instruction for IRAM inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0)
# Create minimal snapshot pe_snap = PESnapshot( pe_id=0, iram={0: inst}, frames=(), tag_store={}, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={0: pe_snap}, sms={}, )
result = graph_loaded_json(ir_graph, snapshot)
# Check structure assert result["type"] == "graph_loaded" assert "graph" in result assert "nodes" in result["graph"] assert "edges" in result["graph"] assert "state" in result assert "pes" in result["state"] assert "sms" in result["state"] assert result["sim_time"] == 0.0 assert result["finished"] is False
def test_graph_loaded_json_node_fields(self): """Test node fields in graph_loaded_json.""" node = IRNode( name="&add", opcode=ArithOp.ADD, pe=0, iram_offset=5, act_id=1, const=42, loc=SourceLoc(1, 1), ) ir_graph = IRGraph( nodes={"&add": node}, edges=[], system=SystemConfig(pe_count=1, sm_count=0), )
inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) pe_snap = PESnapshot( pe_id=0, iram={5: inst}, frames=(), tag_store={}, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={0: pe_snap}, sms={}, )
result = graph_loaded_json(ir_graph, snapshot) nodes = result["graph"]["nodes"]
assert len(nodes) == 1 node_json = nodes[0] assert node_json["id"] == "&add" assert node_json["opcode"] == "add" assert node_json["pe"] == 0 assert node_json["iram_offset"] == 5 assert node_json["act_id"] == 1 assert node_json["const"] == 42 # Check execution overlay fields are present and False assert node_json["active"] is False assert node_json["matched"] is False assert node_json["executed"] is False
def test_graph_loaded_json_pe_state(self): """Test PE state serialization.""" node = IRNode( name="&add", opcode=ArithOp.ADD, pe=0, iram_offset=5, act_id=1, ) ir_graph = IRGraph(nodes={"&add": node})
inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) pe_snap = PESnapshot( pe_id=0, iram={5: inst}, frames=(), tag_store={}, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={0: pe_snap}, sms={}, )
result = graph_loaded_json(ir_graph, snapshot) pe_state = result["state"]["pes"]["0"]
assert "iram" in pe_state assert "frames" in pe_state assert "tag_store" in pe_state assert "free_frames" in pe_state assert pe_state["input_queue_size"] == 0
def test_graph_loaded_json_sm_state(self): """Test SM state serialization.""" ir_graph = IRGraph()
cell_snap = SMCellSnapshot( pres=Presence.FULL, data_l=42, data_r=None, ) sm_snap = SMSnapshot( sm_id=0, cells={10: cell_snap}, deferred_read=None, t0_store=(), input_queue=(), ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={}, sms={0: sm_snap}, )
result = graph_loaded_json(ir_graph, snapshot) sm_state = result["state"]["sms"]["0"]
assert "cells" in sm_state assert "10" in sm_state["cells"] assert sm_state["cells"]["10"]["presence"] == "FULL" assert sm_state["cells"]["10"]["data_l"] == 42 assert sm_state["deferred_read"] is None assert sm_state["t0_store_size"] == 0
class TestGraphJsonWithEvents: """Test execution overlay with events."""
def test_monitor_json_with_token_received(self): """Test TokenReceived sets active flag.""" node = IRNode( name="&add", opcode=ArithOp.ADD, pe=0, iram_offset=0, act_id=0, ) ir_graph = IRGraph(nodes={"&add": node})
token = DyadToken(target=0, offset=0, act_id=0, data=42, port=Port.L) event = TokenReceived(time=1.0, component="pe:0", token=token)
inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) pe_snap = PESnapshot( pe_id=0, iram={0: inst}, frames=(), tag_store={}, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=1.0, next_time=2.0, pes={0: pe_snap}, sms={}, )
result = graph_to_monitor_json(ir_graph, snapshot, [event])
assert result["type"] == "monitor_update" # Find the node in the result nodes = result["graph"]["nodes"] node_json = next((n for n in nodes if n["id"] == "&add"), None) assert node_json is not None assert node_json["active"] is True
def test_monitor_json_with_matched(self): """Test Matched event sets matched flag.""" node = IRNode( name="&add", opcode=ArithOp.ADD, pe=0, iram_offset=5, act_id=0, ) ir_graph = IRGraph(nodes={"&add": node})
event = Matched(time=1.0, component="pe:0", left=42, right=7, act_id=0, offset=5, frame_id=0)
inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) pe_snap = PESnapshot( pe_id=0, iram={5: inst}, frames=(), tag_store={}, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=1.0, next_time=2.0, pes={0: pe_snap}, sms={}, )
result = graph_to_monitor_json(ir_graph, snapshot, [event])
nodes = result["graph"]["nodes"] node_json = next((n for n in nodes if n["id"] == "&add"), None) assert node_json is not None assert node_json["matched"] is True
def test_monitor_json_with_executed(self): """Test Executed event sets executed flag.""" node = IRNode( name="&add", opcode=ArithOp.ADD, pe=0, iram_offset=0, act_id=0, ) ir_graph = IRGraph(nodes={"&add": node})
event = Executed(time=1.0, component="pe:0", op=ArithOp.ADD, result=49, bool_out=False)
inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0) pe_snap = PESnapshot( pe_id=0, iram={0: inst}, frames=(), tag_store={}, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=1.0, next_time=2.0, pes={0: pe_snap}, sms={}, )
result = graph_to_monitor_json(ir_graph, snapshot, [event])
nodes = result["graph"]["nodes"] node_json = next((n for n in nodes if n["id"] == "&add"), None) assert node_json is not None assert node_json["executed"] is True
def test_monitor_json_events_serialization(self): """Test events are properly serialized.""" ir_graph = IRGraph() token = DyadToken(target=0, offset=0, act_id=0, data=42, port=Port.L) events = [ TokenReceived(time=1.0, component="pe:0", token=token), Executed(time=1.1, component="pe:0", op=ArithOp.ADD, result=49, bool_out=False), ]
snapshot = StateSnapshot( sim_time=1.1, next_time=2.0, pes={}, sms={}, )
result = graph_to_monitor_json(ir_graph, snapshot, events)
assert "events" in result assert len(result["events"]) == 2
event_json_0 = result["events"][0] assert event_json_0["type"] == "TokenReceived" assert event_json_0["time"] == 1.0 assert event_json_0["component"] == "pe:0" assert "details" in event_json_0
event_json_1 = result["events"][1] assert event_json_1["type"] == "Executed" assert event_json_1["time"] == 1.1 assert event_json_1["component"] == "pe:0"
class TestGraphJsonEdges: """Test edge serialization."""
def test_edge_serialization(self): """Test edges are properly serialized.""" source = IRNode( name="&a", opcode=ArithOp.ADD, pe=0, iram_offset=0, act_id=0, ) dest = IRNode( name="&b", opcode=ArithOp.SUB, pe=0, iram_offset=1, act_id=0, ) edge = IREdge( source="&a", dest="&b", port=Port.L, source_port=Port.L, ) ir_graph = IRGraph( nodes={"&a": source, "&b": dest}, edges=[edge], )
snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={}, sms={}, )
result = graph_loaded_json(ir_graph, snapshot) edges = result["graph"]["edges"]
assert len(edges) == 1 edge_json = edges[0] assert edge_json["source"] == "&a" assert edge_json["target"] == "&b" assert edge_json["port"] == "L" # source_port can be None or "L" depending on how it's set assert edge_json["source_port"] in [None, "L"] assert edge_json["token_flow"] is False
def test_edge_token_flow(self): """Test token_flow flag is set on emitted edges.""" source = IRNode( name="&a", opcode=ArithOp.ADD, pe=0, iram_offset=0, act_id=0, ) dest = IRNode( name="&b", opcode=ArithOp.SUB, pe=1, # Destination on a different PE iram_offset=1, act_id=0, ) edge = IREdge(source="&a", dest="&b", port=Port.L) ir_graph = IRGraph( nodes={"&a": source, "&b": dest}, edges=[edge], )
# Emitted event with token targeting PE 1 should mark the edge token = DyadToken(target=1, offset=1, act_id=0, data=42, port=Port.L) event = Emitted(time=1.0, component="pe:0", token=token)
snapshot = StateSnapshot( sim_time=1.0, next_time=2.0, pes={}, sms={}, )
result = graph_to_monitor_json(ir_graph, snapshot, [event]) edges = result["graph"]["edges"]
assert len(edges) == 1 edge_json = edges[0] # Edge token_flow should be True when an Emitted event targets the destination PE assert edge_json["token_flow"] is True
class TestGraphJsonFinished: """Test finished flag."""
def test_finished_false_when_next_time_not_inf(self): """Test finished=False when next_time is finite.""" ir_graph = IRGraph() snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={}, sms={}, )
result = graph_loaded_json(ir_graph, snapshot) assert result["finished"] is False
def test_finished_true_when_next_time_inf(self): """Test finished=True when next_time is infinity.""" ir_graph = IRGraph() snapshot = StateSnapshot( sim_time=10.0, next_time=float('inf'), pes={}, sms={}, )
result = graph_loaded_json(ir_graph, snapshot) assert result["finished"] is True
class TestTagStoreSerialisation: """Test non-empty tag_store JSON serialization in PE state."""
def test_non_empty_tag_store_serialization(self): """Test that tag_store with non-empty mapping serializes correctly.""" node = IRNode( name="&add", opcode=ArithOp.ADD, pe=0, iram_offset=0, act_id=0, ) ir_graph = IRGraph(nodes={"&add": node})
inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0)
# Create PESnapshot with non-empty tag_store: act_id 0 maps to frame 2, lane 1 pe_snap = PESnapshot( pe_id=0, iram={0: inst}, frames=(), tag_store={0: (2, 1)}, # act_id=0 -> (frame_id=2, lane=1) presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={0: pe_snap}, sms={}, )
result = graph_loaded_json(ir_graph, snapshot) pe_state = result["state"]["pes"]["0"]
# Verify tag_store is correctly serialized assert "tag_store" in pe_state assert "0" in pe_state["tag_store"] # act_id 0 should be a string key "0" assert pe_state["tag_store"]["0"]["frame_id"] == 2 assert pe_state["tag_store"]["0"]["lane"] == 1 assert pe_state["lane_count"] == 4
def test_multiple_entries_tag_store_serialization(self): """Test tag_store with multiple act_id entries serializes correctly.""" ir_graph = IRGraph()
inst = Instruction(opcode=ArithOp.ADD, output=OutputStyle.INHERIT, has_const=False, dest_count=2, wide=False, fref=0)
# Create PESnapshot with multiple tag_store entries pe_snap = PESnapshot( pe_id=0, iram={0: inst}, frames=(), tag_store={ 0: (2, 1), 1: (3, 0), 5: (7, 2), }, presence=(), port_store=(), match_data=(), free_frames=(), lane_count=4, input_queue=(), output_log=(), ) snapshot = StateSnapshot( sim_time=0.0, next_time=1.0, pes={0: pe_snap}, sms={}, )
result = graph_loaded_json(ir_graph, snapshot) pe_state = result["state"]["pes"]["0"]
# Verify all entries are correctly serialized assert "tag_store" in pe_state assert pe_state["tag_store"]["0"]["frame_id"] == 2 assert pe_state["tag_store"]["0"]["lane"] == 1 assert pe_state["tag_store"]["1"]["frame_id"] == 3 assert pe_state["tag_store"]["1"]["lane"] == 0 assert pe_state["tag_store"]["5"]["frame_id"] == 7 assert pe_state["tag_store"]["5"]["lane"] == 2 assert pe_state["lane_count"] == 4
class TestAC72_NewEventTypesSerialization: """AC7.2: Verify new frame-based event types serialize correctly to JSON."""
def test_frame_allocated_event_serialization(self): """FrameAllocated event should serialize with correct type and fields.""" from emu.events import FrameAllocated
event = FrameAllocated( time=5.0, component="pe:0", act_id=1, frame_id=2, lane=0, )
ir_graph = IRGraph() snapshot = StateSnapshot( sim_time=5.0, next_time=6.0, pes={}, sms={}, )
result = graph_to_monitor_json(ir_graph, snapshot, [event])
# Extract the events from result assert "events" in result events_list = result["events"] assert len(events_list) == 1
event_json = events_list[0] assert event_json["type"] == "FrameAllocated" assert event_json["time"] == 5.0 assert event_json["component"] == "pe:0" assert "details" in event_json assert event_json["details"]["act_id"] == 1 assert event_json["details"]["frame_id"] == 2 assert event_json["details"]["lane"] == 0
def test_frame_freed_event_serialization(self): """FrameFreed event should serialize with correct type and fields.""" from emu.events import FrameFreed
event = FrameFreed( time=10.0, component="pe:1", act_id=3, frame_id=4, lane=0, frame_freed=True, )
ir_graph = IRGraph() snapshot = StateSnapshot( sim_time=10.0, next_time=11.0, pes={}, sms={}, )
result = graph_to_monitor_json(ir_graph, snapshot, [event])
events_list = result["events"] assert len(events_list) == 1
event_json = events_list[0] assert event_json["type"] == "FrameFreed" assert event_json["time"] == 10.0 assert event_json["component"] == "pe:1" assert "details" in event_json assert event_json["details"]["act_id"] == 3 assert event_json["details"]["frame_id"] == 4 assert event_json["details"]["lane"] == 0 assert event_json["details"]["frame_freed"] == True
def test_frame_slot_written_event_serialization(self): """FrameSlotWritten event should serialize with correct type and fields.""" from emu.events import FrameSlotWritten
event = FrameSlotWritten( time=7.0, component="pe:0", frame_id=1, slot=5, value=0x1234, )
ir_graph = IRGraph() snapshot = StateSnapshot( sim_time=7.0, next_time=8.0, pes={}, sms={}, )
result = graph_to_monitor_json(ir_graph, snapshot, [event])
events_list = result["events"] assert len(events_list) == 1
event_json = events_list[0] assert event_json["type"] == "FrameSlotWritten" assert event_json["time"] == 7.0 assert event_json["component"] == "pe:0" assert "details" in event_json assert event_json["details"]["frame_id"] == 1 assert event_json["details"]["slot"] == 5 assert event_json["details"]["value"] == 0x1234
def test_token_rejected_event_serialization(self): """TokenRejected event should serialize with correct type and fields.""" from emu.events import TokenRejected from tokens import DyadToken
token = DyadToken( target=0, offset=1, act_id=99, data=0x5678, port=Port.L, )
event = TokenRejected( time=3.0, component="pe:0", token=token, reason="invalid act_id", )
ir_graph = IRGraph() snapshot = StateSnapshot( sim_time=3.0, next_time=4.0, pes={}, sms={}, )
result = graph_to_monitor_json(ir_graph, snapshot, [event])
events_list = result["events"] assert len(events_list) == 1
event_json = events_list[0] assert event_json["type"] == "TokenRejected" assert event_json["time"] == 3.0 assert event_json["component"] == "pe:0" assert "details" in event_json assert event_json["details"]["reason"] == "invalid act_id"