_monitor_fs does not seem to hold reference to callback

Führe den bereitgestellten Reproducer aus und beobachte, ob Ereignisse nach gc.collect ausbleiben. Verfolge dann den Einstiegspunkt _monitor_fs und die Verarbeitung seines Callbacks; abgeschlossen ist die Aufgabe, wenn der Callback und sein referenziertes FileAndHandle weiterhin verwendbar sind und Leseereignisse nach der Garbage Collection weiterhin eintreffen.

uvloop version: 0.17.0
python version: 3.8.10
platform: linux

The bug can be reproduced with the below script. I call _monitor_fs on the loop, passing in a function callback that itself has a reference to the file I'm reading as well as the handle to the _monitor_fs. I set a timer to later call gc.collect and when that happens, my object gets deleted and I stop getting read events.

This does seem to be a bug because the loop itself should have reference to my callback which has a reference to my object and that object has a reference to the file and handle, so GC should not be collecting them.

import functools
import gc
import sys
import tempfile
import typing

import uvloop
import os.path
import subprocess

class FileAndHandle:
    def __init__(self, f: typing.IO):
        self.f = f
        self.handle = None

    def __del__(self):
        print('goodbye cruel world')

class FileWatcher:
    _proc: typing.Optional[subprocess.Popen]

    def __init__(self):
        self._loop = uvloop.new_event_loop()
        self._proc = None
        self._last_read_event = 0

    def _do_read(self, file_and_handle: FileAndHandle, _os_fn: bytes, _ev_enum: int):
        self._last_read_event = self._loop.time()
        print(f'read: {repr(file_and_handle.f.read())}')

    def _check_read_timer(self):
        t = self._loop.time()
        if t - self._last_read_event > 3:
            print('stopped getting read events')
            self._proc.kill()
            self._loop.stop()
        else:
            self._loop.call_later(1, self._check_read_timer)

    def _start_watching(self, f_path: str) -> None:
        fah = FileAndHandle(open(f_path))
        fah.handle = self._loop._monitor_fs(f_path, functools.partial(self._do_read, fah))

    def main(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_dir:
            pth = os.path.join(tmp_dir, "file.txt")
            open(pth, 'wt').close()
            self._proc = subprocess.Popen([sys.executable, '-c', f'''
import time
with open("{pth}", "wt") as f:
    while True:
        time.sleep(2)
        f.write("hello\\n")
        f.flush()
'''])
            self._start_watching(pth)
            self._loop.call_later(3, self._check_read_timer)
            self._loop.call_later(15, gc.collect)
            self._loop.run_forever()

if __name__ == '__main__':
    FileWatcher().main()