Merge branch 'master' into pr/minrk/3029

diff --cc tornado/platform/asyncio.py
index 3863e78f2e43ad49ad2b78e105012b0f962cb04a,5248c1f9513c0c89da409acb814a13d90c55d327..2e95b801a303b79e1361ea240997cfacac618b88
--- 1/tornado/platform/asyncio.py
--- 2/tornado/platform/asyncio.py
+++ b/tornado/platform/asyncio.py
@@@ -573,70 -646,18 +613,78 @@@ class SelectorThread
          self._writers[fd] = functools.partial(callback, *args)
          self._wake_selector()
  
-     def remove_reader(self, fd: "_FileDescriptorLike") -> None:
-         del self._readers[fd]
+     def remove_reader(self, fd: "_FileDescriptorLike") -> bool:
+         try:
+             del self._readers[fd]
+         except KeyError:
+             return False
          self._wake_selector()
+         return True
  
-     def remove_writer(self, fd: "_FileDescriptorLike") -> None:
-         del self._writers[fd]
+     def remove_writer(self, fd: "_FileDescriptorLike") -> bool:
+         try:
+             del self._writers[fd]
+         except KeyError:
+             return False
          self._wake_selector()
-     def remove_reader(self, fd: "_FileDescriptorLike") -> None:
+         return True
 +
 +
 +class AddThreadSelectorEventLoop(asyncio.AbstractEventLoop):
 +    """Wrap an event loop to add implementations of the ``add_reader`` method family.
 +
 +    Instances of this class start a second thread to run a selector.
 +    This thread is completely hidden from the user; all callbacks are
 +    run on the wrapped event loop's thread.
 +
 +    This class is used automatically by Tornado; applications should not need
 +    to refer to it directly.
 +
 +    It is safe to wrap any event loop with this class, although it only makes sense
 +    for event loops that do not implement the ``add_reader`` family of methods
 +    themselves (i.e. ``WindowsProactorEventLoop``)
 +
 +    Closing the ``AddThreadSelectorEventLoop`` also closes the wrapped event loop.
 +
 +    """
 +
 +    # This class is a __getattribute__-based proxy. All attributes other than those
 +    # in this set are proxied through to the underlying loop.
 +    MY_ATTRIBUTES = {
 +        "_real_loop",
 +        "_selector",
 +        "add_reader",
 +        "add_writer",
 +        "close",
 +        "remove_reader",
 +        "remove_writer",
 +    }
 +
 +    def __getattribute__(self, name: str) -> Any:
 +        if name in AddThreadSelectorEventLoop.MY_ATTRIBUTES:
 +            return super().__getattribute__(name)
 +        return getattr(self._real_loop, name)
 +
 +    def __init__(self, real_loop: asyncio.AbstractEventLoop) -> None:
 +        self._real_loop = real_loop
 +        self._selector = SelectorThread(real_loop)
 +
 +    def close(self) -> None:
 +        self._selector.close()
 +        self._real_loop.close()
 +
 +    def add_reader(
 +        self, fd: "_FileDescriptorLike", callback: Callable[..., None], *args: Any
 +    ) -> None:
 +        return self._selector.add_reader(fd, callback, *args)
 +
 +    def add_writer(
 +        self, fd: "_FileDescriptorLike", callback: Callable[..., None], *args: Any
 +    ) -> None:
 +        return self._selector.add_writer(fd, callback, *args)
 +
-     def remove_writer(self, fd: "_FileDescriptorLike") -> None:
++    def remove_reader(self, fd: "_FileDescriptorLike") -> bool:
 +        return self._selector.remove_reader(fd)
 +
++    def remove_writer(self, fd: "_FileDescriptorLike") -> bool:
 +        return self._selector.remove_writer(fd)

diff --cc tornado/test/asyncio_test.py
index 59825e8f22d9895525a9bba81f112c00f393c757,348c0cebb1b45a8c8bed44232c7c40a47f68e09c..f6b719bb43c3841cfe4e79f640a8b2bdd9a1d954
--- 1/tornado/test/asyncio_test.py
--- 2/tornado/test/asyncio_test.py
+++ b/tornado/test/asyncio_test.py
@@@ -108,11 -105,6 +106,11 @@@ class AsyncIOLoopTest(AsyncTestCase)
              42,
          )
  
-         loop = AddThreadSelectorEventLoop(asyncio.get_event_loop())
 +    def test_add_thread_close_idempotent(self):
++        loop = AddThreadSelectorEventLoop(asyncio.get_event_loop())  # type: ignore
 +        loop.close()
 +        loop.close()
 +
  
  class LeakTest(unittest.TestCase):
      def setUp(self):