Enforce max queue depth on streaming sources

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

@@ -27,7 +27,7 @@ import struct
 import zlib
 
 import cocotb
-from cocotb.queue import Queue
+from cocotb.queue import Queue, QueueFull
 from cocotb.triggers import RisingEdge, Timer, First, Event
 from cocotb.utils import get_sim_time, get_sim_steps
 
@@ -153,6 +153,7 @@ class GmiiSource(Reset):
 
         self.active = False
         self.queue = Queue()
+        self.dequeue_event = Event()
         self.current_frame = None
         self.idle_event = Event()
         self.idle_event.set()
@@ -163,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
 
@@ -179,6 +183,9 @@ class GmiiSource(Reset):
         self._init_reset(reset, reset_active_level)
 
     async def send(self, frame):
+        while self.full():
+            self.dequeue_event.clear()
+            await self.dequeue_event.wait()
         frame = GmiiFrame(frame)
         await self.queue.put(frame)
         self.idle_event.clear()
@@ -186,6 +193,8 @@ class GmiiSource(Reset):
         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()
@@ -198,6 +207,14 @@ class GmiiSource(Reset):
     def empty(self):
         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
 
@@ -206,6 +223,7 @@ class GmiiSource(Reset):
             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
@@ -254,6 +272,7 @@ class GmiiSource(Reset):
                 elif frame is None and not self.queue.empty():
                     # send frame
                     frame = self.queue.get_nowait()
+                    self.dequeue_event.set()
                     self.queue_occupancy_bytes -= len(frame)
                     self.queue_occupancy_frames -= 1
                     self.current_frame = frame

cocotbext/eth/mii.py

@@ -25,7 +25,7 @@ THE SOFTWARE.
 import logging
 
 import cocotb
-from cocotb.queue import Queue
+from cocotb.queue import Queue, QueueFull
 from cocotb.triggers import RisingEdge, Timer, First, Event
 from cocotb.utils import get_sim_time, get_sim_steps
 
@@ -55,6 +55,7 @@ class MiiSource(Reset):
 
         self.active = False
         self.queue = Queue()
+        self.dequeue_event = Event()
         self.current_frame = None
         self.idle_event = Event()
         self.idle_event.set()
@@ -64,6 +65,9 @@ class MiiSource(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 = 4
         self.byte_width = 1
 
@@ -80,6 +84,9 @@ class MiiSource(Reset):
         self._init_reset(reset, reset_active_level)
 
     async def send(self, frame):
+        while self.full():
+            self.dequeue_event.clear()
+            await self.dequeue_event.wait()
         frame = GmiiFrame(frame)
         await self.queue.put(frame)
         self.idle_event.clear()
@@ -87,6 +94,8 @@ class MiiSource(Reset):
         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()
@@ -99,6 +108,14 @@ class MiiSource(Reset):
     def empty(self):
         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
 
@@ -107,6 +124,7 @@ class MiiSource(Reset):
             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
@@ -155,6 +173,7 @@ class MiiSource(Reset):
                 elif frame is None and not self.queue.empty():
                     # send frame
                     frame = self.queue.get_nowait()
+                    self.dequeue_event.set()
                     self.queue_occupancy_bytes -= len(frame)
                     self.queue_occupancy_frames -= 1
                     self.current_frame = frame

cocotbext/eth/rgmii.py

@@ -25,7 +25,7 @@ THE SOFTWARE.
 import logging
 
 import cocotb
-from cocotb.queue import Queue
+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
 
@@ -57,6 +57,7 @@ class RgmiiSource(Reset):
 
         self.active = False
         self.queue = Queue()
+        self.dequeue_event = Event()
         self.current_frame = None
         self.idle_event = Event()
         self.idle_event.set()
@@ -67,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
 
@@ -80,6 +84,9 @@ class RgmiiSource(Reset):
         self._init_reset(reset, reset_active_level)
 
     async def send(self, frame):
+        while self.full():
+            self.dequeue_event.clear()
+            await self.dequeue_event.wait()
         frame = GmiiFrame(frame)
         await self.queue.put(frame)
         self.idle_event.clear()
@@ -87,6 +94,8 @@ class RgmiiSource(Reset):
         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()
@@ -99,6 +108,14 @@ class RgmiiSource(Reset):
     def empty(self):
         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
 
@@ -107,6 +124,7 @@ class RgmiiSource(Reset):
             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
@@ -160,6 +178,7 @@ class RgmiiSource(Reset):
                 elif frame is None and not self.queue.empty():
                     # send frame
                     frame = self.queue.get_nowait()
+                    self.dequeue_event.set()
                     self.queue_occupancy_bytes -= len(frame)
                     self.queue_occupancy_frames -= 1
                     self.current_frame = frame

cocotbext/eth/xgmii.py

@@ -27,7 +27,7 @@ import struct
 import zlib
 
 import cocotb
-from cocotb.queue import Queue
+from cocotb.queue import Queue, QueueFull
 from cocotb.triggers import RisingEdge, Timer, First, Event
 from cocotb.utils import get_sim_time
 
@@ -154,6 +154,7 @@ class XgmiiSource(Reset):
 
         self.active = False
         self.queue = Queue()
+        self.dequeue_event = Event()
         self.current_frame = None
         self.idle_event = Event()
         self.idle_event.set()
@@ -165,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)
 
@@ -185,6 +189,9 @@ class XgmiiSource(Reset):
         self._init_reset(reset, reset_active_level)
 
     async def send(self, frame):
+        while self.full():
+            self.dequeue_event.clear()
+            await self.dequeue_event.wait()
         frame = XgmiiFrame(frame)
         await self.queue.put(frame)
         self.idle_event.clear()
@@ -192,6 +199,8 @@ class XgmiiSource(Reset):
         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()
@@ -204,6 +213,14 @@ class XgmiiSource(Reset):
     def empty(self):
         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
 
@@ -212,6 +229,7 @@ class XgmiiSource(Reset):
             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
@@ -265,6 +283,7 @@ class XgmiiSource(Reset):
                     if not self.queue.empty():
                         # send frame
                         frame = self.queue.get_nowait()
+                        self.dequeue_event.set()
                         self.queue_occupancy_bytes -= len(frame)
                         self.queue_occupancy_frames -= 1
                         self.current_frame = frame