地瓜官网
地瓜开发者社区
地瓜官网
地瓜开发者社区

Flexible模式

Flexible模式下请使用 hbm_infer.hbm_rpc_session_flexible 模块。

接口及参数

  1. 全局方法：init_server

    def init_server( host: str, username: str = "root", password: str = "", remote_root: str = "/map/hbm_infer/", ) -> HbmRpcServer:

    构造 HbmRpcServer 对象。

    • 参数
    参数说明
    host开发板ip地址。
    username板端用户名，默认值为 "root" 。
    remote_root板端暂存文件的根目录路径，默认值为 "/map/hbm_infer/" 。
    password登录开发板的密码，默认为 "" ，代表无需使用密码。
    • 返回值

    HbmRpcServer 对象实例。

  2. 全局方法：deinit_server

    def deinit_server(hbm_rpc_server: HbmRpcServer) -> None:

    清理板端server文件。

    • 参数
    参数说明
    hbm_rpc_serverHbmRpcServer 对象实例。

  3. 全局方法：init_hbm

    def init_hbm( local_hbm_path: str, hbm_rpc_server: HbmRpcServer, ) -> HbmHandle:

    构造 HbmHandle 对象。

    • 参数
    参数说明
    local_hbm_path本地HBM文件路径。
    hbm_rpc_serverHbmRpcServer 对象，可通过调用 init_server 接口获得。
    • 返回值

    HbmHandle 对象实例。

  4. 全局方法：deinit_hbm

    def deinit_hbm(hbm_handle: HbmHandle) -> None:

    清理板端HBM文件。

    • 参数
    参数说明
    hbm_handleHbmHandle 对象实例。

  5. HbmRpcSession成员方法：__init__

    def __init__( self, hbm_handle: HbmHandle, hbm_rpc_server: HbmRpcServer, frame_timeout: Optional[int] = 90, server_timeout: Optional[int] = 30, with_profile: Optional[bool] = False, debug: Optional[bool] = False, compress_option: Optional[str] = "NONE", ) -> None:

    初始化HbmRpcSession对象。

    • 参数
    参数说明
    hbm_handleHbmHandle 对象，可通过调用 init_hbm 接口获得。
    hbm_rpc_serverHbmRpcServer 对象，可通过调用 init_server 接口获得。
    frame_timeoutgRPC 通信的单帧超时时间，以秒为单位，默认值为 90 。
    server_timeout服务端超时时间，以分钟为单位，默认值为 30 ，超时后服务端自动退出并清理 除日志外 的其他文件。
    with_profile是否开启推理各阶段时间统计，默认值为 False 。
    debug开启调试模式，保留更多日志信息，默认值为 False 。
    compress_option开启 gRPC 压缩功能，可选值有 "IN" 、 "INOUT" 及 "NONE" ，分别代表请求数据帧启用压缩、请求和响应数据帧都启用压缩、不启用压缩，默认值为 "NONE" 。
注解

压缩功能是软件处理，因此开启后通常出现推理延时变大的情况，压缩功能优化主要体现在降低网络负载和提高吞吐，并且压缩质量依赖输入输出数据的相关性，对于浮点输入输出一般不建议开启，对于图像输入或者分割输出等可尝试开启。

  1. HbmRpcSession成员方法：get_input_info

    def get_input_info(self) -> dict:

    获取模型输入信息。

    • 返回值

    字典形式的模型输入信息描述，具体格式可参考如下示例：

    { "input_name0": { "valid_shape": [1, 3, 224, 224], "tensor_type": "DATA_TYPE_S8", "quanti_type": "QUANTI_TYPE_SCALE", "quantizeAxis": 0, "scale_data": [0.006861070170998573], "zero_point_data": [0] }, ... }

  2. HbmRpcSession成员方法：get_output_info

    def get_output_info(self) -> dict:

    获取模型输出信息。

    • 返回值

    字典形式的模型输出信息描述，格式与 get_input_info 返回值一致。

  3. HbmRpcSession成员方法：show_input_output_info

    def show_input_output_info(self) -> None:

    打印模型输入输出信息。

  4. HbmRpcSession成员方法：__call__

    def __call__(self, data) -> (dict | list):

    执行模型推理。

    • 参数
    参数说明
    data模型输入，字典类型。 key 为输入张量名称， value 为输入张量，支持 torch.Tensor 和 numpy.ndarray 两种格式。需要注意： 输入数据需要与模型输入信息保持一致（包括名称、数量、形状以及数据元素类型等） 。
    • 返回值

    模型输出，字典类型。 key 为输出张量名称， value 为输出张量，其类型与模型输入保持一致。

  5. HbmRpcSession成员方法：close_server

    def close_server(self) -> None:

    关闭服务端，清理服务端资源。

  6. HbmRpcSession成员方法：get_profile

    def get_profile(self) -> dict:

    获取推理各阶段耗时统计信息，需要 with_profile 参数为 True ，否则报错。返回值为字典类型，其参考格式如下：

    { // 帧总延时（ms） "frame_duration": { "avg": 6, "min": 6, "max": 6, }, // gRPC发送输入帧到返回输出帧的总延时（ms） "sd2rv_duration": { "avg": 5, "min": 5, "max": 5, }, // 网络通信耗时（ms） "commu_duration": { "avg": 4, "min": 4, "max": 4, }, // 板端总耗时（ms） "board_duration": { "avg": 1, "min": 1, "max": 1, }, // 板端纯推理耗时（ms） "infer_duration": { "avg": 0.5, "min": 0.5, "max": 0.5, }, // 板端前处理耗时（ms） "prepr_duration": { "avg": 0.3, "min": 0.3, "max": 0.3, }, // 板端后处理耗时（ms） "pospr_duration": { "avg": 0.2, "min": 0.2, "max": 0.2, }, }

  7. HbmRpcSession成员方法：get_profile_last_frame

    def get_profile_last_frame(self) -> dict:

    获取最近一帧推理各阶段耗时统计信息，需要 with_profile 参数为 True ，否则报错。返回值为字典类型，其参考格式如下：

    { // 帧总延时（ms） "frame_duration": 12, // gRPC发送输入帧到返回输出帧的总延时（ms） "sd2rv_duration": 10, // 网络通信耗时（ms） "commu_duration": 6, // 板端总耗时（ms） "board_duration": 4, // 板端纯推理耗时（ms） "infer_duration": 2, // 板端前处理耗时（ms） "prepr_duration": 0.5, // 板端后处理耗时（ms） "pospr_duration": 0.5, }

使用示例

import torch import multiprocessing as mp from hbm_infer.hbm_rpc_session_flexible import HbmRpcSession, init_server, deinit_server, init_hbm, deinit_hbm def single_session_entry(rpc_server, hbm_handle, run_epoch): # 创建session sess = HbmRpcSession( hbm_rpc_server=rpc_server, hbm_handle=hbm_handle ) # 打印模型输入输出信息 sess.show_input_output_info() # 准备输入数据 input_data = { 'img': torch.ones((1, 3, 224, 224), dtype=torch.int8) } # 执行推理并返回结果 for i in range(run_epoch): output_data = sess(input_data) print([output_data[k].shape for k in output_data]) # 关闭server sess.close_server() def run_hbm_infer(num_process=8, run_epoch=20): # 创建server rpc_server = init_server( host=<available_ip> ) # 加载HBM文件 hbm_handle = init_hbm( hbm_rpc_server=rpc_server, local_hbm_path=<local_hbm_path> ) # 多进程推理 processes = list() for i in range(num_process): p = mp.Process(target=single_session_entry, args=(rpc_server, hbm_handle, run_epoch)) processes.append(p) p.start() for p in processes: p.join() # 清理板端server文件 deinit_server(rpc_server) # 清理板端HBM文件 deinit_hbm(hbm_handle) if __name__ == "__main__": run_hbm_infer()