如何在异步IO中使用kqueue进行文件监视？

Question

我想使用kqueue监视文件的更改。我可以看到如何以线程方式使用select.kqueue（）

我正在寻找一种将其用于asyncio的方法。我可能错过了一些非常明显的东西。我知道python使用kqueue for asyncio on macos。我很高兴任何解决方案只在使用kqueue选择器时才起作用

到目前为止，我能看到的唯一方法是创建一个线程，从另一个线程继续kqueue.control()，然后用asyncio.loop.call_soon_threadsafe()注入事件。我觉得应该有更好的办法

Answer 1

您可以使用loop.add_reader()将来自kqueue对象的FD作为读卡器添加到控制循环中。然后，控制循环将通知您事件已准备好收集

这样做有两个特点，对于熟悉kqueue的人来说可能很奇怪：

• select.kqueue.control是一种一次性方法，它首先更改监视器并等待新事件的到来。因为我们不希望它阻塞，所以必须将这两个操作拆分为一个非阻塞调用来修改监视器，然后再拆分为第二个非阻塞调用来收集结果事件
• 因为我们不想阻塞，所以不能使用超时。这可以通过asyncio.wait_for()重新实现

有更有效的编写方法，但下面是一个如何用异步方法（此处命名为kqueue_control）完全替换select.kqueue.control的示例：

async def kqueue_control(kqueue: select.kqueue,
                         changes: Optional[Iterable[select.kevent]],
                         max_events: int,
                         timeout: Optional[int]):

    def receive_result():
        try:
            # Events are ready to collect; fetch them but do not block
            results = kqueue.control(None, max_events, 0)
        except Exception as ex:
            future.set_exception(ex)
        else:
            future.set_result(results)
        finally:
            loop.remove_reader(kqueue.fileno())
            
    # If this call is non-blocking then just execute it
    if timeout == 0 or max_events == 0:
        return kqueue.control(changes, max_events, 0)
    
    # Apply the changes, but DON'T wait for events
    kqueue.control(changes, 0)
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    loop.add_reader(kqueue.fileno(), receive_result)
    if timeout is None:
        return await future
    else:
        return await asyncio.wait_for(future, timeout)