test: cover poll notification support

Add a pollable file to the test filesystem and exercise the new
Operations.poll() and notify_poll() APIs.

The test opens the synthetic file, starts a userspace poll(2) waiter,
and waits until the filesystem has received and stored a PollHandle.
It then triggers readiness through the existing setxattr command
channel. The filesystem marks the file ready, calls notify_poll() and
the test verifies that the blocked poller wakes with POLLPRI.

This covers the notification path from the low-level FUSE poll callback,
through the Python PollHandle wrapper, to fuse_lowlevel_notify_poll().

Signed-off-by: Christopher Obbard <christopher.obbard@linaro.org>

Author: obbardc

test/test_fs.py

@@ -21,6 +21,7 @@
 import logging
 import multiprocessing
 import os
+import select
 import stat
 import threading
 import time
@@ -34,6 +35,7 @@
     FileInfo,
     FUSEError,
     InodeT,
+    PollHandle,
     ReaddirToken,
     RequestContext,
 )
@@ -118,6 +120,38 @@ def test_notify_store(testfs):
         assert not fs_state.read_called
 
 
+def test_notify_poll(testfs):
+    (mnt_dir, fs_state) = testfs
+    path = os.path.join(mnt_dir, 'pollable')
+
+    with open(path, 'rb', buffering=0) as fh:
+        poller = select.poll()
+        poller.register(fh.fileno(), select.POLLPRI)
+
+        events = []
+
+        def poll_wait():
+            events.extend(poller.poll(5000))
+
+        thread = threading.Thread(target=poll_wait)
+        thread.start()
+
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline and not fs_state.poll_handle_received:
+            time.sleep(0.01)
+
+        assert fs_state.poll_called
+        assert fs_state.poll_handle_received
+        assert not events
+
+        pyfuse3.setxattr(mnt_dir, 'command', b'poll_ready')
+        thread.join(5)
+        assert not thread.is_alive()
+        assert events
+        assert events[0][0] == fh.fileno()
+        assert events[0][1] & select.POLLPRI
+
+
 def test_entry_timeout(testfs):
     (mnt_dir, fs_state) = testfs
     fs_state.entry_timeout = 1
@@ -175,11 +209,17 @@ def __init__(self, cross_process):
         self.hello_name = b"message"
         self.hello_inode = cast(InodeT, pyfuse3.ROOT_INODE + 1)
         self.hello_data = b"hello world\n"
+        self.poll_name = b"pollable"
+        self.poll_inode = cast(InodeT, pyfuse3.ROOT_INODE + 2)
+        self.poll_handle: PollHandle | None = None
         self.status = cross_process
         self.lookup_cnt = 0
         self.status.getattr_called = False
         self.status.lookup_called = False
         self.status.read_called = False
+        self.status.poll_called = False
+        self.status.poll_handle_received = False
+        self.status.poll_ready = False
         self.status.entry_timeout = 99999
         self.status.attr_timeout = 99999
 
@@ -191,6 +231,9 @@ async def getattr(self, inode: InodeT, ctx: RequestContext | None = None) -> Ent
         elif inode == self.hello_inode:
             entry.st_mode = stat.S_IFREG | 0o644
             entry.st_size = len(self.hello_data)
+        elif inode == self.poll_inode:
+            entry.st_mode = stat.S_IFREG | 0o644
+            entry.st_size = 0
         else:
             raise pyfuse3.FUSEError(errno.ENOENT)
 
@@ -212,17 +255,25 @@ async def forget(self, inode_list):
             if inode == self.hello_inode:
                 self.lookup_cnt -= 1
                 assert self.lookup_cnt >= 0
+            elif inode == self.poll_inode:
+                pass
             else:
                 assert inode == pyfuse3.ROOT_INODE
 
     async def lookup(
         self, parent_inode: InodeT, name: bytes, ctx: RequestContext
     ) -> EntryAttributes:
-        if parent_inode != pyfuse3.ROOT_INODE or name != self.hello_name:
+        if parent_inode != pyfuse3.ROOT_INODE:
             raise pyfuse3.FUSEError(errno.ENOENT)
-        self.lookup_cnt += 1
+
         self.status.lookup_called = True
-        return await self.getattr(self.hello_inode, ctx)
+        if name == self.hello_name:
+            self.lookup_cnt += 1
+            return await self.getattr(self.hello_inode, ctx)
+        if name == self.poll_name:
+            return await self.getattr(self.poll_inode, ctx)
+
+        raise pyfuse3.FUSEError(errno.ENOENT)
 
     async def opendir(self, inode, ctx):
         if inode != pyfuse3.ROOT_INODE:
@@ -232,23 +283,53 @@ async def opendir(self, inode, ctx):
 
     async def readdir(self, fh: FileHandleT, start_id: int, token: ReaddirToken) -> None:
         assert fh == pyfuse3.ROOT_INODE
-        if start_id == 0:
-            pyfuse3.readdir_reply(token, self.hello_name, await self.getattr(self.hello_inode), 1)
-        return
+        entries = (
+            (self.hello_name, self.hello_inode),
+            (self.poll_name, self.poll_inode),
+        )
+
+        for idx, (name, inode) in enumerate(entries):
+            if idx < start_id:
+                continue
+            if not pyfuse3.readdir_reply(token, name, await self.getattr(inode), idx + 1):
+                break
 
     async def open(self, inode, flags, ctx):
-        if inode != self.hello_inode:
+        if inode not in (self.hello_inode, self.poll_inode):
             raise pyfuse3.FUSEError(errno.ENOENT)
         if flags & os.O_RDWR or flags & os.O_WRONLY:
             raise pyfuse3.FUSEError(errno.EACCES)
         # For simplicity, we use the inode as file handle
         return FileInfo(fh=FileHandleT(inode))
 
     async def read(self, fh, off, size):
+        if fh == self.poll_inode:
+            return b''
+
         assert fh == self.hello_inode
         self.status.read_called = True
         return self.hello_data[off : off + size]
 
+    async def poll(
+        self,
+        inode: InodeT,
+        fh: FileHandleT,
+        poll_handle: PollHandle | None,
+        ctx: RequestContext,
+    ) -> int:
+        assert inode == self.poll_inode
+        assert fh == self.poll_inode
+
+        self.status.poll_called = True
+        if poll_handle is not None:
+            self.poll_handle = poll_handle
+            self.status.poll_handle_received = True
+
+        if self.status.poll_ready:
+            return select.POLLPRI
+
+        return 0
+
     async def setxattr(self, inode, name, value, ctx):
         if inode != pyfuse3.ROOT_INODE or name != b'command':
             raise FUSEError(errno.ENOTSUP)
@@ -267,6 +348,14 @@ async def setxattr(self, inode, name, value, ctx):
 
         elif value == b'terminate':
             pyfuse3.terminate()
+
+        elif value == b'poll_ready':
+            self.status.poll_ready = True
+            if self.poll_handle is None:
+                raise FUSEError(errno.EINVAL)
+            pyfuse3.notify_poll(self.poll_handle)
+            self.poll_handle = None
+
         else:
             raise FUSEError(errno.EINVAL)