bslathi19/cocotbext-axi
1
0
Fork 0
Code Issues Actions Packages Projects Releases 5 Wiki Activity

Initial commit

Browse Source
This commit is contained in:
Alex Forencich
2020-10-19 14:54:41 -07:00
commit 35ff3fa3b6
11 changed files with 2873 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
LICENSE Normal file
View File

@@ -0,0 +1,21 @@
Copyright (c) 2020 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.

2
README.md Normal file
View File

@@ -0,0 +1,2 @@
# AXI interface modules for Cocotb

34
cocotbext/axi/__init__.py Normal file
View File

@@ -0,0 +1,34 @@
"""
Copyright (c) 2020 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 .constants import *
from .axis import AxiStreamFrame, AxiStreamSource, AxiStreamSink
from .axil_master import AxiLiteMasterWrite, AxiLiteMasterRead, AxiLiteMaster
from .axil_ram import AxiLiteRamWrite, AxiLiteRamRead, AxiLiteRam
from .axi_master import AxiMasterWrite, AxiMasterRead, AxiMaster
from .axi_ram import AxiRamWrite, AxiRamRead, AxiRam

808
cocotbext/axi/axi_master.py Normal file
View File

@@ -0,0 +1,808 @@
"""
Copyright (c) 2020 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 cocotb
from cocotb.triggers import RisingEdge, ReadOnly, Event
from cocotb.drivers import BusDriver
from collections import deque
from .constants import *
class AxiMasterWrite(BusDriver):
_signals = [
# Write address channel
"awid", "awaddr", "awlen", "awsize", "awburst", "awprot", "awvalid", "awready",
# Write data channel
"wdata", "wstrb", "wlast", "wvalid", "wready",
# Write response channel
"bid", "bresp", "bvalid", "bready",
]
_optional_signals = [
# Write address channel
"awlock", "awcache", "awqos", "awregion", "awuser",
# Write data channel
"wuser",
# Write response channel
"buser",
]
def __init__(self, entity, name, clock, reset=None):
super().__init__(entity, name, clock)
self.active_tokens = set()
self.write_command_queue = deque()
self.write_command_sync = Event()
self.write_resp_queue = deque()
self.write_resp_sync = Event()
self.write_resp_set = set()
self.id_queue = deque(range(2**len(self.bus.awid)))
self.id_sync = Event()
self.int_write_addr_queue = deque()
self.int_write_data_queue = deque()
self.int_write_resp_command_queue = deque()
self.int_write_resp_command_sync = Event()
self.int_write_resp_queue_list = {}
self.int_write_resp_sync_list = {}
self.in_flight_operations = 0
self.width = len(self.bus.wdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.strb_mask = 2**len(self.bus.wstrb)-1
self.max_burst_len = 256
self.max_burst_size = (self.byte_width-1).bit_length()
assert self.byte_width == len(self.bus.wstrb)
assert self.byte_width * self.byte_size == self.width
self.reset = reset
self.bus.awid.setimmediatevalue(0)
self.bus.awaddr.setimmediatevalue(0)
assert len(self.bus.awlen) == 8
self.bus.awlen.setimmediatevalue(0)
assert len(self.bus.awsize) == 3
self.bus.awsize.setimmediatevalue(0)
assert len(self.bus.awburst) == 2
self.bus.awburst.setimmediatevalue(0)
if hasattr(self.bus, "awlock"):
assert len(self.bus.awlock) == 1
self.bus.awlock.setimmediatevalue(0)
if hasattr(self.bus, "awcache"):
assert len(self.bus.awcache) == 4
self.bus.awcache.setimmediatevalue(0)
assert len(self.bus.awprot) == 3
self.bus.awprot.setimmediatevalue(0)
if hasattr(self.bus, "awqos"):
assert len(self.bus.awqos) == 4
self.bus.awqos.setimmediatevalue(0)
if hasattr(self.bus, "awregion"):
assert len(self.bus.awregion) == 4
self.bus.awregion.setimmediatevalue(0)
if hasattr(self.bus, "awuser"):
self.bus.awuser.setimmediatevalue(0)
assert len(self.bus.awvalid) == 1
self.bus.awvalid.setimmediatevalue(0)
assert len(self.bus.awready) == 1
self.bus.wdata.setimmediatevalue(0)
self.bus.wstrb.setimmediatevalue(0)
assert len(self.bus.wlast) == 1
self.bus.wlast.setimmediatevalue(0)
if hasattr(self.bus, "wuser"):
self.bus.wuser.setimmediatevalue(0)
assert len(self.bus.wvalid) == 1
self.bus.wvalid.setimmediatevalue(0)
assert len(self.bus.wready) == 1
assert len(self.bus.bid) == len(self.bus.awid)
assert len(self.bus.bresp) == 2
assert len(self.bus.bvalid) == 1
assert len(self.bus.bready) == 1
self.bus.bready.setimmediatevalue(0)
cocotb.fork(self._process_write())
cocotb.fork(self._process_write_resp())
cocotb.fork(self._process_write_addr_if())
cocotb.fork(self._process_write_data_if())
cocotb.fork(self._process_write_resp_if())
def init_write(self, address, data, burst=AxiBurstType.INCR, size=None, lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, token=None):
if token is not None:
if token in self.active_tokens:
raise Exception("Token is not unique")
self.active_tokens.add(token)
self.in_flight_operations += 1
self.write_command_queue.append((address, data, burst, size, lock, cache, prot, qos, region, user, token))
self.write_command_sync.set()
def idle(self):
return not self.in_flight_operations
async def wait(self):
while not self.idle():
self.write_resp_sync.clear()
await self.write_resp_sync.wait()
async def wait_for_token(self, token):
if token not in self.active_tokens:
return
while token not in self.write_resp_set:
self.write_resp_sync.clear()
await self.write_resp_sync.wait()
def write_resp_ready(self, token=None):
if token is not None:
return token in self.write_resp_set
return bool(self.write_resp_queue)
def get_write_resp(self, token=None):
if token is not None:
if token in self.write_resp_set:
for resp in self.write_resp_queue:
if resp[-1] == token:
self.write_resp_queue.remove(resp)
self.active_tokens.remove(resp[-1])
self.write_resp_set.remove(resp[-1])
return resp
return None
if self.write_resp_queue:
resp = self.write_resp_queue.popleft()
if resp[-1] is not None:
self.active_tokens.remove(resp[-1])
self.write_resp_set.remove(resp[-1])
return resp
return None
async def write(self, address, data, burst=AxiBurstType.INCR, size=None, lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
token = object()
self.init_write(address, data, burst, size, lock, cache, prot, qos, region, user, token)
await self.wait_for_token(token)
return self.get_write_resp(token)
async def _process_write(self):
while True:
if not self.write_command_queue:
self.write_command_sync.clear()
await self.write_command_sync.wait()
address, data, burst, size, lock, cache, prot, qos, region, user, token = self.write_command_queue.popleft()
num_bytes = self.byte_width
if size is None:
size = self.max_burst_size
else:
num_bytes = 2**size
assert 0 < num_bytes <= self.byte_width
aligned_addr = (address // num_bytes) * num_bytes
word_addr = (address // self.byte_width) * self.byte_width
start_offset = address % self.byte_width
end_offset = ((address + len(data) - 1) % self.byte_width) + 1
cycles = (len(data) + (address % num_bytes) + num_bytes-1) // num_bytes
cur_addr = aligned_addr
offset = 0
cycle_offset = aligned_addr-word_addr
n = 0
transfer_count = 0
burst_list = []
burst_length = 0
self.log.info(f"Write start addr: {address:#010x} prot: {prot} data: {' '.join((f'{c:02x}' for c in data))}")
for k in range(cycles):
start = cycle_offset
stop = cycle_offset+num_bytes
if k == 0:
start = start_offset
if k == cycles-1:
stop = end_offset
strb = (self.strb_mask << start) & self.strb_mask & (self.strb_mask >> (self.byte_width - stop))
val = 0
for j in range(start, stop):
val |= bytearray(data)[offset] << j*8
offset += 1
if n >= burst_length:
if not self.id_queue:
self.id_sync.clear()
await self.id_sync.wait()
awid = self.id_queue.popleft()
transfer_count += 1
n = 0
burst_length = min(cycles-k, min(max(self.max_burst_len, 1), 256)) # max len
burst_length = (min(burst_length*num_bytes, 0x1000-(cur_addr&0xfff))+num_bytes-1)//num_bytes # 4k align
burst_list.append((awid, burst_length))
self.int_write_addr_queue.append((cur_addr, awid, burst_length-1, size, burst, lock, cache, prot, qos, region, user))
self.log.info(f"Write burst start awid {awid:#x} awaddr: {cur_addr:#010x} awlen: {burst_length-1} awsize: {size}")
n += 1
self.int_write_data_queue.append((val, strb, n >= burst_length, 0))
cur_addr += num_bytes
cycle_offset = (cycle_offset + num_bytes) % self.byte_width
self.int_write_resp_command_queue.append((address, len(data), size, cycles, prot, burst_list, token))
self.int_write_resp_command_sync.set()
async def _process_write_resp(self):
while True:
if not self.int_write_resp_command_queue:
self.int_write_resp_command_sync.clear()
await self.int_write_resp_command_sync.wait()
addr, length, size, cycles, prot, burst_list, token = self.int_write_resp_command_queue.popleft()
resp = AxiResp.OKAY
user = []
for bid, burst_length in burst_list:
self.int_write_resp_queue_list.setdefault(bid, deque())
self.int_write_resp_sync_list.setdefault(bid, Event())
if not self.int_write_resp_queue_list[bid]:
self.int_write_resp_sync_list[bid].clear()
await self.int_write_resp_sync_list[bid].wait()
burst_id, burst_resp, burst_user = self.int_write_resp_queue_list[bid].popleft()
burst_resp = AxiResp(burst_resp)
if burst_resp != AxiResp.OKAY:
resp = burst_resp
if burst_user is not None:
user.append(burst_user)
if bid in self.id_queue:
raise Exception(f"Unexpected burst ID {bid}")
self.id_queue.append(bid)
self.id_sync.set()
self.log.info(f"Write burst complete bid {burst_id:#x} bresp: {burst_resp!s}")
self.log.info(f"Write complete addr: {addr:#010x} prot: {prot} resp: {resp!s} length: {length}")
self.write_resp_queue.append((addr, length, resp, user, token))
self.write_resp_sync.set()
if token is not None:
self.write_resp_set.add(token)
self.in_flight_operations -= 1
async def _process_write_addr_if(self):
while True:
await ReadOnly()
# read handshake signals
awready_sample = self.bus.awready.value
awvalid_sample = self.bus.awvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.awvalid <= 0
continue
await RisingEdge(self.clock)
if (awready_sample and awvalid_sample) or (not awvalid_sample):
if self.int_write_addr_queue:
addr, awid, length, size, burst, lock, cache, prot, qos, region, user = self.int_write_addr_queue.popleft()
self.bus.awaddr <= addr
self.bus.awid <= awid
self.bus.awlen <= length
self.bus.awsize <= size
self.bus.awburst <= burst
if hasattr(self.bus, "awlock"):
self.bus.awlock <= lock
if hasattr(self.bus, "awcache"):
self.bus.awcache <= cache
self.bus.awprot <= prot
if hasattr(self.bus, "awqos"):
self.bus.awqos <= qos
if hasattr(self.bus, "awregion"):
self.bus.awregion <= region
if hasattr(self.bus, "awuser"):
self.bus.awuser <= user
self.bus.awvalid <= 1
else:
self.bus.awvalid <= 0
async def _process_write_data_if(self):
while True:
await ReadOnly()
# read handshake signals
wready_sample = self.bus.wready.value
wvalid_sample = self.bus.wvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.wvalid <= 0
continue
await RisingEdge(self.clock)
if (wready_sample and wvalid_sample) or (not wvalid_sample):
if self.int_write_data_queue:
data, strb, last, user = self.int_write_data_queue.popleft()
self.bus.wdata <= data
self.bus.wstrb <= strb
self.bus.wlast <= last
if hasattr(self.bus, "awuser"):
self.bus.awuser <= user
self.bus.wvalid <= 1
else:
self.bus.wvalid <= 0
async def _process_write_resp_if(self):
while True:
await ReadOnly()
# read handshake signals
bready_sample = self.bus.bready.value
bvalid_sample = self.bus.bvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.bready <= 0
continue
if bready_sample and bvalid_sample:
bid = self.bus.bid.value.integer
bresp = self.bus.bresp.value.integer
buser = self.bus.buser.value.integer if hasattr(self.bus, "buser") else None
self.int_write_resp_queue_list.setdefault(bid, deque())
self.int_write_resp_queue_list[bid].append((bid, bresp, buser))
self.int_write_resp_sync_list.setdefault(bid, Event())
self.int_write_resp_sync_list[bid].set()
await RisingEdge(self.clock)
self.bus.bready <= 1
class AxiMasterRead(BusDriver):
_signals = [
# Read address channel
"arid", "araddr", "arlen", "arsize", "arburst", "arprot", "arvalid", "arready",
# Read data channel
"rid", "rdata", "rresp", "rlast", "rvalid", "rready",
]
_optional_signals = [
# Read address channel
"arlock", "arcache", "arqos", "arregion", "aruser",
# Read data channel
"ruser",
]
def __init__(self, entity, name, clock, reset=None):
super().__init__(entity, name, clock)
self.active_tokens = set()
self.read_command_queue = deque()
self.read_command_sync = Event()
self.read_data_queue = deque()
self.read_data_sync = Event()
self.read_data_set = set()
self.id_queue = deque(range(2**len(self.bus.arid)))
self.id_sync = Event()
self.int_read_addr_queue = deque()
self.int_read_resp_command_queue = deque()
self.int_read_resp_command_sync = Event()
self.int_read_resp_queue_list = {}
self.int_read_resp_sync_list = {}
self.in_flight_operations = 0
self.width = len(self.bus.rdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.max_burst_len = 256
self.max_burst_size = (self.byte_width-1).bit_length()
assert self.byte_width * self.byte_size == self.width
self.reset = reset
self.bus.arid.setimmediatevalue(0)
self.bus.araddr.setimmediatevalue(0)
assert len(self.bus.arlen) == 8
self.bus.arlen.setimmediatevalue(0)
assert len(self.bus.arsize) == 3
self.bus.arsize.setimmediatevalue(0)
assert len(self.bus.arburst) == 2
self.bus.arburst.setimmediatevalue(0)
if hasattr(self.bus, "arlock"):
assert len(self.bus.arlock) == 1
self.bus.arlock.setimmediatevalue(0)
if hasattr(self.bus, "arcache"):
assert len(self.bus.arcache) == 4
self.bus.arcache.setimmediatevalue(0)
assert len(self.bus.arprot) == 3
self.bus.arprot.setimmediatevalue(0)
if hasattr(self.bus, "arqos"):
assert len(self.bus.arqos) == 4
self.bus.arqos.setimmediatevalue(0)
if hasattr(self.bus, "arregion"):
assert len(self.bus.arregion) == 4
self.bus.arregion.setimmediatevalue(0)
if hasattr(self.bus, "aruser"):
self.bus.aruser.setimmediatevalue(0)
assert len(self.bus.arvalid) == 1
self.bus.arvalid.setimmediatevalue(0)
assert len(self.bus.arready) == 1
assert len(self.bus.rid) == len(self.bus.arid)
assert len(self.bus.rresp) == 2
assert len(self.bus.rlast) == 1
assert len(self.bus.rvalid) == 1
assert len(self.bus.rready) == 1
self.bus.rready.setimmediatevalue(0)
cocotb.fork(self._process_read())
cocotb.fork(self._process_read_resp())
cocotb.fork(self._process_read_addr_if())
cocotb.fork(self._process_read_resp_if())
def init_read(self, address, length, burst=AxiBurstType.INCR, size=None, lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, token=None):
if token is not None:
if token in self.active_tokens:
raise Exception("Token is not unique")
self.active_tokens.add(token)
self.in_flight_operations += 1
self.read_command_queue.append((address, length, burst, size, lock, cache, prot, qos, region, user, token))
self.read_command_sync.set()
def idle(self):
return not self.in_flight_operations
async def wait(self):
while not self.idle():
self.read_resp_sync.clear()
await self.read_resp_sync.wait()
async def wait_for_token(self, token):
if token not in self.active_tokens:
return
while token not in self.read_data_set:
self.read_data_sync.clear()
await self.read_data_sync.wait()
def read_data_ready(self, token=None):
if token is not None:
return token in self.read_data_set
return bool(self.read_data_queue)
def get_read_data(self, token=None):
if token is not None:
if token in self.read_data_set:
for resp in self.read_data_queue:
if resp[-1] == token:
self.read_data_queue.remove(resp)
self.active_tokens.remove(resp[-1])
self.read_data_set.remove(resp[-1])
return resp
return None
if self.read_data_queue:
resp = self.read_data_queue.popleft()
if resp[-1] is not None:
self.active_tokens.remove(resp[-1])
self.read_data_set.remove(resp[-1])
return resp
return None
async def read(self, address, length, burst=AxiBurstType.INCR, size=None, lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
token = object()
self.init_read(address, length, burst, size, lock, cache, prot, qos, region, user, token)
await self.wait_for_token(token)
return self.get_read_data(token)
async def _process_read(self):
while True:
if not self.read_command_queue:
self.read_command_sync.clear()
await self.read_command_sync.wait()
address, length, burst, size, lock, cache, prot, qos, region, user, token = self.read_command_queue.popleft()
num_bytes = self.byte_width
if size is None:
size = self.max_burst_size
else:
num_bytes = 2**size
assert 0 < num_bytes <= self.byte_width
aligned_addr = (address // num_bytes) * num_bytes
word_addr = (address // self.byte_width) * self.byte_width
cycles = (length + num_bytes-1 + (address % num_bytes)) // num_bytes
burst_list = []
cur_addr = aligned_addr
n = 0
burst_length = 0
for k in range(cycles):
n += 1
if n >= burst_length:
if not self.id_queue:
self.id_sync.clear()
await self.id_sync.wait()
arid = self.id_queue.popleft()
n = 0
burst_length = min(cycles-k, min(max(self.max_burst_len, 1), 256)) # max len
burst_length = (min(burst_length*num_bytes, 0x1000-(cur_addr&0xfff))+num_bytes-1)//num_bytes # 4k align
burst_list.append((arid, burst_length))
self.int_read_addr_queue.append((cur_addr, arid, burst_length-1, size, burst, lock, cache, prot, qos, region, user))
self.log.info(f"Read burst start arid {arid:#x} araddr: {cur_addr:#010x} arlen: {burst_length-1} arsize: {size}")
cur_addr += num_bytes
self.int_read_resp_command_queue.append((address, length, size, cycles, prot, burst_list, token))
self.int_read_resp_command_sync.set()
async def _process_read_resp(self):
while True:
if not self.int_read_resp_command_queue:
self.int_read_resp_command_sync.clear()
await self.int_read_resp_command_sync.wait()
addr, length, size, cycles, prot, burst_list, token = self.int_read_resp_command_queue.popleft()
num_bytes = 2**size
aligned_addr = (addr // num_bytes) * num_bytes
word_addr = (addr // self.byte_width) * self.byte_width
start_offset = addr % self.byte_width
end_offset = ((addr + length - 1) % self.byte_width) + 1
cycle_offset = aligned_addr - word_addr
data = bytearray()
resp = AxiResp.OKAY
user = []
first = True
for rid, burst_length in burst_list:
for k in range(burst_length):
self.int_read_resp_queue_list.setdefault(rid, deque())
self.int_read_resp_sync_list.setdefault(rid, Event())
if not self.int_read_resp_queue_list[rid]:
self.int_read_resp_sync_list[rid].clear()
await self.int_read_resp_sync_list[rid].wait()
cycle_id, cycle_data, cycle_resp, cycle_last, cycle_user = self.int_read_resp_queue_list[rid].popleft()
cycle_resp = AxiResp(cycle_resp)
if cycle_resp != AxiResp.OKAY:
resp = cycle_resp
if cycle_user is not None:
user.append(cycle_user)
start = cycle_offset
stop = cycle_offset+num_bytes
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
first = False
if rid in self.id_queue:
raise Exception(f"Unexpected burst ID {rid}")
self.id_queue.append(rid)
self.id_sync.set()
self.log.info(f"Read burst complete rid {cycle_id:#x} rresp: {resp!s}")
data = data[:length]
self.log.info(f"Read complete addr: {addr:#010x} prot: {prot} resp: {resp!s} data: {' '.join((f'{c:02x}' for c in data))}")
self.read_data_queue.append((addr, data, resp, user, token))
self.read_data_sync.set()
if token is not None:
self.read_data_set.add(token)
self.in_flight_operations -= 1
async def _process_read_addr_if(self):
while True:
await ReadOnly()
# read handshake signals
arready_sample = self.bus.arready.value
arvalid_sample = self.bus.arvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.arvalid <= 0
continue
await RisingEdge(self.clock)
if (arready_sample and arvalid_sample) or (not arvalid_sample):
if self.int_read_addr_queue:
addr, arid, length, size, burst, lock, cache, prot, qos, region, user = self.int_read_addr_queue.popleft()
self.bus.araddr <= addr
self.bus.arid <= arid
self.bus.arlen <= length
self.bus.arsize <= size
self.bus.arburst <= burst
if hasattr(self.bus, "arlock"):
self.bus.arlock <= lock
if hasattr(self.bus, "arcache"):
self.bus.arcache <= cache
self.bus.arprot <= prot
if hasattr(self.bus, "arqos"):
self.bus.arqos <= qos
if hasattr(self.bus, "arregion"):
self.bus.arregion <= region
if hasattr(self.bus, "aruser"):
self.bus.aruser <= user
self.bus.arvalid <= 1
else:
self.bus.arvalid <= 0
async def _process_read_resp_if(self):
while True:
await ReadOnly()
# read handshake signals
rready_sample = self.bus.rready.value
rvalid_sample = self.bus.rvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.rready <= 0
continue
if rready_sample and rvalid_sample:
rid = self.bus.rid.value.integer
rdata = self.bus.rdata.value.integer
rresp = self.bus.rresp.value.integer
rlast = self.bus.rlast.value.integer
ruser = self.bus.ruser.value.integer if hasattr(self.bus, "ruser") else None
self.int_read_resp_queue_list.setdefault(rid, deque())
self.int_read_resp_queue_list[rid].append((rid, rdata, rresp, rlast, ruser))
self.int_read_resp_sync_list.setdefault(rid, Event())
self.int_read_resp_sync_list[rid].set()
await RisingEdge(self.clock)
self.bus.rready <= 1
class AxiMaster(object):
def __init__(self, entity, name, clock, reset=None):
self.write_if = None
self.read_if = None
self.clock = clock
self.write_if = AxiMasterWrite(entity, name, clock, reset)
self.read_if = AxiMasterRead(entity, name, clock, reset)
def init_read(self, address, length, burst=AxiBurstType.INCR, size=None, lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, token=None):
if not self.read_if:
raise Exception()
self.read_if.init_read(address, length, burst, size, lock, cache, prot, qos, region, user, token)
def init_write(self, address, data, burst=AxiBurstType.INCR, size=None, lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0, token=None):
if not self.write_if:
raise Exception()
self.write_if.init_write(address, data, burst, size, lock, cache, prot, qos, region, user, token)
def idle(self):
return (not self.read_if or self.read_if.idle()) and (not self.write_if or self.write_if.idle())
async def wait(self):
while not self.idle():
await RisingEdge(self.clock)
async def wait_read(self):
if not self.read_if:
raise Exception()
await self.read_if.wait()
async def wait_write(self):
if not self.write_if:
raise Exception()
await self.write_if.wait()
def read_data_ready(self, token=None):
if not self.read_if:
raise Exception()
return self.read_if.read_data_ready(token)
def get_read_data(self, token=None):
if not self.read_if:
raise Exception()
return self.read_if.get_read_data(token)
def write_resp_ready(self, token=None):
if not self.write_if:
raise Exception()
return self.write_if.write_resp_ready(token)
def get_write_resp(self, token=None):
if not self.write_if:
raise Exception()
return self.write_if.get_write_resp(token)
async def read(self, address, length, burst=AxiBurstType.INCR, size=None, lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
if not self.read_if:
raise Exception()
return await self.read_if.read(address, length, burst, size, lock, cache, prot, qos, region, user)
async def write(self, address, data, burst=AxiBurstType.INCR, size=None, lock=AxiLockType.NORMAL, cache=0b0011, prot=AxiProt.NONSECURE, qos=0, region=0, user=0):
if not self.write_if:
raise Exception()
return await self.write_if.write(address, data, burst, size, lock, cache, prot, qos, region, user)

470
cocotbext/axi/axi_ram.py Normal file
View File

@@ -0,0 +1,470 @@
"""
Copyright (c) 2020 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 cocotb
from cocotb.triggers import RisingEdge, ReadOnly, Event
from cocotb.drivers import BusDriver
import mmap
from collections import deque
from .constants import *
class AxiRamWrite(BusDriver):
_signals = [
# Write address channel
"awid", "awaddr", "awlen", "awsize", "awburst", "awprot", "awvalid", "awready",
# Write data channel
"wdata", "wstrb", "wlast", "wvalid", "wready",
# Write response channel
"bid", "bresp", "bvalid", "bready",
]
_optional_signals = [
# Write address channel
"awlock", "awcache", "awqos", "awregion", "awuser",
# Write data channel
"wuser",
# Write response channel
"buser",
]
def __init__(self, entity, name, clock, reset=None, size=1024, mem=None):
super().__init__(entity, name, clock)
if type(mem) is mmap.mmap:
self.mem = mem
else:
self.mem = mmap.mmap(-1, size)
self.size = len(self.mem)
self.int_write_addr_queue = deque()
self.int_write_addr_sync = Event()
self.int_write_data_queue = deque()
self.int_write_data_sync = Event()
self.int_write_resp_queue = deque()
self.int_write_resp_sync = Event()
self.in_flight_operations = 0
self.width = len(self.bus.wdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.strb_mask = 2**len(self.bus.wstrb)-1
assert self.byte_width == len(self.bus.wstrb)
assert self.byte_width * self.byte_size == self.width
self.reset = reset
assert len(self.bus.awlen) == 8
assert len(self.bus.awsize) == 3
assert len(self.bus.awburst) == 2
if hasattr(self.bus, "awlock"):
assert len(self.bus.awlock) == 1
if hasattr(self.bus, "awcache"):
assert len(self.bus.awcache) == 4
assert len(self.bus.awprot) == 3
if hasattr(self.bus, "awqos"):
assert len(self.bus.awqos) == 4
if hasattr(self.bus, "awregion"):
assert len(self.bus.awregion) == 4
assert len(self.bus.awvalid) == 1
assert len(self.bus.awready) == 1
self.bus.awready.setimmediatevalue(0)
assert len(self.bus.wlast) == 1
assert len(self.bus.wvalid) == 1
assert len(self.bus.wready) == 1
self.bus.wready.setimmediatevalue(0)
assert len(self.bus.bid) == len(self.bus.awid)
self.bus.bid.setimmediatevalue(0)
assert len(self.bus.bresp) == 2
self.bus.bresp.setimmediatevalue(0)
assert len(self.bus.bvalid) == 1
if hasattr(self.bus, "buser"):
self.bus.buser.setimmediatevalue(0)
self.bus.bvalid.setimmediatevalue(0)
assert len(self.bus.bready) == 1
cocotb.fork(self._process_write())
cocotb.fork(self._process_write_addr_if())
cocotb.fork(self._process_write_data_if())
cocotb.fork(self._process_write_resp_if())
def read_mem(self, address, length):
self.mem.seek(address)
return self.mem.read(length)
def write_mem(self, address, data):
self.mem.seek(address)
self.mem.write(bytes(data))
async def _process_write(self):
while True:
if not self.int_write_addr_queue:
self.int_write_addr_sync.clear()
await self.int_write_addr_sync.wait()
awid, addr, length, size, burst, prot = self.int_write_addr_queue.popleft()
prot = AxiProt(prot)
self.log.info(f"Write burst awid: {awid:#x} awaddr: {addr:#010x} awlen: {length} awsize: {size} awprot: {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
if not self.int_write_data_queue:
self.int_write_data_sync.clear()
await self.int_write_data_sync.wait()
data, strb, last = self.int_write_data_queue.popleft()
# todo latency
self.mem.seek(cur_word_addr % self.size)
data = data.to_bytes(self.byte_width, 'little')
self.log.info(f"Write word awid: {awid:#x} addr: {cur_addr:#010x} wstrb: {strb:#04x} data: {' '.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
self.int_write_resp_queue.append((awid, AxiResp.OKAY))
self.int_write_resp_sync.set()
async def _process_write_addr_if(self):
while True:
await ReadOnly()
# read handshake signals
awready_sample = self.bus.awready.value
awvalid_sample = self.bus.awvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.awready <= 0
continue
if awready_sample and awvalid_sample:
awid = self.bus.awid.value.integer
awaddr = self.bus.awaddr.value.integer
awlen = self.bus.awlen.value.integer
awsize = self.bus.awsize.value.integer
awburst = self.bus.awburst.value.integer
awprot = self.bus.awprot.value.integer
self.int_write_addr_queue.append((awid, awaddr, awlen, awsize, awburst, awprot))
self.int_write_addr_sync.set()
await RisingEdge(self.clock)
self.bus.awready <= 1
async def _process_write_data_if(self):
while True:
await ReadOnly()
# read handshake signals
wready_sample = self.bus.wready.value
wvalid_sample = self.bus.wvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.wready <= 0
continue
if wready_sample and wvalid_sample:
wdata = self.bus.wdata.value.integer
wstrb = self.bus.wstrb.value.integer
wlast = self.bus.wlast.value.integer
self.int_write_data_queue.append((wdata, wstrb, wlast))
self.int_write_data_sync.set()
await RisingEdge(self.clock)
self.bus.wready <= 1
async def _process_write_resp_if(self):
while True:
await ReadOnly()
# read handshake signals
bready_sample = self.bus.bready.value
bvalid_sample = self.bus.bvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.bvalid <= 0
continue
await RisingEdge(self.clock)
if (bready_sample and bvalid_sample) or (not bvalid_sample):
if self.int_write_resp_queue:
bid, bresp = self.int_write_resp_queue.popleft()
self.bus.bid <= bid
self.bus.bresp <= bresp
self.bus.bvalid <= 1
else:
self.bus.bvalid <= 0
class AxiRamRead(BusDriver):
_signals = [
# Read address channel
"arid", "araddr", "arlen", "arsize", "arburst", "arprot", "arvalid", "arready",
# Read data channel
"rid", "rdata", "rresp", "rlast", "rvalid", "rready",
]
_optional_signals = [
# Read address channel
"arlock", "arcache", "arqos", "arregion", "aruser",
# Read data channel
"ruser",
]
def __init__(self, entity, name, clock, reset=None, size=1024, mem=None):
super().__init__(entity, name, clock)
if type(mem) is mmap.mmap:
self.mem = mem
else:
self.mem = mmap.mmap(-1, size)
self.size = len(self.mem)
self.int_read_addr_queue = deque()
self.int_read_addr_sync = Event()
self.int_read_resp_command_queue = deque()
self.int_read_resp_command_sync = Event()
self.int_read_resp_queue = deque()
self.int_read_resp_sync = Event()
self.in_flight_operations = 0
self.width = len(self.bus.rdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
assert self.byte_width * self.byte_size == self.width
self.reset = reset
assert len(self.bus.arlen) == 8
assert len(self.bus.arsize) == 3
assert len(self.bus.arburst) == 2
if hasattr(self.bus, "arlock"):
assert len(self.bus.arlock) == 1
if hasattr(self.bus, "arcache"):
assert len(self.bus.arcache) == 4
assert len(self.bus.arprot) == 3
if hasattr(self.bus, "arqos"):
assert len(self.bus.arqos) == 4
if hasattr(self.bus, "arregion"):
assert len(self.bus.arregion) == 4
assert len(self.bus.arvalid) == 1
assert len(self.bus.arready) == 1
self.bus.arready.setimmediatevalue(0)
assert len(self.bus.rid) == len(self.bus.arid)
self.bus.rid.setimmediatevalue(0)
self.bus.rdata.setimmediatevalue(0)
assert len(self.bus.rresp) == 2
self.bus.rresp.setimmediatevalue(0)
assert len(self.bus.rlast) == 1
self.bus.rlast.setimmediatevalue(0)
if hasattr(self.bus, "ruser"):
self.bus.ruser.setimmediatevalue(0)
assert len(self.bus.rvalid) == 1
self.bus.rvalid.setimmediatevalue(0)
assert len(self.bus.rready) == 1
cocotb.fork(self._process_read())
cocotb.fork(self._process_read_addr_if())
cocotb.fork(self._process_read_resp_if())
def read_mem(self, address, length):
self.mem.seek(address)
return self.mem.read(length)
def write_mem(self, address, data):
self.mem.seek(address)
self.mem.write(bytes(data))
async def _process_read(self):
while True:
if not self.int_read_addr_queue:
self.int_read_addr_sync.clear()
await self.int_read_addr_sync.wait()
arid, addr, length, size, burst, prot = self.int_read_addr_queue.popleft()
prot = AxiProt(prot)
self.log.info(f"Read burst arid: {arid:#x} araddr: {addr:#010x} arlen: {length} arsize: {size} arprot: {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)
self.int_read_resp_queue.append((arid, int.from_bytes(data, 'little'), AxiResp.OKAY, n == length-1))
self.int_read_resp_sync.set()
self.log.info(f"Read word arid: {arid:#x} addr: {cur_addr:#010x} data: {' '.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 _process_read_addr_if(self):
while True:
await ReadOnly()
# read handshake signals
arready_sample = self.bus.arready.value
arvalid_sample = self.bus.arvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.arready <= 0
continue
if arready_sample and arvalid_sample:
arid = self.bus.arid.value.integer
araddr = self.bus.araddr.value.integer
arlen = self.bus.arlen.value.integer
arsize = self.bus.arsize.value.integer
arburst = self.bus.arburst.value.integer
arprot = self.bus.arprot.value.integer
self.int_read_addr_queue.append((arid, araddr, arlen, arsize, arburst, arprot))
self.int_read_addr_sync.set()
await RisingEdge(self.clock)
self.bus.arready <= 1
async def _process_read_resp_if(self):
while True:
await ReadOnly()
# read handshake signals
rready_sample = self.bus.rready.value
rvalid_sample = self.bus.rvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.rvalid <= 0
continue
await RisingEdge(self.clock)
if (rready_sample and rvalid_sample) or (not rvalid_sample):
if self.int_read_resp_queue:
rid, rdata, rresp, rlast = self.int_read_resp_queue.popleft()
self.bus.rid <= rid
self.bus.rdata <= rdata
self.bus.rresp <= rresp
self.bus.rlast <= rlast
self.bus.rvalid <= 1
else:
self.bus.rvalid <= 0
class AxiRam(object):
def __init__(self, entity, name, clock, reset=None, size=1024, mem=None):
self.write_if = None
self.read_if = None
if type(mem) is mmap.mmap:
self.mem = mem
else:
self.mem = mmap.mmap(-1, size)
self.size = len(self.mem)
self.write_if = AxiRamWrite(entity, name, clock, reset, mem=self.mem)
self.read_if = AxiRamRead(entity, name, clock, reset, mem=self.mem)
def read_mem(self, address, length):
self.mem.seek(address)
return self.mem.read(length)
def write_mem(self, address, data):
self.mem.seek(address)
self.mem.write(bytes(data))

557
cocotbext/axi/axil_master.py Normal file
View File

@@ -0,0 +1,557 @@
"""
Copyright (c) 2020 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 cocotb
from cocotb.triggers import RisingEdge, ReadOnly, Event
from cocotb.drivers import BusDriver
from collections import deque
from .constants import *
class AxiLiteMasterWrite(BusDriver):
_signals = [
# Write address channel
"awaddr", "awprot", "awvalid", "awready",
# Write data channel
"wdata", "wstrb", "wvalid", "wready",
# Write response channel
"bresp", "bvalid", "bready",
]
def __init__(self, entity, name, clock, reset=None):
super().__init__(entity, name, clock)
self.active_tokens = set()
self.write_resp_queue = deque()
self.write_resp_sync = Event()
self.write_resp_set = set()
self.int_write_addr_queue = deque()
self.int_write_data_queue = deque()
self.int_write_resp_command_queue = deque()
self.int_write_resp_command_sync = Event()
self.int_write_resp_queue = deque()
self.int_write_resp_sync = Event()
self.in_flight_operations = 0
self.width = len(self.bus.wdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.strb_mask = 2**len(self.bus.wstrb)-1
assert self.byte_width == len(self.bus.wstrb)
assert self.byte_width * self.byte_size == self.width
self.reset = reset
self.bus.awaddr.setimmediatevalue(0)
assert len(self.bus.awprot) == 3
self.bus.awprot.setimmediatevalue(0)
assert len(self.bus.awvalid) == 1
self.bus.awvalid.setimmediatevalue(0)
assert len(self.bus.awready) == 1
self.bus.wdata.setimmediatevalue(0)
self.bus.wstrb.setimmediatevalue(0)
assert len(self.bus.wvalid) == 1
self.bus.wvalid.setimmediatevalue(0)
assert len(self.bus.wready) == 1
assert len(self.bus.bresp) == 2
assert len(self.bus.bvalid) == 1
assert len(self.bus.bready) == 1
self.bus.bready.setimmediatevalue(0)
cocotb.fork(self._process_write_resp())
cocotb.fork(self._process_write_addr_if())
cocotb.fork(self._process_write_data_if())
cocotb.fork(self._process_write_resp_if())
def init_write(self, address, data, prot=AxiProt.NONSECURE, token=None):
if token is not None:
if token in self.active_tokens:
raise Exception("Token is not unique")
self.active_tokens.add(token)
self.in_flight_operations += 1
word_addr = (address // self.byte_width) * self.byte_width
start_offset = address % self.byte_width
end_offset = ((address + len(data) - 1) % self.byte_width) + 1
strb_start = (self.strb_mask << start_offset) & self.strb_mask
strb_end = self.strb_mask >> (self.byte_width - end_offset)
cycles = (len(data) + (address % self.byte_width) + self.byte_width-1) // self.byte_width
self.int_write_resp_command_queue.append((address, len(data), cycles, prot, token))
self.int_write_resp_command_sync.set()
offset = 0
self.log.info(f"Write start addr: {address:#010x} prot: {prot} data: {' '.join((f'{c:02x}' for c in data))}")
for k in range(cycles):
start = 0
stop = self.byte_width
strb = self.strb_mask
if k == 0:
start = start_offset
strb &= strb_start
if k == cycles-1:
stop = end_offset
strb &= strb_end
val = 0
for j in range(start, stop):
val |= bytearray(data)[offset] << j*8
offset += 1
self.int_write_addr_queue.append((word_addr + start + k*self.byte_width, prot))
self.int_write_data_queue.append((val, strb))
def idle(self):
return not self.in_flight_operations
async def wait(self):
while not self.idle():
self.write_resp_sync.clear()
await self.write_resp_sync.wait()
async def wait_for_token(self, token):
if token not in self.active_tokens:
return
while token not in self.write_resp_set:
self.write_resp_sync.clear()
await self.write_resp_sync.wait()
def write_resp_ready(self, token=None):
if token is not None:
return token in self.write_resp_set
return bool(self.write_resp_queue)
def get_write_resp(self, token=None):
if token is not None:
if token in self.write_resp_set:
for resp in self.write_resp_queue:
if resp[-1] == token:
self.write_resp_queue.remove(resp)
self.active_tokens.remove(resp[-1])
self.write_resp_set.remove(resp[-1])
return resp
return None
if self.write_resp_queue:
resp = self.write_resp_queue.popleft()
if resp[-1] is not None:
self.active_tokens.remove(resp[-1])
self.write_resp_set.remove(resp[-1])
return resp
return None
async def write(self, address, data, prot=AxiProt.NONSECURE):
token = object()
self.init_write(address, data, prot, token)
await self.wait_for_token(token)
return self.get_write_resp(token)
async def _process_write_resp(self):
while True:
if not self.int_write_resp_command_queue:
self.int_write_resp_command_sync.clear()
await self.int_write_resp_command_sync.wait()
addr, length, cycles, prot, token = self.int_write_resp_command_queue.popleft()
resp = AxiResp.OKAY
for k in range(cycles):
if not self.int_write_resp_queue:
self.int_write_resp_sync.clear()
await self.int_write_resp_sync.wait()
cycle_resp = self.int_write_resp_queue.popleft()
cycle_resp = AxiResp(cycle_resp)
if cycle_resp != AxiResp.OKAY:
resp = cycle_resp
self.log.info(f"Write complete addr: {addr:#010x} prot: {prot} resp: {resp!s} length: {length}")
self.write_resp_queue.append((addr, length, resp, token))
self.write_resp_sync.set()
if token is not None:
self.write_resp_set.add(token)
self.in_flight_operations -= 1
async def _process_write_addr_if(self):
while True:
await ReadOnly()
# read handshake signals
awready_sample = self.bus.awready.value
awvalid_sample = self.bus.awvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.awvalid <= 0
continue
await RisingEdge(self.clock)
if (awready_sample and awvalid_sample) or (not awvalid_sample):
if self.int_write_addr_queue:
addr, prot = self.int_write_addr_queue.popleft()
self.bus.awaddr <= addr
self.bus.awprot <= prot
self.bus.awvalid <= 1
else:
self.bus.awvalid <= 0
async def _process_write_data_if(self):
while True:
await ReadOnly()
# read handshake signals
wready_sample = self.bus.wready.value
wvalid_sample = self.bus.wvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.wvalid <= 0
continue
await RisingEdge(self.clock)
if (wready_sample and wvalid_sample) or (not wvalid_sample):
if self.int_write_data_queue:
data, strb = self.int_write_data_queue.popleft()
self.bus.wdata <= data
self.bus.wstrb <= strb
self.bus.wvalid <= 1
else:
self.bus.wvalid <= 0
async def _process_write_resp_if(self):
while True:
await ReadOnly()
# read handshake signals
bready_sample = self.bus.bready.value
bvalid_sample = self.bus.bvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.bready <= 0
continue
if bready_sample and bvalid_sample:
bresp = self.bus.bresp.value.integer
self.int_write_resp_queue.append(bresp)
self.int_write_resp_sync.set()
await RisingEdge(self.clock)
self.bus.bready <= 1
class AxiLiteMasterRead(BusDriver):
_signals = [
# Read address channel
"araddr", "arprot", "arvalid", "arready",
# Read data channel
"rdata", "rresp", "rvalid", "rready",
]
def __init__(self, entity, name, clock, reset=None):
super().__init__(entity, name, clock)
self.active_tokens = set()
self.read_data_queue = deque()
self.read_data_sync = Event()
self.read_data_set = set()
self.int_read_addr_queue = deque()
self.int_read_resp_command_queue = deque()
self.int_read_resp_command_sync = Event()
self.int_read_resp_queue = deque()
self.int_read_resp_sync = Event()
self.in_flight_operations = 0
self.width = len(self.bus.rdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
assert self.byte_width * self.byte_size == self.width
self.reset = reset
self.bus.araddr.setimmediatevalue(0)
assert len(self.bus.arprot) == 3
self.bus.arprot.setimmediatevalue(0)
assert len(self.bus.arvalid) == 1
self.bus.arvalid.setimmediatevalue(0)
assert len(self.bus.arready) == 1
assert len(self.bus.rresp) == 2
assert len(self.bus.rvalid) == 1
assert len(self.bus.rready) == 1
self.bus.rready.setimmediatevalue(0)
cocotb.fork(self._process_read_resp())
cocotb.fork(self._process_read_addr_if())
cocotb.fork(self._process_read_resp_if())
def init_read(self, address, length, prot=AxiProt.NONSECURE, token=None):
if token is not None:
if token in self.active_tokens:
raise Exception("Token is not unique")
self.active_tokens.add(token)
self.in_flight_operations += 1
word_addr = (address // self.byte_width) * self.byte_width
cycles = (length + self.byte_width-1 + (address % self.byte_width)) // self.byte_width
self.int_read_resp_command_queue.append((address, length, cycles, prot, token))
self.int_read_resp_command_sync.set()
self.log.info(f"Read start addr: {address:#010x} prot: {prot} length: {length}")
for k in range(cycles):
self.int_read_addr_queue.append((word_addr + k*self.byte_width, prot))
def idle(self):
return not self.in_flight_operations
async def wait(self):
while not self.idle():
self.read_resp_sync.clear()
await self.read_resp_sync.wait()
async def wait_for_token(self, token):
if token not in self.active_tokens:
return
while token not in self.read_data_set:
self.read_data_sync.clear()
await self.read_data_sync.wait()
def read_data_ready(self, token=None):
if token is not None:
return token in self.read_data_set
return bool(self.read_data_queue)
def get_read_data(self, token=None):
if token is not None:
if token in self.read_data_set:
for resp in self.read_data_queue:
if resp[-1] == token:
self.read_data_queue.remove(resp)
self.active_tokens.remove(resp[-1])
self.read_data_set.remove(resp[-1])
return resp
return None
if self.read_data_queue:
resp = self.read_data_queue.popleft()
if resp[-1] is not None:
self.active_tokens.remove(resp[-1])
self.read_data_set.remove(resp[-1])
return resp
return None
async def read(self, address, length, prot=AxiProt.NONSECURE):
token = object()
self.init_read(address, length, prot, token)
await self.wait_for_token(token)
return self.get_read_data(token)
async def _process_read_resp(self):
while True:
if not self.int_read_resp_command_queue:
self.int_read_resp_command_sync.clear()
await self.int_read_resp_command_sync.wait()
addr, length, cycles, prot, token = self.int_read_resp_command_queue.popleft()
word_addr = (addr // self.byte_width) * self.byte_width
start_offset = addr % self.byte_width
end_offset = ((addr + length - 1) % self.byte_width) + 1
data = bytearray()
resp = AxiResp.OKAY
for k in range(cycles):
if not self.int_read_resp_queue:
self.int_read_resp_sync.clear()
await self.int_read_resp_sync.wait()
cycle_data, cycle_resp = self.int_read_resp_queue.popleft()
cycle_resp = AxiResp(cycle_resp)
if cycle_resp != AxiResp.OKAY:
resp = cycle_resp
start = 0
stop = self.byte_width
if k == 0:
start = start_offset
if k == cycles-1:
stop = end_offset
for j in range(start, stop):
data.extend(bytearray([(cycle_data >> j*8) & 0xff]))
self.log.info(f"Read complete addr: {addr:#010x} prot: {prot} resp: {resp!s} data: {' '.join((f'{c:02x}' for c in data))}")
self.read_data_queue.append((addr, data, resp, token))
self.read_data_sync.set()
if token is not None:
self.read_data_set.add(token)
self.in_flight_operations -= 1
async def _process_read_addr_if(self):
while True:
await ReadOnly()
# read handshake signals
arready_sample = self.bus.arready.value
arvalid_sample = self.bus.arvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.arvalid <= 0
continue
await RisingEdge(self.clock)
if (arready_sample and arvalid_sample) or (not arvalid_sample):
if self.int_read_addr_queue:
addr, prot = self.int_read_addr_queue.popleft()
self.bus.araddr <= addr
self.bus.arprot <= prot
self.bus.arvalid <= 1
else:
self.bus.arvalid <= 0
async def _process_read_resp_if(self):
while True:
await ReadOnly()
# read handshake signals
rready_sample = self.bus.rready.value
rvalid_sample = self.bus.rvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.rready <= 0
continue
if rready_sample and rvalid_sample:
rdata = self.bus.rdata.value.integer
rresp = self.bus.rresp.value.integer
self.int_read_resp_queue.append((rdata, rresp))
self.int_read_resp_sync.set()
await RisingEdge(self.clock)
self.bus.rready <= 1
class AxiLiteMaster(object):
def __init__(self, entity, name, clock, reset=None):
self.write_if = None
self.read_if = None
self.clock = clock
self.write_if = AxiLiteMasterWrite(entity, name, clock, reset)
self.read_if = AxiLiteMasterRead(entity, name, clock, reset)
def init_read(self, address, length, prot=AxiProt.NONSECURE, token=None):
if not self.read_if:
raise Exception()
self.read_if.init_read(address, length, prot, token)
def init_write(self, address, data, prot=AxiProt.NONSECURE, token=None):
if not self.write_if:
raise Exception()
self.write_if.init_write(address, data, prot, token)
def idle(self):
return (not self.read_if or self.read_if.idle()) and (not self.write_if or self.write_if.idle())
async def wait(self):
while not self.idle():
await RisingEdge(self.clock)
async def wait_read(self):
if not self.read_if:
raise Exception()
await self.read_if.wait()
async def wait_write(self):
if not self.write_if:
raise Exception()
await self.write_if.wait()
def read_data_ready(self, token=None):
if not self.read_if:
raise Exception()
return self.read_if.read_data_ready(token)
def get_read_data(self, token=None):
if not self.read_if:
raise Exception()
return self.read_if.get_read_data(token)
def write_resp_ready(self, token=None):
if not self.write_if:
raise Exception()
return self.write_if.write_resp_ready(token)
def get_write_resp(self, token=None):
if not self.write_if:
raise Exception()
return self.write_if.get_write_resp(token)
async def read(self, address, length, prot=AxiProt.NONSECURE):
if not self.read_if:
raise Exception()
return await self.read_if.read(address, length, prot)
async def write(self, address, data, prot=AxiProt.NONSECURE):
if not self.write_if:
raise Exception()
return await self.write_if.write(address, data, prot)

352
cocotbext/axi/axil_ram.py Normal file
View File

@@ -0,0 +1,352 @@
"""
Copyright (c) 2020 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 cocotb
from cocotb.triggers import RisingEdge, ReadOnly, Event
from cocotb.drivers import BusDriver
import mmap
import queue
from collections import deque
from .constants import *
class AxiLiteRamWrite(BusDriver):
_signals = [
# Write address channel
"awaddr", "awprot", "awvalid", "awready",
# Write data channel
"wdata", "wstrb", "wvalid", "wready",
# Write response channel
"bresp", "bvalid", "bready",
]
def __init__(self, entity, name, clock, reset=None, size=1024, mem=None):
super().__init__(entity, name, clock)
if type(mem) is mmap.mmap:
self.mem = mem
else:
self.mem = mmap.mmap(-1, size)
self.size = len(self.mem)
self.int_write_addr_queue = deque()
self.int_write_addr_sync = Event()
self.int_write_data_queue = deque()
self.int_write_data_sync = Event()
self.int_write_resp_queue = deque()
self.int_write_resp_sync = Event()
self.in_flight_operations = 0
self.width = len(self.bus.wdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
self.strb_mask = 2**len(self.bus.wstrb)-1
assert self.byte_width == len(self.bus.wstrb)
assert self.byte_width * self.byte_size == self.width
self.reset = reset
assert len(self.bus.awprot) == 3
assert len(self.bus.awvalid) == 1
assert len(self.bus.awready) == 1
self.bus.awready.setimmediatevalue(0)
assert len(self.bus.wvalid) == 1
assert len(self.bus.wready) == 1
self.bus.wready.setimmediatevalue(0)
assert len(self.bus.bresp) == 2
self.bus.bresp.setimmediatevalue(0)
assert len(self.bus.bvalid) == 1
self.bus.bvalid.setimmediatevalue(0)
assert len(self.bus.bready) == 1
cocotb.fork(self._process_write())
cocotb.fork(self._process_write_addr_if())
cocotb.fork(self._process_write_data_if())
cocotb.fork(self._process_write_resp_if())
def read_mem(self, address, length):
self.mem.seek(address)
return self.mem.read(length)
def write_mem(self, address, data):
self.mem.seek(address)
self.mem.write(bytes(data))
async def _process_write(self):
while True:
if not self.int_write_addr_queue:
self.int_write_addr_sync.clear()
await self.int_write_addr_sync.wait()
addr, prot = self.int_write_addr_queue.popleft()
addr = (addr // self.byte_width) * self.byte_width
prot = AxiProt(prot)
if not self.int_write_data_queue:
self.int_write_data_sync.clear()
await self.int_write_data_sync.wait()
data, strb = self.int_write_data_queue.popleft()
# todo latency
self.mem.seek(addr % self.size)
data = data.to_bytes(self.byte_width, 'little')
self.log.info(f"Write data addr: {addr:#010x} prot: {prot} wstrb: {strb:#04x} data: {' '.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)
self.int_write_resp_queue.append(AxiResp.OKAY)
self.int_write_resp_sync.set()
async def _process_write_addr_if(self):
while True:
await ReadOnly()
# read handshake signals
awready_sample = self.bus.awready.value
awvalid_sample = self.bus.awvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.awready <= 0
continue
if awready_sample and awvalid_sample:
awaddr = self.bus.awaddr.value.integer
awprot = self.bus.awprot.value.integer
self.int_write_addr_queue.append((awaddr, awprot))
self.int_write_addr_sync.set()
await RisingEdge(self.clock)
self.bus.awready <= 1
async def _process_write_data_if(self):
while True:
await ReadOnly()
# read handshake signals
wready_sample = self.bus.wready.value
wvalid_sample = self.bus.wvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.wready <= 0
continue
if wready_sample and wvalid_sample:
wdata = self.bus.wdata.value.integer
wstrb = self.bus.wstrb.value.integer
self.int_write_data_queue.append((wdata, wstrb))
self.int_write_data_sync.set()
await RisingEdge(self.clock)
self.bus.wready <= 1
async def _process_write_resp_if(self):
while True:
await ReadOnly()
# read handshake signals
bready_sample = self.bus.bready.value
bvalid_sample = self.bus.bvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.bvalid <= 0
continue
await RisingEdge(self.clock)
if (bready_sample and bvalid_sample) or (not bvalid_sample):
if self.int_write_resp_queue:
bresp = self.int_write_resp_queue.popleft()
self.bus.bresp <= bresp
self.bus.bvalid <= 1
else:
self.bus.bvalid <= 0
class AxiLiteRamRead(BusDriver):
_signals = [
# Read address channel
"araddr", "arprot", "arvalid", "arready",
# Read data channel
"rdata", "rresp", "rvalid", "rready",
]
def __init__(self, entity, name, clock, reset=None, size=1024, mem=None):
super().__init__(entity, name, clock)
if type(mem) is mmap.mmap:
self.mem = mem
else:
self.mem = mmap.mmap(-1, size)
self.size = len(self.mem)
self.int_read_addr_queue = deque()
self.int_read_addr_sync = Event()
self.int_read_resp_command_queue = deque()
self.int_read_resp_command_sync = Event()
self.int_read_resp_queue = deque()
self.int_read_resp_sync = Event()
self.in_flight_operations = 0
self.width = len(self.bus.rdata)
self.byte_size = 8
self.byte_width = self.width // self.byte_size
assert self.byte_width * self.byte_size == self.width
self.reset = reset
assert len(self.bus.arprot) == 3
assert len(self.bus.arvalid) == 1
assert len(self.bus.arready) == 1
self.bus.arready.setimmediatevalue(0)
self.bus.rdata.setimmediatevalue(0)
assert len(self.bus.rresp) == 2
self.bus.rresp.setimmediatevalue(0)
assert len(self.bus.rvalid) == 1
self.bus.rvalid.setimmediatevalue(0)
assert len(self.bus.rready) == 1
cocotb.fork(self._process_read())
cocotb.fork(self._process_read_addr_if())
cocotb.fork(self._process_read_resp_if())
def read_mem(self, address, length):
self.mem.seek(address)
return self.mem.read(length)
def write_mem(self, address, data):
self.mem.seek(address)
self.mem.write(bytes(data))
async def _process_read(self):
while True:
if not self.int_read_addr_queue:
self.int_read_addr_sync.clear()
await self.int_read_addr_sync.wait()
addr, prot = self.int_read_addr_queue.popleft()
addr = (addr // self.byte_width) * self.byte_width
prot = AxiProt(prot)
# todo latency
self.mem.seek(addr % self.size)
data = self.mem.read(self.byte_width)
self.int_read_resp_queue.append((int.from_bytes(data, 'little'), AxiResp.OKAY))
self.int_read_resp_sync.set()
self.log.info(f"Read data addr: {addr:#010x} prot: {prot} data: {' '.join((f'{c:02x}' for c in data))}")
async def _process_read_addr_if(self):
while True:
await ReadOnly()
# read handshake signals
arready_sample = self.bus.arready.value
arvalid_sample = self.bus.arvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.arready <= 0
continue
if arready_sample and arvalid_sample:
araddr = self.bus.araddr.value.integer
arprot = self.bus.arprot.value.integer
self.int_read_addr_queue.append((araddr, arprot))
self.int_read_addr_sync.set()
await RisingEdge(self.clock)
self.bus.arready <= 1
async def _process_read_resp_if(self):
while True:
await ReadOnly()
# read handshake signals
rready_sample = self.bus.rready.value
rvalid_sample = self.bus.rvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
self.bus.rvalid <= 0
continue
await RisingEdge(self.clock)
if (rready_sample and rvalid_sample) or (not rvalid_sample):
if self.int_read_resp_queue:
rdata, rresp = self.int_read_resp_queue.popleft()
self.bus.rdata <= rdata
self.bus.rresp <= rresp
self.bus.rvalid <= 1
else:
self.bus.rvalid <= 0
class AxiLiteRam(object):
def __init__(self, entity, name, clock, reset=None, size=1024, mem=None):
self.write_if = None
self.read_if = None
if type(mem) is mmap.mmap:
self.mem = mem
else:
self.mem = mmap.mmap(-1, size)
self.size = len(self.mem)
self.write_if = AxiLiteRamWrite(entity, name, clock, reset, mem=self.mem)
self.read_if = AxiLiteRamRead(entity, name, clock, reset, mem=self.mem)
def read_mem(self, address, length):
self.mem.seek(address)
return self.mem.read(length)
def write_mem(self, address, data):
self.mem.seek(address)
self.mem.write(bytes(data))

494
cocotbext/axi/axis.py Normal file
View File

@@ -0,0 +1,494 @@
"""
Copyright (c) 2020 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 cocotb
from cocotb.triggers import RisingEdge, ReadOnly, Timer, First, Event
from cocotb.drivers import BusDriver
from collections import deque
class AxiStreamFrame(object):
def __init__(self, tdata=b'', tkeep=None, tid=None, tdest=None, tuser=None):
self.tdata = bytearray()
self.tkeep = None
self.tid = None
self.tdest = None
self.tuser = None
if type(tdata) is AxiStreamFrame:
if type(tdata.tdata) is bytearray:
self.tdata = bytearray(tdata.tdata)
else:
self.tdata = list(tdata.tdata)
if tdata.tkeep is not None:
self.tkeep = list(tdata.tkeep)
if tdata.tid is not None:
if type(tdata.tid) in (int, bool):
self.tid = tdata.tid
else:
self.tid = list(tdata.tid)
if tdata.tdest is not None:
if type(tdata.tdest) in (int, bool):
self.tdest = tdata.tdest
else:
self.tdest = list(tdata.tdest)
if tdata.tuser is not None:
if type(tdata.tuser) in (int, bool):
self.tuser = tdata.tuser
else:
self.tuser = list(tdata.tuser)
elif type(tdata) in (bytes, bytearray):
self.tdata = bytearray(tdata)
self.tkeep = tkeep
self.tid = tid
self.tdest = tdest
self.tuser = tuser
else:
self.tdata = list(tdata)
self.tkeep = tkeep
self.tid = tid
self.tdest = tdest
self.tuser = tuser
def normalize(self):
# normalize all sideband signals to the same size as tdata
n = len(self.tdata)
if self.tkeep is not None:
self.tkeep = self.tkeep[:n] + [self.tkeep[-1]]*(n-len(self.tkeep))
else:
self.tkeep = [1]*n
if self.tid is not None:
if type(self.tid) in (int, bool):
self.tid = [self.tid]*n
else:
self.tid = self.tid[:n] + [self.tid[-1]]*(n-len(self.tid))
else:
self.tid = [0]*n
if self.tdest is not None:
if type(self.tdest) in (int, bool):
self.tdest = [self.tdest]*n
else:
self.tdest = self.tdest[:n] + [self.tdest[-1]]*(n-len(self.tdest))
else:
self.tdest = [0]*n
if self.tuser is not None:
if type(self.tuser) in (int, bool):
self.tuser = [self.tuser]*n
else:
self.tuser = self.tuser[:n] + [self.tuser[-1]]*(n-len(self.tuser))
else:
self.tuser = [0]*n
def compact(self):
if len(self.tkeep):
# remove tkeep=0 bytes
for k in range(len(self.tdata)-1, -1, -1):
if not self.tkeep[k]:
if k < len(self.tdata):
del self.tdata[k]
if k < len(self.tkeep):
del self.tkeep[k]
if k < len(self.tid):
del self.tid[k]
if k < len(self.tdest):
del self.tdest[k]
if k < len(self.tuser):
del self.tuser[k]
# remove tkeep
self.tkeep = None
# clean up other sideband signals
# either remove or consolidate if values are identical
if len(self.tid) == 0:
self.tid = None
elif all(self.tid[0] == i for i in self.tid):
self.tid = self.tid[0]
if len(self.tdest) == 0:
self.tdest = None
elif all(self.tdest[0] == i for i in self.tdest):
self.tdest = self.tdest[0]
if len(self.tuser) == 0:
self.tuser = None
elif all(self.tuser[0] == i for i in self.tuser):
self.tuser = self.tuser[0]
def __eq__(self, other):
if not isinstance(other, AxiStreamFrame):
return False
if self.tdata != other.tdata:
return False
if self.tkeep is not None and other.tkeep is not None:
if self.tkeep != other.tkeep:
return False
if self.tid is not None and other.tid is not None:
if type(self.tid) in (int, bool) and type(other.tid) is list:
for k in other.tid:
if self.tid != k:
return False
elif type(other.tid) in (int, bool) and type(self.tid) is list:
for k in self.tid:
if other.tid != k:
return False
elif self.tid != other.tid:
return False
if self.tdest is not None and other.tdest is not None:
if type(self.tdest) in (int, bool) and type(other.tdest) is list:
for k in other.tdest:
if self.tdest != k:
return False
elif type(other.tdest) in (int, bool) and type(self.tdest) is list:
for k in self.tdest:
if other.tdest != k:
return False
elif self.tdest != other.tdest:
return False
if self.tuser is not None and other.tuser is not None:
if type(self.tuser) in (int, bool) and type(other.tuser) is list:
for k in other.tuser:
if self.tuser != k:
return False
elif type(other.tuser) in (int, bool) and type(self.tuser) is list:
for k in self.tuser:
if other.tuser != k:
return False
elif self.tuser != other.tuser:
return False
return True
def __repr__(self):
return (
f"{type(self).__name__}(tdata={repr(self.tdata)}, " +
f"tkeep={repr(self.tkeep)}, " +
f"tid={repr(self.tid)}, " +
f"tdest={repr(self.tdest)}, " +
f"tuser={repr(self.tuser)})"
)
def __len__(self):
return len(self.tdata)
def __iter__(self):
return self.tdata.__iter__()
class AxiStreamSource(BusDriver):
_signals = ["tdata"]
_optional_signals = ["tvalid", "tready", "tlast", "tkeep", "tid", "tdest", "tuser"]
def __init__(self, entity, name, clock, reset=None):
super().__init__(entity, name, clock)
self.active = False
self.queue = deque()
self.queue_occupancy_bytes = 0
self.queue_occupancy_frames = 0
self.width = len(self.bus.tdata)
self.byte_width = 1
self.reset = reset
self.bus.tdata.setimmediatevalue(0)
if hasattr(self.bus, "tvalid"):
self.bus.tvalid.setimmediatevalue(0)
if hasattr(self.bus, "tlast"):
self.bus.tlast.setimmediatevalue(0)
if hasattr(self.bus, "tkeep"):
self.byte_width = len(self.bus.tkeep)
self.bus.tkeep.setimmediatevalue(0)
if hasattr(self.bus, "tid"):
self.bus.tid.setimmediatevalue(0)
if hasattr(self.bus, "tdest"):
self.bus.tdest.setimmediatevalue(0)
if hasattr(self.bus, "tuser"):
self.bus.tuser.setimmediatevalue(0)
self.byte_size = self.width // self.byte_width
self.byte_mask = 2**self.byte_size-1
cocotb.fork(self._run())
def send(self, frame):
frame = AxiStreamFrame(frame)
self.queue_occupancy_bytes += len(frame)
self.queue_occupancy_frames += 1
self.queue.append(frame)
def write(self, data):
self.send(data)
def count(self):
return len(self.queue)
def empty(self):
return not self.queue
def idle(self):
return self.empty() and not self.active
async def wait(self):
while not self.idle():
await RisingEdge(self.clock)
async def _run(self):
frame = None
self.active = False
while True:
await ReadOnly()
# read handshake signals
tready_sample = (not hasattr(self.bus, "tready")) or self.bus.tready.value
tvalid_sample = (not hasattr(self.bus, "tvalid")) or self.bus.tvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
frame = None
self.active = False
self.bus.tdata <= 0
if hasattr(self.bus, "tvalid"):
self.bus.tvalid <= 0
if hasattr(self.bus, "tlast"):
self.bus.tlast <= 0
if hasattr(self.bus, "tkeep"):
self.bus.tkeep <= 0
if hasattr(self.bus, "tid"):
self.bus.tid <= 0
if hasattr(self.bus, "tdest"):
self.bus.tdest <= 0
if hasattr(self.bus, "tuser"):
self.bus.tuser <= 0
continue
if frame is None and self.queue:
frame = self.queue.popleft()
self.queue_occupancy_bytes -= len(frame)
self.queue_occupancy_frames -= 1
self.log.info(f"TX frame: {frame}")
frame.normalize()
self.active = True
await RisingEdge(self.clock)
if (tready_sample and tvalid_sample) or not tvalid_sample:
if frame: # and not pause
tdata_val = 0
tlast_val = 0
tkeep_val = 0
tid_val = 0
tdest_val = 0
tuser_val = 0
offset = 0
for offset in range(self.byte_width):
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)
if len(frame.tdata) == 0:
tlast_val = 1
frame = None
break
self.bus.tdata <= tdata_val
if hasattr(self.bus, "tvalid"):
self.bus.tvalid <= 1
if hasattr(self.bus, "tlast"):
self.bus.tlast <= tlast_val
if hasattr(self.bus, "tkeep"):
self.bus.tkeep <= tkeep_val
if hasattr(self.bus, "tid"):
self.bus.tid <= tid_val
if hasattr(self.bus, "tdest"):
self.bus.tdest <= tdest_val
if hasattr(self.bus, "tuser"):
self.bus.tuser <= tuser_val
else:
if hasattr(self.bus, "tvalid"):
self.bus.tvalid <= 0
if hasattr(self.bus, "tlast"):
self.bus.tlast <= 0
self.active = bool(frame)
class AxiStreamSink(BusDriver):
_signals = ["tdata"]
_optional_signals = ["tvalid", "tready", "tlast", "tkeep", "tid", "tdest", "tuser"]
def __init__(self, entity, name, clock, reset=None):
super().__init__(entity, name, clock)
self.active = False
self.queue = deque()
self.sync = Event()
self.read_queue = []
self.queue_occupancy_bytes = 0
self.queue_occupancy_frames = 0
self.queue_occupancy_limit_bytes = None
self.queue_occupancy_limit_frames = None
self.width = len(self.bus.tdata)
self.byte_width = 1
self.reset = reset
if hasattr(self.bus, "tready"):
self.bus.tready.setimmediatevalue(0)
if hasattr(self.bus, "tkeep"):
self.byte_width = len(self.bus.tkeep)
self.byte_size = self.width // self.byte_width
self.byte_mask = 2**self.byte_size-1
cocotb.fork(self._run())
def recv(self, compact=True):
if self.queue:
frame = self.queue.popleft()
self.queue_occupancy_bytes -= len(frame)
self.queue_occupancy_frames -= 1
if compact:
frame.compact()
return frame
def read(self, count=-1):
while True:
frame = self.recv(compact=True)
if frame is None:
break
self.read_queue.extend(frame.tdata)
if count < 0:
count = len(self.read_queue)
data = self.read_queue[:count]
del self.read_queue[:count]
return data
def count(self):
return len(self.queue)
def empty(self):
return not self.queue
def idle(self):
return not self.active
async def wait(self, timeout=0, timeout_unit=None):
if not self.empty():
return
self.sync.clear()
if timeout:
await First(self.sync.wait(), Timer(timeout, timeout_unit))
else:
await self.sync.wait()
async def _run(self):
frame = AxiStreamFrame()
frame.tdata = []
frame.tkeep = []
frame.tid = []
frame.tdest = []
frame.tuser = []
self.active = False
while True:
await ReadOnly()
# read handshake signals
tready_sample = (not hasattr(self.bus, "tready")) or self.bus.tready.value
tvalid_sample = (not hasattr(self.bus, "tvalid")) or self.bus.tvalid.value
if self.reset is not None and self.reset.value:
await RisingEdge(self.clock)
frame = AxiStreamFrame()
frame.tdata = []
frame.tkeep = []
frame.tid = []
frame.tdest = []
frame.tuser = []
self.active = False
if hasattr(self.bus, "tready"):
self.bus.tready <= 0
continue
if tready_sample and tvalid_sample:
for offset in range(self.byte_width):
frame.tdata.append((self.bus.tdata.value >> (offset * self.byte_size)) & self.byte_mask)
if hasattr(self.bus, "tkeep"):
frame.tkeep.append((self.bus.tkeep.value >> offset) & 1)
if hasattr(self.bus, "tid"):
frame.tid.append(self.bus.tid.value)
if hasattr(self.bus, "tdest"):
frame.tdest.append(self.bus.tdest.value)
if hasattr(self.bus, "tuser"):
frame.tuser.append(self.bus.tuser.value)
if not hasattr(self.bus, "tlast") or self.bus.tlast.value:
if self.byte_size == 8:
frame.tdata = bytearray(frame.tdata)
self.log.info(f"RX frame: {frame}")
self.queue_occupancy_bytes += len(frame)
self.queue_occupancy_frames += 1
self.queue.append(frame)
self.sync.set()
frame = AxiStreamFrame()
frame.tdata = []
frame.tkeep = []
frame.tid = []
frame.tdest = []
frame.tuser = []
await RisingEdge(self.clock)
if self.queue_occupancy_limit_bytes and self.queue_occupancy_bytes > self.queue_occupancy_limit_bytes:
self.bus.tready <= 0
elif self.queue_occupancy_limit_frames and self.queue_occupancy_frames > self.queue_occupancy_limit_frames:
self.bus.tready <= 0
else:
self.bus.tready <= 1 # not pause

102
cocotbext/axi/constants.py Normal file
View File

@@ -0,0 +1,102 @@
"""
Copyright (c) 2020 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 enum
# Burst types
# AWBURST/ARBURST
class AxiBurstType(enum.IntEnum):
FIXED = 0b00
INCR = 0b01
WRAP = 0b10
# Burst sizes (per beat)
# AWSIZE/ARSIZE
class AxiBurstSize(enum.IntEnum):
SIZE_1 = 0b000
SIZE_2 = 0b001
SIZE_4 = 0b010
SIZE_8 = 0b011
SIZE_16 = 0b100
SIZE_32 = 0b101
SIZE_64 = 0b110
SIZE_128 = 0b111
# Lock types
# AWLOCK/ARLOCK
class AxiLockType(enum.IntEnum):
NORMAL = 0b0
EXCLUSIVE = 0b1
# Cache control
# AWCACHE/ARCACHE
class AxiCacheBit(enum.IntFlag):
B = 0b0001
M = 0b0010
RA = 0b0100
WA = 0b1000
# ARCACHE
ARCACHE_DEVICE_NON_BUFFERABLE = 0b0000
ARCACHE_DEVICE_BUFFERABLE = 0b0001
ARCACHE_NORMAL_NON_CACHEABLE_NON_BUFFERABLE = 0b0010
ARCACHE_NORMAL_NON_CACHEABLE_BUFFERABLE = 0b0011
ARCACHE_WRITE_THROUGH_NO_ALLOC = 0b1010
ARCACHE_WRITE_THROUGH_READ_ALLOC = 0b1110
ARCACHE_WRITE_THROUGH_WRITE_ALLOC = 0b1010
ARCACHE_WRITE_THROUGH_READ_AND_WRITE_ALLOC = 0b1110
ARCACHE_WRITE_BACK_NO_ALLOC = 0b1011
ARCACHE_WRITE_BACK_READ_ALLOC = 0b1111
ARCACHE_WRITE_BACK_WRITE_ALLOC = 0b1011
ARCACHE_WRITE_BACK_READ_AND_WRITE_ALLOC = 0b1111
# AWCACHE
AWCACHE_DEVICE_NON_BUFFERABLE = 0b0000
AWCACHE_DEVICE_BUFFERABLE = 0b0001
AWCACHE_NORMAL_NON_CACHEABLE_NON_BUFFERABLE = 0b0010
AWCACHE_NORMAL_NON_CACHEABLE_BUFFERABLE = 0b0011
AWCACHE_WRITE_THROUGH_NO_ALLOC = 0b0110
AWCACHE_WRITE_THROUGH_READ_ALLOC = 0b0110
AWCACHE_WRITE_THROUGH_WRITE_ALLOC = 0b1110
AWCACHE_WRITE_THROUGH_READ_AND_WRITE_ALLOC = 0b1110
AWCACHE_WRITE_BACK_NO_ALLOC = 0b0111
AWCACHE_WRITE_BACK_READ_ALLOC = 0b0111
AWCACHE_WRITE_BACK_WRITE_ALLOC = 0b1111
AWCACHE_WRITE_BACK_READ_AND_WRITE_ALLOC = 0b1111
# Protection bits
# AWPROT/ARPROT
class AxiProt(enum.IntFlag):
PRIVILEGED = 0b001
NONSECURE = 0b010
INSTRUCTION = 0b100
# Operation status responses
# BRESP/RRESP
class AxiResp(enum.IntEnum):
OKAY = 0b00
EXOKAY = 0b01
SLVERR = 0b10
DECERR = 0b11

1
cocotbext/axi/version.py Normal file
View File

@@ -0,0 +1 @@
__version__ = 0.1

32
setup.py Normal file
View File

@@ -0,0 +1,32 @@
from setuptools import setup, find_namespace_packages
import os.path
version_py = os.path.join(os.path.dirname(__file__), 'cocotbext', 'axi', 'version.py')
with open(version_py, 'r') as f:
d = dict()
exec(f.read(), d)
version = d['__version__']
with open("README.md", "r") as f:
long_description = f.read()
setup(
name = "cocotbext-axi",
author="Alex Forencich",
author_email="alex@alexforencich.com",
description="AXI modules for Cocotb",
long_description=long_description,
long_description_content_type='text/markdown',
url="https://github.com/alexforencich/cocotbext-axi",
download_url = 'http://github.com/alexforencich/cocotbext-axi/tarball/master',
version = version,
packages = find_namespace_packages(include=['cocotbext.*']),
install_requires = ['cocotb'],
python_requires = '>=3.6',
classifiers = [
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Topic :: Scientific/Engineering :: Electronic Design Automation (EDA)"
]
)