Coverage for Lib/asyncio/events.py: 92%
306 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 01:31 +0000
1"""Event loop and event loop policy."""
3# Contains code from https://github.com/MagicStack/uvloop/tree/v0.16.0
4# SPDX-License-Identifier: PSF-2.0 AND (MIT OR Apache-2.0)
5# SPDX-FileCopyrightText: Copyright (c) 2015-2021 MagicStack Inc. http://magic.io
7__all__ = (
8 "AbstractEventLoop",
9 "AbstractServer",
10 "Handle",
11 "TimerHandle",
12 "get_event_loop",
13 "set_event_loop",
14 "new_event_loop",
15 "_set_running_loop",
16 "get_running_loop",
17 "_get_running_loop",
18)
20import contextvars
21import os
22import signal
23import socket
24import subprocess
25import sys
26import threading
28from . import format_helpers
31class Handle:
32 """Object returned by callback registration methods."""
34 __slots__ = ('_callback', '_args', '_cancelled', '_loop',
35 '_source_traceback', '_repr', '__weakref__',
36 '_context')
38 def __init__(self, callback, args, loop, context=None):
39 if context is None:
40 context = contextvars.copy_context()
41 self._context = context
42 self._loop = loop
43 self._callback = callback
44 self._args = args
45 self._cancelled = False
46 self._repr = None
47 if self._loop.get_debug():
48 self._source_traceback = format_helpers.extract_stack(
49 sys._getframe(1))
50 else:
51 self._source_traceback = None
53 def _repr_info(self):
54 info = [self.__class__.__name__]
55 if self._cancelled:
56 info.append('cancelled')
57 if self._callback is not None:
58 info.append(format_helpers._format_callback_source(
59 self._callback, self._args,
60 debug=self._loop.get_debug()))
61 if self._source_traceback:
62 frame = self._source_traceback[-1]
63 info.append(f'created at {frame[0]}:{frame[1]}')
64 return info
66 def __repr__(self):
67 if self._repr is not None:
68 return self._repr
69 info = self._repr_info()
70 return '<{}>'.format(' '.join(info))
72 def get_context(self):
73 return self._context
75 def cancel(self):
76 if not self._cancelled:
77 self._cancelled = True
78 if self._loop.get_debug():
79 # Keep a representation in debug mode to keep callback and
80 # parameters. For example, to log the warning
81 # "Executing <Handle...> took 2.5 second"
82 self._repr = repr(self)
83 self._callback = None
84 self._args = None
86 def cancelled(self):
87 return self._cancelled
89 def _run(self):
90 try:
91 self._context.run(self._callback, *self._args)
92 except (SystemExit, KeyboardInterrupt):
93 raise
94 except BaseException as exc:
95 cb = format_helpers._format_callback_source(
96 self._callback, self._args,
97 debug=self._loop.get_debug())
98 msg = f'Exception in callback {cb}'
99 context = {
100 'message': msg,
101 'exception': exc,
102 'handle': self,
103 }
104 if self._source_traceback:
105 context['source_traceback'] = self._source_traceback
106 self._loop.call_exception_handler(context)
107 self = None # Needed to break cycles when an exception occurs.
109# _ThreadSafeHandle is used for callbacks scheduled with call_soon_threadsafe
110# and is thread safe unlike Handle which is not thread safe.
111class _ThreadSafeHandle(Handle):
113 __slots__ = ('_lock',)
115 def __init__(self, callback, args, loop, context=None):
116 super().__init__(callback, args, loop, context)
117 self._lock = threading.RLock()
119 def cancel(self):
120 with self._lock:
121 return super().cancel()
123 def cancelled(self):
124 with self._lock:
125 return super().cancelled()
127 def _run(self):
128 # The event loop checks for cancellation without holding the lock
129 # It is possible that the handle is cancelled after the check
130 # but before the callback is called so check it again after acquiring
131 # the lock and return without calling the callback if it is cancelled.
132 with self._lock:
133 if self._cancelled: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true
134 return
135 return super()._run()
138class TimerHandle(Handle):
139 """Object returned by timed callback registration methods."""
141 __slots__ = ['_scheduled', '_when']
143 def __init__(self, when, callback, args, loop, context=None):
144 super().__init__(callback, args, loop, context)
145 if self._source_traceback:
146 del self._source_traceback[-1]
147 self._when = when
148 self._scheduled = False
150 def _repr_info(self):
151 info = super()._repr_info()
152 pos = 2 if self._cancelled else 1
153 info.insert(pos, f'when={self._when}')
154 return info
156 def __hash__(self):
157 return hash(self._when)
159 def __lt__(self, other):
160 if isinstance(other, TimerHandle):
161 return self._when < other._when
162 return NotImplemented
164 def __le__(self, other):
165 if isinstance(other, TimerHandle):
166 return self._when < other._when or self.__eq__(other)
167 return NotImplemented
169 def __gt__(self, other):
170 if isinstance(other, TimerHandle):
171 return self._when > other._when
172 return NotImplemented
174 def __ge__(self, other):
175 if isinstance(other, TimerHandle):
176 return self._when > other._when or self.__eq__(other)
177 return NotImplemented
179 def __eq__(self, other):
180 if isinstance(other, TimerHandle):
181 return (self._when == other._when and
182 self._callback == other._callback and
183 self._args == other._args and
184 self._cancelled == other._cancelled)
185 return NotImplemented
187 def cancel(self):
188 if not self._cancelled: 188 ↛ 190line 188 didn't jump to line 190 because the condition on line 188 was always true
189 self._loop._timer_handle_cancelled(self)
190 super().cancel()
192 def when(self):
193 """Return a scheduled callback time.
195 The time is an absolute timestamp, using the same time
196 reference as loop.time().
197 """
198 return self._when
201class AbstractServer:
202 """Abstract server returned by create_server()."""
204 def close(self):
205 """Stop serving. This leaves existing connections open."""
206 raise NotImplementedError
208 def close_clients(self):
209 """Close all active connections."""
210 raise NotImplementedError
212 def abort_clients(self):
213 """Close all active connections immediately."""
214 raise NotImplementedError
216 def get_loop(self):
217 """Get the event loop the Server object is attached to."""
218 raise NotImplementedError
220 def is_serving(self):
221 """Return True if the server is accepting connections."""
222 raise NotImplementedError
224 async def start_serving(self):
225 """Start accepting connections.
227 This method is idempotent, so it can be called when
228 the server is already being serving.
229 """
230 raise NotImplementedError
232 async def serve_forever(self):
233 """Start accepting connections until the coroutine is cancelled.
235 The server is closed when the coroutine is cancelled.
236 """
237 raise NotImplementedError
239 async def wait_closed(self):
240 """Coroutine to wait until service is closed."""
241 raise NotImplementedError
243 async def __aenter__(self):
244 return self
246 async def __aexit__(self, *exc):
247 self.close()
248 await self.wait_closed()
251class AbstractEventLoop:
252 """Abstract event loop."""
254 # Running and stopping the event loop.
256 def run_forever(self):
257 """Run the event loop until stop() is called."""
258 raise NotImplementedError
260 def run_until_complete(self, future):
261 """Run the event loop until a Future is done.
263 Return the Future's result, or raise its exception.
264 """
265 raise NotImplementedError
267 def stop(self):
268 """Stop the event loop as soon as reasonable.
270 Exactly how soon that is may depend on the implementation, but
271 no more I/O callbacks should be scheduled.
272 """
273 raise NotImplementedError
275 def is_running(self):
276 """Return whether the event loop is currently running."""
277 raise NotImplementedError
279 def is_closed(self):
280 """Returns True if the event loop was closed."""
281 raise NotImplementedError
283 def close(self):
284 """Close the loop.
286 The loop should not be running.
288 This is idempotent and irreversible.
290 No other methods should be called after this one.
291 """
292 raise NotImplementedError
294 async def shutdown_asyncgens(self):
295 """Shutdown all active asynchronous generators."""
296 raise NotImplementedError
298 async def shutdown_default_executor(self):
299 """Schedule the shutdown of the default executor."""
300 raise NotImplementedError
302 # Methods scheduling callbacks. All these return Handles.
304 def _timer_handle_cancelled(self, handle):
305 """Notification that a TimerHandle has been cancelled."""
306 raise NotImplementedError
308 def call_soon(self, callback, *args, context=None):
309 return self.call_later(0, callback, *args, context=context)
311 def call_later(self, delay, callback, *args, context=None):
312 raise NotImplementedError
314 def call_at(self, when, callback, *args, context=None):
315 raise NotImplementedError
317 def time(self):
318 raise NotImplementedError
320 def create_future(self):
321 raise NotImplementedError
323 # Method scheduling a coroutine object: create a task.
325 def create_task(self, coro, **kwargs):
326 raise NotImplementedError
328 # Methods for interacting with threads.
330 def call_soon_threadsafe(self, callback, *args, context=None):
331 raise NotImplementedError
333 def run_in_executor(self, executor, func, *args):
334 raise NotImplementedError
336 def set_default_executor(self, executor):
337 raise NotImplementedError
339 # Network I/O methods returning Futures.
341 async def getaddrinfo(self, host, port, *,
342 family=0, type=0, proto=0, flags=0):
343 raise NotImplementedError
345 async def getnameinfo(self, sockaddr, flags=0):
346 raise NotImplementedError
348 async def create_connection(
349 self, protocol_factory, host=None, port=None,
350 *, ssl=None, family=0, proto=0,
351 flags=0, sock=None, local_addr=None,
352 server_hostname=None,
353 ssl_handshake_timeout=None,
354 ssl_shutdown_timeout=None,
355 happy_eyeballs_delay=None, interleave=None):
356 raise NotImplementedError
358 async def create_server(
359 self, protocol_factory, host=None, port=None,
360 *, family=socket.AF_UNSPEC,
361 flags=socket.AI_PASSIVE, sock=None, backlog=100,
362 ssl=None, reuse_address=None, reuse_port=None,
363 keep_alive=None,
364 ssl_handshake_timeout=None,
365 ssl_shutdown_timeout=None,
366 start_serving=True):
367 """A coroutine which creates a TCP server bound to host and port.
369 The return value is a Server object which can be used to stop
370 the service.
372 If host is an empty string or None all interfaces are assumed
373 and a list of multiple sockets will be returned (most likely
374 one for IPv4 and another one for IPv6). The host parameter can also
375 be a sequence (e.g. list) of hosts to bind to.
377 family can be set to either AF_INET or AF_INET6 to force the
378 socket to use IPv4 or IPv6. If not set it will be determined
379 from host (defaults to AF_UNSPEC).
381 flags is a bitmask for getaddrinfo().
383 sock can optionally be specified in order to use a preexisting
384 socket object.
386 backlog is the maximum number of queued connections passed to
387 listen() (defaults to 100).
389 ssl can be set to an SSLContext to enable SSL over the
390 accepted connections.
392 reuse_address tells the kernel to reuse a local socket in
393 TIME_WAIT state, without waiting for its natural timeout to
394 expire. If not specified will automatically be set to True on
395 UNIX.
397 reuse_port tells the kernel to allow this endpoint to be bound to
398 the same port as other existing endpoints are bound to, so long as
399 they all set this flag when being created. This option is not
400 supported on Windows.
402 keep_alive set to True keeps connections active by enabling the
403 periodic transmission of messages.
405 ssl_handshake_timeout is the time in seconds that an SSL server
406 will wait for completion of the SSL handshake before aborting the
407 connection. Default is 60s.
409 ssl_shutdown_timeout is the time in seconds that an SSL server
410 will wait for completion of the SSL shutdown procedure
411 before aborting the connection. Default is 30s.
413 start_serving set to True (default) causes the created server
414 to start accepting connections immediately. When set to False,
415 the user should await Server.start_serving() or
416 Server.serve_forever() to make the server to start accepting
417 connections.
418 """
419 raise NotImplementedError
421 async def sendfile(self, transport, file, offset=0, count=None,
422 *, fallback=True):
423 """Send a file through a transport.
425 Return an amount of sent bytes.
426 """
427 raise NotImplementedError
429 async def start_tls(self, transport, protocol, sslcontext, *,
430 server_side=False,
431 server_hostname=None,
432 ssl_handshake_timeout=None,
433 ssl_shutdown_timeout=None):
434 """Upgrade a transport to TLS.
436 Return a new transport that *protocol* should start using
437 immediately.
438 """
439 raise NotImplementedError
441 async def create_unix_connection(
442 self, protocol_factory, path=None, *,
443 ssl=None, sock=None,
444 server_hostname=None,
445 ssl_handshake_timeout=None,
446 ssl_shutdown_timeout=None):
447 raise NotImplementedError
449 async def create_unix_server(
450 self, protocol_factory, path=None, *,
451 sock=None, backlog=100, ssl=None,
452 ssl_handshake_timeout=None,
453 ssl_shutdown_timeout=None,
454 start_serving=True, mode=None):
455 """A coroutine which creates a UNIX Domain Socket server.
457 The return value is a Server object, which can be used to stop
458 the service.
460 path is a str, representing a file system path to bind the
461 server socket to.
463 sock can optionally be specified in order to use a preexisting
464 socket object.
466 backlog is the maximum number of queued connections passed to
467 listen() (defaults to 100).
469 ssl can be set to an SSLContext to enable SSL over the
470 accepted connections.
472 ssl_handshake_timeout is the time in seconds that an SSL server
473 will wait for the SSL handshake to complete (defaults to 60s).
475 ssl_shutdown_timeout is the time in seconds that an SSL server
476 will wait for the SSL shutdown to finish (defaults to 30s).
478 start_serving set to True (default) causes the created server
479 to start accepting connections immediately. When set to False,
480 the user should await Server.start_serving() or
481 Server.serve_forever() to make the server to start accepting
482 connections.
484 mode, if not None, is applied to the socket file created for
485 path with os.chmod() after binding and before the server
486 starts accepting connections.
487 """
488 raise NotImplementedError
490 async def connect_accepted_socket(
491 self, protocol_factory, sock,
492 *, ssl=None,
493 ssl_handshake_timeout=None,
494 ssl_shutdown_timeout=None):
495 """Handle an accepted connection.
497 This is used by servers that accept connections outside of
498 asyncio, but use asyncio to handle connections.
500 This method is a coroutine. When completed, the coroutine
501 returns a (transport, protocol) pair.
502 """
503 raise NotImplementedError
505 async def create_datagram_endpoint(self, protocol_factory,
506 local_addr=None, remote_addr=None, *,
507 family=0, proto=0, flags=0,
508 reuse_address=None, reuse_port=None,
509 allow_broadcast=None, sock=None):
510 """A coroutine which creates a datagram endpoint.
512 This method will try to establish the endpoint in the background.
513 When successful, the coroutine returns a (transport, protocol) pair.
515 protocol_factory must be a callable returning a protocol instance.
517 socket family AF_INET, socket.AF_INET6 or socket.AF_UNIX depending
518 on host (or family if specified), socket type SOCK_DGRAM.
520 reuse_address tells the kernel to reuse a local socket in
521 TIME_WAIT state, without waiting for its natural timeout to
522 expire. If not specified it will automatically be set to True on
523 UNIX.
525 reuse_port tells the kernel to allow this endpoint to be bound to
526 the same port as other existing endpoints are bound to, so long as
527 they all set this flag when being created. This option is not
528 supported on Windows and some UNIX's. If the
529 :py:data:`~socket.SO_REUSEPORT` constant is not defined then this
530 capability is unsupported.
532 allow_broadcast tells the kernel to allow this endpoint to send
533 messages to the broadcast address.
535 sock can optionally be specified in order to use a preexisting
536 socket object.
537 """
538 raise NotImplementedError
540 # Pipes and subprocesses.
542 async def connect_read_pipe(self, protocol_factory, pipe):
543 """Register read pipe in event loop. Set the pipe to non-blocking mode.
545 protocol_factory should instantiate object with Protocol interface.
546 pipe is a file-like object.
547 Return pair (transport, protocol), where transport supports the
548 ReadTransport interface."""
549 # The reason to accept file-like object instead of just file descriptor
550 # is: we need to own pipe and close it at transport finishing
551 # Can got complicated errors if pass f.fileno(),
552 # close fd in pipe transport then close f and vice versa.
553 raise NotImplementedError
555 async def connect_write_pipe(self, protocol_factory, pipe):
556 """Register write pipe in event loop.
558 protocol_factory should instantiate object with BaseProtocol
559 interface.
560 Pipe is file-like object already switched to nonblocking.
561 Return pair (transport, protocol), where transport support
562 WriteTransport interface."""
563 # The reason to accept file-like object instead of just file descriptor
564 # is: we need to own pipe and close it at transport finishing
565 # Can got complicated errors if pass f.fileno(),
566 # close fd in pipe transport then close f and vice versa.
567 raise NotImplementedError
569 async def subprocess_shell(self, protocol_factory, cmd, *,
570 stdin=subprocess.PIPE,
571 stdout=subprocess.PIPE,
572 stderr=subprocess.PIPE,
573 **kwargs):
574 raise NotImplementedError
576 async def subprocess_exec(self, protocol_factory, *args,
577 stdin=subprocess.PIPE,
578 stdout=subprocess.PIPE,
579 stderr=subprocess.PIPE,
580 **kwargs):
581 raise NotImplementedError
583 # Ready-based callback registration methods.
584 # The add_*() methods return None.
585 # The remove_*() methods return True if something was removed,
586 # False if there was nothing to delete.
588 def add_reader(self, fd, callback, *args):
589 raise NotImplementedError
591 def remove_reader(self, fd):
592 raise NotImplementedError
594 def add_writer(self, fd, callback, *args):
595 raise NotImplementedError
597 def remove_writer(self, fd):
598 raise NotImplementedError
600 # Completion based I/O methods returning Futures.
602 async def sock_recv(self, sock, nbytes):
603 raise NotImplementedError
605 async def sock_recv_into(self, sock, buf):
606 raise NotImplementedError
608 async def sock_recvfrom(self, sock, bufsize):
609 raise NotImplementedError
611 async def sock_recvfrom_into(self, sock, buf, nbytes=0):
612 raise NotImplementedError
614 async def sock_sendall(self, sock, data):
615 raise NotImplementedError
617 async def sock_sendto(self, sock, data, address):
618 raise NotImplementedError
620 async def sock_connect(self, sock, address):
621 raise NotImplementedError
623 async def sock_accept(self, sock):
624 raise NotImplementedError
626 async def sock_sendfile(self, sock, file, offset=0, count=None,
627 *, fallback=None):
628 raise NotImplementedError
630 # Signal handling.
632 def add_signal_handler(self, sig, callback, *args):
633 raise NotImplementedError
635 def remove_signal_handler(self, sig):
636 raise NotImplementedError
638 # Task factory.
640 def set_task_factory(self, factory):
641 raise NotImplementedError
643 def get_task_factory(self):
644 raise NotImplementedError
646 # Error handlers.
648 def get_exception_handler(self):
649 raise NotImplementedError
651 def set_exception_handler(self, handler):
652 raise NotImplementedError
654 def default_exception_handler(self, context):
655 raise NotImplementedError
657 def call_exception_handler(self, context):
658 raise NotImplementedError
660 # Debug flag management.
662 def get_debug(self):
663 raise NotImplementedError
665 def set_debug(self, enabled):
666 raise NotImplementedError
669class _Local(threading.local):
670 _loop = None
673_local = _Local()
676# A TLS for the running event loop, used by _get_running_loop.
677class _RunningLoop(threading.local):
678 loop_pid = (None, None)
681_running_loop = _RunningLoop()
684def get_running_loop():
685 """Return the running event loop. Raise a RuntimeError if there is none.
687 This function is thread-specific.
688 """
689 # NOTE: this function is implemented in C (see _asynciomodule.c)
690 loop = _get_running_loop()
691 if loop is None:
692 raise RuntimeError('no running event loop')
693 return loop
696def _get_running_loop():
697 """Return the running event loop or None.
699 This is a low-level function intended to be used by event loops.
700 This function is thread-specific.
701 """
702 # NOTE: this function is implemented in C (see _asynciomodule.c)
703 running_loop, pid = _running_loop.loop_pid
704 if running_loop is not None and pid == os.getpid():
705 return running_loop
708def _set_running_loop(loop):
709 """Set the running event loop.
711 This is a low-level function intended to be used by event loops.
712 This function is thread-specific.
713 """
714 # NOTE: this function is implemented in C (see _asynciomodule.c)
715 _running_loop.loop_pid = (loop, os.getpid())
718def _get_event_loop():
719 """Return the event loop set for the current thread.
721 Raise a RuntimeError if no event loop has been set for the current
722 thread. This is the slow path of get_event_loop(); the running loop
723 is checked by the caller.
724 """
725 loop = _local._loop
726 if loop is None:
727 raise RuntimeError('There is no current event loop in thread %r.'
728 % threading.current_thread().name)
729 return loop
732def get_event_loop():
733 """Return an asyncio event loop.
735 When called from a coroutine or a callback (e.g. scheduled with call_soon
736 or similar API), this function will always return the running event loop.
738 If there is no running event loop set, the function will return
739 the loop set by ``set_event_loop()``, or raise a RuntimeError if
740 no loop has been set.
741 """
742 # NOTE: this function is implemented in C (see _asynciomodule.c)
743 current_loop = _get_running_loop()
744 if current_loop is not None:
745 return current_loop
746 return _get_event_loop()
749def set_event_loop(loop):
750 """Set the event loop for the current thread to loop.
752 If loop is None, the current event loop is unset.
753 """
754 if loop is not None and not isinstance(loop, AbstractEventLoop):
755 raise TypeError(f"loop must be an instance of AbstractEventLoop or None, not '{type(loop).__name__}'")
756 _local._loop = loop
759def new_event_loop():
760 """Create and return a new event loop object."""
761 if sys.platform == 'win32': 761 ↛ 762line 761 didn't jump to line 762 because the condition on line 761 was never true
762 from .windows_events import EventLoop
763 else:
764 from .unix_events import EventLoop
765 return EventLoop()
768# Alias pure-Python implementations for testing purposes.
769_py__get_running_loop = _get_running_loop
770_py__set_running_loop = _set_running_loop
771_py_get_running_loop = get_running_loop
772_py_get_event_loop = get_event_loop
775try:
776 # get_event_loop() is one of the most frequently called
777 # functions in asyncio. Pure Python implementation is
778 # about 4 times slower than C-accelerated.
779 from _asyncio import (_get_running_loop, _set_running_loop,
780 get_running_loop, get_event_loop)
781except ImportError:
782 pass
783else:
784 # Alias C implementations for testing purposes.
785 _c__get_running_loop = _get_running_loop
786 _c__set_running_loop = _set_running_loop
787 _c_get_running_loop = get_running_loop
788 _c_get_event_loop = get_event_loop
791if hasattr(os, 'fork'): 791 ↛ exitline 791 didn't exit the module because the condition on line 791 was always true
792 def on_fork():
793 # Reset the loop and wakeupfd in the forked child process.
794 global _local
795 _local = _Local()
796 _set_running_loop(None)
797 signal.set_wakeup_fd(-1)
799 os.register_at_fork(after_in_child=on_fork)