如何让基于GDBus的Python D-Bus服务在空闲时自动退出？

问题背景 #

我用PyGObject绑定写了个基于GDBus的D-Bus服务，因为是Python开发且带有若干依赖，内存占用约30MiB。这个服务仅在用户使用对应客户端应用时才需要，多数时间处于空闲状态。它被配置为systemd的D-Bus激活服务，可按需启动，但我希望它能在一段时间未被使用时自动退出。原本以为systemd会自带该功能，但实际并不支持——不过systemd自身的服务实现了类似逻辑。另外我找到一个无竞争的空闲退出GDBus服务的PoC，但它是用C语言编写的。

实现思路 #

核心逻辑是监听D-Bus的所有外部活动（方法调用、属性访问、信号订阅等），每次捕获到活动就重置空闲计时器；当计时器超时后，标记服务进入关闭状态、拒绝新请求，最后安全退出主循环。这种方式能避免“刚触发退出就有新请求进来”的竞争问题。

Python代码实现示例 #

import gi
gi.require_version('Gio', '2.0')
gi.require_version('GLib', '2.0')
from gi.repository import Gio, GLib

class IdleAwareDBusService:
    def __init__(self):
        self.idle_timeout = 300  # 空闲超时时间，单位：秒（这里设为5分钟）
        self.timeout_source_id = None
        self.is_shutting_down = False
        self.dbus_connection = None

        # 初始化D-Bus连接并导出服务
        self._setup_dbus_connection()
        # 启动空闲计时器
        self._reset_idle_timer()

    def _setup_dbus_connection(self):
        # 替换为你的服务连接逻辑，比如连接系统总线或会话总线
        self.dbus_connection = Gio.bus_get_sync(Gio.BusType.SYSTEM, None)
        # 添加消息过滤器，捕获所有外部发来的D-Bus消息
        self.dbus_connection.add_filter(self._handle_dbus_activity)

        # 这里添加你的服务对象导出逻辑，示例：
        # self.service_object = Gio.DBusExportedObject.wrap_jsongenerator(...)
        # self.service_object.export(self.dbus_connection, "/com/example/MyService")

    def _handle_dbus_activity(self, connection, message, user_data):
        # 忽略服务自身发出的消息
        if message.get_sender() == connection.get_unique_name():
            return Gio.DBusMessageFilterResult.CONTINUE
        
        # 若服务正在关闭，拒绝新请求并返回错误
        if self.is_shutting_down:
            error_reply = message.new_error_reply(
                Gio.dbus_error_quark(),
                "org.freedesktop.DBus.Error.ServiceUnknown",
                "Service is shutting down due to inactivity"
            )
            connection.send_message(error_reply, None)
            return Gio.DBusMessageFilterResult.HANDLED
        
        # 有外部活动，重置空闲计时器
        self._reset_idle_timer()
        return Gio.DBusMessageFilterResult.CONTINUE

    def _reset_idle_timer(self):
        # 移除旧计时器（如果存在）
        if self.timeout_source_id is not None:
            GLib.source_remove(self.timeout_source_id)
        # 启动新的超时计时器
        self.timeout_source_id = GLib.timeout_add_seconds(
            self.idle_timeout,
            self._on_idle_timeout_triggered
        )

    def _on_idle_timeout_triggered(self):
        # 标记服务进入关闭状态
        self.is_shutting_down = True
        # 移除计时器
        if self.timeout_source_id is not None:
            GLib.source_remove(self.timeout_source_id)
        # 退出主循环，结束服务
        GLib.MainLoop.get_default().quit()
        return False  # 不再重复触发计时器

    def run(self):
        main_loop = GLib.MainLoop()
        main_loop.run()

if __name__ == "__main__":
    service = IdleAwareDBusService()
    service.run()

关键细节说明 #

• 消息过滤器：覆盖所有外部D-Bus请求类型，确保不会遗漏任何用户操作导致计时器未重置。
• 关闭状态防护：在超时触发后立即标记关闭状态，拒绝新请求，避免竞争条件。
• 计时器重置：每次有外部活动就刷新超时时间，保证服务仅在最后一次使用后开始倒计时。

配合systemd优化 #

虽然systemd不支持自动空闲退出，但可以在服务配置文件中添加Restart=on-demand，让服务退出后，下次有客户端请求时自动重新激活，保持按需启动的原有逻辑。

示例systemd服务文件（my-dbus-service.service）：

[Unit]
Description=My Idle-Aware D-Bus Service

[Service]
Type=dbus
BusName=com.example.MyService
ExecStart=/usr/bin/python3 /path/to/your/service.py
Restart=on-demand

[Install]
WantedBy=multi-user.target

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