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

Question

我想使用kqueue来监视文件中的变化。我可以看到如何以线程的方式使用select.kqueue()。

我正在寻找一种在异步中使用它的方法。我可能漏掉了一些很明显的东西。我知道python使用macos上用于异步的kqueue。我很高兴任何解决方案只在使用kqueue选择器时才能工作。

到目前为止，我看到做到这一点的唯一方法是创建一个线程，以从另一个线程持续执行kqueue.control()，然后将事件注入到asyncio.loop.call_soon_threadsafe()中。我觉得应该有更好的办法。

Answer 1

您可以使用读者()将kqueue objet中的FD作为读取器添加到控制循环中。然后，控制循环将通知您已准备好收集事件。

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

• select.kqueue.control是一种一次性的方法，它首先改变监视器并等待新事件的到来。因为我们从来不想阻止它，所以必须将这两个操作分割为一个非阻塞调用来修改监视器，另一个是非阻塞调用来收集所产生的事件。
• 因为我们从来不想阻止，所以不能使用超时。这可以用asyncio.wait_for()重新实现。

有更有效的方法来编写这个文件，但下面是一个示例，说明如何用异步方法(此处名为select.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)