add ftb test

tests/FTB/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+pythonpath = ./src ../../out/picker_out_FTB ../../utils
+testpaths = test

tests/FTB/src/ftb_env/ftb.py

@@ -1,4 +1,5 @@
-from .utils import *
+from ftb_utils import *
+
 
 class FTBSlot:
     def __init__(self):
@@ -8,6 +9,10 @@ class FTBSlot:
         self.tarStart = 0
         self.sharing = 0
 
+    def print_slot(self):
+        print("valid ", self.valid, "offset ", self.offset, "lower ", self.lower, "tarStart ", self.tarStart,
+              "sharing ", self.sharing)
+
     def __str__(self, pc, is_cond_branch):
         str = ""
         if not self.valid:
@@ -79,11 +84,14 @@ class FTBEntry:
         d["slot_valids_0"] = self.brSlot.valid
         d["slot_valids_1"] = self.tailSlot.valid
         d["targets_0"] = get_target_addr(pc, self.brSlot.tarStart, self.brSlot.lower, 12)
-        d["targets_1"] = get_target_addr(pc, self.tailSlot.tarStart, self.tailSlot.lower, 12 if self.tailSlot.sharing else 20)
+        d["targets_1"] = get_target_addr(pc, self.tailSlot.tarStart, self.tailSlot.lower,
+                                         12 if self.tailSlot.sharing else 20)
         d["offsets_0"] = self.brSlot.offset
         d["offsets_1"] = self.tailSlot.offset
         d["fallThroughErr"] = get_fallthrough_addr(pc, self.pftAddr, self.carry) <= pc
-        d["fallThroughAddr"] = get_fallthrough_addr(pc, self.pftAddr, self.carry) if not d["fallThroughErr"] else pc + (PREDICT_WIDTH_BYTES)
+        # pdb.set_trace()
+        d["fallThroughAddr"] = get_fallthrough_addr(pc, self.pftAddr, self.carry) if not d["fallThroughErr"] else pc + (
+            PREDICT_WIDTH_BYTES)
         d["is_jal"] = self.isJal
         d["is_jalr"] = self.isJalr
         d["is_call"] = self.isCall
@@ -94,7 +102,6 @@ class FTBEntry:
         d["br_taken_mask_1"] = self.always_taken[1]
         d["jalr_target"] = get_target_addr(pc, self.tailSlot.tarStart, self.tailSlot.lower, 20)
 
-
     def __dict__(self):
         return {
             "brSlots_0_offset": self.brSlot.offset,
@@ -180,6 +187,21 @@ class FTBEntry:
         str += f"*\tlast_may_be_rvi_call: {self.last_may_be_rvi_call}, always_taken: {self.always_taken}\n"
         return str
 
+    def print_entry(self):
+        # pdb.set_trace()
+        print("valid", self.valid)
+        self.brSlot.print_slot()
+        self.tailSlot.print_slot()
+        print("pftAddr", self.pftAddr)
+        print("carry", self.carry)
+        print("isCall", self.isCall)
+        print("isRet", self.isRet)
+        print("isJal", self.isJal)
+        print("isJalr", self.isJalr)
+        print("last_may_be_rvi_call", self.last_may_be_rvi_call)
+        print("always_taken", self.always_taken)
+
+
 class FTBProvider():
     def __init__(self):
         self.entries = {}
@@ -187,9 +209,19 @@ class FTBProvider():
     def update(self, update_request):
         if update_request["valid"]:
             self.entries[update_request["bits_pc"]] = FTBEntry.from_dict(update_request["ftb_entry"])
+            # print("==========update provider ", hex(update_request["bits_pc"]))
+            # if self.entries[update_request["bits_pc"]] is not None:
+            #     self.entries[update_request["bits_pc"]].print_entry()
+            # pdb.set_trace()
+            # print(self.entries[update_request["bits_pc"]])
 
     def provide_ftb_entry(self, fire, pc):
         if fire and pc in self.entries:
+            # print("==========get provider ", hex(pc))
+            # if self.entries[pc] is not None:
+            #     self.entries[pc].print_entry()
+            # pdb.set_trace()
+            # print(self.entries[pc])
             return self.entries[pc]
         else:
             return None

tests/FTB/src/ftb_env/ftb_wrapper.py

@@ -0,0 +1,191 @@
+import random
+
+from mlvp import Bundle, setup_logging, WriteMode
+from mlvp.logger import ERROR
+
+from ftb_env.ftb import *
+
+setup_logging(ERROR)
+
+
+class PipelineCtrlBundle(Bundle):
+    signals = ["s0_fire_0", "s0_fire_1", "s0_fire_2", "s0_fire_3",
+               "s1_fire_0", "s1_fire_1", "s1_fire_2", "s1_fire_3",
+               "s2_fire_0", "s2_fire_1", "s2_fire_2", "s2_fire_3",
+               "s3_fire_0", "s3_fire_1", "s3_fire_2", "s3_fire_3",
+               "s1_ready", "s2_ready", "s3_ready",
+               "s2_redirect", "s3_redirect"]
+
+
+class EnableCtrlBundle(Bundle):
+    signals = ["ubtb_enable", "btb_enable", "bim_enable", "tage_enable",
+               "sc_enable", "ras_enable", "loop_enable"]
+
+
+class FTBEntryBundle(Bundle):
+    signals = ["brSlots_0_offset", "brSlots_0_lower", "brSlots_0_tarStat", "brSlots_0_sharing", "brSlots_0_valid",
+               "tailSlot_offset", "tailSlot_lower", "tailSlot_tarStat", "tailSlot_sharing", "tailSlot_valid",
+               "pftAddr", "carry", "isCall", "isRet", "isJalr", "last_may_be_rvi_call",
+               "always_taken_0", "always_taken_1", "valid"]
+
+
+class UpdateBundle(Bundle):
+    signals = ["valid", "bits_pc", "bits_old_entry", "bits_meta"]
+
+    def __init__(self):
+        super().__init__()
+        self.ftb_entry = FTBEntryBundle.from_prefix("bits_ftb_entry_")
+
+
+class FullBranchPredirectionBundle(Bundle):
+    signals = ["hit", "slot_valids_0", "slot_valids_1", "targets_0", "targets_1",
+               "offsets_0", "offsets_1", "fallThroughAddr", "fallThroughErr",
+               "is_jal", "is_jalr", "is_call", "is_ret", "is_br_sharing",
+               "last_may_be_rvi_call",
+               "br_taken_mask_0", "br_taken_mask_1",
+               "jalr_target"]
+
+
+class BranchPredictionBundle(Bundle):
+    signals = ["pc_3", "valid", "hasRedirect", "ftq_idx"]
+
+    def __init__(self):
+        super().__init__()
+        self.full_pred = FullBranchPredirectionBundle.from_regex(r"full_pred_\d_(.*)")
+
+
+class BranchPredictionResp(Bundle):
+    signals = ["last_stage_meta"]
+
+    def __init__(self):
+        super().__init__()
+        self.s1 = BranchPredictionBundle.from_prefix("s1_")
+        self.s2 = BranchPredictionBundle.from_prefix("s2_")
+        self.s3 = BranchPredictionBundle.from_prefix("s3_")
+        self.last_stage_ftb_entry = FTBEntryBundle.from_prefix("last_stage_ftb_entry_")
+
+
+class FTBWrapper(Bundle):
+
+    def __init__(self, dut):
+        super().__init__()
+        self.dut = dut
+
+        self.dut_out = BranchPredictionResp.from_prefix("io_out_").set_name("FTB_out").bind(self.dut)
+        self.dut_update = UpdateBundle.from_prefix("io_update_").set_name("FTB_update").bind(self.dut)
+        self.pipeline_ctrl = PipelineCtrlBundle.from_prefix("io_").set_name("pipeline_ctrl").bind(self.dut)
+        self.enable_ctrl = EnableCtrlBundle.from_prefix("io_ctrl_").set_name("enable_ctrl").bind(self.dut)
+
+        self.ftb_provider = FTBProvider()
+
+        self.enable_ctrl.btb_enable.value = 1
+
+        ##### reset #####
+        self.dut.reset.value = 1
+        self.dut.Step(10)
+        self.dut.reset.value = 0
+        self.dut.Step(1000)
+        print("$$$ FINISH RESET !")
+
+        self.pipeline_ctrl.s0_fire_0.value = 1
+        self.pipeline_ctrl.s1_fire_0.value = 1
+        self.pipeline_ctrl.s2_fire_0.value = 1
+        self.pipeline_ctrl.s3_fire_0.value = 1
+
+    def predict(self, s0_pc, do_assert, do_hit):
+        # s0
+        self.dut.io_update_valid.value = 0
+        self.dut.io_in_bits_s0_pc_0.value = s0_pc
+        self.dut.Step(1)
+        # s1
+        std_ftb_entry = self.ftb_provider.provide_ftb_entry(1, s0_pc)
+        std_full_pred = {}
+        if std_ftb_entry is not None:
+            std_ftb_entry.put_to_full_pred_dict(s0_pc, std_full_pred)
+        self.dut.Step(1)
+        # s2
+        a = self.dut.io_in_bits_resp_in_0_s2_full_pred_0_br_taken_mask_0.value = random.randint(0, 1)
+        b = self.dut.io_in_bits_resp_in_0_s2_full_pred_0_br_taken_mask_1.value = random.randint(0, 1)
+        if std_ftb_entry is not None:
+            s2_mask_0 = std_full_pred["br_taken_mask_0"] or a
+            s2_mask_1 = std_full_pred["br_taken_mask_1"] or b
+        self.dut.Step(1)
+        if std_ftb_entry is not None and do_hit == 1:
+            assert (self.dut.io_out_s2_full_pred_0_hit.value == 1)
+        else:
+            assert (self.dut.io_out_s2_full_pred_0_hit.value == 0)
+        if do_assert == 1:
+            assert (s2_mask_0 == self.dut.io_out_s2_full_pred_0_br_taken_mask_0.value)
+            assert (s2_mask_1 == self.dut.io_out_s2_full_pred_0_br_taken_mask_1.value)
+            assert (std_full_pred["fallThroughAddr"] == self.dut.io_out_s2_full_pred_0_fallThroughAddr.value)
+
+        # s3
+        a = self.dut.io_in_bits_resp_in_0_s3_full_pred_0_br_taken_mask_0.value = random.randint(0, 1)
+        b = self.dut.io_in_bits_resp_in_0_s3_full_pred_0_br_taken_mask_1.value = random.randint(0, 1)
+        if std_ftb_entry is not None:
+            s3_mask_0 = std_full_pred["br_taken_mask_0"] or a
+            s3_mask_1 = std_full_pred["br_taken_mask_1"] or b
+        self.dut.Step(1)
+        if std_ftb_entry is not None and do_hit == 1:
+            assert (self.dut.io_out_s3_full_pred_0_hit.value == 1)
+        else:
+            assert (self.dut.io_out_s3_full_pred_0_hit.value == 0)
+        if do_assert == 1:
+            assert (s3_mask_0 == self.dut.io_out_s3_full_pred_0_br_taken_mask_0.value)
+            assert (s3_mask_1 == self.dut.io_out_s3_full_pred_0_br_taken_mask_1.value)
+            assert (std_full_pred["fallThroughErr"] == self.dut.io_out_s3_full_pred_0_fallThroughErr.value)
+            assert (std_full_pred["fallThroughAddr"] == self.dut.io_out_s3_full_pred_0_fallThroughAddr.value)
+
+    def update(self, update_pc, meta_hit, update_valid, old_entry, entry_valid):
+        """
+        放入provider
+        """
+
+        self.dut.io_update_valid.value = update_valid
+        self.dut.io_update_bits_pc.value = update_pc
+        self.dut.io_update_bits_ftb_entry_valid.value = entry_valid
+        self.dut.io_update_bits_old_entry.value = old_entry
+        self.dut.io_update_bits_meta.value = meta_hit
+        a = self.dut.io_update_bits_ftb_entry_always_taken_0.value = random.randint(0, 1)
+        b = self.dut.io_update_bits_ftb_entry_always_taken_1.value = random.randint(0, 1)
+        h = self.dut.io_update_bits_ftb_entry_brSlots_0_lower.value = random.randint(0, 2 ** 12 - 1)
+        self.dut.io_update_bits_ftb_entry_brSlots_0_offset.value = 0
+        self.dut.io_update_bits_ftb_entry_brSlots_0_sharing.value = 0
+        e = self.dut.io_update_bits_ftb_entry_brSlots_0_tarStat.value = random.randint(0, 2)
+        self.dut.io_update_bits_ftb_entry_brSlots_0_valid.value = 1
+        g = self.dut.io_update_bits_ftb_entry_tailSlot_sharing.value = random.randint(0, 1)
+        h = self.dut.io_update_bits_ftb_entry_tailSlot_lower.value = random.randint(0,
+                                                                                    2 ** 20 - 1) if g else random.randint(
+            0, 2 ** 12 - 1)
+        self.dut.io_update_bits_ftb_entry_tailSlot_offset.value = 0
+        f = self.dut.io_update_bits_ftb_entry_tailSlot_tarStat.value = random.randint(0, 2)
+        self.dut.io_update_bits_ftb_entry_tailSlot_valid.value = 1
+        self.dut.io_update_bits_ftb_entry_pftAddr.value = 0
+        d = self.dut.io_update_bits_ftb_entry_carry.value = random.randint(0, 1)
+
+        update_request = {}
+        update_request["ftb_entry"] = {}
+        update_request["valid"] = update_valid
+        update_request["bits_pc"] = update_pc
+        update_request["ftb_entry"]["valid"] = entry_valid
+        update_request["ftb_entry"]["brSlots_0_offset"] = 0
+        update_request["ftb_entry"]["brSlots_0_lower"] = h
+        update_request["ftb_entry"]["brSlots_0_tarStat"] = e
+        update_request["ftb_entry"]["brSlots_0_valid"] = 1
+        update_request["ftb_entry"]["tailSlot_offset"] = 0
+        update_request["ftb_entry"]["tailSlot_lower"] = h
+        update_request["ftb_entry"]["tailSlot_tarStat"] = f
+        update_request["ftb_entry"]["tailSlot_sharing"] = g
+        update_request["ftb_entry"]["tailSlot_valid"] = 1
+        update_request["ftb_entry"]["pftAddr"] = 0  #
+        update_request["ftb_entry"]["carry"] = d
+        update_request["ftb_entry"]["isCall"] = 0
+        update_request["ftb_entry"]["isRet"] = 0
+        update_request["ftb_entry"]["isJalr"] = 0
+        update_request["ftb_entry"]["last_may_be_rvi_call"] = 0
+        update_request["ftb_entry"]["always_taken_0"] = a
+        update_request["ftb_entry"]["always_taken_1"] = b
+        update_request["bits_old_entry"] = old_entry
+        update_request["bits_meta"] = meta_hit
+
+        self.ftb_provider.update(update_request)

tests/FTB/src/ftb_utils.py

@@ -1,8 +1,24 @@
-from .config import *
+INST_OFFSET_BITS = 1
+PREDICT_WIDTH_OFFSET_BITS = 4
+
+PREDICT_WIDTH_BYTES = 32
+
+UFTB_WAYS_NUM = 32
+UFTB_TAG_SIZE = 16
+
+FTB_WAYS_NUM = 4
+FTB_TAG_SIZE = 20
+INDEX_BITS = 9
+
+TAR_OVF = 1
+TAR_UDF = 2
+TAR_FIT = 0
+
 
 def get_slot_offset(pc, target):
     return ((target - pc) >> INST_OFFSET_BITS) & ((1 << PREDICT_WIDTH_OFFSET_BITS) - 1)
 
+
 def get_slot_addr(pc, offset):
     return pc + (offset << INST_OFFSET_BITS)
 
@@ -10,20 +26,22 @@ def get_slot_addr(pc, offset):
 def get_pftaddr(target):
     return (target >> INST_OFFSET_BITS) & ((1 << PREDICT_WIDTH_OFFSET_BITS) - 1)
 
+
 def get_pftaddr_carry(pc, target):
     pc_higher = pc >> (INST_OFFSET_BITS + PREDICT_WIDTH_OFFSET_BITS)
     target_higher = target >> (INST_OFFSET_BITS + PREDICT_WIDTH_OFFSET_BITS)
     return (target_higher - pc_higher) & 1
 
+
 def get_fallthrough_addr(pc, part_addr, carry):
     higher = (pc >> (INST_OFFSET_BITS + PREDICT_WIDTH_OFFSET_BITS)) + carry
     return (higher << (INST_OFFSET_BITS + PREDICT_WIDTH_OFFSET_BITS)) | (part_addr << INST_OFFSET_BITS)
 
 
-
 def get_lower_addr(pc, bits):
     return (pc >> INST_OFFSET_BITS) & ((1 << bits) - 1)
 
+
 def get_target_stat(pc_higher, target_higher):
     if target_higher < pc_higher:
         return TAR_UDF
@@ -32,6 +50,7 @@ def get_target_stat(pc_higher, target_higher):
     else:
         return TAR_FIT
 
+
 def get_target_addr(pc, target_stat, target_lower, target_lower_bits):
     target_higher = pc >> (target_lower_bits + INST_OFFSET_BITS)
     if target_stat == TAR_UDF:
@@ -54,6 +73,7 @@ def get_cfi_addr_from_full_pred_dict(pc, d):
     else:
         return None
 
+
 def get_target_from_full_pred_dict(pc, d):
     if not d["hit"]:
         return pc + PREDICT_WIDTH_BYTES
@@ -67,13 +87,22 @@ def get_target_from_full_pred_dict(pc, d):
     else:
         return d["fallThroughAddr"]
 
+
 def set_all_none_item_to_zero(d):
     for k, v in d.items():
         if v is None:
             d[k] = 0
 
+
 def parse_uftb_meta(meta):
     return {
         "pred_way": meta >> 1,
         "hit": meta & 1
     }
+
+
+def parse_ftb_meta(meta):
+    return {
+        "pred_way": meta >> 1,
+        "hit": meta & 1
+    }

tests/FTB/tests/conftest.py

@@ -0,0 +1,14 @@
+import pytest
+from mlvp.reporter import process_context, process_func_coverage
+
+
+@pytest.hookimpl(trylast=True, optionalhook=True)
+def pytest_reporter_context(context, config):
+    process_context(context, config)
+
+
+@pytest.hookimpl(tryfirst=True, hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+    outcome = yield
+    report = outcome.get_result()
+    return process_func_coverage(item, call, report)

tests/FTB/tests/test_ftb_predict.py

@@ -0,0 +1,88 @@
+import os
+
+import mlvp.funcov as fc
+import pytest
+from mlvp.reporter import set_func_coverage, set_line_coverage
+
+from UT_FTB import DUTFTB, xspcomm
+from ftb_env import ftb_wrapper
+
+g = fc.CovGroup("test_ftb_pred")  ###
+
+
+def init_ftb_pins(func_name):
+    waveform_dir = "report/FTB/fst"
+    coverage_dir = "report/FTB/cov_dat"
+    if not os.path.exists(coverage_dir):
+        os.makedirs(coverage_dir, exist_ok=True)
+    if not os.path.exists(waveform_dir):
+        os.makedirs(waveform_dir, exist_ok=True)
+    dut = DUTFTB(
+        ["+verilator+seed+19198107"],
+        waveform_filename=os.path.join(waveform_dir, f"FTB-{func_name}.vcd"),
+        coverage_filename=os.path.join(coverage_dir, f"FTB-{func_name}.dat"),
+    )
+    dut.InitClock("clock")
+    dut.Step()
+    while dut.io_s1_ready.value == 0:
+        dut.Step()
+    return ftb_wrapper.FTBWrapper(dut)
+
+
+def finish_ftb_pins(request, group, func_name):
+    set_func_coverage(request, group)
+    coverage_file = f"report/FTB/cov_dat/FTB-{func_name}.dat"
+    if not os.path.exists(coverage_file):
+        raise FileNotFoundError(f"File not found: {coverage_file}")
+    set_line_coverage(request, coverage_file)
+
+
+@pytest.fixture()
+def ftb_fixture(request):
+    func_name = request.node.name
+    ftb_pins = init_ftb_pins(func_name)
+    yield ftb_pins
+    ftb_pins.dut.Finish()
+    finish_ftb_pins(request, [g], func_name)
+    g.clear()
+
+
+def test_ftb_pred_result(ftb_fixture):
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_hit, {"s2_hit": fc.Eq(1), "s2_not_hit": fc.Eq(0)},
+                      name="s2_hit")
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_br_taken_mask_0,
+                      {"s2_mask_0_is_1": fc.Eq(1), "s2_mask_0_is_0": fc.Eq(0)}, name="s2_mask_0")
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_br_taken_mask_1,
+                      {"s2_mask_1_is_1": fc.Eq(1), "s2_mask_1_is_0": fc.Eq(0)}, name="s2_mask_1")
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_targets_0, {"s2_targets_0": fc.Ne(0), },
+                      name="s2_targets_0")
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_targets_1, {"s2_targets_1": fc.Ne(0), },
+                      name="s2_targets_1")
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_jalr_target, {"s2_jalr_target": fc.Ne(0), },
+                      name="s2_jalr_target")
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_fallThroughAddr, {"s2_fallThroughAddr": fc.Ne(0), },
+                      name="s2_fallThroughAddr")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_hit, {"s3_hit": fc.Eq(1), "s3_not_hit": fc.Eq(0)},
+                      name="s3_hit")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_br_taken_mask_0,
+                      {"s3_mask_0_is_1": fc.Eq(1), "s3_mask_0_is_0": fc.Eq(0)}, name="s3_mask_0")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_br_taken_mask_1,
+                      {"s3_mask_1_is_1": fc.Eq(1), "s3_mask_1_is_0": fc.Eq(0)}, name="s3_mask_1")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_targets_0, {"s3_targets_0": fc.Ne(0), },
+                      name="s3_targets_0")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_targets_1, {"s3_targets_1": fc.Ne(0), },
+                      name="s3_targets_1")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_jalr_target, {"s3_jalr_target": fc.Ne(0), },
+                      name="s3_jalr_target")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_fallThroughAddr, {"s3_fallThroughAddr": fc.Ne(0), },
+                      name="s3_fallThroughAddr")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_fallThroughErr,
+                      {"s3_fallThroughErr": fc.Eq(1), "s3_not_fallThroughErr": fc.Eq(0)}, name="s3_fallThroughErr")
+    g.add_watch_point(ftb_fixture.dut.io_out_last_stage_meta, {"meta=0": fc.Eq(0), "meta=1": fc.Eq(1)},
+                      name="last_stage_meta")
+    ftb_fixture.dut.xclock.StepRis(lambda _: g.sample())
+    for i in range(1000):
+        ftb_fixture.update(0x80000000 + i * 32, 0, 1, 0, 1)
+        ftb_fixture.dut.Step(10)
+        ftb_fixture.predict(0x80000000 + i * 32, 1, 1)
+        ftb_fixture.dut.Step(10)

tests/FTB/tests/test_ftb_update.py

@@ -0,0 +1,131 @@
+import os
+import mlvp
+import random
+from datetime import datetime
+
+import mlvp.funcov as fc
+import pytest
+from mlvp.reporter import set_func_coverage, set_line_coverage
+
+from UT_FTB import DUTFTB
+from ftb_env import ftb_wrapper
+
+g = fc.CovGroup("test_ftb_update")  ###
+
+
+def init_ftb_pins(func_name):
+    waveform_dir = "report/FTB/fst"
+    coverage_dir = "report/FTB/cov_dat"
+    if not os.path.exists(coverage_dir):
+        os.makedirs(coverage_dir, exist_ok=True)
+    if not os.path.exists(waveform_dir):
+        os.makedirs(waveform_dir, exist_ok=True)
+
+    random.seed(datetime.now().timestamp() * 100000)
+    v = 1919810 + random.randint(7, 114514)
+    dut = DUTFTB(
+        [f"+verilator+seed+{v}"],
+        waveform_filename=os.path.join(waveform_dir, f"FTB-{func_name}.fst"),
+        coverage_filename=os.path.join(coverage_dir, f"FTB-{func_name}.dat"),
+    )
+    dut.InitClock("clock")
+    dut.Step()
+    while dut.io_s1_ready.value == 0:
+        dut.Step()
+    return ftb_wrapper.FTBWrapper(dut)
+
+
+def finish_ftb_pins(request, group, func_name):
+    set_func_coverage(request, group)
+    coverage_file = f"report/FTB/cov_dat/FTB-{func_name}.dat"
+    if not os.path.exists(coverage_file):
+        raise FileNotFoundError(f"File not found: {coverage_file}")
+    set_line_coverage(request, coverage_file)
+
+
+@pytest.fixture()
+def ftb_fixture(request):
+    func_name = request.node.name
+    ftb_pins = init_ftb_pins(func_name)
+    yield ftb_pins
+    ftb_pins.dut.Finish()
+    finish_ftb_pins(request, [g], func_name)
+    g.clear()
+
+
+def test_ftb_meta_hit_is_1(ftb_fixture):
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_hit, {"s2_hit=1": fc.Eq(1)}, name="meta_hit_1_s2")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_hit, {"s3_hit=1": fc.Eq(1)}, name="meta_hit_1_s3")
+    ftb_fixture.dut.xclock.StepRis(lambda _: g.sample())
+    for i in range(1000):
+        ftb_fixture.update(0x80000000 + i * 32, 1, 1, 0, 1)
+        ftb_fixture.dut.Step(1)
+        ftb_fixture.predict(0x80000000 + i * 32, 0, 1)
+        ftb_fixture.dut.Step(10)
+
+
+def test_ftb_meta_hit_is_0(ftb_fixture):
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_hit, {"s2_hit=1": fc.Eq(1)}, name="meta_hit_0_s2")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_hit, {"s3_hit=1": fc.Eq(1)}, name="meta_hit_0_s3")
+    ftb_fixture.dut.xclock.StepRis(lambda _: g.sample())
+    for i in range(1000):
+        ftb_fixture.update(0x80000000 + i * 32, 0, 1, 0, 1)
+        ftb_fixture.dut.Step(3)  ##
+        ftb_fixture.predict(0x80000000 + i * 32, 0, 1)
+        ftb_fixture.dut.Step(10)
+
+
+def test_ftb_update_queue(ftb_fixture):
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_hit, {"update_queue_s2": fc.Eq(1)}, name="update_queue_s2")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_hit, {"update_queue_s3": fc.Eq(1)}, name="update_queue_s3")
+    ftb_fixture.dut.xclock.StepRis(lambda _: g.sample())
+    for i in range(100):
+        ftb_fixture.update(0x80000000 + i * 32, 0, 1, 0, 1)
+        ftb_fixture.dut.Step(1)
+    for i in range(100):
+        ftb_fixture.predict(0x80000000 + i * 32, 0, 1)
+        ftb_fixture.dut.Step(10)
+
+
+def test_ftb_update_valid(ftb_fixture):
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_hit, {"update_valid_s2": fc.Eq(0)}, name="update_valid_s2")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_hit, {"update_valid_s3": fc.Eq(0)}, name="update_valid_s3")
+    ftb_fixture.dut.xclock.StepRis(lambda _: g.sample())
+    for i in range(1000):
+        ftb_fixture.update(0x80000000 + i * 32, 0, 0, 0, 1)
+        ftb_fixture.dut.Step(10)
+        ftb_fixture.predict(0x80000000 + i * 32, 0, 0)
+        ftb_fixture.dut.Step(10)
+
+
+def test_ftb_old_entry(ftb_fixture):
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_hit, {"old_entry_s2": fc.Eq(0)}, name="old_entry_s2")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_hit, {"old_entry_s3": fc.Eq(0)}, name="old_entry_s3")
+    ftb_fixture.dut.xclock.StepRis(lambda _: g.sample())
+    for i in range(1000):
+        ftb_fixture.update(0x80000000 + i * 32, 0, 1, 1, 1)
+        ftb_fixture.dut.Step(10)
+        ftb_fixture.predict(0x80000000 + i * 32, 0, 0)
+        ftb_fixture.dut.Step(10)
+
+
+def test_ftb_entry_valid(ftb_fixture):
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_hit, {"entry_valid_s2": fc.Eq(0)}, name="entry_valid_s2")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_hit, {"entry_valid_s3": fc.Eq(0)}, name="entry_valid_s3")
+    ftb_fixture.dut.xclock.StepRis(lambda _: g.sample())
+    for i in range(1000):
+        ftb_fixture.update(0x80000000 + i * 32, 0, 1, 0, 0)
+        ftb_fixture.dut.Step(10)
+        ftb_fixture.predict(0x80000000 + i * 32, 0, 0)
+        ftb_fixture.dut.Step(10)
+
+
+def test_ftb_cover(ftb_fixture):
+    g.add_watch_point(ftb_fixture.dut.io_out_s2_full_pred_0_hit, {"cover_s2": fc.Eq(0)}, name="cover_s2")
+    g.add_watch_point(ftb_fixture.dut.io_out_s3_full_pred_0_hit, {"cover_s3": fc.Eq(0)}, name="cover_s3")
+    ftb_fixture.dut.xclock.StepRis(lambda _: g.sample())
+    for i in range(1000):
+        ftb_fixture.update(0x80000000 + i * 0b10, 0, 1, 0, 1)
+        ftb_fixture.dut.Step(10)
+        ftb_fixture.predict(0x80000000 + i * 0b10, 0, 1)
+        ftb_fixture.dut.Step(10)

tests/Makefile

@@ -1,9 +1,18 @@
 TEST=.
 TEST_FOLDER=./$(TEST)
+
 PYTHON=python3
 
+START_CODE="from mlvp.reporter import *;\
+set_meta_info('test_case', '$(TEST)');\
+report = 'report/$(TEST)/report.html';\
+generate_pytest_report(report, args=['-s', '-n=auto', '$(TEST_FOLDER)'], );\
+"
+
 run:
-	pytest --mlvp-report -n=auto -sv $(TEST_FOLDER)
+	@echo "Running test $(TEST)..."
+	@mkdir report/$(TEST) -p
+	@$(PYTHON) -c $(START_CODE)
 
 clean:
-	rm -rf reports/
+	rm -rf report/ *.fst *.dat *.log *.hier

tests/conftest.py

@@ -0,0 +1,14 @@
+import pytest
+from mlvp.reporter import process_context, process_func_coverage
+
+
+@pytest.hookimpl(trylast=True, optionalhook=True)
+def pytest_reporter_context(context, config):
+    process_context(context, config)
+
+
+@pytest.hookimpl(tryfirst=True, hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+    outcome = yield
+    report = outcome.get_result()
+    return process_func_coverage(item, call, report)

tests/pytest.ini

@@ -0,0 +1,20 @@
+[pytest]
+pythonpath = 
+  ../utils
+  ./FTB/src
+  ../out/picker_out_FTB
+  ./uFTB/src
+  ../out/picker_out_uFTB
+  ./TAGE-SC/src
+  ../out/picker_out_TageSC
+  ./ITTAGE/src
+  ../out/picker_out_ITTAGE
+  ./RAS/src
+  ../out/picker_out_RAS
+
+testpaths = 
+  ./FTB/tests
+  ./uFTB/tests
+  ./TAGE-SC/tests
+  ./ITTAGE/tests
+  ./RAS/tests

tests/uFTB-raw/FTBEntry.py

@@ -1,63 +0,0 @@
-from types import SimpleNamespace
-import random
-
-
-class FTBEntry:
-
-    
-    class BrSlot:
-        valid=False
-        offset=0
-        target=0
-        targetCoA=0
-        alwaysTaken=False
-
-    class TailSlot:
-        valid=False
-        offset=0
-        target=0
-        targetCoA=0
-        alwaysTaken=False
-        is_br_sharing=False
-
-    def __init__(self):
-        self.valid = False
-        self.brSlot = self.BrSlot()
-        self.tailSlot = self.TailSlot()
-        
-        self.pftAddr = 0
-        self.carry = False
-        self.isCall = False
-        self.isRet = False
-        self.isJalr = False
-        self.rviCall = False
-
-        self.brSlot.valid = False
-        self.tailSlot.valid = False
-
-
-def gen_ftb_entry(pc, br_slotv, tail_slotv) -> FTBEntry:
-    entry = FTBEntry()
-    entry.valid = True
-    if br_slotv:
-        entry.brSlot.valid = True
-        entry.brSlot.target = random.randint(0, 2**20)
-        entry.brSlot.offset = random.randint(0, 2**12)
-        entry.brSlot.alwaysTaken = random.choice([True, False])
-    if tail_slotv:
-        entry.tailSlot.valid = True
-        entry.tailSlot.target = random.randint(0, 2**20)
-        entry.tailSlot.offset = random.randint(0, 2**12)
-        entry.tailSlot.is_br_sharing = random.choice([True, False])
-    entry.pftAddr = pc + random.randint(0, 2**12)
-    if tail_slotv and entry.tailSlot.is_br_sharing == False:
-        flag = random.randint(0, 3)
-        if flag == 0:
-            entry.isCall = True
-        elif flag == 1:
-            entry.isRet = True
-        elif flag == 2:
-            entry.isJalr = True
-        else:
-            entry.rviCall = True
-    return entry

tests/uFTB-raw/FauFTB.py

@@ -1,203 +0,0 @@
-import os
-ROOT_PATH = os.path.dirname(os.path.abspath(__file__)) + "/../.."
-DUT_PATH = ROOT_PATH + "/out/picker_out_uFTB"
-os.sys.path.append(DUT_PATH)
-
-from typing import Any, Tuple
-from UT_FauFTB import *
-from FTBEntry import *
-from collections import namedtuple
-
-
-class FauFTB(DUTFauFTB):
-
-    class Io_in_bits_s0_pc:
-
-        def __init__(self, outer_instance: "FauFTB"):
-            self.outer_instance = outer_instance
-            for i in range(4):
-                setattr(self, f"_{i}", getattr(outer_instance, f"io_in_bits_s0_pc_{i}"))
-
-        def set(self, arg):
-            for i in range(4):
-                getattr(self, f"_{i}").value = arg
-
-    class Io_out_s1_pc:
-
-        def __init__(self, outer_instance: "FauFTB"):
-            self.outer_instance = outer_instance
-            for i in range(4):
-                setattr(self, f"_{i}", getattr(outer_instance, f"io_out_s1_pc_{i}"))
-
-        def get(self):
-            assert all(self._0.value == getattr(self, f"_{i}").value for i in range(4))
-            return self._0
-
-    class Io_s0_fire:
-
-        def __init__(self, outer_instance: "FauFTB"):
-            self.outer_instance = outer_instance
-            for i in range(4):
-                setattr(self, f"_{i}", getattr(outer_instance, f"io_s0_fire_{i}"))
-
-        def set(self,arg):
-            for i in range(4):
-                getattr(self, f"_{i}").value = arg
-
-    class Io_out_s1_full_pred:
-
-        def __init__(self, outer_instance: "FauFTB"):
-            self.outer_instance = outer_instance
-            self.sig_array = [
-                "br_taken_mask_0",
-                "br_taken_mask_1",
-                "slot_valids_0",
-                "slot_valids_1",
-                "targets_0",
-                "targets_1",
-                "offsets_0",
-                "offsets_1",
-                "fallThroughAddr",
-                "is_br_sharing",
-                "hit",
-            ]
-            for i in range(4):
-                setattr(
-                    self,
-                    f"_{i}",
-                    namedtuple(f"_{i}", self.sig_array)(
-                        *[
-                            getattr(outer_instance, f"io_out_s1_full_pred_{i}_{sig}")
-                            for sig in self.sig_array
-                        ]
-                    ),
-                )
-
-        def get(self):
-            for sig in self.sig_array:
-                assert all(
-                    getattr(self, f"_{i}")._asdict()[sig].value
-                    == getattr(self, f"_{i}")._asdict()[sig].value
-                    for i in range(4)
-                )
-            return self._0
-
-    class Io_out_last_stage_meta:
-
-        def __init__(self, outer_instance: "FauFTB"):
-            self.outer_instance = outer_instance
-
-            self.resp_meta_hit_r_1 = xsp.XPin(
-                outer_instance.io_out_last_stage_meta.xdata.SubDataRef(
-                    0,
-                    0,
-                    "resp_meta_hit_r_1"
-                ),
-                outer_instance.event,
-            )
-
-            self.resp_meta_pred_way_r_1 = xsp.XPin(
-                outer_instance.io_out_last_stage_meta.xdata.SubDataRef(
-                    1,
-                    5,
-                    "resp_meta_pred_way_r_1"
-                ),
-                outer_instance.event,
-            )
-
-            pass
-
-        pass
-
-    class Io_update_bits_ftb_entry:
-
-        def __init__(self, outer_instance: "FauFTB"):
-            self.outer_instance = outer_instance
-            self.sig_array = [
-                "brSlots_0_valid",
-                "brSlots_0_offset",
-                "brSlots_0_lower",
-                "brSlots_0_tarStat",
-                "always_taken_0",
-                "tailSlot_valid",
-                "tailSlot_offset",
-                "tailSlot_lower",
-                "tailSlot_tarStat",
-                "tailSlot_sharing",
-                "always_taken_1",
-                "pftAddr",
-                "carry",
-            ]
-            namedtuple(
-                "Io_update_bits_ftb_entry",
-                self.sig_array,
-            )(
-                *[
-                    setattr(
-                        self,
-                        sig,
-                        getattr(outer_instance, f"io_update_bits_ftb_entry_{sig}"),
-                    )
-                    for sig in self.sig_array
-                ]
-            )
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.io_in_bits_s0_pc = self.Io_in_bits_s0_pc(self)
-        self.io_out_s1_pc = self.Io_out_s1_pc(self)
-        self.io_s0_fire = self.Io_s0_fire(self)
-        self.io_out_s1_full_pred = self.Io_out_s1_full_pred(self)
-        self._io_out_last_stage_meta = self.Io_out_last_stage_meta(self)
-        self.io_update_bits_ftb_entry = self.Io_update_bits_ftb_entry(self)
-
-    def check_dup_equation(self, *inputs):
-        return all(i == inputs[0] for i in inputs)
-
-    def s1_full_pred(self) -> Tuple[int, FTBEntry, int, int]:
-        io_out_s1_full_pred = self.io_out_s1_full_pred.get()
-
-        entry = FTBEntry()
-
-        # FauFTB
-        entry.valid = io_out_s1_full_pred.hit.value
-        entry.pftAddr = io_out_s1_full_pred.fallThroughAddr.value
-        # Br slot
-        entry.brSlot.valid = io_out_s1_full_pred.slot_valids_0.value
-        entry.brSlot.offset = io_out_s1_full_pred.offsets_0.value
-        entry.brSlot.target = io_out_s1_full_pred.targets_0.value
-        entry.brSlot.targetCoA = None
-        entry.brSlot.alwaysTaken = io_out_s1_full_pred.br_taken_mask_0.value
-        # Tail slot
-        entry.tailSlot.valid = io_out_s1_full_pred.slot_valids_1.value
-        entry.tailSlot.offset = io_out_s1_full_pred.offsets_1.value
-        entry.tailSlot.target = io_out_s1_full_pred.targets_1.value
-        entry.tailSlot.targetCoA = None
-        entry.tailSlot.alwaysTaken = io_out_s1_full_pred.br_taken_mask_1.value
-        entry.tailSlot.is_br_sharing = io_out_s1_full_pred.is_br_sharing.value
-
-        return (self.io_out_s1_pc.get().value, entry, io_out_s1_full_pred.br_taken_mask_0.value, io_out_s1_full_pred.br_taken_mask_1.value)
-
-    def update_ftb_entry(self, update_pc: int, entry: FTBEntry, taken: Tuple[int, int]):
-
-        self.io_update_bits_pc.value = update_pc
-        self.io_update_bits_br_taken_mask_0.value = taken[0]
-        self.io_update_bits_br_taken_mask_1.value = taken[1]
-
-        update = self.io_update_bits_ftb_entry
-
-        update.brSlots_0_valid.value = entry.brSlot.valid
-        update.brSlots_0_offset.value = entry.brSlot.offset
-        update.brSlots_0_lower.value = entry.brSlot.target
-        update.brSlots_0_tarStat.value = entry.brSlot.targetCoA
-        update.always_taken_0.value = entry.brSlot.alwaysTaken
-
-        update.tailSlot_valid.value = entry.tailSlot.valid
-        update.tailSlot_offset.value = entry.tailSlot.offset
-        update.tailSlot_lower.value = entry.tailSlot.target
-        update.tailSlot_tarStat.value = entry.tailSlot.targetCoA
-        update.tailSlot_sharing.value = entry.tailSlot.is_br_sharing
-        update.always_taken_1.value = entry.tailSlot.alwaysTaken
-
-        update.pftAddr.value = entry.pftAddr
-        update.carry.value = entry.carry

tests/uFTB-raw/README.md

@@ -1,58 +0,0 @@
-# uFTB-raw
-
-## 介绍
-
-本测试用例提供了基于模拟随机数据的香山处理器 uFTB 分支预测器的仿真验证环境，用于验证 uFTB 的缓存功能和预测功能。
-
-为此，我们为 uFTB 提供了简易的 uFTB Wrapper，以向 uFTB 提供时序控制和输入输出处理。具体而言，我们提供了三种操作：
-
-1. 生成数据队列：由于uFTB实际上可以被是为FTB表项的一个缓存，所以我们将传递的信息封装为 FTBEntry，以便于 uFTB 的使用。而 FTBEntry 本身是一组受约束的数据，因此只要随机数据符合结果即可。
-2. 读取操作：uFTB 会根据传递的信息，从自身缓存的 FTB 表项中读取预测结果。因为我们的测试用例是随机生成的，所以我们不需要真实的预测结果，只需要保证 uFTB 的读取操作正确即可。
-3. 更新操作：uFTB 会根据传递的信息，更新自身缓存的 FTB 表项。只要更新生效，即可认为 uFTB 的更新操作正确。
-
-对读取和更新操作，我们将对Pin接口的操作封装为 `get_pred` 和 `set_update` 方法，以便于 uFTB 的使用。
-同时对原始的 uFTB 模型进行了封装，由于存在很多重复的接口，我们使用 python 的元编程机制，将重复的接口合并，以减少对pin接口操作时的代码量。
-
-
-## 快速使用
-
-### 环境配置
-
-**1. 安装 mlvp**
-
-具体步骤参见 https://github.com/XS-MLVP/mlvp
-
-**2. 编译 DUT**
-
-在本仓库根目录下执行
-
-```shell
-make uftb TL=python
-```
-
-即可生成 DUT 编译结果，编译结果无需移动，程序会自动检索对应目录。
-
-### 仿真验证
-
-在 `tests` 目录下执行
-
-```shell
-make TEST=uFTB-raw run
-```
-
-即可开始仿真验证。
-
-程序运行结束后，会生成对应的波形文件及覆盖率报告。波形文件位于 `tests/report/uFTB-raw` 目录下，覆盖率报告位于 `tests/report/uFTB-raw.html`。
-
-
-## 使用说明
-
-### 目录结构
-
-```bash
-uFTB-raw                    # 测试用例名称
-|-- FTBEntry.py             # 针对 FTB 表项的封装
-|-- FauFTB.py               # 针对 uFTB 的二次封装，将重复信号合并，并恢复信号结构体层次
-|-- README.md               
-`-- test_raw.py             # 测试入口，用于驱动随机数据生成、uFTB 操作
-```

tests/uFTB-raw/test_raw.py

@@ -1,62 +0,0 @@
-from typing import List, Tuple
-from FauFTB import *
-from FTBEntry import *
-
-EntryList: List[Tuple[int, FTBEntry, bool, bool]] = []
-
-
-def ftb_entry_list():
-    for i in range(10000):
-        pc = random.randint(0, 2**39)
-        gentry = gen_ftb_entry(pc, True, True)
-        EntryList.append(
-            (pc, gentry, random.choice([True, False]), random.choice([True, False]))
-        )
-        # print("ftb_entry_list", EntryList[i][1].brSlot)
-    return EntryList
-
-
-def get_pred(uFTB: FauFTB, pc: int) -> Tuple[int, FTBEntry, bool, bool]:
-    uFTB.io_in_bits_s0_pc.set(pc)
-    uFTB.io_ctrl_ubtb_enable.value = 1
-    uFTB.io_s0_fire.set(1)
-    uFTB.io_s1_fire_0.value = 1
-    uFTB.io_s2_fire_0.value = 1
-    return uFTB.s1_full_pred()
-
-
-def set_update(uFTB: FauFTB, entry: Tuple[int, FTBEntry, bool, bool]):
-    uFTB.io_update_valid.value = True
-    uFTB.io_update_bits_pc.value = entry[0]
-    # print("set_update", entry[1].brSlot)
-    uFTB.update_ftb_entry(entry[0], entry[1], (entry[2], entry[3]))
-
-
-from mlvp import *
-import mlvp.funcov as fc
-from mlvp.reporter import *
-
-
-def test_raw(mlvp_request):
-    uFTB: FauFTB = mlvp_request
-
-    ftb_entry_list()
-    uFTB.reset.value = 1
-    uFTB.Step(100)
-    uFTB.reset.value = 0
-
-    for i in range(10000):
-        # print("main1", EntryList[i - 10][1].brSlot)
-        j = i
-        pred = get_pred(uFTB, EntryList[j][0])
-        if i > 9:
-            # print("main2", EntryList[i - 10][1].brSlot)
-            set_update(uFTB, EntryList[(i - 10)%10000]) # update data 10 cycles ago
-            pass
-        print("main", pred[0], pred[1].__dict__)
-        uFTB.Step(1)
-
-import pytest
-@pytest.fixture()
-def mlvp_request(mlvp_pre_request: PreRequest):
-    return mlvp_pre_request.create_dut(FauFTB, "clock")

tests/uFTB-with-ftq/README.md

@@ -1,113 +0,0 @@
-# uFTB-env
-
-## 介绍
-
-本测试用例提供了基于真实指令流的香山处理器 uFTB 分支预测器的仿真验证环境，以及时钟精确的 uFTB 参考模型，最终可给出 uFTB 的分支预测准确率。
-
-为此，我们为 uFTB 提供了简易的 BPU Top Wrapper，以向 uFTB 提供时序控制和输入输出处理。并向 BPU Top 提供了简易的 FTQ 实现，FTQ 中实例化了一个真实的程序仿真器，用于生成真实指令流，FTQ 会处理 BPU 产生的预测结果，并向 BPU 提供更新请求与重定向请求的执行反馈。大致的结构可参考下图：
-
-![env](README.assets/env.png)
-
-$\color{red}{对于香山 BPU 中其他子预测器的验证，可复用本项目中的真实指令执行环境，但需要对 DUT 的接口、时序以及需要使用的预测结果通道等进行适配。}$
-
-## 快速使用
-
-### 环境配置
-
-**1. 安装 mlvp**
-
-具体步骤参见 https://github.com/XS-MLVP/mlvp
-
-**2. 生成 BRTParser Trace 工具**
-
-为了生成真实的指令流，BRTParser 作为一个自定义的工具已经被放置在仓库根目录下的 `utils` 文件夹中，但其中缺少了模拟器仿真程序，需要自行编译生成，具体步骤参见 https://github.com/yaozhicheng/NEMU
-
-生成编译结果 `NemuBR` 后，将其放置在 `utils/BRTParser` 目录下，工具即可正常使用。
-
-**3. 编译 DUT**
-
-在本仓库根目录下执行
-
-```shell
-make uftb TL=python
-```
-
-即可生成 DUT 编译结果，编译结果无需移动，程序会自动检索对应目录。
-
-### 仿真验证
-
-在 `tests` 目录下执行
-
-```shell
-make TEST=uFTB-with-ftq run
-```
-
-即可开始仿真验证。
-
-程序运行结束后，会打印出分支预测的统计信息。
-
-若要更改需要执行的程序，可在 `config.py` 中更改相应变量的值，仿真所需的真实程序已经放置仓库 `utils/ready-to-run` 目录下。若要更改仿真所持续的周期数，可在 `config.py` 中更改 `MAX_CYCLE` 的值。
-
-如果需要启用随机trace，可以设置环境变量 RANDOM_BPT=1
-```shell
-RANDOM_BPT=1 make TEST=uFTB-with-ftq run
-```
-
-## 使用说明
-
-### 目录结构
-
-```
-uftb-env/                   # uFTB 环境源码
-├── bpu_top.py                  # BPU Top Wrapper
-├── bundle.py                   # 定义了 DUT 相关接口
-├── config.py                   # 与 uFTB 相关的配置信息
-├── executor.py                 # 对 BRTParser 工具的封装
-├── ftb.py                      # FTB 项相关结构
-├── ftq.py                      # FTQ 实现
-├── tb.py                       # 测试用例
-├── uftb_model.py               # uFTB 参考模型
-└── utils.py                    # 相关工具函数
-```
-
-### 指令执行器
-
-`executor.py` 中定义了 `Executor` 类，用于对 BRTParser 工具的封装，提供了生成真实指令流的功能。
-
-实现中，由于 BRTParser 工具只提供了分支指令的跳转 Trace，因此普通指令的长度无法获取，为此普通指令的长度是在 `Executor` 中进行随机生成的。
-
-使用时需要用到两个主要方法：
-
-- `current_inst` 用于获取当前指令。调用时返回当前指令 PC、指令长度及分支指令信息。分支指令如果为空则表示当前指令不是分支指令，否则给出分支指令相关信息。
-- `next_inst` 用于执行当前指令。
-
-### FTQ
-
-`ftq.py` 中实现了 FTQ 的相关逻辑，指令执行器也在此被实例化，因此 FTQ 具备了获取真实指令执行情况的能力。
-
-
-FTQ 的工作流程如下：
-
-1. 在每个周期 `update` 方法被调用，用于更新 FTQ 的状态。此时，如果传入的 BPU 输出信息中，s1 通道有效，则 FTQ 会将 s1 产生的预测结果存入一个 FTQ 项中。
-2. 执行一个预测块。FTQ 检测队列中是否还有尚未执行的预测块，如果没有则跳过，如有则执行此预测块，分为两种情况。
-    - 如果预测块指示 FTB 项没有 hit，这说明预测结果无效。FTB 会调用执行器，生成一个完整的 FTB 项。
-    - 如果预测块指示 FTB 项 hit，并且预测结果中的起始 PC 与执行器当前 PC 相同，则说明本次预测有效。FTQ 会根据预测结果调用执行器，若执行过程中出现与预测结果不符的情况，则 FTQ 生成重定向请求，以供 BPU 恢复到正确状态。
-3. 生成更新和重定向请求。FTQ 会使用新生成或者更新后的 FTB 项生成更新请求，如果有预测错误还会生成重定向请求。最终，FTQ 会将更新请求和重定向请求传递给 BPU。
-
-在该 FTQ 实现中，仅仅根据 s1 通道的预测结果来更新 FTQ 队列，对于 s2, s3 通道的预测结果没有进行响应。因此，若需要验证 s2, s3 通道的预测结果，需要对 FTQ 的该部分进行相应的修改。
-
-### BPU Top Wrapper
-
-`bpu_top.py` 中实现了 BPU Top Wrapper，用于向 uFTB 提供时序控制和输入输出处理。由于 uFTB 只在 s1 阶段工作，因此 BPU Top 并没有对 s2, s3 通道的预测结果进行处理，并且在 BPU Top 中将 DUT 中的 `s2_fire` 及 `s3_fire` 端口持续置高，以获取 DUT 在 s3 阶段输出的 `meta` 信息 。如果需要验证 s2, s3 通道的预测结果，需要对 BPU Top 的该部分进行相应的修改。
-
-具体地，`BPU Top` 会维护流水线控制信息，并驱动 DUT。在每个周期，BPU 的工作流程如下：
-
-1. 更新 DUT 的流水线控制信号
-2. 获取 DUT 的预测结果并进行加工。BPU Top 会获取 DUT 的预测结果，并且对其中需要 BPU 赋值的部分进行赋值，生成 BPU 的输出信息。
-3. 获取 uFTB Model 的预测结果，并进行对比。以此来验证 uFTB 实现的正确性。
-4. 将 BPU 输出信息传递给 FTQ，获取 FTQ 的更新请求和重定向请求。
-5. 将 FTQ 的更新请求和重定向请求传递给 DUT 和 uFTB Model，并更新流水线控制信号。
-
-在本项目中还实现了一个 `FTBProvider` 用于提供基于 FTB 项的基础预测结果，如果需要验证非 FTB 项的预测结果，需要将 `ftb_provider_stage_enable` 中相应阶段开关打开，便可以在相应阶段添加 FTB 的预测结果。
-
-

tests/uFTB-with-ftq/env/bpu_top.py

@@ -1,221 +0,0 @@
-from mlvp import *
-from mlvp.triggers import *
-from .bundle import *
-from .ftq import *
-from .uftb_model import uFTBModel
-
-def assert_equal(a, b):
-    if a != b:
-        error(f"[Error] Expected is {a}, but actual is {b}")
-        exit(1)
-
-def compare_uftb_full_pred(uftb_output, std_output):
-    need_compare = ["hit", "slot_valids_0", "slot_valids_1", "targets_0", "targets_1",
-                    "offsets_0", "offsets_1", "fallThroughAddr", "is_br_sharing",
-                    "br_taken_mask_0", "br_taken_mask_1"]
-    for key in need_compare:
-        assert_equal(uftb_output[key], std_output[key])
-
-class BPUTop:
-    def __init__(self, dut, dut_out: BranchPredictionResp, dut_update: UpdateBundle, pipeline_ctrl: PipelineCtrlBundle, enable_ctrl: EnableCtrlBundle):
-        self.dut = dut
-
-        self.dut_out = dut_out
-        self.dut_update = dut_update
-        self.pipeline_ctrl = pipeline_ctrl
-        self.enable_ctrl = enable_ctrl
-
-        self.s0_fire = 0
-        self.s1_fire = 0
-        self.s2_fire = 0
-        self.s3_fire = 0
-        self.s0_pc = 0
-        self.s1_pc = 0
-        self.s2_pc = 0
-        self.s3_pc = 0
-        self.s1_hit_way = 0
-        self.s2_hit_way = 0
-        self.s3_hit_way = 0
-
-        self.ftq = FTQ()
-        self.uftb_model = uFTBModel()
-        self.ftb_provider = FTBProvider()
-
-    def pipeline_assign(self):
-        self.pipeline_ctrl.s0_fire_0.value = self.s0_fire
-        self.pipeline_ctrl.s0_fire_1.value = self.s0_fire
-        self.pipeline_ctrl.s0_fire_2.value = self.s0_fire
-        self.pipeline_ctrl.s0_fire_3.value = self.s0_fire
-
-        self.pipeline_ctrl.s1_fire_0.value = self.s1_fire
-        self.pipeline_ctrl.s1_fire_1.value = self.s1_fire
-        self.pipeline_ctrl.s1_fire_2.value = self.s1_fire
-        self.pipeline_ctrl.s1_fire_3.value = self.s1_fire
-
-        self.pipeline_ctrl.s2_fire_0.value = self.s2_fire
-        self.pipeline_ctrl.s2_fire_1.value = self.s2_fire
-        self.pipeline_ctrl.s2_fire_2.value = self.s2_fire
-        self.pipeline_ctrl.s2_fire_3.value = self.s2_fire
-
-        self.pipeline_ctrl.s3_fire_0.value = self.s3_fire
-        self.pipeline_ctrl.s3_fire_1.value = self.s3_fire
-        self.pipeline_ctrl.s3_fire_2.value = self.s3_fire
-        self.pipeline_ctrl.s3_fire_3.value = self.s3_fire
-
-        # Set the value to 1 forcibly to obtain meta information
-        self.dut.io_s1_fire_0.value = 1
-        self.dut.io_s2_fire_0.value = 1
-
-        self.dut.io_in_bits_s0_pc_0.value = self.s0_pc
-        self.dut.io_in_bits_s0_pc_1.value = self.s0_pc
-        self.dut.io_in_bits_s0_pc_2.value = self.s0_pc
-        self.dut.io_in_bits_s0_pc_3.value = self.s0_pc
-
-    def generate_bpu_output(self, dut_output):
-        dut_output["s1"]["valid"] = self.s1_fire
-        dut_output["s2"]["valid"] = self.s2_fire
-        dut_output["s3"]["valid"] = self.s3_fire
-
-        dut_output["s2"]["pc_3"] = self.s2_pc
-        dut_output["s3"]["pc_3"] = self.s3_pc
-
-        # Provide Basic FTB Prediction
-        ftb_provider_stage_enable = (False, False, False)
-
-        if self.s1_fire and ftb_provider_stage_enable[0]:
-            ftb_entry = self.ftb_provider.provide_ftb_entry(self.s1_fire, self.s1_pc)
-            if ftb_entry is not None:
-                ftb_entry.put_to_full_pred_dict(self.s1_pc, dut_output["s1"]["full_pred"])
-            else:
-                set_all_none_item_to_zero(dut_output["s1"]["full_pred"])
-
-        if self.s2_fire and ftb_provider_stage_enable[1]:
-            ftb_entry = self.ftb_provider.provide_ftb_entry(self.s2_fire, self.s2_pc)
-            if ftb_entry is not None:
-                ftb_entry.put_to_full_pred_dict(self.s2_pc, dut_output["s2"]["full_pred"])
-            else:
-                set_all_none_item_to_zero(dut_output["s2"]["full_pred"])
-
-        if self.s3_fire and ftb_provider_stage_enable[2]:
-            ftb_entry = self.ftb_provider.provide_ftb_entry(self.s3_fire, self.s3_pc)
-            if ftb_entry is not None:
-                ftb_entry.put_to_full_pred_dict(self.s3_pc, dut_output["s3"]["full_pred"])
-                dut_output["last_stage_ftb_entry"] = ftb_entry.__dict__()
-            else:
-                set_all_none_item_to_zero(dut_output["s3"]["full_pred"])
-                dut_output["last_stage_ftb_entry"] = FTBEntry().__dict__()
-
-        return dut_output
-
-    async def run(self):
-        self.enable_ctrl.ubtb_enable.value = 1
-        self.s0_pc = RESET_VECTOR
-
-        self.dut.reset.value = 1
-        await ClockCycles(self.dut, 10)
-        self.dut.reset.value = 0
-        await ClockCycles(self.dut, 10)
-
-        while True:
-            self.pipeline_assign()
-            await ClockCycles(self.dut, 1)
-
-            self.s3_fire = self.s2_fire
-            self.s2_fire = self.s1_fire
-            self.s1_fire = self.s0_fire
-            self.s3_pc = self.s2_pc
-            self.s2_pc = self.s1_pc
-            self.s1_pc = self.s0_pc
-            self.s3_hit_way = self.s2_hit_way
-            self.s2_hit_way = self.s1_hit_way
-
-            npc_gen = self.s0_pc
-            next_s0_fire = 1
-            s1_flush = False
-            s2_flush = False
-            s3_flush = False
-
-
-            # Get dut output and generate bpu output
-            dut_output = self.dut_out.as_dict()
-            bpu_output = self.generate_bpu_output(dut_output)
-
-            ftb_entry = FTBEntry.from_full_pred_dict(self.s1_pc, dut_output["s1"]["full_pred"])
-            model_output = self.uftb_model.generate_output(self.s1_fire, self.s1_pc)
-            std_ftb_entry = self.ftb_provider.provide_ftb_entry(self.s1_fire, self.s1_pc)
-
-            if model_output:
-                self.s1_hit_way = model_output[2]
-            else:
-                self.s1_hit_way = None
-
-            debug("-" * 30)
-            if self.s1_fire:
-                # Debug Imformation
-                debug("[BPU]")
-                debug(f"New prediction at {hex(self.s1_pc)}")
-                if bpu_output["s1"]["full_pred"]["hit"]:
-                    debug("Dut Hit")
-
-                debug("FTB Entry in pred result: ")
-                if bpu_output["s1"]["full_pred"]["hit"]:
-                    debug(ftb_entry.__str__(self.s1_pc))
-                else:
-                    debug("No FTB Entry")
-                debug(f"br_taken_mask: {bpu_output['s1']['full_pred']['br_taken_mask_0']}, {bpu_output['s1']['full_pred']['br_taken_mask_1']}")
-
-                debug("FTB Entry in uFTB Model: ")
-                if model_output:
-                    debug(model_output[0].__str__(self.s1_pc))
-                    debug(f"br_taken_mask: {model_output[1]}")
-                else:
-                    debug("No FTB Entry")
-
-                # Compare dut output and uFTB model output
-                expected_hit = model_output is not None
-                actual_hit = bpu_output["s1"]["full_pred"]["hit"]
-                assert_equal(expected_hit, actual_hit)
-                if parse_uftb_meta(dut_output["last_stage_meta"])["hit"] or self.s3_hit_way is not None:
-                    expected_hit_way = self.s3_hit_way
-                    actual_hit_way = parse_uftb_meta(dut_output["last_stage_meta"])["pred_way"]
-                    assert_equal(expected_hit_way, actual_hit_way)
-
-                if model_output:
-                    std_full_pred = {}
-                    model_output[0].put_to_full_pred_dict(self.s1_pc, std_full_pred)
-                    std_full_pred["br_taken_mask_0"] = model_output[1][0]
-                    std_full_pred["br_taken_mask_1"] = model_output[1][1]
-                    compare_uftb_full_pred(bpu_output["s1"]["full_pred"], std_full_pred)
-
-
-            # Forward to FTQ and get update and redirect request
-            if self.s1_fire:
-                npc_gen = get_target_from_full_pred_dict(self.s1_pc, dut_output["s1"]["full_pred"])
-            update_request, redirect_request = self.ftq.update(bpu_output, std_ftb_entry)
-
-            ## Update Request
-            if update_request:
-                self.uftb_model.update(update_request)
-                self.ftb_provider.update(update_request)
-                self.dut_update.assign(update_request)
-                self.dut_update.valid.value = 1
-            else:
-                self.dut_update.valid.value = 0
-
-            ## Redirect Request
-            if redirect_request:
-                next_s0_fire = 1
-                s1_flush = True
-                s2_flush = True
-                s3_flush = True
-                npc_gen = redirect_request["cfiUpdate"]["target"]
-
-            # Add new control information
-            self.s0_fire = next_s0_fire
-            self.s0_pc = npc_gen
-            if s1_flush:
-                self.s1_fire = 0
-            if s2_flush:
-                self.s2_fire = 0
-            if s3_flush:
-                self.s3_fire = 0

tests/uFTB-with-ftq/env/bundle.py

@@ -1,53 +0,0 @@
-from mlvp import Bundle
-
-class PipelineCtrlBundle(Bundle):
-    signals = ["s0_fire_0", "s0_fire_1", "s0_fire_2", "s0_fire_3",
-               "s1_fire_0", "s1_fire_1", "s1_fire_2", "s1_fire_3",
-               "s2_fire_0", "s2_fire_1", "s2_fire_2", "s2_fire_3",
-               "s3_fire_0", "s3_fire_1", "s3_fire_2", "s3_fire_3",
-               "s1_ready", "s2_ready", "s3_ready",
-               "s2_redirect", "s3_redirect"]
-
-class EnableCtrlBundle(Bundle):
-    signals = ["ubtb_enable", "btb_enable", "bim_enable", "tage_enable",
-               "sc_enable", "ras_enable", "loop_enable"]
-
-
-class FTBEntryBundle(Bundle):
-    signals = ["brSlots_0_offset", "brSlots_0_lower", "brSlots_0_tarStat", "brSlots_0_valid",
-               "tailSlot_offset", "tailSlot_lower", "tailSlot_tarStat", "tailSlot_sharing", "tailSlot_valid",
-               "pftAddr", "carry", "isCall", "isRet", "isJalr", "last_may_be_rvi_call",
-               "always_taken_0", "always_taken_1"]
-
-class UpdateBundle(Bundle):
-    signals = ["valid", "bits_pc", "bits_br_taken_mask_0", "bits_br_taken_mask_1"]
-
-    def __init__(self):
-        super().__init__()
-        self.ftb_entry = FTBEntryBundle.from_prefix("bits_ftb_entry_")
-
-class FullBranchPredirectionBundle(Bundle):
-    signals = ["hit", "slot_valids_0", "slot_valids_1", "targets_0", "targets_1",
-               "offsets_0", "offsets_1", "fallThroughAddr", "fallThroughErr",
-               "is_jal", "is_jalr", "is_call", "is_ret", "is_br_sharing",
-               "last_may_be_rvi_call",
-               "br_taken_mask_0", "br_taken_mask_1",
-               "jalr_target"]
-
-class BranchPredictionBundle(Bundle):
-    signals = ["pc_3", "valid", "hasRedirect", "ftq_idx"]
-
-    def __init__(self):
-        super().__init__()
-        self.full_pred = FullBranchPredirectionBundle.from_regex(r"full_pred_\d_(.*)")
-
-
-class BranchPredictionResp(Bundle):
-    signals = ["last_stage_meta"]
-
-    def __init__(self):
-        super().__init__()
-        self.s1 = BranchPredictionBundle.from_prefix("s1_")
-        self.s2 = BranchPredictionBundle.from_prefix("s2_")
-        self.s3 = BranchPredictionBundle.from_prefix("s3_")
-        self.last_stage_ftb_entry = FTBEntryBundle.from_prefix("last_stage_ftb_entry_")

tests/uFTB-with-ftq/env/config.py

@@ -1,26 +0,0 @@
-PROGRAM_NAME = "microbench.bin"
-MAX_CYCLE = 10000
-
-import os
-ROOT_PATH = os.path.dirname(os.path.abspath(__file__)) + "/../../.."
-DUT_PATH = ROOT_PATH + "/out/picker_out_uFTB"
-UTILS_PATH = ROOT_PATH + "/utils"
-RROGRAM_FORDER_PATH = UTILS_PATH + "/ready-to-run"
-PROGRAM_PATH = RROGRAM_FORDER_PATH + "/" + PROGRAM_NAME
-
-
-INST_OFFSET_BITS = 1
-PREDICT_WIDTH_OFFSET_BITS = 4
-
-PREDICT_WIDTH_BYTES = 32
-RESET_VECTOR = 0x80000000
-
-UFTB_WAYS_NUM = 32
-UFTB_TAG_SIZE = 16
-
-
-TAR_OVF = 1
-TAR_UDF = 2
-TAR_FIT = 0
-
-

tests/uFTB-with-ftq/env/executor.py

@@ -1,106 +0,0 @@
-from .config import *
-
-import os
-os.sys.path.append(UTILS_PATH)
-
-from BRTParser import BRTParser, RandomBPTTrace
-
-class Executor:
-    """Get program real execution instruction flow."""
-
-    def __init__(self, filename, reset_vector=0x80000000):
-        if str(os.getenv("RANDOM_BPT")).lower() in ["1", "true"]:
-            self._executor = RandomBPTTrace().gen(start_address=reset_vector, pc_range_size=100_000)
-        else:
-            self._executor = BRTParser().fetch(filename)
-        self._current_branch = next(self._executor)
-        self._current_pc = reset_vector
-
-        self._last_exec_result = {
-            "pc": 0,
-            "inst_len": 0,
-            "branch": 0
-        }
-
-        self._exec_once()
-
-    def current_inst(self):
-        """Return current instruction information."""
-        return self._last_exec_result["pc"], self._last_exec_result["inst_len"], self._last_exec_result["branch"]
-
-    def next_inst(self):
-        """Move to next instruction."""
-        self._exec_once()
-
-    def _exec_once(self):
-        # print(f"- Executor: pc: {hex(self._last_exec_result['pc'])}, inst_len: {self._last_exec_result['inst_len']},\
-                # branch: {self._last_exec_result['branch']}")
-
-        self._last_exec_result["pc"] = self._current_pc
-
-        inst_len, branch = 0, None
-        if (2 <= self._current_branch["pc"] - self._current_pc <= 4):
-            inst_len = self._current_branch["pc"] - self._current_pc
-            self._current_pc = self._current_branch["pc"]
-
-        elif (self._current_branch["pc"] <= self._current_pc):
-            # When current pc < next branch pc, force to execute next branch
-            inst_len = Executor.branch_inst_len(self._current_branch)
-            self._current_pc = self._current_branch["target"] if self._current_branch["taken"] \
-                else self._current_pc + inst_len
-
-            branch = self._current_branch
-            self._current_branch = next(self._executor)
-
-        else:
-            inst_len = Executor.random_inst_len(self._current_pc)
-            self._current_pc += Executor.random_inst_len(self._current_pc)
-
-        self._last_exec_result["inst_len"] = inst_len
-        self._last_exec_result["branch"] = branch
-
-    @staticmethod
-    def random_inst_len(pc):
-        xor_ans = 0
-        for i in range(8):
-            xor_ans ^= (pc >> i) & 1
-        return 2 if xor_ans else 4
-
-    @staticmethod
-    def is_cond_branch_inst(branch):
-        return branch["type"] == "*.CBR"
-
-    @staticmethod
-    def is_jump_inst(branch):
-        return not Executor.is_cond_branch_inst(branch)
-
-    @staticmethod
-    def is_call_inst(branch):
-        return ".CALL" in branch["type"]
-
-    @staticmethod
-    def is_ret_inst(branch):
-        return ".RET" in branch["type"]
-
-    @staticmethod
-    def is_jal_inst(branch):
-        return branch["type"] == "I.JAL" or branch["type"] == "P.JAL"
-
-    @staticmethod
-    def is_jalr_inst(branch):
-        return ".JALR" in branch["type"] or ".JR" in branch["type"]
-
-    @staticmethod
-    def is_compressed_inst(branch):
-        type = branch["type"]
-        if "C." in type:
-            return True
-        elif Executor.is_cond_branch_inst(branch):
-            return Executor.random_inst_len(branch["pc"]) == 2
-        else:
-            return False
-
-    @staticmethod
-    def branch_inst_len(branch):
-        return 2 if Executor.is_compressed_inst(branch) else 4
-

tests/uFTB-with-ftq/env/ftq.py

@@ -1,315 +0,0 @@
-from mlvp import *
-from random import random
-from .bundle import *
-from .config import *
-from .utils import *
-from .executor import Executor
-from .ftb import *
-
-class PredictionStatistician:
-    """Predictive condition statistician for branch instructions"""
-
-    def __init__(self):
-        # { pc : [number, right_number]}
-        self.cond_branches_list = {}
-
-        # { pc : [type, number, right_number]}
-        self.jmp_branches_list = {}
-
-
-    def record_cond_branch(self, pc, correct):
-        if pc in self.cond_branches_list:
-            self.cond_branches_list[pc][0] += 1
-            self.cond_branches_list[pc][1] += correct
-        else:
-            self.cond_branches_list[pc] = [1, int(correct)]
-
-    def record_jmp_branch(self, pc, branch_type, correct):
-        if pc in self.jmp_branches_list:
-            self.jmp_branches_list[pc][1] += 1
-            self.jmp_branches_list[pc][2] += correct
-        else:
-            self.jmp_branches_list[pc] = [branch_type, 1, int(correct)]
-
-    def summary(self):
-        summary_str = ""
-        summary_str += "=" * 30 + "\n"
-        summary_str += "Summary\n"
-        summary_str += "[Conditional Branches]\n"
-        cond_branches_total = sum([record[0] for record in self.cond_branches_list.values()])
-        cond_branches_correct = sum([record[1] for record in self.cond_branches_list.values()])
-        summary_str += f"Total: {cond_branches_total}, Correct: {cond_branches_correct}, Accuracy: {cond_branches_correct / max(1,cond_branches_total)}\n"
-
-        for pc, record in self.cond_branches_list.items():
-            summary_str += f"PC: {hex(pc)}\tTotal: {record[0]}\tCorrect: {record[1]}\tAccuracy: {record[1] / record[0]}\n"
-
-        summary_str += "[Jump Branches]\n"
-        jmp_branches_total = sum([record[1] for record in self.jmp_branches_list.values()])
-        jmp_branches_correct = sum([record[2] for record in self.jmp_branches_list.values()])
-        summary_str += f"Total: {jmp_branches_total}, Correct: {jmp_branches_correct}, Accuracy: {jmp_branches_correct / max(1,jmp_branches_total)}\n"
-        for pc, record in self.jmp_branches_list.items():
-            summary_str += f"PC: {hex(pc)}\tType: {record[0]}\tTotal: {record[1]}\tCorrect: {record[2]}\tAccuracy: {record[2] / max(1,record[1])}\n"
-
-        summary_str += "[All Branches]\n"
-        total = cond_branches_total + jmp_branches_total
-        correct = cond_branches_correct + jmp_branches_correct
-        summary_str += f"Total: {total}, Correct: {correct}, Accuracy: {correct /max(1, total)}\n"
-
-        info(summary_str)
-
-    @staticmethod
-    def get_type(is_call, is_ret, is_jalr, is_jal):
-        if is_call:
-            return "call"
-        elif is_ret:
-            return "ret"
-        elif is_jalr:
-            return "jalr"
-        elif is_jal:
-            return "jal"
-        else:
-            return "jmp"
-
-pred_stat = PredictionStatistician()
-
-
-
-class FTQEntry:
-    """Stores all the information that FTQ entries need to record."""
-
-    def __init__(self):
-        self.pc = None
-        self.ftb = None
-        self.full_pred = None
-
-class FTQ:
-    """Simulate FTQ behavior."""
-
-    def __init__(self):
-        self.executor = Executor(filename=PROGRAM_PATH, reset_vector=RESET_VECTOR)
-
-        self.entries = [FTQEntry() for _ in range(32)]
-        self.bpu_ptr = 0
-        self.exec_ptr = 0
-
-        self.update_queue = []
-        self.redirect_queue = []
-
-    def update(self, bpu_out, ftb_entry):
-        # print("[FTQ]")
-
-        # Get the result from BPU out and update the FTQ entry
-        self._update_entries(bpu_out, ftb_entry)
-
-        # Execute a FTQ entry
-        self._exec_one_ftq_entry()
-
-        # Generate update and redirect request
-        update_request, redirect_request = None, None
-        if self.update_queue:
-            update_request = self._generate_update_request(self.update_queue.pop(0))
-            debug(f"Send Update Request: {hex(update_request['bits_pc'])}\
-                            br_taken_mask: {update_request['bits_br_taken_mask_0']}, {update_request['bits_br_taken_mask_1']}")
-        if self.redirect_queue:
-            cfi_target = self.redirect_queue.pop(0)
-            redirect_request = self._generate_redirect_request(cfi_target)
-            debug("Send Redirect Request: (target: %s)" % hex(cfi_target))
-
-        return (update_request, redirect_request)
-
-
-
-
-
-    def _get_entry(self, ptr):
-        return self.entries[ptr % 32]
-
-    def _exec_one_ftq_entry(self):
-        if self.exec_ptr >= self.bpu_ptr:
-            return None
-
-        # Get a FTQ entry
-        entry = self._get_entry(self.exec_ptr)
-        executor_current_pc = self.executor.current_inst()[0]
-        self.exec_ptr += 1
-        debug("Executing FTQ entry at pc %s" % hex(entry.pc))
-
-        # Prediction Block Hit
-        if entry.full_pred["hit"] and entry.pc == executor_current_pc:
-            debug("Prediction Block Hit")
-
-            # Execute the prediction block
-            all_branches, redirect_addr, br_taken_mask = self._execute_this_pred_block(entry.pc, entry.full_pred)
-            if redirect_addr is None:
-                debug("Predicition is correct")
-            else:
-                debug("Prediction is wrong, redirect to %s" % hex(redirect_addr))
-            new_ftb_entry = self._update_ftb_entry_from_branches(entry.pc, entry.ftb, all_branches, br_taken_mask)
-            self.update_queue.append((entry.pc, new_ftb_entry, br_taken_mask))
-            if redirect_addr is not None:
-                self.redirect_queue.append((redirect_addr))
-
-        # Prediction Block Miss
-        else:
-            debug("Prediction Block Miss")
-            if entry.pc != executor_current_pc:
-                debug("Target Error: actual: %s expected: %s" % (hex(entry.pc), hex(executor_current_pc)))
-
-            # Create a new FTB entry and update & redirect
-            new_ftb_entry, br_taken_mask = self._generate_new_ftb_entry(executor_current_pc)
-            self.update_queue.append((executor_current_pc, new_ftb_entry, br_taken_mask))
-            self.redirect_queue.append((self.executor.current_inst()[0]))
-
-    def _generate_update_request(self, update_queue_item):
-        pc, new_ftb_entry, br_taken_mask = update_queue_item[0], update_queue_item[1], update_queue_item[2]
-        update_request = {}
-
-        update_request["valid"] = True
-        update_request["bits_pc"] = pc
-        update_request["ftb_entry"] = new_ftb_entry.__dict__()
-        update_request["bits_br_taken_mask_0"] = 0 if len(br_taken_mask) == 0 else br_taken_mask[0]
-        update_request["bits_br_taken_mask_1"] = 0 if len(br_taken_mask) < 2 else br_taken_mask[1]
-
-        return update_request
-
-    def _generate_redirect_request(self, cfi_target):
-        redirect_request = {}
-        redirect_request["cfiUpdate"] = {}
-        redirect_request["cfiUpdate"]["target"] = cfi_target
-
-        return redirect_request
-
-    def _update_ftb_entry_from_branches(self, pc, ftb_entry, branches, br_taken_mask):
-        # update always_taken
-        if len(br_taken_mask) >= 1:
-            ftb_entry.always_taken[0] &= br_taken_mask[0]
-        if len(br_taken_mask) >= 2:
-            ftb_entry.always_taken[1] &= br_taken_mask[1]
-
-        # update jmp target
-        for branch in branches:
-            if Executor.is_jump_inst(branch):
-                ftb_entry.tailSlot.lower = get_lower_addr(branch["target"], 20)
-                ftb_entry.tailSlot.tarStart = get_target_stat(pc >> 20, branch["target"] >> 20)
-
-        return ftb_entry
-
-    def _record_branch_helper(self, branch, cfi_addr, cfi_target):
-        if Executor.is_cond_branch_inst(branch):
-            correct = None
-            if branch["taken"]:
-                correct = cfi_addr is not None and branch["pc"] == cfi_addr
-            else:
-                correct = cfi_addr is None or branch["pc"] != cfi_addr
-            pred_stat.record_cond_branch(branch["pc"], correct)
-        else:
-            correct = cfi_addr is not None and branch["pc"] == cfi_addr and branch["target"] == cfi_target
-            pred_stat.record_jmp_branch(branch["pc"], PredictionStatistician.get_type(Executor.is_call_inst(branch),
-                                                                                        Executor.is_ret_inst(branch),
-                                                                                        Executor.is_jalr_inst(branch),
-                                                                                        Executor.is_jal_inst(branch)),
-                                            correct)
-
-    def _execute_this_pred_block(self, pc, full_pred):
-        end_pc = full_pred["fallThroughAddr"]
-        cfi_addr = get_cfi_addr_from_full_pred_dict(pc, full_pred)
-        cfi_target = get_target_from_full_pred_dict(pc, full_pred)
-
-        all_branches = []
-        br_taken_mask = []
-        redirect_addr = None
-        while pc < end_pc:
-            _, inst_len, branch = self.executor.current_inst()
-            self.executor.next_inst()
-            if branch is not None:
-                br_taken_mask.append(branch["taken"])
-                all_branches.append(branch)
-                self._record_branch_helper(branch, cfi_addr, cfi_target)
-
-            pred_cfi_valid = cfi_addr is not None and pc == cfi_addr
-            exec_cfi_valid = branch is not None and branch["taken"]
-            pc += inst_len
-
-            if pred_cfi_valid and exec_cfi_valid:
-                if cfi_target != branch["target"]:
-                    redirect_addr = branch["target"]
-                break
-            elif pred_cfi_valid and not exec_cfi_valid:
-                redirect_addr = pc
-                break
-            elif not pred_cfi_valid and exec_cfi_valid:
-                redirect_addr = branch["target"]
-                break
-
-        return all_branches, redirect_addr, br_taken_mask
-
-    def _generate_new_ftb_entry(self, pc):
-        br_taken_mask = []
-        ftb_entry = FTBEntry()
-
-        fallthrough_addr = pc
-        while fallthrough_addr < pc + PREDICT_WIDTH_BYTES:
-            _, inst_len, branch = self.executor.current_inst()
-
-            if branch is not None:
-                if Executor.is_cond_branch_inst(branch):
-                    success = ftb_entry.add_cond_branch_inst(pc, branch["pc"], branch["taken"], branch["target"])
-                    br_taken_mask.append(branch["taken"])
-
-                    if not success:
-                        break
-                    else:
-                        pred_stat.record_cond_branch(branch["pc"], False)
-                        self.executor.next_inst()
-                        fallthrough_addr += inst_len
-                        if branch["taken"]:
-                            break
-                else:
-                    success = ftb_entry.add_jmp_inst(pc,
-                                                  branch["pc"],
-                                                  branch["target"],
-                                                  inst_len,
-                                                  Executor.is_call_inst(branch),
-                                                  Executor.is_ret_inst(branch),
-                                                  Executor.is_jalr_inst(branch),
-                                                  Executor.is_jal_inst(branch))
-                    if success:
-                        pred_stat.record_jmp_branch(branch["pc"], PredictionStatistician.get_type(Executor.is_call_inst(branch),
-                                                                                                    Executor.is_ret_inst(branch),
-                                                                                                    Executor.is_jalr_inst(branch),
-                                                                                                    Executor.is_jal_inst(branch)),
-                                                        False)
-                        fallthrough_addr += 2
-                        self.executor.next_inst()
-
-                    break
-            else:
-                fallthrough_addr += inst_len
-                self.executor.next_inst()
-
-        ftb_entry.valid = True
-        ftb_entry.pftAddr = get_pftaddr(fallthrough_addr)
-        ftb_entry.carry = get_pftaddr_carry(pc, fallthrough_addr)
-
-        debug("Generate FTB Entry")
-        debug(ftb_entry.__str__(pc))
-
-        return ftb_entry, br_taken_mask
-
-    def _update_entries(self, bpu_out, ftb_entry):
-        if bpu_out["s1"]["valid"]:
-            debug("Add ftq entry (pc: %s)" % hex(bpu_out["s1"]["pc_3"]))
-            entry = self._get_entry(self.bpu_ptr)
-            entry.full_pred = bpu_out["s1"]["full_pred"]
-            entry.pc = bpu_out["s1"]["pc_3"]
-            entry.ftb = ftb_entry
-            self.bpu_ptr += 1
-
-
-
-if __name__ == "__main__":
-    parser = Executor()
-    for _ in range (100):
-        print(parser.current_inst())
-        parser.next_inst()
-

tests/uFTB-with-ftq/env/uftb_model.py

@@ -1,138 +0,0 @@
-from mlvp.utils import PLRU, TwoBitsCounter
-from mlvp import *
-from .ftb import *
-
-class uFTBWay:
-    def __init__(self):
-        self.valid = 0
-        self.tag = 0
-        self.ftb_entry = FTBEntry()
-
-    @staticmethod
-    def get_tag(pc):
-        return pc >> INST_OFFSET_BITS & ((1 << UFTB_TAG_SIZE) - 1)
-
-class uFTBModel:
-    def __init__(self):
-        self.replacer = PLRU(UFTB_WAYS_NUM)
-        self.ftbways = [uFTBWay() for _ in range(UFTB_WAYS_NUM)]
-        self.counters = [[TwoBitsCounter(), TwoBitsCounter()] for _ in range(UFTB_WAYS_NUM)]
-
-        # Update requests are used to update FTBways and counters.
-        self.update_queue = []
-
-        # The update queue of the replacement algorithm, and there are two channels,
-        # the first channel has a higher priority.
-        self.replacer_update_queue = [[], []]
-
-    def update(self, update_request):
-        self.update_queue.append((update_request, 2, None))
-
-    def generate_output(self, s1_fire, s1_pc):
-        self._process_update()
-        if s1_fire:
-            hit_way = self._find_hit_way(s1_pc)
-            if hit_way is None:
-                return None
-            self.replacer_update_queue[0].append((hit_way, 1))
-
-            ftb_entry = self.ftbways[hit_way].ftb_entry
-            br_taken_mask = self._generate_br_taken_mask(hit_way)
-
-            return ftb_entry, br_taken_mask, hit_way
-
-    def print_all_ftb_ways(self):
-        for i in range(UFTB_WAYS_NUM):
-            debug(f"way {i}: valid: {self.ftbways[i].valid}, tag: {hex(self.ftbways[i].tag << 1)}")
-
-    def _generate_br_taken_mask(self, hit_way):
-        ftb_entry = self.ftbways[hit_way].ftb_entry
-        br_taken_mask = [self.counters[hit_way][0].get_prediction(), self.counters[hit_way][1].get_prediction()]
-        for i in range(2):
-            if ftb_entry.always_taken[i]:
-                br_taken_mask[i] = 1
-        return br_taken_mask
-
-    def _process_update(self):
-        # Update replacement algorithm
-        for i in range(2):
-            new_update_queue = []
-            for j in range(len(self.replacer_update_queue[i])):
-                if self.replacer_update_queue[i][j][1] == 0:
-                    self.replacer.update(self.replacer_update_queue[i][j][0])
-                else:
-                    new_update_queue.append((self.replacer_update_queue[i][j][0], self.replacer_update_queue[i][j][1] - 1))
-            self.replacer_update_queue[i] = new_update_queue
-
-        # Processing update requests
-
-        # Find the item for the next cycle update to fit the dut hit mode
-        next_cycle_update_item = []
-        for i in range(len(self.update_queue)):
-            selected_way = self.update_queue[i][2]
-            if self.update_queue[i][1] == 1:
-                if selected_way is None:
-                    selected_way = self.replacer.get()
-                next_cycle_update_item.append((self.update_queue[i][0], selected_way))
-                self.update_queue[i] = (self.update_queue[i][0], self.update_queue[i][1], selected_way)
-                self.replacer_update_queue[1].insert(0, (selected_way, 0))
-
-        # Update request processing
-        new_update_queue = []
-        for i in range(len(self.update_queue)):
-            if self.update_queue[i][1] == 0:
-                self._update_all(self.update_queue[i][0], self.update_queue[i][2])
-            else:
-                selected_way = self.update_queue[i][2]
-                if self.update_queue[i][1] == 2:
-                    selected_way = self._find_hit_way(self.update_queue[i][0]['bits_pc'])
-
-                    for (update_request, way) in next_cycle_update_item:
-                        if uFTBWay.get_tag(self.update_queue[i][0]['bits_pc']) == uFTBWay.get_tag(update_request["bits_pc"]):
-                            if selected_way is None or way < selected_way:
-                                selected_way = way
-                                break
-                    debug(f"Hit selected way is {selected_way}")
-
-                new_update_queue.append((self.update_queue[i][0], self.update_queue[i][1] - 1, selected_way))
-        self.update_queue = new_update_queue
-
-    def _find_hit_way(self, pc):
-        tag = uFTBWay.get_tag(pc)
-        for i in range(UFTB_WAYS_NUM):
-            if self.ftbways[i].valid and self.ftbways[i].tag == tag:
-                return i
-        return None
-
-    def _update_ftb_ways(self, update_request, selected_way):
-        if not update_request["valid"]:
-            return
-
-        debug(f"ftb entry {hex(update_request['bits_pc'])} is put into way {selected_way}")
-        self.ftbways[selected_way].valid = 1
-        self.ftbways[selected_way].tag = uFTBWay.get_tag(update_request["bits_pc"])
-        self.ftbways[selected_way].ftb_entry = FTBEntry.from_dict(update_request["ftb_entry"])
-
-    def _update_counters(self, update_request, selected_way):
-        if not update_request["valid"]:
-            return
-
-        need_to_update = [False, False]
-        brslot_valid = [update_request["ftb_entry"]["brSlots_0_valid"], update_request["ftb_entry"]["tailSlot_valid"] and update_request["ftb_entry"]["tailSlot_sharing"]]
-        br_taken_mask = [update_request["bits_br_taken_mask_0"], update_request["bits_br_taken_mask_1"]]
-        always_taken = [update_request["ftb_entry"]["always_taken_0"], update_request["ftb_entry"]["always_taken_1"]]
-
-        cfi_pos = 0 if br_taken_mask[0] else (1 if br_taken_mask[1] else 2)
-        for i in range(2):
-            need_to_update[i] = i <= cfi_pos \
-                                and not always_taken[i] \
-                                and brslot_valid[i]
-
-        for i in range(2):
-            if need_to_update[i]:
-                self.counters[selected_way][i].update(br_taken_mask[i])
-
-    def _update_all(self, update_request, selected_way):
-        self._update_ftb_ways(update_request, selected_way)
-        self._update_counters(update_request, selected_way)
-

tests/uFTB-with-ftq/tests/test_with_ftq.py

@@ -1,71 +0,0 @@
-import mlvp
-import pytest
-from mlvp.triggers import *
-
-import os
-os.sys.path.append(os.path.dirname(os.path.abspath(__file__)) + "/..")
-
-from env.bundle import *
-from env.bpu_top import *
-from env.config import *
-
-os.sys.path.append(DUT_PATH)
-
-from UT_FauFTB import *
-
-def set_imm_mode(uFTB):
-    imm_mode = uFTB.io_s0_fire_0.xdata.Imme
-    need_to_write_imm = ["io_s0_fire_0", "io_s0_fire_1", "io_s0_fire_2", "io_s0_fire_3",
-                        "io_s1_fire_0", "io_s2_fire_0", "io_in_bits_s0_pc_0", "io_in_bits_s0_pc_1",
-                        "io_in_bits_s0_pc_2", "io_in_bits_s0_pc_3"]
-    for name in need_to_write_imm:
-        getattr(uFTB, name).xdata.SetWriteMode(imm_mode)
-
-
-@pytest.mark.mlvp_async
-async def test_uftb(mlvp_request):
-    uFTB = mlvp_request
-    set_imm_mode(uFTB)
-
-    uFTB_update = UpdateBundle.from_prefix("io_update_").set_name("uFTB_update").bind(uFTB)
-    uFTB_out = BranchPredictionResp.from_prefix("io_out_").set_name("uFTB_out").bind(uFTB)
-    pipeline_ctrl = PipelineCtrlBundle.from_prefix("io_").set_name("pipeline_ctrl").bind(uFTB)
-    enable_ctrl = EnableCtrlBundle.from_prefix("io_ctrl_").set_name("enable_ctrl").bind(uFTB)
-
-    mlvp.start_clock(uFTB)
-    mlvp.create_task(BPUTop(uFTB, uFTB_out, uFTB_update, pipeline_ctrl, enable_ctrl).run())
-
-    await ClockCycles(uFTB, MAX_CYCLE)
-
-    pred_stat.summary()
-
-import mlvp.funcov as fc
-from mlvp.reporter import *
-from mlvp import PreRequest
-
-@pytest.fixture()
-def mlvp_request(mlvp_pre_request: PreRequest):
-    mlvp.setup_logging(mlvp.INFO)
-    uFTB = mlvp_pre_request.create_dut(DUTFauFTB, "clock")
-
-    # Set Coverage
-    g1 = fc.CovGroup("interaction")
-    g1.add_watch_point(uFTB.reset, { "reset": fc.Eq(1), }, name="reset")
-    g1.add_watch_point(uFTB.io_ctrl_ubtb_enable, { "ubtb_enable": fc.Eq(1), }, name="ubtb_enable")
-    g1.add_watch_point(uFTB.io_s0_fire_0, { "s0_fire": fc.Eq(1), }, name="s0_fire_0")
-    g1.add_watch_point(uFTB.io_s1_fire_0, { "s1_fire": fc.Eq(1), }, name="s1_fire_0")
-    g1.add_watch_point(uFTB.io_s2_fire_0, { "s2_fire": fc.Eq(1), }, name="s2_fire_0")
-    g1.add_watch_point(uFTB.io_update_valid, { "update_valid": fc.Eq(1), }, name="update_valid")
-
-    g2 = fc.CovGroup("ftb_entry")
-    g2.add_watch_point(uFTB.io_out_s1_full_pred_0_hit, { "hit": fc.Eq(1), "not_hit": fc.Eq(0) }, name="s1_full_pred_0_hit")
-    g2.add_watch_point(uFTB.io_out_s1_full_pred_3_fallThroughErr, { "fallThroughErr": fc.Eq(1), "not_fallThroughErr": fc.Eq(0) }, name="s1_full_pred_3_fallThroughErr")
-    g2.add_watch_point(uFTB.io_out_s1_full_pred_0_slot_valids_0, { "slot_valids_0": fc.Eq(1), "slot_valids_0_invalid": fc.Eq(0) }, name="s1_full_pred_0_slot_valids_0")
-    g2.add_watch_point(uFTB.io_out_s1_full_pred_0_slot_valids_1, { "slot_valids_1": fc.Eq(1), "slot_valids_1_invalid": fc.Eq(0) }, name="s1_full_pred_0_slot_valids_1")
-    g2.add_watch_point(uFTB.io_out_s1_full_pred_0_br_taken_mask_0, { "br_taken_mask_0": fc.Eq(1), "br_taken_mask_0_invalid": fc.Eq(0) }, name="s1_full_pred_0_br_taken_mask_0")
-    g2.add_watch_point(uFTB.io_out_s1_full_pred_0_br_taken_mask_1, { "br_taken_mask_1": fc.Eq(1), "br_taken_mask_1_invalid": fc.Eq(0) }, name="s1_full_pred_0_br_taken_mask_1")
-    g2.add_watch_point(uFTB.io_out_s1_full_pred_0_is_br_sharing, { "is_br_sharing": fc.Eq(1), "is_br_sharing_invalid": fc.Eq(0) }, name="s1_full_pred_0_is_br_sharing")
-
-    mlvp_pre_request.add_cov_groups([g1, g2])
-
-    return uFTB