37 Commits

Author SHA1 Message Date
Alex Forencich
1608af26e5 Release v0.1.16 2021-11-16 22:54:37 -08:00
Alex Forencich
31fb855311 Update readme 2021-11-16 22:40:42 -08:00
Alex Forencich
cb4b0e1738 Wrap access on RAM size 2021-11-16 17:13:18 -08:00
Alex Forencich
ea95eeaf0d Don't pass through extra positional args 2021-11-16 17:01:31 -08:00
Alex Forencich
079f4009b3 Rewrite RAM modules to use common slave implementation 2021-11-16 17:00:48 -08:00
Alex Forencich
612a94c97a Add AXI and AXI lite slave modules 2021-11-16 17:00:05 -08:00
Alex Forencich
757e3a6f2d AXI master modules extend Region 2021-11-16 16:58:50 -08:00
Alex Forencich
b9b9a2da72 Add address space abstraction 2021-11-16 16:55:53 -08:00
Alex Forencich
f7660e9038 Add buddy allocator 2021-11-16 16:53:40 -08:00
Alex Forencich
78693a63d5 Fix types 2021-11-10 23:59:11 -08:00
Alex Forencich
34498f6e5d Add write data type check 2021-11-10 23:49:13 -08:00
Alex Forencich
6329187ced Add address range checks 2021-11-10 23:48:47 -08:00
Alex Forencich
da24857dd2 Cast write data to bytes instead of bytearray 2021-11-10 23:45:46 -08:00
Alex Forencich
c08f22c710 Bring out address and ID signal widths 2021-11-10 21:56:08 -08:00
Alex Forencich
d874d91d05 Use typing.NamedTuple instead of collections.namedtuple to add __bytes__ cast 2021-11-10 21:49:58 -08:00
Alex Forencich
3fd016a84c Lazy logging 2021-11-09 00:53:37 -08:00
Alex Forencich
43de2ea9b0 Use getattr with default value when accessing optional signals 2021-11-09 00:46:37 -08:00
Alex Forencich
558ba51c91 Use correct transaction object 2021-11-09 00:13:19 -08:00
Alex Forencich
f6426bd8f3 Bump to dev version 2021-11-07 13:12:32 -08:00
Alex Forencich
74dd47ca99 Release v0.1.14 2021-11-07 12:39:42 -08:00
Reto Meier
cde2056bb0 Remove deprecated <= assignments
Starting from cocotb v1.6 the use of <= syntax has been deprecated. This
commit replaces all use of this syntax with the ``.value =`` syntax.
2021-10-25 18:27:18 +02:00
Alex Forencich
b6870716ed Fix active state tracking for AXI stream sink/monitor 2021-09-15 00:46:01 -07:00
Alex Forencich
44da562db9 Add cocotb framework classifier 2021-08-31 14:43:59 -07:00
Alex Forencich
8dcdbfefb8 Bump to dev version 2021-04-12 22:56:51 -07:00
Alex Forencich
b8919a095b Release v0.1.12 2021-04-12 22:46:32 -07:00
Alex Forencich
bc7edec289 Make resp and prot signals optional 2021-04-12 22:04:22 -07:00
Alex Forencich
e7c3a31eb0 Improve handling for optional signals 2021-04-12 21:24:33 -07:00
Alex Forencich
ce907ffbb9 Print out signal summary 2021-04-12 19:29:41 -07:00
Alex Forencich
95e2d5800d Store parameters 2021-04-12 19:27:38 -07:00
Alex Forencich
82853b31ff Rename byte_width to byte_lanes 2021-04-12 15:08:30 -07:00
Alex Forencich
8bbabd92df Update readme 2021-04-12 15:07:26 -07:00
Alex Forencich
c060f6c963 Transmit data without using pop 2021-04-12 13:53:28 -07:00
Alex Forencich
a767e00ce5 Transmit frames without using pop 2021-04-12 13:49:20 -07:00
Alex Forencich
d1d7313b98 Add tag context manager to AXI master to reuse per-ID processing components 2021-04-08 19:03:46 -07:00
Alex Forencich
01b43b97f2 Update readme 2021-04-08 19:01:27 -07:00
Alex Forencich
b5b6df84fe Improve burst handling in AXI master 2021-03-25 18:03:36 -07:00
Alex Forencich
babe69f4d3 Bump to dev version 2021-03-24 21:50:10 -07:00
21 changed files with 1786 additions and 958 deletions

103
README.md
View File

@@ -32,7 +32,7 @@ See the `tests` directory, [verilog-axi](https://github.com/alexforencich/verilo
### AXI and AXI lite master
The `AxiMaster` and `AxiLiteMaster` classes implement AXI masters and are capable of generating read and write operations against AXI slaves. Requested operations will be split and aligned according to the AXI specification. The `AxiMaster` module is capable of generating narrow bursts, handling multiple in-flight operations, and handling reordering and interleaving in responses across different transaction IDs.
The `AxiMaster` and `AxiLiteMaster` classes implement AXI masters and are capable of generating read and write operations against AXI slaves. Requested operations will be split and aligned according to the AXI specification. The `AxiMaster` module is capable of generating narrow bursts, handling multiple in-flight operations, and handling reordering and interleaving in responses across different transaction IDs. `AxiMaster` and `AxiLiteMaster` and related objects all extend `Region`, so they can be attached to `AddressSpace` objects to handle memory operations in the specified region.
The `AxiMaster` is a wrapper around `AxiMasterWrite` and `AxiMasterRead`. Similarly, `AxiLiteMaster` is a wrapper around `AxiLiteMasterWrite` and `AxiLiteMasterRead`. If a read-only or write-only interface is required instead of a full interface, use the corresponding read-only or write-only variant, the usage and API are exactly the same.
@@ -46,12 +46,12 @@ The first argument to the constructor accepts an `AxiBus` or `AxiLiteBus` object
Once the module is instantiated, read and write operations can be initiated in a couple of different ways.
First, blocking operations can be carried out with `read()` and `write()` and their associated word-access wrappers. Multiple concurrent operations started from different coroutines are handled correctly. For example:
First, operations can be carried out with async blocking `read()`, `write()`, and their associated word-access wrappers. Multiple concurrent operations started from different coroutines are handled correctly, with results returned in the order that the operations complete. For example:
await axi_master.write(0x0000, b'test')
data = await axi_master.read(0x0000, 4)
`read()` and `write()` return `namedtuple` objects containing _address_, _data_ or _length_, and _resp_. This is the preferred style, and this is the only style supported by the word-access wrappers.
Additional parameters can be specified to control sideband signals and burst settings. The transfer will be split into one or more bursts according to the AXI specification. All bursts generated from the same call to `read()` or `write()` will use the same ID, which will be automatically generated if not specified. `read()` and `write()` return `namedtuple` objects containing _address_, _data_ or _length_, and _resp_. This is the preferred style, and this is the only style supported by the word-access wrappers.
Alternatively, operations can be initiated with non-blocking `init_read()` and `init_write()`. These functions return `Event` objects which are triggered when the operation completes, and the result can be retrieved from `Event.data`. For example:
@@ -123,9 +123,39 @@ With this method, it is possible to start multiple concurrent operations from th
The `AxiBus`, `AxiLiteBus`, and related objects are containers for the interface signals. These hold instances of bus objects for the individual channels, which are currently extensions of `cocotb_bus.bus.Bus`. Class methods `from_entity` and `from_prefix` are provided to facilitate signal name matching. For AXI interfaces use `AxiBus`, `AxiReadBus`, or `AxiWriteBus`, as appropriate. For AXI lite interfaces, use `AxiLiteBus`, `AxiLiteReadBus`, or `AxiLiteWriteBus`, as appropriate.
### AXI and AXI lite slave
The `AxiSlave` and `AxiLiteSlave` classes implement AXI slaves and are capable of completing read and write operations from upstream AXI masters. The `AxiSlave` module is capable of handling narrow bursts. These modules can either be used to perform memory reads and writes on a `MemoryInterface` on behalf of the DUT, or they can be extended to implement customized functionality.
The `AxiSlave` is a wrapper around `AxiSlaveWrite` and `AxiSlaveRead`. Similarly, `AxiLiteSlave` is a wrapper around `AxiLiteSlaveWrite` and `AxiLiteSlaveRead`. If a read-only or write-only interface is required instead of a full interface, use the corresponding read-only or write-only variant, the usage and API are exactly the same.
To use these modules, import the one you need and connect it to the DUT:
from cocotbext.axi import AxiBus, AxiSlave, MemoryRegion
axi_slave = AxiSlave(AxiBus.from_prefix(dut, "m_axi"), dut.clk, dut.rst)
region = MemoryRegion(2**axi_slave.read_if.address_width)
axi_slave.target = region
The first argument to the constructor accepts an `AxiBus` or `AxiLiteBus` object. These objects are containers for the interface signals and include class methods to automate connections.
It is also possible to extend these modules; operation can be customized by overriding the internal `_read()` and `_write()` methods. See `AxiRam` and `AxiLiteRam` for an example.
#### `AxiSlave` and `AxiLiteSlave` constructor parameters
* _bus_: `AxiBus` or `AxiLiteBus` object containing AXI interface signals
* _clock_: clock signal
* _reset_: reset signal (optional)
* _reset_active_level_: reset active level (optional, default `True`)
* _target_: target region (optional, default `None`)
#### Attributes:
* _target_: target region
### AXI and AXI lite RAM
The `AxiRam` and `AxiLiteRam` classes implement AXI RAMs and are capable of completing read and write operations from upstream AXI masters. The `AxiRam` module is capable of handling narrow bursts.
The `AxiRam` and `AxiLiteRam` classes implement AXI RAMs and are capable of completing read and write operations from upstream AXI masters. The `AxiRam` module is capable of handling narrow bursts. These modules are extensions of the corresponding `AxiSlave` and `AxiLiteSlave` modules.
The `AxiRam` is a wrapper around `AxiRamWrite` and `AxiRamRead`. Similarly, `AxiLiteRam` is a wrapper around `AxiLiteRamWrite` and `AxiLiteRamRead`. If a read-only or write-only interface is required instead of a full interface, use the corresponding read-only or write-only variant, the usage and API are exactly the same.
@@ -141,6 +171,7 @@ Once the module is instantiated, the memory contents can be accessed in a couple
axi_ram.write(0x0000, b'test')
data = axi_ram.read(0x0000, 4)
axi_ram.hexdump(0x0000, 4, prefix="RAM")
Multi-port memories can be constructed by passing the `mem` object of the first instance to the other instances. For example, here is how to create a four-port RAM:
@@ -242,8 +273,8 @@ Note: _byte_size_, _byte_lanes_, `len(tdata)`, and `len(tkeep)` are all related,
* _pause_: stall the interface (deassert `tready` or `tvalid`) (source/sink only)
* _queue_occupancy_bytes_: number of bytes in queue (all)
* _queue_occupancy_frames_: number of frames in queue (all)
* _queue_occupancy_limit_bytes_: max number of bytes in queue allowed before tready deassert (sink only)
* _queue_occupancy_limit_frames_: max number of frames in queue allowed before tready deassert (sink only)
* _queue_occupancy_limit_bytes_: max number of bytes in queue allowed before backpressure is applied (source/sink only)
* _queue_occupancy_limit_frames_: max number of frames in queue allowed before backpressure is applied (source/sink only)
#### Methods
@@ -289,6 +320,66 @@ Methods:
* `normalize()`: pack `tkeep`, `tid`, `tdest`, and `tuser` to the same length as `tdata`, replicating last element if necessary, initialize `tkeep` to list of `1` and `tid`, `tdest`, and `tuser` to list of `0` if not specified.
* `compact()`: remove `tdata`, `tid`, `tdest`, and `tuser` values based on `tkeep`, remove `tkeep`, compact `tid`, `tdest`, and `tuser` to an int if all values are identical.
### Address space abstraction
The address space abstraction provides a framework for cross-connecting multiple memory-mapped interfaces for testing components that interface with complex systems, including components with DMA engines.
`MemoryInterface` is the base class for all components in the address space abstraction. `MemoryInterface` provides the core `read()` and `write()` methods, which implement bounds checking, as well as word-access wrappers. Methods for creating `Window` and `WindowPool` objects are also provided. The function `get_absolute_address()` translates addresses to the system address space. `MemoryInterface` can be extended to implement custom functionality by overriding `_read()` and `_write()`.
`Window` objects represent views onto a parent address space with some length and offset. `read()` and `write()` operations on a `Window` are translated to the equivalent operations on the parent address space. Multiple `Window` instances can overlap and access the same portion of address space.
`WindowPool` provides a method for dynamically allocating windows from a section of address space. It uses a standard memory management algorithm to provide naturally-aligned `Window` objects of the requested size.
`Region` is the base class for all components which implement a portion of address space. `Region` objects can be registered with `AddressSpace` objects to handle `read()` and `write()` operations in a specified region. `Region` can be extended by components that implement a portion of address space.
`MemoryRegion` is an extension of `Region` that uses an `mmap` instance to handle memory operations. `MemoryRegion` also provides hex dump methods as well as indexing and slicing.
`PeripheralRegion` is an extension of `Region` that can wrap another object that implements `read()` and `write()`, as an alternative to extending `Region`.
`AddressSpace` is the core object for handling address spaces. `Region` objects can be registered with `AddressSpace` with specified base address, size, and offset. The `AddressSpace` object will then direct `read()` and `write()` operations to the appropriate `Region`s, splitting requests appropriately when necessary and translating addresses. Regions registered with `offset` other than `None` are translated such that accesses to base address + N map to N + offset. Regions registered with an `offset` of `None` are not translated. `Region` objects registered with the same `AddressSpace` cannot overlap, however the same `Region` can be registered multiple times. `AddressSpace` also provides a method for creating `Pool` objects.
`Pool` is an extension of `AddressSpace` that supports dynamic allocation of `MemoryRegion`s. It uses a standard memory management algorithm to provide naturally-aligned `MemoryRegion` objects of the requested size.
#### Example
This is a simple example that shows how the address space abstraction components can be used to connect a DUT to a simulated host system, including simulated RAM, an AXI interface from the DUT for DMA, and an AXI lite interface to the DUT for control.
from cocotbext.axi import AddressSpace, MemoryRegion
from cocotbext.axi import AxiBus, AxiLiteMaster, AxiSlave
# system address space
address_space = AddressSpace(2**32)
# RAM
ram = MemoryRegion(2**24)
address_space.register_region(ram, 0x0000_0000)
ram_pool = address_space.create_window_pool(0x0000_0000, 2**20)
# DUT control register interface
axil_master = AxiLiteMaster(AxiLiteBus.from_prefix(dut, "s_axil_ctrl"), dut.clk, dut.rst)
address_space.register_region(axil_master, 0x8000_0000)
ctrl_regs = address_space.create_window(0x8000_0000, axil_master.size)
# DMA from DUT
axi_slave = AxiSlave(AxiBus.from_prefix(dut, "m_axi_dma"), dut.clk, dut.rst, target=address_space)
# exercise DUT DMA functionality
src_block = ram_pool.alloc_window(1024)
dst_block = ram_pool.alloc_window(1024)
test_data = b'test data'
await src_block.write(0, test_data)
await ctrl_regs.write_dword(DMA_SRC_ADDR, src_block.get_absolute_address(0))
await ctrl_regs.write_dword(DMA_DST_ADDR, dst_block.get_absolute_address(0))
await ctrl_regs.write_dword(DMA_LEN, len(test_data))
await ctrl_regs.write_dword(DMA_CONTROL, 1)
while await ctrl_regs.read_dword(DMA_STATUS) == 0:
pass
assert await dst_block.read(0, len(test_data)) == test_data
### AXI signals
* Write address channel

View File

@@ -26,14 +26,20 @@ from .version import __version__
from .constants import AxiBurstType, AxiBurstSize, AxiLockType, AxiCacheBit, AxiProt, AxiResp
from .address_space import MemoryInterface, Window, WindowPool
from .address_space import Region, MemoryRegion, PeripheralRegion
from .address_space import AddressSpace, Pool
from .axis import AxiStreamFrame, AxiStreamBus, AxiStreamSource, AxiStreamSink, AxiStreamMonitor
from .axil_channels import AxiLiteAWBus, AxiLiteWBus, AxiLiteBBus, AxiLiteARBus, AxiLiteRBus
from .axil_channels import AxiLiteWriteBus, AxiLiteReadBus, AxiLiteBus
from .axil_master import AxiLiteMasterWrite, AxiLiteMasterRead, AxiLiteMaster
from .axil_slave import AxiLiteSlaveWrite, AxiLiteSlaveRead, AxiLiteSlave
from .axil_ram import AxiLiteRamWrite, AxiLiteRamRead, AxiLiteRam
from .axi_channels import AxiAWBus, AxiWBus, AxiBBus, AxiARBus, AxiRBus
from .axi_channels import AxiWriteBus, AxiReadBus, AxiBus
from .axi_master import AxiMasterWrite, AxiMasterRead, AxiMaster
from .axi_slave import AxiSlaveWrite, AxiSlaveRead, AxiSlave
from .axi_ram import AxiRamWrite, AxiRamRead, AxiRam

View File

@@ -0,0 +1,320 @@
"""
Copyright (c) 2021 Alex Forencich
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""
import mmap
from .buddy_allocator import BuddyAllocator
from .utils import hexdump, hexdump_lines, hexdump_str
class MemoryInterface:
def __init__(self, size, base=0, parent=None, **kwargs):
self._parent = parent
self._size = size
self._base = base
super().__init__(**kwargs)
@property
def parent(self):
return self._parent
@property
def size(self):
return self._size
@property
def base(self):
return self._base
def check_range(self, address, length=0):
if address < 0 or address >= self.size:
raise ValueError("address out of range")
if length < 0:
raise ValueError("invalid length")
if address+length > self.size:
raise ValueError("operation out of range")
def get_absolute_address(self, address):
if self.base is None:
return None
self.check_range(address)
return address+self.base
async def _read(self, address, length, **kwargs):
raise NotImplementedError()
async def read(self, address, length, **kwargs):
self.check_range(address, length)
return await self._read(address, length, **kwargs)
async def read_words(self, address, count, byteorder='little', ws=2, **kwargs):
data = bytes(await self.read(address, count*ws, **kwargs))
words = []
for k in range(count):
words.append(int.from_bytes(data[ws*k:ws*(k+1)], byteorder))
return words
async def read_dwords(self, address, count, byteorder='little', **kwargs):
return await self.read_words(address, count, byteorder, 4, **kwargs)
async def read_qwords(self, address, count, byteorder='little', **kwargs):
return await self.read_words(address, count, byteorder, 8, **kwargs)
async def read_byte(self, address, **kwargs):
return (await self.read(address, 1, **kwargs)).data[0]
async def read_word(self, address, byteorder='little', ws=2, **kwargs):
return (await self.read_words(address, 1, byteorder, ws, **kwargs))[0]
async def read_dword(self, address, byteorder='little', **kwargs):
return (await self.read_dwords(address, 1, byteorder, **kwargs))[0]
async def read_qword(self, address, byteorder='little', **kwargs):
return (await self.read_qwords(address, 1, byteorder, **kwargs))[0]
async def _write(self, address, data, **kwargs):
raise NotImplementedError()
async def write(self, address, data, **kwargs):
self.check_range(address, len(data))
await self._write(address, data, **kwargs)
async def write_words(self, address, data, byteorder='little', ws=2, **kwargs):
words = data
data = bytearray()
for w in words:
data.extend(w.to_bytes(ws, byteorder))
await self.write(address, data, **kwargs)
async def write_dwords(self, address, data, byteorder='little', **kwargs):
await self.write_words(address, data, byteorder, 4, **kwargs)
async def write_qwords(self, address, data, byteorder='little', **kwargs):
await self.write_words(address, data, byteorder, 8, **kwargs)
async def write_byte(self, address, data, **kwargs):
await self.write(address, [data], **kwargs)
async def write_word(self, address, data, byteorder='little', ws=2, **kwargs):
await self.write_words(address, [data], byteorder, ws, **kwargs)
async def write_dword(self, address, data, byteorder='little', **kwargs):
await self.write_dwords(address, [data], byteorder, **kwargs)
async def write_qword(self, address, data, byteorder='little', **kwargs):
await self.write_qwords(address, [data], byteorder, **kwargs)
def create_window(self, offset, size):
self.check_range(offset, size)
return Window(self, offset, size, base=self.get_absolute_address(offset))
def create_window_pool(self, offset=None, size=None):
if offset is None:
offset = 0
if size is None:
size = self.size - offset
self.check_range(offset, size)
return WindowPool(self, offset, size, base=self.get_absolute_address(offset))
def __len__(self):
return self._size
class Window(MemoryInterface):
def __init__(self, parent, offset, size, base=0, **kwargs):
super().__init__(size, base=base, parent=parent, **kwargs)
self._offset = offset
@property
def offset(self):
return self._offset
def get_parent_address(self, address):
if address < 0 or address >= self.size:
raise ValueError("address out of range")
return address+self.offset
async def _read(self, address, length, **kwargs):
return await self.parent.read(self.get_parent_address(address), length, **kwargs)
async def _write(self, address, data, **kwargs):
await self.parent.write(self.get_parent_address(address), data, **kwargs)
class WindowPool(Window):
def __init__(self, parent, offset, size, base=None, **kwargs):
super().__init__(parent, offset, size, base=base, **kwargs)
self.allocator = BuddyAllocator(size)
def alloc_window(self, size):
return self.create_window(self.allocator.alloc(size), size)
class Region(MemoryInterface):
def __init__(self, size, **kwargs):
super().__init__(size, **kwargs)
class MemoryRegion(Region):
def __init__(self, size=4096, mem=None, **kwargs):
super().__init__(size, **kwargs)
if mem is None:
mem = mmap.mmap(-1, size)
self.mem = mem
async def _read(self, address, length, **kwargs):
return self.mem[address:address+length]
async def _write(self, address, data, **kwargs):
self.mem[address:address+len(data)] = data
def hexdump(self, address, length, prefix=""):
hexdump(self.mem[address:address+length], prefix=prefix, offset=address)
def hexdump_lines(self, address, length, prefix=""):
return hexdump_lines(self.mem[address:address+length], prefix=prefix, offset=address)
def hexdump_str(self, address, length, prefix=""):
return hexdump_str(self.mem[address:address+length], prefix=prefix, offset=address)
def __getitem__(self, key):
return self.mem[key]
def __setitem__(self, key, value):
self.mem[key] = value
def __bytes__(self):
return bytes(self.mem)
class PeripheralRegion(Region):
def __init__(self, obj, size, **kwargs):
super().__init__(size, **kwargs)
self.obj = obj
async def _read(self, address, length, **kwargs):
try:
return await self.obj.read(address, length, **kwargs)
except TypeError:
return self.obj.read(address, length, **kwargs)
async def _write(self, address, data, **kwargs):
try:
await self.obj.write(address, data, **kwargs)
except TypeError:
self.obj.write(address, data, **kwargs)
class AddressSpace(Region):
def __init__(self, size=2**64, base=0, parent=None, **kwargs):
super().__init__(size=size, base=base, parent=parent, **kwargs)
self.regions = []
def find_regions(self, address, length=1):
regions = []
if address < 0 or address >= self.size:
raise ValueError("address out of range")
if length < 0:
raise ValueError("invalid length")
length = max(length, 1)
for (base, size, translate, region) in self.regions:
if address < base+size and base < address+length:
regions.append((base, size, translate, region))
regions.sort()
return regions
def register_region(self, region, base, size=None, offset=0):
if size is None:
size = region.size
if self.find_regions(base, size):
raise ValueError("overlaps existing region")
region._parent = self
if offset == 0:
region._base = self.get_absolute_address(base)
else:
region._base = None
self.regions.append((base, size, offset, region))
async def read(self, address, length, **kwargs):
regions = self.find_regions(address, length)
data = bytearray()
if not regions:
raise Exception("Invalid address")
for base, size, offset, region in regions:
if base > address:
raise Exception("Invalid address")
seg_addr = address - base
seg_len = min(size-seg_addr, length)
if offset is None:
seg_addr = address
offset = 0
data.extend(bytes(await region.read(seg_addr+offset, seg_len, **kwargs)))
address += seg_len
length -= seg_len
if length > 0:
raise Exception("Invalid address")
return bytes(data)
async def write(self, address, data, **kwargs):
start = 0
length = len(data)
regions = self.find_regions(address, length)
if not regions:
raise Exception("Invalid address")
for base, size, offset, region in regions:
if base > address:
raise Exception("Invalid address")
seg_addr = address - base
seg_len = min(size-seg_addr, length)
if offset is None:
seg_addr = address
offset = 0
await region.write(seg_addr+offset, data[start:start+seg_len], **kwargs)
address += seg_len
start += seg_len
length -= seg_len
if length > 0:
raise Exception("Invalid address")
def create_pool(self, base=None, size=None):
if base is None:
base = 0
if size is None:
size = self.size - base
self.check_range(base, size)
pool = Pool(self, base, size)
self.register_region(pool, base, size)
return pool
class Pool(AddressSpace):
def __init__(self, parent, base, size, **kwargs):
super().__init__(parent=parent, base=base, size=size, **kwargs)
self.allocator = BuddyAllocator(size)
def alloc_region(self, size):
base = self.allocator.alloc(size)
region = MemoryRegion(size)
self.register_region(region, base)
return region

View File

@@ -26,8 +26,8 @@ from .stream import define_stream
# Write address channel
AxiAWBus, AxiAWTransaction, AxiAWSource, AxiAWSink, AxiAWMonitor = define_stream("AxiAW",
signals=["awid", "awaddr", "awlen", "awsize", "awburst", "awprot", "awvalid", "awready"],
optional_signals=["awlock", "awcache", "awqos", "awregion", "awuser"],
signals=["awid", "awaddr", "awlen", "awsize", "awburst", "awvalid", "awready"],
optional_signals=["awlock", "awcache", "awprot", "awqos", "awregion", "awuser"],
signal_widths={"awlen": 8, "awsize": 3, "awburst": 2, "awlock": 1,
"awcache": 4, "awprot": 3, "awqos": 4, "awregion": 4}
)
@@ -41,23 +41,23 @@ AxiWBus, AxiWTransaction, AxiWSource, AxiWSink, AxiWMonitor = define_stream("Axi
# Write response channel
AxiBBus, AxiBTransaction, AxiBSource, AxiBSink, AxiBMonitor = define_stream("AxiB",
signals=["bid", "bresp", "bvalid", "bready"],
optional_signals=["buser"],
signals=["bid", "bvalid", "bready"],
optional_signals=["bresp", "buser"],
signal_widths={"bresp": 2}
)
# Read address channel
AxiARBus, AxiARTransaction, AxiARSource, AxiARSink, AxiARMonitor = define_stream("AxiAR",
signals=["arid", "araddr", "arlen", "arsize", "arburst", "arprot", "arvalid", "arready"],
optional_signals=["arlock", "arcache", "arqos", "arregion", "aruser"],
signals=["arid", "araddr", "arlen", "arsize", "arburst", "arvalid", "arready"],
optional_signals=["arlock", "arcache", "arprot", "arqos", "arregion", "aruser"],
signal_widths={"arlen": 8, "arsize": 3, "arburst": 2, "arlock": 1,
"arcache": 4, "arprot": 3, "arqos": 4, "arregion": 4}
)
# Read data channel
AxiRBus, AxiRTransaction, AxiRSource, AxiRSink, AxiRMonitor = define_stream("AxiR",
signals=["rid", "rdata", "rresp", "rlast", "rvalid", "rready"],
optional_signals=["ruser"],
signals=["rid", "rdata", "rlast", "rvalid", "rready"],
optional_signals=["rresp", "ruser"],
signal_widths={"rresp": 2, "rlast": 1}
)

View File

@@ -23,7 +23,8 @@ THE SOFTWARE.
"""
import logging
from collections import namedtuple, Counter
from collections import Counter
from typing import List, NamedTuple, Union
import cocotb
from cocotb.queue import Queue
@@ -32,25 +33,171 @@ from cocotb.triggers import Event
from .version import __version__
from .constants import AxiBurstType, AxiLockType, AxiProt, AxiResp
from .axi_channels import AxiAWSource, AxiWSource, AxiBSink, AxiARSource, AxiRSink
from .address_space import Region
from .reset import Reset
# AXI master write helper objects
AxiWriteCmd = namedtuple("AxiWriteCmd", ["address", "data", "awid", "burst", "size",
"lock", "cache", "prot", "qos", "region", "user", "wuser", "event"])
AxiWriteRespCmd = namedtuple("AxiWriteRespCmd", ["address", "length", "size", "cycles",
"prot", "burst_list", "event"])
AxiWriteResp = namedtuple("AxiWriteResp", ["address", "length", "resp", "user"])
class AxiWriteCmd(NamedTuple):
address: int
data: bytes
awid: int
burst: AxiBurstType
size: int
lock: AxiLockType
cache: int
prot: AxiProt
qos: int
region: int
user: int
wuser: Union[list, int, None]
event: Event
class AxiWriteRespCmd(NamedTuple):
address: int
length: int
size: int
cycles: int
prot: AxiProt
burst_list: List[int]
event: Event
class AxiWriteResp(NamedTuple):
address: int
length: int
resp: AxiResp
user: Union[list, None]
# AXI master read helper objects
AxiReadCmd = namedtuple("AxiReadCmd", ["address", "length", "arid", "burst", "size",
"lock", "cache", "prot", "qos", "region", "user", "event"])
AxiReadRespCmd = namedtuple("AxiReadRespCmd", ["address", "length", "size", "cycles",
"prot", "burst_list", "event"])
AxiReadResp = namedtuple("AxiReadResp", ["address", "data", "resp", "user"])
class AxiReadCmd(NamedTuple):
address: int
length: int
arid: int
burst: AxiBurstType
size: int
lock: AxiLockType
cache: int
prot: AxiProt
qos: int
region: int
user: int
event: Event
class AxiMasterWrite(Reset):
def __init__(self, bus, clock, reset=None, reset_active_level=True, max_burst_len=256):
class AxiReadRespCmd(NamedTuple):
address: int
length: int
size: int
cycles: int
prot: AxiProt
burst_list: List[int]
event: Event
class AxiReadResp(NamedTuple):
address: int
data: bytes
resp: AxiResp
user: Union[list, None]
def __bytes__(self):
return self.data
class TagContext:
def __init__(self, manager):
self.current_tag = 0
self._cmd_queue = Queue()
self._current_cmd = None
self._resp_queue = Queue()
self._cr = None
self._manager = manager
async def get_resp(self):
return await self._resp_queue.get()
def get_resp_nowait(self):
return self._resp_queue.get_nowait()
def _start(self):
if self._cr is None:
self._cr = cocotb.fork(self._process_queue())
def _flush(self):
flushed_cmds = []
if self._cr is not None:
self._cr.kill()
self._cr = None
self._manager._set_idle(self)
if self._current_cmd is not None:
flushed_cmds.append(self._current_cmd)
self._current_cmd = None
while not self._cmd_queue.empty():
flushed_cmds.append(self._cmd_queue.get_nowait())
while not self._resp_queue.empty():
self._resp_queue.get_nowait()
return flushed_cmds
async def _process_queue(self):
while True:
cmd = await self._cmd_queue.get()
self._current_cmd = cmd
await self._manager._process(self, cmd)
self._current_cmd = None
if self._cmd_queue.empty() and self._resp_queue.empty():
self._manager._set_idle(self)
class TagContextManager:
def __init__(self, process):
self._context_list = []
self._context_idle_list = []
self._context_mapping = {}
self._process = process
def _get_context(self, tag):
if tag in self._context_mapping:
return self._context_mapping[tag]
elif self._context_idle_list:
context = self._context_idle_list.pop()
else:
context = TagContext(self)
self._context_list.append(context)
context._start()
context.current_tag = tag
self._context_mapping[tag] = context
return context
def start_cmd(self, tag, cmd):
context = self._get_context(tag)
context._cmd_queue.put_nowait(cmd)
def put_resp(self, tag, resp):
context = self._get_context(tag)
context._resp_queue.put_nowait(resp)
def _set_idle(self, context):
if context.current_tag in self._context_mapping:
del self._context_mapping[context.current_tag]
self._context_idle_list.append(context)
context.current_tag = None
def flush(self):
flushed_cmds = []
for c in self._context_list:
flushed_cmds.extend(c._flush())
return flushed_cmds
class AxiMasterWrite(Region, Reset):
def __init__(self, bus, clock, reset=None, reset_active_level=True, max_burst_len=256, **kwargs):
self.bus = bus
self.clock = clock
self.reset = reset
self.log = logging.getLogger(f"cocotb.{bus.aw._entity._name}.{bus.aw._name}")
self.log.info("AXI master (write)")
@@ -72,39 +219,57 @@ class AxiMasterWrite(Reset):
self.cur_id = 0
self.active_id = Counter()
self.int_write_resp_command_queue = [Queue() for k in range(self.id_count)]
self.current_write_resp_command = [None for k in range(self.id_count)]
self.int_write_resp_queue_list = [Queue() for k in range(self.id_count)]
self.tag_context_manager = TagContextManager(self._process_write_resp_id)
self.in_flight_operations = 0
self._idle = Event()
self._idle.set()
self.address_width = len(self.aw_channel.bus.awaddr)
self.id_width = len(self.aw_channel.bus.awid)
self.width = len(self.w_channel.bus.wdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.strb_mask = 2**self.byte_width-1
self.byte_lanes = self.width // self.byte_size
self.strb_mask = 2**self.byte_lanes-1
self.max_burst_len = max(min(max_burst_len, 256), 1)
self.max_burst_size = (self.byte_width-1).bit_length()
self.max_burst_size = (self.byte_lanes-1).bit_length()
self.awlock_present = hasattr(self.bus.aw, "awlock")
self.awcache_present = hasattr(self.bus.aw, "awcache")
self.awprot_present = hasattr(self.bus.aw, "awprot")
self.awqos_present = hasattr(self.bus.aw, "awqos")
self.awregion_present = hasattr(self.bus.aw, "awregion")
self.awuser_present = hasattr(self.bus.aw, "awuser")
self.wuser_present = hasattr(self.bus.w, "wuser")
self.buser_present = hasattr(self.bus.b, "buser")
super().__init__(2**self.address_width, **kwargs)
self.log.info("AXI master configuration:")
self.log.info(" Address width: %d bits", len(self.aw_channel.bus.awaddr))
self.log.info(" ID width: %d bits", len(self.aw_channel.bus.awid))
self.log.info(" Address width: %d bits", self.address_width)
self.log.info(" ID width: %d bits", self.id_width)
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_width)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_lanes)
self.log.info(" Max burst size: %d (%d bytes)", self.max_burst_size, 2**self.max_burst_size)
self.log.info(" Max burst length: %d cycles (%d bytes)",
self.max_burst_len, self.max_burst_len*self.byte_width)
self.max_burst_len, self.max_burst_len*self.byte_lanes)
assert self.byte_width == len(self.w_channel.bus.wstrb)
assert self.byte_width * self.byte_size == self.width
self.log.info("AXI master signals:")
for bus in (self.bus.aw, self.bus.w, self.bus.b):
for sig in sorted(list(set().union(bus._signals, bus._optional_signals))):
if hasattr(bus, sig):
self.log.info(" %s width: %d bits", sig, len(getattr(bus, sig)))
else:
self.log.info(" %s: not present", sig)
assert self.byte_lanes == len(self.w_channel.bus.wstrb)
assert self.byte_lanes * self.byte_size == self.width
assert len(self.b_channel.bus.bid) == len(self.aw_channel.bus.awid)
self._process_write_cr = None
self._process_write_resp_cr = None
self._process_write_resp_id_cr = None
self._init_reset(reset, reset_active_level)
@@ -117,6 +282,12 @@ class AxiMasterWrite(Reset):
if not isinstance(event, Event):
raise ValueError("Expected event object")
if address < 0 or address >= 2**self.address_width:
raise ValueError("Address out of range")
if isinstance(data, int):
raise ValueError("Expected bytes or bytearray for data")
if awid is None or awid < 0:
awid = None
elif awid > self.id_count:
@@ -132,6 +303,27 @@ class AxiMasterWrite(Reset):
lock = AxiLockType(lock)
prot = AxiProt(prot)
if not self.awlock_present and lock != AxiLockType.NORMAL:
raise ValueError("awlock sideband signal value specified, but signal is not connected")
if not self.awcache_present and cache != 0b0011:
raise ValueError("awcache sideband signal value specified, but signal is not connected")
if not self.awprot_present and prot != AxiProt.NONSECURE:
raise ValueError("awprot sideband signal value specified, but signal is not connected")
if not self.awqos_present and qos != 0:
raise ValueError("awqos sideband signal value specified, but signal is not connected")
if not self.awregion_present and region != 0:
raise ValueError("awregion sideband signal value specified, but signal is not connected")
if not self.awuser_present and user != 0:
raise ValueError("awuser sideband signal value specified, but signal is not connected")
if not self.wuser_present and wuser != 0:
raise ValueError("wuser sideband signal value specified, but signal is not connected")
if wuser is None:
wuser = 0
elif isinstance(wuser, int):
@@ -142,7 +334,7 @@ class AxiMasterWrite(Reset):
self.in_flight_operations += 1
self._idle.clear()
cmd = AxiWriteCmd(address, bytearray(data), awid, burst, size, lock,
cmd = AxiWriteCmd(address, bytes(data), awid, burst, size, lock,
cache, prot, qos, region, user, wuser, event)
self.write_command_queue.put_nowait(cmd)
@@ -161,43 +353,6 @@ class AxiMasterWrite(Reset):
await event.wait()
return event.data
async def write_words(self, address, data, byteorder='little', ws=2, awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
words = data
data = bytearray()
for w in words:
data.extend(w.to_bytes(ws, byteorder))
await self.write(address, data, awid, burst, size, lock, cache, prot, qos, region, user, wuser)
async def write_dwords(self, address, data, byteorder='little', awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
await self.write_words(address, data, byteorder, 4, awid, burst, size,
lock, cache, prot, qos, region, user, wuser)
async def write_qwords(self, address, data, byteorder='little', awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
await self.write_words(address, data, byteorder, 8, awid, burst, size,
lock, cache, prot, qos, region, user, wuser)
async def write_byte(self, address, data, awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
await self.write(address, [data], awid, burst, size, lock, cache, prot, qos, region, user, wuser)
async def write_word(self, address, data, byteorder='little', ws=2, awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
await self.write_words(address, [data], byteorder, ws, awid, burst, size,
lock, cache, prot, qos, region, user, wuser)
async def write_dword(self, address, data, byteorder='little', awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
await self.write_dwords(address, [data], byteorder, awid, burst, size,
lock, cache, prot, qos, region, user, wuser)
async def write_qword(self, address, data, byteorder='little', awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
await self.write_qwords(address, [data], byteorder, awid, burst, size,
lock, cache, prot, qos, region, user, wuser)
def _handle_reset(self, state):
if state:
self.log.info("Reset asserted")
@@ -207,10 +362,6 @@ class AxiMasterWrite(Reset):
if self._process_write_resp_cr is not None:
self._process_write_resp_cr.kill()
self._process_write_resp_cr = None
if self._process_write_resp_id_cr is not None:
for cr in self._process_write_resp_id_cr:
cr.kill()
self._process_write_resp_id_cr = None
self.aw_channel.clear()
self.w_channel.clear()
@@ -230,21 +381,9 @@ class AxiMasterWrite(Reset):
self.current_write_command = None
flush_cmd(cmd)
for q in self.int_write_resp_command_queue:
while not q.empty():
cmd = q.get_nowait()
for cmd in self.tag_context_manager.flush():
flush_cmd(cmd)
for k in range(len(self.current_write_resp_command)):
if self.current_write_resp_command[k]:
cmd = self.current_write_resp_command[k]
self.current_write_resp_command[k] = None
flush_cmd(cmd)
for q in self.int_write_resp_queue_list:
while not q.empty():
q.get_nowait()
self.cur_id = 0
self.active_id = Counter()
@@ -256,8 +395,6 @@ class AxiMasterWrite(Reset):
self._process_write_cr = cocotb.fork(self._process_write())
if self._process_write_resp_cr is None:
self._process_write_resp_cr = cocotb.fork(self._process_write_resp())
if self._process_write_resp_id_cr is None:
self._process_write_resp_id_cr = [cocotb.fork(self._process_write_resp_id(i)) for i in range(self.id_count)]
async def _process_write(self):
while True:
@@ -267,10 +404,10 @@ class AxiMasterWrite(Reset):
num_bytes = 2**cmd.size
aligned_addr = (cmd.address // num_bytes) * num_bytes
word_addr = (cmd.address // self.byte_width) * self.byte_width
word_addr = (cmd.address // self.byte_lanes) * self.byte_lanes
start_offset = cmd.address % self.byte_width
end_offset = ((cmd.address + len(cmd.data) - 1) % self.byte_width) + 1
start_offset = cmd.address % self.byte_lanes
end_offset = ((cmd.address + len(cmd.data) - 1) % self.byte_lanes) + 1
cycles = (len(cmd.data) + (cmd.address % num_bytes) + num_bytes-1) // num_bytes
@@ -291,6 +428,7 @@ class AxiMasterWrite(Reset):
wuser = cmd.wuser
if self.log.isEnabledFor(logging.INFO):
self.log.info("Write start addr: 0x%08x awid: 0x%x prot: %s data: %s",
cmd.address, awid, cmd.prot, ' '.join((f'{c:02x}' for c in cmd.data)))
@@ -303,7 +441,7 @@ class AxiMasterWrite(Reset):
if k == cycles-1:
stop = end_offset
strb = (self.strb_mask << start) & self.strb_mask & (self.strb_mask >> (self.byte_width - stop))
strb = (self.strb_mask << start) & self.strb_mask & (self.strb_mask >> (self.byte_lanes - stop))
val = 0
for j in range(start, stop):
@@ -350,18 +488,18 @@ class AxiMasterWrite(Reset):
if isinstance(wuser, int):
w.wuser = wuser
else:
if wuser:
w.wuser = wuser.pop(0)
if wuser and k < len(wuser):
w.wuser = wuser[k]
else:
w.wuser = 0
await self.w_channel.send(w)
cur_addr += num_bytes
cycle_offset = (cycle_offset + num_bytes) % self.byte_width
cycle_offset = (cycle_offset + num_bytes) % self.byte_lanes
resp_cmd = AxiWriteRespCmd(cmd.address, len(cmd.data), cmd.size, cycles, cmd.prot, burst_list, cmd.event)
await self.int_write_resp_command_queue[awid].put(resp_cmd)
self.tag_context_manager.start_cmd(awid, resp_cmd)
self.current_write_command = None
@@ -369,27 +507,24 @@ class AxiMasterWrite(Reset):
while True:
b = await self.b_channel.recv()
bid = int(b.bid)
bid = int(getattr(b, 'bid', 0))
if self.active_id[bid] <= 0:
raise Exception(f"Unexpected burst ID {bid}")
await self.int_write_resp_queue_list[bid].put(b)
self.tag_context_manager.put_resp(bid, b)
async def _process_write_resp_id(self, bid):
while True:
cmd = await self.int_write_resp_command_queue[bid].get()
self.current_write_resp_command[bid] = cmd
async def _process_write_resp_id(self, context, cmd):
bid = context.current_tag
resp = AxiResp.OKAY
user = []
for burst_length in cmd.burst_list:
b = await self.int_write_resp_queue_list[bid].get()
b = await context.get_resp()
burst_id = int(b.bid)
burst_resp = AxiResp(b.bresp)
burst_user = int(b.buser)
burst_resp = AxiResp(getattr(b, 'bresp', AxiResp.OKAY))
burst_user = int(getattr(b, 'buser', 0))
if burst_resp != AxiResp.OKAY:
resp = burst_resp
@@ -402,7 +537,10 @@ class AxiMasterWrite(Reset):
self.active_id[bid] -= 1
self.log.info("Write burst complete bid: 0x%x bresp: %s", burst_id, burst_resp)
self.log.info("Write burst complete bid: 0x%x bresp: %s", bid, burst_resp)
if not self.buser_present:
user = None
self.log.info("Write complete addr: 0x%08x prot: %s resp: %s length: %d",
cmd.address, cmd.prot, resp, cmd.length)
@@ -411,16 +549,17 @@ class AxiMasterWrite(Reset):
cmd.event.set(write_resp)
self.current_write_resp_command[bid] = None
self.in_flight_operations -= 1
if self.in_flight_operations == 0:
self._idle.set()
class AxiMasterRead(Reset):
def __init__(self, bus, clock, reset=None, reset_active_level=True, max_burst_len=256):
class AxiMasterRead(Region, Reset):
def __init__(self, bus, clock, reset=None, reset_active_level=True, max_burst_len=256, **kwargs):
self.bus = bus
self.clock = clock
self.reset = reset
self.log = logging.getLogger(f"cocotb.{bus.ar._entity._name}.{bus.ar._name}")
self.log.info("AXI master (read)")
@@ -440,37 +579,54 @@ class AxiMasterRead(Reset):
self.cur_id = 0
self.active_id = Counter()
self.int_read_resp_command_queue = [Queue() for k in range(self.id_count)]
self.current_read_resp_command = [None for k in range(self.id_count)]
self.int_read_resp_queue_list = [Queue() for k in range(self.id_count)]
self.tag_context_manager = TagContextManager(self._process_read_resp_id)
self.in_flight_operations = 0
self._idle = Event()
self._idle.set()
self.address_width = len(self.ar_channel.bus.araddr)
self.id_width = len(self.ar_channel.bus.arid)
self.width = len(self.r_channel.bus.rdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.byte_lanes = self.width // self.byte_size
self.max_burst_len = max(min(max_burst_len, 256), 1)
self.max_burst_size = (self.byte_width-1).bit_length()
self.max_burst_size = (self.byte_lanes-1).bit_length()
self.arlock_present = hasattr(self.bus.ar, "arlock")
self.arcache_present = hasattr(self.bus.ar, "arcache")
self.arprot_present = hasattr(self.bus.ar, "arprot")
self.arqos_present = hasattr(self.bus.ar, "arqos")
self.arregion_present = hasattr(self.bus.ar, "arregion")
self.aruser_present = hasattr(self.bus.ar, "aruser")
self.ruser_present = hasattr(self.bus.r, "ruser")
super().__init__(2**self.address_width, **kwargs)
self.log.info("AXI master configuration:")
self.log.info(" Address width: %d bits", len(self.ar_channel.bus.araddr))
self.log.info(" ID width: %d bits", len(self.ar_channel.bus.arid))
self.log.info(" Address width: %d bits", self.address_width)
self.log.info(" ID width: %d bits", self.id_width)
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_width)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_lanes)
self.log.info(" Max burst size: %d (%d bytes)", self.max_burst_size, 2**self.max_burst_size)
self.log.info(" Max burst length: %d cycles (%d bytes)",
self.max_burst_len, self.max_burst_len*self.byte_width)
self.max_burst_len, self.max_burst_len*self.byte_lanes)
assert self.byte_width * self.byte_size == self.width
self.log.info("AXI master signals:")
for bus in (self.bus.ar, self.bus.r):
for sig in sorted(list(set().union(bus._signals, bus._optional_signals))):
if hasattr(bus, sig):
self.log.info(" %s width: %d bits", sig, len(getattr(bus, sig)))
else:
self.log.info(" %s: not present", sig)
assert self.byte_lanes * self.byte_size == self.width
assert len(self.r_channel.bus.rid) == len(self.ar_channel.bus.arid)
self._process_read_cr = None
self._process_read_resp_cr = None
self._process_read_resp_id_cr = None
self._init_reset(reset, reset_active_level)
@@ -483,6 +639,9 @@ class AxiMasterRead(Reset):
if not isinstance(event, Event):
raise ValueError("Expected event object")
if address < 0 or address >= 2**self.address_width:
raise ValueError("Address out of range")
if length < 0:
raise ValueError("Read length must be positive")
@@ -501,6 +660,24 @@ class AxiMasterRead(Reset):
lock = AxiLockType(lock)
prot = AxiProt(prot)
if not self.arlock_present and lock != AxiLockType.NORMAL:
raise ValueError("arlock sideband signal value specified, but signal is not connected")
if not self.arcache_present and cache != 0b0011:
raise ValueError("arcache sideband signal value specified, but signal is not connected")
if not self.arprot_present and prot != AxiProt.NONSECURE:
raise ValueError("arprot sideband signal value specified, but signal is not connected")
if not self.arqos_present and qos != 0:
raise ValueError("arqos sideband signal value specified, but signal is not connected")
if not self.arregion_present and region != 0:
raise ValueError("arregion sideband signal value specified, but signal is not connected")
if not self.aruser_present and user != 0:
raise ValueError("aruser sideband signal value specified, but signal is not connected")
self.in_flight_operations += 1
self._idle.clear()
@@ -522,43 +699,6 @@ class AxiMasterRead(Reset):
await event.wait()
return event.data
async def read_words(self, address, count, byteorder='little', ws=2, arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
data = await self.read(address, count*ws, arid, burst, size, lock, cache, prot, qos, region, user)
words = []
for k in range(count):
words.append(int.from_bytes(data.data[ws*k:ws*(k+1)], byteorder))
return words
async def read_dwords(self, address, count, byteorder='little', arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return await self.read_words(address, count, byteorder, 4, arid, burst, size,
lock, cache, prot, qos, region, user)
async def read_qwords(self, address, count, byteorder='little', arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return await self.read_words(address, count, byteorder, 8, arid, burst, size,
lock, cache, prot, qos, region, user)
async def read_byte(self, address, arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return (await self.read(address, 1, arid, burst, size, lock, cache, prot, qos, region, user)).data[0]
async def read_word(self, address, byteorder='little', ws=2, arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return (await self.read_words(address, 1, byteorder, ws, arid, burst, size,
lock, cache, prot, qos, region, user))[0]
async def read_dword(self, address, byteorder='little', arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return (await self.read_dwords(address, 1, byteorder, arid, burst, size,
lock, cache, prot, qos, region, user))[0]
async def read_qword(self, address, byteorder='little', arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return (await self.read_qwords(address, 1, byteorder, arid, burst, size,
lock, cache, prot, qos, region, user))[0]
def _handle_reset(self, state):
if state:
self.log.info("Reset asserted")
@@ -568,10 +708,6 @@ class AxiMasterRead(Reset):
if self._process_read_resp_cr is not None:
self._process_read_resp_cr.kill()
self._process_read_resp_cr = None
if self._process_read_resp_id_cr is not None:
for cr in self._process_read_resp_id_cr:
cr.kill()
self._process_read_resp_id_cr = None
self.ar_channel.clear()
self.r_channel.clear()
@@ -590,21 +726,9 @@ class AxiMasterRead(Reset):
self.current_read_command = None
flush_cmd(cmd)
for q in self.int_read_resp_command_queue:
while not q.empty():
cmd = q.get_nowait()
for cmd in self.tag_context_manager.flush():
flush_cmd(cmd)
for k in range(len(self.current_read_resp_command)):
if self.current_read_resp_command[k]:
cmd = self.current_read_resp_command[k]
self.current_read_resp_command[k] = None
flush_cmd(cmd)
for q in self.int_read_resp_queue_list:
while not q.empty():
q.get_nowait()
self.cur_id = 0
self.active_id = Counter()
@@ -616,8 +740,6 @@ class AxiMasterRead(Reset):
self._process_read_cr = cocotb.fork(self._process_read())
if self._process_read_resp_cr is None:
self._process_read_resp_cr = cocotb.fork(self._process_read_resp())
if self._process_read_resp_id_cr is None:
self._process_read_resp_id_cr = [cocotb.fork(self._process_read_resp_id(i)) for i in range(self.id_count)]
async def _process_read(self):
while True:
@@ -658,7 +780,7 @@ class AxiMasterRead(Reset):
burst_list.append(burst_length)
ar = self.r_channel._transaction_obj()
ar = self.ar_channel._transaction_obj()
ar.arid = arid
ar.araddr = cur_addr
ar.arlen = burst_length-1
@@ -680,32 +802,42 @@ class AxiMasterRead(Reset):
cur_addr += num_bytes
resp_cmd = AxiReadRespCmd(cmd.address, cmd.length, cmd.size, cycles, cmd.prot, burst_list, cmd.event)
await self.int_read_resp_command_queue[arid].put(resp_cmd)
self.tag_context_manager.start_cmd(arid, resp_cmd)
self.current_read_command = None
async def _process_read_resp(self):
burst = []
cur_rid = None
while True:
r = await self.r_channel.recv()
rid = int(r.rid)
rid = int(getattr(r, 'rid', 0))
if cur_rid is not None and cur_rid != rid:
raise Exception(f"ID not constant within burst (expected {cur_rid}, got {rid})")
if self.active_id[rid] <= 0:
raise Exception(f"Unexpected burst ID {rid}")
await self.int_read_resp_queue_list[rid].put(r)
burst.append(r)
cur_rid = rid
async def _process_read_resp_id(self, rid):
while True:
cmd = await self.int_read_resp_command_queue[rid].get()
self.current_read_resp_command[rid] = cmd
if int(r.rlast):
self.tag_context_manager.put_resp(rid, burst)
burst = []
cur_rid = None
async def _process_read_resp_id(self, context, cmd):
rid = context.current_tag
num_bytes = 2**cmd.size
aligned_addr = (cmd.address // num_bytes) * num_bytes
word_addr = (cmd.address // self.byte_width) * self.byte_width
word_addr = (cmd.address // self.byte_lanes) * self.byte_lanes
start_offset = cmd.address % self.byte_width
start_offset = cmd.address % self.byte_lanes
cycle_offset = aligned_addr - word_addr
data = bytearray()
@@ -716,14 +848,15 @@ class AxiMasterRead(Reset):
first = True
for burst_length in cmd.burst_list:
for k in range(burst_length):
r = await self.int_read_resp_queue_list[rid].get()
burst = await context.get_resp()
cycle_id = int(r.rid)
if len(burst) != burst_length:
raise Exception(f"Burst length incorrect (ID {rid}, expected {burst_length}, got {len(burst)}")
for r in burst:
cycle_data = int(r.rdata)
cycle_resp = AxiResp(r.rresp)
cycle_last = int(r.rlast)
cycle_user = int(r.ruser)
cycle_resp = AxiResp(getattr(r, "rresp", AxiResp.OKAY))
cycle_user = int(getattr(r, "ruser", 0))
if cycle_resp != AxiResp.OKAY:
resp = cycle_resp
@@ -737,46 +870,45 @@ class AxiMasterRead(Reset):
if first:
start = start_offset
assert cycle_last == (k == burst_length - 1)
for j in range(start, stop):
data.append((cycle_data >> j*8) & 0xff)
cycle_offset = (cycle_offset + num_bytes) % self.byte_width
cycle_offset = (cycle_offset + num_bytes) % self.byte_lanes
first = False
if self.active_id[rid] <= 0:
raise Exception(f"Unexpected burst ID {rid}")
self.active_id[rid] -= 1
self.log.info("Read burst complete rid: 0x%x rresp: %s", cycle_id, resp)
self.log.info("Read burst complete rid: 0x%x rresp: %s", rid, resp)
data = data[:cmd.length]
if not self.ruser_present:
user = None
if self.log.isEnabledFor(logging.INFO):
self.log.info("Read complete addr: 0x%08x prot: %s resp: %s data: %s",
cmd.address, cmd.prot, resp, ' '.join((f'{c:02x}' for c in data)))
read_resp = AxiReadResp(cmd.address, data, resp, user)
read_resp = AxiReadResp(cmd.address, bytes(data), resp, user)
cmd.event.set(read_resp)
self.current_read_resp_command[rid] = None
self.in_flight_operations -= 1
if self.in_flight_operations == 0:
self._idle.set()
class AxiMaster:
def __init__(self, bus, clock, reset=None, reset_active_level=True, max_burst_len=256):
class AxiMaster(Region):
def __init__(self, bus, clock, reset=None, reset_active_level=True, max_burst_len=256, **kwargs):
self.write_if = None
self.read_if = None
self.write_if = AxiMasterWrite(bus.write, clock, reset, reset_active_level, max_burst_len)
self.read_if = AxiMasterRead(bus.read, clock, reset, reset_active_level, max_burst_len)
self.write_if = AxiMasterWrite(bus.write, clock, reset, reset_active_level, max_burst_len, **kwargs)
self.read_if = AxiMasterRead(bus.read, clock, reset, reset_active_level, max_burst_len, **kwargs)
super().__init__(max(self.write_if.size, self.read_if.size), **kwargs)
def init_read(self, address, length, arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, event=None):
@@ -805,77 +937,7 @@ class AxiMaster:
return await self.read_if.read(address, length, arid,
burst, size, lock, cache, prot, qos, region, user)
async def read_words(self, address, count, byteorder='little', ws=2, arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return await self.read_if.read_words(address, count, byteorder, ws, arid,
burst, size, lock, cache, prot, qos, region, user)
async def read_dwords(self, address, count, byteorder='little', arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return await self.read_if.read_dwords(address, count, byteorder, arid,
burst, size, lock, cache, prot, qos, region, user)
async def read_qwords(self, address, count, byteorder='little', arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return await self.read_if.read_qwords(address, count, byteorder, arid,
burst, size, lock, cache, prot, qos, region, user)
async def read_byte(self, address, arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return await self.read_if.read_byte(address, arid,
burst, size, lock, cache, prot, qos, region, user)
async def read_word(self, address, byteorder='little', ws=2, arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return await self.read_if.read_word(address, byteorder, ws, arid,
burst, size, lock, cache, prot, qos, region, user)
async def read_dword(self, address, byteorder='little', arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return await self.read_if.read_dword(address, byteorder, arid,
burst, size, lock, cache, prot, qos, region, user)
async def read_qword(self, address, byteorder='little', arid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
return await self.read_if.read_qword(address, byteorder, arid,
burst, size, lock, cache, prot, qos, region, user)
async def write(self, address, data, awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
return await self.write_if.write(address, data, awid,
burst, size, lock, cache, prot, qos, region, user, wuser)
async def write_words(self, address, data, byteorder='little', ws=2, awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
return await self.write_if.write_words(address, data, byteorder, ws, awid,
burst, size, lock, cache, prot, qos, region, user, wuser)
async def write_dwords(self, address, data, byteorder='little', awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
return await self.write_if.write_dwords(address, data, byteorder, awid,
burst, size, lock, cache, prot, qos, region, user, wuser)
async def write_qwords(self, address, data, byteorder='little', awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
return await self.write_if.write_qwords(address, data, byteorder, awid,
burst, size, lock, cache, prot, qos, region, user, wuser)
async def write_byte(self, address, data, awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
return await self.write_if.write_byte(address, data, awid,
burst, size, lock, cache, prot, qos, region, user, wuser)
async def write_word(self, address, data, byteorder='little', ws=2, awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
return await self.write_if.write_word(address, data, byteorder, ws, awid,
burst, size, lock, cache, prot, qos, region, user, wuser)
async def write_dword(self, address, data, byteorder='little', awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
return await self.write_if.write_dword(address, data, byteorder, awid,
burst, size, lock, cache, prot, qos, region, user, wuser)
async def write_qword(self, address, data, byteorder='little', awid=None, burst=AxiBurstType.INCR, size=None,
lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, wuser=0):
return await self.write_if.write_qword(address, data, byteorder, awid,
burst, size, lock, cache, prot, qos, region, user, wuser)

View File

@@ -1,6 +1,6 @@
"""
Copyright (c) 2020 Alex Forencich
Copyright (c) 2021 Alex Forencich
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
@@ -22,256 +22,32 @@ THE SOFTWARE.
"""
import logging
import cocotb
from .version import __version__
from .constants import AxiBurstType, AxiProt, AxiResp
from .axi_channels import AxiAWSink, AxiWSink, AxiBSource, AxiARSink, AxiRSource
from .axi_slave import AxiSlaveWrite, AxiSlaveRead
from .memory import Memory
from .reset import Reset
class AxiRamWrite(Memory, Reset):
def __init__(self, bus, clock, reset=None, reset_active_level=True, size=1024, mem=None, *args, **kwargs):
self.log = logging.getLogger(f"cocotb.{bus.aw._entity._name}.{bus.aw._name}")
class AxiRamWrite(AxiSlaveWrite, Memory):
def __init__(self, bus, clock, reset=None, reset_active_level=True, size=1024, mem=None, **kwargs):
super().__init__(bus, clock, reset, reset_active_level=reset_active_level, size=size, mem=mem, **kwargs)
self.log.info("AXI RAM model (write)")
self.log.info("cocotbext-axi version %s", __version__)
self.log.info("Copyright (c) 2020 Alex Forencich")
self.log.info("https://github.com/alexforencich/cocotbext-axi")
async def _write(self, address, data):
self.write(address % self.size, data)
super().__init__(size, mem, *args, **kwargs)
self.aw_channel = AxiAWSink(bus.aw, clock, reset, reset_active_level)
self.aw_channel.queue_occupancy_limit = 2
self.w_channel = AxiWSink(bus.w, clock, reset, reset_active_level)
self.w_channel.queue_occupancy_limit = 2
self.b_channel = AxiBSource(bus.b, clock, reset, reset_active_level)
self.b_channel.queue_occupancy_limit = 2
class AxiRamRead(AxiSlaveRead, Memory):
def __init__(self, bus, clock, reset=None, reset_active_level=True, size=1024, mem=None, **kwargs):
super().__init__(bus, clock, reset, reset_active_level=reset_active_level, size=size, mem=mem, **kwargs)
self.width = len(self.w_channel.bus.wdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.strb_mask = 2**self.byte_width-1
self.log.info("AXI RAM model configuration:")
self.log.info(" Memory size: %d bytes", len(self.mem))
self.log.info(" Address width: %d bits", len(self.aw_channel.bus.awaddr))
self.log.info(" ID width: %d bits", len(self.aw_channel.bus.awid))
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_width)
assert self.byte_width == len(self.w_channel.bus.wstrb)
assert self.byte_width * self.byte_size == self.width
assert len(self.b_channel.bus.bid) == len(self.aw_channel.bus.awid)
self._process_write_cr = None
self._init_reset(reset, reset_active_level)
def _handle_reset(self, state):
if state:
self.log.info("Reset asserted")
if self._process_write_cr is not None:
self._process_write_cr.kill()
self._process_write_cr = None
self.aw_channel.clear()
self.w_channel.clear()
self.b_channel.clear()
else:
self.log.info("Reset de-asserted")
if self._process_write_cr is None:
self._process_write_cr = cocotb.fork(self._process_write())
async def _process_write(self):
while True:
aw = await self.aw_channel.recv()
awid = int(aw.awid)
addr = int(aw.awaddr)
length = int(aw.awlen)
size = int(aw.awsize)
burst = int(aw.awburst)
prot = AxiProt(int(aw.awprot))
self.log.info("Write burst awid: 0x%x awaddr: 0x%08x awlen: %d awsize: %d awprot: %s",
awid, addr, length, size, prot)
num_bytes = 2**size
assert 0 < num_bytes <= self.byte_width
aligned_addr = (addr // num_bytes) * num_bytes
length += 1
transfer_size = num_bytes*length
if burst == AxiBurstType.WRAP:
lower_wrap_boundary = (addr // transfer_size) * transfer_size
upper_wrap_boundary = lower_wrap_boundary + transfer_size
if burst == AxiBurstType.INCR:
# check 4k boundary crossing
assert 0x1000-(aligned_addr & 0xfff) >= transfer_size
cur_addr = aligned_addr
for n in range(length):
cur_word_addr = (cur_addr // self.byte_width) * self.byte_width
w = await self.w_channel.recv()
data = int(w.wdata)
strb = int(w.wstrb)
last = int(w.wlast)
# todo latency
self.mem.seek(cur_word_addr % self.size)
data = data.to_bytes(self.byte_width, 'little')
self.log.debug("Write word awid: 0x%x addr: 0x%08x wstrb: 0x%02x data: %s",
awid, cur_addr, strb, ' '.join((f'{c:02x}' for c in data)))
for i in range(self.byte_width):
if strb & (1 << i):
self.mem.write(data[i:i+1])
else:
self.mem.seek(1, 1)
assert last == (n == length-1)
if burst != AxiBurstType.FIXED:
cur_addr += num_bytes
if burst == AxiBurstType.WRAP:
if cur_addr == upper_wrap_boundary:
cur_addr = lower_wrap_boundary
b = self.b_channel._transaction_obj()
b.bid = awid
b.bresp = AxiResp.OKAY
await self.b_channel.send(b)
class AxiRamRead(Memory, Reset):
def __init__(self, bus, clock, reset=None, reset_active_level=True, size=1024, mem=None, *args, **kwargs):
self.log = logging.getLogger(f"cocotb.{bus.ar._entity._name}.{bus.ar._name}")
self.log.info("AXI RAM model (read)")
self.log.info("cocotbext-axi version %s", __version__)
self.log.info("Copyright (c) 2020 Alex Forencich")
self.log.info("https://github.com/alexforencich/cocotbext-axi")
super().__init__(size, mem, *args, **kwargs)
self.ar_channel = AxiARSink(bus.ar, clock, reset, reset_active_level)
self.ar_channel.queue_occupancy_limit = 2
self.r_channel = AxiRSource(bus.r, clock, reset, reset_active_level)
self.r_channel.queue_occupancy_limit = 2
self.width = len(self.r_channel.bus.rdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.log.info("AXI RAM model configuration:")
self.log.info(" Memory size: %d bytes", len(self.mem))
self.log.info(" Address width: %d bits", len(self.ar_channel.bus.araddr))
self.log.info(" ID width: %d bits", len(self.ar_channel.bus.arid))
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_width)
assert self.byte_width * self.byte_size == self.width
assert len(self.r_channel.bus.rid) == len(self.ar_channel.bus.arid)
self._process_read_cr = None
self._init_reset(reset, reset_active_level)
def _handle_reset(self, state):
if state:
self.log.info("Reset asserted")
if self._process_read_cr is not None:
self._process_read_cr.kill()
self._process_read_cr = None
self.ar_channel.clear()
self.r_channel.clear()
else:
self.log.info("Reset de-asserted")
if self._process_read_cr is None:
self._process_read_cr = cocotb.fork(self._process_read())
async def _process_read(self):
while True:
ar = await self.ar_channel.recv()
arid = int(ar.arid)
addr = int(ar.araddr)
length = int(ar.arlen)
size = int(ar.arsize)
burst = int(ar.arburst)
prot = AxiProt(ar.arprot)
self.log.info("Read burst arid: 0x%x araddr: 0x%08x arlen: %d arsize: %d arprot: %s",
arid, addr, length, size, prot)
num_bytes = 2**size
assert 0 < num_bytes <= self.byte_width
aligned_addr = (addr // num_bytes) * num_bytes
length += 1
transfer_size = num_bytes*length
if burst == AxiBurstType.WRAP:
lower_wrap_boundary = (addr // transfer_size) * transfer_size
upper_wrap_boundary = lower_wrap_boundary + transfer_size
if burst == AxiBurstType.INCR:
# check 4k boundary crossing
assert 0x1000-(aligned_addr & 0xfff) >= transfer_size
cur_addr = aligned_addr
for n in range(length):
cur_word_addr = (cur_addr // self.byte_width) * self.byte_width
self.mem.seek(cur_word_addr % self.size)
data = self.mem.read(self.byte_width)
r = self.r_channel._transaction_obj()
r.rid = arid
r.rdata = int.from_bytes(data, 'little')
r.rlast = n == length-1
r.rresp = AxiResp.OKAY
await self.r_channel.send(r)
self.log.debug("Read word awid: 0x%x addr: 0x%08x data: %s",
arid, cur_addr, ' '.join((f'{c:02x}' for c in data)))
if burst != AxiBurstType.FIXED:
cur_addr += num_bytes
if burst == AxiBurstType.WRAP:
if cur_addr == upper_wrap_boundary:
cur_addr = lower_wrap_boundary
async def _read(self, address, length):
return self.read(address % self.size, length)
class AxiRam(Memory):
def __init__(self, bus, clock, reset=None, reset_active_level=True, size=1024, mem=None, *args, **kwargs):
def __init__(self, bus, clock, reset=None, reset_active_level=True, size=1024, mem=None, **kwargs):
self.write_if = None
self.read_if = None
super().__init__(size, mem, *args, **kwargs)
super().__init__(size, mem, **kwargs)
self.write_if = AxiRamWrite(bus.write, clock, reset, reset_active_level, mem=self.mem)
self.read_if = AxiRamRead(bus.read, clock, reset, reset_active_level, mem=self.mem)

335
cocotbext/axi/axi_slave.py Normal file
View File

@@ -0,0 +1,335 @@
"""
Copyright (c) 2021 Alex Forencich
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""
import logging
import cocotb
from .version import __version__
from .constants import AxiBurstType, AxiProt, AxiResp
from .axi_channels import AxiAWSink, AxiWSink, AxiBSource, AxiARSink, AxiRSource
from .reset import Reset
class AxiSlaveWrite(Reset):
def __init__(self, bus, clock, reset=None, target=None, reset_active_level=True, **kwargs):
self.bus = bus
self.clock = clock
self.reset = reset
self.target = target
self.log = logging.getLogger(f"cocotb.{bus.aw._entity._name}.{bus.aw._name}")
self.log.info("AXI slave model (write)")
self.log.info("cocotbext-axi version %s", __version__)
self.log.info("Copyright (c) 2021 Alex Forencich")
self.log.info("https://github.com/alexforencich/cocotbext-axi")
super().__init__(**kwargs)
self.aw_channel = AxiAWSink(bus.aw, clock, reset, reset_active_level)
self.aw_channel.queue_occupancy_limit = 2
self.w_channel = AxiWSink(bus.w, clock, reset, reset_active_level)
self.w_channel.queue_occupancy_limit = 2
self.b_channel = AxiBSource(bus.b, clock, reset, reset_active_level)
self.b_channel.queue_occupancy_limit = 2
self.address_width = len(self.aw_channel.bus.awaddr)
self.id_width = len(self.aw_channel.bus.awid)
self.width = len(self.w_channel.bus.wdata)
self.byte_size = 8
self.byte_lanes = self.width // self.byte_size
self.strb_mask = 2**self.byte_lanes-1
self.max_burst_size = (self.byte_lanes-1).bit_length()
self.log.info("AXI slave model configuration:")
self.log.info(" Address width: %d bits", self.address_width)
self.log.info(" ID width: %d bits", self.id_width)
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_lanes)
self.log.info("AXI slave model signals:")
for bus in (self.bus.aw, self.bus.w, self.bus.b):
for sig in sorted(list(set().union(bus._signals, bus._optional_signals))):
if hasattr(bus, sig):
self.log.info(" %s width: %d bits", sig, len(getattr(bus, sig)))
else:
self.log.info(" %s: not present", sig)
assert self.byte_lanes == len(self.w_channel.bus.wstrb)
assert self.byte_lanes * self.byte_size == self.width
assert len(self.b_channel.bus.bid) == len(self.aw_channel.bus.awid)
self._process_write_cr = None
self._init_reset(reset, reset_active_level)
async def _write(self, address, data):
await self.target.write(address, data)
def _handle_reset(self, state):
if state:
self.log.info("Reset asserted")
if self._process_write_cr is not None:
self._process_write_cr.kill()
self._process_write_cr = None
self.aw_channel.clear()
self.w_channel.clear()
self.b_channel.clear()
else:
self.log.info("Reset de-asserted")
if self._process_write_cr is None:
self._process_write_cr = cocotb.fork(self._process_write())
async def _process_write(self):
while True:
aw = await self.aw_channel.recv()
awid = int(getattr(aw, 'awid', 0))
addr = int(aw.awaddr)
length = int(getattr(aw, 'awlen', 0))
size = int(getattr(aw, 'awsize', self.max_burst_size))
burst = AxiBurstType(getattr(aw, 'awburst', AxiBurstType.INCR))
prot = AxiProt(getattr(aw, 'awprot', AxiProt.NONSECURE))
self.log.info("Write burst awid: 0x%x awaddr: 0x%08x awlen: %d awsize: %d awprot: %s",
awid, addr, length, size, prot)
num_bytes = 2**size
assert 0 < num_bytes <= self.byte_lanes
aligned_addr = (addr // num_bytes) * num_bytes
length += 1
transfer_size = num_bytes*length
if burst == AxiBurstType.WRAP:
lower_wrap_boundary = (addr // transfer_size) * transfer_size
upper_wrap_boundary = lower_wrap_boundary + transfer_size
if burst == AxiBurstType.INCR:
# check 4k boundary crossing
assert 0x1000-(aligned_addr & 0xfff) >= transfer_size
cur_addr = aligned_addr
b = self.b_channel._transaction_obj()
b.bid = awid
b.bresp = AxiResp.OKAY
for n in range(length):
cur_word_addr = (cur_addr // self.byte_lanes) * self.byte_lanes
w = await self.w_channel.recv()
data = int(w.wdata)
strb = int(getattr(w, 'wstrb', self.strb_mask))
last = int(w.wlast)
# generate operation list
offset = 0
start_offset = None
write_ops = []
data = data.to_bytes(self.byte_lanes, 'little')
if self.log.isEnabledFor(logging.DEBUG):
self.log.debug("Write word awid: 0x%x addr: 0x%08x wstrb: 0x%02x data: %s",
awid, cur_addr, strb, ' '.join((f'{c:02x}' for c in data)))
for i in range(self.byte_lanes):
if strb & (1 << i):
if start_offset is None:
start_offset = offset
else:
if start_offset is not None and offset != start_offset:
write_ops.append((cur_word_addr+start_offset, data[start_offset:offset]))
start_offset = None
offset += 1
if start_offset is not None and offset != start_offset:
write_ops.append((cur_word_addr+start_offset, data[start_offset:offset]))
# perform writes
try:
for addr, data in write_ops:
await self._write(addr, data)
except Exception:
self.log.warning("Write operation failed")
b.bresp = AxiResp.SLVERR
assert last == (n == length-1)
if burst != AxiBurstType.FIXED:
cur_addr += num_bytes
if burst == AxiBurstType.WRAP:
if cur_addr == upper_wrap_boundary:
cur_addr = lower_wrap_boundary
await self.b_channel.send(b)
class AxiSlaveRead(Reset):
def __init__(self, bus, clock, reset=None, target=None, reset_active_level=True, **kwargs):
self.bus = bus
self.clock = clock
self.reset = reset
self.target = target
self.log = logging.getLogger(f"cocotb.{bus.ar._entity._name}.{bus.ar._name}")
self.log.info("AXI slave model (read)")
self.log.info("cocotbext-axi version %s", __version__)
self.log.info("Copyright (c) 2021 Alex Forencich")
self.log.info("https://github.com/alexforencich/cocotbext-axi")
super().__init__(**kwargs)
self.ar_channel = AxiARSink(bus.ar, clock, reset, reset_active_level)
self.ar_channel.queue_occupancy_limit = 2
self.r_channel = AxiRSource(bus.r, clock, reset, reset_active_level)
self.r_channel.queue_occupancy_limit = 2
self.address_width = len(self.ar_channel.bus.araddr)
self.id_width = len(self.ar_channel.bus.arid)
self.width = len(self.r_channel.bus.rdata)
self.byte_size = 8
self.byte_lanes = self.width // self.byte_size
self.max_burst_size = (self.byte_lanes-1).bit_length()
self.log.info("AXI slave model configuration:")
self.log.info(" Address width: %d bits", self.address_width)
self.log.info(" ID width: %d bits", self.id_width)
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_lanes)
self.log.info("AXI slave model signals:")
for bus in (self.bus.ar, self.bus.r):
for sig in sorted(list(set().union(bus._signals, bus._optional_signals))):
if hasattr(bus, sig):
self.log.info(" %s width: %d bits", sig, len(getattr(bus, sig)))
else:
self.log.info(" %s: not present", sig)
assert self.byte_lanes * self.byte_size == self.width
assert len(self.r_channel.bus.rid) == len(self.ar_channel.bus.arid)
self._process_read_cr = None
self._init_reset(reset, reset_active_level)
async def _read(self, address, length):
return await self.target.read(address, length)
def _handle_reset(self, state):
if state:
self.log.info("Reset asserted")
if self._process_read_cr is not None:
self._process_read_cr.kill()
self._process_read_cr = None
self.ar_channel.clear()
self.r_channel.clear()
else:
self.log.info("Reset de-asserted")
if self._process_read_cr is None:
self._process_read_cr = cocotb.fork(self._process_read())
async def _process_read(self):
while True:
ar = await self.ar_channel.recv()
arid = int(getattr(ar, 'arid', 0))
addr = int(ar.araddr)
length = int(getattr(ar, 'arlen', 0))
size = int(getattr(ar, 'arsize', self.max_burst_size))
burst = AxiBurstType(getattr(ar, 'arburst', AxiBurstType.INCR))
prot = AxiProt(getattr(ar, 'arprot', AxiProt.NONSECURE))
self.log.info("Read burst arid: 0x%x araddr: 0x%08x arlen: %d arsize: %d arprot: %s",
arid, addr, length, size, prot)
num_bytes = 2**size
assert 0 < num_bytes <= self.byte_lanes
aligned_addr = (addr // num_bytes) * num_bytes
length += 1
transfer_size = num_bytes*length
if burst == AxiBurstType.WRAP:
lower_wrap_boundary = (addr // transfer_size) * transfer_size
upper_wrap_boundary = lower_wrap_boundary + transfer_size
if burst == AxiBurstType.INCR:
# check 4k boundary crossing
assert 0x1000-(aligned_addr & 0xfff) >= transfer_size
cur_addr = aligned_addr
for n in range(length):
cur_word_addr = (cur_addr // self.byte_lanes) * self.byte_lanes
r = self.r_channel._transaction_obj()
r.rid = arid
r.rlast = n == length-1
r.rresp = AxiResp.OKAY
try:
data = await self._read(cur_word_addr, self.byte_lanes)
except Exception:
self.log.warning("Read operation failed")
data = bytes(self.byte_lanes)
r.rresp = AxiResp.SLVERR
r.rdata = int.from_bytes(data, 'little')
await self.r_channel.send(r)
if self.log.isEnabledFor(logging.DEBUG):
self.log.debug("Read word awid: 0x%x addr: 0x%08x data: %s",
arid, cur_addr, ' '.join((f'{c:02x}' for c in data)))
if burst != AxiBurstType.FIXED:
cur_addr += num_bytes
if burst == AxiBurstType.WRAP:
if cur_addr == upper_wrap_boundary:
cur_addr = lower_wrap_boundary
class AxiSlave:
def __init__(self, bus, clock, reset=None, target=None, reset_active_level=True, **kwargs):
self.write_if = None
self.read_if = None
super().__init__(**kwargs)
self.write_if = AxiSlaveWrite(bus.write, clock, reset, target, reset_active_level)
self.read_if = AxiSlaveRead(bus.read, clock, reset, target, reset_active_level)

View File

@@ -26,7 +26,8 @@ from .stream import define_stream
# Write address channel
AxiLiteAWBus, AxiLiteAWTransaction, AxiLiteAWSource, AxiLiteAWSink, AxiLiteAWMonitor = define_stream("AxiLiteAW",
signals=["awaddr", "awprot", "awvalid", "awready"],
signals=["awaddr", "awvalid", "awready"],
optional_signals=["awprot"],
signal_widths={"awprot": 3}
)
@@ -37,19 +38,22 @@ AxiLiteWBus, AxiLiteWTransaction, AxiLiteWSource, AxiLiteWSink, AxiLiteWMonitor
# Write response channel
AxiLiteBBus, AxiLiteBTransaction, AxiLiteBSource, AxiLiteBSink, AxiLiteBMonitor = define_stream("AxiLiteB",
signals=["bresp", "bvalid", "bready"],
signals=["bvalid", "bready"],
optional_signals=["bresp"],
signal_widths={"bresp": 2}
)
# Read address channel
AxiLiteARBus, AxiLiteARTransaction, AxiLiteARSource, AxiLiteARSink, AxiLiteARMonitor = define_stream("AxiLiteAR",
signals=["araddr", "arprot", "arvalid", "arready"],
signals=["araddr", "arvalid", "arready"],
optional_signals=["arprot"],
signal_widths={"arprot": 3}
)
# Read data channel
AxiLiteRBus, AxiLiteRTransaction, AxiLiteRSource, AxiLiteRSink, AxiLiteRMonitor = define_stream("AxiLiteR",
signals=["rdata", "rresp", "rvalid", "rready"],
signals=["rdata", "rvalid", "rready"],
optional_signals=["rresp"],
signal_widths={"rresp": 2}
)

View File

@@ -23,7 +23,7 @@ THE SOFTWARE.
"""
import logging
from collections import namedtuple
from typing import NamedTuple
import cocotb
from cocotb.queue import Queue
@@ -32,21 +32,62 @@ from cocotb.triggers import Event
from .version import __version__
from .constants import AxiProt, AxiResp
from .axil_channels import AxiLiteAWSource, AxiLiteWSource, AxiLiteBSink, AxiLiteARSource, AxiLiteRSink
from .address_space import Region
from .reset import Reset
# AXI lite master write
AxiLiteWriteCmd = namedtuple("AxiLiteWriteCmd", ["address", "data", "prot", "event"])
AxiLiteWriteRespCmd = namedtuple("AxiLiteWriteRespCmd", ["address", "length", "cycles", "prot", "event"])
AxiLiteWriteResp = namedtuple("AxiLiteWriteResp", ["address", "length", "resp"])
# AXI lite master read
AxiLiteReadCmd = namedtuple("AxiLiteReadCmd", ["address", "length", "prot", "event"])
AxiLiteReadRespCmd = namedtuple("AxiLiteReadRespCmd", ["address", "length", "cycles", "prot", "event"])
AxiLiteReadResp = namedtuple("AxiLiteReadResp", ["address", "data", "resp"])
# AXI lite master write helper objects
class AxiLiteWriteCmd(NamedTuple):
address: int
data: bytes
prot: AxiProt
event: Event
class AxiLiteMasterWrite(Reset):
def __init__(self, bus, clock, reset=None, reset_active_level=True):
class AxiLiteWriteRespCmd(NamedTuple):
address: int
length: int
cycles: int
prot: AxiProt
event: Event
class AxiLiteWriteResp(NamedTuple):
address: int
length: int
resp: AxiResp
# AXI lite master read helper objects
class AxiLiteReadCmd(NamedTuple):
address: int
length: int
prot: AxiProt
event: Event
class AxiLiteReadRespCmd(NamedTuple):
address: int
length: int
cycles: int
prot: AxiProt
event: Event
class AxiLiteReadResp(NamedTuple):
address: int
data: bytes
resp: AxiResp
def __bytes__(self):
return self.data
class AxiLiteMasterWrite(Region, Reset):
def __init__(self, bus, clock, reset=None, reset_active_level=True, **kwargs):
self.bus = bus
self.clock = clock
self.reset = reset
self.log = logging.getLogger(f"cocotb.{bus.aw._entity._name}.{bus.aw._name}")
self.log.info("AXI lite master (write)")
@@ -71,18 +112,31 @@ class AxiLiteMasterWrite(Reset):
self._idle = Event()
self._idle.set()
self.address_width = len(self.aw_channel.bus.awaddr)
self.width = len(self.w_channel.bus.wdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.strb_mask = 2**self.byte_width-1
self.byte_lanes = self.width // self.byte_size
self.strb_mask = 2**self.byte_lanes-1
self.awprot_present = hasattr(self.bus.aw, "awprot")
super().__init__(2**self.address_width, **kwargs)
self.log.info("AXI lite master configuration:")
self.log.info(" Address width: %d bits", len(self.aw_channel.bus.awaddr))
self.log.info(" Address width: %d bits", self.address_width)
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_width)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_lanes)
assert self.byte_width == len(self.w_channel.bus.wstrb)
assert self.byte_width * self.byte_size == self.width
self.log.info("AXI lite master signals:")
for bus in (self.bus.aw, self.bus.w, self.bus.b):
for sig in sorted(list(set().union(bus._signals, bus._optional_signals))):
if hasattr(bus, sig):
self.log.info(" %s width: %d bits", sig, len(getattr(bus, sig)))
else:
self.log.info(" %s: not present", sig)
assert self.byte_lanes == len(self.w_channel.bus.wstrb)
assert self.byte_lanes * self.byte_size == self.width
self._process_write_cr = None
self._process_write_resp_cr = None
@@ -96,10 +150,19 @@ class AxiLiteMasterWrite(Reset):
if not isinstance(event, Event):
raise ValueError("Expected event object")
if address < 0 or address >= 2**self.address_width:
raise ValueError("Address out of range")
if isinstance(data, int):
raise ValueError("Expected bytes or bytearray for data")
if not self.awprot_present and prot != AxiProt.NONSECURE:
raise ValueError("awprot sideband signal value specified, but signal is not connected")
self.in_flight_operations += 1
self._idle.clear()
self.write_command_queue.put_nowait(AxiLiteWriteCmd(address, bytearray(data), prot, event))
self.write_command_queue.put_nowait(AxiLiteWriteCmd(address, bytes(data), prot, event))
return event
@@ -115,31 +178,6 @@ class AxiLiteMasterWrite(Reset):
await event.wait()
return event.data
async def write_words(self, address, data, byteorder='little', ws=2, prot=AxiProt.NONSECURE):
words = data
data = bytearray()
for w in words:
data.extend(w.to_bytes(ws, byteorder))
await self.write(address, data, prot)
async def write_dwords(self, address, data, byteorder='little', prot=AxiProt.NONSECURE):
await self.write_words(address, data, byteorder, 4, prot)
async def write_qwords(self, address, data, byteorder='little', prot=AxiProt.NONSECURE):
await self.write_words(address, data, byteorder, 8, prot)
async def write_byte(self, address, data, prot=AxiProt.NONSECURE):
await self.write(address, [data], prot)
async def write_word(self, address, data, byteorder='little', ws=2, prot=AxiProt.NONSECURE):
await self.write_words(address, [data], byteorder, ws, prot)
async def write_dword(self, address, data, byteorder='little', prot=AxiProt.NONSECURE):
await self.write_dwords(address, [data], byteorder, prot)
async def write_qword(self, address, data, byteorder='little', prot=AxiProt.NONSECURE):
await self.write_qwords(address, [data], byteorder, prot)
def _handle_reset(self, state):
if state:
self.log.info("Reset asserted")
@@ -191,27 +229,28 @@ class AxiLiteMasterWrite(Reset):
cmd = await self.write_command_queue.get()
self.current_write_command = cmd
word_addr = (cmd.address // self.byte_width) * self.byte_width
word_addr = (cmd.address // self.byte_lanes) * self.byte_lanes
start_offset = cmd.address % self.byte_width
end_offset = ((cmd.address + len(cmd.data) - 1) % self.byte_width) + 1
start_offset = cmd.address % self.byte_lanes
end_offset = ((cmd.address + len(cmd.data) - 1) % self.byte_lanes) + 1
strb_start = (self.strb_mask << start_offset) & self.strb_mask
strb_end = self.strb_mask >> (self.byte_width - end_offset)
strb_end = self.strb_mask >> (self.byte_lanes - end_offset)
cycles = (len(cmd.data) + (cmd.address % self.byte_width) + self.byte_width-1) // self.byte_width
cycles = (len(cmd.data) + (cmd.address % self.byte_lanes) + self.byte_lanes-1) // self.byte_lanes
resp_cmd = AxiLiteWriteRespCmd(cmd.address, len(cmd.data), cycles, cmd.prot, cmd.event)
await self.int_write_resp_command_queue.put(resp_cmd)
offset = 0
if self.log.isEnabledFor(logging.INFO):
self.log.info("Write start addr: 0x%08x prot: %s data: %s",
cmd.address, cmd.prot, ' '.join((f'{c:02x}' for c in cmd.data)))
for k in range(cycles):
start = 0
stop = self.byte_width
stop = self.byte_lanes
strb = self.strb_mask
if k == 0:
@@ -227,7 +266,7 @@ class AxiLiteMasterWrite(Reset):
offset += 1
aw = self.aw_channel._transaction_obj()
aw.awaddr = word_addr + k*self.byte_width
aw.awaddr = word_addr + k*self.byte_lanes
aw.awprot = cmd.prot
w = self.w_channel._transaction_obj()
@@ -249,7 +288,7 @@ class AxiLiteMasterWrite(Reset):
for k in range(cmd.cycles):
b = await self.b_channel.recv()
cycle_resp = AxiResp(b.bresp)
cycle_resp = AxiResp(getattr(b, 'bresp', AxiResp.OKAY))
if cycle_resp != AxiResp.OKAY:
resp = cycle_resp
@@ -269,8 +308,11 @@ class AxiLiteMasterWrite(Reset):
self._idle.set()
class AxiLiteMasterRead(Reset):
def __init__(self, bus, clock, reset=None, reset_active_level=True):
class AxiLiteMasterRead(Region, Reset):
def __init__(self, bus, clock, reset=None, reset_active_level=True, **kwargs):
self.bus = bus
self.clock = clock
self.reset = reset
self.log = logging.getLogger(f"cocotb.{bus.ar._entity._name}.{bus.ar._name}")
self.log.info("AXI lite master (read)")
@@ -293,16 +335,29 @@ class AxiLiteMasterRead(Reset):
self._idle = Event()
self._idle.set()
self.address_width = len(self.ar_channel.bus.araddr)
self.width = len(self.r_channel.bus.rdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.byte_lanes = self.width // self.byte_size
self.arprot_present = hasattr(self.bus.ar, "arprot")
super().__init__(2**self.address_width, **kwargs)
self.log.info("AXI lite master configuration:")
self.log.info(" Address width: %d bits", len(self.ar_channel.bus.araddr))
self.log.info(" Address width: %d bits", self.address_width)
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_width)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_lanes)
assert self.byte_width * self.byte_size == self.width
self.log.info("AXI lite master signals:")
for bus in (self.bus.ar, self.bus.r):
for sig in sorted(list(set().union(bus._signals, bus._optional_signals))):
if hasattr(bus, sig):
self.log.info(" %s width: %d bits", sig, len(getattr(bus, sig)))
else:
self.log.info(" %s: not present", sig)
assert self.byte_lanes * self.byte_size == self.width
self._process_read_cr = None
self._process_read_resp_cr = None
@@ -316,6 +371,12 @@ class AxiLiteMasterRead(Reset):
if not isinstance(event, Event):
raise ValueError("Expected event object")
if address < 0 or address >= 2**self.address_width:
raise ValueError("Address out of range")
if not self.arprot_present and prot != AxiProt.NONSECURE:
raise ValueError("arprot sideband signal value specified, but signal is not connected")
self.in_flight_operations += 1
self._idle.clear()
@@ -335,31 +396,6 @@ class AxiLiteMasterRead(Reset):
await event.wait()
return event.data
async def read_words(self, address, count, byteorder='little', ws=2, prot=AxiProt.NONSECURE):
data = await self.read(address, count*ws, prot)
words = []
for k in range(count):
words.append(int.from_bytes(data.data[ws*k:ws*(k+1)], byteorder))
return words
async def read_dwords(self, address, count, byteorder='little', prot=AxiProt.NONSECURE):
return await self.read_words(address, count, byteorder, 4, prot)
async def read_qwords(self, address, count, byteorder='little', prot=AxiProt.NONSECURE):
return await self.read_words(address, count, byteorder, 8, prot)
async def read_byte(self, address, prot=AxiProt.NONSECURE):
return (await self.read(address, 1, prot)).data[0]
async def read_word(self, address, byteorder='little', ws=2, prot=AxiProt.NONSECURE):
return (await self.read_words(address, 1, byteorder, ws, prot))[0]
async def read_dword(self, address, byteorder='little', prot=AxiProt.NONSECURE):
return (await self.read_dwords(address, 1, byteorder, prot))[0]
async def read_qword(self, address, byteorder='little', prot=AxiProt.NONSECURE):
return (await self.read_qwords(address, 1, byteorder, prot))[0]
def _handle_reset(self, state):
if state:
self.log.info("Reset asserted")
@@ -410,9 +446,9 @@ class AxiLiteMasterRead(Reset):
cmd = await self.read_command_queue.get()
self.current_read_command = cmd
word_addr = (cmd.address // self.byte_width) * self.byte_width
word_addr = (cmd.address // self.byte_lanes) * self.byte_lanes
cycles = (cmd.length + self.byte_width-1 + (cmd.address % self.byte_width)) // self.byte_width
cycles = (cmd.length + self.byte_lanes-1 + (cmd.address % self.byte_lanes)) // self.byte_lanes
resp_cmd = AxiLiteReadRespCmd(cmd.address, cmd.length, cycles, cmd.prot, cmd.event)
await self.int_read_resp_command_queue.put(resp_cmd)
@@ -422,7 +458,7 @@ class AxiLiteMasterRead(Reset):
for k in range(cycles):
ar = self.ar_channel._transaction_obj()
ar.araddr = word_addr + k*self.byte_width
ar.araddr = word_addr + k*self.byte_lanes
ar.arprot = cmd.prot
await self.ar_channel.send(ar)
@@ -434,8 +470,8 @@ class AxiLiteMasterRead(Reset):
cmd = await self.int_read_resp_command_queue.get()
self.current_read_resp_command = cmd
start_offset = cmd.address % self.byte_width
end_offset = ((cmd.address + cmd.length - 1) % self.byte_width) + 1
start_offset = cmd.address % self.byte_lanes
end_offset = ((cmd.address + cmd.length - 1) % self.byte_lanes) + 1
data = bytearray()
@@ -445,13 +481,13 @@ class AxiLiteMasterRead(Reset):
r = await self.r_channel.recv()
cycle_data = int(r.rdata)
cycle_resp = AxiResp(r.rresp)
cycle_resp = AxiResp(getattr(r, 'rresp', AxiResp.OKAY))
if cycle_resp != AxiResp.OKAY:
resp = cycle_resp
start = 0
stop = self.byte_width
stop = self.byte_lanes
if k == 0:
start = start_offset
@@ -461,10 +497,11 @@ class AxiLiteMasterRead(Reset):
for j in range(start, stop):
data.extend(bytearray([(cycle_data >> j*8) & 0xff]))
if self.log.isEnabledFor(logging.INFO):
self.log.info("Read complete addr: 0x%08x prot: %s resp: %s data: %s",
cmd.address, cmd.prot, resp, ' '.join((f'{c:02x}' for c in data)))
read_resp = AxiLiteReadResp(cmd.address, data, resp)
read_resp = AxiLiteReadResp(cmd.address, bytes(data), resp)
cmd.event.set(read_resp)
@@ -476,13 +513,15 @@ class AxiLiteMasterRead(Reset):
self._idle.set()
class AxiLiteMaster:
def __init__(self, bus, clock, reset=None, reset_active_level=True):
class AxiLiteMaster(Region):
def __init__(self, bus, clock, reset=None, reset_active_level=True, **kwargs):
self.write_if = None
self.read_if = None
self.write_if = AxiLiteMasterWrite(bus.write, clock, reset, reset_active_level)
self.read_if = AxiLiteMasterRead(bus.read, clock, reset, reset_active_level)
self.write_if = AxiLiteMasterWrite(bus.write, clock, reset, reset_active_level, **kwargs)
self.read_if = AxiLiteMasterRead(bus.read, clock, reset, reset_active_level, **kwargs)
super().__init__(max(self.write_if.size, self.read_if.size), **kwargs)
def init_read(self, address, length, prot=AxiProt.NONSECURE, event=None):
return self.read_if.init_read(address, length, prot, event)
@@ -507,47 +546,5 @@ class AxiLiteMaster:
async def read(self, address, length, prot=AxiProt.NONSECURE):
return await self.read_if.read(address, length, prot)
async def read_words(self, address, count, byteorder='little', ws=2, prot=AxiProt.NONSECURE):
return await self.read_if.read_words(address, count, byteorder, ws, prot)
async def read_dwords(self, address, count, byteorder='little', prot=AxiProt.NONSECURE):
return await self.read_if.read_dwords(address, count, byteorder, prot)
async def read_qwords(self, address, count, byteorder='little', prot=AxiProt.NONSECURE):
return await self.read_if.read_qwords(address, count, byteorder, prot)
async def read_byte(self, address, prot=AxiProt.NONSECURE):
return await self.read_if.read_byte(address, prot)
async def read_word(self, address, byteorder='little', ws=2, prot=AxiProt.NONSECURE):
return await self.read_if.read_word(address, byteorder, ws, prot)
async def read_dword(self, address, byteorder='little', prot=AxiProt.NONSECURE):
return await self.read_if.read_dword(address, byteorder, prot)
async def read_qword(self, address, byteorder='little', prot=AxiProt.NONSECURE):
return await self.read_if.read_qword(address, byteorder, prot)
async def write(self, address, data, prot=AxiProt.NONSECURE):
return await self.write_if.write(address, data, prot)
async def write_words(self, address, data, byteorder='little', ws=2, prot=AxiProt.NONSECURE):
return await self.write_if.write_words(address, data, byteorder, ws, prot)
async def write_dwords(self, address, data, byteorder='little', prot=AxiProt.NONSECURE):
return await self.write_if.write_dwords(address, data, byteorder, prot)
async def write_qwords(self, address, data, byteorder='little', prot=AxiProt.NONSECURE):
return await self.write_if.write_qwords(address, data, byteorder, prot)
async def write_byte(self, address, data, prot=AxiProt.NONSECURE):
return await self.write_if.write_byte(address, data, prot)
async def write_word(self, address, data, byteorder='little', ws=2, prot=AxiProt.NONSECURE):
return await self.write_if.write_word(address, data, byteorder, ws, prot)
async def write_dword(self, address, data, byteorder='little', prot=AxiProt.NONSECURE):
return await self.write_if.write_dword(address, data, byteorder, prot)
async def write_qword(self, address, data, byteorder='little', prot=AxiProt.NONSECURE):
return await self.write_if.write_qword(address, data, byteorder, prot)

View File

@@ -1,6 +1,6 @@
"""
Copyright (c) 2020 Alex Forencich
Copyright (c) 2021 Alex Forencich
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
@@ -22,176 +22,32 @@ THE SOFTWARE.
"""
import logging
import cocotb
from .version import __version__
from .constants import AxiProt, AxiResp
from .axil_channels import AxiLiteAWSink, AxiLiteWSink, AxiLiteBSource, AxiLiteARSink, AxiLiteRSource
from .axil_slave import AxiLiteSlaveWrite, AxiLiteSlaveRead
from .memory import Memory
from .reset import Reset
class AxiLiteRamWrite(Memory, Reset):
def __init__(self, bus, clock, reset=None, reset_active_level=True, size=1024, mem=None, *args, **kwargs):
self.log = logging.getLogger(f"cocotb.{bus.aw._entity._name}.{bus.aw._name}")
class AxiLiteRamWrite(AxiLiteSlaveWrite, Memory):
def __init__(self, bus, clock, reset=None, reset_active_level=True, size=1024, mem=None, **kwargs):
super().__init__(bus, clock, reset, reset_active_level=reset_active_level, size=size, mem=mem, **kwargs)
self.log.info("AXI lite RAM model (write)")
self.log.info("cocotbext-axi version %s", __version__)
self.log.info("Copyright (c) 2020 Alex Forencich")
self.log.info("https://github.com/alexforencich/cocotbext-axi")
super().__init__(size, mem, *args, **kwargs)
self.aw_channel = AxiLiteAWSink(bus.aw, clock, reset, reset_active_level)
self.aw_channel.queue_occupancy_limit = 2
self.w_channel = AxiLiteWSink(bus.w, clock, reset, reset_active_level)
self.w_channel.queue_occupancy_limit = 2
self.b_channel = AxiLiteBSource(bus.b, clock, reset, reset_active_level)
self.b_channel.queue_occupancy_limit = 2
self.width = len(self.w_channel.bus.wdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.strb_mask = 2**self.byte_width-1
self.log.info("AXI lite RAM model configuration:")
self.log.info(" Memory size: %d bytes", len(self.mem))
self.log.info(" Address width: %d bits", len(self.aw_channel.bus.awaddr))
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_width)
assert self.byte_width == len(self.w_channel.bus.wstrb)
assert self.byte_width * self.byte_size == self.width
self._process_write_cr = None
self._init_reset(reset, reset_active_level)
def _handle_reset(self, state):
if state:
self.log.info("Reset asserted")
if self._process_write_cr is not None:
self._process_write_cr.kill()
self._process_write_cr = None
self.aw_channel.clear()
self.w_channel.clear()
self.b_channel.clear()
else:
self.log.info("Reset de-asserted")
if self._process_write_cr is None:
self._process_write_cr = cocotb.fork(self._process_write())
async def _process_write(self):
while True:
aw = await self.aw_channel.recv()
addr = (int(aw.awaddr) // self.byte_width) * self.byte_width
prot = AxiProt(aw.awprot)
w = await self.w_channel.recv()
data = int(w.wdata)
strb = int(w.wstrb)
# todo latency
self.mem.seek(addr % self.size)
data = data.to_bytes(self.byte_width, 'little')
self.log.info("Write data awaddr: 0x%08x awprot: %s wstrb: 0x%02x data: %s",
addr, prot, strb, ' '.join((f'{c:02x}' for c in data)))
for i in range(self.byte_width):
if strb & (1 << i):
self.mem.write(data[i:i+1])
else:
self.mem.seek(1, 1)
b = self.b_channel._transaction_obj()
b.bresp = AxiResp.OKAY
await self.b_channel.send(b)
async def _write(self, address, data):
self.write(address % self.size, data)
class AxiLiteRamRead(Memory, Reset):
def __init__(self, bus, clock, reset=None, reset_active_level=True, size=1024, mem=None, *args, **kwargs):
self.log = logging.getLogger(f"cocotb.{bus.ar._entity._name}.{bus.ar._name}")
class AxiLiteRamRead(AxiLiteSlaveRead, Memory):
def __init__(self, bus, clock, reset=None, reset_active_level=True, size=1024, mem=None, **kwargs):
super().__init__(bus, clock, reset, reset_active_level=reset_active_level, size=size, mem=mem, **kwargs)
self.log.info("AXI lite RAM model (read)")
self.log.info("cocotbext-axi version %s", __version__)
self.log.info("Copyright (c) 2020 Alex Forencich")
self.log.info("https://github.com/alexforencich/cocotbext-axi")
super().__init__(size, mem, *args, **kwargs)
self.ar_channel = AxiLiteARSink(bus.ar, clock, reset, reset_active_level)
self.ar_channel.queue_occupancy_limit = 2
self.r_channel = AxiLiteRSource(bus.r, clock, reset, reset_active_level)
self.r_channel.queue_occupancy_limit = 2
self.width = len(self.r_channel.bus.rdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.log.info("AXI lite RAM model configuration:")
self.log.info(" Memory size: %d bytes", len(self.mem))
self.log.info(" Address width: %d bits", len(self.ar_channel.bus.araddr))
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_width)
assert self.byte_width * self.byte_size == self.width
self._process_read_cr = None
self._init_reset(reset, reset_active_level)
def _handle_reset(self, state):
if state:
self.log.info("Reset asserted")
if self._process_read_cr is not None:
self._process_read_cr.kill()
self._process_read_cr = None
self.ar_channel.clear()
self.r_channel.clear()
else:
self.log.info("Reset de-asserted")
if self._process_read_cr is None:
self._process_read_cr = cocotb.fork(self._process_read())
async def _process_read(self):
while True:
ar = await self.ar_channel.recv()
addr = (int(ar.araddr) // self.byte_width) * self.byte_width
prot = AxiProt(ar.arprot)
# todo latency
self.mem.seek(addr % self.size)
data = self.mem.read(self.byte_width)
r = self.r_channel._transaction_obj()
r.rdata = int.from_bytes(data, 'little')
r.rresp = AxiResp.OKAY
await self.r_channel.send(r)
self.log.info("Read data araddr: 0x%08x arprot: %s data: %s",
addr, prot, ' '.join((f'{c:02x}' for c in data)))
async def _read(self, address, length):
return self.read(address % self.size, length)
class AxiLiteRam(Memory):
def __init__(self, bus, clock, reset=None, reset_active_level=True, size=1024, mem=None, *args, **kwargs):
def __init__(self, bus, clock, reset=None, reset_active_level=True, size=1024, mem=None, **kwargs):
self.write_if = None
self.read_if = None
super().__init__(size, mem, *args, **kwargs)
super().__init__(size, mem, **kwargs)
self.write_if = AxiLiteRamWrite(bus.write, clock, reset, reset_active_level, mem=self.mem)
self.read_if = AxiLiteRamRead(bus.read, clock, reset, reset_active_level, mem=self.mem)

249
cocotbext/axi/axil_slave.py Normal file
View File

@@ -0,0 +1,249 @@
"""
Copyright (c) 2021 Alex Forencich
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""
import logging
import cocotb
from .version import __version__
from .constants import AxiProt, AxiResp
from .axil_channels import AxiLiteAWSink, AxiLiteWSink, AxiLiteBSource, AxiLiteARSink, AxiLiteRSource
from .reset import Reset
class AxiLiteSlaveWrite(Reset):
def __init__(self, bus, clock, reset=None, target=None, reset_active_level=True, **kwargs):
self.bus = bus
self.clock = clock
self.reset = reset
self.target = target
self.log = logging.getLogger(f"cocotb.{bus.aw._entity._name}.{bus.aw._name}")
self.log.info("AXI lite slave model (write)")
self.log.info("cocotbext-axi version %s", __version__)
self.log.info("Copyright (c) 2021 Alex Forencich")
self.log.info("https://github.com/alexforencich/cocotbext-axi")
super().__init__(**kwargs)
self.aw_channel = AxiLiteAWSink(bus.aw, clock, reset, reset_active_level)
self.aw_channel.queue_occupancy_limit = 2
self.w_channel = AxiLiteWSink(bus.w, clock, reset, reset_active_level)
self.w_channel.queue_occupancy_limit = 2
self.b_channel = AxiLiteBSource(bus.b, clock, reset, reset_active_level)
self.b_channel.queue_occupancy_limit = 2
self.address_width = len(self.aw_channel.bus.awaddr)
self.width = len(self.w_channel.bus.wdata)
self.byte_size = 8
self.byte_lanes = self.width // self.byte_size
self.strb_mask = 2**self.byte_lanes-1
self.log.info("AXI lite slave model configuration:")
self.log.info(" Memory size: %d bytes", len(self.mem))
self.log.info(" Address width: %d bits", self.address_width)
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_lanes)
self.log.info("AXI lite slave model signals:")
for bus in (self.bus.aw, self.bus.w, self.bus.b):
for sig in sorted(list(set().union(bus._signals, bus._optional_signals))):
if hasattr(bus, sig):
self.log.info(" %s width: %d bits", sig, len(getattr(bus, sig)))
else:
self.log.info(" %s: not present", sig)
assert self.byte_lanes == len(self.w_channel.bus.wstrb)
assert self.byte_lanes * self.byte_size == self.width
self._process_write_cr = None
self._init_reset(reset, reset_active_level)
async def _write(self, address, data):
await self.target.write(address, data)
def _handle_reset(self, state):
if state:
self.log.info("Reset asserted")
if self._process_write_cr is not None:
self._process_write_cr.kill()
self._process_write_cr = None
self.aw_channel.clear()
self.w_channel.clear()
self.b_channel.clear()
else:
self.log.info("Reset de-asserted")
if self._process_write_cr is None:
self._process_write_cr = cocotb.fork(self._process_write())
async def _process_write(self):
while True:
aw = await self.aw_channel.recv()
addr = (int(aw.awaddr) // self.byte_lanes) * self.byte_lanes
prot = AxiProt(getattr(aw, 'awprot', AxiProt.NONSECURE))
w = await self.w_channel.recv()
data = int(w.wdata)
strb = int(getattr(w, 'wstrb', self.strb_mask))
# generate operation list
offset = 0
start_offset = None
write_ops = []
data = data.to_bytes(self.byte_lanes, 'little')
b = self.b_channel._transaction_obj()
b.bresp = AxiResp.OKAY
if self.log.isEnabledFor(logging.INFO):
self.log.info("Write data awaddr: 0x%08x awprot: %s wstrb: 0x%02x data: %s",
addr, prot, strb, ' '.join((f'{c:02x}' for c in data)))
for i in range(self.byte_lanes):
if strb & (1 << i):
if start_offset is None:
start_offset = offset
else:
if start_offset is not None and offset != start_offset:
write_ops.append((addr+start_offset, data[start_offset:offset]))
start_offset = None
offset += 1
if start_offset is not None and offset != start_offset:
write_ops.append((addr+start_offset, data[start_offset:offset]))
# perform writes
try:
for addr, data in write_ops:
await self._write(addr, data)
except Exception:
self.log.warning("Write operation failed")
b.bresp = AxiResp.SLVERR
await self.b_channel.send(b)
class AxiLiteSlaveRead(Reset):
def __init__(self, bus, clock, reset=None, target=None, reset_active_level=True, **kwargs):
self.bus = bus
self.clock = clock
self.reset = reset
self.target = target
self.log = logging.getLogger(f"cocotb.{bus.ar._entity._name}.{bus.ar._name}")
self.log.info("AXI lite slave model (read)")
self.log.info("cocotbext-axi version %s", __version__)
self.log.info("Copyright (c) 2021 Alex Forencich")
self.log.info("https://github.com/alexforencich/cocotbext-axi")
super().__init__(**kwargs)
self.ar_channel = AxiLiteARSink(bus.ar, clock, reset, reset_active_level)
self.ar_channel.queue_occupancy_limit = 2
self.r_channel = AxiLiteRSource(bus.r, clock, reset, reset_active_level)
self.r_channel.queue_occupancy_limit = 2
self.address_width = len(self.ar_channel.bus.araddr)
self.width = len(self.r_channel.bus.rdata)
self.byte_size = 8
self.byte_lanes = self.width // self.byte_size
self.log.info("AXI lite slave model configuration:")
self.log.info(" Memory size: %d bytes", len(self.mem))
self.log.info(" Address width: %d bits", self.address_width)
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_lanes)
self.log.info("AXI lite slave model signals:")
for bus in (self.bus.ar, self.bus.r):
for sig in sorted(list(set().union(bus._signals, bus._optional_signals))):
if hasattr(bus, sig):
self.log.info(" %s width: %d bits", sig, len(getattr(bus, sig)))
else:
self.log.info(" %s: not present", sig)
assert self.byte_lanes * self.byte_size == self.width
self._process_read_cr = None
self._init_reset(reset, reset_active_level)
async def _read(self, address, length):
return await self.target.read(address, length)
def _handle_reset(self, state):
if state:
self.log.info("Reset asserted")
if self._process_read_cr is not None:
self._process_read_cr.kill()
self._process_read_cr = None
self.ar_channel.clear()
self.r_channel.clear()
else:
self.log.info("Reset de-asserted")
if self._process_read_cr is None:
self._process_read_cr = cocotb.fork(self._process_read())
async def _process_read(self):
while True:
ar = await self.ar_channel.recv()
addr = (int(ar.araddr) // self.byte_lanes) * self.byte_lanes
prot = AxiProt(getattr(ar, 'arprot', AxiProt.NONSECURE))
r = self.r_channel._transaction_obj()
r.rresp = AxiResp.OKAY
try:
data = await self._read(addr, self.byte_lanes)
except Exception:
self.log.warning("Read operation failed")
data = bytes(self.byte_lanes)
r.rresp = AxiResp.SLVERR
r.rdata = int.from_bytes(data, 'little')
await self.r_channel.send(r)
if self.log.isEnabledFor(logging.INFO):
self.log.info("Read data araddr: 0x%08x arprot: %s data: %s",
addr, prot, ' '.join((f'{c:02x}' for c in data)))
class AxiLiteSlave:
def __init__(self, bus, clock, reset=None, target=None, reset_active_level=True, **kwargs):
self.write_if = None
self.read_if = None
super().__init__(**kwargs)
self.write_if = AxiLiteSlaveWrite(target, bus.write, clock, reset, reset_active_level)
self.read_if = AxiLiteSlaveRead(target, bus.read, clock, reset, reset_active_level)

View File

@@ -319,25 +319,13 @@ class AxiStreamBase(Reset):
self.log.info("AXI stream %s configuration:", self._type)
self.log.info(" Byte size: %d bits", self.byte_size)
self.log.info(" Data width: %d bits (%d bytes)", self.width, self.byte_lanes)
self.log.info(" tvalid: %s", "present" if hasattr(self.bus, "tvalid") else "not present")
self.log.info(" tready: %s", "present" if hasattr(self.bus, "tready") else "not present")
self.log.info(" tlast: %s", "present" if hasattr(self.bus, "tlast") else "not present")
if hasattr(self.bus, "tkeep"):
self.log.info(" tkeep width: %d bits", len(self.bus.tkeep))
self.log.info("AXI stream %s signals:", self._type)
for sig in sorted(list(set().union(self.bus._signals, self.bus._optional_signals))):
if hasattr(self.bus, sig):
self.log.info(" %s width: %d bits", sig, len(getattr(self.bus, sig)))
else:
self.log.info(" tkeep: not present")
if hasattr(self.bus, "tid"):
self.log.info(" tid width: %d bits", len(self.bus.tid))
else:
self.log.info(" tid: not present")
if hasattr(self.bus, "tdest"):
self.log.info(" tdest width: %d bits", len(self.bus.tdest))
else:
self.log.info(" tdest: not present")
if hasattr(self.bus, "tuser"):
self.log.info(" tuser width: %d bits", len(self.bus.tuser))
else:
self.log.info(" tuser: not present")
self.log.info(" %s: not present", sig)
if self.byte_lanes * self.byte_size != self.width:
raise ValueError(f"Bus does not evenly divide into byte lanes "
@@ -471,19 +459,19 @@ class AxiStreamSource(AxiStreamBase, AxiStreamPause):
super()._handle_reset(state)
if state:
self.bus.tdata <= 0
self.bus.tdata.value = 0
if hasattr(self.bus, "tvalid"):
self.bus.tvalid <= 0
self.bus.tvalid.value = 0
if hasattr(self.bus, "tlast"):
self.bus.tlast <= 0
self.bus.tlast.value = 0
if hasattr(self.bus, "tkeep"):
self.bus.tkeep <= 0
self.bus.tkeep.value = 0
if hasattr(self.bus, "tid"):
self.bus.tid <= 0
self.bus.tid.value = 0
if hasattr(self.bus, "tdest"):
self.bus.tdest <= 0
self.bus.tdest.value = 0
if hasattr(self.bus, "tuser"):
self.bus.tuser <= 0
self.bus.tuser.value = 0
if self.current_frame:
self.log.warning("Flushed transmit frame during reset: %s", self.current_frame)
@@ -492,6 +480,7 @@ class AxiStreamSource(AxiStreamBase, AxiStreamPause):
async def _run(self):
frame = None
frame_offset = 0
self.active = False
while True:
@@ -513,6 +502,7 @@ class AxiStreamSource(AxiStreamBase, AxiStreamPause):
self.log.info("TX frame: %s", frame)
frame.normalize()
self.active = True
frame_offset = 0
if frame and not self.pause:
tdata_val = 0
@@ -523,13 +513,14 @@ class AxiStreamSource(AxiStreamBase, AxiStreamPause):
tuser_val = 0
for offset in range(self.byte_lanes):
tdata_val |= (frame.tdata.pop(0) & self.byte_mask) << (offset * self.byte_size)
tkeep_val |= (frame.tkeep.pop(0) & 1) << offset
tid_val = frame.tid.pop(0)
tdest_val = frame.tdest.pop(0)
tuser_val = frame.tuser.pop(0)
tdata_val |= (frame.tdata[frame_offset] & self.byte_mask) << (offset * self.byte_size)
tkeep_val |= (frame.tkeep[frame_offset] & 1) << offset
tid_val = frame.tid[frame_offset]
tdest_val = frame.tdest[frame_offset]
tuser_val = frame.tuser[frame_offset]
frame_offset += 1
if len(frame.tdata) == 0:
if frame_offset >= len(frame.tdata):
tlast_val = 1
frame.sim_time_end = get_sim_time()
frame.handle_tx_complete()
@@ -537,24 +528,24 @@ class AxiStreamSource(AxiStreamBase, AxiStreamPause):
self.current_frame = None
break
self.bus.tdata <= tdata_val
self.bus.tdata.value = tdata_val
if hasattr(self.bus, "tvalid"):
self.bus.tvalid <= 1
self.bus.tvalid.value = 1
if hasattr(self.bus, "tlast"):
self.bus.tlast <= tlast_val
self.bus.tlast.value = tlast_val
if hasattr(self.bus, "tkeep"):
self.bus.tkeep <= tkeep_val
self.bus.tkeep.value = tkeep_val
if hasattr(self.bus, "tid"):
self.bus.tid <= tid_val
self.bus.tid.value = tid_val
if hasattr(self.bus, "tdest"):
self.bus.tdest <= tdest_val
self.bus.tdest.value = tdest_val
if hasattr(self.bus, "tuser"):
self.bus.tuser <= tuser_val
self.bus.tuser.value = tuser_val
else:
if hasattr(self.bus, "tvalid"):
self.bus.tvalid <= 0
self.bus.tvalid.value = 0
if hasattr(self.bus, "tlast"):
self.bus.tlast <= 0
self.bus.tlast.value = 0
self.active = bool(frame)
if not frame and self.queue.empty():
self.idle_event.set()
@@ -638,9 +629,9 @@ class AxiStreamMonitor(AxiStreamBase):
else:
frame = AxiStreamFrame([], [], [], [], [])
frame.sim_time_start = get_sim_time()
self.active = True
for offset in range(self.byte_lanes):
frame.tdata.append((self.bus.tdata.value.integer >> (offset * self.byte_size)) & self.byte_mask)
if hasattr(self.bus, "tkeep"):
frame.tkeep.append((self.bus.tkeep.value.integer >> offset) & 1)
@@ -662,6 +653,8 @@ class AxiStreamMonitor(AxiStreamBase):
self.active_event.set()
frame = None
else:
self.active = bool(frame)
class AxiStreamSink(AxiStreamMonitor, AxiStreamPause):
@@ -694,7 +687,7 @@ class AxiStreamSink(AxiStreamMonitor, AxiStreamPause):
if state:
if hasattr(self.bus, "tready"):
self.bus.tready <= 0
self.bus.tready.value = 0
async def _run(self):
frame = None
@@ -714,9 +707,9 @@ class AxiStreamSink(AxiStreamMonitor, AxiStreamPause):
else:
frame = AxiStreamFrame([], [], [], [], [])
frame.sim_time_start = get_sim_time()
self.active = True
for offset in range(self.byte_lanes):
frame.tdata.append((self.bus.tdata.value.integer >> (offset * self.byte_size)) & self.byte_mask)
if hasattr(self.bus, "tkeep"):
frame.tkeep.append((self.bus.tkeep.value.integer >> offset) & 1)
@@ -738,6 +731,8 @@ class AxiStreamSink(AxiStreamMonitor, AxiStreamPause):
self.active_event.set()
frame = None
else:
self.active = bool(frame)
if hasattr(self.bus, "tready"):
self.bus.tready <= (not self.full() and not self.pause)
self.bus.tready.value = (not self.full() and not self.pause)

View File

@@ -0,0 +1,92 @@
"""
Copyright (c) 2021 Alex Forencich
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""
class BuddyAllocator:
def __init__(self, size, min_alloc=1):
self.size = size
self.min_alloc = min_alloc
self.free_lists = [[] for x in range((self.size-1).bit_length())]
self.free_lists.append([0])
self.allocations = {}
def alloc(self, size):
if size < 1 or size > self.size:
raise ValueError("size out of range")
size = max(size, self.min_alloc)
bucket = (size-1).bit_length()
orig_bucket = bucket
while bucket < len(self.free_lists):
if not self.free_lists[bucket]:
# find free block
bucket += 1
continue
while bucket > orig_bucket:
# split block
block = self.free_lists[bucket].pop(0)
bucket -= 1
self.free_lists[bucket].append(block)
self.free_lists[bucket].append(block+2**bucket)
if self.free_lists[bucket]:
# allocate
block = self.free_lists[bucket].pop(0)
self.allocations[block] = bucket
return block
break
raise Exception("out of memory")
def free(self, addr):
if addr not in self.allocations:
raise ValueError("unknown allocation")
bucket = self.allocations.pop(addr)
while bucket < len(self.free_lists):
size = 2**bucket
# find buddy
if (addr // size) % 2:
buddy = addr - size
else:
buddy = addr + size
if buddy in self.free_lists[bucket]:
# buddy is free, merge
self.free_lists[bucket].remove(buddy)
addr = min(addr, buddy)
bucket += 1
else:
# buddy is not free, so add to free list
self.free_lists[bucket].append(addr)
return
raise Exception("failed to free memory")

View File

@@ -28,13 +28,13 @@ from .utils import hexdump, hexdump_lines, hexdump_str
class Memory:
def __init__(self, size=1024, mem=None, *args, **kwargs):
def __init__(self, size=1024, mem=None, **kwargs):
if mem is not None:
self.mem = mem
else:
self.mem = mmap.mmap(-1, size)
self.size = len(self.mem)
super().__init__(*args, **kwargs)
super().__init__(**kwargs)
def read(self, address, length):
self.mem.seek(address)

View File

@@ -228,7 +228,7 @@ class StreamSource(StreamBase, StreamPause):
if state:
if self.valid is not None:
self.valid <= 0
self.valid.value = 0
async def _run(self):
while True:
@@ -243,11 +243,11 @@ class StreamSource(StreamBase, StreamPause):
self.bus.drive(self.queue.get_nowait())
self.dequeue_event.set()
if self.valid is not None:
self.valid <= 1
self.valid.value = 1
self.active = True
else:
if self.valid is not None:
self.valid <= 0
self.valid.value = 0
self.active = not self.queue.empty()
if self.queue.empty():
self.idle_event.set()
@@ -319,7 +319,7 @@ class StreamSink(StreamMonitor, StreamPause):
if state:
if self.ready is not None:
self.ready <= 0
self.ready.value = 0
async def _run(self):
while True:
@@ -336,7 +336,7 @@ class StreamSink(StreamMonitor, StreamPause):
self.active_event.set()
if self.ready is not None:
self.ready <= (not self.full() and not self.pause)
self.ready.value = (not self.full() and not self.pause)
def define_stream(name, signals, optional_signals=None, valid_signal=None, ready_signal=None, signal_widths=None):

View File

@@ -1 +1 @@
__version__ = "0.1.10"
__version__ = "0.1.16"

View File

@@ -17,9 +17,10 @@ long-description-content-type = text/markdown
platforms = any
classifiers =
Development Status :: 3 - Alpha
Programming Language :: Python :: 3
Framework :: cocotb
License :: OSI Approved :: MIT License
Operating System :: OS Independent
Programming Language :: Python :: 3
Topic :: Scientific/Engineering :: Electronic Design Automation (EDA)
[options]

View File

@@ -73,10 +73,10 @@ class TB:
self.dut.rst.setimmediatevalue(0)
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
self.dut.rst <= 1
self.dut.rst.value = 1
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
self.dut.rst <= 0
self.dut.rst.value = 0
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
@@ -85,7 +85,7 @@ async def run_test_write(dut, idle_inserter=None, backpressure_inserter=None, si
tb = TB(dut)
byte_width = tb.axi_master.write_if.byte_width
byte_lanes = tb.axi_master.write_if.byte_lanes
max_burst_size = tb.axi_master.write_if.max_burst_size
if size is None:
@@ -96,8 +96,8 @@ async def run_test_write(dut, idle_inserter=None, backpressure_inserter=None, si
tb.set_idle_generator(idle_inserter)
tb.set_backpressure_generator(backpressure_inserter)
for length in list(range(1, byte_width*2))+[1024]:
for offset in list(range(byte_width))+list(range(4096-byte_width, 4096)):
for length in list(range(1, byte_lanes*2))+[1024]:
for offset in list(range(byte_lanes))+list(range(4096-byte_lanes, 4096)):
tb.log.info("length %d, offset %d", length, offset)
addr = offset+0x1000
test_data = bytearray([x % 256 for x in range(length)])
@@ -120,7 +120,7 @@ async def run_test_read(dut, idle_inserter=None, backpressure_inserter=None, siz
tb = TB(dut)
byte_width = tb.axi_master.write_if.byte_width
byte_lanes = tb.axi_master.write_if.byte_lanes
max_burst_size = tb.axi_master.write_if.max_burst_size
if size is None:
@@ -131,8 +131,8 @@ async def run_test_read(dut, idle_inserter=None, backpressure_inserter=None, siz
tb.set_idle_generator(idle_inserter)
tb.set_backpressure_generator(backpressure_inserter)
for length in list(range(1, byte_width*2))+[1024]:
for offset in list(range(byte_width))+list(range(4096-byte_width, 4096)):
for length in list(range(1, byte_lanes*2))+[1024]:
for offset in list(range(byte_lanes))+list(range(4096-byte_lanes, 4096)):
tb.log.info("length %d, offset %d", length, offset)
addr = offset+0x1000
test_data = bytearray([x % 256 for x in range(length)])
@@ -151,12 +151,12 @@ async def run_test_write_words(dut):
tb = TB(dut)
byte_width = tb.axi_master.write_if.byte_width
byte_lanes = tb.axi_master.write_if.byte_lanes
await tb.cycle_reset()
for length in list(range(1, 4)):
for offset in list(range(byte_width)):
for offset in list(range(byte_lanes)):
tb.log.info("length %d, offset %d", length, offset)
addr = offset+0x1000
@@ -205,12 +205,12 @@ async def run_test_read_words(dut):
tb = TB(dut)
byte_width = tb.axi_master.write_if.byte_width
byte_lanes = tb.axi_master.write_if.byte_lanes
await tb.cycle_reset()
for length in list(range(1, 4)):
for offset in list(range(byte_width)):
for offset in list(range(byte_lanes)):
tb.log.info("length %d, offset %d", length, offset)
addr = offset+0x1000
@@ -299,8 +299,8 @@ def cycle_pause():
if cocotb.SIM_NAME:
data_width = len(cocotb.top.axi_wdata)
byte_width = data_width // 8
max_burst_size = (byte_width-1).bit_length()
byte_lanes = data_width // 8
max_burst_size = (byte_lanes-1).bit_length()
for test in [run_test_write, run_test_read]:

View File

@@ -70,10 +70,10 @@ class TB:
self.dut.rst.setimmediatevalue(0)
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
self.dut.rst <= 1
self.dut.rst.value = 1
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
self.dut.rst <= 0
self.dut.rst.value = 0
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
@@ -82,15 +82,15 @@ async def run_test_write(dut, data_in=None, idle_inserter=None, backpressure_ins
tb = TB(dut)
byte_width = tb.axil_master.write_if.byte_width
byte_lanes = tb.axil_master.write_if.byte_lanes
await tb.cycle_reset()
tb.set_idle_generator(idle_inserter)
tb.set_backpressure_generator(backpressure_inserter)
for length in range(1, byte_width*2):
for offset in range(byte_width):
for length in range(1, byte_lanes*2):
for offset in range(byte_lanes):
tb.log.info("length %d, offset %d", length, offset)
addr = offset+0x1000
test_data = bytearray([x % 256 for x in range(length)])
@@ -113,15 +113,15 @@ async def run_test_read(dut, data_in=None, idle_inserter=None, backpressure_inse
tb = TB(dut)
byte_width = tb.axil_master.write_if.byte_width
byte_lanes = tb.axil_master.write_if.byte_lanes
await tb.cycle_reset()
tb.set_idle_generator(idle_inserter)
tb.set_backpressure_generator(backpressure_inserter)
for length in range(1, byte_width*2):
for offset in range(byte_width):
for length in range(1, byte_lanes*2):
for offset in range(byte_lanes):
tb.log.info("length %d, offset %d", length, offset)
addr = offset+0x1000
test_data = bytearray([x % 256 for x in range(length)])
@@ -140,12 +140,12 @@ async def run_test_write_words(dut):
tb = TB(dut)
byte_width = tb.axil_master.write_if.byte_width
byte_lanes = tb.axil_master.write_if.byte_lanes
await tb.cycle_reset()
for length in list(range(1, 4)):
for offset in list(range(byte_width)):
for offset in list(range(byte_lanes)):
tb.log.info("length %d, offset %d", length, offset)
addr = offset+0x1000
@@ -194,12 +194,12 @@ async def run_test_read_words(dut):
tb = TB(dut)
byte_width = tb.axil_master.write_if.byte_width
byte_lanes = tb.axil_master.write_if.byte_lanes
await tb.cycle_reset()
for length in list(range(1, 4)):
for offset in list(range(byte_width)):
for offset in list(range(byte_lanes)):
tb.log.info("length %d, offset %d", length, offset)
addr = offset+0x1000

View File

@@ -63,10 +63,10 @@ class TB:
self.dut.rst.setimmediatevalue(0)
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
self.dut.rst <= 1
self.dut.rst.value = 1
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
self.dut.rst <= 0
self.dut.rst.value = 0
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)

View File

@@ -0,0 +1,44 @@
"""
Copyright (c) 2021 Alex Forencich
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""
from cocotbext.axi.buddy_allocator import BuddyAllocator
def test_allocator():
ba = BuddyAllocator(1024)
lst = []
for k in range(1, 32):
print(f"Alloc {k} bytes")
addr = ba.alloc(k)
print(f"Got address {addr}")
assert addr & (2**((k-1).bit_length())-1) == 0
lst.append(addr)
for addr in lst:
print(f"Free {addr}")
ba.free(addr)
assert ba.free_lists[-1] == [0]