Convert send/recv to blocking, add nonblocking send_nowait/recv_nowait

README.md

@@ -244,9 +244,15 @@ To receive data with an `AxiStreamSink` or `AxiStreamMonitor`, call `recv()` or
 
 #### Methods
 
-* `send(frame)`: send _frame_ (source)
+
-* `write(data)`: send _data_ (alias of send) (source)
+* `send(frame)`: send _frame_ (blocking) (source)
-* `recv(compact=True)`: receive a frame as an `AxiStreamFrame`, optionally compact frame (sink/monitor)
+* `send_nowait(frame)`: send _frame_ (non-blocking) (source)
+* `write(data)`: send _data_ (alias of send) (blocking) (source)
+* `write_nowait(data)`: send _data_ (alias of send_nowait) (non-blocking) (source)
+* `recv(compact=True)`: receive a frame as a `GmiiFrame` (blocking) (sink)
+* `recv_nowait(compact=True)`: receive a frame as a `GmiiFrame` (non-blocking) (sink)
+* `read(count)`: read _count_ bytes from buffer (blocking) (sink/monitor)
+* `read_nowait(count)`: read _count_ bytes from buffer (non-blocking) (sink/monitor)
 * `read(count)`: read _count_ bytes from buffer (sink/monitor)
 * `count()`: returns the number of items in the queue (all)
 * `empty()`: returns _True_ if the queue is empty (all)

cocotbext/axi/axi_master.py

@@ -297,7 +297,7 @@ class AxiMasterWrite(object):
                     else:
                         w.wuser = 0
 
-                self.w_channel.send(w)
+                await self.w_channel.send(w)
 
                 cur_addr += num_bytes
                 cycle_offset = (cycle_offset + num_bytes) % self.byte_width
@@ -319,8 +319,7 @@ class AxiMasterWrite(object):
 
             for bid, burst_length in cmd.burst_list:
                 while not self.int_write_resp_queue_list[bid]:
-                    await self.b_channel.wait()
+                    b = await self.b_channel.recv()
-                    b = self.b_channel.recv()
 
                     i = int(b.bid)
 
@@ -598,8 +597,7 @@ class AxiMasterRead(object):
             for rid, burst_length in cmd.burst_list:
                 for k in range(burst_length):
                     while not self.int_read_resp_queue_list[rid]:
-                        await self.r_channel.wait()
+                        r = await self.r_channel.recv()
-                        r = self.r_channel.recv()
 
                         i = int(r.rid)
 

cocotbext/axi/axi_ram.py

@@ -72,8 +72,7 @@ class AxiRamWrite(Memory):
 
     async def _process_write(self):
         while True:
-            await self.aw_channel.wait()
+            aw = await self.aw_channel.recv()
-            aw = self.aw_channel.recv()
 
             awid = int(aw.awid)
             addr = int(aw.awaddr)
@@ -106,8 +105,7 @@ class AxiRamWrite(Memory):
             for n in range(length):
                 cur_word_addr = (cur_addr // self.byte_width) * self.byte_width
 
-                await self.w_channel.wait()
+                w = await self.w_channel.recv()
-                w = self.w_channel.recv()
 
                 data = int(w.wdata)
                 strb = int(w.wstrb)
@@ -141,7 +139,7 @@ class AxiRamWrite(Memory):
             b.bid = awid
             b.bresp = AxiResp.OKAY
 
-            self.b_channel.send(b)
+            await self.b_channel.send(b)
 
 
 class AxiRamRead(Memory):
@@ -181,8 +179,7 @@ class AxiRamRead(Memory):
 
     async def _process_read(self):
         while True:
-            await self.ar_channel.wait()
+            ar = await self.ar_channel.recv()
-            ar = self.ar_channel.recv()
 
             arid = int(ar.arid)
             addr = int(ar.araddr)
@@ -225,7 +222,7 @@ class AxiRamRead(Memory):
                 r.rlast = n == length-1
                 r.rresp = AxiResp.OKAY
 
-                self.r_channel.send(r)
+                await self.r_channel.send(r)
 
                 self.log.debug("Read word awid: 0x%x addr: 0x%08x data: %s",
                     arid, cur_addr, ' '.join((f'{c:02x}' for c in data)))

cocotbext/axi/axil_master.py

@@ -194,7 +194,7 @@ class AxiLiteMasterWrite(object):
                 w.wstrb = strb
 
                 await self.aw_channel.drive(aw)
-                self.w_channel.send(w)
+                await self.w_channel.send(w)
 
     async def _process_write_resp(self):
         while True:
@@ -207,8 +207,7 @@ class AxiLiteMasterWrite(object):
             resp = AxiResp.OKAY
 
             for k in range(cmd.cycles):
-                await self.b_channel.wait()
+                b = await self.b_channel.recv()
-                b = self.b_channel.recv()
 
                 cycle_resp = AxiResp(b.bresp)
 
@@ -366,8 +365,7 @@ class AxiLiteMasterRead(object):
             resp = AxiResp.OKAY
 
             for k in range(cmd.cycles):
-                await self.r_channel.wait()
+                r = await self.r_channel.recv()
-                r = self.r_channel.recv()
 
                 cycle_data = int(r.rdata)
                 cycle_resp = AxiResp(r.rresp)

cocotbext/axi/axil_ram.py

@@ -69,14 +69,12 @@ class AxiLiteRamWrite(Memory):
 
     async def _process_write(self):
         while True:
-            await self.aw_channel.wait()
+            aw = await self.aw_channel.recv()
-            aw = self.aw_channel.recv()
 
             addr = (int(aw.awaddr) // self.byte_width) * self.byte_width
             prot = AxiProt(aw.awprot)
 
-            await self.w_channel.wait()
+            w = await self.w_channel.recv()
-            w = self.w_channel.recv()
 
             data = int(w.wdata)
             strb = int(w.wstrb)
@@ -99,7 +97,7 @@ class AxiLiteRamWrite(Memory):
             b = self.b_channel._transaction_obj()
             b.bresp = AxiResp.OKAY
 
-            self.b_channel.send(b)
+            await self.b_channel.send(b)
 
 
 class AxiLiteRamRead(Memory):
@@ -136,8 +134,7 @@ class AxiLiteRamRead(Memory):
 
     async def _process_read(self):
         while True:
-            await self.ar_channel.wait()
+            ar = await self.ar_channel.recv()
-            ar = self.ar_channel.recv()
 
             addr = (int(ar.araddr) // self.byte_width) * self.byte_width
             prot = AxiProt(ar.arprot)
@@ -152,7 +149,7 @@ class AxiLiteRamRead(Memory):
             r.rdata = int.from_bytes(data, 'little')
             r.rresp = AxiResp.OKAY
 
-            self.r_channel.send(r)
+            await self.r_channel.send(r)
 
             self.log.info("Read data araddr: 0x%08x arprot: %s data: %s",
                 addr, prot, ' '.join((f'{c:02x}' for c in data)))

cocotbext/axi/axis.py

@@ -304,14 +304,20 @@ class AxiStreamSource(object):
 
         cocotb.fork(self._run())
 
-    def send(self, frame):
+    async def send(self, frame):
+        self.send_nowait(frame)
+
+    def send_nowait(self, frame):
         frame = AxiStreamFrame(frame)
         self.queue_occupancy_bytes += len(frame)
         self.queue_occupancy_frames += 1
         self.queue.append(frame)
 
-    def write(self, data):
+    async def write(self, data):
-        self.send(data)
+        await self.send(data)
+
+    def write_nowait(self, data):
+        self.send_nowait(data)
 
     def count(self):
         return len(self.queue)
@@ -511,7 +517,13 @@ class AxiStreamSink(object):
 
         cocotb.fork(self._run())
 
-    def recv(self, compact=True):
+    async def recv(self, compact=True):
+        while self.empty():
+            self.sync.clear()
+            await self.sync.wait()
+        return self.recv_nowait(compact)
+
+    def recv_nowait(self, compact=True):
         if self.queue:
             frame = self.queue.popleft()
             self.queue_occupancy_bytes -= len(frame)
@@ -521,11 +533,15 @@ class AxiStreamSink(object):
             return frame
         return None
 
-    def read(self, count=-1):
+    async def read(self, count=-1):
-        while True:
+        while not self.read_queue:
-            frame = self.recv(compact=True)
+            frame = await self.recv(compact=True)
-            if frame is None:
+            self.read_queue.extend(frame.tdata)
-                break
+        return self.read_nowait(count)
+
+    def read_nowait(self, count=-1):
+        while not self.empty():
+            frame = self.recv_nowait(compact=True)
             self.read_queue.extend(frame.tdata)
         if count < 0:
             count = len(self.read_queue)
@@ -705,7 +721,13 @@ class AxiStreamMonitor(object):
 
         cocotb.fork(self._run())
 
-    def recv(self, compact=True):
+    async def recv(self, compact=True):
+        while self.empty():
+            self.sync.clear()
+            await self.sync.wait()
+        return self.recv_nowait(compact)
+
+    def recv_nowait(self, compact=True):
         if self.queue:
             frame = self.queue.popleft()
             self.queue_occupancy_bytes -= len(frame)
@@ -715,11 +737,15 @@ class AxiStreamMonitor(object):
             return frame
         return None
 
-    def read(self, count=-1):
+    async def read(self, count=-1):
-        while True:
+        while not self.read_queue:
-            frame = self.recv(compact=True)
+            frame = await self.recv(compact=True)
-            if frame is None:
+            self.read_queue.extend(frame.tdata)
-                break
+        return self.read_nowait(count)
+
+    def read_nowait(self, count=-1):
+        while not self.empty():
+            frame = self.recv_nowait(compact=True)
             self.read_queue.extend(frame.tdata)
         if count < 0:
             count = len(self.read_queue)

cocotbext/axi/stream.py

@@ -169,7 +169,10 @@ class StreamSource(StreamBase, StreamPause):
 
         self.drive_obj = obj
 
-    def send(self, obj):
+    async def send(self, obj):
+        self.send_nowait(obj)
+
+    def send_nowait(self, obj):
         self.queue.append(obj)
         self.queue_sync.set()
 
@@ -246,7 +249,13 @@ class StreamSink(StreamBase, StreamPause):
 
         cocotb.fork(self._run_sink())
 
-    def recv(self):
+    async def recv(self):
+        while self.empty():
+            self.queue_sync.clear()
+            await self.queue_sync.wait()
+        return self.recv_nowait()
+
+    def recv_nowait(self):
         if self.queue:
             return self.queue.popleft()
         return None
@@ -310,7 +319,13 @@ class StreamMonitor(StreamBase):
 
         cocotb.fork(self._run_monitor())
 
-    def recv(self):
+    async def recv(self):
+        while self.empty():
+            self.queue_sync.clear()
+            await self.queue_sync.wait()
+        return self.recv_nowait()
+
+    def recv_nowait(self):
         if self.queue:
             return self.queue.popleft()
         return None

tests/axis/test_axis.py

@@ -90,23 +90,21 @@ async def run_test(dut, payload_lengths=None, payload_data=None, idle_inserter=N
         test_frame = AxiStreamFrame(test_data)
         test_frame.tid = cur_id
         test_frame.tdest = cur_id
-        tb.source.send(test_frame)
+        await tb.source.send(test_frame)
 
         test_frames.append(test_frame)
 
         cur_id = (cur_id + 1) % id_count
 
     for test_frame in test_frames:
-        await tb.sink.wait()
+        rx_frame = await tb.sink.recv()
-        rx_frame = tb.sink.recv()
 
         assert rx_frame.tdata == test_frame.tdata
         assert rx_frame.tid == test_frame.tid
         assert rx_frame.tdest == test_frame.tdest
         assert not rx_frame.tuser
 
-        await tb.monitor.wait()
+        rx_frame = await tb.monitor.recv()
-        rx_frame = tb.monitor.recv()
 
         assert rx_frame.tdata == test_frame.tdata
         assert rx_frame.tid == test_frame.tid