Allow mapping a device to non-contiguous addresses

Author: mnaberez

k0emu/bus.py

@@ -2,11 +2,13 @@
 
 
 class Bus(object):
+    ADDRESS_SPACE_SIZE = 2**16  # 64K address space
+
     def __init__(self, processor):
         self.processor = processor
         self._unmapped = BaseDevice("unmapped")
-        self._devices_by_address = [self._unmapped] * 0x10000
-        self._dev_bases_by_address = [0] * 0x10000
+        self._devices_by_address = [self._unmapped] * self.ADDRESS_SPACE_SIZE
+        self._device_registers_by_address = [0] * self.ADDRESS_SPACE_SIZE
         self._all_devices = set()
 
     def __getitem__(self, address):
@@ -17,12 +19,35 @@ def __setitem__(self, address, value):
 
     # device registration
 
-    def add_device(self, start, end, device):
-        """Register a device for an address range (inclusive).
-        The device sees local addresses 0 through (end - start)."""
-        for address in range(start, end + 1):
-            self._devices_by_address[address] = device
-            self._dev_bases_by_address[address] = start
+    def add_device(self, address_ranges, device):
+        """Register a device for one or more address ranges.
+
+        address_ranges: list of (start, end) tuples of bus addresses,
+                inclusive.  The device's registers are mapped sequentially
+                across the given ranges, allowing a single device model to
+                be mapped to non-contigous addresses in the memory map.
+        """
+        register = 0
+        for start, end in address_ranges:
+            if (start < 0) or (end >= self.ADDRESS_SPACE_SIZE) or (start > end):
+                raise ValueError("address range 0x%04X-0x%04X out of bounds" %
+                                 (start, end))
+
+            for address in range(start, end + 1):
+                existing_device = self._devices_by_address[address]
+                if existing_device is not self._unmapped:
+                    raise ValueError("address 0x%04X already mapped to %s" %
+                                     (address, existing_device.name))
+
+                self._devices_by_address[address] = device
+                self._device_registers_by_address[address] = register
+                register += 1
+
+        if register != device.size:
+            raise ValueError("%s: ranges cover %d bytes but device size is %d" %
+                             (device.name, register, device.size))
+
+        device.bus = self
         self._all_devices.add(device)
 
     def device(self, name):
@@ -46,10 +71,10 @@ def tick(self, cycles):
 
     def read(self, address):
         device = self._devices_by_address[address]
-        local_address = address - self._dev_bases_by_address[address]
-        return device.read(local_address)
+        register = self._device_registers_by_address[address]
+        return device.read(register)
 
     def write(self, address, value):
         device = self._devices_by_address[address]
-        local_address = address - self._dev_bases_by_address[address]
-        device.write(local_address, value)
+        register = self._device_registers_by_address[address]
+        device.write(register, value)

k0emu/devices.py

@@ -1,6 +1,7 @@
 class BaseDevice(object):
     def __init__(self, name):
         self.name = name
+        self.bus = None
         self.size = 0
         self.ticks = 0
 

k0emu/system.py

@@ -8,21 +8,18 @@ def make_processor():
     proc = Processor()
 
     rom = MemoryDevice("rom", size=0xF000, fill=0xFF, writable=False)
-    proc.bus.add_device(0x0000, 0xEFFF, rom)
+    proc.bus.add_device([(0x0000, 0xEFFF)], rom)
 
     expansion_ram = MemoryDevice("expansion_ram", size=0x0800)
-    proc.bus.add_device(0xF000, 0xF7FF, expansion_ram)
+    proc.bus.add_device([(0xF000, 0xF7FF)], expansion_ram)
 
     reserved = MemoryDevice("reserved", size=0x0300, fill=0x08, writable=False)
-    proc.bus.add_device(0xF800, 0xFAFF, reserved)
+    proc.bus.add_device([(0xF800, 0xFAFF)], reserved)
 
     high_speed_ram = MemoryDevice("high_speed_ram", size=0x03E0)
-    proc.bus.add_device(0xFB00, 0xFEDF, high_speed_ram)
+    proc.bus.add_device([(0xFB00, 0xFEDF)], high_speed_ram)
 
     register_file = RegisterFileDevice()
-    proc.bus.add_device(0xFEE0, 0xFEFF, register_file)
-
-    sfr = MemoryDevice("sfr", size=0x0100)
-    proc.bus.add_device(0xFF00, 0xFFFF, sfr)
+    proc.bus.add_device([(0xFEE0, 0xFEFF)], register_file)
 
     return proc

k0emu/tests/test_bus.py

@@ -19,15 +19,15 @@ def test_read_routes_to_device(self):
         proc = _FakeProcessor()
         bus = Bus(proc)
         mem = MemoryDevice("test", size=4)
-        bus.add_device(0x1000, 0x1003, mem)
+        bus.add_device([(0x1000, 0x1003)], mem)
         mem.write(2, 0x42)
         self.assertEqual(bus.read(0x1002), 0x42)
 
     def test_write_routes_to_device(self):
         proc = _FakeProcessor()
         bus = Bus(proc)
         mem = MemoryDevice("test", size=4)
-        bus.add_device(0x1000, 0x1003, mem)
+        bus.add_device([(0x1000, 0x1003)], mem)
         bus.write(0x1001, 0x42)
         self.assertEqual(mem.read(1), 0x42)
 
@@ -37,7 +37,7 @@ def test_device_sees_local_address(self):
         proc = _FakeProcessor()
         bus = Bus(proc)
         mem = MemoryDevice("test", size=4)
-        bus.add_device(0x8000, 0x8003, mem)
+        bus.add_device([(0x8000, 0x8003)], mem)
         bus.write(0x8003, 0x42)
         self.assertEqual(mem.read(3), 0x42)
 
@@ -60,30 +60,83 @@ def test_two_devices_at_different_ranges(self):
         bus = Bus(proc)
         mem_a = MemoryDevice("a", size=4)
         mem_b = MemoryDevice("b", size=4)
-        bus.add_device(0x1000, 0x1003, mem_a)
-        bus.add_device(0x2000, 0x2003, mem_b)
+        bus.add_device([(0x1000, 0x1003)], mem_a)
+        bus.add_device([(0x2000, 0x2003)], mem_b)
         bus.write(0x1000, 0x11)
         bus.write(0x2000, 0x22)
         self.assertEqual(bus.read(0x1000), 0x11)
         self.assertEqual(bus.read(0x2000), 0x22)
 
-    def test_later_device_overlays_earlier(self):
+    # registration validation
+
+    def test_overlapping_device_raises(self):
+        proc = _FakeProcessor()
+        bus = Bus(proc)
+        mem_a = MemoryDevice("a", size=4)
+        mem_b = MemoryDevice("b", size=2)
+        bus.add_device([(0x1000, 0x1003)], mem_a)
+        with self.assertRaises(ValueError):
+            bus.add_device([(0x1000, 0x1001)], mem_b)
+
+    def test_start_negative_raises(self):
+        proc = _FakeProcessor()
+        bus = Bus(proc)
+        mem = MemoryDevice("test", size=1)
+        with self.assertRaises(ValueError):
+            bus.add_device([(-1, -1)], mem)
+
+    def test_end_past_address_space_raises(self):
+        proc = _FakeProcessor()
+        bus = Bus(proc)
+        mem = MemoryDevice("test", size=2)
+        with self.assertRaises(ValueError):
+            bus.add_device([(0xFFFF, 0x10000)], mem)
+
+    def test_start_greater_than_end_raises(self):
         proc = _FakeProcessor()
         bus = Bus(proc)
-        mem_a = MemoryDevice("a", size=4, fill=0xAA)
-        mem_b = MemoryDevice("b", size=2, fill=0xBB)
-        bus.add_device(0x1000, 0x1003, mem_a)
-        bus.add_device(0x1000, 0x1001, mem_b)
-        self.assertEqual(bus.read(0x1000), 0xBB)  # overlaid by b
-        self.assertEqual(bus.read(0x1002), 0xAA)  # still a
+        mem = MemoryDevice("test", size=4)
+        with self.assertRaises(ValueError):
+            bus.add_device([(0x1003, 0x1000)], mem)
+
+    def test_range_size_mismatch_raises(self):
+        proc = _FakeProcessor()
+        bus = Bus(proc)
+        mem = MemoryDevice("test", size=4)
+        with self.assertRaises(ValueError):
+            bus.add_device([(0x1000, 0x1001)], mem)
+
+    # non-contiguous ranges
+
+    def test_non_contiguous_ranges_assign_sequential_registers(self):
+        proc = _FakeProcessor()
+        bus = Bus(proc)
+        mem = MemoryDevice("test", size=3)
+        bus.add_device([(0x1000, 0x1000), (0x2000, 0x2001)], mem)
+        bus.write(0x1000, 0xAA)  # register 0
+        bus.write(0x2000, 0xBB)  # register 1
+        bus.write(0x2001, 0xCC)  # register 2
+        self.assertEqual(mem.read(0), 0xAA)
+        self.assertEqual(mem.read(1), 0xBB)
+        self.assertEqual(mem.read(2), 0xCC)
+
+    # add_device sets bus reference
+
+    def test_add_device_sets_bus_on_device(self):
+        proc = _FakeProcessor()
+        bus = Bus(proc)
+        mem = MemoryDevice("test", size=4)
+        self.assertIsNone(mem.bus)
+        bus.add_device([(0x1000, 0x1003)], mem)
+        self.assertIs(mem.bus, bus)
 
     # reset
 
     def test_reset_resets_all_devices(self):
         proc = _FakeProcessor()
         bus = Bus(proc)
         mem = MemoryDevice("test", size=4)
-        bus.add_device(0x1000, 0x1003, mem)
+        bus.add_device([(0x1000, 0x1003)], mem)
         mem.ticks = 99
         bus.reset()
         # BaseDevice.reset() is a no-op, but processor.reset() is called
@@ -102,8 +155,8 @@ def test_tick_fans_out_to_all_devices(self):
         bus = Bus(proc)
         mem_a = MemoryDevice("a", size=4)
         mem_b = MemoryDevice("b", size=4)
-        bus.add_device(0x1000, 0x1003, mem_a)
-        bus.add_device(0x2000, 0x2003, mem_b)
+        bus.add_device([(0x1000, 0x1003)], mem_a)
+        bus.add_device([(0x2000, 0x2003)], mem_b)
         bus.tick(5)
         self.assertEqual(mem_a.ticks, 5)
         self.assertEqual(mem_b.ticks, 5)
@@ -112,7 +165,7 @@ def test_tick_accumulates(self):
         proc = _FakeProcessor()
         bus = Bus(proc)
         mem = MemoryDevice("test", size=4)
-        bus.add_device(0x1000, 0x1003, mem)
+        bus.add_device([(0x1000, 0x1003)], mem)
         bus.tick(3)
         bus.tick(7)
         self.assertEqual(mem.ticks, 10)
@@ -123,7 +176,7 @@ def test_device_finds_by_name(self):
         proc = _FakeProcessor()
         bus = Bus(proc)
         mem = MemoryDevice("my_ram", size=4)
-        bus.add_device(0x1000, 0x1003, mem)
+        bus.add_device([(0x1000, 0x1003)], mem)
         self.assertIs(bus.device("my_ram"), mem)
 
     def test_device_raises_for_unknown_name(self):
@@ -138,15 +191,15 @@ def test_getitem_reads_address(self):
         proc = _FakeProcessor()
         bus = Bus(proc)
         mem = MemoryDevice("test", size=4)
-        bus.add_device(0x1000, 0x1003, mem)
+        bus.add_device([(0x1000, 0x1003)], mem)
         mem.write(0, 0x42)
         self.assertEqual(bus[0x1000], 0x42)
 
     def test_setitem_writes_address(self):
         proc = _FakeProcessor()
         bus = Bus(proc)
         mem = MemoryDevice("test", size=4)
-        bus.add_device(0x1000, 0x1003, mem)
+        bus.add_device([(0x1000, 0x1003)], mem)
         bus[0x1001] = 0x42
         self.assertEqual(mem.read(1), 0x42)
 

k0emu/tests/test_devices.py

@@ -1,4 +1,5 @@
 import unittest
+from k0emu.bus import Bus
 from k0emu.devices import MemoryDevice, RegisterFileDevice
 
 
@@ -115,5 +116,3 @@ def test_out_of_bounds_raises(self):
         rf = RegisterFileDevice()
         with self.assertRaises(IndexError):
             rf.read(32)
-
-

k0emu/tests/test_processor.py

@@ -7,7 +7,7 @@
 def _make_processor():
     proc = Processor()
     mem = MemoryDevice("test_memory", size=0x10000)
-    proc.bus.add_device(0x0000, 0xFFFF, mem)
+    proc.bus.add_device([(0x0000, 0xFFFF)], mem)
     return proc, mem
 
 

k0emu/tests/test_processor_cycles.py

@@ -6,7 +6,7 @@
 def _make_processor():
     proc = Processor()
     mem = MemoryDevice("test_memory", size=0x10000)
-    proc.bus.add_device(0x0000, 0xFFFF, mem)
+    proc.bus.add_device([(0x0000, 0xFFFF)], mem)
     return proc
 
 

k0emu/tests/test_run.py

@@ -8,7 +8,7 @@
 def _make_runner(code):
     proc = Processor()
     mem = MemoryDevice("test_memory", size=0x10000)
-    proc.bus.add_device(0x0000, 0xFFFF, mem)
+    proc.bus.add_device([(0x0000, 0xFFFF)], mem)
     for i, byte in enumerate(code):
         mem.write(i, byte)
     output = io.StringIO()