feat: add aimrt_py benchmark publisher and subscriber applications (#28)

* feat: add aimrt_py benchmark publisher and subscriber applications

Introduce new benchmark publisher and subscriber modules, enabling performance testing with custom configuration. Includes necessary YAML configuration files and scripts for simulation of message publishing and receiving.

* style: correct message size formatting

Update the log message to use "bytes" for clarity and consistency in reporting metrics.

* feat: add python benchmark example and update dependencies

Introduce an aimrt_py benchmark example for Python testing. Replace boost dependency with the standalone asio library to reduce overall dependencies and enhance configuration options for zenoh and mqtt.

* feat: add aimrt_py channel benchmark example

Introduce a new benchmark example for aimrt_py to enhance Python benchmarking capabilities and provide clearer insights into performance metrics.

* format code

* style: format loss rate output

Remove the percentage symbol from the loss rate display for a cleaner presentation.

* docs: update release notes for v0.9.0

Clarify new features, config options, and dependency changes, including the addition of the aimrt_py channel benchmark example and removal of boost dependencies in favor of asio.

* docs: add benchmark example to Python interfaces section

Include a new link for the pb_chn_bench example to enhance the documentation and provide users with more comprehensive usage scenarios.

* docs: add README for protobuf channel benchmark example

Provide instructions on setting up and running a Python channel benchmark using protobuf with HTTP backend. Include details on core modules, configuration, and execution steps to enhance usability and understanding.

* docs: update installation instructions to link to quick start guide

Enhance user experience by providing direct access to the installation guide for the `aimrt_py` package, improving clarity and reducing search time for new users.

* docs: correct grpc reference in README

Update the section title to accurately reflect the protocol being used, enhancing clarity for users.

document/sphinx-cn/release_notes/v0_9_0.md

@@ -8,9 +8,9 @@
 - mqtt 新增配置项以支持加密传输；
 - 新增了第三方库 asio，runtime::core 不再引用 boost，改为引用独立的 asio 库，以减轻依赖；
 
-
 **次要修改**：
 - 缩短了一些 examples 的文件路径长度；
 - 修复了一些轻微问题；
 - 优化代码结构，移动代码 src/runtime/common/net 至新位置 src/common/net；
 - 升级 jsoncpp 至 1.9.6 版本以优化一些 cmake 问题；
+- 新增了 aimrt_py channel benchmark 示例；

document/sphinx-cn/tutorials/examples/examples_py.md

@@ -7,6 +7,7 @@ AimRT 提供了以下 Python 接口使用示例：
   - {{ '[helloworld]({}/src/examples/py/helloworld)'.format(code_site_root_path_url) }}
   - {{ '[pb_chn]({}/src/examples/py/pb_chn)'.format(code_site_root_path_url) }}
   - {{ '[pb_rpc]({}/src/examples/py/pb_rpc)'.format(code_site_root_path_url) }}
+  - {{ '[pb_chn_bench]({}/src/examples/py/pb_chn_bench)'.format(code_site_root_path_url) }}
 
 关于这些示例的说明：
 - 每个示例都有自己独立的 readme 文档，详情请点击示例链接进入后查看；

src/examples/cpp/pb_chn/module/benchmark_subscriber_module/benchmark_subscriber_module.cc

@@ -191,7 +191,7 @@ void BenchmarkSubscriberModule::Evaluate() const {
   AIMRT_INFO(R"str(Benchmark plan {} completed, report:
 frequency: {} hz
 topic number: {}
-msg size: {} byte
+msg size: {} bytes
 msg count per topic: {}
 send count : {}
 recv count: {}

src/examples/py/helloworld/README.md

@@ -16,7 +16,7 @@
 
 
 运行方式（linux）：
-- 安装 `aimrt_py` 包；
+- [安装 `aimrt_py` 包](../../../../document/sphinx-cn/tutorials/quick_start/installation_py.md)；
 - 直接运行本目录下[start_examples_py_helloworld_app_mode.sh](./start_examples_py_helloworld_app_mode.sh)脚本启动进程；
 - 键入`ctrl-c`停止进程；
 
@@ -45,7 +45,7 @@
 
 
 运行方式（linux）：
-- 安装 `aimrt_py` 包；
+- [安装 `aimrt_py` 包](../../../../document/sphinx-cn/tutorials/quick_start/installation_py.md)；
 - 直接运行本目录下[start_examples_py_helloworld_registration_mode.sh](./start_examples_py_helloworld_registration_mode.sh)脚本启动进程；
 - 键入`ctrl-c`停止进程；
 

src/examples/py/parameter/README.md

@@ -14,7 +14,7 @@
 
 
 运行方式（linux）：
-- 安装 `aimrt_py` 包；
+- [安装 `aimrt_py` 包](../../../../document/sphinx-cn/tutorials/quick_start/installation_py.md)；
 - 直接运行本目录下[start_examples_py_parameter_app.sh](./start_examples_py_parameter_app.sh)脚本启动进程；
 - 键入`ctrl-c`停止进程；
 

src/examples/py/pb_chn/README.md

@@ -23,7 +23,7 @@
 
 
 运行方式（linux）：
-- 安装 `aimrt_py` 包；
+- [安装 `aimrt_py` 包](../../../../document/sphinx-cn/tutorials/quick_start/installation_py.md)；
 - 运行本目录下的[build_examples_py_pb_chn.sh](./build_examples_py_pb_chn.sh)脚本，生成协议桩代码文件；
   - 如果本地没有 protoc 或者 protoc 版本小于 3.20，请安装或升级 protoc，或直接修改脚本中的 `protoc_cmd` 变量指向合适的路径；
 - 运行本目录下的[start_examples_py_pb_chn_http_sub.sh](./start_examples_py_pb_chn_http_sub.sh)脚本，启动 subscriber；
@@ -60,7 +60,7 @@
 
 
 运行方式（linux）：
-- 安装 `aimrt_py` 包；
+- [安装 `aimrt_py` 包](../../../../document/sphinx-cn/tutorials/quick_start/installation_py.md)；
 - 运行本目录下的[build_examples_py_pb_chn.sh](./build_examples_py_pb_chn.sh)脚本，生成协议桩代码文件；
   - 如果本地没有 protoc 或者 protoc 版本小于 3.20，请安装或升级 protoc，或直接修改脚本中的 `protoc_cmd` 变量指向合适的路径；
 - 运行本目录下的[start_examples_py_pb_chn_ros2_sub.sh](./start_examples_py_pb_chn_ros2_sub.sh)脚本，启动 subscriber；

src/examples/py/pb_chn_bench/README.md

@@ -0,0 +1,34 @@
+# protobuf channel benchmark
+
+
+一个基于 protobuf 协议与 http 后端的 python channel benchmark 示例，演示内容包括：
+- 如何在 python 中使用 protobuf 协议作为 channel 传输协议；
+- 如何基于 aimrt_py 注册模块的方式使用 Channel publish 和 subscribe 功能；
+- 如何使用 http 类型的 channel 后端；
+
+核心代码：
+- [benchmark.proto](../../../protocols/example/benchmark.proto)
+- [benchmark_publisher_module.py](./benchmark_publisher_module.py)
+- [benchmark_publisher_app.py](./benchmark_publisher_app.py)
+- [benchmark_subscriber_module.py](./benchmark_subscriber_module.py)
+- [benchmark_subscriber_app.py](./benchmark_subscriber_app.py)
+
+配置文件：
+- [benchmark_publisher_cfg.yaml](./cfg/benchmark_publisher_cfg.yaml)
+- [benchmark_subscriber_cfg.yaml](./cfg/benchmark_subscriber_cfg.yaml)
+
+
+运行方式（linux）：
+- [安装 `aimrt_py` 包](../../../../document/sphinx-cn/tutorials/quick_start/installation_py.md)；
+- 运行本目录下的[build_examples_py_pb_chn_bench.sh](./build_examples_py_pb_chn_bench.sh)脚本，生成协议桩代码文件；
+  - 如果本地没有 protoc 或者 protoc 版本小于 3.20，请安装或升级 protoc，或直接修改脚本中的 `protoc_cmd` 变量指向合适的路径；
+- 运行本目录下的[start_benchmark_subscriber.sh](./start_benchmark_subscriber.sh)脚本，启动 subscriber；
+- 在新终端里运行本目录下的[start_benchmark_publisher.sh](./start_benchmark_publisher.sh)脚本，启动 publisher；
+- Benchmark 运行结束后会输出 benchmark 结果并自动结束进程；
+
+
+说明：
+- 此示例创建了以下两个模块：
+  - `BenchmarkPublisherModule`：会在启动后根据配置好的 bench plan 向指定的 test_topic_xx 中发布类型为 `BenchmarkMsg` 的消息，每个 bench plan 前后会发送一个 `BenchmarkSignal` 类型的消息通知 subscriber 当前的 benchmark 状态，所有 bench plan 结束后会发送一个 `BenchmarkSignal` 类型的消息通知 subscriber 当前的 benchmark 结束；
+  - `BenchmarkSubscriberModule`：会订阅 channel 中的 test_topic_xx 消息，并统计接收到的消息的延迟分布；
+- 此示例使用 http 类型的 channel 后端进行通信，请确保相关端口未被占用；

src/examples/py/pb_chn_bench/benchmark_publisher_app.py

@@ -0,0 +1,61 @@
+# Copyright (c) 2024 The AimRT Authors.
+# AimRT is licensed under Mulan PSL v2.
+
+import argparse
+import signal
+import sys
+import threading
+
+import aimrt_py
+import benchmark_publisher_module
+
+global_aimrt_core = None
+
+
+def signal_handler(sig, _):
+    global global_aimrt_core
+
+    if (global_aimrt_core and (sig == signal.SIGINT or sig == signal.SIGTERM)):
+        global_aimrt_core.Shutdown()
+        return
+
+    sys.exit(0)
+
+
+def main():
+    parser = argparse.ArgumentParser(description='Example helloworld registration mode.')
+    parser.add_argument('--cfg_file_path', type=str, default="", help='config file path')
+    args = parser.parse_args()
+
+    signal.signal(signal.SIGINT, signal_handler)
+    signal.signal(signal.SIGTERM, signal_handler)
+
+    print("AimRT start.")
+
+    aimrt_core = aimrt_py.Core()
+
+    global global_aimrt_core
+    global_aimrt_core = aimrt_core
+
+    module = benchmark_publisher_module.BenchmarkPublisher()
+    aimrt_core.RegisterModule(module)
+
+    core_options = aimrt_py.CoreOptions()
+    core_options.cfg_file_path = args.cfg_file_path
+    aimrt_core.Initialize(core_options)
+
+    thread = threading.Thread(target=aimrt_core.Start)
+    thread.start()
+
+    while thread.is_alive():
+        thread.join(1.0)
+
+    aimrt_core.Shutdown()
+
+    global_aimrt_core = None
+
+    print("AimRT exit.")
+
+
+if __name__ == '__main__':
+    main()

src/examples/py/pb_chn_bench/benchmark_publisher_module.py

@@ -0,0 +1,234 @@
+# Copyright (c) 2024 The AimRT Authors.
+# AimRT is licensed under Mulan PSL v2.
+
+import os
+import random
+import signal
+import string
+import threading
+import time
+
+import aimrt_py
+import benchmark_pb2
+import yaml
+
+
+class BenchmarkPublisher(aimrt_py.ModuleBase):
+    def __init__(self):
+        super().__init__()
+        self.core = aimrt_py.CoreRef()
+        self.logger = aimrt_py.LoggerRef()
+        self.max_topic_number = 0
+        self.bench_plans = []
+        self.run_flag = True
+        self.stop_sig = threading.Event()
+        self.shutdown_delay = 1  # seconds
+        self.bench_interval = 1  # seconds
+        self.publish_complete_event = threading.Event()
+
+    def Info(self) -> aimrt_py.ModuleInfo:
+        info = aimrt_py.ModuleInfo()
+        info.name = "BenchmarkPublisherModule"
+        return info
+
+    def Initialize(self, core: aimrt_py.CoreRef) -> bool:
+        assert isinstance(core, aimrt_py.CoreRef)
+
+        self.core = core
+        self.logger = self.core.GetLogger()
+
+        aimrt_py.info(self.logger, "Module initializing ...")
+
+        try:
+            # read config
+            file_path = self.core.GetConfigurator().GetConfigFilePath()
+            with open(file_path, "r") as f:
+                cfg_node = yaml.safe_load(f)
+
+            self.max_topic_number = cfg_node.get("max_topic_number", 0)
+
+            if "bench_plans" in cfg_node and isinstance(cfg_node["bench_plans"], list):
+                for bench_plan_node in cfg_node["bench_plans"]:
+                    bench_plan = {
+                        "channel_frq": bench_plan_node["channel_frq"],
+                        "msg_size": bench_plan_node["msg_size"],
+                        "topic_number": bench_plan_node["topic_number"],
+                        "msg_count": bench_plan_node["msg_count"]
+                    }
+
+                    if bench_plan["topic_number"] > self.max_topic_number:
+                        raise ValueError(f"Bench plan topic number ({bench_plan['topic_number']}) "
+                                         f"is greater than max topic number ({self.max_topic_number})")
+
+                    self.bench_plans.append(bench_plan)
+
+            aimrt_py.info(self.logger, f"Module config: max_topic_number={self.max_topic_number}, "
+                                       f"bench_plans={self.bench_plans}")
+
+            # controller
+            self.publish_control_executor = self.core.GetExecutorManager().GetExecutor("publish_control_executor")
+            if not self.publish_control_executor:
+                raise RuntimeError("Get executor 'publish_control_executor' failed.")
+
+            # signal publisher
+            signal_topic = "benchmark_signal"
+            self.signal_publisher = self.core.GetChannelHandle().GetPublisher(signal_topic)
+            if not self.signal_publisher:
+                raise RuntimeError(f"Get publisher for topic '{signal_topic}' failed.")
+
+            ret = aimrt_py.RegisterPublishType(self.signal_publisher, benchmark_pb2.BenchmarkSignal)
+            if not ret:
+                raise RuntimeError(f"Register publish type for topic '{signal_topic}' failed.")
+
+            # message publishers
+            self.publisher_wrapper_vec = []
+            for ii in range(self.max_topic_number):
+                executor_name = f"publish_executor_{ii}"
+                executor = self.core.GetExecutorManager().GetExecutor(executor_name)
+                if not executor:
+                    raise RuntimeError(f"Get executor '{executor_name}' failed.")
+
+                topic_name = f"test_topic_{ii}"
+                publisher = self.core.GetChannelHandle().GetPublisher(topic_name)
+                if not publisher:
+                    raise RuntimeError(f"Get publisher for topic '{topic_name}' failed.")
+
+                ret = aimrt_py.RegisterPublishType(publisher, benchmark_pb2.BenchmarkMessage)
+                if not ret:
+                    raise RuntimeError(f"Register publish type for topic '{topic_name}' failed.")
+
+                self.publisher_wrapper_vec.append({
+                    "publish_executor": executor,
+                    "publisher": publisher
+                })
+
+        except Exception as e:
+            aimrt_py.error(self.logger, f"Initialize failed: {e}")
+            return False
+
+        aimrt_py.info(self.logger, "Init succeeded")
+        return True
+
+    def Start(self) -> bool:
+        try:
+            aimrt_py.info(self.logger, "Module starting ...")
+            self.publish_control_executor.Execute(self.MainLoop)
+
+        except Exception as e:
+            aimrt_py.error(self.logger, f"Start failed: {e}")
+            return False
+
+        aimrt_py.info(self.logger, "Start succeeded")
+        return True
+
+    def Shutdown(self) -> bool:
+        try:
+            aimrt_py.info(self.logger, "Module shutting down ...")
+
+            self.run_flag = False
+            self.publish_complete_event.set()
+            self.stop_sig.wait()
+        except Exception as e:
+            aimrt_py.error(self.logger, f"Shutdown failed: {e}")
+            return False
+
+        aimrt_py.info(self.logger, "Shutdown succeeded")
+        return True
+
+    def MainLoop(self) -> None:
+        try:
+            aimrt_py.info(self.logger, "Start Bench.")
+
+            # warm up
+            for ii in range(10):
+                begin_signal = benchmark_pb2.BenchmarkSignal()
+                begin_signal.status = benchmark_pb2.BenchmarkStatus.WarmUp
+                aimrt_py.Publish(self.signal_publisher, begin_signal)
+                time.sleep(0.1)
+                aimrt_py.info(self.logger, f"Warm up {ii}")
+
+            # benchmark
+            for ii, bench_plan in enumerate(self.bench_plans):
+                if not self.run_flag:
+                    break
+
+                print(f"Start bench plan {ii}")
+                self.StartSinglePlan(ii, bench_plan)
+                print(f"End bench plan {ii}")
+
+                time.sleep(self.bench_interval)
+
+            self.ShutdownPeer()
+
+        except Exception as e:
+            aimrt_py.error(self.logger, f"Exit MainLoop with exception: {e}")
+
+        self.stop_sig.set()
+
+    def StartSinglePlan(self, plan_id: int, plan: dict) -> None:
+        # publish start signal
+        begin_signal = benchmark_pb2.BenchmarkSignal()
+        begin_signal.status = benchmark_pb2.BenchmarkStatus.Begin
+        begin_signal.bench_plan_id = plan_id
+        begin_signal.topic_number = plan['topic_number']
+        begin_signal.send_num = plan['msg_count']
+        begin_signal.message_size = plan['msg_size']
+        begin_signal.send_frequency = plan['channel_frq']
+
+        aimrt_py.info(self.logger, f"Publish benchmark start signal, data: \n{begin_signal}")
+        aimrt_py.Publish(self.signal_publisher, begin_signal)
+
+        self.publish_complete_event.clear()
+        self.completed_tasks = 0
+        self.total_tasks = plan['topic_number']
+
+        time.sleep(1)
+
+        # publish topic
+        for ii in range(plan['topic_number']):
+            publish_executor = self.publisher_wrapper_vec[ii]["publish_executor"]
+            publish_executor.Execute(lambda index=ii: self.PublishTask(self.publisher_wrapper_vec[index], plan))
+
+        self.publish_complete_event.wait()
+
+        # publish end signal
+        end_signal = benchmark_pb2.BenchmarkSignal()
+        end_signal.status = benchmark_pb2.BenchmarkStatus.End
+        end_signal.bench_plan_id = plan_id
+        aimrt_py.info(self.logger, f"Publish benchmark end signal, data: {end_signal}")
+        aimrt_py.Publish(self.signal_publisher, end_signal)
+
+    def PublishTask(self, publisher_wrapper: dict, plan: dict) -> None:
+        publisher = publisher_wrapper['publisher']
+        msg = benchmark_pb2.BenchmarkMessage()
+        msg.data = self.GenerateRandomString(plan['msg_size'])
+
+        send_count = 0
+        sleep_time = 1 / plan['channel_frq']
+
+        while send_count < plan['msg_count'] and self.run_flag:
+            msg.seq = send_count
+            msg.timestamp = int(time.time() * 1e9)  # ns
+
+            aimrt_py.Publish(publisher, msg)
+
+            send_count += 1
+            time.sleep(sleep_time)
+
+        with threading.Lock():
+            self.completed_tasks += 1
+            if self.completed_tasks == self.total_tasks:
+                self.publish_complete_event.set()
+
+    def ShutdownPeer(self) -> None:
+        shutdown_peer_signal = benchmark_pb2.BenchmarkSignal()
+        shutdown_peer_signal.status = benchmark_pb2.BenchmarkStatus.ShutdownPeer
+        aimrt_py.info(self.logger, f"Publish benchmark shutdown peer signal, data: {shutdown_peer_signal}")
+        aimrt_py.Publish(self.signal_publisher, shutdown_peer_signal)
+
+        time.sleep(self.shutdown_delay)
+        os.kill(os.getpid(), signal.SIGINT)
+
+    @staticmethod
+    def GenerateRandomString(length: int) -> bytes:
+        return ''.join(random.choices(string.ascii_letters + string.digits, k=length)).encode()

src/examples/py/pb_chn_bench/benchmark_subscriber_app.py

@@ -0,0 +1,61 @@
+# Copyright (c) 2024 The AimRT Authors.
+# AimRT is licensed under Mulan PSL v2.
+
+import argparse
+import signal
+import sys
+import threading
+
+import aimrt_py
+import benchmark_subscriber_module
+
+global_aimrt_core = None
+
+
+def signal_handler(sig, _):
+    global global_aimrt_core
+
+    if (global_aimrt_core and (sig == signal.SIGINT or sig == signal.SIGTERM)):
+        global_aimrt_core.Shutdown()
+        return
+
+    sys.exit(0)
+
+
+def main():
+    parser = argparse.ArgumentParser(description='Example benchmark subscriber application.')
+    parser.add_argument('--cfg_file_path', type=str, default="", help='config file path')
+    args = parser.parse_args()
+
+    signal.signal(signal.SIGINT, signal_handler)
+    signal.signal(signal.SIGTERM, signal_handler)
+
+    print("AimRT start.")
+
+    aimrt_core = aimrt_py.Core()
+
+    global global_aimrt_core
+    global_aimrt_core = aimrt_core
+
+    module = benchmark_subscriber_module.BenchmarkSubscriber()
+    aimrt_core.RegisterModule(module)
+
+    core_options = aimrt_py.CoreOptions()
+    core_options.cfg_file_path = args.cfg_file_path
+    aimrt_core.Initialize(core_options)
+
+    thread = threading.Thread(target=aimrt_core.Start)
+    thread.start()
+
+    while thread.is_alive():
+        thread.join(1.0)
+
+    aimrt_core.Shutdown()
+
+    global_aimrt_core = None
+
+    print("AimRT exit.")
+
+
+if __name__ == '__main__':
+    main()

src/examples/py/pb_chn_bench/benchmark_subscriber_module.py

@@ -0,0 +1,185 @@
+# Copyright (c) 2024 The AimRT Authors.
+# AimRT is licensed under Mulan PSL v2.
+
+import os
+import signal
+import time
+from dataclasses import dataclass, field
+from typing import Dict, List
+
+import aimrt_py
+import benchmark_pb2
+import yaml
+from google.protobuf.json_format import MessageToJson
+
+
+@dataclass
+class MsgRecord:
+    recv: bool = False
+    send_timestamp: float = 0
+    recv_timestamp: float = 0
+
+
+@dataclass
+class TopicRecord:
+    topic_name: str
+    msg_record_vec: List[MsgRecord] = field(default_factory=list)
+
+
+class BenchmarkSubscriber(aimrt_py.ModuleBase):
+    def __init__(self):
+        super().__init__()
+        self.core = aimrt_py.CoreRef()
+        self.logger = aimrt_py.LoggerRef()
+        self.max_topic_number = 0
+
+        self.cur_bench_plan = benchmark_pb2.BenchmarkSignal
+        self.topic_record_map: Dict[str, TopicRecord] = {}
+
+    def Info(self) -> aimrt_py.ModuleInfo:
+        info = aimrt_py.ModuleInfo()
+        info.name = "BenchmarkSubscriberModule"
+        return info
+
+    def Initialize(self, core: aimrt_py.CoreRef) -> bool:
+        assert isinstance(core, aimrt_py.CoreRef)
+
+        self.core = core
+        self.logger = self.core.GetLogger()
+
+        aimrt_py.info(self.logger, "Module initializing ...")
+
+        try:
+            # read config
+            file_path = self.core.GetConfigurator().GetConfigFilePath()
+            with open(file_path, "r") as f:
+                cfg_node = yaml.safe_load(f)
+
+            self.max_topic_number = cfg_node.get("max_topic_number", 0)
+
+            aimrt_py.info(self.logger, f"Module config: max_topic_number={self.max_topic_number}")
+
+            # signal subscriber
+            signal_topic = "benchmark_signal"
+            self.signal_subscriber = self.core.GetChannelHandle().GetSubscriber(signal_topic)
+            if not self.signal_subscriber:
+                raise RuntimeError(f"Get subscriber for topic '{signal_topic}' failed.")
+
+            # set signal and message callbacks
+            aimrt_py.Subscribe(self.signal_subscriber, benchmark_pb2.BenchmarkSignal, self.SignalCallback)
+
+            # message subscribers
+            for ii in range(self.max_topic_number):
+                topic_name = f"test_topic_{ii}"
+                subscriber = self.core.GetChannelHandle().GetSubscriber(topic_name)
+                if not subscriber:
+                    raise RuntimeError(f"Get subscriber for topic '{topic_name}' failed.")
+                aimrt_py.Subscribe(subscriber,
+                                   benchmark_pb2.BenchmarkMessage,
+                                   lambda msg, topic_index=ii: self.MessageCallback(topic_index, msg))
+
+        except Exception as e:
+            aimrt_py.error(self.logger, f"Initialize failed: {e}")
+            return False
+
+        aimrt_py.info(self.logger, "Init succeeded")
+        return True
+
+    def Start(self) -> bool:
+        return True
+
+    def Shutdown(self) -> bool:
+        try:
+            aimrt_py.info(self.logger, "Module is shutting down...")
+        except Exception as e:
+            aimrt_py.error(self.logger, f"Shutdown failed: {e}")
+            return False
+
+        aimrt_py.info(self.logger, "Shutdown succeeded")
+        return True
+
+    def SignalCallback(self, signal_msg: benchmark_pb2.BenchmarkSignal) -> None:
+        aimrt_py.info(self.logger, f"Received signal: {MessageToJson(signal_msg)}")
+
+        match signal_msg.status:
+            case benchmark_pb2.BenchmarkStatus.WarmUp:
+                pass
+
+            case benchmark_pb2.BenchmarkStatus.Begin:
+                self.cur_bench_plan = signal_msg
+                if self.cur_bench_plan.topic_number > self.max_topic_number:
+                    raise RuntimeError(f"Topic number in bench plan is larger than max topic number: "
+                                       f"{self.cur_bench_plan.topic_number} > {self.max_topic_number}")
+
+                for ii in range(self.cur_bench_plan.topic_number):
+                    topic_name = f"test_topic_{ii}"
+                    self.topic_record_map[topic_name] = TopicRecord(topic_name=topic_name)
+                    self.topic_record_map[topic_name].msg_record_vec = [MsgRecord()
+                                                                        for _ in range(self.cur_bench_plan.send_num)]
+
+            case benchmark_pb2.BenchmarkStatus.End:
+                self.Evaluate()
+
+            case benchmark_pb2.BenchmarkStatus.ShutdownPeer:
+                aimrt_py.info(self.logger, f"Received shutdown peer signal, shutdown aimrt core now.")
+                os.kill(os.getpid(), signal.SIGINT)
+
+            case _:
+                aimrt_py.error(self.logger, f"Unknown signal status: {signal_msg.status}")
+
+    def MessageCallback(self, topic_index: int, benchmark_msg: benchmark_pb2.BenchmarkMessage) -> None:
+        recv_timestamp = time.time() * 1e9  # ns
+
+        topic_name = f"test_topic_{topic_index}"
+        self.topic_record_map[topic_name].msg_record_vec[benchmark_msg.seq].recv = True
+        self.topic_record_map[topic_name].msg_record_vec[benchmark_msg.seq].recv_timestamp = recv_timestamp
+        self.topic_record_map[topic_name].msg_record_vec[benchmark_msg.seq].send_timestamp = benchmark_msg.timestamp
+
+    def Evaluate(self) -> None:
+        # calculate metrics
+        latency_vec = []
+        for ii in range(self.cur_bench_plan.topic_number):
+            topic_name = f"test_topic_{ii}"
+            topic_record = self.topic_record_map[topic_name]
+            for jj in range(self.cur_bench_plan.send_num):
+                msg_record = topic_record.msg_record_vec[jj]
+                if not msg_record.recv:
+                    aimrt_py.error(self.logger, f"topic '{topic_name}' message seq '{jj}' not received")
+                    continue
+
+                if msg_record.recv_timestamp < msg_record.send_timestamp:
+                    aimrt_py.error(self.logger, f"topic '{topic_name}' message seq '{jj}' "
+                                   "recv timestamp is smaller than send timestamp")
+                    continue
+
+                latency_vec.append((msg_record.recv_timestamp - msg_record.send_timestamp) / 1e3)  # us
+
+        latency_vec.sort()
+        recv_count = len(latency_vec)
+
+        min_latency = latency_vec[0]
+        max_latency = latency_vec[-1]
+        p90_latency = latency_vec[int(recv_count * 0.9)]
+        p99_latency = latency_vec[int(recv_count * 0.99)]
+        p999_latency = latency_vec[int(recv_count * 0.999)]
+
+        send_count = self.cur_bench_plan.send_num * self.cur_bench_plan.topic_number
+        loss_rate = (send_count - recv_count) / send_count * 100
+        avg_latency = sum(latency_vec) / recv_count
+
+        result_str = f"Benchmark plan {self.cur_bench_plan.bench_plan_id} completed, report:"
+        result_str += f"\nfrequency: {self.cur_bench_plan.send_frequency} hz"
+        result_str += f"\ntopic number: {self.cur_bench_plan.topic_number}"
+        result_str += f"\nmsg size: {self.cur_bench_plan.message_size} bytes"
+        result_str += f"\nmsg count per topic: {self.cur_bench_plan.send_num}"
+        result_str += f"\nsend count: {send_count}"
+        result_str += f"\nrecv count: {recv_count}"
+        result_str += f"\nloss rate: {loss_rate:.2f}"
+        result_str += f"\nmin latency: {min_latency:.3f} us"
+        result_str += f"\nmax latency: {max_latency:.3f} us"
+        result_str += f"\navg latency: {avg_latency:.3f} us"
+        result_str += f"\np90 latency: {p90_latency:.3f} us"
+        result_str += f"\np99 latency: {p99_latency:.3f} us"
+        result_str += f"\np999 latency: {p999_latency:.3f} us\n"
+
+        aimrt_py.info(self.logger, result_str)

src/examples/py/pb_chn_bench/build_py_pb_chn.sh

@@ -0,0 +1,27 @@
+#!/bin/bash
+
+protoc_cmd=protoc
+protocols_dir=../../../protocols/example/
+
+if ! command -v ${protoc_cmd} &> /dev/null; then
+    echo "Can not find protoc!"
+    exit 1
+fi
+
+PROTOC_VERSION=$(${protoc_cmd} --version | awk '{print $2}')
+
+version_greater_equal() {
+    local version1=$1
+    local version2=$2
+
+    [ "$(printf '%s\n' "$version1" "$version2" | sort -V | head -n1)" != "$version1" ]
+}
+
+if version_greater_equal "$PROTOC_VERSION" "3.20"; then
+    echo "protoc version $PROTOC_VERSION."
+else
+    echo "protoc version $PROTOC_VERSION, need 3.20 at least."
+    exit 1
+fi
+
+${protoc_cmd} -I${protocols_dir} --python_out=./ ${protocols_dir}/benchmark.proto

src/examples/py/pb_chn_bench/cfg/benchmark_publisher_cfg.yaml

@@ -0,0 +1,56 @@
+# Copyright (c) 2024 The AimRT Authors.
+# AimRT is licensed under Mulan PSL v2.
+
+aimrt:
+  plugin:
+    plugins:
+      - name: net_plugin
+        path: ${AIMRT_PLUGIN_DIR}/libaimrt_net_plugin.so
+        options:
+          thread_num: 4
+          http_options:
+            listen_ip: 127.0.0.1
+            listen_port: 50081
+  log:
+    core_lvl: INFO # Trace/Debug/Info/Warn/Error/Fatal/Off
+    backends:
+      - type: console
+  executor:
+    executors:
+      - name: publish_control_executor
+        type: simple_thread
+      - name: publish_executor_0
+        type: simple_thread
+      - name: publish_executor_1
+        type: simple_thread
+      - name: publish_executor_2
+        type: simple_thread
+      - name: publish_executor_3
+        type: simple_thread
+  channel:
+    backends:
+      - type: http
+        options:
+          pub_topics_options:
+            - topic_name: "(.*)"
+              server_url_list: ["http://127.0.0.1:50080"]
+    pub_topics_options:
+      - topic_name: "(.*)"
+        enable_backends: [http]
+  module:
+    modules:
+      - name: BenchmarkPublisherModule
+        log_lvl: INFO
+
+# Module custom configuration
+BenchmarkPublisherModule:
+  max_topic_number: 4
+  bench_plans:
+    - channel_frq: 1000
+      msg_size: 512
+      topic_number: 4
+      msg_count: 5000
+    - channel_frq: 1000
+      msg_size: 4096
+      topic_number: 1
+      msg_count: 1000

src/examples/py/pb_chn_bench/cfg/benchmark_subscriber_cfg.yaml

@@ -0,0 +1,31 @@
+# Copyright (c) 2024 The AimRT Authors.
+# AimRT is licensed under Mulan PSL v2.
+
+aimrt:
+  plugin:
+    plugins:
+      - name: net_plugin
+        path: ${AIMRT_PLUGIN_DIR}/libaimrt_net_plugin.so
+        options:
+          thread_num: 4
+          http_options:
+            listen_ip: 127.0.0.1
+            listen_port: 50080
+  log:
+    core_lvl: INFO # Trace/Debug/Info/Warn/Error/Fatal/Off
+    backends:
+      - type: console
+  channel:
+    backends:
+      - type: http
+    sub_topics_options:
+      - topic_name: "(.*)"
+        enable_backends: [http]
+  module:
+    modules:
+      - name: BenchmarkSubscriberModule
+        log_lvl: INFO
+
+
+BenchmarkSubscriberModule:
+  max_topic_number: 4

src/examples/py/pb_chn_bench/start_benchmark_publisher.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+export AIMRT_PLUGIN_DIR=$(pip show aimrt_py | grep Location | awk '{print $2}')/aimrt_py
+
+python3 benchmark_publisher_app.py --cfg_file_path ./cfg/benchmark_publisher_cfg.yaml

src/examples/py/pb_chn_bench/start_benchmark_subscriber.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+export AIMRT_PLUGIN_DIR=$(pip show aimrt_py | grep Location | awk '{print $2}')/aimrt_py
+
+python3 benchmark_subscriber_app.py --cfg_file_path ./cfg/benchmark_subscriber_cfg.yaml

src/examples/py/pb_rpc/README.md

@@ -23,7 +23,7 @@
 
 
 运行方式（linux）：
-- 安装 `aimrt_py` 包；
+- [安装 `aimrt_py` 包](../../../../document/sphinx-cn/tutorials/quick_start/installation_py.md)；
 - 运行本目录下的[build_examples_py_pb_rpc.sh](./build_examples_py_pb_rpc.sh)脚本，生成协议桩代码文件；
   - 如果本地没有 protoc 或者 protoc 版本小于 3.20，请安装或升级 protoc，或直接修改脚本中的 `protoc_cmd` 变量指向合适的路径；
 - 运行本目录下的[start_examples_py_pb_rpc_http_server.sh](./start_examples_py_pb_rpc_http_server.sh)脚本，启动 RPC Server；
@@ -39,7 +39,7 @@
 
 
 
-## protobuf rpc grpcc
+## protobuf rpc grpc
 
 
 一个基于 protobuf 协议与 grpc 后端的 python rpc 示例，演示内容包括：
@@ -61,7 +61,7 @@
 
 
 运行方式（linux）：
-- 安装 `aimrt_py` 包；
+- [安装 `aimrt_py` 包](../../../../document/sphinx-cn/tutorials/quick_start/installation_py.md)；
 - 运行本目录下的[build_examples_py_pb_rpc.sh](./build_examples_py_pb_rpc.sh)脚本，生成协议桩代码文件；
   - 如果本地没有 protoc 或者 protoc 版本小于 3.20，请安装或升级 protoc，或直接修改脚本中的 `protoc_cmd` 变量指向合适的路径；
 - 运行本目录下的[start_examples_py_pb_rpc_grpc_server.sh](./start_examples_py_pb_rpc_grpc_server.sh)脚本，启动 RPC Server；
@@ -100,7 +100,7 @@
 
 
 运行方式（linux）：
-- 安装 `aimrt_py` 包；
+- [安装 `aimrt_py` 包](../../../../document/sphinx-cn/tutorials/quick_start/installation_py.md)；
 - 运行本目录下的[build_examples_py_pb_rpc.sh](./build_examples_py_pb_rpc.sh)脚本，生成协议桩代码文件；
   - 如果本地没有 protoc 或者 protoc 版本小于 3.20，请安装或升级 protoc，或直接修改脚本中的 `protoc_cmd` 变量指向合适的路径；
 - 运行本目录下的[start_examples_py_pb_rpc_ros2_server.sh](./start_examples_py_pb_rpc_ros2_server.sh)脚本，启动 RPC Server；

src/protocols/example/benchmark.proto

@@ -9,6 +9,7 @@ enum BenchmarkStatus {
   Begin = 0;
   End = 1;
   WarmUp = 2;
+  ShutdownPeer = 3;
 }
 
 message BenchmarkSignal {