You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

为何Python方法调用时长呈周期性非均匀分布?

问题:C++调用Python方法出现双耗时类别及周期性延迟的原因

我在C++中调用Python脚本,流程是加载模块、实例化类,之后调用类的方法约100万次,用std::chrono::high_resolution_clock和std::chrono::duration_cast测量每次调用耗时。结果发现方法调用时长分为两类,约50微秒和约100微秒(见图1),且时长呈现周期性特征(见图2,仅展示部分数据)。请问导致该现象的原因可能是什么?


被调用的Python方法代码

def data_received(self, data, time_us, first_sample_of_experiment):
    self.eeg_data_index += 1

    c3 = data[4]
    others = [data[20], data[22], data[24], data[26]]
    filtered = self.average(c3, others)

    self.data.append(filtered)
    if len(self.data) > 20:
        self.data.pop(0)

    signal = self.peak_detection.thresholding_algo(c3)
    if signal == 0 and not self.peak_over:
        self.peak_over = True

    peak = signal != 0

    if peak and self.peak_over:
        self.peak_over = False
        self.peak_at = self.eeg_data_index
        self.peaks_detected += 1

    return [charge_event, charge_event]

峰值检测类代码(改编自实时时序数据峰值检测实现)

import numpy as np


class RealtimePeakDetection:
    def __init__(self, array, lag, threshold, influence):
        self.y = list(array)
        self.length = len(self.y)
        self.lag = lag
        self.threshold = threshold
        self.influence = influence

        self.signals = [0] * len(self.y)
        self.filteredY = np.array(self.y).tolist()
        self.avgFilter = [0] * len(self.y)
        self.stdFilter = [0] * len(self.y)
        self.avgFilter[self.lag - 1] = np.mean(self.y[0:self.lag]).tolist()
        self.stdFilter[self.lag - 1] = np.std(self.y[0:self.lag]).tolist()

    def thresholding_algo(self, new_value):
        i = len(self.y) - 1
        self.y.append(new_value)

        self.signals += [0]
        self.filteredY += [0]
        self.avgFilter += [0]
        self.stdFilter += [0]

        if len(self.y) > self.length:
            self.y.pop(0)
        if len(self.signals) > self.length:
            self.signals.pop(0)
        if len(self.filteredY) > self.length:
            self.filteredY.pop(0)
        if len(self.avgFilter) > self.length:
            self.avgFilter.pop(0)
        if len(self.stdFilter) > self.length:
            self.stdFilter.pop(0)

        if abs(self.y[i] - self.avgFilter[i - 1]) > (self.threshold * self.stdFilter[i - 1]):

            if self.y[i] > self.avgFilter[i - 1]:
                self.signals[i] = 1
            else:
                self.signals[i] = -1

            self.filteredY[i] = self.influence * self.y[i] + (1 - self.influence) * self.filteredY[i - 1]
            self.avgFilter[i] = np.mean(self.filteredY[(i - self.lag):i])
            self.stdFilter[i] = np.std(self.filteredY[(i - self.lag):i])
        else:
            self.signals[i] = 0
            self.filteredY[i] = self.y[i]
            self.avgFilter[i] = np.mean(self.filteredY[(i - self.lag):i])
            self.stdFilter[i] = np.std(self.filteredY[(i - self.lag):i])

        return self.signals[i]

环境与构建信息

  • 内核:Linux PREEMPT_RT 5.15.55-rt48
  • C++程序:ROS节点,RTPRIO优先级为-98
  • 构建命令:colcon build --packages-select <ros package> --cmake-args -DCMAKE_BUILD_TYPE=Release
  • 编译器:GNU 9.4.0

CMakeLists.txt代码

cmake_minimum_required(VERSION 3.8)
project(data_processor)

if (CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
    add_compile_options(-Wall -Wextra -Wpedantic)
endif ()

find_package(ament_cmake REQUIRED)
find_package(rclcpp REQUIRED)
find_package(std_msgs REQUIRED)
find_package(mtms_interfaces REQUIRED)
find_package(fpga_interfaces REQUIRED)

set(MATLAB_FIND_DEBUG true)
# MATLAB
find_package(Matlab)
if (Matlab_FOUND)
    # Fixes runtime error "error while loading shared libraries: libMatlabDataArray.so: cannot open shared object file: No such file or directory"
    SET(CMAKE_SKIP_BUILD_RPATH FALSE)
    SET(CMAKE_BUILD_WITH_INSTALL_RPATH FALSE)
    SET(CMAKE_INSTALL_RPATH "${CMAKE_INSTALL_PREFIX}/lib64")
    SET(CMAKE_INSTALL_RPATH_USE_LINK_PATH TRUE)
    SET(CMAKE_INSTALL_RPATH "${CMAKE_INSTALL_PREFIX}/lib64")

    set(LD_LIBRARY_PATH ${LD_LIBRARY_PATH}:${Matlab_ROOT_DIR}/extern/bin/glnxa64:${Matlab_ROOT_DIR}/sys/os/glnxa64)
    include_directories(${Matlab_ROOT_DIR}/extern/include/)
    link_directories(${Matlab_ROOT_DIR}/extern/bin/glnxa64)
else ()
    message(STATUS "MATLAB NOT FOUND")
endif ()

add_executable(
        data_processor
        src/data_processor.cpp
        src/processor.cpp
        src/headers/processor.h
        src/python_processor.cpp
        src/matlab_processor.cpp
        src/compiled_matlab_processor.cpp
        src/matlab_processor_interface.cpp
        src/headers/matlab_processor.h
        src/headers/python_processor.h
        src/headers/compiled_matlab_processor.h
        src/headers/scheduling_utils.h
        src/headers/scheduling_utils.cpp
        src/headers/matlab_processor_interface.h
        src/headers/fpga_event.h
        src/headers/data_processor.h
        src/headers/matlab_helpers.h
        src/matlab_helpers.cpp)

target_include_directories(data_processor
        PUBLIC
        $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/lib>
        $<INSTALL_INTERFACE:lib>)

ament_target_dependencies(data_processor rclcpp std_msgs mtms_interfaces fpga_interfaces)

if (Matlab_FOUND)
    # MATLAB, Linker to libMatlabEngine in link_directories
    target_link_libraries(data_processor MatlabDataArray)
    target_link_libraries(data_processor MatlabEngine)
endif ()

# Python
find_package(PythonLibs)
if (PYTHONLIBS_FOUND)
    message(STATUS "Python found")
    include_directories(${PYTHON_INCLUDE_DIRS})
    target_link_libraries(data_processor ${PYTHON_LIBRARIES})
else ()
    message(STATUS "Python not found")
endif ()

install(TARGETS
        data_processor
        DESTINATION lib/${PROJECT_NAME}
        )

install(
        DIRECTORY launch
        DESTINATION share/${PROJECT_NAME}
)

if (BUILD_TESTING)
    find_package(ament_lint_auto REQUIRED)
    ament_lint_auto_find_test_dependencies()
endif ()

ament_package()

耗时分布与周期性特征图

图1:图1:调用耗时分布

图2:图2:调用耗时周期性特征


可能的原因分析

  • Python GIL调度开销:CPython的全局解释器锁(GIL)会定期释放并重新获取,即使单线程调用Python,内部的垃圾回收线程等也会触发GIL切换。每次切换都会带来额外耗时,且GIL调度有固定周期,和你观察到的周期性延迟匹配。
  • Numpy操作的内存波动:峰值检测中频繁调用np.mean和np.std,这类函数会临时分配内存,内存分配器(如glibc ptmalloc)在重复处理小内存块时,可能出现周期性的缓存失效或分配策略切换,导致耗时翻倍。另外,列表与Numpy数组的转换(.tolist())也会带来不稳定的开销。
  • 实时调度的潜在干扰:虽然用了PREEMPT_RT内核并设置了高优先级,但ROS节点依赖的rclcpp内部线程、系统守护进程仍可能抢占进程;同时Python解释器本身并非实时安全,其内部的垃圾回收、GIL调度等操作不受实时调度控制,会引发周期性延迟。
  • 列表操作的缓存局部性变化:self.data每次append+pop(0)是O(n)操作(头部删除需移动所有元素),当列表元素处于CPU缓存中时操作更快,缓存失效则需从主存加载,百万次重复操作下,缓存命中/失效会呈现周期性规律。
  • 垃圾回收周期性触发:Python对象的引用计数维护会在达到阈值时触发垃圾回收,该操作会带来额外耗时,且垃圾回收有固定触发周期,对应耗时的周期性峰值。

内容的提问来源于stack exchange,提问作者Alqio

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.18 23:01:07