DBUS通讯中出现NoReply异常

Question

我正在开发一个异步通信脚本，它将充当反应本机应用程序和另一个代理之间的中间人。为此，我使用 python 和 DBUS 来实现两者之间的通信。

为了实现这一点，我们创建了两个进程，一个用于 BLE，另一个用于与代理通信。如果客服人员立即回复（通过非阻塞呼叫），则通信始终按预期进行。对于我们附加信号以持续监视更新状态的情况，此错误大多数时候会在过程中的随机点发生：

dbus.exceptions.DBusException: org.freedesktop.DBus.Error.NoReply: Did not receive a reply. Possible causes include: the remote application did not send a reply, the message bus security policy blocked the reply, the reply timeout expired, or the network connection was broken.

我已经分别测试了 BLE 流程和代理流程，它们按预期工作。

我目前怀疑它可能与系统总线上的消息“崩溃”或某些竞争条件有关，但我们不确定如何验证这一点。

关于可能导致此问题的原因或如何避免它有什么建议吗？

为了完整起见，我附加了处理与代理通信的类的简化版本。

import multiprocessing
from enum import Enum
import dbus
import dbus.mainloop.glib
from dbus.proxies import ProxyObject
from gi.repository import GLib
from omegaconf import DictConfig
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)


class ClientUpdateStatus(Enum):
    SUCCESS = 0
    PENDING = 1
    IN_PROGRESS = 2
    FAILED = 3

class DBUSManager:

    GLIB_LOOP = GLib.MainLoop()
    COMMUNICATION_QUEUE = multiprocessing.Queue()
    
    def __init__(self, config: DictConfig) -> None:
        
        dbus_system_bus = dbus.SystemBus()
        
        dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
        self._config = config
        
        self._dbus_object = dbus_system_bus.get_object(self._config['dbus_interface'],
                                                            self._config['dbus_object_path'], introspect=False)

    def get_version(self) -> str:

        version = self._dbus_object.GetVersion("clientSimulator", dbus_interface=self._config['dbus_interface'])
        return version

    def check_for_update(self) -> str:
        update_version = self._dbus_object.CheckForUpdate("clientSimulator",
                                                               dbus_interface=self._config['dbus_interface'])
        return update_version

    def run_update(self) -> ClientUpdateStatus:
        raw_status = self._dbus_object.ExecuteUpdate(dbus_interface=self._config['dbus_interface'])
        
        update_status = ClientUpdateStatus(raw_status)
        
        # Launch listening process
        signal_update_proc = multiprocessing.Process(target=DBUSManager.start_listener_process,
                                                     args=(self._dbus_object, self._config['dbus_interface'],))
        signal_update_proc.start()

        while True:
            raw_status = DBUSManager.COMMUNICATION_QUEUE.get()
            update_status = ClientUpdateStatus(raw_status)
            
            if ClientUpdateStatus.SUCCESS == update_status:
                break

        signal_update_proc.join()

        return update_status

    @staticmethod
    def start_listener_process(dbus_object: ProxyObject, dbus_interface: str) -> None:
       
        dbus_object.connect_to_signal("UpdateStatusChanged", DBUSManager.status_change_handler,
                                           dbus_interface=dbus_interface)
        # Launch loop to acquire signals
        DBUSManager.GLIB_LOOP.run()  # This listening loop exits on GLIB_LOOP.quit()

    @staticmethod
    def status_change_handler(new_status: int) -> None:
        DBUSManager.COMMUNICATION_QUEUE.put(new_status)
        if ClientUpdateStatus.SUCCESS == ClientUpdateStatus(new_status):
            DBUSManager.GLIB_LOOP.quit()

Answer 1

在此阶段，我建议执行

dbus-monitor

来查看代理和 BLE 是否对请求做出正确反应。

Answer 2

也许为了将来帮助其他人，我还没有找到解决方案，但至少找到了解决它的方法。我们在项目的各个地方都遇到过这个问题，不要问我为什么，通常有帮助的是总是重新实例化所有需要的 dbus 对象。因此，我们总是会重新实例化它们，而不是拥有一个具有系统总线变量

self._system_bus = dbus.SystemBus()

或接口变量

self._manager_interface = dbus.Interface(proxy_object, "org.freedesktop.DBus.ObjectManager")

的单个类。

如果有人知道问题是什么，我很高兴听到。