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

@@ -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
 
@@ -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):
@@ -153,6 +153,8 @@ 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()
 
@@ -162,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
 
@@ -178,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()
@@ -185,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()
@@ -197,12 +207,23 @@ 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
 
     def clear(self):
         while not self.queue.empty():
-            self.queue.get_nowait()
+            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
@@ -216,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
@@ -243,8 +272,10 @@ 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
                     frame.sim_time_start = get_sim_time()
                     frame.sim_time_sfd = None
                     frame.sim_time_end = None
@@ -281,6 +312,7 @@ 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:
@@ -331,23 +363,22 @@ class GmiiSink(Reset):
 
         self._init_reset(reset, reset_active_level)
 
-    async def recv(self, compact=True):
-        frame = await self.queue.get()
+    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):
+        frame = await self.queue.get()
+        return self._recv(frame, compact)
+
     def recv_nowait(self, compact=True):
-        if not self.queue.empty():
-            frame = self.queue.get_nowait()
-            if self.queue.empty():
-                self.active_event.clear()
-            self.queue_occupancy_bytes -= len(frame)
-            self.queue_occupancy_frames -= 1
-            return frame
-        return None
+        frame = self.queue.get_nowait()
+        return self._recv(frame, compact)
 
     def count(self):
         return self.queue.qsize()
@@ -379,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

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,8 @@ 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()
 
@@ -63,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
 
@@ -79,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()
@@ -86,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()
@@ -98,12 +108,23 @@ 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
 
     def clear(self):
         while not self.queue.empty():
-            self.queue.get_nowait()
+            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
@@ -117,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
@@ -144,8 +173,10 @@ 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
                     frame.sim_time_start = get_sim_time()
                     frame.sim_time_sfd = None
                     frame.sim_time_end = None
@@ -178,6 +209,7 @@ 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:
@@ -225,23 +257,22 @@ class MiiSink(Reset):
 
         self._init_reset(reset, reset_active_level)
 
-    async def recv(self, compact=True):
-        frame = await self.queue.get()
+    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):
+        frame = await self.queue.get()
+        return self._recv(frame, compact)
+
     def recv_nowait(self, compact=True):
-        if not self.queue.empty():
-            frame = self.queue.get_nowait()
-            if self.queue.empty():
-                self.active_event.clear()
-            self.queue_occupancy_bytes -= len(frame)
-            self.queue_occupancy_frames -= 1
-            return frame
-        return None
+        frame = self.queue.get_nowait()
+        return self._recv(frame, compact)
 
     def count(self):
         return self.queue.qsize()
@@ -273,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

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

@@ -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,8 @@ 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()
 
@@ -66,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
 
@@ -79,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()
@@ -86,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()
@@ -98,12 +108,23 @@ 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
 
     def clear(self):
         while not self.queue.empty():
-            self.queue.get_nowait()
+            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
@@ -117,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
@@ -149,8 +178,10 @@ 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
                     frame.sim_time_start = get_sim_time()
                     frame.sim_time_sfd = None
                     frame.sim_time_end = None
@@ -186,6 +217,7 @@ 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
@@ -239,23 +271,22 @@ class RgmiiSink(Reset):
 
         self._init_reset(reset, reset_active_level)
 
-    async def recv(self, compact=True):
-        frame = await self.queue.get()
+    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):
+        frame = await self.queue.get()
+        return self._recv(frame, compact)
+
     def recv_nowait(self, compact=True):
-        if not self.queue.empty():
-            frame = self.queue.get_nowait()
-            if self.queue.empty():
-                self.active_event.clear()
-            self.queue_occupancy_bytes -= len(frame)
-            self.queue_occupancy_frames -= 1
-            return frame
-        return None
+        frame = self.queue.get_nowait()
+        return self._recv(frame, compact)
 
     def count(self):
         return self.queue.qsize()
@@ -287,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

cocotbext/eth/version.py

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

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
 
@@ -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):
@@ -154,6 +154,8 @@ 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()
 
@@ -164,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)
 
@@ -184,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()
@@ -191,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()
@@ -203,12 +213,23 @@ 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
 
     def clear(self):
         while not self.queue.empty():
-            self.queue.get_nowait()
+            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
@@ -222,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
@@ -254,8 +283,10 @@ 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
                         frame.sim_time_start = get_sim_time()
                         frame.sim_time_sfd = None
                         frame.sim_time_end = None
@@ -307,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
@@ -353,23 +385,22 @@ class XgmiiSink(Reset):
 
         self._init_reset(reset, reset_active_level)
 
-    async def recv(self, compact=True):
-        frame = await self.queue.get()
+    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):
+        frame = await self.queue.get()
+        return self._recv(frame, compact)
+
     def recv_nowait(self, compact=True):
-        if not self.queue.empty():
-            frame = self.queue.get_nowait()
-            if self.queue.empty():
-                self.active_event.clear()
-            self.queue_occupancy_bytes -= len(frame)
-            self.queue_occupancy_frames -= 1
-            return frame
-        return None
+        frame = self.queue.get_nowait()
+        return self._recv(frame, compact)
 
     def count(self):
         return self.queue.qsize()
@@ -401,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