Release v0.1.10

README.md

@@ -101,6 +101,7 @@ The `GmiiPhy` class provides a model of a GMII PHY chip.  It wraps instances of
 * `recv_nowait()`: receive a frame as a `GmiiFrame` (non-blocking) (sink)
 * `count()`: returns the number of items in the queue (all)
 * `empty()`: returns _True_ if the queue is empty (all)
+* `full()`: returns _True_ if the queue occupancy limits are met (source)
 * `idle()`: returns _True_ if no transfer is in progress (all) or if the queue is not empty (source)
 * `clear()`: drop all data in queue (all)
 * `wait()`: wait for idle (source)
@@ -217,6 +218,7 @@ The `MiiPhy` class provides a model of an MII PHY chip.  It wraps instances of `
 * `recv_nowait()`: receive a frame as a `GmiiFrame` (non-blocking) (sink)
 * `count()`: returns the number of items in the queue (all)
 * `empty()`: returns _True_ if the queue is empty (all)
+* `full()`: returns _True_ if the queue occupancy limits are met (source)
 * `idle()`: returns _True_ if no transfer is in progress (all) or if the queue is not empty (source)
 * `clear()`: drop all data in queue (all)
 * `wait()`: wait for idle (source)
@@ -307,6 +309,7 @@ The `RgmiiPhy` class provides a model of an RGMII PHY chip.  It wraps instances
 * `recv_nowait()`: receive a frame as a `GmiiFrame` (non-blocking) (sink)
 * `count()`: returns the number of items in the queue (all)
 * `empty()`: returns _True_ if the queue is empty (all)
+* `full()`: returns _True_ if the queue occupancy limits are met (source)
 * `idle()`: returns _True_ if no transfer is in progress (all) or if the queue is not empty (source)
 * `clear()`: drop all data in queue (all)
 * `wait()`: wait for idle (source)
@@ -381,6 +384,7 @@ To receive data with an `XgmiiSink`, call `recv()` or `recv_nowait()`.  Optional
 * `recv_nowait()`: receive a frame as an `XgmiiFrame` (non-blocking) (sink)
 * `count()`: returns the number of items in the queue (all)
 * `empty()`: returns _True_ if the queue is empty (all)
+* `full()`: returns _True_ if the queue occupancy limits are met (source)
 * `idle()`: returns _True_ if no transfer is in progress (all) or if the queue is not empty (source)
 * `clear()`: drop all data in queue (all)
 * `wait()`: wait for idle (source)

cocotbext/eth/gmii.py

@@ -25,9 +25,9 @@ THE SOFTWARE.
 import logging
 import struct
 import zlib
-from collections import deque
 
 import cocotb
+from cocotb.queue import Queue, QueueFull
 from cocotb.triggers import RisingEdge, Timer, First, Event
 from cocotb.utils import get_sim_time, get_sim_steps
 
@@ -100,7 +100,7 @@ class GmiiFrame:
             self.error = [0]*n
 
     def compact(self):
-        if not any(self.error):
+        if self.error is not None and not any(self.error):
             self.error = None
 
     def handle_tx_complete(self):
@@ -152,7 +152,11 @@ class GmiiSource(Reset):
         super().__init__(*args, **kwargs)
 
         self.active = False
-        self.queue = deque()
+        self.queue = Queue()
+        self.dequeue_event = Event()
+        self.current_frame = None
+        self.idle_event = Event()
+        self.idle_event.set()
 
         self.ifg = 12
         self.mii_mode = False
@@ -160,6 +164,9 @@ class GmiiSource(Reset):
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
 
+        self.queue_occupancy_limit_bytes = -1
+        self.queue_occupancy_limit_frames = -1
+
         self.width = 8
         self.byte_width = 1
 
@@ -176,31 +183,53 @@ class GmiiSource(Reset):
         self._init_reset(reset, reset_active_level)
 
     async def send(self, frame):
-        self.send_nowait(frame)
+        while self.full():
-
+            self.dequeue_event.clear()
-    def send_nowait(self, frame):
+            await self.dequeue_event.wait()
         frame = GmiiFrame(frame)
+        await self.queue.put(frame)
+        self.idle_event.clear()
+        self.queue_occupancy_bytes += len(frame)
+        self.queue_occupancy_frames += 1
+
+    def send_nowait(self, frame):
+        if self.full():
+            raise QueueFull()
+        frame = GmiiFrame(frame)
+        self.queue.put_nowait(frame)
+        self.idle_event.clear()
         self.queue_occupancy_bytes += len(frame)
         self.queue_occupancy_frames += 1
-        self.queue.append(frame)
 
     def count(self):
-        return len(self.queue)
+        return self.queue.qsize()
 
     def empty(self):
-        return not self.queue
+        return self.queue.empty()
+
+    def full(self):
+        if self.queue_occupancy_limit_bytes > 0 and self.queue_occupancy_bytes > self.queue_occupancy_limit_bytes:
+            return True
+        elif self.queue_occupancy_limit_frames > 0 and self.queue_occupancy_frames > self.queue_occupancy_limit_frames:
+            return True
+        else:
+            return False
 
     def idle(self):
         return self.empty() and not self.active
 
     def clear(self):
-        self.queue.clear()
+        while not self.queue.empty():
+            frame = self.queue.get_nowait()
+            frame.sim_time_end = None
+            frame.handle_tx_complete()
+        self.dequeue_event.set()
+        self.idle_event.set()
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
 
     async def wait(self):
-        while not self.idle():
+        await self.idle_event.wait()
-            await RisingEdge(self.clock)
 
     def _handle_reset(self, state):
         if state:
@@ -208,17 +237,25 @@ class GmiiSource(Reset):
             if self._run_cr is not None:
                 self._run_cr.kill()
                 self._run_cr = None
+
+            self.active = False
+            self.data <= 0
+            if self.er is not None:
+                self.er <= 0
+            self.dv <= 0
+
+            if self.current_frame:
+                self.log.warning("Flushed transmit frame during reset: %s", self.current_frame)
+                self.current_frame.handle_tx_complete()
+                self.current_frame = None
+
+            if self.queue.empty():
+                self.idle_event.set()
         else:
             self.log.info("Reset de-asserted")
             if self._run_cr is None:
                 self._run_cr = cocotb.fork(self._run())
 
-        self.active = False
-        self.data <= 0
-        if self.er is not None:
-            self.er <= 0
-        self.dv <= 0
-
     async def _run(self):
         frame = None
         ifg_cnt = 0
@@ -232,11 +269,13 @@ class GmiiSource(Reset):
                     # in IFG
                     ifg_cnt -= 1
 
-                elif frame is None and self.queue:
+                elif frame is None and not self.queue.empty():
                     # send frame
-                    frame = self.queue.popleft()
+                    frame = self.queue.get_nowait()
+                    self.dequeue_event.set()
                     self.queue_occupancy_bytes -= len(frame)
                     self.queue_occupancy_frames -= 1
+                    self.current_frame = frame
                     frame.sim_time_start = get_sim_time()
                     frame.sim_time_sfd = None
                     frame.sim_time_end = None
@@ -273,12 +312,14 @@ class GmiiSource(Reset):
                         frame.sim_time_end = get_sim_time()
                         frame.handle_tx_complete()
                         frame = None
+                        self.current_frame = None
                 else:
                     self.data <= 0
                     if self.er is not None:
                         self.er <= 0
                     self.dv <= 0
                     self.active = False
+                    self.idle_event.set()
 
 
 class GmiiSink(Reset):
@@ -301,8 +342,8 @@ class GmiiSink(Reset):
         super().__init__(*args, **kwargs)
 
         self.active = False
-        self.queue = deque()
+        self.queue = Queue()
-        self.sync = Event()
+        self.active_event = Event()
 
         self.mii_mode = False
 
@@ -322,42 +363,46 @@ class GmiiSink(Reset):
 
         self._init_reset(reset, reset_active_level)
 
+    def _recv(self, frame, compact=True):
+        if self.queue.empty():
+            self.active_event.clear()
+        self.queue_occupancy_bytes -= len(frame)
+        self.queue_occupancy_frames -= 1
+        if compact:
+            frame.compact()
+        return frame
+
     async def recv(self, compact=True):
-        while self.empty():
+        frame = await self.queue.get()
-            self.sync.clear()
+        return self._recv(frame, compact)
-            await self.sync.wait()
-        return self.recv_nowait(compact)
 
     def recv_nowait(self, compact=True):
-        if self.queue:
+        frame = self.queue.get_nowait()
-            frame = self.queue.popleft()
+        return self._recv(frame, compact)
-            self.queue_occupancy_bytes -= len(frame)
-            self.queue_occupancy_frames -= 1
-            return frame
-        return None
 
     def count(self):
-        return len(self.queue)
+        return self.queue.qsize()
 
     def empty(self):
-        return not self.queue
+        return self.queue.empty()
 
     def idle(self):
         return not self.active
 
     def clear(self):
-        self.queue.clear()
+        while not self.queue.empty():
+            self.queue.get_nowait()
+        self.active_event.clear()
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
 
     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))
+            await First(self.active_event.wait(), Timer(timeout, timeout_unit))
         else:
-            await self.sync.wait()
+            await self.active_event.wait()
 
     def _handle_reset(self, state):
         if state:
@@ -365,13 +410,13 @@ class GmiiSink(Reset):
             if self._run_cr is not None:
                 self._run_cr.kill()
                 self._run_cr = None
+
+            self.active = False
         else:
             self.log.info("Reset de-asserted")
             if self._run_cr is None:
                 self._run_cr = cocotb.fork(self._run())
 
-        self.active = False
-
     async def _run(self):
         frame = None
         self.active = False
@@ -424,8 +469,8 @@ class GmiiSink(Reset):
                         self.queue_occupancy_bytes += len(frame)
                         self.queue_occupancy_frames += 1
 
-                        self.queue.append(frame)
+                        self.queue.put_nowait(frame)
-                        self.sync.set()
+                        self.active_event.set()
 
                         frame = None
 

cocotbext/eth/mii.py

@@ -23,9 +23,9 @@ THE SOFTWARE.
 """
 
 import logging
-from collections import deque
 
 import cocotb
+from cocotb.queue import Queue, QueueFull
 from cocotb.triggers import RisingEdge, Timer, First, Event
 from cocotb.utils import get_sim_time, get_sim_steps
 
@@ -54,13 +54,20 @@ class MiiSource(Reset):
         super().__init__(*args, **kwargs)
 
         self.active = False
-        self.queue = deque()
+        self.queue = Queue()
+        self.dequeue_event = Event()
+        self.current_frame = None
+        self.idle_event = Event()
+        self.idle_event.set()
 
         self.ifg = 12
 
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
 
+        self.queue_occupancy_limit_bytes = -1
+        self.queue_occupancy_limit_frames = -1
+
         self.width = 4
         self.byte_width = 1
 
@@ -77,31 +84,53 @@ class MiiSource(Reset):
         self._init_reset(reset, reset_active_level)
 
     async def send(self, frame):
-        self.send_nowait(frame)
+        while self.full():
-
+            self.dequeue_event.clear()
-    def send_nowait(self, frame):
+            await self.dequeue_event.wait()
         frame = GmiiFrame(frame)
+        await self.queue.put(frame)
+        self.idle_event.clear()
+        self.queue_occupancy_bytes += len(frame)
+        self.queue_occupancy_frames += 1
+
+    def send_nowait(self, frame):
+        if self.full():
+            raise QueueFull()
+        frame = GmiiFrame(frame)
+        self.queue.put_nowait(frame)
+        self.idle_event.clear()
         self.queue_occupancy_bytes += len(frame)
         self.queue_occupancy_frames += 1
-        self.queue.append(frame)
 
     def count(self):
-        return len(self.queue)
+        return self.queue.qsize()
 
     def empty(self):
-        return not self.queue
+        return self.queue.empty()
+
+    def full(self):
+        if self.queue_occupancy_limit_bytes > 0 and self.queue_occupancy_bytes > self.queue_occupancy_limit_bytes:
+            return True
+        elif self.queue_occupancy_limit_frames > 0 and self.queue_occupancy_frames > self.queue_occupancy_limit_frames:
+            return True
+        else:
+            return False
 
     def idle(self):
         return self.empty() and not self.active
 
     def clear(self):
-        self.queue.clear()
+        while not self.queue.empty():
+            frame = self.queue.get_nowait()
+            frame.sim_time_end = None
+            frame.handle_tx_complete()
+        self.dequeue_event.set()
+        self.idle_event.set()
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
 
     async def wait(self):
-        while not self.idle():
+        await self.idle_event.wait()
-            await RisingEdge(self.clock)
 
     def _handle_reset(self, state):
         if state:
@@ -109,17 +138,25 @@ class MiiSource(Reset):
             if self._run_cr is not None:
                 self._run_cr.kill()
                 self._run_cr = None
+
+            self.active = False
+            self.data <= 0
+            if self.er is not None:
+                self.er <= 0
+            self.dv <= 0
+
+            if self.current_frame:
+                self.log.warning("Flushed transmit frame during reset: %s", self.current_frame)
+                self.current_frame.handle_tx_complete()
+                self.current_frame = None
+
+            if self.queue.empty():
+                self.idle_event.set()
         else:
             self.log.info("Reset de-asserted")
             if self._run_cr is None:
                 self._run_cr = cocotb.fork(self._run())
 
-        self.active = False
-        self.data <= 0
-        if self.er is not None:
-            self.er <= 0
-        self.dv <= 0
-
     async def _run(self):
         frame = None
         ifg_cnt = 0
@@ -133,11 +170,13 @@ class MiiSource(Reset):
                     # in IFG
                     ifg_cnt -= 1
 
-                elif frame is None and self.queue:
+                elif frame is None and not self.queue.empty():
                     # send frame
-                    frame = self.queue.popleft()
+                    frame = self.queue.get_nowait()
+                    self.dequeue_event.set()
                     self.queue_occupancy_bytes -= len(frame)
                     self.queue_occupancy_frames -= 1
+                    self.current_frame = frame
                     frame.sim_time_start = get_sim_time()
                     frame.sim_time_sfd = None
                     frame.sim_time_end = None
@@ -170,12 +209,14 @@ class MiiSource(Reset):
                         frame.sim_time_end = get_sim_time()
                         frame.handle_tx_complete()
                         frame = None
+                        self.current_frame = None
                 else:
                     self.data <= 0
                     if self.er is not None:
                         self.er <= 0
                     self.dv <= 0
                     self.active = False
+                    self.idle_event.set()
 
 
 class MiiSink(Reset):
@@ -197,8 +238,8 @@ class MiiSink(Reset):
         super().__init__(*args, **kwargs)
 
         self.active = False
-        self.queue = deque()
+        self.queue = Queue()
-        self.sync = Event()
+        self.active_event = Event()
 
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
@@ -216,42 +257,46 @@ class MiiSink(Reset):
 
         self._init_reset(reset, reset_active_level)
 
+    def _recv(self, frame, compact=True):
+        if self.queue.empty():
+            self.active_event.clear()
+        self.queue_occupancy_bytes -= len(frame)
+        self.queue_occupancy_frames -= 1
+        if compact:
+            frame.compact()
+        return frame
+
     async def recv(self, compact=True):
-        while self.empty():
+        frame = await self.queue.get()
-            self.sync.clear()
+        return self._recv(frame, compact)
-            await self.sync.wait()
-        return self.recv_nowait(compact)
 
     def recv_nowait(self, compact=True):
-        if self.queue:
+        frame = self.queue.get_nowait()
-            frame = self.queue.popleft()
+        return self._recv(frame, compact)
-            self.queue_occupancy_bytes -= len(frame)
-            self.queue_occupancy_frames -= 1
-            return frame
-        return None
 
     def count(self):
-        return len(self.queue)
+        return self.queue.qsize()
 
     def empty(self):
-        return not self.queue
+        return self.queue.empty()
 
     def idle(self):
         return not self.active
 
     def clear(self):
-        self.queue.clear()
+        while not self.queue.empty():
+            self.queue.get_nowait()
+        self.active_event.clear()
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
 
     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))
+            await First(self.active_event.wait(), Timer(timeout, timeout_unit))
         else:
-            await self.sync.wait()
+            await self.active_event.wait()
 
     def _handle_reset(self, state):
         if state:
@@ -259,13 +304,13 @@ class MiiSink(Reset):
             if self._run_cr is not None:
                 self._run_cr.kill()
                 self._run_cr = None
+
+            self.active = False
         else:
             self.log.info("Reset de-asserted")
             if self._run_cr is None:
                 self._run_cr = cocotb.fork(self._run())
 
-        self.active = False
-
     async def _run(self):
         frame = None
         self.active = False
@@ -313,8 +358,8 @@ class MiiSink(Reset):
                         self.queue_occupancy_bytes += len(frame)
                         self.queue_occupancy_frames += 1
 
-                        self.queue.append(frame)
+                        self.queue.put_nowait(frame)
-                        self.sync.set()
+                        self.active_event.set()
 
                         frame = None
 

cocotbext/eth/ptp.py

@@ -186,26 +186,26 @@ class PtpClock(Reset):
             if self._run_cr is not None:
                 self._run_cr.kill()
                 self._run_cr = None
+
+            self.ts_96_s = 0
+            self.ts_96_ns = 0
+            self.ts_96_fns = 0
+            self.ts_64_ns = 0
+            self.ts_64_fns = 0
+            self.drift_cnt = 0
+            if self.ts_96 is not None:
+                self.ts_96 <= 0
+            if self.ts_64 is not None:
+                self.ts_64 <= 0
+            if self.ts_step is not None:
+                self.ts_step <= 0
+            if self.pps is not None:
+                self.pps <= 0
         else:
             self.log.info("Reset de-asserted")
             if self._run_cr is None:
                 self._run_cr = cocotb.fork(self._run())
 
-        self.ts_96_s = 0
-        self.ts_96_ns = 0
-        self.ts_96_fns = 0
-        self.ts_64_ns = 0
-        self.ts_64_fns = 0
-        self.drift_cnt = 0
-        if self.ts_96 is not None:
-            self.ts_96 <= 0
-        if self.ts_64 is not None:
-            self.ts_64 <= 0
-        if self.ts_step is not None:
-            self.ts_step <= 0
-        if self.pps is not None:
-            self.pps <= 0
-
     async def _run(self):
         while True:
             await RisingEdge(self.clock)

cocotbext/eth/rgmii.py

@@ -23,9 +23,9 @@ THE SOFTWARE.
 """
 
 import logging
-from collections import deque
 
 import cocotb
+from cocotb.queue import Queue, QueueFull
 from cocotb.triggers import RisingEdge, FallingEdge, Timer, First, Event
 from cocotb.utils import get_sim_time, get_sim_steps
 
@@ -56,7 +56,11 @@ class RgmiiSource(Reset):
         super().__init__(*args, **kwargs)
 
         self.active = False
-        self.queue = deque()
+        self.queue = Queue()
+        self.dequeue_event = Event()
+        self.current_frame = None
+        self.idle_event = Event()
+        self.idle_event.set()
 
         self.ifg = 12
         self.mii_mode = False
@@ -64,6 +68,9 @@ class RgmiiSource(Reset):
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
 
+        self.queue_occupancy_limit_bytes = -1
+        self.queue_occupancy_limit_frames = -1
+
         self.width = 8
         self.byte_width = 1
 
@@ -77,31 +84,53 @@ class RgmiiSource(Reset):
         self._init_reset(reset, reset_active_level)
 
     async def send(self, frame):
-        self.send_nowait(frame)
+        while self.full():
-
+            self.dequeue_event.clear()
-    def send_nowait(self, frame):
+            await self.dequeue_event.wait()
         frame = GmiiFrame(frame)
+        await self.queue.put(frame)
+        self.idle_event.clear()
+        self.queue_occupancy_bytes += len(frame)
+        self.queue_occupancy_frames += 1
+
+    def send_nowait(self, frame):
+        if self.full():
+            raise QueueFull()
+        frame = GmiiFrame(frame)
+        self.queue.put_nowait(frame)
+        self.idle_event.clear()
         self.queue_occupancy_bytes += len(frame)
         self.queue_occupancy_frames += 1
-        self.queue.append(frame)
 
     def count(self):
-        return len(self.queue)
+        return self.queue.qsize()
 
     def empty(self):
-        return not self.queue
+        return self.queue.empty()
+
+    def full(self):
+        if self.queue_occupancy_limit_bytes > 0 and self.queue_occupancy_bytes > self.queue_occupancy_limit_bytes:
+            return True
+        elif self.queue_occupancy_limit_frames > 0 and self.queue_occupancy_frames > self.queue_occupancy_limit_frames:
+            return True
+        else:
+            return False
 
     def idle(self):
         return self.empty() and not self.active
 
     def clear(self):
-        self.queue.clear()
+        while not self.queue.empty():
+            frame = self.queue.get_nowait()
+            frame.sim_time_end = None
+            frame.handle_tx_complete()
+        self.dequeue_event.set()
+        self.idle_event.set()
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
 
     async def wait(self):
-        while not self.idle():
+        await self.idle_event.wait()
-            await RisingEdge(self.clock)
 
     def _handle_reset(self, state):
         if state:
@@ -109,15 +138,23 @@ class RgmiiSource(Reset):
             if self._run_cr is not None:
                 self._run_cr.kill()
                 self._run_cr = None
+
+            self.active = False
+            self.data <= 0
+            self.ctrl <= 0
+
+            if self.current_frame:
+                self.log.warning("Flushed transmit frame during reset: %s", self.current_frame)
+                self.current_frame.handle_tx_complete()
+                self.current_frame = None
+
+            if self.queue.empty():
+                self.idle_event.set()
         else:
             self.log.info("Reset de-asserted")
             if self._run_cr is None:
                 self._run_cr = cocotb.fork(self._run())
 
-        self.active = False
-        self.data <= 0
-        self.ctrl <= 0
-
     async def _run(self):
         frame = None
         ifg_cnt = 0
@@ -138,11 +175,13 @@ class RgmiiSource(Reset):
                     # in IFG
                     ifg_cnt -= 1
 
-                elif frame is None and self.queue:
+                elif frame is None and not self.queue.empty():
                     # send frame
-                    frame = self.queue.popleft()
+                    frame = self.queue.get_nowait()
+                    self.dequeue_event.set()
                     self.queue_occupancy_bytes -= len(frame)
                     self.queue_occupancy_frames -= 1
+                    self.current_frame = frame
                     frame.sim_time_start = get_sim_time()
                     frame.sim_time_sfd = None
                     frame.sim_time_end = None
@@ -178,11 +217,13 @@ class RgmiiSource(Reset):
                         frame.sim_time_end = get_sim_time()
                         frame.handle_tx_complete()
                         frame = None
+                        self.current_frame = None
                 else:
                     d = 0
                     er = 0
                     en = 0
                     self.active = False
+                    self.idle_event.set()
 
             await FallingEdge(self.clock)
 
@@ -212,8 +253,8 @@ class RgmiiSink(Reset):
         super().__init__(*args, **kwargs)
 
         self.active = False
-        self.queue = deque()
+        self.queue = Queue()
-        self.sync = Event()
+        self.active_event = Event()
 
         self.mii_mode = False
 
@@ -230,42 +271,46 @@ class RgmiiSink(Reset):
 
         self._init_reset(reset, reset_active_level)
 
+    def _recv(self, frame, compact=True):
+        if self.queue.empty():
+            self.active_event.clear()
+        self.queue_occupancy_bytes -= len(frame)
+        self.queue_occupancy_frames -= 1
+        if compact:
+            frame.compact()
+        return frame
+
     async def recv(self, compact=True):
-        while self.empty():
+        frame = await self.queue.get()
-            self.sync.clear()
+        return self._recv(frame, compact)
-            await self.sync.wait()
-        return self.recv_nowait(compact)
 
     def recv_nowait(self, compact=True):
-        if self.queue:
+        frame = self.queue.get_nowait()
-            frame = self.queue.popleft()
+        return self._recv(frame, compact)
-            self.queue_occupancy_bytes -= len(frame)
-            self.queue_occupancy_frames -= 1
-            return frame
-        return None
 
     def count(self):
-        return len(self.queue)
+        return self.queue.qsize()
 
     def empty(self):
-        return not self.queue
+        return self.queue.empty()
 
     def idle(self):
         return not self.active
 
     def clear(self):
-        self.queue.clear()
+        while not self.queue.empty():
+            self.queue.get_nowait()
+        self.active_event.clear()
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
 
     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))
+            await First(self.active_event.wait(), Timer(timeout, timeout_unit))
         else:
-            await self.sync.wait()
+            await self.active_event.wait()
 
     def _handle_reset(self, state):
         if state:
@@ -273,13 +318,13 @@ class RgmiiSink(Reset):
             if self._run_cr is not None:
                 self._run_cr.kill()
                 self._run_cr = None
+
+            self.active = False
         else:
             self.log.info("Reset de-asserted")
             if self._run_cr is None:
                 self._run_cr = cocotb.fork(self._run())
 
-        self.active = False
-
     async def _run(self):
         frame = None
         self.active = False
@@ -342,8 +387,8 @@ class RgmiiSink(Reset):
                         self.queue_occupancy_bytes += len(frame)
                         self.queue_occupancy_frames += 1
 
-                        self.queue.append(frame)
+                        self.queue.put_nowait(frame)
-                        self.sync.set()
+                        self.active_event.set()
 
                         frame = None
 

cocotbext/eth/version.py

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

cocotbext/eth/xgmii.py

@@ -25,9 +25,9 @@ THE SOFTWARE.
 import logging
 import struct
 import zlib
-from collections import deque
 
 import cocotb
+from cocotb.queue import Queue, QueueFull
 from cocotb.triggers import RisingEdge, Timer, First, Event
 from cocotb.utils import get_sim_time
 
@@ -102,7 +102,7 @@ class XgmiiFrame:
             self.ctrl = [0]*n
 
     def compact(self):
-        if not any(self.ctrl):
+        if self.ctrl is not None and not any(self.ctrl):
             self.ctrl = None
 
     def handle_tx_complete(self):
@@ -153,7 +153,11 @@ class XgmiiSource(Reset):
         super().__init__(*args, **kwargs)
 
         self.active = False
-        self.queue = deque()
+        self.queue = Queue()
+        self.dequeue_event = Event()
+        self.current_frame = None
+        self.idle_event = Event()
+        self.idle_event.set()
 
         self.enable_dic = True
         self.ifg = 12
@@ -162,6 +166,9 @@ class XgmiiSource(Reset):
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
 
+        self.queue_occupancy_limit_bytes = -1
+        self.queue_occupancy_limit_frames = -1
+
         self.width = len(self.data)
         self.byte_width = len(self.ctrl)
 
@@ -182,31 +189,53 @@ class XgmiiSource(Reset):
         self._init_reset(reset, reset_active_level)
 
     async def send(self, frame):
-        self.send_nowait(frame)
+        while self.full():
-
+            self.dequeue_event.clear()
-    def send_nowait(self, frame):
+            await self.dequeue_event.wait()
         frame = XgmiiFrame(frame)
+        await self.queue.put(frame)
+        self.idle_event.clear()
+        self.queue_occupancy_bytes += len(frame)
+        self.queue_occupancy_frames += 1
+
+    def send_nowait(self, frame):
+        if self.full():
+            raise QueueFull()
+        frame = XgmiiFrame(frame)
+        self.queue.put_nowait(frame)
+        self.idle_event.clear()
         self.queue_occupancy_bytes += len(frame)
         self.queue_occupancy_frames += 1
-        self.queue.append(frame)
 
     def count(self):
-        return len(self.queue)
+        return self.queue.qsize()
 
     def empty(self):
-        return not self.queue
+        return self.queue.empty()
+
+    def full(self):
+        if self.queue_occupancy_limit_bytes > 0 and self.queue_occupancy_bytes > self.queue_occupancy_limit_bytes:
+            return True
+        elif self.queue_occupancy_limit_frames > 0 and self.queue_occupancy_frames > self.queue_occupancy_limit_frames:
+            return True
+        else:
+            return False
 
     def idle(self):
         return self.empty() and not self.active
 
     def clear(self):
-        self.queue.clear()
+        while not self.queue.empty():
+            frame = self.queue.get_nowait()
+            frame.sim_time_end = None
+            frame.handle_tx_complete()
+        self.dequeue_event.set()
+        self.idle_event.set()
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
 
     async def wait(self):
-        while not self.idle():
+        await self.idle_event.wait()
-            await RisingEdge(self.clock)
 
     def _handle_reset(self, state):
         if state:
@@ -214,15 +243,23 @@ class XgmiiSource(Reset):
             if self._run_cr is not None:
                 self._run_cr.kill()
                 self._run_cr = None
+
+            self.active = False
+            self.data <= 0
+            self.ctrl <= 0
+
+            if self.current_frame:
+                self.log.warning("Flushed transmit frame during reset: %s", self.current_frame)
+                self.current_frame.handle_tx_complete()
+                self.current_frame = None
+
+            if self.queue.empty():
+                self.idle_event.set()
         else:
             self.log.info("Reset de-asserted")
             if self._run_cr is None:
                 self._run_cr = cocotb.fork(self._run())
 
-        self.active = False
-        self.data <= 0
-        self.ctrl <= 0
-
     async def _run(self):
         frame = None
         ifg_cnt = 0
@@ -243,11 +280,13 @@ class XgmiiSource(Reset):
 
                 elif frame is None:
                     # idle
-                    if self.queue:
+                    if not self.queue.empty():
                         # send frame
-                        frame = self.queue.popleft()
+                        frame = self.queue.get_nowait()
+                        self.dequeue_event.set()
                         self.queue_occupancy_bytes -= len(frame)
                         self.queue_occupancy_frames -= 1
+                        self.current_frame = frame
                         frame.sim_time_start = get_sim_time()
                         frame.sim_time_sfd = None
                         frame.sim_time_end = None
@@ -299,6 +338,7 @@ class XgmiiSource(Reset):
                                 frame.sim_time_end = get_sim_time()
                                 frame.handle_tx_complete()
                                 frame = None
+                                self.current_frame = None
                         else:
                             d_val |= XgmiiCtrl.IDLE << k*8
                             c_val |= 1 << k
@@ -309,6 +349,7 @@ class XgmiiSource(Reset):
                     self.data <= self.idle_d
                     self.ctrl <= self.idle_c
                     self.active = False
+                    self.idle_event.set()
 
 
 class XgmiiSink(Reset):
@@ -329,8 +370,8 @@ class XgmiiSink(Reset):
         super().__init__(*args, **kwargs)
 
         self.active = False
-        self.queue = deque()
+        self.queue = Queue()
-        self.sync = Event()
+        self.active_event = Event()
 
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
@@ -344,42 +385,46 @@ class XgmiiSink(Reset):
 
         self._init_reset(reset, reset_active_level)
 
+    def _recv(self, frame, compact=True):
+        if self.queue.empty():
+            self.active_event.clear()
+        self.queue_occupancy_bytes -= len(frame)
+        self.queue_occupancy_frames -= 1
+        if compact:
+            frame.compact()
+        return frame
+
     async def recv(self, compact=True):
-        while self.empty():
+        frame = await self.queue.get()
-            self.sync.clear()
+        return self._recv(frame, compact)
-            await self.sync.wait()
-        return self.recv_nowait(compact)
 
     def recv_nowait(self, compact=True):
-        if self.queue:
+        frame = self.queue.get_nowait()
-            frame = self.queue.popleft()
+        return self._recv(frame, compact)
-            self.queue_occupancy_bytes -= len(frame)
-            self.queue_occupancy_frames -= 1
-            return frame
-        return None
 
     def count(self):
-        return len(self.queue)
+        return self.queue.qsize()
 
     def empty(self):
-        return not self.queue
+        return self.queue.empty()
 
     def idle(self):
         return not self.active
 
     def clear(self):
-        self.queue.clear()
+        while not self.queue.empty():
+            self.queue.get_nowait()
+        self.active_event.clear()
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
 
     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))
+            await First(self.active_event.wait(), Timer(timeout, timeout_unit))
         else:
-            await self.sync.wait()
+            await self.active_event.wait()
 
     def _handle_reset(self, state):
         if state:
@@ -387,13 +432,13 @@ class XgmiiSink(Reset):
             if self._run_cr is not None:
                 self._run_cr.kill()
                 self._run_cr = None
+
+            self.active = False
         else:
             self.log.info("Reset de-asserted")
             if self._run_cr is None:
                 self._run_cr = cocotb.fork(self._run())
 
-        self.active = False
-
     async def _run(self):
         frame = None
         self.active = False
@@ -427,8 +472,8 @@ class XgmiiSink(Reset):
                             self.queue_occupancy_bytes += len(frame)
                             self.queue_occupancy_frames += 1
 
-                            self.queue.append(frame)
+                            self.queue.put_nowait(frame)
-                            self.sync.set()
+                            self.active_event.set()
 
                             frame = None
                         else:

setup.cfg

@@ -65,7 +65,6 @@ deps =
     cocotb-test
     coverage
     pytest-cov
-    git+https://github.com/cocotb/cocotb.git@e892a3ea48#egg=cocotb
 
 commands =
     pytest --cov=cocotbext --cov=tests --cov-branch -n auto