| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946 |
- #
- # Copyright 2009 Facebook
- #
- # Licensed under the Apache License, Version 2.0 (the "License"); you may
- # not use this file except in compliance with the License. You may obtain
- # a copy of the License at
- #
- # http://www.apache.org/licenses/LICENSE-2.0
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- # License for the specific language governing permissions and limitations
- # under the License.
- """An I/O event loop for non-blocking sockets.
- In Tornado 6.0, `.IOLoop` is a wrapper around the `asyncio` event
- loop, with a slightly different interface for historical reasons.
- Applications can use either the `.IOLoop` interface or the underlying
- `asyncio` event loop directly (unless compatibility with older
- versions of Tornado is desired, in which case `.IOLoop` must be used).
- Typical applications will use a single `IOLoop` object, accessed via
- `IOLoop.current` class method. The `IOLoop.start` method (or
- equivalently, `asyncio.AbstractEventLoop.run_forever`) should usually
- be called at the end of the ``main()`` function. Atypical applications
- may use more than one `IOLoop`, such as one `IOLoop` per thread, or
- per `unittest` case.
- """
- import asyncio
- import concurrent.futures
- import datetime
- import functools
- import logging
- import numbers
- import os
- import sys
- import time
- import math
- import random
- from tornado.concurrent import (
- Future,
- is_future,
- chain_future,
- future_set_exc_info,
- future_add_done_callback,
- )
- from tornado.log import app_log
- from tornado.util import Configurable, TimeoutError, import_object
- import typing
- from typing import Union, Any, Type, Optional, Callable, TypeVar, Tuple, Awaitable
- if typing.TYPE_CHECKING:
- from typing import Dict, List # noqa: F401
- from typing_extensions import Protocol
- else:
- Protocol = object
- class _Selectable(Protocol):
- def fileno(self) -> int:
- pass
- def close(self) -> None:
- pass
- _T = TypeVar("_T")
- _S = TypeVar("_S", bound=_Selectable)
- class IOLoop(Configurable):
- """An I/O event loop.
- As of Tornado 6.0, `IOLoop` is a wrapper around the `asyncio` event
- loop.
- Example usage for a simple TCP server:
- .. testcode::
- import errno
- import functools
- import socket
- import tornado.ioloop
- from tornado.iostream import IOStream
- async def handle_connection(connection, address):
- stream = IOStream(connection)
- message = await stream.read_until_close()
- print("message from client:", message.decode().strip())
- def connection_ready(sock, fd, events):
- while True:
- try:
- connection, address = sock.accept()
- except socket.error as e:
- if e.args[0] not in (errno.EWOULDBLOCK, errno.EAGAIN):
- raise
- return
- connection.setblocking(0)
- io_loop = tornado.ioloop.IOLoop.current()
- io_loop.spawn_callback(handle_connection, connection, address)
- if __name__ == '__main__':
- sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
- sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- sock.setblocking(0)
- sock.bind(("", 8888))
- sock.listen(128)
- 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()
- .. 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.
- In general, an `IOLoop` cannot survive a fork or be shared across
- processes in any way. When multiple processes are being used, each
- process should create its own `IOLoop`, which also implies that
- any objects which depend on the `IOLoop` (such as
- `.AsyncHTTPClient`) must also be created in the child processes.
- As a guideline, anything that starts processes (including the
- `tornado.process` and `multiprocessing` modules) should do so as
- early as possible, ideally the first thing the application does
- after loading its configuration in ``main()``.
- .. versionchanged:: 4.2
- Added the ``make_current`` keyword argument to the `IOLoop`
- constructor.
- .. versionchanged:: 5.0
- Uses the `asyncio` event loop by default. The
- ``IOLoop.configure`` method cannot be used on Python 3 except
- to redundantly specify the `asyncio` event loop.
- """
- # These constants were originally based on constants from the epoll module.
- NONE = 0
- READ = 0x001
- WRITE = 0x004
- ERROR = 0x018
- # In Python 3, _ioloop_for_asyncio maps from asyncio loops to IOLoops.
- _ioloop_for_asyncio = dict() # type: Dict[asyncio.AbstractEventLoop, IOLoop]
- @classmethod
- def configure(
- cls, impl: "Union[None, str, Type[Configurable]]", **kwargs: Any
- ) -> None:
- if asyncio is not None:
- from tornado.platform.asyncio import BaseAsyncIOLoop
- if isinstance(impl, str):
- impl = import_object(impl)
- if isinstance(impl, type) and not issubclass(impl, BaseAsyncIOLoop):
- raise RuntimeError(
- "only AsyncIOLoop is allowed when asyncio is available"
- )
- super(IOLoop, cls).configure(impl, **kwargs)
- @staticmethod
- def instance() -> "IOLoop":
- """Deprecated alias for `IOLoop.current()`.
- .. versionchanged:: 5.0
- Previously, this method returned a global singleton
- `IOLoop`, in contrast with the per-thread `IOLoop` returned
- by `current()`. In nearly all cases the two were the same
- (when they differed, it was generally used from non-Tornado
- threads to communicate back to the main thread's `IOLoop`).
- This distinction is not present in `asyncio`, so in order
- to facilitate integration with that package `instance()`
- was changed to be an alias to `current()`. Applications
- using the cross-thread communications aspect of
- `instance()` should instead set their own global variable
- to point to the `IOLoop` they want to use.
- .. deprecated:: 5.0
- """
- return IOLoop.current()
- def install(self) -> None:
- """Deprecated alias for `make_current()`.
- .. versionchanged:: 5.0
- Previously, this method would set this `IOLoop` as the
- global singleton used by `IOLoop.instance()`. Now that
- `instance()` is an alias for `current()`, `install()`
- is an alias for `make_current()`.
- .. deprecated:: 5.0
- """
- self.make_current()
- @staticmethod
- def clear_instance() -> None:
- """Deprecated alias for `clear_current()`.
- .. versionchanged:: 5.0
- Previously, this method would clear the `IOLoop` used as
- the global singleton by `IOLoop.instance()`. Now that
- `instance()` is an alias for `current()`,
- `clear_instance()` is an alias for `clear_current()`.
- .. deprecated:: 5.0
- """
- IOLoop.clear_current()
- @typing.overload
- @staticmethod
- def current() -> "IOLoop":
- pass
- @typing.overload # noqa: F811
- @staticmethod
- def current(instance: bool = True) -> Optional["IOLoop"]:
- pass
- @staticmethod # noqa: F811
- def current(instance: bool = True) -> Optional["IOLoop"]:
- """Returns the current thread's `IOLoop`.
- If an `IOLoop` is currently running or has been marked as
- current by `make_current`, returns that instance. If there is
- no current `IOLoop` and ``instance`` is true, creates one.
- .. versionchanged:: 4.1
- Added ``instance`` argument to control the fallback to
- `IOLoop.instance()`.
- .. versionchanged:: 5.0
- On Python 3, control of the current `IOLoop` is delegated
- to `asyncio`, with this and other methods as pass-through accessors.
- The ``instance`` argument now controls whether an `IOLoop`
- is created automatically when there is none, instead of
- whether we fall back to `IOLoop.instance()` (which is now
- 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.
- """
- try:
- loop = asyncio.get_event_loop()
- except (RuntimeError, AssertionError):
- if not instance:
- return None
- raise
- try:
- return IOLoop._ioloop_for_asyncio[loop]
- except KeyError:
- if instance:
- from tornado.platform.asyncio import AsyncIOMainLoop
- current = AsyncIOMainLoop(make_current=True) # type: Optional[IOLoop]
- else:
- current = None
- return current
- def make_current(self) -> None:
- """Makes this the `IOLoop` for the current thread.
- An `IOLoop` automatically becomes current for its thread
- when it is started, but it is sometimes useful to call
- `make_current` explicitly before starting the `IOLoop`,
- so that code run at startup time can find the right
- instance.
- .. versionchanged:: 4.1
- An `IOLoop` created while there is no current `IOLoop`
- will automatically become current.
- .. versionchanged:: 5.0
- This method also sets the current `asyncio` event loop.
- """
- # The asyncio event loops override this method.
- raise NotImplementedError()
- @staticmethod
- def clear_current() -> None:
- """Clears the `IOLoop` for the current thread.
- Intended primarily for use by test frameworks in between tests.
- .. versionchanged:: 5.0
- This method also clears the current `asyncio` event loop.
- """
- old = IOLoop.current(instance=False)
- if old is not None:
- old._clear_current_hook()
- if asyncio is None:
- IOLoop._current.instance = None
- def _clear_current_hook(self) -> None:
- """Instance method called when an IOLoop ceases to be current.
- May be overridden by subclasses as a counterpart to make_current.
- """
- pass
- @classmethod
- def configurable_base(cls) -> Type[Configurable]:
- return IOLoop
- @classmethod
- def configurable_default(cls) -> Type[Configurable]:
- from tornado.platform.asyncio import AsyncIOLoop
- return AsyncIOLoop
- def initialize(self, make_current: bool = None) -> None:
- if make_current is None:
- if IOLoop.current(instance=False) is None:
- self.make_current()
- elif make_current:
- current = IOLoop.current(instance=False)
- # AsyncIO loops can already be current by this point.
- if current is not None and current is not self:
- raise RuntimeError("current IOLoop already exists")
- self.make_current()
- def close(self, all_fds: bool = False) -> None:
- """Closes the `IOLoop`, freeing any resources used.
- If ``all_fds`` is true, all file descriptors registered on the
- IOLoop will be closed (not just the ones created by the
- `IOLoop` itself).
- Many applications will only use a single `IOLoop` that runs for the
- entire lifetime of the process. In that case closing the `IOLoop`
- is not necessary since everything will be cleaned up when the
- process exits. `IOLoop.close` is provided mainly for scenarios
- such as unit tests, which create and destroy a large number of
- ``IOLoops``.
- An `IOLoop` must be completely stopped before it can be closed. This
- means that `IOLoop.stop()` must be called *and* `IOLoop.start()` must
- be allowed to return before attempting to call `IOLoop.close()`.
- Therefore the call to `close` will usually appear just after
- the call to `start` rather than near the call to `stop`.
- .. versionchanged:: 3.1
- If the `IOLoop` implementation supports non-integer objects
- for "file descriptors", those objects will have their
- ``close`` method when ``all_fds`` is true.
- """
- raise NotImplementedError()
- @typing.overload
- def add_handler(
- self, fd: int, handler: Callable[[int, int], None], events: int
- ) -> None:
- pass
- @typing.overload # noqa: F811
- def add_handler(
- self, fd: _S, handler: Callable[[_S, int], None], events: int
- ) -> None:
- pass
- def add_handler( # noqa: F811
- self, fd: Union[int, _Selectable], handler: Callable[..., None], events: int
- ) -> None:
- """Registers the given handler to receive the given events for ``fd``.
- The ``fd`` argument may either be an integer file descriptor or
- a file-like object with a ``fileno()`` and ``close()`` method.
- The ``events`` argument is a bitwise or of the constants
- ``IOLoop.READ``, ``IOLoop.WRITE``, and ``IOLoop.ERROR``.
- When an event occurs, ``handler(fd, events)`` will be run.
- .. versionchanged:: 4.0
- Added the ability to pass file-like objects in addition to
- raw file descriptors.
- """
- raise NotImplementedError()
- def update_handler(self, fd: Union[int, _Selectable], events: int) -> None:
- """Changes the events we listen for ``fd``.
- .. versionchanged:: 4.0
- Added the ability to pass file-like objects in addition to
- raw file descriptors.
- """
- raise NotImplementedError()
- def remove_handler(self, fd: Union[int, _Selectable]) -> None:
- """Stop listening for events on ``fd``.
- .. versionchanged:: 4.0
- Added the ability to pass file-like objects in addition to
- raw file descriptors.
- """
- raise NotImplementedError()
- def start(self) -> None:
- """Starts the I/O loop.
- The loop will run until one of the callbacks calls `stop()`, which
- will make the loop stop after the current event iteration completes.
- """
- raise NotImplementedError()
- def _setup_logging(self) -> None:
- """The IOLoop catches and logs exceptions, so it's
- important that log output be visible. However, python's
- default behavior for non-root loggers (prior to python
- 3.2) is to print an unhelpful "no handlers could be
- found" message rather than the actual log entry, so we
- must explicitly configure logging if we've made it this
- far without anything.
- This method should be called from start() in subclasses.
- """
- if not any(
- [
- logging.getLogger().handlers,
- logging.getLogger("tornado").handlers,
- logging.getLogger("tornado.application").handlers,
- ]
- ):
- logging.basicConfig()
- def stop(self) -> None:
- """Stop the I/O loop.
- If the event loop is not currently running, the next call to `start()`
- will return immediately.
- Note that even after `stop` has been called, the `IOLoop` is not
- completely stopped until `IOLoop.start` has also returned.
- Some work that was scheduled before the call to `stop` may still
- be run before the `IOLoop` shuts down.
- """
- raise NotImplementedError()
- def run_sync(self, func: Callable, timeout: float = None) -> Any:
- """Starts the `IOLoop`, runs the given function, and stops the loop.
- The function must return either an awaitable object or
- ``None``. If the function returns an awaitable object, the
- `IOLoop` will run until the awaitable is resolved (and
- `run_sync()` will return the awaitable's result). If it raises
- an exception, the `IOLoop` will stop and the exception will be
- re-raised to the caller.
- The keyword-only argument ``timeout`` may be used to set
- a maximum duration for the function. If the timeout expires,
- a `tornado.util.TimeoutError` is raised.
- This method is useful to allow asynchronous calls in a
- ``main()`` function::
- async def main():
- # do stuff...
- if __name__ == '__main__':
- IOLoop.current().run_sync(main)
- .. versionchanged:: 4.3
- Returning a non-``None``, non-awaitable value is now an error.
- .. versionchanged:: 5.0
- If a timeout occurs, the ``func`` coroutine will be cancelled.
- """
- future_cell = [None] # type: List[Optional[Future]]
- def run() -> None:
- try:
- result = func()
- if result is not None:
- from tornado.gen import convert_yielded
- result = convert_yielded(result)
- except Exception:
- fut = Future() # type: Future[Any]
- future_cell[0] = fut
- future_set_exc_info(fut, sys.exc_info())
- else:
- if is_future(result):
- future_cell[0] = result
- else:
- fut = Future()
- future_cell[0] = fut
- fut.set_result(result)
- assert future_cell[0] is not None
- self.add_future(future_cell[0], lambda future: self.stop())
- self.add_callback(run)
- if timeout is not None:
- def timeout_callback() -> None:
- # If we can cancel the future, do so and wait on it. If not,
- # Just stop the loop and return with the task still pending.
- # (If we neither cancel nor wait for the task, a warning
- # will be logged).
- assert future_cell[0] is not None
- if not future_cell[0].cancel():
- self.stop()
- timeout_handle = self.add_timeout(self.time() + timeout, timeout_callback)
- self.start()
- if timeout is not None:
- self.remove_timeout(timeout_handle)
- assert future_cell[0] is not None
- if future_cell[0].cancelled() or not future_cell[0].done():
- raise TimeoutError("Operation timed out after %s seconds" % timeout)
- return future_cell[0].result()
- def time(self) -> float:
- """Returns the current time according to the `IOLoop`'s clock.
- The return value is a floating-point number relative to an
- unspecified time in the past.
- Historically, the IOLoop could be customized to use e.g.
- `time.monotonic` instead of `time.time`, but this is not
- currently supported and so this method is equivalent to
- `time.time`.
- """
- return time.time()
- def add_timeout(
- self,
- deadline: Union[float, datetime.timedelta],
- callback: Callable[..., None],
- *args: Any,
- **kwargs: Any
- ) -> object:
- """Runs the ``callback`` at the time ``deadline`` from the I/O loop.
- Returns an opaque handle that may be passed to
- `remove_timeout` to cancel.
- ``deadline`` may be a number denoting a time (on the same
- scale as `IOLoop.time`, normally `time.time`), or a
- `datetime.timedelta` object for a deadline relative to the
- current time. Since Tornado 4.0, `call_later` is a more
- convenient alternative for the relative case since it does not
- require a timedelta object.
- Note that it is not safe to call `add_timeout` from other threads.
- Instead, you must use `add_callback` to transfer control to the
- `IOLoop`'s thread, and then call `add_timeout` from there.
- Subclasses of IOLoop must implement either `add_timeout` or
- `call_at`; the default implementations of each will call
- the other. `call_at` is usually easier to implement, but
- subclasses that wish to maintain compatibility with Tornado
- versions prior to 4.0 must use `add_timeout` instead.
- .. versionchanged:: 4.0
- Now passes through ``*args`` and ``**kwargs`` to the callback.
- """
- if isinstance(deadline, numbers.Real):
- return self.call_at(deadline, callback, *args, **kwargs)
- elif isinstance(deadline, datetime.timedelta):
- return self.call_at(
- self.time() + deadline.total_seconds(), callback, *args, **kwargs
- )
- else:
- raise TypeError("Unsupported deadline %r" % deadline)
- def call_later(
- self, delay: float, callback: Callable[..., None], *args: Any, **kwargs: Any
- ) -> object:
- """Runs the ``callback`` after ``delay`` seconds have passed.
- Returns an opaque handle that may be passed to `remove_timeout`
- to cancel. Note that unlike the `asyncio` method of the same
- name, the returned object does not have a ``cancel()`` method.
- See `add_timeout` for comments on thread-safety and subclassing.
- .. versionadded:: 4.0
- """
- return self.call_at(self.time() + delay, callback, *args, **kwargs)
- def call_at(
- self, when: float, callback: Callable[..., None], *args: Any, **kwargs: Any
- ) -> object:
- """Runs the ``callback`` at the absolute time designated by ``when``.
- ``when`` must be a number using the same reference point as
- `IOLoop.time`.
- Returns an opaque handle that may be passed to `remove_timeout`
- to cancel. Note that unlike the `asyncio` method of the same
- name, the returned object does not have a ``cancel()`` method.
- See `add_timeout` for comments on thread-safety and subclassing.
- .. versionadded:: 4.0
- """
- return self.add_timeout(when, callback, *args, **kwargs)
- def remove_timeout(self, timeout: object) -> None:
- """Cancels a pending timeout.
- The argument is a handle as returned by `add_timeout`. It is
- safe to call `remove_timeout` even if the callback has already
- been run.
- """
- raise NotImplementedError()
- def add_callback(self, callback: Callable, *args: Any, **kwargs: Any) -> None:
- """Calls the given callback on the next I/O loop iteration.
- It is safe to call this method from any thread at any time,
- except from a signal handler. Note that this is the **only**
- method in `IOLoop` that makes this thread-safety guarantee; all
- other interaction with the `IOLoop` must be done from that
- `IOLoop`'s thread. `add_callback()` may be used to transfer
- control from other threads to the `IOLoop`'s thread.
- To add a callback from a signal handler, see
- `add_callback_from_signal`.
- """
- raise NotImplementedError()
- def add_callback_from_signal(
- self, callback: Callable, *args: Any, **kwargs: Any
- ) -> None:
- """Calls the given callback on the next I/O loop iteration.
- Safe for use from a Python signal handler; should not be used
- otherwise.
- """
- raise NotImplementedError()
- def spawn_callback(self, callback: Callable, *args: Any, **kwargs: Any) -> None:
- """Calls the given callback on the next IOLoop iteration.
- As of Tornado 6.0, this method is equivalent to `add_callback`.
- .. versionadded:: 4.0
- """
- self.add_callback(callback, *args, **kwargs)
- def add_future(
- self,
- future: "Union[Future[_T], concurrent.futures.Future[_T]]",
- callback: Callable[["Future[_T]"], None],
- ) -> None:
- """Schedules a callback on the ``IOLoop`` when the given
- `.Future` is finished.
- The callback is invoked with one argument, the
- `.Future`.
- This method only accepts `.Future` objects and not other
- awaitables (unlike most of Tornado where the two are
- interchangeable).
- """
- if isinstance(future, Future):
- # Note that we specifically do not want the inline behavior of
- # tornado.concurrent.future_add_done_callback. We always want
- # this callback scheduled on the next IOLoop iteration (which
- # asyncio.Future always does).
- #
- # Wrap the callback in self._run_callback so we control
- # the error logging (i.e. it goes to tornado.log.app_log
- # instead of asyncio's log).
- future.add_done_callback(
- lambda f: self._run_callback(functools.partial(callback, future))
- )
- else:
- assert is_future(future)
- # For concurrent futures, we use self.add_callback, so
- # it's fine if future_add_done_callback inlines that call.
- future_add_done_callback(
- future, lambda f: self.add_callback(callback, future)
- )
- def run_in_executor(
- self,
- executor: Optional[concurrent.futures.Executor],
- func: Callable[..., _T],
- *args: Any
- ) -> Awaitable[_T]:
- """Runs a function in a ``concurrent.futures.Executor``. If
- ``executor`` is ``None``, the IO loop's default executor will be used.
- Use `functools.partial` to pass keyword arguments to ``func``.
- .. versionadded:: 5.0
- """
- if executor is None:
- if not hasattr(self, "_executor"):
- from tornado.process import cpu_count
- self._executor = concurrent.futures.ThreadPoolExecutor(
- max_workers=(cpu_count() * 5)
- ) # type: concurrent.futures.Executor
- executor = self._executor
- c_future = executor.submit(func, *args)
- # Concurrent Futures are not usable with await. Wrap this in a
- # Tornado Future instead, using self.add_future for thread-safety.
- t_future = Future() # type: Future[_T]
- self.add_future(c_future, lambda f: chain_future(f, t_future))
- return t_future
- def set_default_executor(self, executor: concurrent.futures.Executor) -> None:
- """Sets the default executor to use with :meth:`run_in_executor`.
- .. versionadded:: 5.0
- """
- self._executor = executor
- def _run_callback(self, callback: Callable[[], Any]) -> None:
- """Runs a callback with error handling.
- .. versionchanged:: 6.0
- CancelledErrors are no longer logged.
- """
- try:
- ret = callback()
- if ret is not None:
- from tornado import gen
- # Functions that return Futures typically swallow all
- # exceptions and store them in the Future. If a Future
- # makes it out to the IOLoop, ensure its exception (if any)
- # gets logged too.
- try:
- ret = gen.convert_yielded(ret)
- except gen.BadYieldError:
- # It's not unusual for add_callback to be used with
- # methods returning a non-None and non-yieldable
- # result, which should just be ignored.
- pass
- else:
- self.add_future(ret, self._discard_future_result)
- except asyncio.CancelledError:
- pass
- except Exception:
- app_log.error("Exception in callback %r", callback, exc_info=True)
- def _discard_future_result(self, future: Future) -> None:
- """Avoid unhandled-exception warnings from spawned coroutines."""
- future.result()
- def split_fd(
- self, fd: Union[int, _Selectable]
- ) -> Tuple[int, Union[int, _Selectable]]:
- # """Returns an (fd, obj) pair from an ``fd`` parameter.
- # We accept both raw file descriptors and file-like objects as
- # input to `add_handler` and related methods. When a file-like
- # object is passed, we must retain the object itself so we can
- # close it correctly when the `IOLoop` shuts down, but the
- # poller interfaces favor file descriptors (they will accept
- # file-like objects and call ``fileno()`` for you, but they
- # always return the descriptor itself).
- # This method is provided for use by `IOLoop` subclasses and should
- # not generally be used by application code.
- # .. versionadded:: 4.0
- # """
- if isinstance(fd, int):
- return fd, fd
- return fd.fileno(), fd
- def close_fd(self, fd: Union[int, _Selectable]) -> None:
- # """Utility method to close an ``fd``.
- # If ``fd`` is a file-like object, we close it directly; otherwise
- # we use `os.close`.
- # This method is provided for use by `IOLoop` subclasses (in
- # implementations of ``IOLoop.close(all_fds=True)`` and should
- # not generally be used by application code.
- # .. versionadded:: 4.0
- # """
- try:
- if isinstance(fd, int):
- os.close(fd)
- else:
- fd.close()
- except OSError:
- pass
- class _Timeout(object):
- """An IOLoop timeout, a UNIX timestamp and a callback"""
- # Reduce memory overhead when there are lots of pending callbacks
- __slots__ = ["deadline", "callback", "tdeadline"]
- def __init__(
- self, deadline: float, callback: Callable[[], None], io_loop: IOLoop
- ) -> None:
- if not isinstance(deadline, numbers.Real):
- raise TypeError("Unsupported deadline %r" % deadline)
- self.deadline = deadline
- self.callback = callback
- self.tdeadline = (
- deadline,
- next(io_loop._timeout_counter),
- ) # type: Tuple[float, int]
- # Comparison methods to sort by deadline, with object id as a tiebreaker
- # to guarantee a consistent ordering. The heapq module uses __le__
- # in python2.5, and __lt__ in 2.6+ (sort() and most other comparisons
- # use __lt__).
- def __lt__(self, other: "_Timeout") -> bool:
- return self.tdeadline < other.tdeadline
- def __le__(self, other: "_Timeout") -> bool:
- return self.tdeadline <= other.tdeadline
- class PeriodicCallback(object):
- """Schedules the given callback to be called periodically.
- The callback is called every ``callback_time`` milliseconds.
- Note that the timeout is given in milliseconds, while most other
- time-related functions in Tornado use seconds.
- If ``jitter`` is specified, each callback time will be randomly selected
- within a window of ``jitter * callback_time`` milliseconds.
- Jitter can be used to reduce alignment of events with similar periods.
- A jitter of 0.1 means allowing a 10% variation in callback time.
- The window is centered on ``callback_time`` so the total number of calls
- within a given interval should not be significantly affected by adding
- jitter.
- If the callback runs for longer than ``callback_time`` milliseconds,
- subsequent invocations will be skipped to get back on schedule.
- `start` must be called after the `PeriodicCallback` is created.
- .. versionchanged:: 5.0
- The ``io_loop`` argument (deprecated since version 4.1) has been removed.
- .. versionchanged:: 5.1
- The ``jitter`` argument is added.
- """
- def __init__(
- self, callback: Callable[[], None], callback_time: float, jitter: float = 0
- ) -> None:
- self.callback = callback
- if callback_time <= 0:
- raise ValueError("Periodic callback must have a positive callback_time")
- self.callback_time = callback_time
- self.jitter = jitter
- self._running = False
- self._timeout = None # type: object
- def start(self) -> None:
- """Starts the timer."""
- # Looking up the IOLoop here allows to first instantiate the
- # PeriodicCallback in another thread, then start it using
- # IOLoop.add_callback().
- self.io_loop = IOLoop.current()
- self._running = True
- self._next_timeout = self.io_loop.time()
- self._schedule_next()
- def stop(self) -> None:
- """Stops the timer."""
- self._running = False
- if self._timeout is not None:
- self.io_loop.remove_timeout(self._timeout)
- self._timeout = None
- def is_running(self) -> bool:
- """Returns ``True`` if this `.PeriodicCallback` has been started.
- .. versionadded:: 4.1
- """
- return self._running
- def _run(self) -> None:
- if not self._running:
- return
- try:
- return self.callback()
- except Exception:
- app_log.error("Exception in callback %r", self.callback, exc_info=True)
- finally:
- self._schedule_next()
- def _schedule_next(self) -> None:
- if self._running:
- self._update_next(self.io_loop.time())
- self._timeout = self.io_loop.add_timeout(self._next_timeout, self._run)
- def _update_next(self, current_time: float) -> None:
- callback_time_sec = self.callback_time / 1000.0
- if self.jitter:
- # apply jitter fraction
- callback_time_sec *= 1 + (self.jitter * (random.random() - 0.5))
- if self._next_timeout <= current_time:
- # The period should be measured from the start of one call
- # to the start of the next. If one call takes too long,
- # skip cycles to get back to a multiple of the original
- # schedule.
- self._next_timeout += (
- math.floor((current_time - self._next_timeout) / callback_time_sec) + 1
- ) * callback_time_sec
- else:
- # If the clock moved backwards, ensure we advance the next
- # timeout instead of recomputing the same value again.
- # This may result in long gaps between callbacks if the
- # clock jumps backwards by a lot, but the far more common
- # scenario is a small NTP adjustment that should just be
- # ignored.
- #
- # Note that on some systems if time.time() runs slower
- # than time.monotonic() (most common on windows), we
- # effectively experience a small backwards time jump on
- # every iteration because PeriodicCallback uses
- # time.time() while asyncio schedules callbacks using
- # time.monotonic().
- # https://github.com/tornadoweb/tornado/issues/2333
- self._next_timeout += callback_time_sec
|