import time
import math
import random
+import warnings
from inspect import isawaitable
from tornado.concurrent import (
.. testcode::
+ import asyncio
import errno
import functools
import socket
io_loop = tornado.ioloop.IOLoop.current()
io_loop.spawn_callback(handle_connection, connection, address)
- if __name__ == '__main__':
+ async def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setblocking(0)
io_loop = tornado.ioloop.IOLoop.current()
callback = functools.partial(connection_ready, sock)
io_loop.add_handler(sock.fileno(), callback, io_loop.READ)
- io_loop.start()
+ await asyncio.Event().wait()
+
+ if __name__ == "__main__":
+ asyncio.run(main())
.. testoutput::
:hide:
- By default, a newly-constructed `IOLoop` becomes the thread's current
- `IOLoop`, unless there already is a current `IOLoop`. This behavior
- can be controlled with the ``make_current`` argument to the `IOLoop`
- constructor: if ``make_current=True``, the new `IOLoop` will always
- try to become current and it raises an error if there is already a
- current instance. If ``make_current=False``, the new `IOLoop` will
- not try to become current.
+ Do not attempt to construct an `IOLoop` directly; this is deprecated
+ since Tornado 6.2. Instead, initialize the `asyncio` event loop and
+ use `IOLoop.current()` to access an `IOLoop` wrapper around the
+ current event loop.
In general, an `IOLoop` cannot survive a fork or be shared across
processes in any way. When multiple processes are being used, each
``IOLoop.configure`` method cannot be used on Python 3 except
to redundantly specify the `asyncio` event loop.
+ .. deprecated:: 6.2
+ It is deprecated to create an event loop that is "current" but not
+ currently running. This means it is deprecated to pass
+ ``make_current=True`` to the ``IOLoop`` constructor, or to create
+ an ``IOLoop`` while no asyncio event loop is running unless
+ ``make_current=False`` is used.
"""
# These constants were originally based on constants from the epoll module.
an alias for this method). ``instance=False`` is deprecated,
since even if we do not create an `IOLoop`, this method
may initialize the asyncio loop.
+
+ .. deprecated:: 6.2
+ It is deprecated to call ``IOLoop.current()`` when no `asyncio`
+ event loop is running.
"""
try:
loop = asyncio.get_event_loop()
.. versionchanged:: 5.0
This method also sets the current `asyncio` event loop.
+
+ .. deprecated:: 6.2
+ The concept of an event loop that is "current" without
+ currently running is deprecated in asyncio since Python
+ 3.10. All related functionality in Tornado is also
+ deprecated. Instead, start the event loop with `asyncio.run`
+ before interacting with it.
"""
# The asyncio event loops override this method.
raise NotImplementedError()
.. versionchanged:: 5.0
This method also clears the current `asyncio` event loop.
+ .. deprecated:: 6.2
"""
+ warnings.warn("clear_current is deprecated", DeprecationWarning)
old = IOLoop.current(instance=False)
if old is not None:
old._clear_current_hook()
def close(self, all_fds: bool = False) -> None:
if self.is_current:
- self.clear_current()
+ with warnings.catch_warnings():
+ # We can't get here unless the warning in make_current
+ # was swallowed, so swallow the one from clear_current too.
+ warnings.simplefilter("ignore", DeprecationWarning)
+ self.clear_current()
super().close(all_fds=all_fds)
def make_current(self) -> None:
+ warnings.warn(
+ "make_current is deprecated; start the event loop first",
+ DeprecationWarning,
+ )
if not self.is_current:
try:
self.old_asyncio = asyncio.get_event_loop()
class AsyncIOLoopTest(AsyncTestCase):
def get_new_ioloop(self):
- io_loop = AsyncIOLoop()
+ io_loop = AsyncIOLoop(make_current=False)
return io_loop
def test_asyncio_callback(self):
class LeakTest(unittest.TestCase):
def setUp(self):
# Trigger a cleanup of the mapping so we start with a clean slate.
- AsyncIOLoop().close()
+ AsyncIOLoop(make_current=False).close()
# If we don't clean up after ourselves other tests may fail on
# py34.
self.orig_policy = asyncio.get_event_loop_policy()
orig_count = len(IOLoop._ioloop_for_asyncio)
for i in range(10):
# Create and close an AsyncIOLoop using Tornado interfaces.
- loop = AsyncIOLoop()
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore", DeprecationWarning)
+ loop = AsyncIOLoop()
loop.close()
new_count = len(IOLoop._ioloop_for_asyncio) - orig_count
self.assertEqual(new_count, 0)
class SyncHTTPClientTest(unittest.TestCase):
def setUp(self):
- self.server_ioloop = IOLoop()
+ self.server_ioloop = IOLoop(make_current=False)
event = threading.Event()
@gen.coroutine
from tornado.ioloop import IOLoop, TimeoutError, PeriodicCallback
from tornado.log import app_log
from tornado.testing import AsyncTestCase, bind_unused_port, ExpectLog, gen_test
-from tornado.test.util import skipIfNonUnix, skipOnTravis
+from tornado.test.util import (
+ ignore_deprecation,
+ setup_with_context_manager,
+ skipIfNonUnix,
+ skipOnTravis,
+)
import typing
# automatically set as current.
class TestIOLoopCurrent(unittest.TestCase):
def setUp(self):
+ setup_with_context_manager(self, ignore_deprecation())
self.io_loop = None # type: typing.Optional[IOLoop]
IOLoop.clear_current()
class TestIOLoopCurrentAsync(AsyncTestCase):
+ def setUp(self):
+ super().setUp()
+ setup_with_context_manager(self, ignore_deprecation())
+
@gen_test
def test_clear_without_current(self):
# If there is no current IOLoop, clear_current is a no-op (but
class TestIOLoopRunSync(unittest.TestCase):
def setUp(self):
- self.io_loop = IOLoop()
+ self.io_loop = IOLoop(make_current=False)
def tearDown(self):
self.io_loop.close()
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
yield
+
+
+# From https://nedbatchelder.com/blog/201508/using_context_managers_in_test_setup.html
+def setup_with_context_manager(testcase, cm):
+ """Use a contextmanager to setUp a test case."""
+ val = cm.__enter__()
+ testcase.addCleanup(cm.__exit__, None, None, None)
+ return val
import socket
import sys
import unittest
+import warnings
from tornado import gen
from tornado.httpclient import AsyncHTTPClient, HTTPResponse
def setUp(self) -> None:
super().setUp()
- self.io_loop = self.get_new_ioloop()
- self.io_loop.make_current()
+ # NOTE: this code attempts to navigate deprecation warnings introduced
+ # in Python 3.10. The idea of an implicit current event loop is
+ # deprecated in that version, with the intention that tests like this
+ # explicitly create a new event loop and run on it. However, other
+ # packages such as pytest-asyncio (as of version 0.16.0) still rely on
+ # the implicit current event loop and we want to be compatible with them
+ # (even when run on 3.10, but not, of course, on the future version of
+ # python that removes the get/set_event_loop methods completely).
+ #
+ # Deprecation warnings were introduced inconsistently:
+ # asyncio.get_event_loop warns, but
+ # asyncio.get_event_loop_policy().get_event_loop does not. Similarly,
+ # none of the set_event_loop methods warn, although comments on
+ # https://bugs.python.org/issue39529 indicate that they are also
+ # intended for future removal.
+ #
+ # Therefore, we first attempt to access the event loop with the
+ # (non-warning) policy method, and if it fails, fall back to creating a
+ # new event loop. We do not have effective test coverage of the
+ # new event loop case; this will have to be watched when/if
+ # get_event_loop is actually removed.
+ self.should_close_asyncio_loop = False
+ try:
+ self.asyncio_loop = asyncio.get_event_loop_policy().get_event_loop()
+ except Exception:
+ self.asyncio_loop = asyncio.new_event_loop()
+ self.should_close_asyncio_loop = True
+
+ async def get_loop() -> IOLoop:
+ return self.get_new_ioloop()
+
+ self.io_loop = self.asyncio_loop.run_until_complete(get_loop())
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore", DeprecationWarning)
+ self.io_loop.make_current()
def tearDown(self) -> None:
# Native coroutines tend to produce warnings if they're not
# Clean up Subprocess, so it can be used again with a new ioloop.
Subprocess.uninitialize()
- self.io_loop.clear_current()
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore", DeprecationWarning)
+ self.io_loop.clear_current()
if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):
# Try to clean up any file descriptors left open in the ioloop.
# This avoids leaks, especially when tests are run repeatedly
# in the same process with autoreload (because curl does not
# set FD_CLOEXEC on its file descriptors)
self.io_loop.close(all_fds=True)
+ if self.should_close_asyncio_loop:
+ self.asyncio_loop.close()
super().tearDown()
# In case an exception escaped or the StackContext caught an exception
# when there wasn't a wait() to re-raise it, do so here.
loop is being provided by another system (such as
``pytest-asyncio``).
"""
- return IOLoop()
+ return IOLoop(make_current=False)
def _handle_exception(
self, typ: Type[Exception], value: Exception, tb: TracebackType
projects / thirdparty / tornado.git / commitdiff
