如何在Python后台监控文件描述符？（非线程/子进程方案）

需要在Python后台监控libgpiod的文件描述符，尝试以下两种方案均遇到问题，希望找到除真实线程、子进程外的可行方案：

方案一：使用select.poll + asyncio.run_in_executor #

import gpiod
import asyncio
import select

async def read():
    name = "GPOUT"
    line = chip.get_line(INPUT)
    line.request(consumer=name, type=gpiod.LINE_REQ_EV_BOTH_EDGES, flags=gpiod.LINE_REQ_FLAG_BIAS_PULL_UP)
    fd = line.event_get_fd()
    poll = select.poll()
    poll.register(fd)
    loop = asyncio.get_event_loop()
    while True:
        await loop.run_in_executor(None, poll.poll, None)
        event = line.event_read()
        print(event)

def main():
    asyncio.run(read())
    while True:
        input = input("Enter something: ")
        print(input)

main()

问题：执行时在poll.poll调用处阻塞，主线程的输入循环永远无法执行。

方案二：使用asyncio.add_reader #

import gpiod
import asyncio

def readCb(line):
    print(f"{line.consumer}: {line.event_read()}")

def read():
    name = "GPOUT"
    line = chip.get_line(INPUT)
    line.request(consumer=name, type=gpiod.LINE_REQ_EV_BOTH_EDGES, flags=gpiod.LINE_REQ_FLAG_BIAS_PULL_UP)
    fd = line.event_get_fd()
    loop = asyncio.get_event_loop()
    loop.add_reader(fd, readCb, line)
    loop.run_forever()

def main():
    read()
    while True:
        input = input("Enter something: ")
        print(input)

main()

问题：read函数中loop.run_forever()直接阻塞主线程，输入循环无法执行；若将read改为async函数并通过asyncio.run()调用，会抛出错误：

File "/usr/lib/python3.9/asyncio/base_events.py", line 578, in _check_running
    raise RuntimeError('This event loop is already running')

可行方案：整合asyncio事件循环与同步输入 #

核心思路是将GPIO监控和用户输入都纳入asyncio事件循环的调度，同步输入操作通过线程池避免阻塞事件循环。

import gpiod
import asyncio
from concurrent.futures import ThreadPoolExecutor

# 替换为你的GPIO编号和芯片路径
INPUT_GPIO = 17
GPIO_CHIP_PATH = "gpiochip0"

async def monitor_gpio(chip):
    consumer_name = "GPOUT"
    line = chip.get_line(INPUT_GPIO)
    
    # 请求GPIO事件监听
    line.request(
        consumer=consumer_name,
        type=gpiod.LINE_REQ_EV_BOTH_EDGES,
        flags=gpiod.LINE_REQ_FLAG_BIAS_PULL_UP
    )
    fd = line.event_get_fd()
    loop = asyncio.get_running_loop()

    def handle_gpio_event():
        try:
            event = line.event_read()
            print(f"GPIO事件: {event}")
        except Exception as e:
            print(f"处理GPIO事件出错: {e}")

    # 注册文件描述符监听
    loop.add_reader(fd, handle_gpio_event)
    
    try:
        # 保持任务运行，直到被取消
        await asyncio.Future()
    finally:
        # 清理资源
        loop.remove_reader(fd)
        line.release()

async def handle_user_input():
    # 用线程池处理同步input，避免阻塞事件循环
    with ThreadPoolExecutor(max_workers=1) as executor:
        loop = asyncio.get_running_loop()
        while True:
            user_input = await loop.run_in_executor(
                executor, input, "Enter something: "
            )
            print(f"输入内容: {user_input}")

async def main():
    chip = gpiod.Chip(GPIO_CHIP_PATH)
    try:
        # 同时运行GPIO监控和输入处理任务
        await asyncio.gather(monitor_gpio(chip), handle_user_input())
    finally:
        chip.close()

if __name__ == "__main__":
    asyncio.run(main())

方案说明 #

1. 事件循环整合：通过asyncio.gather同时运行GPIO监控和用户输入两个异步任务，事件循环会自动调度两者，不会互相阻塞。
2. 同步输入处理：input是同步阻塞操作，通过run_in_executor将其放到线程池执行，避免阻塞事件循环的主逻辑。
3. GPIO监听优化：使用loop.add_reader直接监听GPIO的文件描述符，事件触发时自动调用回调函数，无需手动poll。
4. 资源清理：在任务结束时正确释放GPIO line和chip资源，避免资源泄漏。

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