feat(bus): Add SECDED integrity for TileLink-UL
This commit implements SECDED (Single Error Correction, Double Error
Detection) for the TileLink-UL bus to ensure data integrity. The
implementation is compatible with OpenTitan's `prim_secded_inv` logic.
Key changes include:
- A parameterized `SecdedEncoder` Chisel module that supports 32-bit,
57-bit, and 128-bit data widths. The 128-bit implementation uses a
folded ECC scheme.
- `RequestIntegrityGen/Check` and `ResponseIntegrityGen/Check` modules
to generate and verify integrity codes for the TileLink A and D
channels.
- A Python-based golden model (`secded_golden.py`) for the SECDED
logic to ensure correctness.
- A new `TileLinkULInterface` cocotb utility for simplified,
transaction-based testing of the TileLink bus.
- Comprehensive cocotb tests that verify the `SecdedEncoder` against
the golden model and test the full `TlulIntegrity` functionality,
including fault injection.
Change-Id: I20a059b78a47699f145ae397b0e037d8c56dab69
diff --git a/hdl/chisel/src/bus/BUILD b/hdl/chisel/src/bus/BUILD
index 0a257c2..48737ef 100644
--- a/hdl/chisel/src/bus/BUILD
+++ b/hdl/chisel/src/bus/BUILD
@@ -22,6 +22,11 @@
package(default_visibility = ["//visibility:public"])
+load(
+ "//tests/cocotb:build_defs.bzl",
+ "VERILATOR_BUILD_ARGS",
+)
+
chisel_library(
name = "bus",
srcs = [
@@ -30,7 +35,10 @@
"Axi2TLUL.scala",
"KelvinMemIO.scala",
"KelvinToTlul.scala",
+ "SecdedEncoderTestbench.scala",
"TileLinkUL.scala",
+ "TlulIntegrity.scala",
+ "TlulIntegrityTestbench.scala",
],
deps = [
"//hdl/chisel/src/kelvin:kelvin_params",
@@ -121,4 +129,64 @@
},
vcs_verilog_sources = ["//hdl/chisel/src/bus:axi2tlul_cc_library_verilog"],
verilator_model = ":axi2tlul_model",
+)
+
+chisel_cc_library(
+ name = "tlul_integrity_testbench_cc_library",
+ chisel_lib = ":bus",
+ emit_class = "bus.EmitTlulIntegrityTestbench",
+ module_name = "TlulIntegrityTestbench",
+)
+
+verilator_cocotb_model(
+ name = "tlul_integrity_testbench_model",
+ cflags = VERILATOR_BUILD_ARGS,
+ hdl_toplevel = "TlulIntegrityTestbench",
+ trace = True,
+ verilog_source = "//hdl/chisel/src/bus:TlulIntegrityTestbench.sv",
+)
+
+chisel_cc_library(
+ name = "secded_encoder_testbench_cc_library",
+ chisel_lib = ":bus",
+ emit_class = "bus.EmitSecdedEncoderTestbench",
+ module_name = "SecdedEncoderTestbench128",
+)
+
+verilator_cocotb_model(
+ name = "secded_encoder_testbench_model",
+ cflags = VERILATOR_BUILD_ARGS,
+ hdl_toplevel = "SecdedEncoderTestbench128",
+ trace = True,
+ verilog_source = "//hdl/chisel/src/bus:SecdedEncoderTestbench128.sv",
+)
+
+chisel_cc_library(
+ name = "secded_encoder_testbench_32_cc_library",
+ chisel_lib = ":bus",
+ emit_class = "bus.EmitSecdedEncoderTestbench32",
+ module_name = "SecdedEncoderTestbench32",
+)
+
+verilator_cocotb_model(
+ name = "secded_encoder_testbench_32_model",
+ cflags = VERILATOR_BUILD_ARGS,
+ hdl_toplevel = "SecdedEncoderTestbench32",
+ trace = True,
+ verilog_source = "//hdl/chisel/src/bus:SecdedEncoderTestbench32.sv",
+)
+
+chisel_cc_library(
+ name = "secded_encoder_testbench_57_cc_library",
+ chisel_lib = ":bus",
+ emit_class = "bus.EmitSecdedEncoderTestbench57",
+ module_name = "SecdedEncoderTestbench57",
+)
+
+verilator_cocotb_model(
+ name = "secded_encoder_testbench_57_model",
+ cflags = VERILATOR_BUILD_ARGS,
+ hdl_toplevel = "SecdedEncoderTestbench57",
+ trace = True,
+ verilog_source = "//hdl/chisel/src/bus:SecdedEncoderTestbench57.sv",
)
\ No newline at end of file
diff --git a/hdl/chisel/src/bus/SecdedEncoderTestbench.scala b/hdl/chisel/src/bus/SecdedEncoderTestbench.scala
new file mode 100644
index 0000000..f35df89
--- /dev/null
+++ b/hdl/chisel/src/bus/SecdedEncoderTestbench.scala
@@ -0,0 +1,65 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package bus
+
+import chisel3._
+import kelvin.Parameters
+
+/**
+ * A testbench DUT that instantiates a single SecdedEncoder so it can be
+ * tested with cocotb.
+ */
+class SecdedEncoderTestbench(val w: Int, val moduleName: String) extends Module {
+ override def desiredName = moduleName
+
+ val io = IO(new Bundle {
+ val data_i = Input(UInt(w.W))
+ val ecc_o = Output(UInt(7.W))
+ })
+
+ val encoder = Module(new SecdedEncoder(w))
+ encoder.io.data_i := io.data_i
+ io.ecc_o := encoder.io.ecc_o
+}
+
+import _root_.circt.stage.{ChiselStage,FirtoolOption}
+import chisel3.stage.ChiselGeneratorAnnotation
+import scala.annotation.nowarn
+
+@nowarn
+object EmitSecdedEncoderTestbench extends App {
+ val p = new Parameters
+ p.lsuDataBits = 128
+ (new ChiselStage).execute(
+ Array("--target", "systemverilog") ++ args,
+ Seq(ChiselGeneratorAnnotation(() => new SecdedEncoderTestbench(p.lsuDataBits, "SecdedEncoderTestbench128"))) ++ Seq(FirtoolOption("-enable-layers=Verification"))
+ )
+}
+
+@nowarn
+object EmitSecdedEncoderTestbench32 extends App {
+ (new ChiselStage).execute(
+ Array("--target", "systemverilog") ++ args,
+ Seq(ChiselGeneratorAnnotation(() => new SecdedEncoderTestbench(32, "SecdedEncoderTestbench32"))) ++ Seq(FirtoolOption("-enable-layers=Verification"))
+ )
+}
+
+@nowarn
+object EmitSecdedEncoderTestbench57 extends App {
+ (new ChiselStage).execute(
+ Array("--target", "systemverilog") ++ args,
+ Seq(ChiselGeneratorAnnotation(() => new SecdedEncoderTestbench(57, "SecdedEncoderTestbench57"))) ++ Seq(FirtoolOption("-enable-layers=Verification"))
+ )
+}
\ No newline at end of file
diff --git a/hdl/chisel/src/bus/TlulIntegrity.scala b/hdl/chisel/src/bus/TlulIntegrity.scala
new file mode 100644
index 0000000..fbde102
--- /dev/null
+++ b/hdl/chisel/src/bus/TlulIntegrity.scala
@@ -0,0 +1,253 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package bus
+
+import chisel3._
+import chisel3.util._
+
+
+/**
+ * Contains pure combinational functions for calculating SECDED ECC codes.
+ * These are based on the `prim_secded_inv` functions from OpenTitan to ensure
+ * compatibility.
+ */
+object Secded {
+ /**
+ * Calculates a 39-bit word (32-bit data, 7-bit ECC) using the same
+ * logic as OpenTitan's `prim_secded_inv_39_32_enc`.
+ */
+ def ecc39_32(data: UInt): UInt = {
+ val checksum = Wire(Vec(7, Bool()))
+
+ // ECC bit calculation based on Verilog implementation.
+ checksum(0) := (data & "h002606BD25".U).xorR
+ checksum(1) := (data & "h00DEBA8050".U).xorR
+ checksum(2) := (data & "h00413D89AA".U).xorR
+ checksum(3) := (data & "h0031234ED1".U).xorR
+ checksum(4) := (data & "h00C2C1323B".U).xorR
+ checksum(5) := (data & "h002DCC624C".U).xorR
+ checksum(6) := (data & "h0098505586".U).xorR
+
+ // Final inversion for `secded_inv` compatibility.
+ val data_o = Cat(checksum.asUInt, data)
+ data_o.asUInt ^ "h2A00000000".U
+ }
+
+ /**
+ * Calculates a 64-bit word (57-bit data, 7-bit ECC) using the same
+ * logic as OpenTitan's `prim_secded_inv_64_57_enc`.
+ */
+ def ecc64_57(data: UInt): UInt = {
+ val checksum = Wire(Vec(7, Bool()))
+
+ // ECC bit calculation based on Verilog implementation.
+ checksum(0) := (data & "h0103FFF800007FFF".U).xorR
+ checksum(1) := (data & "h017C1FF801FF801F".U).xorR
+ checksum(2) := (data & "h01BDE1F87E0781E1".U).xorR
+ checksum(3) := (data & "h01DEEE3B8E388E22".U).xorR
+ checksum(4) := (data & "h01EF76CDB2C93244".U).xorR
+ checksum(5) := (data & "h01F7BB56D5525488".U).xorR
+ checksum(6) := (data & "h01FBDDA769A46910".U).xorR
+
+ // Final inversion for `secded_inv` compatibility.
+ val data_o = Cat(checksum.asUInt, data)
+ data_o.asUInt ^ "h5400000000000000".U
+ }
+}
+
+/**
+ * A parameterized SECDED encoder.
+ *
+ * @param DATA_W The width of the data input. Supported values are 32, 57, 128.
+ */
+class SecdedEncoder(val DATA_W: Int) extends Module {
+ override val desiredName = s"SecdedEncoder_${DATA_W}"
+ val IO_W = DATA_W match {
+ case 32 => 39
+ case 57 => 64
+ case 128 => 128 + 7 // 128-bit data uses a 7-bit folded ECC.
+ }
+ val ECC_W = IO_W - DATA_W
+
+ val io = IO(new Bundle {
+ val data_i = Input(UInt(DATA_W.W))
+ val data_o = Output(UInt(IO_W.W))
+ val ecc_o = Output(UInt(ECC_W.W))
+ })
+
+ if (DATA_W == 32) {
+ io.data_o := Secded.ecc39_32(io.data_i)
+ } else if (DATA_W == 57) {
+ io.data_o := Secded.ecc64_57(io.data_i)
+ } else if (DATA_W == 128) {
+ // For 128-bit data, we use the "folding" scheme: the data is split into
+ // four 32-bit chunks, and their 7-bit ECC codes are XORed together.
+ val ecc0 = Secded.ecc39_32(io.data_i(31, 0))(38, 32)
+ val ecc1 = Secded.ecc39_32(io.data_i(63, 32))(38, 32)
+ val ecc2 = Secded.ecc39_32(io.data_i(95, 64))(38, 32)
+ val ecc3 = Secded.ecc39_32(io.data_i(127, 96))(38, 32)
+ io.data_o := Cat(ecc0 ^ ecc1 ^ ecc2 ^ ecc3, io.data_i)
+ } else {
+ // Ensure we don't try to synthesize for an unsupported width.
+ assert(false, "Unsupported DATA_W for SecdedEncoder")
+ io.data_o := 0.U // Default assignment to avoid compilation errors
+ }
+
+ // Convenient output for just the ECC bits.
+ io.ecc_o := io.data_o(IO_W - 1, DATA_W)
+}
+
+/**
+ * Generates TileLink integrity fields for the A-channel (Request).
+ */
+class RequestIntegrityGen(p: TLULParameters) extends Module {
+ override val desiredName = s"RequestIntegrityGen_${p.w}"
+ val io = IO(new Bundle {
+ val a_i = Input(new OpenTitanTileLink.A_Channel(p))
+ val a_o = Output(new OpenTitanTileLink.A_Channel(p))
+ })
+ // Ensure that we don't optimize out any parts of the bundle, at least
+ // via the Chisel toolchain.
+ dontTouch(io.a_i)
+ dontTouch(io.a_o)
+
+ // Passthrough for most fields.
+ io.a_o := io.a_i
+
+ // Recreate the tl_h2d_cmd_intg_t struct for command integrity.
+ val cmd_w = 57
+ val cmd_data = Wire(UInt(cmd_w.W))
+ cmd_data := Cat(
+ io.a_i.user.instr_type,
+ io.a_i.address,
+ io.a_i.opcode,
+ io.a_i.mask
+ )
+
+ val cmd_encoder = Module(new SecdedEncoder(cmd_w))
+ cmd_encoder.io.data_i := cmd_data
+ io.a_o.user.cmd_intg := cmd_encoder.io.ecc_o
+
+ // Data integrity calculation.
+ val data_encoder = Module(new SecdedEncoder(p.w * 8))
+ data_encoder.io.data_i := io.a_i.data
+ io.a_o.user.data_intg := data_encoder.io.ecc_o
+}
+
+/**
+ * Checks TileLink integrity fields for the A-channel (Request).
+ */
+class RequestIntegrityCheck(p: TLULParameters) extends Module {
+ override val desiredName = s"RequestIntegrityCheck_${p.w}"
+ val io = IO(new Bundle {
+ val a_i = Input(new OpenTitanTileLink.A_Channel(p))
+ val fault = Output(Bool())
+ })
+
+ // Recreate the tl_h2d_cmd_intg_t struct for command integrity.
+ val cmd_w = 57
+ val cmd_data = Wire(UInt(cmd_w.W))
+ cmd_data := Cat(
+ io.a_i.user.instr_type,
+ io.a_i.address,
+ io.a_i.opcode,
+ io.a_i.mask
+ )
+
+ val cmd_encoder = Module(new SecdedEncoder(cmd_w))
+ cmd_encoder.io.data_i := cmd_data
+ val expected_cmd_intg = cmd_encoder.io.ecc_o
+
+ // Data integrity calculation.
+ val data_encoder = Module(new SecdedEncoder(p.w * 8))
+ data_encoder.io.data_i := io.a_i.data
+ val expected_data_intg = data_encoder.io.ecc_o
+
+ // A fault is generated if the received integrity does not match the
+ // calculated integrity.
+ io.fault := (expected_cmd_intg =/= io.a_i.user.cmd_intg) ||
+ (expected_data_intg =/= io.a_i.user.data_intg)
+}
+
+/**
+ * Generates TileLink integrity fields for the D-channel (Response).
+ */
+class ResponseIntegrityGen(p: TLULParameters) extends Module {
+ override val desiredName = s"ResponseIntegrityGen_${p.w}"
+ val io = IO(new Bundle {
+ val d_i = Input(new OpenTitanTileLink.D_Channel(p))
+ val d_o = Output(new OpenTitanTileLink.D_Channel(p))
+ })
+ // Ensure that we don't optimize out any parts of the bundle, at least
+ // via the Chisel toolchain.
+ dontTouch(io.d_i)
+ dontTouch(io.d_o)
+
+ // Passthrough for most fields.
+ io.d_o := io.d_i
+
+ // Recreate the tl_d2h_rsp_intg_t struct for response integrity.
+ val rsp_w = 57
+ val rsp_data = Wire(UInt(rsp_w.W))
+ rsp_data := Cat(
+ io.d_i.opcode,
+ io.d_i.size,
+ io.d_i.error
+ )
+
+
+ val rsp_encoder = Module(new SecdedEncoder(rsp_w))
+ rsp_encoder.io.data_i := rsp_data
+ io.d_o.user.rsp_intg := rsp_encoder.io.ecc_o
+
+ // Data integrity calculation.
+ val data_encoder = Module(new SecdedEncoder(p.w * 8))
+ data_encoder.io.data_i := io.d_i.data
+ io.d_o.user.data_intg := data_encoder.io.ecc_o
+}
+
+/**
+ * Checks TileLink integrity fields for the D-channel (Response).
+ */
+class ResponseIntegrityCheck(p: TLULParameters) extends Module {
+ override val desiredName = s"ResponseIntegrityCheck_${p.w}"
+ val io = IO(new Bundle {
+ val d_i = Input(new OpenTitanTileLink.D_Channel(p))
+ val fault = Output(Bool())
+ })
+
+ // Recreate the tl_d2h_rsp_intg_t struct for response integrity.
+ val rsp_w = 57
+ val rsp_data = Wire(UInt(rsp_w.W))
+ rsp_data := Cat(
+ io.d_i.opcode,
+ io.d_i.size,
+ io.d_i.error
+ )
+
+ val rsp_encoder = Module(new SecdedEncoder(rsp_w))
+ rsp_encoder.io.data_i := rsp_data
+ val expected_rsp_intg = rsp_encoder.io.ecc_o
+
+ // Data integrity calculation.
+ val data_encoder = Module(new SecdedEncoder(p.w * 8))
+ data_encoder.io.data_i := io.d_i.data
+ val expected_data_intg = data_encoder.io.ecc_o
+
+ // A fault is generated if the received integrity does not match the
+ // calculated integrity.
+ io.fault := (expected_rsp_intg =/= io.d_i.user.rsp_intg) ||
+ (expected_data_intg =/= io.d_i.user.data_intg)
+}
\ No newline at end of file
diff --git a/hdl/chisel/src/bus/TlulIntegrityTestbench.scala b/hdl/chisel/src/bus/TlulIntegrityTestbench.scala
new file mode 100644
index 0000000..285e109
--- /dev/null
+++ b/hdl/chisel/src/bus/TlulIntegrityTestbench.scala
@@ -0,0 +1,85 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package bus
+
+import chisel3._
+import chisel3.util._
+import kelvin.Parameters
+
+/**
+ * A testbench DUT that instantiates all TlulIntegrity modules so they can be
+ * tested with cocotb in a single simulation.
+ */
+class TlulIntegrityTestbench(p: TLULParameters) extends Module {
+ val io = IO(new Bundle {
+ // 1. RequestIntegrityGen instance
+ val req_gen_a_i = Flipped(Decoupled(new OpenTitanTileLink.A_Channel(p)))
+ val req_gen_a_o = Decoupled(new OpenTitanTileLink.A_Channel(p))
+
+ // 2. RequestIntegrityCheck instance
+ val req_check_a_i = Flipped(Decoupled(new OpenTitanTileLink.A_Channel(p)))
+ val req_check_fault = Output(Bool())
+
+ // 4. ResponseIntegrityGen instance
+ val rsp_gen_d_i = Flipped(Decoupled(new OpenTitanTileLink.D_Channel(p)))
+ val rsp_gen_d_o = Decoupled(new OpenTitanTileLink.D_Channel(p))
+
+ // 5. ResponseIntegrityCheck instance
+ val rsp_check_d_i = Flipped(Decoupled(new OpenTitanTileLink.D_Channel(p)))
+ val rsp_check_fault = Output(Bool())
+
+ })
+
+ // 1. RequestIntegrityGen instance
+ val req_gen = Module(new RequestIntegrityGen(p))
+ req_gen.io.a_i <> io.req_gen_a_i.bits
+ io.req_gen_a_o.bits := req_gen.io.a_o
+ io.req_gen_a_o.valid := io.req_gen_a_i.valid
+ io.req_gen_a_i.ready := io.req_gen_a_o.ready
+
+ // 2. RequestIntegrityCheck instance
+ val req_check = Module(new RequestIntegrityCheck(p))
+ req_check.io.a_i := io.req_check_a_i.bits
+ io.req_check_fault := req_check.io.fault
+ io.req_check_a_i.ready := true.B // Always ready to check
+
+ // 4. ResponseIntegrityGen instance
+ val rsp_gen = Module(new ResponseIntegrityGen(p))
+ rsp_gen.io.d_i := io.rsp_gen_d_i.bits
+ io.rsp_gen_d_o.bits := rsp_gen.io.d_o
+ io.rsp_gen_d_o.valid := io.rsp_gen_d_i.valid
+ io.rsp_gen_d_i.ready := io.rsp_gen_d_o.ready
+
+ // 5. ResponseIntegrityCheck instance
+ val rsp_check = Module(new ResponseIntegrityCheck(p))
+ rsp_check.io.d_i := io.rsp_check_d_i.bits
+ io.rsp_check_fault := rsp_check.io.fault
+ io.rsp_check_d_i.ready := true.B
+
+}
+
+import _root_.circt.stage.{ChiselStage,FirtoolOption}
+import chisel3.stage.ChiselGeneratorAnnotation
+import scala.annotation.nowarn
+
+@nowarn
+object EmitTlulIntegrityTestbench extends App {
+ val p = new Parameters
+ p.lsuDataBits = 128
+ (new ChiselStage).execute(
+ Array("--target", "systemverilog") ++ args,
+ Seq(ChiselGeneratorAnnotation(() => new TlulIntegrityTestbench(new bus.TLULParameters(p)))) ++ Seq(FirtoolOption("-enable-layers=Verification"))
+ )
+}
diff --git a/kelvin_test_utils/BUILD b/kelvin_test_utils/BUILD
index a358b45..94c7b6a 100644
--- a/kelvin_test_utils/BUILD
+++ b/kelvin_test_utils/BUILD
@@ -15,6 +15,22 @@
load("@kelvin_hw//third_party/python:requirements.bzl", "requirement")
py_library(
+ name = "TileLinkULInterface",
+ srcs = ["TileLinkULInterface.py"],
+ deps = [
+ requirement("cocotb"),
+ "//kelvin_test_utils:secded_golden",
+ ],
+ visibility = ["//visibility:public"],
+)
+
+py_library(
+ name = "secded_golden",
+ srcs = ["secded_golden.py"],
+ visibility = ["//visibility:public"],
+)
+
+py_library(
name = "core_mini_axi_sim_interface",
srcs = [
"core_mini_axi_interface.py",
diff --git a/kelvin_test_utils/TileLinkULInterface.py b/kelvin_test_utils/TileLinkULInterface.py
new file mode 100644
index 0000000..57c1cb5
--- /dev/null
+++ b/kelvin_test_utils/TileLinkULInterface.py
@@ -0,0 +1,273 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import cocotb
+from cocotb.clock import Clock
+from cocotb.queue import Queue
+from cocotb.triggers import FallingEdge, RisingEdge, with_timeout
+import cocotb.result
+import math
+
+from kelvin_test_utils.secded_golden import get_cmd_intg, get_data_intg, get_rsp_intg
+
+
+def create_a_channel_req(address,
+ data,
+ mask,
+ source=1,
+ size=None,
+ param=0,
+ width=32):
+ """Creates a standard TileLink-UL PutFullData request dictionary."""
+ num_bytes = width // 8
+ if size is None:
+ size = int(math.log2(num_bytes))
+
+ full_mask = (1 << num_bytes) - 1
+ opcode = 0 if mask == full_mask else 1 # PutFull vs PutPartial
+ txn = {
+ "opcode": opcode,
+ "param": param,
+ "size": size,
+ "source": source,
+ "address": address,
+ "mask": mask,
+ "data": data,
+ "user": {
+ "cmd_intg": 0,
+ "data_intg": 0,
+ "instr_type": 0,
+ "rsvd": 0
+ }
+ }
+ txn["user"]["cmd_intg"] = get_cmd_intg(txn, width=width)
+ txn["user"]["data_intg"] = get_data_intg(txn["data"], width=width)
+ return txn
+
+
+class TileLinkULInterface:
+ """A testbench interface for a TileLink-UL bus.
+
+ This class provides a high-level, transaction-based interface to a TileLink-UL
+ bus in the DUT. It uses cocotb queues and background coroutines ("agents")
+ to handle the low-level signal handshaking.
+
+ Args:
+ dut: The cocotb DUT object.
+ host_if_name (str, optional): The prefix for the host-side interface signals.
+ device_if_name (str, optional): The prefix for the device-side interface signals.
+ """
+
+ def __init__(self,
+ dut,
+ host_if_name=None,
+ device_if_name=None,
+ clock_name="clock",
+ reset_name="reset",
+ width=32):
+ self.dut = dut
+ self.clock = getattr(dut, clock_name)
+ self.reset = getattr(dut, reset_name)
+ self.width = width
+ self.name = host_if_name or device_if_name
+
+ if host_if_name is None and device_if_name is None:
+ raise ValueError(
+ "At least one of host_if_name or device_if_name must be provided."
+ )
+
+ self._agents = []
+
+ if host_if_name:
+ self.host_a_fifo = Queue()
+ self.host_d_fifo = Queue()
+ self._agents.append(
+ cocotb.start_soon(self._host_a_driver(host_if_name)))
+ self._agents.append(
+ cocotb.start_soon(self._host_d_monitor(host_if_name)))
+
+ if device_if_name:
+ self.device_a_fifo = Queue()
+ self.device_d_fifo = Queue()
+ self._device_a_ready = True # Default to being ready
+ self._agents.append(
+ cocotb.start_soon(self._device_a_monitor(device_if_name)))
+ self._agents.append(
+ cocotb.start_soon(self._device_d_driver(device_if_name)))
+
+ def device_a_set_ready(self, value):
+ """Set the ready signal for the device A channel monitor."""
+ self._device_a_ready = value
+
+ async def init(self):
+ """Starts the agents."""
+ # This method is currently a placeholder for starting agents.
+ # In this implementation, agents are started in the constructor.
+ # This can be extended if more complex initialization is needed.
+ pass
+
+ # --- Private Methods (Agents) ---
+
+ # slave_a{r|w}agent
+ async def _host_a_driver(self, prefix, timeout=4096):
+ """Drives the host A channel from the host_a_fifo."""
+ a_valid = getattr(self.dut, f"{prefix}_a_valid")
+ a_ready = getattr(self.dut, f"{prefix}_a_ready")
+
+ a_valid.value = 0
+ for prop in ["opcode", "param", "size", "source", "address", "mask", "data"]:
+ getattr(self.dut, f"{prefix}_a_bits_{prop}").value = 0
+
+ while True:
+ while True:
+ await RisingEdge(self.clock)
+ a_valid.value = 0
+ if self.host_a_fifo.qsize():
+ break
+ txn = await self.host_a_fifo.get()
+ a_valid.value = 1
+ for prop in ["opcode", "param", "size", "source", "address", "mask", "data"]:
+ getattr(self.dut, f"{prefix}_a_bits_{prop}").value = txn[prop]
+ for field, value in txn["user"].items():
+ getattr(self.dut,
+ f"{prefix}_a_bits_user_{field}").value = value
+ await FallingEdge(self.clock)
+ timeout_count = 0
+ while a_ready.value == 0:
+ await FallingEdge(self.clock)
+ timeout_count += 1
+ if timeout_count >= timeout:
+ assert False, "timeout waiting for a_ready"
+
+ # slave_bagent
+ async def _host_d_monitor(self, prefix):
+ """Monitors the host D channel and puts transactions into host_d_fifo."""
+ d_valid = getattr(self.dut, f"{prefix}_d_valid")
+ d_ready = getattr(self.dut, f"{prefix}_d_ready")
+
+ d_ready.value = 1
+ while True:
+ await RisingEdge(self.clock)
+ try:
+ if d_valid.value:
+ # Capture the transaction
+ txn = {'user': {}}
+ for prop in ["opcode", "param", "size", "source", "sink", "data", "error"]:
+ txn[prop] = getattr(self.dut, f"{prefix}_d_bits_{prop}").value
+ user_fields = ["rsp_intg", "data_intg"]
+ for field in user_fields:
+ signal_name = f"{prefix}_d_bits_user_{field}"
+ if hasattr(self.dut, signal_name):
+ txn["user"][field] = getattr(self.dut, signal_name).value
+
+ await self.host_d_fifo.put(txn)
+ except Exception as e:
+ print('X seen in _host_d_monitor: ' + str(e) + ' ' + prefix)
+ # raise e
+
+ # master_aragent
+ async def _device_a_monitor(self, prefix):
+ """Monitors the device A channel and puts transactions into device_a_fifo."""
+ a_valid = getattr(self.dut, f"{prefix}_a_valid")
+ a_ready = getattr(self.dut, f"{prefix}_a_ready")
+
+ a_ready.value = 1
+ while True:
+ await RisingEdge(self.clock)
+ try:
+ if a_valid.value:
+ txn = {"user": {}}
+ for prop in ["opcode", "param", "size", "source", "address", "mask", "data"]:
+ txn[prop] = getattr(self.dut, f"{prefix}_a_bits_{prop}").value
+ user_fields = ["cmd_intg", "data_intg", "instr_type", "rsvd"]
+ for field in user_fields:
+ signal_name = f"{prefix}_a_bits_user_{field}"
+ if hasattr(self.dut, signal_name):
+ txn["user"][field] = getattr(self.dut,
+ signal_name).value
+ await self.device_a_fifo.put(txn)
+ except Exception as e:
+ print('X seen in _device_a_monitor: ' + str(e) + ' ' + prefix)
+
+ # master_bagent
+ async def _device_d_driver(self, prefix, timeout=4096):
+ """Drives the device D channel from the device_d_fifo."""
+ d_valid = getattr(self.dut, f"{prefix}_d_valid")
+ d_ready = getattr(self.dut, f"{prefix}_d_ready")
+
+ d_valid.value = 0
+ for prop in ["opcode", "param", "size", "source", "sink", "data", "error"]:
+ getattr(self.dut, f"{prefix}_d_bits_{prop}").value = 0
+
+ while True:
+ while True:
+ await RisingEdge(self.clock)
+ d_valid.value = 0
+ if self.device_d_fifo.qsize():
+ break
+ txn = await self.device_d_fifo.get()
+ d_valid.value = 1
+ for prop in ["opcode", "param", "size", "source", "sink", "data", "error"]:
+ getattr(self.dut, f"{prefix}_d_bits_{prop}").value = txn[prop]
+ for field, value in txn["user"].items():
+ getattr(self.dut,
+ f"{prefix}_d_bits_user_{field}").value = value
+ await FallingEdge(self.clock)
+ timeout_count = 0
+ while d_ready.value == 0:
+ await FallingEdge(self.clock)
+ timeout_count += 1
+ if timeout_count >= timeout:
+ assert False, "timeout waiting for d_ready"
+
+ # --- Public API Methods ---
+
+ async def host_put(self, txn):
+ """Send a PutFullData or PutPartialData request from the host."""
+ await self.host_a_fifo.put(txn)
+
+ async def host_get_response(self):
+ """Get a response from the host D channel."""
+ return await self.host_d_fifo.get()
+
+ async def device_get_request(self):
+ """Get a request from the device A channel."""
+ return await self.device_a_fifo.get()
+
+ async def device_respond(self,
+ opcode,
+ param,
+ size,
+ source,
+ sink=0,
+ data=0,
+ error=0,
+ width=32):
+ """Send a response from the device."""
+ txn = {
+ "opcode": opcode,
+ "param": param,
+ "size": size,
+ "source": source,
+ "sink": sink,
+ "data": data,
+ "error": error,
+ "user": {
+ "rsp_intg": 0,
+ "data_intg": 0
+ }
+ }
+ txn["user"]["rsp_intg"] = get_rsp_intg(txn, width)
+ txn["user"]["data_intg"] = get_data_intg(txn["data"], width)
+ await self.device_d_fifo.put(txn)
diff --git a/kelvin_test_utils/secded_golden.py b/kelvin_test_utils/secded_golden.py
new file mode 100644
index 0000000..c4d2877
--- /dev/null
+++ b/kelvin_test_utils/secded_golden.py
@@ -0,0 +1,136 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Golden model for TileLink-UL integrity calculations."""
+import struct
+
+
+def _parity(n):
+ """Calculates the parity of an integer."""
+ p = 0
+ while n > 0:
+ p ^= (n & 1)
+ n >>= 1
+ return p
+
+
+def secded_inv_39_32_enc(data):
+ """Golden model for prim_secded_inv_39_32_enc. Returns 7-bit ECC."""
+ data_o = data # Start with the 32-bit data
+
+ # Calculate parity bits based on the initial 32-bit data
+ p0 = _parity(data_o & 0x002606BD25)
+ p1 = _parity(data_o & 0x00DEBA8050)
+ p2 = _parity(data_o & 0x00413D89AA)
+ p3 = _parity(data_o & 0x0031234ED1)
+ p4 = _parity(data_o & 0x00C2C1323B)
+ p5 = _parity(data_o & 0x002DCC624C)
+ p6 = _parity(data_o & 0x0098505586)
+
+ # Assemble the 39-bit word
+ data_o |= p0 << 32
+ data_o |= p1 << 33
+ data_o |= p2 << 34
+ data_o |= p3 << 35
+ data_o |= p4 << 36
+ data_o |= p5 << 37
+ data_o |= p6 << 38
+
+ # XOR the full 39-bit word with the inversion constant
+ inverted_data = data_o ^ 0x2A00000000
+
+ # Return the top 7 bits (the ECC)
+ return inverted_data >> 32
+
+
+def secded_inv_64_57_enc(data):
+ """Golden model for prim_secded_inv_64_57_enc. Returns 7-bit ECC."""
+ data_o = data # Start with the 57-bit data
+
+ # Calculate parity bits based on the initial 57-bit data
+ p0 = _parity(data_o & 0x0103FFF800007FFF)
+ p1 = _parity(data_o & 0x017C1FF801FF801F)
+ p2 = _parity(data_o & 0x01BDE1F87E0781E1)
+ p3 = _parity(data_o & 0x01DEEE3B8E388E22)
+ p4 = _parity(data_o & 0x01EF76CDB2C93244)
+ p5 = _parity(data_o & 0x01F7BB56D5525488)
+ p6 = _parity(data_o & 0x01FBDDA769A46910)
+
+ # Assemble the 64-bit word
+ data_o |= p0 << 57
+ data_o |= p1 << 58
+ data_o |= p2 << 59
+ data_o |= p3 << 60
+ data_o |= p4 << 61
+ data_o |= p5 << 62
+ data_o |= p6 << 63
+
+ # XOR the full 64-bit word with the inversion constant
+ inverted_data = data_o ^ 0x5400000000000000
+
+ # Return the top 7 bits (the ECC)
+ return inverted_data >> 57
+
+
+def get_cmd_intg(a_channel, width=128):
+ """Packs A-channel fields and returns the command integrity."""
+ # Packing order (MSB to LSB) from TlulIntegrity.scala
+ # Cat(instr_type, address, opcode, mask)
+ # instr_type: 4 bits
+ # address: 32 bits
+ # opcode: 3 bits
+ # mask: variable bits
+ mask_width = width // 8
+ packed = ((int(a_channel["user"]["instr_type"]) << (32 + 3 + mask_width)) |
+ (int(a_channel["address"]) <<
+ (3 + mask_width)) | (int(a_channel["opcode"]) << mask_width) |
+ (int(a_channel["mask"])))
+
+ return secded_inv_64_57_enc(packed)
+
+
+def get_data_intg(data, width=32):
+ """Returns the data integrity."""
+ dataint = int(data)
+ if width == 32:
+ return secded_inv_39_32_enc(dataint)
+ elif width == 128:
+ # Folded scheme
+ d0 = dataint & 0xFFFFFFFF
+ d1 = (dataint >> 32) & 0xFFFFFFFF
+ d2 = (dataint >> 64) & 0xFFFFFFFF
+ d3 = (dataint >> 96) & 0xFFFFFFFF
+ ecc0 = secded_inv_39_32_enc(d0)
+ ecc1 = secded_inv_39_32_enc(d1)
+ ecc2 = secded_inv_39_32_enc(d2)
+ ecc3 = secded_inv_39_32_enc(d3)
+ return ecc0 ^ ecc1 ^ ecc2 ^ ecc3
+ else:
+ raise ValueError(f"Unsupported data width: {width}")
+
+
+import math
+
+
+def get_rsp_intg(d_channel, width=128):
+ """Packs D-channel fields and returns the response integrity."""
+ # Packing order (MSB to LSB) from TlulIntegrity.scala
+ # Cat(opcode, size, error)
+ # opcode: 3 bits
+ # size: variable bits
+ # error: 1 bit
+ size_width = math.ceil(math.log2(width // 8))
+ packed = ((int(d_channel["opcode"]) << (size_width + 1)) |
+ (int(d_channel["size"]) << 1) | (int(d_channel["error"])))
+
+ return secded_inv_64_57_enc(packed)
diff --git a/tests/cocotb/tlul/BUILD b/tests/cocotb/tlul/BUILD
new file mode 100644
index 0000000..9b16b9c
--- /dev/null
+++ b/tests/cocotb/tlul/BUILD
@@ -0,0 +1,123 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+load("@kelvin_hw//rules:coco_tb.bzl", "cocotb_test_suite", "verilator_cocotb_model")
+load("@kelvin_hw//third_party/python:requirements.bzl", "requirement")
+
+package(default_visibility = ["//visibility:public"])
+
+load(
+ "//tests/cocotb:build_defs.bzl",
+ "VCS_BUILD_ARGS",
+ "VCS_DEFINES",
+ "VCS_TEST_ARGS",
+)
+
+# BEGIN_TESTCASES_FOR_tlul_integrity_cocotb_test
+TLUL_INTEGRITY_TESTCASES = [
+ "test_request_integrity_gen",
+ "test_request_integrity_check",
+ "test_response_integrity_gen",
+ "test_response_integrity_check",
+]
+# END_TESTCASES_FOR_tlul_integrity_cocotb_test
+
+cocotb_test_suite(
+ name = "tlul_integrity_cocotb_test",
+ simulators = ["verilator", "vcs"],
+ testcases = TLUL_INTEGRITY_TESTCASES,
+ testcases_vname = "TLUL_INTEGRITY_TESTCASES",
+ tests_kwargs = {
+ "hdl_toplevel": "TlulIntegrityTestbench",
+ "test_module": ["test_tlul_integrity.py"],
+ "deps": [
+ "//kelvin_test_utils:TileLinkULInterface",
+ "//kelvin_test_utils:secded_golden",
+ ],
+ "waves": True,
+ },
+ verilator_model = "//hdl/chisel/src/bus:tlul_integrity_testbench_model",
+ vcs_verilog_sources = ["//hdl/chisel/src/bus:tlul_integrity_testbench_cc_library_verilog"],
+ vcs_build_args = VCS_BUILD_ARGS,
+ vcs_test_args = VCS_TEST_ARGS,
+ vcs_defines = VCS_DEFINES,
+)
+
+
+
+# BEGIN_TESTCASES_FOR_secded_encoder_cocotb_test
+SECDED_ENCODER_TESTCASES = [
+ "test_secded_encoder",
+]
+# END_TESTCASES_FOR_secded_encoder_cocotb_test
+
+cocotb_test_suite(
+ name = "secded_encoder_cocotb_test",
+ simulators = ["verilator", "vcs"],
+ testcases = SECDED_ENCODER_TESTCASES,
+ testcases_vname = "SECDED_ENCODER_TESTCASES",
+ tests_kwargs = {
+ "hdl_toplevel": "SecdedEncoderTestbench128",
+ "test_module": ["test_secded_encoder.py"],
+ "deps": [
+ "//kelvin_test_utils:secded_golden",
+ ],
+ "waves": True,
+ },
+ verilator_model = "//hdl/chisel/src/bus:secded_encoder_testbench_model",
+ vcs_verilog_sources = ["//hdl/chisel/src/bus:secded_encoder_testbench_cc_library_verilog"],
+ vcs_build_args = VCS_BUILD_ARGS,
+ vcs_test_args = VCS_TEST_ARGS,
+ vcs_defines = VCS_DEFINES,
+)
+
+cocotb_test_suite(
+ name = "secded_encoder_32_cocotb_test",
+ simulators = ["verilator", "vcs"],
+ testcases = SECDED_ENCODER_TESTCASES,
+ testcases_vname = "SECDED_ENCODER_TESTCASES",
+ tests_kwargs = {
+ "hdl_toplevel": "SecdedEncoderTestbench32",
+ "test_module": ["test_secded_encoder.py"],
+ "deps": [
+ "//kelvin_test_utils:secded_golden",
+ ],
+ "waves": True,
+ },
+ verilator_model = "//hdl/chisel/src/bus:secded_encoder_testbench_32_model",
+ vcs_verilog_sources = ["//hdl/chisel/src/bus:secded_encoder_testbench_32_cc_library_verilog"],
+ vcs_build_args = VCS_BUILD_ARGS,
+ vcs_test_args = VCS_TEST_ARGS,
+ vcs_defines = VCS_DEFINES,
+)
+
+cocotb_test_suite(
+ name = "secded_encoder_57_cocotb_test",
+ simulators = ["verilator", "vcs"],
+ testcases = SECDED_ENCODER_TESTCASES,
+ testcases_vname = "SECDED_ENCODER_TESTCASES",
+ tests_kwargs = {
+ "hdl_toplevel": "SecdedEncoderTestbench57",
+ "test_module": ["test_secded_encoder.py"],
+ "deps": [
+ "//kelvin_test_utils:secded_golden",
+ ],
+ "waves": True,
+ },
+ verilator_model = "//hdl/chisel/src/bus:secded_encoder_testbench_57_model",
+ vcs_verilog_sources = ["//hdl/chisel/src/bus:secded_encoder_testbench_57_cc_library_verilog"],
+ vcs_build_args = VCS_BUILD_ARGS,
+ vcs_test_args = VCS_TEST_ARGS,
+ vcs_defines = VCS_DEFINES,
+)
diff --git a/tests/cocotb/tlul/test_secded_encoder.py b/tests/cocotb/tlul/test_secded_encoder.py
new file mode 100644
index 0000000..175e59f
--- /dev/null
+++ b/tests/cocotb/tlul/test_secded_encoder.py
@@ -0,0 +1,69 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import cocotb
+from cocotb.clock import Clock
+from cocotb.triggers import RisingEdge, ClockCycles
+import random
+
+from kelvin_test_utils.secded_golden import get_data_intg, secded_inv_39_32_enc, secded_inv_64_57_enc
+
+
+async def setup_dut(dut):
+ """Common setup for all tests."""
+ clock = Clock(dut.clock, 10, unit="us")
+ cocotb.start_soon(clock.start())
+
+ dut.reset.value = 1
+ await ClockCycles(dut.clock, 2)
+ dut.reset.value = 0
+ await RisingEdge(dut.clock)
+
+
+@cocotb.test()
+async def test_secded_encoder(dut):
+ """Test that the SecdedEncoder module matches the golden model for random data."""
+ await setup_dut(dut)
+
+ # Determine the data width from the DUT.
+ data_width = len(dut.io_data_i)
+ num_iterations = 1000
+
+ for i in range(num_iterations):
+ # Generate a random integer of the correct width.
+ random_data = random.getrandbits(data_width)
+
+ # Drive the random data into the DUT.
+ dut.io_data_i.value = random_data
+ await RisingEdge(dut.clock)
+
+ # Get the ECC from the DUT.
+ dut_ecc = dut.io_ecc_o.value
+
+ # Calculate the expected ECC using the golden model.
+ if data_width == 32:
+ golden_ecc = secded_inv_39_32_enc(random_data)
+ elif data_width == 57:
+ golden_ecc = secded_inv_64_57_enc(random_data)
+ elif data_width == 128:
+ golden_ecc = get_data_intg(random_data, width=data_width)
+ else:
+ raise ValueError(f"Unsupported data width: {data_width}")
+
+ # Compare the DUT's output with the golden model.
+ assert dut_ecc == golden_ecc, f"Mismatch on iteration {i}: data={hex(random_data)}, dut_ecc={hex(dut_ecc)}, golden_ecc={hex(golden_ecc)}"
+
+ dut._log.info(
+ f"Successfully compared {num_iterations} random data values for data width {data_width}."
+ )
diff --git a/tests/cocotb/tlul/test_tlul_integrity.py b/tests/cocotb/tlul/test_tlul_integrity.py
new file mode 100644
index 0000000..13cce8e
--- /dev/null
+++ b/tests/cocotb/tlul/test_tlul_integrity.py
@@ -0,0 +1,280 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import cocotb
+from cocotb.clock import Clock
+from cocotb.triggers import RisingEdge, ClockCycles, with_timeout
+
+from kelvin_test_utils.TileLinkULInterface import TileLinkULInterface, create_a_channel_req
+from kelvin_test_utils.secded_golden import get_cmd_intg, get_data_intg, get_rsp_intg
+
+
+def create_d_channel_rsp(opcode,
+ data,
+ size,
+ source,
+ param=0,
+ sink=0,
+ error=False):
+ """Creates a standard TileLink-UL D-channel response dictionary."""
+ return {
+ "opcode": opcode,
+ "param": param,
+ "size": size,
+ "source": source,
+ "sink": sink,
+ "data": data,
+ "error": error,
+ "user": {
+ "rsp_intg": 0,
+ "data_intg": 0
+ }
+ }
+
+
+async def setup_dut(dut):
+ """Common setup for all tests."""
+ clock = Clock(dut.clock, 10, unit="us")
+ cocotb.start_soon(clock.start())
+
+ dut.reset.value = 1
+ await ClockCycles(dut.clock, 2)
+ dut.reset.value = 0
+ await RisingEdge(dut.clock)
+
+
+@cocotb.test()
+async def test_request_integrity_gen(dut):
+ """Test that the RequestIntegrityGen module generates correct integrity."""
+ await setup_dut(dut)
+
+ # Drive the input A-channel
+ req = create_a_channel_req(address=0x1000,
+ data=0x112233445566778899aabbccddeeff00,
+ mask=0xFFFF,
+ width=128)
+ dut.io_req_gen_a_i_valid.value = 1
+ dut.io_req_gen_a_i_bits_opcode.value = req["opcode"]
+ dut.io_req_gen_a_i_bits_param.value = req["param"]
+ dut.io_req_gen_a_i_bits_size.value = req["size"]
+ dut.io_req_gen_a_i_bits_source.value = req["source"]
+ dut.io_req_gen_a_i_bits_address.value = req["address"]
+ dut.io_req_gen_a_i_bits_mask.value = req["mask"]
+ dut.io_req_gen_a_i_bits_data.value = req["data"]
+ dut.io_req_gen_a_i_bits_user_cmd_intg.value = get_cmd_intg(req)
+ dut.io_req_gen_a_i_bits_user_data_intg.value = get_data_intg(req["data"],
+ width=128)
+ dut.io_req_gen_a_i_bits_user_rsvd.value = 0
+ dut.io_req_gen_a_i_bits_user_instr_type.value = 0
+
+ # Signal that we are ready to accept the output.
+ dut.io_req_gen_a_o_ready.value = 1
+
+ # TODO: Timeout loop
+ await RisingEdge(dut.clock)
+
+ # Check the output A-channel
+ assert dut.io_req_gen_a_o_valid.value
+ assert dut.io_req_gen_a_o_bits_opcode.value == req["opcode"]
+ assert dut.io_req_gen_a_o_bits_address.value == req["address"]
+ assert dut.io_req_gen_a_o_bits_data.value == req["data"]
+
+ assert dut.io_req_gen_a_o_bits_user_cmd_intg.value == get_cmd_intg(req)
+ assert dut.io_req_gen_a_o_bits_user_data_intg.value == get_data_intg(
+ req["data"], width=128)
+
+
+@cocotb.test()
+async def test_request_integrity_check(dut):
+ """Test that the RequestIntegrityCheck module correctly identifies faults."""
+ await setup_dut(dut)
+ req = create_a_channel_req(address=0x1000,
+ data=0x112233445566778899aabbccddeeff00,
+ mask=0xFFFF,
+ width=128)
+
+ # --- Transaction 1: Correct integrity ---
+ dut.io_req_check_a_i_valid.value = 1
+ dut.io_req_check_a_i_bits_opcode.value = req["opcode"]
+ dut.io_req_check_a_i_bits_param.value = req["param"]
+ dut.io_req_check_a_i_bits_size.value = req["size"]
+ dut.io_req_check_a_i_bits_source.value = req["source"]
+ dut.io_req_check_a_i_bits_address.value = req["address"]
+ dut.io_req_check_a_i_bits_mask.value = req["mask"]
+ dut.io_req_check_a_i_bits_data.value = req["data"]
+ dut.io_req_check_a_i_bits_user_cmd_intg.value = get_cmd_intg(req)
+ dut.io_req_check_a_i_bits_user_data_intg.value = get_data_intg(req["data"],
+ width=128)
+ dut.io_req_check_a_i_bits_user_rsvd.value = 0
+ dut.io_req_check_a_i_bits_user_instr_type.value = 0
+
+ for _ in range(10):
+ if dut.io_req_check_a_i_ready.value:
+ break
+ await RisingEdge(dut.clock)
+ else:
+ assert False, "Timeout waiting for dut.io_req_check_a_i_ready"
+
+ await RisingEdge(dut.clock)
+ dut.io_req_check_a_i_valid.value = 0
+ await RisingEdge(dut.clock)
+ assert not dut.io_req_check_fault.value
+ await ClockCycles(dut.clock, 5) # Delay for clarity in logs/waves
+
+ # --- Transaction 2: Command integrity fault ---
+ dut.io_req_check_a_i_valid.value = 1
+ dut.io_req_check_a_i_bits_data.value = req["data"]
+ correct_cmd_intg = get_cmd_intg(req)
+ dut.io_req_check_a_i_bits_user_cmd_intg.value = ~correct_cmd_intg & 0x7F
+
+ for _ in range(10):
+ if dut.io_req_check_a_i_ready.value:
+ break
+ await RisingEdge(dut.clock)
+ else:
+ assert False, "Timeout waiting for dut.io_req_check_a_i_ready"
+
+ await RisingEdge(dut.clock)
+ dut.io_req_check_a_i_valid.value = 0
+ await RisingEdge(dut.clock)
+ assert dut.io_req_check_fault.value
+ await ClockCycles(dut.clock, 5) # Delay for clarity in logs/waves
+
+ # --- Transaction 3: Data integrity fault ---
+ dut.io_req_check_a_i_valid.value = 1
+ dut.io_req_check_a_i_bits_data.value = req["data"]
+ dut.io_req_check_a_i_bits_user_cmd_intg.value = get_cmd_intg(
+ req) # Restore cmd_intg
+ correct_data_intg = get_data_intg(req["data"], width=128)
+ dut.io_req_check_a_i_bits_user_data_intg.value = ~correct_data_intg & 0x7F
+
+ for _ in range(10):
+ if dut.io_req_check_a_i_ready.value:
+ break
+ await RisingEdge(dut.clock)
+ else:
+ assert False, "Timeout waiting for dut.io_req_check_a_i_ready"
+
+ await RisingEdge(dut.clock)
+ dut.io_req_check_a_i_valid.value = 0
+ await RisingEdge(dut.clock)
+ assert dut.io_req_check_fault.value
+ await RisingEdge(dut.clock)
+
+
+@cocotb.test()
+async def test_response_integrity_gen(dut):
+ """Test that the ResponseIntegrityGen module generates correct integrity."""
+ await setup_dut(dut)
+
+ # Drive the input D-channel
+ rsp = create_d_channel_rsp(opcode=1,
+ data=0x112233445566778899aabbccddeeff00,
+ size=4,
+ source=1)
+ dut.io_rsp_gen_d_i_valid.value = 1
+ dut.io_rsp_gen_d_i_bits_opcode.value = rsp["opcode"]
+ dut.io_rsp_gen_d_i_bits_param.value = rsp["param"]
+ dut.io_rsp_gen_d_i_bits_size.value = rsp["size"]
+ dut.io_rsp_gen_d_i_bits_source.value = rsp["source"]
+ dut.io_rsp_gen_d_i_bits_sink.value = rsp["sink"]
+ dut.io_rsp_gen_d_i_bits_data.value = rsp["data"]
+ dut.io_rsp_gen_d_i_bits_error.value = rsp["error"]
+
+ # Signal that we are ready to accept the output.
+ dut.io_rsp_gen_d_o_ready.value = 1
+
+ await RisingEdge(dut.clock)
+
+ # Check the output D-channel
+ assert dut.io_rsp_gen_d_o_valid.value
+ assert dut.io_rsp_gen_d_o_bits_opcode.value == rsp["opcode"]
+ assert dut.io_rsp_gen_d_o_bits_data.value == rsp["data"]
+
+ assert dut.io_rsp_gen_d_o_bits_user_rsp_intg.value == get_rsp_intg(rsp)
+ assert dut.io_rsp_gen_d_o_bits_user_data_intg.value == get_data_intg(
+ rsp["data"], width=128)
+
+
+@cocotb.test()
+async def test_response_integrity_check(dut):
+ """Test that the ResponseIntegrityCheck module correctly identifies faults."""
+ await setup_dut(dut)
+ rsp = create_d_channel_rsp(opcode=1,
+ data=0x112233445566778899aabbccddeeff00,
+ size=4,
+ source=1)
+
+ # --- Transaction 1: Correct integrity ---
+ dut.io_rsp_check_d_i_valid.value = 1
+ dut.io_rsp_check_d_i_bits_opcode.value = rsp["opcode"]
+ dut.io_rsp_check_d_i_bits_param.value = rsp["param"]
+ dut.io_rsp_check_d_i_bits_size.value = rsp["size"]
+ dut.io_rsp_check_d_i_bits_source.value = rsp["source"]
+ dut.io_rsp_check_d_i_bits_sink.value = rsp["sink"]
+ dut.io_rsp_check_d_i_bits_data.value = rsp["data"]
+ dut.io_rsp_check_d_i_bits_error.value = rsp["error"]
+ dut.io_rsp_check_d_i_bits_user_rsp_intg.value = get_rsp_intg(rsp)
+ dut.io_rsp_check_d_i_bits_user_data_intg.value = get_data_intg(rsp["data"],
+ width=128)
+
+ for _ in range(10):
+ if dut.io_rsp_check_d_i_ready.value:
+ break
+ await RisingEdge(dut.clock)
+ else:
+ assert False, "Timeout waiting for dut.io_rsp_check_d_i_ready"
+ await RisingEdge(dut.clock)
+ dut.io_rsp_check_d_i_valid.value = 0
+ await RisingEdge(dut.clock)
+ assert not dut.io_rsp_check_fault.value
+ await ClockCycles(dut.clock, 5) # Delay for clarity in logs/waves
+
+ # --- Transaction 2: Response integrity fault ---
+ dut.io_rsp_check_d_i_valid.value = 1
+ dut.io_rsp_check_d_i_bits_data.value = rsp["data"] # Keep data same
+ correct_rsp_intg = get_rsp_intg(rsp)
+ dut.io_rsp_check_d_i_bits_user_rsp_intg.value = ~correct_rsp_intg & 0x7F
+
+ for _ in range(10):
+ if dut.io_rsp_check_d_i_ready.value:
+ break
+ await RisingEdge(dut.clock)
+ else:
+ assert False, "Timeout waiting for dut.io_rsp_check_d_i_ready"
+ await RisingEdge(dut.clock)
+ dut.io_rsp_check_d_i_valid.value = 0
+ await RisingEdge(dut.clock)
+ assert dut.io_rsp_check_fault.value
+ await ClockCycles(dut.clock, 5) # Delay for clarity in logs/waves
+
+ # --- Transaction 3: Data integrity fault ---
+ dut.io_rsp_check_d_i_valid.value = 1
+ dut.io_rsp_check_d_i_bits_data.value = rsp["data"]
+ dut.io_rsp_check_d_i_bits_user_rsp_intg.value = get_rsp_intg(
+ rsp) # Restore rsp_intg
+ correct_data_intg = get_data_intg(rsp["data"], width=128)
+ dut.io_rsp_check_d_i_bits_user_data_intg.value = ~correct_data_intg & 0x7F
+
+ for _ in range(10):
+ if dut.io_rsp_check_d_i_ready.value:
+ break
+ await RisingEdge(dut.clock)
+ else:
+ assert False, "Timeout waiting for dut.io_rsp_check_d_i_ready"
+ await RisingEdge(dut.clock)
+ dut.io_rsp_check_d_i_valid.value = 0
+ await RisingEdge(dut.clock)
+ assert dut.io_rsp_check_fault.value
+ await RisingEdge(dut.clock)
