nonbinary.computer
OR-1 dataflow CPU sketch
1"""
2Tests for initialization API smoke tests.
3
4Verifies:
5- or1-emu.AC5.1: IRAM initialization — PE has expected instructions at expected offsets
6- or1-emu.AC5.2: SM cell initialization — SM cells match config
7- or1-emu.AC5.3: Token injection — inject() routes tokens to correct stores by type
8"""
9
10import simpy
11
12from cm_inst import ALUInst, Addr, ArithOp, CfgOp, MemOp, Port, RoutingOp, SMInst
13from emu import build_topology, PEConfig, SMConfig
14from sm_mod import Presence
15from tokens import CMToken, CfgToken, DyadToken, LoadInstToken, MonadToken, SMToken
16
17
18class TestAC51IRAMInitialization:
19 """Test AC5.1: IRAM initialization"""
20
21 def test_iram_contains_instructions_at_configured_offsets(self):
22 """IRAM contains ALUInst at offsets specified in PEConfig."""
23 env = simpy.Environment()
24
25 pe_iram = {
26 0: ALUInst(
27 op=ArithOp.ADD,
28 dest_l=None,
29 dest_r=None,
30 const=None,
31 ),
32 5: ALUInst(
33 op=RoutingOp.CONST,
34 dest_l=None,
35 dest_r=None,
36 const=99,
37 ),
38 }
39
40 sys = build_topology(
41 env,
42 [PEConfig(pe_id=0, iram=pe_iram)],
43 [],
44 )
45
46 # Verify instructions are at expected offsets
47 assert 0 in sys.pes[0].iram
48 assert sys.pes[0].iram[0].op == ArithOp.ADD
49 assert 5 in sys.pes[0].iram
50 assert sys.pes[0].iram[5].op == RoutingOp.CONST
51 assert sys.pes[0].iram[5].const == 99
52
53 def test_iram_does_not_contain_uninitialized_offsets(self):
54 """IRAM does not contain offsets not specified in config."""
55 env = simpy.Environment()
56
57 pe_iram = {
58 0: ALUInst(op=ArithOp.ADD, dest_l=None, dest_r=None, const=None),
59 5: ALUInst(op=RoutingOp.CONST, dest_l=None, dest_r=None, const=99),
60 }
61
62 sys = build_topology(
63 env,
64 [PEConfig(pe_id=0, iram=pe_iram)],
65 [],
66 )
67
68 # Verify uninitialized offsets are NOT in IRAM
69 assert 3 not in sys.pes[0].iram
70 assert 10 not in sys.pes[0].iram
71
72
73class TestAC52SMCellInitialization:
74 """Test AC5.2: SM cell initialization"""
75
76 def test_sm_cells_initialized_with_presence_and_data(self):
77 """SM cells initialized via config contain expected presence and data."""
78 env = simpy.Environment()
79
80 sm_config = SMConfig(
81 sm_id=0,
82 cell_count=512,
83 initial_cells={
84 0: (Presence.FULL, 42),
85 10: (Presence.RESERVED, None),
86 },
87 )
88
89 sys = build_topology(
90 env,
91 [],
92 [sm_config],
93 )
94
95 # Verify cell 0: FULL with data 42
96 assert sys.sms[0].cells[0].pres == Presence.FULL
97 assert sys.sms[0].cells[0].data_l == 42
98
99 # Verify cell 10: RESERVED with no data
100 assert sys.sms[0].cells[10].pres == Presence.RESERVED
101 assert sys.sms[0].cells[10].data_l is None
102
103 def test_uninitialized_sm_cells_are_empty(self):
104 """SM cells not in initial_cells config are EMPTY."""
105 env = simpy.Environment()
106
107 sm_config = SMConfig(
108 sm_id=0,
109 cell_count=512,
110 initial_cells={
111 0: (Presence.FULL, 42),
112 },
113 )
114
115 sys = build_topology(
116 env,
117 [],
118 [sm_config],
119 )
120
121 # Verify uninitialized cells are EMPTY
122 assert sys.sms[0].cells[1].pres == Presence.EMPTY
123 assert sys.sms[0].cells[1].data_l is None
124 assert sys.sms[0].cells[100].pres == Presence.EMPTY
125 assert sys.sms[0].cells[100].data_l is None
126
127
128class TestAC53TokenInjection:
129 """Test AC5.3: Token injection API"""
130
131 def test_inject_monad_token_to_pe(self):
132 """inject() delivers MonadToken to correct PE's input_store."""
133 env = simpy.Environment()
134
135 sys = build_topology(
136 env,
137 [PEConfig(pe_id=0, iram={}), PEConfig(pe_id=1, iram={})],
138 [],
139 )
140
141 # Create and inject token to PE0
142 token = MonadToken(
143 target=0,
144 offset=0,
145 ctx=0,
146 data=42,
147 inline=False,
148 )
149
150 sys.inject(token)
151
152 # Verify token is in PE0's input_store
153 assert len(sys.pes[0].input_store.items) == 1
154 assert sys.pes[0].input_store.items[0] == token
155
156 def test_inject_multiple_tokens_to_correct_pes(self):
157 """inject() places tokens in correct PE based on target."""
158 env = simpy.Environment()
159
160 sys = build_topology(
161 env,
162 [PEConfig(pe_id=0, iram={}), PEConfig(pe_id=1, iram={})],
163 [],
164 )
165
166 # Inject token to PE0
167 token0 = MonadToken(target=0, offset=0, ctx=0, data=10, inline=False)
168 sys.inject(token0)
169
170 # Inject token to PE1
171 token1 = MonadToken(target=1, offset=0, ctx=0, data=20, inline=False)
172 sys.inject(token1)
173
174 # Verify tokens arrived at correct PEs
175 assert len(sys.pes[0].input_store.items) == 1
176 assert sys.pes[0].input_store.items[0].data == 10
177
178 assert len(sys.pes[1].input_store.items) == 1
179 assert sys.pes[1].input_store.items[0].data == 20
180
181 def test_inject_sm_token_to_sm(self):
182 """inject() delivers SMToken to correct SM's input_store."""
183 env = simpy.Environment()
184
185 sys = build_topology(
186 env,
187 [PEConfig(pe_id=0, iram={})],
188 [SMConfig(sm_id=0, cell_count=512), SMConfig(sm_id=1, cell_count=512)],
189 )
190
191 # Create and inject token to SM0
192 token = SMToken(
193 target=0,
194 addr=5,
195 op=MemOp.READ,
196 flags=None,
197 data=None,
198 ret=CMToken(target=0, offset=0, ctx=0, data=0),
199 )
200
201 sys.inject(token)
202
203 # Verify token is in SM0's input_store
204 assert len(sys.sms[0].input_store.items) == 1
205 assert sys.sms[0].input_store.items[0] == token
206
207 def test_inject_sm_multiple_tokens_to_correct_sms(self):
208 """inject() routes SMTokens to correct SM based on token.target."""
209 env = simpy.Environment()
210
211 sys = build_topology(
212 env,
213 [PEConfig(pe_id=0, iram={})],
214 [SMConfig(sm_id=0, cell_count=512), SMConfig(sm_id=1, cell_count=512)],
215 )
216
217 # Inject token to SM0
218 token0 = SMToken(
219 target=0,
220 addr=10,
221 op=MemOp.WRITE,
222 flags=None,
223 data=42,
224 ret=None,
225 )
226 sys.inject(token0)
227
228 # Inject token to SM1
229 token1 = SMToken(
230 target=1,
231 addr=20,
232 op=MemOp.READ,
233 flags=None,
234 data=None,
235 ret=CMToken(target=0, offset=0, ctx=0, data=0),
236 )
237 sys.inject(token1)
238
239 # Verify tokens arrived at correct SMs
240 assert len(sys.sms[0].input_store.items) == 1
241 assert sys.sms[0].input_store.items[0].addr == 10
242
243 assert len(sys.sms[1].input_store.items) == 1
244 assert sys.sms[1].input_store.items[0].addr == 20
245
246
247class TestAC51GenCounterInitialization:
248 """Test AC5.1 extended: gen_counter initialization"""
249
250 def test_gen_counters_initialized_from_config(self):
251 """PEConfig with gen_counters list initializes PE's gen_counters."""
252 env = simpy.Environment()
253
254 sys = build_topology(
255 env,
256 [PEConfig(pe_id=0, iram={}, gen_counters=[1, 0, 2, 3])],
257 [],
258 )
259
260 # Verify gen_counters match config
261 assert sys.pes[0].gen_counters == [1, 0, 2, 3]
262
263 def test_gen_counters_default_to_zero_when_none(self):
264 """PEConfig with gen_counters=None (default) initializes all to 0."""
265 env = simpy.Environment()
266
267 sys = build_topology(
268 env,
269 [PEConfig(pe_id=0, iram={}, gen_counters=None)],
270 [],
271 )
272
273 # Verify all gen_counters are 0 (ctx_slots default is 16)
274 assert sys.pes[0].gen_counters == [0] * 16
275
276 def test_gen_counters_with_custom_ctx_slots(self):
277 """gen_counters list length matches ctx_slots."""
278 env = simpy.Environment()
279
280 sys = build_topology(
281 env,
282 [PEConfig(pe_id=0, iram={}, ctx_slots=8, gen_counters=[1, 2, 3, 4, 5, 6, 7, 8])],
283 [],
284 )
285
286 # Verify gen_counters match provided list
287 assert sys.pes[0].gen_counters == [1, 2, 3, 4, 5, 6, 7, 8]
288 assert len(sys.pes[0].gen_counters) == 8
289
290
291class TestAC61E2EConstFedsAdd:
292 """Test AC6.1: CONST on PE0 feeds ADD on PE1"""
293
294 def test_const_feeds_add(self):
295 """CONST on PE0 emits tokens that arrive at PE1, trigger ADD, produce correct result."""
296 env = simpy.Environment()
297
298 # PE0 IRAM: offset 0 = CONST(7), offset 1 = CONST(3)
299 pe0_iram = {
300 0: ALUInst(
301 op=RoutingOp.CONST,
302 dest_l=Addr(a=0, port=Port.L, pe=1),
303 dest_r=None,
304 const=7,
305 ),
306 1: ALUInst(
307 op=RoutingOp.CONST,
308 dest_l=Addr(a=0, port=Port.R, pe=1),
309 dest_r=None,
310 const=3,
311 ),
312 }
313
314 # PE1 IRAM: offset 0 = ADD routing to PE2 (collector)
315 pe1_iram = {
316 0: ALUInst(
317 op=ArithOp.ADD,
318 dest_l=Addr(a=0, port=Port.L, pe=2),
319 dest_r=None,
320 const=None,
321 ),
322 }
323
324 # Build topology with 3 PEs (PE2 has no IRAM, acts as collector)
325 sys = build_topology(
326 env,
327 [
328 PEConfig(pe_id=0, iram=pe0_iram),
329 PEConfig(pe_id=1, iram=pe1_iram),
330 PEConfig(pe_id=2, iram={}),
331 ],
332 [],
333 )
334
335 # Set up PE1's routing to direct output to a collector store (not PE2's input_store)
336 # This simulates a sink where results are collected without being consumed
337 collector_store = simpy.Store(env, capacity=100)
338 sys.pes[1].route_table[2] = collector_store
339
340 # Inject tokens via SimPy process
341 def injector():
342 yield sys.pes[0].input_store.put(MonadToken(target=0, offset=0, ctx=0, data=0, inline=False))
343 yield sys.pes[0].input_store.put(MonadToken(target=0, offset=1, ctx=0, data=0, inline=False))
344
345 env.process(injector())
346
347 # Run simulation until quiescence
348 env.run(until=1000)
349
350 # Verify collector receives exactly one token with data=10 (7+3)
351 assert len(collector_store.items) == 1
352 result_token = collector_store.items[0]
353 assert result_token.data == 10
354
355
356class TestAC62E2ESMRoundTrip:
357 """Test AC6.2: SM round-trip (PE writes, PE reads)"""
358
359 def test_sm_round_trip(self):
360 """PE writes to SM, another PE reads from SM, receives correct data."""
361 env = simpy.Environment()
362
363 # PE0 IRAM: offset 0 = SM WRITE, offset 1 = SM READ
364 pe0_iram = {
365 0: SMInst(op=MemOp.WRITE, sm_id=0, const=0), # Write to cell 0
366 1: SMInst(
367 op=MemOp.READ,
368 sm_id=0,
369 const=0, # Read from cell 0
370 ret=Addr(a=0, port=Port.L, pe=1),
371 ),
372 }
373
374 # PE1 IRAM: offset 0 = PASS (to pass through SM READ result)
375 pe1_iram = {
376 0: ALUInst(
377 op=RoutingOp.PASS,
378 dest_l=Addr(a=0, port=Port.L, pe=1),
379 dest_r=None,
380 const=None,
381 ),
382 }
383
384 # Build topology
385 sys = build_topology(
386 env,
387 [
388 PEConfig(pe_id=0, iram=pe0_iram),
389 PEConfig(pe_id=1, iram=pe1_iram),
390 ],
391 [
392 # SM0: cell 0 starts EMPTY
393 SMConfig(sm_id=0, cell_count=512, initial_cells={}),
394 ],
395 )
396
397 # Set up collector for PE1's output
398 collector_store = simpy.Store(env, capacity=100)
399 sys.pes[1].route_table[1] = collector_store
400
401 # Inject tokens via SimPy process (FIFO order ensures WRITE before READ)
402 def injector():
403 yield sys.pes[0].input_store.put(MonadToken(target=0, offset=0, ctx=0, data=42, inline=False))
404 yield sys.pes[0].input_store.put(MonadToken(target=0, offset=1, ctx=0, data=0, inline=False))
405
406 env.process(injector())
407
408 # Run simulation
409 env.run(until=1000)
410
411 # Verify collector receives a token with data=42 (the value written and read back)
412 assert len(collector_store.items) >= 1
413 result_token = collector_store.items[0]
414 assert result_token.data == 42
415
416
417class TestAC63E2EDualFanout:
418 """Test AC6.3: DUAL mode fan-out to two consumers"""
419
420 def test_dual_fanout(self):
421 """DUAL mode emits same result to both PE1 and PE2."""
422 env = simpy.Environment()
423
424 # PE0 IRAM: offset 0 = PASS with dual destinations
425 pe0_iram = {
426 0: ALUInst(
427 op=RoutingOp.PASS,
428 dest_l=Addr(a=0, port=Port.L, pe=1),
429 dest_r=Addr(a=0, port=Port.L, pe=2),
430 const=None,
431 ),
432 }
433
434 # Build topology
435 sys = build_topology(
436 env,
437 [
438 PEConfig(pe_id=0, iram=pe0_iram),
439 PEConfig(pe_id=1, iram={}), # PE1: collector
440 PEConfig(pe_id=2, iram={}), # PE2: collector
441 ],
442 [],
443 )
444
445 # Set up collectors for PE1 and PE2
446 collector_1 = simpy.Store(env, capacity=100)
447 collector_2 = simpy.Store(env, capacity=100)
448 sys.pes[0].route_table[1] = collector_1
449 sys.pes[0].route_table[2] = collector_2
450
451 # Inject token via SimPy process
452 def injector():
453 yield sys.pes[0].input_store.put(MonadToken(target=0, offset=0, ctx=0, data=99, inline=False))
454
455 env.process(injector())
456
457 # Run simulation
458 env.run(until=1000)
459
460 # Verify each collector receives one token with data=99
461 assert len(collector_1.items) == 1
462 assert collector_1.items[0].data == 99
463
464 assert len(collector_2.items) == 1
465 assert collector_2.items[0].data == 99
466
467
468class TestAC64E2ESwitchRouting:
469 """Test AC6.4: SWITCH mode conditional routing"""
470
471 def test_switch_routing_condition_true(self):
472 """SWEQ with equal operands routes data to dest_l, trigger to dest_r."""
473 env = simpy.Environment()
474
475 # PE0 IRAM: offset 0 = SWEQ
476 pe0_iram = {
477 0: ALUInst(
478 op=RoutingOp.SWEQ,
479 dest_l=Addr(a=0, port=Port.L, pe=1),
480 dest_r=Addr(a=0, port=Port.L, pe=2),
481 const=None,
482 ),
483 }
484
485 # Build topology with gen_counters initialized for dyadic matching
486 sys = build_topology(
487 env,
488 [
489 PEConfig(pe_id=0, iram=pe0_iram, gen_counters=[0, 0, 0, 0]),
490 PEConfig(pe_id=1, iram={}), # PE1: receives data token
491 PEConfig(pe_id=2, iram={}), # PE2: receives inline trigger
492 ],
493 [],
494 )
495
496 # Set up collectors
497 collector_1 = simpy.Store(env, capacity=100)
498 collector_2 = simpy.Store(env, capacity=100)
499 sys.pes[0].route_table[1] = collector_1
500 sys.pes[0].route_table[2] = collector_2
501
502 # Inject two DyadTokens with same data (5, 5) via SimPy process
503 def injector():
504 yield sys.pes[0].input_store.put(
505 DyadToken(
506 target=0,
507 offset=0,
508 ctx=0,
509 data=5,
510 port=Port.L,
511 gen=0,
512 wide=False,
513 )
514 )
515 yield sys.pes[0].input_store.put(
516 DyadToken(
517 target=0,
518 offset=0,
519 ctx=0,
520 data=5,
521 port=Port.R,
522 gen=0,
523 wide=False,
524 )
525 )
526
527 env.process(injector())
528
529 # Run simulation
530 env.run(until=1000)
531
532 # When bool_out=True (equal): data → dest_l (collector_1), trigger → dest_r (collector_2)
533 assert len(collector_1.items) == 1
534 data_token = collector_1.items[0]
535 assert data_token.data == 5
536
537 assert len(collector_2.items) == 1
538 trigger_token = collector_2.items[0]
539 assert trigger_token.inline is True
540 assert trigger_token.data == 0
541
542 def test_switch_routing_condition_false(self):
543 """SWEQ with unequal operands routes data to dest_r, trigger to dest_l."""
544 env = simpy.Environment()
545
546 # PE0 IRAM: offset 0 = SWEQ
547 pe0_iram = {
548 0: ALUInst(
549 op=RoutingOp.SWEQ,
550 dest_l=Addr(a=0, port=Port.L, pe=1),
551 dest_r=Addr(a=0, port=Port.L, pe=2),
552 const=None,
553 ),
554 }
555
556 # Build topology
557 sys = build_topology(
558 env,
559 [
560 PEConfig(pe_id=0, iram=pe0_iram, gen_counters=[0, 0, 0, 0]),
561 PEConfig(pe_id=1, iram={}), # PE1: receives inline trigger
562 PEConfig(pe_id=2, iram={}), # PE2: receives data token
563 ],
564 [],
565 )
566
567 # Set up collectors
568 collector_1 = simpy.Store(env, capacity=100)
569 collector_2 = simpy.Store(env, capacity=100)
570 sys.pes[0].route_table[1] = collector_1
571 sys.pes[0].route_table[2] = collector_2
572
573 # Inject two DyadTokens with different data (5, 10) via SimPy process
574 def injector():
575 yield sys.pes[0].input_store.put(
576 DyadToken(
577 target=0,
578 offset=0,
579 ctx=0,
580 data=5,
581 port=Port.L,
582 gen=0,
583 wide=False,
584 )
585 )
586 yield sys.pes[0].input_store.put(
587 DyadToken(
588 target=0,
589 offset=0,
590 ctx=0,
591 data=10,
592 port=Port.R,
593 gen=0,
594 wide=False,
595 )
596 )
597
598 env.process(injector())
599
600 # Run simulation
601 env.run(until=1000)
602
603 # When bool_out=False (not equal): data → dest_r (collector_2), trigger → dest_l (collector_1)
604 assert len(collector_2.items) == 1
605 data_token = collector_2.items[0]
606 assert data_token.data == 5 # First operand goes to dest_r
607
608 assert len(collector_1.items) == 1
609 trigger_token = collector_1.items[0]
610 assert trigger_token.inline is True
611 assert trigger_token.data == 0
612
613
614class TestTask5CfgTokenLoadInst:
615 """Test CfgToken LOAD_INST handling in PE"""
616
617 def test_cfg_token_load_inst(self):
618 """CfgToken with LOAD_INST dynamically loads instructions into PE IRAM."""
619 env = simpy.Environment()
620
621 # Build topology: PE0 starts with empty IRAM, PE1 is collector
622 sys = build_topology(
623 env,
624 [
625 PEConfig(pe_id=0, iram={}), # Empty IRAM initially
626 PEConfig(pe_id=1, iram={}), # Collector
627 ],
628 [],
629 )
630
631 # Set up collector for output
632 collector_store = simpy.Store(env, capacity=100)
633 sys.pes[0].route_table[1] = collector_store
634
635 # Create instruction to load: CONST(42) routing to PE1
636 const_inst = ALUInst(
637 op=RoutingOp.CONST,
638 dest_l=Addr(a=0, port=Port.L, pe=1),
639 dest_r=None,
640 const=42,
641 )
642
643 # Create LoadInstToken to load instruction at offset 0
644 cfg_token = LoadInstToken(
645 target=0,
646 addr=0,
647 op=CfgOp.LOAD_INST,
648 instructions=(const_inst,),
649 )
650
651 # Function to send CfgToken then seed token
652 def injector():
653 # First, send CfgToken to load the instruction
654 yield sys.pes[0].input_store.put(cfg_token)
655 # Then inject a MonadToken seed to trigger the loaded instruction
656 yield sys.pes[0].input_store.put(
657 MonadToken(target=0, offset=0, ctx=0, data=0, inline=False)
658 )
659
660 env.process(injector())
661
662 # Run simulation
663 env.run(until=1000)
664
665 # Verify:
666 # 1. The instruction was loaded into IRAM
667 assert 0 in sys.pes[0].iram
668 assert sys.pes[0].iram[0].op == RoutingOp.CONST
669 assert sys.pes[0].iram[0].const == 42
670
671 # 2. The loaded instruction executed and produced output
672 assert len(collector_store.items) == 1
673 result_token = collector_store.items[0]
674 assert result_token.data == 42