forked from OSchip/UnityChipVerification
Add uFTB test cases with ftq
This commit is contained in:
parent
aeea5bb381
commit
2e8c2c953f
|
|
@ -3,4 +3,5 @@ split_verilogs
|
|||
.vscode
|
||||
**/__pycache__/
|
||||
*.fst
|
||||
**/UT_*
|
||||
**/UT_*
|
||||
NemuBR/
|
||||
|
|
|
|||
Binary file not shown.
|
After Width: | Height: | Size: 38 KiB |
|
|
@ -0,0 +1,128 @@
|
|||
# uFTB-env
|
||||
|
||||
## 介绍
|
||||
|
||||
本项目提供了基于真实指令流的香山处理器 uFTB 分支预测器的仿真验证环境，以及时钟精确的 uFTB 参考模型，最终可给出 uFTB 的分支预测准确率。
|
||||
|
||||
为此，本项目为 uFTB 提供了简易的 BPU Top Wrapper，以向 uFTB 提供时序控制和输入输出处理。并向 BPU Top 提供了简易的 FTQ 实现，FTQ 中实例化了一个真实的程序仿真器，用于生成真实指令流，FTQ 会处理 BPU 产生的预测结果，并向 BPU 提供更新请求与重定向请求的执行反馈。大致的结构可参考下图：
|
||||
|
||||
|
||||
|
||||
对于香山 BPU 中其他子预测器的验证，可复用本项目中的真实指令执行环境，但需要对 DUT 的接口、时序以及需要使用的预测结果通道等进行适配。
|
||||
|
||||
## 快速使用
|
||||
|
||||
### 环境配置
|
||||
|
||||
**1. 安装 mlvp**
|
||||
|
||||
具体步骤参见 https://github.com/XS-MLVP/mlvp
|
||||
|
||||
**2. 生成 BRTParser Trace 工具**
|
||||
|
||||
为了生成真实的指令流，BRTParser 作为一个自定义的工具已经被放置在了本仓库中，但其中缺少了模拟器仿真程序，需要自行编译生成，具体步骤参见 https://github.com/yaozhicheng/NEMU
|
||||
|
||||
生成编译结果 `NemuBR` 后，将其放置在 `BRTParser` 目录下，工具即可正常使用。
|
||||
|
||||
**3. 编译 DUT**
|
||||
|
||||
参照 https://github.com/XS-MLVP/env-xs-ov-00-bpu 编译 DUT，将编译结果（`UT_FauFTB`）放置本仓库根目录下。
|
||||
|
||||
### 仿真验证
|
||||
|
||||
在本仓库根目录下执行
|
||||
|
||||
```shell
|
||||
python uftb-env/tb.py
|
||||
```
|
||||
|
||||
即可开始仿真验证。
|
||||
|
||||
程序运行结束后，会打印出分支预测的统计信息。
|
||||
|
||||
若要更改需要执行的程序，可在 `config.py` 中更改相应变量的值，仿真所需的真实程序已经放置在 `ready-to-run` 目录下。若要更改仿真所持续的周期数，可在 `config.py` 中更改 `MAX_CYCLE` 的值。
|
||||
|
||||
## 使用说明
|
||||
|
||||
### 目录结构
|
||||
|
||||
```
|
||||
uFTB-env/
|
||||
├── BRTParser # BRTParser 工具，用于生成真实指令流
|
||||
│ ├── __init__.py
|
||||
│ ├── NemuBR.txt
|
||||
│ └── NemuBR # 编译生成的模拟器仿真程序
|
||||
├── LICENSE
|
||||
├── README.assets
|
||||
│ └── env.png
|
||||
├── README.md
|
||||
├── ready-to-run # 可供直接运行的真实程序
|
||||
│ ├── coremark-2-iteration.bin
|
||||
│ ├── linux-0xa0000.bin
|
||||
│ ├── linux.bin
|
||||
│ └── microbench.bin
|
||||
├── 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 # 相关工具函数
|
||||
└── UT_FauFTB # DUT 编译结果
|
||||
```
|
||||
|
||||
### 指令执行器
|
||||
|
||||
`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 的预测结果。
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,220 @@
|
|||
from mlvp import *
|
||||
from bundle import *
|
||||
from ftq import *
|
||||
from uftb_model import uFTBModel
|
||||
|
||||
def assert_equal(a, b):
|
||||
if a != b:
|
||||
print(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.collect()
|
||||
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
|
||||
|
||||
# print("-" * 30)
|
||||
if self.s1_fire:
|
||||
# Debug Imformation
|
||||
# print("[BPU]")
|
||||
# print("New prediction at", hex(self.s1_pc))
|
||||
# if bpu_output["s1"]["full_pred"]["hit"]:
|
||||
# print("Dut Hit")
|
||||
|
||||
# print("FTB Entry in pred result: ")
|
||||
# if bpu_output["s1"]["full_pred"]["hit"]:
|
||||
# ftb_entry.print(self.s1_pc)
|
||||
# else:
|
||||
# print("No FTB Entry")
|
||||
# print("br_taken_mask:", bpu_output["s1"]["full_pred"]["br_taken_mask_0"], bpu_output["s1"]["full_pred"]["br_taken_mask_1"])
|
||||
|
||||
# print("FTB Entry in uFTB Model: ")
|
||||
# if model_output:
|
||||
# model_output[0].print(self.s1_pc)
|
||||
# print("br_taken_mask:", model_output[1])
|
||||
# else:
|
||||
# print("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
|
||||
|
|
@ -0,0 +1,51 @@
|
|||
from mlvp import Interface
|
||||
|
||||
class PipelineCtrlBundle(Interface):
|
||||
signals_list = ["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(Interface):
|
||||
signals_list = ["ubtb_enable", "btb_enable", "bim_enable", "tage_enable",
|
||||
"sc_enable", "ras_enable", "loop_enable"]
|
||||
|
||||
|
||||
class FTBEntryBundle(Interface):
|
||||
signals_list = ["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(Interface):
|
||||
signals_list = ["valid", "bits_pc", "bits_br_taken_mask_0", "bits_br_taken_mask_1"]
|
||||
|
||||
sub_interfaces = [
|
||||
("ftb_entry", lambda dut: FTBEntryBundle.from_prefix(dut, "bits_ftb_entry_"))
|
||||
]
|
||||
|
||||
class FullBranchPredirectionBundle(Interface):
|
||||
signals_list = ["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(Interface):
|
||||
signals_list = ["pc_3", "valid", "hasRedirect", "ftq_idx"]
|
||||
sub_interfaces = [
|
||||
("full_pred", lambda dut: FullBranchPredirectionBundle.from_regex(dut, r"full_pred_\d_(.*)"))
|
||||
]
|
||||
|
||||
|
||||
class BranchPredictionResp(Interface):
|
||||
signals_list = ["last_stage_meta"]
|
||||
sub_interfaces = [
|
||||
("s1", lambda dut: BranchPredictionBundle.from_prefix(dut, "s1_")),
|
||||
("s2", lambda dut: BranchPredictionBundle.from_prefix(dut, "s2_")),
|
||||
("s3", lambda dut: BranchPredictionBundle.from_prefix(dut, "s3_")),
|
||||
("last_stage_ftb_entry", lambda dut: FTBEntryBundle.from_prefix(dut, "last_stage_ftb_entry_"))
|
||||
]
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
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
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,102 @@
|
|||
from config import *
|
||||
|
||||
import os
|
||||
os.sys.path.append(UTILS_PATH)
|
||||
|
||||
from BRTParser import BRTParser
|
||||
|
||||
class Executor:
|
||||
"""Get program real execution instruction flow."""
|
||||
|
||||
def __init__(self, filename, reset_vector=0x80000000):
|
||||
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):
|
||||
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
|
||||
|
||||
|
|
@ -0,0 +1,191 @@
|
|||
from utils import *
|
||||
|
||||
class FTBSlot:
|
||||
def __init__(self):
|
||||
self.valid = 0
|
||||
self.offset = 0
|
||||
self.lower = 0
|
||||
self.tarStart = 0
|
||||
self.sharing = 0
|
||||
|
||||
def print(self, pc, is_cond_branch):
|
||||
if not self.valid:
|
||||
print("*\tInvalid FTBSlot")
|
||||
return
|
||||
|
||||
if is_cond_branch:
|
||||
print(f"*\t[Conditional Branch Inst] at PC {hex(get_slot_addr(pc, self.offset))}: Target: {hex(get_target_addr(pc, self.tarStart, self.lower, 12))}")
|
||||
else:
|
||||
print(f"*\t[Jump Inst] PC {hex(get_slot_addr(pc, self.offset))}: Target: {hex(get_target_addr(pc, self.tarStart, self.lower, 20))}")
|
||||
|
||||
|
||||
class FTBEntry:
|
||||
def __init__(self):
|
||||
self.valid = 0
|
||||
self.brSlot = FTBSlot()
|
||||
self.tailSlot = FTBSlot()
|
||||
self.pftAddr = 0
|
||||
self.carry = 0
|
||||
self.isCall = False
|
||||
self.isRet = False
|
||||
self.isJal = False
|
||||
self.isJalr = False
|
||||
self.last_may_be_rvi_call = False
|
||||
self.always_taken = [0, 0]
|
||||
|
||||
def add_cond_branch_inst(self, start_pc, inst_pc, is_taken, target_addr):
|
||||
if self.brSlot.valid and self.tailSlot.valid:
|
||||
return False
|
||||
|
||||
slot = FTBSlot()
|
||||
slot.valid = True
|
||||
slot.offset = get_slot_offset(start_pc, inst_pc)
|
||||
slot.lower = get_lower_addr(target_addr, 12)
|
||||
slot.tarStart = get_target_stat(start_pc >> 12, target_addr >> 12)
|
||||
|
||||
if self.brSlot.valid:
|
||||
self.tailSlot = slot
|
||||
self.tailSlot.sharing = True
|
||||
self.always_taken[1] = is_taken
|
||||
else:
|
||||
self.brSlot = slot
|
||||
self.always_taken[0] = is_taken
|
||||
|
||||
return True
|
||||
|
||||
def add_jmp_inst(self, start_pc, inst_pc, target_addr, inst_len, is_call, is_ret, is_jalr, is_jal):
|
||||
if self.tailSlot.valid:
|
||||
return False
|
||||
|
||||
self.tailSlot.valid = True
|
||||
self.tailSlot.offset = get_slot_offset(start_pc, inst_pc)
|
||||
self.tailSlot.lower = get_lower_addr(target_addr, 20)
|
||||
self.tailSlot.tarStart = get_target_stat(start_pc >> 20, target_addr >> 20)
|
||||
self.tailSlot.sharing = False
|
||||
|
||||
self.isCall = is_call
|
||||
self.isRet = is_ret
|
||||
self.isJalr = is_jalr
|
||||
self.isJal = is_jal
|
||||
self.last_may_be_rvi_call = is_call and inst_len == 4
|
||||
|
||||
return True
|
||||
|
||||
def put_to_full_pred_dict(self, pc, d):
|
||||
d["hit"] = 1
|
||||
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["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)
|
||||
d["is_jal"] = self.isJal
|
||||
d["is_jalr"] = self.isJalr
|
||||
d["is_call"] = self.isCall
|
||||
d["is_ret"] = self.isRet
|
||||
d["is_br_sharing"] = self.tailSlot.sharing
|
||||
d["last_may_be_rvi_call"] = self.last_may_be_rvi_call
|
||||
d["br_taken_mask_0"] = self.always_taken[0]
|
||||
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,
|
||||
"brSlots_0_lower": self.brSlot.lower,
|
||||
"brSlots_0_tarStat": self.brSlot.tarStart,
|
||||
"brSlots_0_valid": self.brSlot.valid,
|
||||
"tailSlot_offset": self.tailSlot.offset,
|
||||
"tailSlot_lower": self.tailSlot.lower,
|
||||
"tailSlot_tarStat": self.tailSlot.tarStart,
|
||||
"tailSlot_sharing": self.tailSlot.sharing,
|
||||
"tailSlot_valid": self.tailSlot.valid,
|
||||
"pftAddr": self.pftAddr,
|
||||
"carry": self.carry,
|
||||
"isCall": self.isCall,
|
||||
"isRet": self.isRet,
|
||||
"isJalr": self.isJalr,
|
||||
"last_may_be_rvi_call": self.last_may_be_rvi_call,
|
||||
"always_taken_0": self.always_taken[0],
|
||||
"always_taken_1": self.always_taken[1]
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(self, d):
|
||||
entry = FTBEntry()
|
||||
entry.brSlot.offset = d["brSlots_0_offset"]
|
||||
entry.brSlot.lower = d["brSlots_0_lower"]
|
||||
entry.brSlot.tarStart = d["brSlots_0_tarStat"]
|
||||
entry.brSlot.valid = d["brSlots_0_valid"]
|
||||
entry.tailSlot.offset = d["tailSlot_offset"]
|
||||
entry.tailSlot.lower = d["tailSlot_lower"]
|
||||
entry.tailSlot.tarStart = d["tailSlot_tarStat"]
|
||||
entry.tailSlot.sharing = d["tailSlot_sharing"]
|
||||
entry.tailSlot.valid = d["tailSlot_valid"]
|
||||
entry.pftAddr = d["pftAddr"]
|
||||
entry.carry = d["carry"]
|
||||
entry.isCall = d["isCall"]
|
||||
entry.isRet = d["isRet"]
|
||||
entry.isJalr = d["isJalr"]
|
||||
entry.last_may_be_rvi_call = d["last_may_be_rvi_call"]
|
||||
entry.always_taken[0] = d["always_taken_0"]
|
||||
entry.always_taken[1] = d["always_taken_1"]
|
||||
return entry
|
||||
|
||||
@classmethod
|
||||
def from_full_pred_dict(self, pc, d):
|
||||
entry = FTBEntry()
|
||||
entry.brSlot.valid = d["slot_valids_0"]
|
||||
entry.brSlot.offset = d["offsets_0"]
|
||||
entry.brSlot.lower = get_lower_addr(d["targets_0"], 12)
|
||||
entry.brSlot.tarStart = get_target_stat(pc >> 12, d["targets_0"] >> 12)
|
||||
entry.tailSlot.valid = d["slot_valids_1"]
|
||||
entry.tailSlot.offset = d["offsets_1"]
|
||||
entry.tailSlot.sharing = d["is_br_sharing"]
|
||||
|
||||
if entry.tailSlot.sharing:
|
||||
entry.tailSlot.lower = get_lower_addr(d["targets_1"], 12)
|
||||
entry.tailSlot.tarStart = get_target_stat(pc >> 12, d["targets_1"] >> 12)
|
||||
else:
|
||||
entry.tailSlot.lower = get_lower_addr(d["targets_1"], 20)
|
||||
entry.tailSlot.tarStart = get_target_stat(pc >> 20, d["targets_1"] >> 20)
|
||||
|
||||
entry.pftAddr = get_pftaddr(d["fallThroughAddr"])
|
||||
entry.carry = get_pftaddr_carry(pc, d["fallThroughAddr"])
|
||||
entry.isCall = d["is_call"]
|
||||
entry.isRet = d["is_ret"]
|
||||
entry.isJal = d["is_jal"]
|
||||
entry.isJalr = d["is_jalr"]
|
||||
entry.last_may_be_rvi_call = d["last_may_be_rvi_call"]
|
||||
entry.always_taken[0] = d["br_taken_mask_0"]
|
||||
entry.always_taken[1] = d["br_taken_mask_1"]
|
||||
|
||||
return entry
|
||||
|
||||
|
||||
def print(self, pc):
|
||||
print(f"[FTBEntry at {hex(pc)}]")
|
||||
print(f"* Slots:")
|
||||
self.brSlot.print(pc, True)
|
||||
self.tailSlot.print(pc, self.tailSlot.sharing)
|
||||
print("* Other Info:")
|
||||
print(f"*\tFallthrough Addr: {hex(get_fallthrough_addr(pc, self.pftAddr, self.carry))}")
|
||||
print(f"*\tisCall: {self.isCall}, isRet: {self.isRet}, isJalr: {self.isJalr}, isJal: {self.isJal}")
|
||||
print(f"*\tlast_may_be_rvi_call: {self.last_may_be_rvi_call}, always_taken: {self.always_taken}")
|
||||
|
||||
class FTBProvider():
|
||||
def __init__(self):
|
||||
self.entries = {}
|
||||
|
||||
def update(self, update_request):
|
||||
if update_request["valid"]:
|
||||
self.entries[update_request["bits_pc"]] = FTBEntry.from_dict(update_request["ftb_entry"])
|
||||
|
||||
def provide_ftb_entry(self, fire, pc):
|
||||
if fire and pc in self.entries:
|
||||
return self.entries[pc]
|
||||
else:
|
||||
return None
|
||||
|
|
@ -0,0 +1,312 @@
|
|||
from bundle import *
|
||||
from config import *
|
||||
from utils import *
|
||||
from executor import Executor
|
||||
from ftb import *
|
||||
from random import random
|
||||
|
||||
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):
|
||||
print("=" * 30)
|
||||
print("Summary")
|
||||
print("[Conditional Branches]")
|
||||
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()])
|
||||
print(f"Total: {cond_branches_total}, Correct: {cond_branches_correct}, Accuracy: {cond_branches_correct / cond_branches_total}")
|
||||
|
||||
for pc, record in self.cond_branches_list.items():
|
||||
print(f"PC: {hex(pc)}\tTotal: {record[0]}\tCorrect: {record[1]}\tAccuracy: {record[1] / record[0]}")
|
||||
|
||||
print("[Jump Branches]")
|
||||
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()])
|
||||
print(f"Total: {jmp_branches_total}, Correct: {jmp_branches_correct}, Accuracy: {jmp_branches_correct / jmp_branches_total}")
|
||||
for pc, record in self.jmp_branches_list.items():
|
||||
print(f"PC: {hex(pc)}\tType: {record[0]}\tTotal: {record[1]}\tCorrect: {record[2]}\tAccuracy: {record[2] / record[1]}")
|
||||
|
||||
print("[All Branches]")
|
||||
total = cond_branches_total + jmp_branches_total
|
||||
correct = cond_branches_correct + jmp_branches_correct
|
||||
print(f"Total: {total}, Correct: {correct}, Accuracy: {correct / total}")
|
||||
|
||||
@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))
|
||||
# print("Send Update Request: %s" % 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)
|
||||
# print("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
|
||||
# print("Executing FTQ entry at pc %s" % hex(entry.pc))
|
||||
|
||||
# Prediction Block Hit
|
||||
if entry.full_pred["hit"] and entry.pc == executor_current_pc:
|
||||
# print("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:
|
||||
# print("Predicition is correct")
|
||||
# else:
|
||||
# print("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:
|
||||
# print("Prediction Block Miss")
|
||||
# if entry.pc != executor_current_pc:
|
||||
# print("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)
|
||||
|
||||
# print("Generate FTB Entry")
|
||||
# ftb_entry.print(pc)
|
||||
|
||||
return ftb_entry, br_taken_mask
|
||||
|
||||
def _update_entries(self, bpu_out, ftb_entry):
|
||||
if bpu_out["s1"]["valid"]:
|
||||
# print("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()
|
||||
|
||||
|
|
@ -0,0 +1,39 @@
|
|||
import mlvp
|
||||
|
||||
from bundle import *
|
||||
from bpu_top import *
|
||||
from config import MAX_CYCLE
|
||||
|
||||
import os
|
||||
os.sys.path.append(DUT_PATH)
|
||||
|
||||
from UT_FauFTB import *
|
||||
uFTB = DUTFauFTB()
|
||||
uFTB.init_clock("clock")
|
||||
|
||||
def set_imm_mode():
|
||||
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)
|
||||
|
||||
set_imm_mode()
|
||||
|
||||
async def uftb_test():
|
||||
uFTB_update = UpdateBundle.from_prefix(uFTB, "io_update_")
|
||||
uFTB_out = BranchPredictionResp.from_prefix(uFTB, "io_out_")
|
||||
pipeline_ctrl = PipelineCtrlBundle.from_prefix(uFTB, "io_")
|
||||
enable_ctrl = EnableCtrlBundle.from_prefix(uFTB, "io_ctrl_")
|
||||
|
||||
mlvp.create_task(mlvp.start_clock(uFTB))
|
||||
mlvp.create_task(BPUTop(uFTB, uFTB_out, uFTB_update, pipeline_ctrl, enable_ctrl).run())
|
||||
|
||||
await ClockCycles(uFTB, MAX_CYCLE)
|
||||
|
||||
if __name__ == "__main__":
|
||||
mlvp.run(uftb_test())
|
||||
uFTB.finalize()
|
||||
|
||||
pred_stat.summary()
|
||||
|
|
@ -0,0 +1,137 @@
|
|||
from mlvp.utils import PLRU, TwoBitsCounter
|
||||
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):
|
||||
print(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
|
||||
# print(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
|
||||
|
||||
# print(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)
|
||||
|
||||
|
|
@ -0,0 +1,79 @@
|
|||
from config import *
|
||||
|
||||
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)
|
||||
|
||||
|
||||
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
|
||||
elif target_higher > pc_higher:
|
||||
return TAR_OVF
|
||||
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:
|
||||
target_higher -= 1
|
||||
elif target_stat == TAR_OVF:
|
||||
target_higher += 1
|
||||
|
||||
return (target_higher << (target_lower_bits + INST_OFFSET_BITS)) | (target_lower << INST_OFFSET_BITS)
|
||||
|
||||
|
||||
def get_cfi_addr_from_full_pred_dict(pc, d):
|
||||
if not d["hit"]:
|
||||
return None
|
||||
elif d["slot_valids_0"] and d["br_taken_mask_0"]:
|
||||
return get_slot_addr(pc, d["offsets_0"])
|
||||
elif d["slot_valids_1"] and d["br_taken_mask_1"] and d["is_br_sharing"]:
|
||||
return get_slot_addr(pc, d["offsets_1"])
|
||||
elif d["slot_valids_1"] and not d["is_br_sharing"]:
|
||||
return get_slot_addr(pc, d["offsets_1"])
|
||||
else:
|
||||
return None
|
||||
|
||||
def get_target_from_full_pred_dict(pc, d):
|
||||
if not d["hit"]:
|
||||
return pc + PREDICT_WIDTH_BYTES
|
||||
elif d["slot_valids_0"] and d["br_taken_mask_0"]:
|
||||
return d["targets_0"]
|
||||
elif d["slot_valids_1"] and d["br_taken_mask_1"] and d["is_br_sharing"]:
|
||||
return d["targets_1"]
|
||||
elif d["slot_valids_1"] and not d["is_br_sharing"]:
|
||||
# return d["jalr_target"]
|
||||
return d["targets_1"]
|
||||
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
|
||||
}
|
||||
|
|
@ -0,0 +1 @@
|
|||
Place the generated NemuBR folder in this folder.
|
||||
|
|
@ -0,0 +1,131 @@
|
|||
#!coding=utf8
|
||||
|
||||
import os
|
||||
from .util import *
|
||||
import time
|
||||
|
||||
class BRTParser:
|
||||
|
||||
def __init__(self, check_trace=True) -> None:
|
||||
self.magic_head = b'\xbe\xbe\xbe\xbe\xbe\xbe\xbe\xbe\xbe\xbe\xbe\xbe\xbe\xbe\xbe\xbe'
|
||||
self.magic_tail = b'\xed\xed\xed\xed\xed\xed\xed\xed\xed\xed\xed\xed\xed\xed\xed\xed'
|
||||
self.type_map = {
|
||||
101 : "C.J",
|
||||
102 : "C.JR",
|
||||
103 : "C.CALL",
|
||||
104 : "C.RET",
|
||||
105 : "C.JALR",
|
||||
201 : "P.JAL",
|
||||
203 : "P.CALL",
|
||||
204 : "P.RET",
|
||||
0 : "*.CBR",
|
||||
1 : "I.JAL",
|
||||
2 : "I.JALR",
|
||||
3 : "I.CALL",
|
||||
4 : "I.RET",
|
||||
}
|
||||
self.logger = get_logger(self.__class__.__name__)
|
||||
self.check_trace = check_trace
|
||||
self.clear()
|
||||
|
||||
def clear(self):
|
||||
self.branchs = {}
|
||||
self.branchs_step = []
|
||||
self.statistics_type = {}
|
||||
|
||||
def disable_check(self):
|
||||
self.check_trace = False
|
||||
|
||||
def enable_check(self):
|
||||
self.check_trace = True
|
||||
|
||||
def parse_data(self, data):
|
||||
index = int.from_bytes(data[0:8], byteorder='little')
|
||||
pc = int.from_bytes(data[8:16], byteorder='little')
|
||||
target = int.from_bytes(data[16:24], byteorder='little')
|
||||
taken = int.from_bytes(data[24:28], byteorder='little') > 0
|
||||
btype = int.from_bytes(data[28:32], byteorder='little')
|
||||
if (btype not in self.type_map) and self.check_trace:
|
||||
self.logger.warning("Find Unrecognized Type: %d" % btype)
|
||||
btype = self.type_map.get(btype, "ERROR-%s"%btype)
|
||||
key = pc
|
||||
return key, index, pc, target, taken, btype
|
||||
|
||||
def load(self, file):
|
||||
if not os.path.isfile(file):
|
||||
self.logger.error("file: %s not find!" % file)
|
||||
return
|
||||
self.logger.debug("Load file: %s"%file)
|
||||
if not self.check_trace:
|
||||
self.logger.warning("Trace check is disabled!")
|
||||
time_start = time.time()
|
||||
with open(file, "rb") as fp:
|
||||
# read header
|
||||
header = fp.read(16)
|
||||
if(header != self.magic_head):
|
||||
self.logger.error("file[%s] is not a branch/jump trace")
|
||||
return
|
||||
pre_pc = -1
|
||||
while True:
|
||||
data = fp.read(32)
|
||||
if(data == self.magic_tail):
|
||||
break
|
||||
key, index, pc, target, taken, btype = self.parse_data(data)
|
||||
if pc < pre_pc and self.check_trace:
|
||||
self.logger.warning("Detect disordered PC (0x%x => 0x%x) sequence; potentially indicating a corrupted trace file." % (pre_pc, pc))
|
||||
if taken:
|
||||
pre_pc = target
|
||||
if key not in self.branchs:
|
||||
self.branchs[key] = {"pc": pc, "index": [index], "target": [target], "taken": [taken], "type": btype}
|
||||
# statistic
|
||||
if btype not in self.statistics_type:
|
||||
self.statistics_type[btype] = {"count":1, "taken": int(taken), "notaken": int(not taken)}
|
||||
else:
|
||||
self.statistics_type[btype]["count"] += 1
|
||||
self.statistics_type[btype]["taken"] += int(taken)
|
||||
self.statistics_type[btype]["notaken"] += int(not taken)
|
||||
else:
|
||||
self.branchs[key]["index"].append(index)
|
||||
self.branchs[key]["target"].append(target)
|
||||
self.branchs[key]["taken"].append(taken)
|
||||
self.statistics_type[btype]["taken"] += int(taken)
|
||||
self.statistics_type[btype]["notaken"] += int(not taken)
|
||||
|
||||
self.branchs_step.append((index, self.branchs[key], len(self.branchs[key]["index"]) - 1))
|
||||
self.logger.debug("%d branchs (%d checks), loaded! time cost: %s"%(len(self.branchs), len(self.branchs_step), fmt_seconds(time.time() - time_start)))
|
||||
|
||||
def fetch(self, file):
|
||||
from . import NemuBR as nbr
|
||||
nbr.br_monitor_init(["", "-b", file])
|
||||
while True:
|
||||
data = nbr.br_monitor_get()
|
||||
if data.index < 0:
|
||||
return None
|
||||
pc, index, target, taken, btype = data.pc, data.index, data.target, data.taken, data.type
|
||||
if (btype not in self.type_map) and self.check_trace:
|
||||
self.logger.warning("Find Unrecognized Type: %d" % btype)
|
||||
btype = self.type_map.get(btype, "ERROR-%s"%btype)
|
||||
if btype not in self.statistics_type:
|
||||
self.statistics_type[btype] = {"count":1, "taken": int(taken), "notaken": int(not taken)}
|
||||
else:
|
||||
self.statistics_type[btype]["count"] += 1
|
||||
self.statistics_type[btype]["taken"] += int(taken)
|
||||
self.statistics_type[btype]["notaken"] += int(not taken)
|
||||
yield {"pc": pc, "index": index, "target": target, "taken": taken, "type": btype}
|
||||
|
||||
def print_stat(self):
|
||||
keys = self.statistics_type.keys()
|
||||
print("\n%5s %8s %8s %8s %8s" % ("Index", "Type", "icount", "taken", "notaken"))
|
||||
count, taken, notaken = 0, 0, 0
|
||||
all_cal, all_ret = 0, 0
|
||||
for i, k in enumerate(sorted(keys)):
|
||||
data = self.statistics_type[k]
|
||||
print("%5d %8s %8d %8d %8d" % (i, k, data["count"], data["taken"], data["notaken"]))
|
||||
count += data["count"]
|
||||
taken += data["taken"]
|
||||
notaken += data["notaken"]
|
||||
if ".RET" in k:
|
||||
all_ret += data["taken"]
|
||||
elif ".CALL" in k:
|
||||
all_cal += data["taken"]
|
||||
print("%5d %8s %8d %8d %8d (%d checks, ins.ret - ins.call = %d)\n" % (len(keys), "ALL", count, taken, notaken, taken + notaken, all_ret - all_cal))
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
from .logger import *
|
||||
from .functions import *
|
||||
|
||||
|
|
@ -0,0 +1,24 @@
|
|||
|
||||
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
def fmt_time(t = None, fmt="%Y-%m-%d %H:%M:%S"):
|
||||
if t is None:
|
||||
t = time.time()
|
||||
return datetime.fromtimestamp(t).strftime(fmt)
|
||||
|
||||
def fmt_seconds(t: float):
|
||||
if t == 0:
|
||||
return "0 seconds"
|
||||
ret = ""
|
||||
for d, f, u in [(60, " %.2f", "second"), (60, " %d", "minute"), (24, " %d", "hour"), (sys.maxsize, " %d", "day")]:
|
||||
if t == 0:
|
||||
break
|
||||
v = t % d
|
||||
t = t // d
|
||||
ret = (f + " %s%s")%(v, u, "" if v == 1 else "s") + ret
|
||||
return ret.strip()
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
#coding=utf8
|
||||
|
||||
import logging
|
||||
default_fmt="[%(asctime)s %(levelname)s %(filename)s:%(lineno)d] %(message)s"
|
||||
default_level = logging.DEBUG
|
||||
|
||||
log_warn = logging.WARNING
|
||||
log_erro = logging.ERROR
|
||||
|
||||
logging.basicConfig(format=default_fmt, level=default_level)
|
||||
|
||||
def get_logger(name=None, level=default_level, fmt=None):
|
||||
logger = logging.getLogger(name)
|
||||
logger.setLevel(level)
|
||||
if fmt is not None:
|
||||
for h in logger.handlers:
|
||||
h.setFormatter(logging.Formatter(fmt))
|
||||
return logger
|
||||
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Loading…
Reference in New Issue