Make wstrb optional

Author: alexforencich

cocotbext/axi/axi_channels.py

@@ -34,8 +34,8 @@
 
 # Write data channel
 AxiWBus, AxiWTransaction, AxiWSource, AxiWSink, AxiWMonitor = define_stream("AxiW",
-    signals=["wdata", "wstrb", "wlast", "wvalid", "wready"],
-    optional_signals=["wuser"],
+    signals=["wdata", "wlast", "wvalid", "wready"],
+    optional_signals=["wstrb", "wuser"],
     signal_widths={"wlast": 1}
 )
 

cocotbext/axi/axi_master.py

@@ -241,6 +241,7 @@ def __init__(self, bus, clock, reset=None, reset_active_level=True, max_burst_le
         self.awqos_present = hasattr(self.bus.aw, "awqos")
         self.awregion_present = hasattr(self.bus.aw, "awregion")
         self.awuser_present = hasattr(self.bus.aw, "awuser")
+        self.wstrb_present = hasattr(self.bus.w, "wstrb")
         self.wuser_present = hasattr(self.bus.w, "wuser")
         self.buser_present = hasattr(self.bus.b, "buser")
 
@@ -263,7 +264,8 @@ def __init__(self, bus, clock, reset=None, reset_active_level=True, max_burst_le
                 else:
                     self.log.info("  %s: not present", sig)
 
-        assert self.byte_lanes == len(self.w_channel.bus.wstrb)
+        if self.wstrb_present:
+            assert self.byte_lanes == len(self.w_channel.bus.wstrb)
         assert self.byte_lanes * self.byte_size == self.width
 
         assert len(self.b_channel.bus.bid) == len(self.aw_channel.bus.awid)
@@ -480,6 +482,9 @@ async def _process_write(self):
 
                 n += 1
 
+                if not self.wstrb_present and strb != self.strb_mask:
+                    self.log.warning("Partial operation requested with wstrb not connected, write will be zero-padded (0x%x != 0x%x)", strb, self.strb_mask)
+
                 w = self.w_channel._transaction_obj()
                 w.wdata = val
                 w.wstrb = strb

cocotbext/axi/axi_slave.py

@@ -63,6 +63,8 @@ def __init__(self, bus, clock, reset=None, target=None, reset_active_level=True,
 
         self.max_burst_size = (self.byte_lanes-1).bit_length()
 
+        self.wstrb_present = hasattr(self.bus.w, "wstrb")
+
         self.log.info("AXI slave model configuration:")
         self.log.info("  Address width: %d bits", self.address_width)
         self.log.info("  ID width: %d bits", self.id_width)
@@ -77,7 +79,8 @@ def __init__(self, bus, clock, reset=None, target=None, reset_active_level=True,
                 else:
                     self.log.info("  %s: not present", sig)
 
-        assert self.byte_lanes == len(self.w_channel.bus.wstrb)
+        if self.wstrb_present:
+            assert self.byte_lanes == len(self.w_channel.bus.wstrb)
         assert self.byte_lanes * self.byte_size == self.width
 
         assert len(self.b_channel.bus.bid) == len(self.aw_channel.bus.awid)
@@ -146,7 +149,10 @@ async def _process_write(self):
                 w = await self.w_channel.recv()
 
                 data = int(w.wdata)
-                strb = int(getattr(w, 'wstrb', self.strb_mask))
+                if self.wstrb_present:
+                    strb = int(getattr(w, 'wstrb', self.strb_mask))
+                else:
+                    strb = self.strb_mask
                 last = int(w.wlast)
 
                 # generate operation list

cocotbext/axi/axil_channels.py

@@ -33,7 +33,8 @@
 
 # Write data channel
 AxiLiteWBus, AxiLiteWTransaction, AxiLiteWSource, AxiLiteWSink, AxiLiteWMonitor = define_stream("AxiLiteW",
-    signals=["wdata", "wstrb", "wvalid", "wready"]
+    signals=["wdata", "wvalid", "wready"],
+    optional_signals=["wstrb"]
 )
 
 # Write response channel

cocotbext/axi/axil_master.py

@@ -119,6 +119,7 @@ def __init__(self, bus, clock, reset=None, reset_active_level=True, **kwargs):
         self.strb_mask = 2**self.byte_lanes-1
 
         self.awprot_present = hasattr(self.bus.aw, "awprot")
+        self.wstrb_present = hasattr(self.bus.w, "wstrb")
 
         super().__init__(2**self.address_width, **kwargs)
 
@@ -135,7 +136,8 @@ def __init__(self, bus, clock, reset=None, reset_active_level=True, **kwargs):
                 else:
                     self.log.info("  %s: not present", sig)
 
-        assert self.byte_lanes == len(self.w_channel.bus.wstrb)
+        if self.wstrb_present:
+            assert self.byte_lanes == len(self.w_channel.bus.wstrb)
         assert self.byte_lanes * self.byte_size == self.width
 
         self._process_write_cr = None
@@ -269,6 +271,9 @@ async def _process_write(self):
                 aw.awaddr = word_addr + k*self.byte_lanes
                 aw.awprot = cmd.prot
 
+                if not self.wstrb_present and strb != self.strb_mask:
+                    self.log.warning("Partial operation requested with wstrb not connected, write will be zero-padded (0x%x != 0x%x)", strb, self.strb_mask)
+
                 w = self.w_channel._transaction_obj()
                 w.wdata = val
                 w.wstrb = strb

cocotbext/axi/axil_slave.py

@@ -60,6 +60,8 @@ def __init__(self, bus, clock, reset=None, target=None, reset_active_level=True,
         self.byte_lanes = self.width // self.byte_size
         self.strb_mask = 2**self.byte_lanes-1
 
+        self.wstrb_present = hasattr(self.bus.w, "wstrb")
+
         self.log.info("AXI lite slave model configuration:")
         self.log.info("  Memory size: %d bytes", len(self.mem))
         self.log.info("  Address width: %d bits", self.address_width)
@@ -74,7 +76,8 @@ def __init__(self, bus, clock, reset=None, target=None, reset_active_level=True,
                 else:
                     self.log.info("  %s: not present", sig)
 
-        assert self.byte_lanes == len(self.w_channel.bus.wstrb)
+        if self.wstrb_present:
+            assert self.byte_lanes == len(self.w_channel.bus.wstrb)
         assert self.byte_lanes * self.byte_size == self.width
 
         self._process_write_cr = None
@@ -109,7 +112,10 @@ async def _process_write(self):
             w = await self.w_channel.recv()
 
             data = int(w.wdata)
-            strb = int(getattr(w, 'wstrb', self.strb_mask))
+            if self.wstrb_present:
+                strb = int(getattr(w, 'wstrb', self.strb_mask))
+            else:
+                strb = self.strb_mask
 
             # generate operation list
             offset = 0