Put sinks to sleep when idle

Signed-off-by: Alex Forencich <alex@alexforencich.com>

cocotbext/axi/axis.py

@@ -282,6 +282,7 @@ class AxiStreamBase(Reset):
         self.idle_event = Event()
         self.idle_event.set()
         self.active_event = Event()
+        self.wake_event = Event()
 
         self.queue_occupancy_bytes = 0
         self.queue_occupancy_frames = 0
@@ -376,10 +377,23 @@ class AxiStreamPause:
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-        self.pause = False
+        self._pause = False
         self._pause_generator = None
         self._pause_cr = None
 
+    def _pause_update(self, val):
+        pass
+
+    @property
+    def pause(self):
+        return self._pause
+
+    @pause.setter
+    def pause(self, val):
+        if self._pause != val:
+            self._pause_update(val)
+        self._pause = val
+
     def set_pause_generator(self, generator=None):
         if self._pause_cr is not None:
             self._pause_cr.kill()
@@ -584,11 +598,20 @@ class AxiStreamMonitor(AxiStreamBase):
 
         self.read_queue = []
 
+        if hasattr(self.bus, "tvalid"):
+            cocotb.start_soon(self._run_tvalid_monitor())
+        if hasattr(self.bus, "tready"):
+            cocotb.start_soon(self._run_tready_monitor())
+
+    def _dequeue(self, frame):
+        pass
+
     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
+        self._dequeue(frame)
         if compact:
             frame.compact()
         return frame
@@ -628,6 +651,20 @@ class AxiStreamMonitor(AxiStreamBase):
         else:
             await self.active_event.wait()
 
+    async def _run_tvalid_monitor(self):
+        event = RisingEdge(self.bus.tvalid)
+
+        while True:
+            await event
+            self.wake_event.set()
+
+    async def _run_tready_monitor(self):
+        event = RisingEdge(self.bus.tready)
+
+        while True:
+            await event
+            self.wake_event.set()
+
     async def _run(self):
         frame = None
         self.active = False
@@ -642,6 +679,8 @@ class AxiStreamMonitor(AxiStreamBase):
 
         clock_edge_event = RisingEdge(self.clock)
 
+        wake_event = self.wake_event.wait()
+
         while True:
             await clock_edge_event
 
@@ -683,6 +722,9 @@ class AxiStreamMonitor(AxiStreamBase):
             else:
                 self.active = bool(frame)
 
+                self.wake_event.clear()
+                await wake_event
+
 
 class AxiStreamSink(AxiStreamMonitor, AxiStreamPause):
 
@@ -716,6 +758,12 @@ class AxiStreamSink(AxiStreamMonitor, AxiStreamPause):
             if hasattr(self.bus, "tready"):
                 self.bus.tready.value = 0
 
+    def _pause_update(self, val):
+        self.wake_event.set()
+
+    def _dequeue(self, frame):
+        self.wake_event.set()
+
     async def _run(self):
         frame = None
         self.active = False
@@ -730,7 +778,11 @@ class AxiStreamSink(AxiStreamMonitor, AxiStreamPause):
 
         clock_edge_event = RisingEdge(self.clock)
 
+        wake_event = self.wake_event.wait()
+
         while True:
+            pause_sample = self.pause
+
             await clock_edge_event
 
             # read handshake signals
@@ -772,4 +824,8 @@ class AxiStreamSink(AxiStreamMonitor, AxiStreamPause):
                 self.active = bool(frame)
 
             if has_tready:
-                self.bus.tready.value = (not self.full() and not self.pause)
+                self.bus.tready.value = (not self.full() and not pause_sample)
+
+            if not tvalid_sample or (self.pause and pause_sample) or self.full():
+                self.wake_event.clear()
+                await wake_event

cocotbext/axi/stream.py

@@ -99,6 +99,7 @@ class StreamBase(Reset):
         self.idle_event = Event()
         self.idle_event.set()
         self.active_event = Event()
+        self.wake_event = Event()
 
         self.ready = None
         self.valid = None
@@ -163,10 +164,23 @@ class StreamPause:
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-        self.pause = False
+        self._pause = False
         self._pause_generator = None
         self._pause_cr = None
 
+    def _pause_update(self, val):
+        pass
+
+    @property
+    def pause(self):
+        return self._pause
+
+    @pause.setter
+    def pause(self, val):
+        if self._pause != val:
+            self._pause_update(val)
+        self._pause = val
+
     def set_pause_generator(self, generator=None):
         if self._pause_cr is not None:
             self._pause_cr.kill()
@@ -269,9 +283,21 @@ class StreamMonitor(StreamBase):
     _valid_init = None
     _ready_init = None
 
+    def __init__(self, bus, clock, reset=None, reset_active_level=True, *args, **kwargs):
+        super().__init__(bus, clock, reset, reset_active_level, *args, **kwargs)
+
+        if self.valid is not None:
+            cocotb.start_soon(self._run_valid_monitor())
+        if self.ready is not None:
+            cocotb.start_soon(self._run_ready_monitor())
+
+    def _dequeue(self, item):
+        pass
+
     def _recv(self, item):
         if self.queue.empty():
             self.active_event.clear()
+        self._dequeue(item)
         return item
 
     async def recv(self):
@@ -290,9 +316,25 @@ class StreamMonitor(StreamBase):
         else:
             await self.active_event.wait()
 
+    async def _run_valid_monitor(self):
+        event = RisingEdge(self.valid)
+
+        while True:
+            await event
+            self.wake_event.set()
+
+    async def _run_ready_monitor(self):
+        event = RisingEdge(self.ready)
+
+        while True:
+            await event
+            self.wake_event.set()
+
     async def _run(self):
         clock_edge_event = RisingEdge(self.clock)
 
+        wake_event = self.wake_event.wait()
+
         while True:
             await clock_edge_event
 
@@ -305,6 +347,9 @@ class StreamMonitor(StreamBase):
                 self.bus.sample(obj)
                 self.queue.put_nowait(obj)
                 self.active_event.set()
+            else:
+                self.wake_event.clear()
+                await wake_event
 
 
 class StreamSink(StreamMonitor, StreamPause):
@@ -332,10 +377,20 @@ class StreamSink(StreamMonitor, StreamPause):
             if self.ready is not None:
                 self.ready.value = 0
 
+    def _pause_update(self, val):
+        self.wake_event.set()
+
+    def _dequeue(self, item):
+        self.wake_event.set()
+
     async def _run(self):
         clock_edge_event = RisingEdge(self.clock)
 
+        wake_event = self.wake_event.wait()
+
         while True:
+            pause_sample = self.pause
+
             await clock_edge_event
 
             # read handshake signals
@@ -349,7 +404,11 @@ class StreamSink(StreamMonitor, StreamPause):
                 self.active_event.set()
 
             if self.ready is not None:
-                self.ready.value = (not self.full() and not self.pause)
+                self.ready.value = (not self.full() and not pause_sample)
+
+            if not valid_sample or (self.pause and pause_sample) or self.full():
+                self.wake_event.clear()
+                await wake_event
 
 
 def define_stream(name, signals, optional_signals=None, valid_signal=None, ready_signal=None, signal_widths=None):