CAEN DT5742采集程序单核心100%CPU无响应问题求助

问题背景 #

使用CAEN DT5742 16通道数字化仪开展测量工作，基于CAENPy（CAENDigitizer库的Python封装）开发多进程程序，通过步进电机控制激光扫描区域，同时读取模拟读出板数据。程序运行整体正常，但会随机出现无响应情况，且两个进程会占用单个核心100%的CPU资源。通过py-spy性能分析定位到问题根源在CAENPy库的_GetNumEvents方法。

核心采集代码 #

def read_and_save_events(self, max_num_events: int = 1):
    """Reads a specified number of events from the digitizer.

    Arguments
    ---------
    max_num_events: int, default 1
        Number of events to read.

    Returns
    -------
    nevts: int
        Number of events read.
    """
    nevts: int = 0
    data = []
    retries = 0
    while retries < MAX_RETRIES:
        retries += 1
        try:
            with self.device:
                self.log.info("Reading %d events...", max_num_events)
                while nevts < max_num_events:
                    time.sleep(0.05)
                    waveforms = self.get_waveforms()
                    current_nevts = len(waveforms)
                    nevts += current_nevts
                    data += waveforms
                    self.log.info(
                        "Read %d out of %d events...", nevts, max_num_events
                    )
            break
        except RuntimeError:
            self.log.error("Encountered error during read. Retrying...")
            self.hard_reset(self._device_id)
            self.close()
            self.device = CAEN_DT5742_Digitizer(self._device_id)
            self.init()
            time.sleep(RETRY_TIMEOUT)
    else:
        self.log.error("Too many retries, aborting read...")

    if self._save_path is None:
        self.log.warning("No save path specified, waveforms not saved!")
        return 0

    # Disentangle data and save to file
    df = pd.DataFrame(data)
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    data_file = os.path.join(self._save_path, f"waveforms_{timestamp}.h5")
    self.curr_savefile = data_file

    with pd.HDFStore(data_file, "w") as store:
        for channel in df.columns:
            channel_df = []
            for eventid, event in enumerate(df[channel]):
                event_df = pd.DataFrame(event)
                for column in event_df.columns:
                    col = pd.Series(
                        event_df[column].values,
                        name=f"{eventid}_{column.split()[0]}",
                    )
                    channel_df.append(col)
            channel_df = pd.concat(channel_df, axis=1)
            store.put(channel, channel_df)

修复可能性判断 #

1. 开源库场景：如果CAENPy是开源项目，可直接查看_GetNumEvents的实现代码，排查是否存在死循环、未处理异常或无超时的阻塞逻辑，针对性修复后重新编译安装库；若该方法是对CAENDigitizer官方C库的封装，可查阅官方文档确认是否存在已知Bug，尝试升级库版本，或向官方提交Issue反馈问题。
2. 闭源库场景：无法直接修改底层库的Bug，只能通过上层程序逻辑规避或监控处理。

规避与监控方案 #

1. 进程级CPU占用监控与重启 #

通过psutil库监控采集进程的CPU使用率，当进程持续100%占用核心超过设定阈值（如30秒）时，强制终止并重启该进程，同时重新初始化数字化仪设备。

示例逻辑：

import psutil
import time

def monitor_worker_process(pid, threshold=30):
    while True:
        try:
            proc = psutil.Process(pid)
            cpu_percent = proc.cpu_percent(interval=1)
            if cpu_percent >= 99:
                # 多次确认避免误判
                consecutive_high = 0
                for _ in range(5):
                    if proc.cpu_percent(interval=1) >=99:
                        consecutive_high +=1
                if consecutive_high >=5:
                    proc.terminate()
                    proc.wait()
                    # 执行采集进程重启逻辑
                    restart_acquisition_process()
                    break
            time.sleep(5)
        except psutil.NoSuchProcess:
            break

2. 采集逻辑添加心跳超时检测 #

在采集循环中添加心跳检测，若长时间未读取到新事件（如10秒），主动抛出异常触发重置流程：

修改内层采集循环：

import datetime

NO_EVENT_TIMEOUT = 10  # 10秒无事件触发超时
last_event_time = datetime.datetime.now()

while nevts < max_num_events:
    time.sleep(0.05)
    waveforms = self.get_waveforms()
    current_nevts = len(waveforms)
    
    if current_nevts > 0:
        last_event_time = datetime.datetime.now()
    
    nevts += current_nevts
    data += waveforms
    self.log.info("Read %d out of %d events...", nevts, max_num_events)
    
    # 检查超时
    elapsed = (datetime.datetime.now() - last_event_time).total_seconds()
    if elapsed > NO_EVENT_TIMEOUT:
        self.log.error(f"No events received for {elapsed}s, triggering reset")
        raise RuntimeError("Event acquisition timeout")

3. 封装采集方法的超时控制 #

将get_waveforms()调用放到子线程中，主线程设置超时等待，超时则终止子线程并抛出异常：

import threading

def _get_waveforms_with_timeout(self, timeout=10):
    result = []
    exception = None
    
    def target():
        nonlocal result, exception
        try:
            result = self.get_waveforms()
        except Exception as e:
            exception = e
    
    thread = threading.Thread(target=target)
    thread.start()
    thread.join(timeout=timeout)
    
    if thread.is_alive():
        raise RuntimeError("Waveform read timed out")
    if exception is not None:
        raise exception
    return result

在采集循环中替换原waveforms = self.get_waveforms()为waveforms = self._get_waveforms_with_timeout()。

4. 强化设备重置逻辑 #

在触发重置时，不仅重启设备对象，还可调用设备的硬件重置接口（若有），确保设备回到初始状态后再重新初始化。

内容的提问来源于stack exchange，提问作者TensaZangetsu