third-party/cocotbext-eth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve transfer tracking

Browse Source
This commit is contained in:
Alex Forencich
2021-01-03 22:55:09 -08:00
parent 1d5688778a
commit cfbc80c0cb
6 changed files with 67 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
README.md
View File

@@ -119,7 +119,10 @@ Attributes:
* `data`: bytearray
* `error`: error field, optional; list, each entry qualifies the corresponding entry in `data`.
* `rx_sim_time`: simulation time when packet was received by sink.
* `sim_time_start`: simulation time of first transfer cycle of frame.
* `sim_time_end`: simulation time of last transfer cycle of frame.
* `start_lane`: byte lane in which the start control character was transferred.
* `tx_complete`: event or callable triggered when frame is transmitted.
Methods:
@@ -383,8 +386,10 @@ Attributes:
* `data`: bytearray
* `ctrl`: control field, optional; list, each entry qualifies the corresponding entry in `data` as an XGMII control character.
* `rx_sim_time`: simulation time when packet was received by sink.
* `rx_start_lane`: byte lane that the frame start control character was received in.
* `sim_time_start`: simulation time of first transfer cycle of frame.
* `sim_time_end`: simulation time of last transfer cycle of frame.
* `start_lane`: byte lane in which the start control character was transferred.
* `tx_complete`: event or callable triggered when frame is transmitted.
Methods:

26
cocotbext/eth/gmii.py
View File

@@ -37,18 +37,23 @@ from .reset import Reset
class GmiiFrame:
def __init__(self, data=None, error=None):
def __init__(self, data=None, error=None, tx_complete=None):
self.data = bytearray()
self.error = None
self.rx_sim_time = None
self.sim_time_start = None
self.sim_time_end = None
self.tx_complete = None
if type(data) is GmiiFrame:
self.data = bytearray(data.data)
self.error = data.error
self.rx_sim_time = data.rx_sim_time
self.sim_time_start = data.sim_time_start
self.sim_time_end = data.sim_time_end
self.tx_complete = data.tx_complete
else:
self.data = bytearray(data)
self.error = error
self.tx_complete = tx_complete
@classmethod
def from_payload(cls, payload, min_len=60):
@@ -94,6 +99,12 @@ class GmiiFrame:
if not any(self.error):
self.error = None
def handle_tx_complete(self):
if isinstance(self.tx_complete, Event):
self.tx_complete.set(self)
elif callable(self.tx_complete):
self.tx_complete(self)
def __eq__(self, other):
if type(other) is GmiiFrame:
return self.data == other.data
@@ -102,7 +113,8 @@ class GmiiFrame:
return (
f"{type(self).__name__}(data={self.data!r}, "
f"error={self.error!r}, "
f"rx_sim_time={self.rx_sim_time!r})"
f"sim_time_start={self.sim_time_start!r}, "
f"sim_time_end={self.sim_time_end!r})"
)
def __len__(self):
@@ -220,6 +232,7 @@ class GmiiSource(Reset):
frame = self.queue.popleft()
self.queue_occupancy_bytes -= len(frame)
self.queue_occupancy_frames -= 1
frame.sim_time_start = get_sim_time()
self.log.info("TX frame: %s", frame)
frame.normalize()
@@ -247,6 +260,8 @@ class GmiiSource(Reset):
if not frame.data:
ifg_cnt = max(self.ifg, 1)
frame.sim_time_end = get_sim_time()
frame.handle_tx_complete()
frame = None
else:
self.data <= 0
@@ -363,7 +378,7 @@ class GmiiSink(Reset):
if dv_val:
# start of frame
frame = GmiiFrame(bytearray(), [])
frame.rx_sim_time = get_sim_time()
frame.sim_time_start = get_sim_time()
else:
if not dv_val:
# end of frame
@@ -393,6 +408,7 @@ class GmiiSink(Reset):
frame.error = error
frame.compact()
frame.sim_time_end = get_sim_time()
self.log.info("RX frame: %s", frame)
self.queue_occupancy_bytes += len(frame)

6
cocotbext/eth/mii.py
View File

@@ -138,6 +138,7 @@ class MiiSource(Reset):
frame = self.queue.popleft()
self.queue_occupancy_bytes -= len(frame)
self.queue_occupancy_frames -= 1
frame.sim_time_start = get_sim_time()
self.log.info("TX frame: %s", frame)
frame.normalize()
@@ -161,6 +162,8 @@ class MiiSource(Reset):
if not frame.data:
ifg_cnt = max(self.ifg, 1)
frame.sim_time_end = get_sim_time()
frame.handle_tx_complete()
frame = None
else:
self.data <= 0
@@ -274,7 +277,7 @@ class MiiSink(Reset):
if dv_val:
# start of frame
frame = GmiiFrame(bytearray(), [])
frame.rx_sim_time = get_sim_time()
frame.sim_time_start = get_sim_time()
else:
if not dv_val:
# end of frame
@@ -299,6 +302,7 @@ class MiiSink(Reset):
frame.error = error
frame.compact()
frame.sim_time_end = get_sim_time()
self.log.info("RX frame: %s", frame)
self.queue_occupancy_bytes += len(frame)

6
cocotbext/eth/rgmii.py
View File

@@ -142,6 +142,7 @@ class RgmiiSource(Reset):
frame = self.queue.popleft()
self.queue_occupancy_bytes -= len(frame)
self.queue_occupancy_frames -= 1
frame.sim_time_start = get_sim_time()
self.log.info("TX frame: %s", frame)
frame.normalize()
@@ -168,6 +169,8 @@ class RgmiiSource(Reset):
if not frame.data:
ifg_cnt = max(self.ifg, 1)
frame.sim_time_end = get_sim_time()
frame.handle_tx_complete()
frame = None
else:
d = 0
@@ -295,7 +298,7 @@ class RgmiiSink(Reset):
if dv_val:
# start of frame
frame = GmiiFrame(bytearray(), [])
frame.rx_sim_time = get_sim_time()
frame.sim_time_start = get_sim_time()
else:
if not dv_val:
# end of frame
@@ -325,6 +328,7 @@ class RgmiiSink(Reset):
frame.error = error
frame.compact()
frame.sim_time_end = get_sim_time()
self.log.info("RX frame: %s", frame)
self.queue_occupancy_bytes += len(frame)

36
cocotbext/eth/xgmii.py
View File

@@ -37,20 +37,25 @@ from .reset import Reset
class XgmiiFrame:
def __init__(self, data=None, ctrl=None):
def __init__(self, data=None, ctrl=None, tx_complete=None):
self.data = bytearray()
self.ctrl = None
self.rx_sim_time = None
self.rx_start_lane = None
self.sim_time_start = None
self.sim_time_end = None
self.start_lane = None
self.tx_complete = None
if type(data) is XgmiiFrame:
self.data = bytearray(data.data)
self.ctrl = data.ctrl
self.rx_sim_time = data.rx_sim_time
self.rx_start_lane = data.rx_start_lane
self.sim_time_start = data.sim_time_start
self.sim_time_end = data.sim_time_end
self.start_lane = data.start_lane
self.tx_complete = data.tx_complete
else:
self.data = bytearray(data)
self.ctrl = ctrl
self.tx_complete = tx_complete
@classmethod
def from_payload(cls, payload, min_len=60):
@@ -96,6 +101,12 @@ class XgmiiFrame:
if not any(self.ctrl):
self.ctrl = None
def handle_tx_complete(self):
if isinstance(self.tx_complete, Event):
self.tx_complete.set(self)
elif callable(self.tx_complete):
self.tx_complete(self)
def __eq__(self, other):
if type(other) is XgmiiFrame:
return self.data == other.data
@@ -104,8 +115,9 @@ class XgmiiFrame:
return (
f"{type(self).__name__}(data={self.data!r}, "
f"ctrl={self.ctrl!r}, "
f"rx_sim_time={self.rx_sim_time!r}, "
f"rx_start_lane={self.rx_start_lane!r})"
f"sim_time_start={self.sim_time_start!r}, "
f"sim_time_end={self.sim_time_end!r}, "
f"start_lane={self.start_lane!r})"
)
def __len__(self):
@@ -231,8 +243,10 @@ class XgmiiSource(Reset):
frame = self.queue.popleft()
self.queue_occupancy_bytes -= len(frame)
self.queue_occupancy_frames -= 1
frame.sim_time_start = get_sim_time()
self.log.info("TX frame: %s", frame)
frame.normalize()
frame.start_lane = 0
assert frame.data[0] == EthPre.PRE
assert frame.ctrl[0] == 0
frame.data[0] = XgmiiCtrl.START
@@ -248,6 +262,7 @@ class XgmiiSource(Reset):
if self.byte_width > 4 and (ifg_cnt > min_ifg or self.force_offset_start):
ifg_cnt = ifg_cnt-4
frame.start_lane = 4
frame.data = bytearray([XgmiiCtrl.IDLE]*4)+frame.data
frame.ctrl = [1]*4+frame.ctrl
@@ -271,6 +286,8 @@ class XgmiiSource(Reset):
if not frame.data:
ifg_cnt = max(self.ifg - (self.byte_width-k), 0)
frame.sim_time_end = get_sim_time()
frame.handle_tx_complete()
frame = None
else:
d_val |= XgmiiCtrl.IDLE << k*8
@@ -383,8 +400,8 @@ class XgmiiSink(Reset):
if c_val and d_val == XgmiiCtrl.START:
# start
frame = XgmiiFrame(bytearray([EthPre.PRE]), [0])
frame.rx_sim_time = get_sim_time()
frame.rx_start_lane = offset
frame.sim_time_start = get_sim_time()
frame.start_lane = offset
else:
if c_val:
# got a control character; terminate frame reception
@@ -394,6 +411,7 @@ class XgmiiSink(Reset):
frame.ctrl.append(c_val)
frame.compact()
frame.sim_time_end = get_sim_time()
self.log.info("RX frame: %s", frame)
self.queue_occupancy_bytes += len(frame)

2
tests/xgmii/test_xgmii.py
View File

@@ -150,7 +150,7 @@ async def run_test_alignment(dut, payload_data=None, ifg=12, enable_dic=True,
assert rx_frame.check_fcs()
assert rx_frame.ctrl is None
start_lane.append(rx_frame.rx_start_lane)
start_lane.append(rx_frame.start_lane)
tb.log.info("length: %d", length)
tb.log.info("start_lane: %s", start_lane)