_monitor_fs does not seem to hold reference to callback

Run the supplied reproducer and observe whether events stop after gc.collect. Then trace the _monitor_fs entry point and its callback handling; done means the callback and its referenced FileAndHandle remain usable and read events continue after garbage collection.

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()