Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/tornado/ioloop.py: 35%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#
2# Copyright 2009 Facebook
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
16"""An I/O event loop for non-blocking sockets.
18In Tornado 6.0, `.IOLoop` is a wrapper around the `asyncio` event loop, with a
19slightly different interface. The `.IOLoop` interface is now provided primarily
20for backwards compatibility; new code should generally use the `asyncio` event
21loop interface directly. The `IOLoop.current` class method provides the
22`IOLoop` instance corresponding to the running `asyncio` event loop.
24"""
26from __future__ import annotations
28import asyncio
29import concurrent.futures
30import datetime
31import functools
32import math
33import numbers
34import os
35import random
36import sys
37import time
38import typing
39import warnings
40from collections.abc import Awaitable, Callable
41from inspect import isawaitable
42from typing import Any, Protocol, TypedDict, TypeVar
44from tornado.concurrent import (
45 Future,
46 chain_future,
47 future_add_done_callback,
48 future_set_exc_info,
49 is_future,
50)
51from tornado.log import app_log
52from tornado.util import ( # noqa: F401 (TimeoutError re-exported for compatibility)
53 Configurable,
54 TimeoutError,
55 import_object,
56)
59class _Selectable(Protocol):
60 def fileno(self) -> int:
61 pass
63 def close(self) -> None:
64 pass
67_T = TypeVar("_T")
68_S = TypeVar("_S", bound=_Selectable)
71class IOLoop(Configurable):
72 """An I/O event loop.
74 As of Tornado 6.0, `IOLoop` is a wrapper around the `asyncio` event loop.
76 Example usage for a simple TCP server:
78 .. testcode::
80 import asyncio
81 import errno
82 import functools
83 import socket
85 import tornado
86 from tornado.iostream import IOStream
88 async def handle_connection(connection, address):
89 stream = IOStream(connection)
90 message = await stream.read_until_close()
91 print("message from client:", message.decode().strip())
93 def connection_ready(sock, fd, events):
94 while True:
95 try:
96 connection, address = sock.accept()
97 except BlockingIOError:
98 return
99 connection.setblocking(0)
100 io_loop = tornado.ioloop.IOLoop.current()
101 io_loop.spawn_callback(handle_connection, connection, address)
103 async def main():
104 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
105 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
106 sock.setblocking(0)
107 sock.bind(("", 8888))
108 sock.listen(128)
110 io_loop = tornado.ioloop.IOLoop.current()
111 callback = functools.partial(connection_ready, sock)
112 io_loop.add_handler(sock.fileno(), callback, io_loop.READ)
113 await asyncio.Event().wait()
115 if __name__ == "__main__":
116 asyncio.run(main())
118 Most applications should not attempt to construct an `IOLoop` directly,
119 and instead initialize the `asyncio` event loop and use `IOLoop.current()`.
120 In some cases, such as in test frameworks when initializing an `IOLoop`
121 to be run in a secondary thread, it may be appropriate to construct
122 an `IOLoop` with ``IOLoop(make_current=False)``.
124 In general, an `IOLoop` cannot survive a fork or be shared across processes
125 in any way. When multiple processes are being used, each process should
126 create its own `IOLoop`, which also implies that any objects which depend on
127 the `IOLoop` (such as `.AsyncHTTPClient`) must also be created in the child
128 processes. As a guideline, anything that starts processes (including the
129 `tornado.process` and `multiprocessing` modules) should do so as early as
130 possible, ideally the first thing the application does after loading its
131 configuration, and *before* any calls to `.IOLoop.start` or `asyncio.run`.
133 .. versionchanged:: 4.2
134 Added the ``make_current`` keyword argument to the `IOLoop`
135 constructor.
137 .. versionchanged:: 5.0
139 Uses the `asyncio` event loop by default. The ``IOLoop.configure`` method
140 cannot be used on Python 3 except to redundantly specify the `asyncio`
141 event loop.
143 .. versionchanged:: 6.3
144 ``make_current=True`` is now the default when creating an IOLoop -
145 previously the default was to make the event loop current if there wasn't
146 already a current one.
147 """
149 # These constants were originally based on constants from the epoll module.
150 NONE = 0
151 READ = 0x001
152 WRITE = 0x004
153 ERROR = 0x018
155 # In Python 3, _ioloop_for_asyncio maps from asyncio loops to IOLoops.
156 _ioloop_for_asyncio: dict[asyncio.AbstractEventLoop, IOLoop] = dict()
158 # Maintain a set of all pending tasks to follow the warning in the docs
159 # of asyncio.create_tasks:
160 # https://docs.python.org/3.11/library/asyncio-task.html#asyncio.create_task
161 # This ensures that all pending tasks have a strong reference so they
162 # will not be garbage collected before they are finished.
163 # (Thus avoiding "task was destroyed but it is pending" warnings)
164 # An analogous change has been proposed in cpython for 3.13:
165 # https://github.com/python/cpython/issues/91887
166 # If that change is accepted, this can eventually be removed.
167 # If it is not, we will consider the rationale and may remove this.
168 _pending_tasks: set[Future] = set()
170 @classmethod
171 def configure(cls, impl: None | str | type[Configurable], **kwargs: Any) -> None:
172 from tornado.platform.asyncio import BaseAsyncIOLoop
174 if isinstance(impl, str):
175 impl = import_object(impl)
176 if isinstance(impl, type) and not issubclass(impl, BaseAsyncIOLoop):
177 raise RuntimeError("only AsyncIOLoop is allowed when asyncio is available")
178 super().configure(impl, **kwargs)
180 @staticmethod
181 def instance() -> IOLoop:
182 """Deprecated alias for `IOLoop.current()`.
184 .. versionchanged:: 5.0
186 Previously, this method returned a global singleton
187 `IOLoop`, in contrast with the per-thread `IOLoop` returned
188 by `current()`. In nearly all cases the two were the same
189 (when they differed, it was generally used from non-Tornado
190 threads to communicate back to the main thread's `IOLoop`).
191 This distinction is not present in `asyncio`, so in order
192 to facilitate integration with that package `instance()`
193 was changed to be an alias to `current()`. Applications
194 using the cross-thread communications aspect of
195 `instance()` should instead set their own global variable
196 to point to the `IOLoop` they want to use.
198 .. deprecated:: 5.0
199 """
200 return IOLoop.current()
202 def install(self) -> None:
203 """Deprecated alias for `make_current()`.
205 .. versionchanged:: 5.0
207 Previously, this method would set this `IOLoop` as the
208 global singleton used by `IOLoop.instance()`. Now that
209 `instance()` is an alias for `current()`, `install()`
210 is an alias for `make_current()`.
212 .. deprecated:: 5.0
213 """
214 self.make_current()
216 @staticmethod
217 def clear_instance() -> None:
218 """Deprecated alias for `clear_current()`.
220 .. versionchanged:: 5.0
222 Previously, this method would clear the `IOLoop` used as
223 the global singleton by `IOLoop.instance()`. Now that
224 `instance()` is an alias for `current()`,
225 `clear_instance()` is an alias for `clear_current()`.
227 .. deprecated:: 5.0
229 """
230 IOLoop.clear_current()
232 @typing.overload
233 @staticmethod
234 def current() -> IOLoop:
235 pass
237 @typing.overload
238 @staticmethod
239 def current(instance: bool = True) -> IOLoop | None:
240 pass
242 @staticmethod
243 def current(instance: bool = True) -> IOLoop | None:
244 """Returns the current thread's `IOLoop`.
246 If an `IOLoop` is currently running or has been marked as
247 current by `make_current`, returns that instance. If there is
248 no current `IOLoop` and ``instance`` is true, creates one.
250 .. versionchanged:: 4.1
251 Added ``instance`` argument to control the fallback to
252 `IOLoop.instance()`.
253 .. versionchanged:: 5.0
254 On Python 3, control of the current `IOLoop` is delegated
255 to `asyncio`, with this and other methods as pass-through accessors.
256 The ``instance`` argument now controls whether an `IOLoop`
257 is created automatically when there is none, instead of
258 whether we fall back to `IOLoop.instance()` (which is now
259 an alias for this method). ``instance=False`` is deprecated,
260 since even if we do not create an `IOLoop`, this method
261 may initialize the asyncio loop.
263 .. deprecated:: 6.2
264 It is deprecated to call ``IOLoop.current()`` when no `asyncio`
265 event loop is running.
266 """
267 try:
268 loop = asyncio.get_event_loop()
269 except RuntimeError:
270 if not instance:
271 return None
272 # Create a new asyncio event loop for this thread.
273 loop = asyncio.new_event_loop()
274 asyncio.set_event_loop(loop)
276 try:
277 return IOLoop._ioloop_for_asyncio[loop]
278 except KeyError:
279 if instance:
280 from tornado.platform.asyncio import AsyncIOMainLoop
282 current: IOLoop | None = AsyncIOMainLoop()
283 else:
284 current = None
285 return current
287 def make_current(self) -> None:
288 """Makes this the `IOLoop` for the current thread.
290 An `IOLoop` automatically becomes current for its thread
291 when it is started, but it is sometimes useful to call
292 `make_current` explicitly before starting the `IOLoop`,
293 so that code run at startup time can find the right
294 instance.
296 .. versionchanged:: 4.1
297 An `IOLoop` created while there is no current `IOLoop`
298 will automatically become current.
300 .. versionchanged:: 5.0
301 This method also sets the current `asyncio` event loop.
303 .. deprecated:: 6.2
304 Setting and clearing the current event loop through Tornado is
305 deprecated. Use ``asyncio.set_event_loop`` instead if you need this.
306 """
307 warnings.warn(
308 "make_current is deprecated; start the event loop first",
309 DeprecationWarning,
310 stacklevel=2,
311 )
312 self._make_current()
314 def _make_current(self) -> None:
315 # The asyncio event loops override this method.
316 raise NotImplementedError()
318 @staticmethod
319 def clear_current() -> None:
320 """Clears the `IOLoop` for the current thread.
322 Intended primarily for use by test frameworks in between tests.
324 .. versionchanged:: 5.0
325 This method also clears the current `asyncio` event loop.
326 .. deprecated:: 6.2
327 """
328 warnings.warn(
329 "clear_current is deprecated",
330 DeprecationWarning,
331 stacklevel=2,
332 )
333 IOLoop._clear_current()
335 @staticmethod
336 def _clear_current() -> None:
337 old = IOLoop.current(instance=False)
338 if old is not None:
339 old._clear_current_hook()
341 def _clear_current_hook(self) -> None:
342 """Instance method called when an IOLoop ceases to be current.
344 May be overridden by subclasses as a counterpart to make_current.
345 """
346 pass
348 @classmethod
349 def configurable_base(cls) -> type[Configurable]:
350 return IOLoop
352 @classmethod
353 def configurable_default(cls) -> type[Configurable]:
354 from tornado.platform.asyncio import AsyncIOLoop
356 return AsyncIOLoop
358 def initialize(self, make_current: bool = True) -> None:
359 if make_current:
360 self._make_current()
362 def close(self, all_fds: bool = False) -> None:
363 """Closes the `IOLoop`, freeing any resources used.
365 If ``all_fds`` is true, all file descriptors registered on the
366 IOLoop will be closed (not just the ones created by the
367 `IOLoop` itself).
369 Many applications will only use a single `IOLoop` that runs for the
370 entire lifetime of the process. In that case closing the `IOLoop`
371 is not necessary since everything will be cleaned up when the
372 process exits. `IOLoop.close` is provided mainly for scenarios
373 such as unit tests, which create and destroy a large number of
374 ``IOLoops``.
376 An `IOLoop` must be completely stopped before it can be closed. This
377 means that `IOLoop.stop()` must be called *and* `IOLoop.start()` must
378 be allowed to return before attempting to call `IOLoop.close()`.
379 Therefore the call to `close` will usually appear just after
380 the call to `start` rather than near the call to `stop`.
382 .. versionchanged:: 3.1
383 If the `IOLoop` implementation supports non-integer objects
384 for "file descriptors", those objects will have their
385 ``close`` method when ``all_fds`` is true.
386 """
387 raise NotImplementedError()
389 @typing.overload
390 def add_handler(
391 self, fd: int, handler: Callable[[int, int], None], events: int
392 ) -> None:
393 pass
395 @typing.overload
396 def add_handler(
397 self, fd: _S, handler: Callable[[_S, int], None], events: int
398 ) -> None:
399 pass
401 def add_handler(
402 self, fd: int | _Selectable, handler: Callable[..., None], events: int
403 ) -> None:
404 """Registers the given handler to receive the given events for ``fd``.
406 The ``fd`` argument may either be an integer file descriptor or
407 a file-like object with a ``fileno()`` and ``close()`` method.
409 The ``events`` argument is a bitwise or of the constants
410 ``IOLoop.READ``, ``IOLoop.WRITE``, and ``IOLoop.ERROR``.
412 When an event occurs, ``handler(fd, events)`` will be run.
414 .. versionchanged:: 4.0
415 Added the ability to pass file-like objects in addition to
416 raw file descriptors.
417 """
418 raise NotImplementedError()
420 def update_handler(self, fd: int | _Selectable, events: int) -> None:
421 """Changes the events we listen for ``fd``.
423 .. versionchanged:: 4.0
424 Added the ability to pass file-like objects in addition to
425 raw file descriptors.
426 """
427 raise NotImplementedError()
429 def remove_handler(self, fd: int | _Selectable) -> None:
430 """Stop listening for events on ``fd``.
432 .. versionchanged:: 4.0
433 Added the ability to pass file-like objects in addition to
434 raw file descriptors.
435 """
436 raise NotImplementedError()
438 def start(self) -> None:
439 """Starts the I/O loop.
441 The loop will run until one of the callbacks calls `stop()`, which
442 will make the loop stop after the current event iteration completes.
443 """
444 raise NotImplementedError()
446 def stop(self) -> None:
447 """Stop the I/O loop.
449 If the event loop is not currently running, the next call to `start()`
450 will return immediately.
452 Note that even after `stop` has been called, the `IOLoop` is not
453 completely stopped until `IOLoop.start` has also returned.
454 Some work that was scheduled before the call to `stop` may still
455 be run before the `IOLoop` shuts down.
456 """
457 raise NotImplementedError()
459 def run_sync(self, func: Callable, timeout: float | None = None) -> Any:
460 """Starts the `IOLoop`, runs the given function, and stops the loop.
462 The function must return either an awaitable object or
463 ``None``. If the function returns an awaitable object, the
464 `IOLoop` will run until the awaitable is resolved (and
465 `run_sync()` will return the awaitable's result). If it raises
466 an exception, the `IOLoop` will stop and the exception will be
467 re-raised to the caller.
469 The keyword-only argument ``timeout`` may be used to set
470 a maximum duration for the function. If the timeout expires,
471 a `TimeoutError` is raised.
473 This method is useful to allow asynchronous calls in a
474 ``main()`` function::
476 async def main():
477 # do stuff...
479 if __name__ == '__main__':
480 IOLoop.current().run_sync(main)
482 .. versionchanged:: 4.3
483 Returning a non-``None``, non-awaitable value is now an error.
485 .. versionchanged:: 5.0
486 If a timeout occurs, the ``func`` coroutine will be cancelled.
488 .. versionchanged:: 6.2
489 ``tornado.util.TimeoutError`` is now an alias to ``asyncio.TimeoutError``.
490 """
492 class FutureCell(TypedDict):
493 future: Future | None
494 timeout_called: bool
496 future_cell: FutureCell = {"future": None, "timeout_called": False}
498 def run() -> None:
499 try:
500 result = func()
501 if result is not None:
502 from tornado.gen import convert_yielded
504 result = convert_yielded(result)
505 except Exception:
506 fut: Future[Any] = Future()
507 future_cell["future"] = fut
508 future_set_exc_info(fut, sys.exc_info())
509 else:
510 if is_future(result):
511 future_cell["future"] = result
512 else:
513 fut = Future()
514 future_cell["future"] = fut
515 fut.set_result(result)
516 assert future_cell["future"] is not None
517 self.add_future(future_cell["future"], lambda future: self.stop())
519 self.add_callback(run)
520 if timeout is not None:
522 def timeout_callback() -> None:
523 # signal that timeout is triggered
524 future_cell["timeout_called"] = True
525 # If we can cancel the future, do so and wait on it. If not,
526 # Just stop the loop and return with the task still pending.
527 # (If we neither cancel nor wait for the task, a warning
528 # will be logged).
529 assert future_cell["future"] is not None
530 if not future_cell["future"].cancel():
531 self.stop()
533 timeout_handle = self.add_timeout(self.time() + timeout, timeout_callback)
534 self.start()
535 if timeout is not None:
536 self.remove_timeout(timeout_handle)
537 assert future_cell["future"] is not None
538 if future_cell["future"].cancelled() or not future_cell["future"].done():
539 if future_cell["timeout_called"]:
540 raise TimeoutError("Operation timed out after %s seconds" % timeout)
541 else:
542 # timeout not called; maybe stop() was called explicitly
543 # or some other cancellation
544 raise RuntimeError("Event loop stopped before Future completed.")
545 return future_cell["future"].result()
547 def time(self) -> float:
548 """Returns the current time according to the `IOLoop`'s clock.
550 The return value is a floating-point number relative to an
551 unspecified time in the past.
553 Historically, the IOLoop could be customized to use e.g.
554 `time.monotonic` instead of `time.time`, but this is not
555 currently supported and so this method is equivalent to
556 `time.time`.
558 """
559 return time.time()
561 def add_timeout(
562 self,
563 deadline: float | datetime.timedelta,
564 callback: Callable,
565 *args: Any,
566 **kwargs: Any,
567 ) -> object:
568 """Runs the ``callback`` at the time ``deadline`` from the I/O loop.
570 Returns an opaque handle that may be passed to
571 `remove_timeout` to cancel.
573 ``deadline`` may be a number denoting a time (on the same
574 scale as `IOLoop.time`, normally `time.time`), or a
575 `datetime.timedelta` object for a deadline relative to the
576 current time. Since Tornado 4.0, `call_later` is a more
577 convenient alternative for the relative case since it does not
578 require a timedelta object.
580 Note that it is not safe to call `add_timeout` from other threads.
581 Instead, you must use `add_callback` to transfer control to the
582 `IOLoop`'s thread, and then call `add_timeout` from there.
584 Subclasses of IOLoop must implement either `add_timeout` or
585 `call_at`; the default implementations of each will call
586 the other. `call_at` is usually easier to implement, but
587 subclasses that wish to maintain compatibility with Tornado
588 versions prior to 4.0 must use `add_timeout` instead.
590 .. versionchanged:: 4.0
591 Now passes through ``*args`` and ``**kwargs`` to the callback.
592 """
593 if isinstance(deadline, numbers.Real):
594 return self.call_at(deadline, callback, *args, **kwargs)
595 elif isinstance(deadline, datetime.timedelta):
596 return self.call_at(
597 self.time() + deadline.total_seconds(), callback, *args, **kwargs
598 )
599 else:
600 raise TypeError("Unsupported deadline %r" % deadline)
602 def call_later(
603 self, delay: float, callback: Callable, *args: Any, **kwargs: Any
604 ) -> object:
605 """Runs the ``callback`` after ``delay`` seconds have passed.
607 Returns an opaque handle that may be passed to `remove_timeout`
608 to cancel. Note that unlike the `asyncio` method of the same
609 name, the returned object does not have a ``cancel()`` method.
611 See `add_timeout` for comments on thread-safety and subclassing.
613 .. versionadded:: 4.0
614 """
615 return self.call_at(self.time() + delay, callback, *args, **kwargs)
617 def call_at(
618 self, when: float, callback: Callable, *args: Any, **kwargs: Any
619 ) -> object:
620 """Runs the ``callback`` at the absolute time designated by ``when``.
622 ``when`` must be a number using the same reference point as
623 `IOLoop.time`.
625 Returns an opaque handle that may be passed to `remove_timeout`
626 to cancel. Note that unlike the `asyncio` method of the same
627 name, the returned object does not have a ``cancel()`` method.
629 See `add_timeout` for comments on thread-safety and subclassing.
631 .. versionadded:: 4.0
632 """
633 return self.add_timeout(when, callback, *args, **kwargs)
635 def remove_timeout(self, timeout: object) -> None:
636 """Cancels a pending timeout.
638 The argument is a handle as returned by `add_timeout`. It is
639 safe to call `remove_timeout` even if the callback has already
640 been run.
641 """
642 raise NotImplementedError()
644 def add_callback(self, callback: Callable, *args: Any, **kwargs: Any) -> None:
645 """Calls the given callback on the next I/O loop iteration.
647 It is safe to call this method from any thread at any time,
648 except from a signal handler. Note that this is the **only**
649 method in `IOLoop` that makes this thread-safety guarantee; all
650 other interaction with the `IOLoop` must be done from that
651 `IOLoop`'s thread. `add_callback()` may be used to transfer
652 control from other threads to the `IOLoop`'s thread.
653 """
654 raise NotImplementedError()
656 def add_callback_from_signal(
657 self, callback: Callable, *args: Any, **kwargs: Any
658 ) -> None:
659 """Calls the given callback on the next I/O loop iteration.
661 Intended to be safe for use from a Python signal handler; should not be
662 used otherwise.
664 .. deprecated:: 6.4
665 Use ``asyncio.AbstractEventLoop.add_signal_handler`` instead.
666 This method is suspected to have been broken since Tornado 5.0 and
667 will be removed in version 7.0.
668 """
669 raise NotImplementedError()
671 def spawn_callback(self, callback: Callable, *args: Any, **kwargs: Any) -> None:
672 """Calls the given callback on the next IOLoop iteration.
674 As of Tornado 6.0, this method is equivalent to `add_callback`.
676 .. versionadded:: 4.0
677 """
678 self.add_callback(callback, *args, **kwargs)
680 def add_future(
681 self,
682 future: Future[_T] | concurrent.futures.Future[_T],
683 callback: Callable[[Future[_T]], None],
684 ) -> None:
685 """Schedules a callback on the ``IOLoop`` when the given
686 `.Future` is finished.
688 The callback is invoked with one argument, the
689 `.Future`.
691 This method only accepts `.Future` objects and not other
692 awaitables (unlike most of Tornado where the two are
693 interchangeable).
694 """
695 if isinstance(future, Future):
696 # Note that we specifically do not want the inline behavior of
697 # tornado.concurrent.future_add_done_callback. We always want
698 # this callback scheduled on the next IOLoop iteration (which
699 # asyncio.Future always does).
700 #
701 # Wrap the callback in self._run_callback so we control
702 # the error logging (i.e. it goes to tornado.log.app_log
703 # instead of asyncio's log).
704 future.add_done_callback(
705 lambda f: self._run_callback(functools.partial(callback, f))
706 )
707 else:
708 assert is_future(future)
709 # For concurrent futures, we use self.add_callback, so
710 # it's fine if future_add_done_callback inlines that call.
711 future_add_done_callback(future, lambda f: self.add_callback(callback, f))
713 def run_in_executor(
714 self,
715 executor: concurrent.futures.Executor | None,
716 func: Callable[..., _T],
717 *args: Any,
718 ) -> Future[_T]:
719 """Runs a function in a ``concurrent.futures.Executor``. If
720 ``executor`` is ``None``, the IO loop's default executor will be used.
722 Use `functools.partial` to pass keyword arguments to ``func``.
724 .. versionadded:: 5.0
725 """
726 if executor is None:
727 if not hasattr(self, "_executor"):
728 from tornado.process import cpu_count
730 self._executor: concurrent.futures.Executor = (
731 concurrent.futures.ThreadPoolExecutor(max_workers=(cpu_count() * 5))
732 )
733 executor = self._executor
734 c_future = executor.submit(func, *args)
735 # Concurrent Futures are not usable with await. Wrap this in a
736 # Tornado Future instead, using self.add_future for thread-safety.
737 t_future: Future[_T] = Future()
738 self.add_future(c_future, lambda f: chain_future(f, t_future))
739 return t_future
741 def set_default_executor(self, executor: concurrent.futures.Executor) -> None:
742 """Sets the default executor to use with :meth:`run_in_executor`.
744 .. versionadded:: 5.0
745 """
746 self._executor = executor
748 def _run_callback(self, callback: Callable[[], Any]) -> None:
749 """Runs a callback with error handling.
751 .. versionchanged:: 6.0
753 CancelledErrors are no longer logged.
754 """
755 try:
756 ret = callback()
757 if ret is not None:
758 from tornado import gen
760 # Functions that return Futures typically swallow all
761 # exceptions and store them in the Future. If a Future
762 # makes it out to the IOLoop, ensure its exception (if any)
763 # gets logged too.
764 try:
765 ret = gen.convert_yielded(ret)
766 except gen.BadYieldError:
767 # It's not unusual for add_callback to be used with
768 # methods returning a non-None and non-yieldable
769 # result, which should just be ignored.
770 pass
771 else:
772 self.add_future(ret, self._discard_future_result)
773 except asyncio.CancelledError:
774 pass
775 except Exception:
776 app_log.error("Exception in callback %r", callback, exc_info=True)
778 def _discard_future_result(self, future: Future) -> None:
779 """Avoid unhandled-exception warnings from spawned coroutines."""
780 future.result()
782 def split_fd(self, fd: int | _Selectable) -> tuple[int, int | _Selectable]:
783 # """Returns an (fd, obj) pair from an ``fd`` parameter.
785 # We accept both raw file descriptors and file-like objects as
786 # input to `add_handler` and related methods. When a file-like
787 # object is passed, we must retain the object itself so we can
788 # close it correctly when the `IOLoop` shuts down, but the
789 # poller interfaces favor file descriptors (they will accept
790 # file-like objects and call ``fileno()`` for you, but they
791 # always return the descriptor itself).
793 # This method is provided for use by `IOLoop` subclasses and should
794 # not generally be used by application code.
796 # .. versionadded:: 4.0
797 # """
798 if isinstance(fd, int):
799 return fd, fd
800 return fd.fileno(), fd
802 def close_fd(self, fd: int | _Selectable) -> None:
803 # """Utility method to close an ``fd``.
805 # If ``fd`` is a file-like object, we close it directly; otherwise
806 # we use `os.close`.
808 # This method is provided for use by `IOLoop` subclasses (in
809 # implementations of ``IOLoop.close(all_fds=True)`` and should
810 # not generally be used by application code.
812 # .. versionadded:: 4.0
813 # """
814 try:
815 if isinstance(fd, int):
816 os.close(fd)
817 else:
818 fd.close()
819 except OSError:
820 pass
822 def _register_task(self, f: Future) -> None:
823 self._pending_tasks.add(f)
825 def _unregister_task(self, f: Future) -> None:
826 self._pending_tasks.discard(f)
829class _Timeout:
830 """An IOLoop timeout, a UNIX timestamp and a callback"""
832 # Reduce memory overhead when there are lots of pending callbacks
833 __slots__ = ["deadline", "callback", "tdeadline"]
835 def __init__(
836 self, deadline: float, callback: Callable[[], None], io_loop: IOLoop
837 ) -> None:
838 if not isinstance(deadline, numbers.Real):
839 raise TypeError("Unsupported deadline %r" % deadline)
840 self.deadline = deadline
841 self.callback = callback
842 self.tdeadline: tuple[float, int] = (
843 deadline,
844 next(io_loop._timeout_counter),
845 )
847 # Comparison methods to sort by deadline, with object id as a tiebreaker
848 # to guarantee a consistent ordering. The heapq module uses __le__
849 # in python2.5, and __lt__ in 2.6+ (sort() and most other comparisons
850 # use __lt__).
851 def __lt__(self, other: _Timeout) -> bool:
852 return self.tdeadline < other.tdeadline
854 def __le__(self, other: _Timeout) -> bool:
855 return self.tdeadline <= other.tdeadline
858class PeriodicCallback:
859 """Schedules the given callback to be called periodically.
861 The callback is called every ``callback_time`` milliseconds when
862 ``callback_time`` is a float. Note that the timeout is given in
863 milliseconds, while most other time-related functions in Tornado use
864 seconds. ``callback_time`` may alternatively be given as a
865 `datetime.timedelta` object.
867 If ``jitter`` is specified, each callback time will be randomly selected
868 within a window of ``jitter * callback_time`` milliseconds.
869 Jitter can be used to reduce alignment of events with similar periods.
870 A jitter of 0.1 means allowing a 10% variation in callback time.
871 The window is centered on ``callback_time`` so the total number of calls
872 within a given interval should not be significantly affected by adding
873 jitter.
875 If the callback runs for longer than ``callback_time`` milliseconds,
876 subsequent invocations will be skipped to get back on schedule.
878 `start` must be called after the `PeriodicCallback` is created.
880 .. versionchanged:: 5.0
881 The ``io_loop`` argument (deprecated since version 4.1) has been removed.
883 .. versionchanged:: 5.1
884 The ``jitter`` argument is added.
886 .. versionchanged:: 6.2
887 If the ``callback`` argument is a coroutine, and a callback runs for
888 longer than ``callback_time``, subsequent invocations will be skipped.
889 Previously this was only true for regular functions, not coroutines,
890 which were "fire-and-forget" for `PeriodicCallback`.
892 The ``callback_time`` argument now accepts `datetime.timedelta` objects,
893 in addition to the previous numeric milliseconds.
894 """
896 def __init__(
897 self,
898 callback: Callable[[], Awaitable | None],
899 callback_time: datetime.timedelta | float,
900 jitter: float = 0,
901 ) -> None:
902 self.callback = callback
903 if isinstance(callback_time, datetime.timedelta):
904 self.callback_time = callback_time / datetime.timedelta(milliseconds=1)
905 else:
906 if callback_time <= 0:
907 raise ValueError("Periodic callback must have a positive callback_time")
908 self.callback_time = callback_time
909 self.jitter = jitter
910 self._running = False
911 self._timeout: object = None
913 def start(self) -> None:
914 """Starts the timer."""
915 # Looking up the IOLoop here allows to first instantiate the
916 # PeriodicCallback in another thread, then start it using
917 # IOLoop.add_callback().
918 self.io_loop = IOLoop.current()
919 self._running = True
920 self._next_timeout = self.io_loop.time()
921 self._schedule_next()
923 def stop(self) -> None:
924 """Stops the timer."""
925 self._running = False
926 if self._timeout is not None:
927 self.io_loop.remove_timeout(self._timeout)
928 self._timeout = None
930 def is_running(self) -> bool:
931 """Returns ``True`` if this `.PeriodicCallback` has been started.
933 .. versionadded:: 4.1
934 """
935 return self._running
937 async def _run(self) -> None:
938 if not self._running:
939 return
940 try:
941 val = self.callback()
942 if val is not None and isawaitable(val):
943 await val
944 except Exception:
945 app_log.error("Exception in callback %r", self.callback, exc_info=True)
946 finally:
947 self._schedule_next()
949 def _schedule_next(self) -> None:
950 if self._running:
951 self._update_next(self.io_loop.time())
952 self._timeout = self.io_loop.add_timeout(self._next_timeout, self._run)
954 def _update_next(self, current_time: float) -> None:
955 callback_time_sec = self.callback_time / 1000.0
956 if self.jitter:
957 # apply jitter fraction
958 callback_time_sec *= 1 + (self.jitter * (random.random() - 0.5))
959 if self._next_timeout <= current_time:
960 # The period should be measured from the start of one call
961 # to the start of the next. If one call takes too long,
962 # skip cycles to get back to a multiple of the original
963 # schedule.
964 self._next_timeout += (
965 math.floor((current_time - self._next_timeout) / callback_time_sec) + 1
966 ) * callback_time_sec
967 else:
968 # If the clock moved backwards, ensure we advance the next
969 # timeout instead of recomputing the same value again.
970 # This may result in long gaps between callbacks if the
971 # clock jumps backwards by a lot, but the far more common
972 # scenario is a small NTP adjustment that should just be
973 # ignored.
974 #
975 # Note that on some systems if time.time() runs slower
976 # than time.monotonic() (most common on windows), we
977 # effectively experience a small backwards time jump on
978 # every iteration because PeriodicCallback uses
979 # time.time() while asyncio schedules callbacks using
980 # time.monotonic().
981 # https://github.com/tornadoweb/tornado/issues/2333
982 self._next_timeout += callback_time_sec