nonbinary.computer
OR-1 dataflow CPU sketch
1"""
2Tests for Structure Memory operations.
3
4Verifies all acceptance criteria:
5- or1-emu.AC3.1: READ on FULL cell returns data immediately via result token
6- or1-emu.AC3.2: READ on EMPTY cell with empty deferred register stashes return route, sets WAITING
7- or1-emu.AC3.3: WRITE on WAITING cell satisfies deferred read — emits result token to stashed return route
8- or1-emu.AC3.4: WRITE on EMPTY/RESERVED sets cell to FULL
9- or1-emu.AC3.5: CLEAR sets cell to EMPTY, cancels deferred read if targeting that cell
10- or1-emu.AC3.6: READ_INC/READ_DEC atomically modify and return value (lower 256 cells only)
11- or1-emu.AC3.7: Depth-1 constraint: second blocking READ on different empty cell stalls until first deferred read is satisfied
12- or1-emu.AC3.8: WRITE on FULL cell overwrites data (diagnostic flag set if modelled)
13- or1-emu.AC3.9: CAS on FULL cell: if current value == expected (SMToken.flags), writes new value (SMToken.data) and returns old value; if mismatch, cell unchanged and returns old value (lower 256 cells only)
14"""
15
16import simpy
17from hypothesis import given, strategies as st
18
19from cm_inst import MemOp
20from emu.sm import StructureMemory
21from sm_mod import Presence
22from tests.conftest import sm_token, sm_return_route, uint16
23from tokens import CMToken, SMToken
24
25
26def inject_token(env: simpy.Environment, store: simpy.Store, token):
27 """Helper to inject token into store via a process."""
28 def _injector():
29 yield store.put(token)
30
31 env.process(_injector())
32
33
34class TestAC3_1ReadOnFull:
35 """AC3.1: READ on FULL cell returns data immediately via result token."""
36
37 def test_read_on_full_cell(self):
38 env = simpy.Environment()
39 sm = StructureMemory(env, 0, cell_count=512)
40
41 # Pre-populate cell 10 to FULL with data 0xBEEF
42 sm.cells[10].pres = Presence.FULL
43 sm.cells[10].data_l = 0xBEEF
44
45 # Create collector store for results
46 collector = simpy.Store(env)
47 sm.route_table[0] = collector
48
49 # Inject READ token targeting cell 10
50 ret_route = CMToken(target=0, offset=5, act_id=1, data=0)
51 read_token = SMToken(target=0, addr=10, op=MemOp.READ, flags=None, data=None, ret=ret_route)
52 inject_token(env, sm.input_store, read_token)
53
54 # Run simulation
55 env.run(until=100)
56
57 # Verify result token in collector with correct data
58 assert len(collector.items) == 1
59 result = collector.items[0]
60 assert result.data == 0xBEEF
61 assert result.offset == 5
62 assert result.act_id == 1
63
64
65class TestAC3_2ReadOnEmpty:
66 """AC3.2: READ on EMPTY cell with empty deferred register stashes return route, sets WAITING."""
67
68 def test_read_on_empty_cell(self):
69 env = simpy.Environment()
70 sm = StructureMemory(env, 0, cell_count=512)
71
72 # Cell 20 starts EMPTY
73 assert sm.cells[20].pres == Presence.EMPTY
74
75 # Create collector store
76 collector = simpy.Store(env)
77 sm.route_table[0] = collector
78
79 # Inject READ token targeting cell 20
80 ret_route = CMToken(target=0, offset=7, act_id=2, data=0)
81 read_token = SMToken(target=0, addr=20, op=MemOp.READ, flags=None, data=None, ret=ret_route)
82 inject_token(env, sm.input_store, read_token)
83
84 # Run simulation
85 env.run(until=100)
86
87 # Verify cell is now WAITING
88 assert sm.cells[20].pres == Presence.WAITING
89
90 # Verify deferred_read is set
91 assert sm.deferred_read is not None
92 assert sm.deferred_read.cell_addr == 20
93 assert sm.deferred_read.return_route == ret_route
94
95 # Verify no result token emitted yet
96 assert len(collector.items) == 0
97
98
99class TestAC3_3DeferredReadSatisfaction:
100 """AC3.3: WRITE on WAITING cell satisfies deferred read — emits result token to stashed return route."""
101
102 def test_write_satisfies_deferred_read(self):
103 env = simpy.Environment()
104 sm = StructureMemory(env, 0, cell_count=512)
105
106 # Create collector store
107 collector = simpy.Store(env)
108 sm.route_table[0] = collector
109
110 # Set up deferred read on cell 30
111 ret_route = CMToken(target=0, offset=8, act_id=3, data=0)
112 read_token = SMToken(target=0, addr=30, op=MemOp.READ, flags=None, data=None, ret=ret_route)
113 inject_token(env, sm.input_store, read_token)
114
115 # Run to let deferred read be set up
116 env.run(until=10)
117
118 # Now inject WRITE to cell 30
119 write_token = SMToken(target=0, addr=30, op=MemOp.WRITE, flags=None, data=0xDEAD, ret=None)
120 inject_token(env, sm.input_store, write_token)
121
122 # Continue simulation
123 env.run(until=100)
124
125 # Verify cell is now FULL
126 assert sm.cells[30].pres == Presence.FULL
127 assert sm.cells[30].data_l == 0xDEAD
128
129 # Verify result token was emitted with written data
130 assert len(collector.items) == 1
131 result = collector.items[0]
132 assert result.data == 0xDEAD
133 assert result.offset == 8
134 assert result.act_id == 3
135
136 # Verify deferred_read is cleared
137 assert sm.deferred_read is None
138
139
140class TestAC3_4WriteOnEmptyOrReserved:
141 """AC3.4: WRITE on EMPTY/RESERVED sets cell to FULL."""
142
143 def test_write_on_empty_cell(self):
144 env = simpy.Environment()
145 sm = StructureMemory(env, 0, cell_count=512)
146
147 # Cell 40 starts EMPTY
148 assert sm.cells[40].pres == Presence.EMPTY
149
150 # Inject WRITE
151 write_token = SMToken(target=0, addr=40, op=MemOp.WRITE, flags=None, data=0xCAFE, ret=None)
152 inject_token(env, sm.input_store, write_token)
153
154 env.run(until=100)
155
156 # Verify cell is FULL with correct data
157 assert sm.cells[40].pres == Presence.FULL
158 assert sm.cells[40].data_l == 0xCAFE
159
160 def test_write_on_reserved_cell(self):
161 env = simpy.Environment()
162 sm = StructureMemory(env, 0, cell_count=512)
163
164 # Set cell 50 to RESERVED
165 sm.cells[50].pres = Presence.RESERVED
166
167 # Inject WRITE
168 write_token = SMToken(target=0, addr=50, op=MemOp.WRITE, flags=None, data=0xF00D, ret=None)
169 inject_token(env, sm.input_store, write_token)
170
171 env.run(until=100)
172
173 # Verify cell is FULL with correct data
174 assert sm.cells[50].pres == Presence.FULL
175 assert sm.cells[50].data_l == 0xF00D
176
177
178class TestAC3_5Clear:
179 """AC3.5: CLEAR sets cell to EMPTY, cancels deferred read if targeting that cell."""
180
181 def test_clear_sets_empty(self):
182 env = simpy.Environment()
183 sm = StructureMemory(env, 0, cell_count=512)
184
185 # Set cell 60 to FULL
186 sm.cells[60].pres = Presence.FULL
187 sm.cells[60].data_l = 0x1234
188
189 # Inject CLEAR
190 clear_token = SMToken(target=0, addr=60, op=MemOp.CLEAR, flags=None, data=None, ret=None)
191 inject_token(env, sm.input_store, clear_token)
192
193 env.run(until=100)
194
195 # Verify cell is EMPTY
196 assert sm.cells[60].pres == Presence.EMPTY
197 assert sm.cells[60].data_l is None
198
199 def test_clear_cancels_deferred_read(self):
200 env = simpy.Environment()
201 sm = StructureMemory(env, 0, cell_count=512)
202
203 # Create collector
204 collector = simpy.Store(env)
205 sm.route_table[0] = collector
206
207 # Set up deferred read on cell 70
208 ret_route = CMToken(target=0, offset=10, act_id=0, data=0)
209 read_token = SMToken(target=0, addr=70, op=MemOp.READ, flags=None, data=None, ret=ret_route)
210 inject_token(env, sm.input_store, read_token)
211
212 env.run(until=10)
213
214 # Now inject CLEAR on cell 70
215 clear_token = SMToken(target=0, addr=70, op=MemOp.CLEAR, flags=None, data=None, ret=None)
216 inject_token(env, sm.input_store, clear_token)
217
218 env.run(until=100)
219
220 # Verify cell is EMPTY
221 assert sm.cells[70].pres == Presence.EMPTY
222
223 # Verify deferred_read is cleared
224 assert sm.deferred_read is None
225
226 # Verify no result token was emitted (deferred read was cancelled)
227 assert len(collector.items) == 0
228
229
230class TestAC3_8WriteOnFull:
231 """AC3.8: WRITE on FULL cell overwrites data."""
232
233 def test_write_overwrites_full_cell(self):
234 env = simpy.Environment()
235 sm = StructureMemory(env, 0, cell_count=512)
236
237 # Set cell 80 to FULL with data X
238 sm.cells[80].pres = Presence.FULL
239 sm.cells[80].data_l = 0x5555
240
241 # Inject WRITE with data Y
242 write_token = SMToken(target=0, addr=80, op=MemOp.WRITE, flags=None, data=0xAAAA, ret=None)
243 inject_token(env, sm.input_store, write_token)
244
245 env.run(until=100)
246
247 # Verify cell still FULL but data is overwritten
248 assert sm.cells[80].pres == Presence.FULL
249 assert sm.cells[80].data_l == 0xAAAA
250
251
252class TestAC3_6AtomicOps:
253 """AC3.6: READ_INC/READ_DEC atomically modify and return value (lower 256 cells only)."""
254
255 def test_read_inc_returns_old_value(self):
256 env = simpy.Environment()
257 sm = StructureMemory(env, 0, cell_count=512)
258
259 # Set cell 100 to FULL with value 42
260 sm.cells[100].pres = Presence.FULL
261 sm.cells[100].data_l = 42
262
263 # Create collector
264 collector = simpy.Store(env)
265 sm.route_table[0] = collector
266
267 # Inject RD_INC
268 ret_route = CMToken(target=0, offset=11, act_id=0, data=0)
269 inc_token = SMToken(target=0, addr=100, op=MemOp.RD_INC, flags=None, data=None, ret=ret_route)
270 inject_token(env, sm.input_store, inc_token)
271
272 env.run(until=100)
273
274 # Verify cell was incremented
275 assert sm.cells[100].data_l == 43
276
277 # Verify result token has old value
278 assert len(collector.items) == 1
279 assert collector.items[0].data == 42
280
281 def test_read_inc_wraps_at_0xFFFF(self):
282 env = simpy.Environment()
283 sm = StructureMemory(env, 0, cell_count=512)
284
285 # Set cell 110 to FULL with max value
286 sm.cells[110].pres = Presence.FULL
287 sm.cells[110].data_l = 0xFFFF
288
289 # Create collector
290 collector = simpy.Store(env)
291 sm.route_table[0] = collector
292
293 # Inject RD_INC
294 ret_route = CMToken(target=0, offset=12, act_id=0, data=0)
295 inc_token = SMToken(target=0, addr=110, op=MemOp.RD_INC, flags=None, data=None, ret=ret_route)
296 inject_token(env, sm.input_store, inc_token)
297
298 env.run(until=100)
299
300 # Verify cell wrapped to 0
301 assert sm.cells[110].data_l == 0
302
303 # Verify result token has old value
304 assert len(collector.items) == 1
305 assert collector.items[0].data == 0xFFFF
306
307 def test_read_dec_returns_old_value(self):
308 env = simpy.Environment()
309 sm = StructureMemory(env, 0, cell_count=512)
310
311 # Set cell 120 to FULL with value 100
312 sm.cells[120].pres = Presence.FULL
313 sm.cells[120].data_l = 100
314
315 # Create collector
316 collector = simpy.Store(env)
317 sm.route_table[0] = collector
318
319 # Inject RD_DEC
320 ret_route = CMToken(target=0, offset=13, act_id=0, data=0)
321 dec_token = SMToken(target=0, addr=120, op=MemOp.RD_DEC, flags=None, data=None, ret=ret_route)
322 inject_token(env, sm.input_store, dec_token)
323
324 env.run(until=100)
325
326 # Verify cell was decremented
327 assert sm.cells[120].data_l == 99
328
329 # Verify result token has old value
330 assert len(collector.items) == 1
331 assert collector.items[0].data == 100
332
333 def test_atomic_op_rejected_on_non_full_cell(self):
334 env = simpy.Environment()
335 sm = StructureMemory(env, 0, cell_count=512)
336
337 # Cell 130 starts EMPTY
338 assert sm.cells[130].pres == Presence.EMPTY
339
340 # Inject RD_INC
341 ret_route = CMToken(target=0, offset=14, act_id=0, data=0)
342 inc_token = SMToken(target=0, addr=130, op=MemOp.RD_INC, flags=None, data=None, ret=ret_route)
343 inject_token(env, sm.input_store, inc_token)
344
345 # Create collector
346 collector = simpy.Store(env)
347 sm.route_table[0] = collector
348
349 env.run(until=100)
350
351 # Verify no result token emitted and cell still EMPTY
352 assert sm.cells[130].pres == Presence.EMPTY
353 assert len(collector.items) == 0
354
355 def test_atomic_op_rejected_on_addr_gte_256(self):
356 env = simpy.Environment()
357 sm = StructureMemory(env, 0, cell_count=512)
358
359 # Set cell 256 to FULL
360 sm.cells[256].pres = Presence.FULL
361 sm.cells[256].data_l = 50
362
363 # Create collector
364 collector = simpy.Store(env)
365 sm.route_table[0] = collector
366
367 # Inject RD_INC on cell 256 (should be rejected)
368 ret_route = CMToken(target=0, offset=15, act_id=0, data=0)
369 inc_token = SMToken(target=0, addr=256, op=MemOp.RD_INC, flags=None, data=None, ret=ret_route)
370 inject_token(env, sm.input_store, inc_token)
371
372 env.run(until=100)
373
374 # Verify no result token emitted and cell unchanged
375 assert sm.cells[256].data_l == 50
376 assert len(collector.items) == 0
377
378
379class TestAC3_7DepthOneConstraint:
380 """AC3.7: Depth-1 constraint - SM enforces only one outstanding deferred read."""
381
382 def test_two_blocking_reads_stall_and_unblock(self):
383 """
384 AC3.7: Inject READ A then READ B while deferred register is occupied.
385 SM stalls on second READ. WRITE A satisfies first, SM unblocks and
386 retries READ B. Then WRITE B satisfies second. Both results arrive.
387 """
388 env = simpy.Environment()
389 sm = StructureMemory(env, 0, cell_count=512)
390
391 collector = simpy.Store(env)
392 sm.route_table[0] = collector
393
394 ret_a = CMToken(target=0, offset=20, act_id=0, data=0)
395 ret_b = CMToken(target=0, offset=21, act_id=1, data=0)
396 read_a = SMToken(target=0, addr=140, op=MemOp.READ, flags=None, data=None, ret=ret_a)
397 read_b = SMToken(target=0, addr=150, op=MemOp.READ, flags=None, data=None, ret=ret_b)
398
399 # Inject both READs before any WRITE
400 inject_token(env, sm.input_store, read_a)
401 inject_token(env, sm.input_store, read_b)
402 env.run(until=10)
403
404 # Cell A is WAITING, deferred register holds A's route
405 assert sm.cells[140].pres == Presence.WAITING
406 assert sm.deferred_read is not None
407 assert sm.deferred_read.cell_addr == 140
408 # Cell B still EMPTY — SM stalled on second READ
409 assert sm.cells[150].pres == Presence.EMPTY
410
411 # Satisfy first deferred by writing to A
412 write_a = SMToken(target=0, addr=140, op=MemOp.WRITE, flags=None, data=0x1111, ret=None)
413 inject_token(env, sm.input_store, write_a)
414 env.run(until=50)
415
416 # SM unblocked, retried READ B → cell B now WAITING
417 assert sm.cells[150].pres == Presence.WAITING
418 assert sm.deferred_read is not None
419 assert sm.deferred_read.cell_addr == 150
420
421 # Satisfy second deferred
422 write_b = SMToken(target=0, addr=150, op=MemOp.WRITE, flags=None, data=0x2222, ret=None)
423 inject_token(env, sm.input_store, write_b)
424 env.run(until=200)
425
426 # Both results collected
427 assert len(collector.items) == 2
428 assert collector.items[0].data == 0x1111
429 assert collector.items[1].data == 0x2222
430 assert sm.cells[150].pres == Presence.FULL
431 assert sm.cells[150].data_l == 0x2222
432
433
434class TestAC3_9CAS:
435 """AC3.9: CAS on FULL cell with compare-and-swap semantics."""
436
437 def test_cas_match_swaps_value(self):
438 env = simpy.Environment()
439 sm = StructureMemory(env, 0, cell_count=512)
440
441 # Set cell 160 to FULL with value 10
442 sm.cells[160].pres = Presence.FULL
443 sm.cells[160].data_l = 10
444
445 # Create collector
446 collector = simpy.Store(env)
447 sm.route_table[0] = collector
448
449 # Inject CMP_SW with flags=10 (expected), data=99 (new)
450 ret_route = CMToken(target=0, offset=22, act_id=0, data=0)
451 cas_token = SMToken(target=0, addr=160, op=MemOp.CMP_SW, flags=10, data=99, ret=ret_route)
452 inject_token(env, sm.input_store, cas_token)
453
454 env.run(until=100)
455
456 # Verify cell now has new value
457 assert sm.cells[160].data_l == 99
458
459 # Verify result token has old value
460 assert len(collector.items) == 1
461 assert collector.items[0].data == 10
462
463 def test_cas_mismatch_no_swap(self):
464 env = simpy.Environment()
465 sm = StructureMemory(env, 0, cell_count=512)
466
467 # Set cell 170 to FULL with value 10
468 sm.cells[170].pres = Presence.FULL
469 sm.cells[170].data_l = 10
470
471 # Create collector
472 collector = simpy.Store(env)
473 sm.route_table[0] = collector
474
475 # Inject CMP_SW with flags=20 (mismatch), data=99 (new)
476 ret_route = CMToken(target=0, offset=23, act_id=0, data=0)
477 cas_token = SMToken(target=0, addr=170, op=MemOp.CMP_SW, flags=20, data=99, ret=ret_route)
478 inject_token(env, sm.input_store, cas_token)
479
480 env.run(until=100)
481
482 # Verify cell unchanged
483 assert sm.cells[170].data_l == 10
484
485 # Verify result token has old value
486 assert len(collector.items) == 1
487 assert collector.items[0].data == 10
488
489 def test_cas_rejected_on_addr_gte_256(self):
490 env = simpy.Environment()
491 sm = StructureMemory(env, 0, cell_count=512)
492
493 # Set cell 256 to FULL
494 sm.cells[256].pres = Presence.FULL
495 sm.cells[256].data_l = 10
496
497 # Create collector
498 collector = simpy.Store(env)
499 sm.route_table[0] = collector
500
501 # Inject CMP_SW on cell 256 (should be rejected)
502 ret_route = CMToken(target=0, offset=24, act_id=0, data=0)
503 cas_token = SMToken(target=0, addr=256, op=MemOp.CMP_SW, flags=10, data=99, ret=ret_route)
504 inject_token(env, sm.input_store, cas_token)
505
506 env.run(until=100)
507
508 # Verify no result token and cell unchanged
509 assert sm.cells[256].data_l == 10
510 assert len(collector.items) == 0
511
512
513class TestPresenceStateMachineInvariant:
514 """Property-based test: presence state machine invariant."""
515
516 @given(sm_token())
517 def test_valid_presence_state_after_operations(self, token):
518 """Verify that after any valid operation, cell is in a valid Presence state."""
519 env = simpy.Environment()
520 sm = StructureMemory(env, 0, cell_count=512)
521
522 # Create collector for results
523 collector = simpy.Store(env)
524 sm.route_table[0] = collector
525
526 # Ensure return route is valid
527 if token.ret is None:
528 token = SMToken(
529 target=token.target,
530 addr=token.addr,
531 op=token.op,
532 flags=token.flags,
533 data=token.data,
534 ret=CMToken(target=0, offset=0, act_id=0, data=0),
535 )
536
537 # Pre-populate target cell as FULL if it's an atomic operation
538 if token.op in (MemOp.RD_INC, MemOp.RD_DEC, MemOp.CMP_SW):
539 sm.cells[token.addr].pres = Presence.FULL
540 sm.cells[token.addr].data_l = 0x1234
541
542 inject_token(env, sm.input_store, token)
543
544 # Run simulation
545 env.run(until=100)
546
547 # Verify cell is in valid state and specific transitions per op type
548 cell = sm.cells[token.addr]
549 assert cell.pres in (Presence.EMPTY, Presence.RESERVED, Presence.FULL, Presence.WAITING)
550
551 # Verify specific state transitions per operation type
552 if token.op == MemOp.WRITE:
553 # WRITE should set cell to FULL
554 assert cell.pres == Presence.FULL
555 assert cell.data_l == token.data
556 elif token.op == MemOp.READ:
557 # READ on empty cell should set to WAITING
558 if cell.pres == Presence.WAITING:
559 assert sm.deferred_read is not None
560 else:
561 assert cell.pres in (Presence.EMPTY, Presence.FULL)
562 elif token.op == MemOp.CLEAR:
563 # CLEAR should set to EMPTY
564 assert cell.pres == Presence.EMPTY
565 elif token.op == MemOp.ALLOC:
566 # ALLOC on EMPTY should set to RESERVED
567 assert cell.pres == Presence.RESERVED
568 elif token.op in (MemOp.RD_INC, MemOp.RD_DEC):
569 # Atomic ops on FULL should stay FULL with modified data
570 assert cell.pres == Presence.FULL
571 elif token.op == MemOp.CMP_SW:
572 # CMP_SW on FULL should stay FULL (data may change)
573 assert cell.pres == Presence.FULL
574
575
576class TestWriteAlwaysSetsDataL:
577 """Property-based test: WRITE always sets data_l to the written value."""
578
579 @given(sm_token(op=MemOp.WRITE))
580 def test_write_sets_data_l(self, token):
581 """Verify WRITE operations always set data_l to the written value."""
582 env = simpy.Environment()
583 sm = StructureMemory(env, 0, cell_count=512)
584
585 # Inject token
586 inject_token(env, sm.input_store, token)
587
588 # Run simulation
589 env.run(until=100)
590
591 # Verify data_l is set to token.data
592 assert sm.cells[token.addr].data_l == token.data
593 assert sm.cells[token.addr].pres == Presence.FULL
594
595
596class TestWriteReadRoundtrip:
597 """Property-based test: WRITE→READ roundtrip always returns written value."""
598
599 @given(sm_token(op=MemOp.WRITE), sm_return_route())
600 def test_write_read_roundtrip(self, write_token, read_route):
601 """WRITE→READ roundtrip always returns written value."""
602 env = simpy.Environment()
603 sm = StructureMemory(env, 0, cell_count=512)
604
605 # Create collector for results
606 collector = simpy.Store(env)
607 sm.route_table[0] = collector
608
609 # Inject WRITE
610 inject_token(env, sm.input_store, write_token)
611 env.run(until=10)
612
613 # Now READ from same cell with return route
614 read_token = SMToken(
615 target=write_token.target,
616 addr=write_token.addr,
617 op=MemOp.READ,
618 flags=None,
619 data=None,
620 ret=read_route,
621 )
622 inject_token(env, sm.input_store, read_token)
623 env.run(until=100)
624
625 # Verify result token has written data
626 assert len(collector.items) == 1
627 result = collector.items[0]
628 assert result.data == write_token.data
629
630
631class TestClearAlwaysEmptifies:
632 """Property-based test: CLEAR on any state produces EMPTY."""
633
634 @given(sm_token(op=MemOp.CLEAR))
635 def test_clear_on_any_state(self, token):
636 """CLEAR on any state produces EMPTY."""
637 env = simpy.Environment()
638 sm = StructureMemory(env, 0, cell_count=512)
639
640 # Pre-set cell to a random state (FULL with data)
641 sm.cells[token.addr].pres = Presence.FULL
642 sm.cells[token.addr].data_l = 0xBEEF
643
644 # Inject CLEAR
645 inject_token(env, sm.input_store, token)
646 env.run(until=100)
647
648 # Verify cell is EMPTY
649 assert sm.cells[token.addr].pres == Presence.EMPTY
650 assert sm.cells[token.addr].data_l is None
651
652
653class TestRDIncDecRestores:
654 """Property-based test: RD_INC then RD_DEC restores original value (mod 16-bit)."""
655
656 @given(uint16, sm_return_route(), sm_return_route())
657 def test_rd_inc_dec_restores(self, original_value, return_route_inc, return_route_dec):
658 """RD_INC then RD_DEC restores original value."""
659 env = simpy.Environment()
660 sm = StructureMemory(env, 0, cell_count=512)
661
662 # Create collector for results
663 collector = simpy.Store(env)
664 sm.route_table[0] = collector
665
666 # Set cell to original value
667 sm.cells[100].pres = Presence.FULL
668 sm.cells[100].data_l = original_value
669
670 # Inject RD_INC
671 inc_token = SMToken(
672 target=0,
673 addr=100,
674 op=MemOp.RD_INC,
675 flags=None,
676 data=None,
677 ret=return_route_inc,
678 )
679 inject_token(env, sm.input_store, inc_token)
680 env.run(until=10)
681
682 # Cell now has (original_value + 1) & 0xFFFF
683 # Inject RD_DEC
684 dec_token = SMToken(
685 target=0,
686 addr=100,
687 op=MemOp.RD_DEC,
688 flags=None,
689 data=None,
690 ret=return_route_dec,
691 )
692 inject_token(env, sm.input_store, dec_token)
693 env.run(until=100)
694
695 # Cell should be back to original_value
696 assert sm.cells[100].data_l == original_value
697
698
699class TestAtomicOpsOnNonFullRejected:
700 """Property-based test: Atomic ops on non-FULL cells are rejected."""
701
702 @given(sm_token(op=MemOp.RD_INC), sm_return_route())
703 def test_rd_inc_on_non_full_rejected(self, token, return_route):
704 """RD_INC on non-FULL cell is rejected."""
705 env = simpy.Environment()
706 sm = StructureMemory(env, 0, cell_count=512)
707
708 # Create collector
709 collector = simpy.Store(env)
710 sm.route_table[0] = collector
711
712 # Ensure cell is EMPTY (not FULL)
713 assert sm.cells[token.addr].pres == Presence.EMPTY
714
715 # Create RD_INC token with return route
716 inc_token = SMToken(
717 target=token.target,
718 addr=token.addr,
719 op=MemOp.RD_INC,
720 flags=None,
721 data=None,
722 ret=return_route,
723 )
724
725 inject_token(env, sm.input_store, inc_token)
726 env.run(until=100)
727
728 # Verify no result token and cell still EMPTY
729 assert len(collector.items) == 0
730 assert sm.cells[token.addr].pres == Presence.EMPTY
731
732 @given(sm_token(op=MemOp.RD_DEC), sm_return_route())
733 def test_rd_dec_on_non_full_rejected(self, token, return_route):
734 """RD_DEC on non-FULL cell is rejected."""
735 env = simpy.Environment()
736 sm = StructureMemory(env, 0, cell_count=512)
737
738 # Create collector
739 collector = simpy.Store(env)
740 sm.route_table[0] = collector
741
742 # Ensure cell is EMPTY
743 assert sm.cells[token.addr].pres == Presence.EMPTY
744
745 # Create RD_DEC token
746 dec_token = SMToken(
747 target=token.target,
748 addr=token.addr,
749 op=MemOp.RD_DEC,
750 flags=None,
751 data=None,
752 ret=return_route,
753 )
754
755 inject_token(env, sm.input_store, dec_token)
756 env.run(until=100)
757
758 # Verify no result token and cell still EMPTY
759 assert len(collector.items) == 0
760 assert sm.cells[token.addr].pres == Presence.EMPTY
761
762 @given(sm_token(op=MemOp.CMP_SW), sm_return_route())
763 def test_cmp_sw_on_non_full_rejected(self, token, return_route):
764 """CMP_SW on non-FULL cell is rejected."""
765 env = simpy.Environment()
766 sm = StructureMemory(env, 0, cell_count=512)
767
768 # Create collector
769 collector = simpy.Store(env)
770 sm.route_table[0] = collector
771
772 # Ensure cell is EMPTY
773 assert sm.cells[token.addr].pres == Presence.EMPTY
774
775 # Create CMP_SW token
776 cas_token = SMToken(
777 target=token.target,
778 addr=token.addr,
779 op=MemOp.CMP_SW,
780 flags=0x1234,
781 data=0x5678,
782 ret=return_route,
783 )
784
785 inject_token(env, sm.input_store, cas_token)
786 env.run(until=100)
787
788 # Verify no result token and cell still EMPTY
789 assert len(collector.items) == 0
790 assert sm.cells[token.addr].pres == Presence.EMPTY
791
792
793class TestBoundaryEdgeCases:
794 """Test boundary and edge cases for SM operations."""
795
796 def test_rd_dec_wrapping_0x0000_to_0xffff(self):
797 """RD_DEC on 0x0000 wraps to 0xFFFF."""
798 env = simpy.Environment()
799 sm = StructureMemory(env, 0, cell_count=512)
800
801 # Set cell 150 to FULL with value 0
802 sm.cells[150].pres = Presence.FULL
803 sm.cells[150].data_l = 0x0000
804
805 # Create collector
806 collector = simpy.Store(env)
807 sm.route_table[0] = collector
808
809 # Inject RD_DEC
810 ret_route = CMToken(target=0, offset=16, act_id=0, data=0)
811 dec_token = SMToken(target=0, addr=150, op=MemOp.RD_DEC, flags=None, data=None, ret=ret_route)
812 inject_token(env, sm.input_store, dec_token)
813
814 env.run(until=100)
815
816 # Verify cell wrapped to 0xFFFF
817 assert sm.cells[150].data_l == 0xFFFF
818
819 # Verify result token has old value (0)
820 assert len(collector.items) == 1
821 assert collector.items[0].data == 0x0000
822
823 def test_alloc_empty_to_reserved(self):
824 """ALLOC changes cell from EMPTY to RESERVED."""
825 env = simpy.Environment()
826 sm = StructureMemory(env, 0, cell_count=512)
827
828 # Cell 200 starts EMPTY
829 assert sm.cells[200].pres == Presence.EMPTY
830
831 # Inject ALLOC
832 alloc_token = SMToken(target=0, addr=200, op=MemOp.ALLOC, flags=None, data=None, ret=None)
833 inject_token(env, sm.input_store, alloc_token)
834
835 env.run(until=100)
836
837 # Verify cell is now RESERVED
838 assert sm.cells[200].pres == Presence.RESERVED
839
840 def test_cmp_sw_on_non_full_cell_rejected(self):
841 """CMP_SW on non-FULL cell (EMPTY state) is rejected."""
842 env = simpy.Environment()
843 sm = StructureMemory(env, 0, cell_count=512)
844
845 # Create collector
846 collector = simpy.Store(env)
847 sm.route_table[0] = collector
848
849 # Cell 210 is EMPTY
850 assert sm.cells[210].pres == Presence.EMPTY
851
852 # Inject CMP_SW on EMPTY cell
853 ret_route = CMToken(target=0, offset=17, act_id=0, data=0)
854 cas_token = SMToken(
855 target=0,
856 addr=210,
857 op=MemOp.CMP_SW,
858 flags=0x1111,
859 data=0x2222,
860 ret=ret_route
861 )
862 inject_token(env, sm.input_store, cas_token)
863
864 env.run(until=100)
865
866 # Verify no result token and cell still EMPTY
867 assert len(collector.items) == 0
868 assert sm.cells[210].pres == Presence.EMPTY
869
870
871# Task 3: T0/T1 Tier Split and EXEC Opcode Tests
872