# cython: language_level=3, embedsignature=True
import asyncio
cimport cython
from .includes.debug cimport UVLOOP_DEBUG
from .includes cimport uv
from .includes cimport system
from .includes.python cimport (
PY_VERSION_HEX,
PyMem_RawMalloc, PyMem_RawFree,
PyMem_RawCalloc, PyMem_RawRealloc,
PyUnicode_EncodeFSDefault,
PyErr_SetInterrupt,
_Py_RestoreSignals,
Context_CopyCurrent,
Context_Enter,
Context_Exit,
PyMemoryView_FromMemory, PyBUF_WRITE,
PyMemoryView_FromObject, PyMemoryView_Check,
PyOS_AfterFork_Parent, PyOS_AfterFork_Child,
PyOS_BeforeFork,
PyUnicode_FromString
)
from .includes.flowcontrol cimport add_flowcontrol_defaults
from libc.stdint cimport uint64_t
from libc.string cimport memset, strerror, memcpy
from libc cimport errno
from cpython cimport PyObject
from cpython cimport PyErr_CheckSignals, PyErr_Occurred
from cpython cimport PyThread_get_thread_ident
from cpython cimport Py_INCREF, Py_DECREF, Py_XDECREF, Py_XINCREF
from cpython cimport (
PyObject_GetBuffer, PyBuffer_Release, PyBUF_SIMPLE,
Py_buffer, PyBytes_AsString, PyBytes_CheckExact,
PyBytes_AsStringAndSize,
Py_SIZE, PyBytes_AS_STRING, PyBUF_WRITABLE
)
from cpython.pycapsule cimport PyCapsule_New, PyCapsule_GetPointer
from . import _noop
include "includes/consts.pxi"
include "includes/stdlib.pxi"
include "errors.pyx"
cdef:
int PY39 = PY_VERSION_HEX >= 0x03090000
int PY311 = PY_VERSION_HEX >= 0x030b0000
uint64_t MAX_SLEEP = 3600 * 24 * 365 * 100
cdef _is_sock_stream(sock_type):
if SOCK_NONBLOCK == -1:
return sock_type == uv.SOCK_STREAM
else:
# Linux's socket.type is a bitmask that can include extra info
# about socket (like SOCK_NONBLOCK bit), therefore we can't do simple
# `sock_type == socket.SOCK_STREAM`, see
# https://github.com/torvalds/linux/blob/v4.13/include/linux/net.h#L77
# for more details.
return (sock_type & 0xF) == uv.SOCK_STREAM
cdef _is_sock_dgram(sock_type):
if SOCK_NONBLOCK == -1:
return sock_type == uv.SOCK_DGRAM
else:
# Read the comment in `_is_sock_stream`.
return (sock_type & 0xF) == uv.SOCK_DGRAM
cdef isfuture(obj):
if aio_isfuture is None:
return isinstance(obj, aio_Future)
else:
return aio_isfuture(obj)
cdef inline socket_inc_io_ref(sock):
if isinstance(sock, socket_socket):
sock._io_refs += 1
cdef inline socket_dec_io_ref(sock):
if isinstance(sock, socket_socket):
sock._decref_socketios()
cdef inline run_in_context(context, method):
# This method is internally used to workaround a reference issue that in
# certain circumstances, inlined context.run() will not hold a reference to
# the given method instance, which - if deallocated - will cause segfault.
# See also: edgedb/edgedb#2222
Py_INCREF(method)
try:
return context.run(method)
finally:
Py_DECREF(method)
cdef inline run_in_context1(context, method, arg):
Py_INCREF(method)
try:
return context.run(method, arg)
finally:
Py_DECREF(method)
cdef inline run_in_context2(context, method, arg1, arg2):
Py_INCREF(method)
try:
return context.run(method, arg1, arg2)
finally:
Py_DECREF(method)
# Used for deprecation and removal of `loop.create_datagram_endpoint()`'s
# *reuse_address* parameter
_unset = object()
@cython.no_gc_clear
cdef class Loop:
def __cinit__(self):
cdef int err
# Install PyMem* memory allocators if they aren't installed yet.
__install_pymem()
# Install pthread_atfork handlers
__install_atfork()
self.uvloop = <uv.uv_loop_t*>PyMem_RawMalloc(sizeof(uv.uv_loop_t))
if self.uvloop is NULL:
raise MemoryError()
self.slow_callback_duration = 0.1
self._closed = 0
self._debug = 0
self._thread_id = 0
self._running = 0
self._stopping = 0
self._transports = weakref_WeakValueDictionary()
self._processes = set()
# Used to keep a reference (and hence keep the fileobj alive)
# for as long as its registered by add_reader or add_writer.
# This is how the selector module and hence asyncio behaves.
self._fd_to_reader_fileobj = {}
self._fd_to_writer_fileobj = {}
self._timers = set()
self._polls = {}
self._recv_buffer_in_use = 0
err = uv.uv_loop_init(self.uvloop)
if err < 0:
raise convert_error(err)
self.uvloop.data = <void*> self
self._init_debug_fields()
self.active_process_handler = None
self._last_error = None
self._task_factory = None
self._exception_handler = None
self._default_executor = None
self._queued_streams = set()
self._executing_streams = set()
self._ready = col_deque()
self._ready_len = 0
self.handler_async = UVAsync.new(
self, <method_t>self._on_wake, self)
self.handler_idle = UVIdle.new(
self,
new_MethodHandle(
self, "loop._on_idle", <method_t>self._on_idle, None, self))
# Needed to call `UVStream._exec_write` for writes scheduled
# during `Protocol.data_received`.
self.handler_check__exec_writes = UVCheck.new(
self,
new_MethodHandle(
self, "loop._exec_queued_writes",
<method_t>self._exec_queued_writes, None, self))
self._signals = set()
self._ssock = self._csock = None
self._signal_handlers = {}
self._listening_signals = False
self._old_signal_wakeup_id = -1
self._coroutine_debug_set = False
# A weak set of all asynchronous generators that are
# being iterated by the loop.
self._asyncgens = weakref_WeakSet()
# Set to True when `loop.shutdown_asyncgens` is called.
self._asyncgens_shutdown_called = False
# Set to True when `loop.shutdown_default_executor` is called.
self._executor_shutdown_called = False
self._servers = set()
cdef inline _is_main_thread(self):
cdef uint64_t main_thread_id = system.MAIN_THREAD_ID
if system.MAIN_THREAD_ID_SET == 0:
main_thread_id = <uint64_t>threading_main_thread().ident
system.setMainThreadID(main_thread_id)
return main_thread_id == PyThread_get_thread_ident()
def __init__(self):
self.set_debug(
sys_dev_mode or (not sys_ignore_environment
and bool(os_environ.get('PYTHONASYNCIODEBUG'))))
def __dealloc__(self):
if self._running == 1:
raise RuntimeError('deallocating a running event loop!')
if self._closed == 0:
aio_logger.error("deallocating an open event loop")
return
PyMem_RawFree(self.uvloop)
self.uvloop = NULL
cdef _init_debug_fields(self):
self._debug_cc = bool(UVLOOP_DEBUG)
if UVLOOP_DEBUG:
self._debug_handles_current = col_Counter()
self._debug_handles_closed = col_Counter()
self._debug_handles_total = col_Counter()
else:
self._debug_handles_current = None
self._debug_handles_closed = None
self._debug_handles_total = None
self._debug_uv_handles_total = 0
self._debug_uv_handles_freed = 0
self._debug_stream_read_cb_total = 0
self._debug_stream_read_eof_total = 0
self._debug_stream_read_errors_total = 0
self._debug_stream_read_cb_errors_total = 0
self._debug_stream_read_eof_cb_errors_total = 0
self._debug_stream_shutdown_errors_total = 0
self._debug_stream_listen_errors_total = 0
self._debug_stream_write_tries = 0
self._debug_stream_write_errors_total = 0
self._debug_stream_write_ctx_total = 0
self._debug_stream_write_ctx_cnt = 0
self._debug_stream_write_cb_errors_total = 0
self._debug_cb_handles_total = 0
self._debug_cb_handles_count = 0
self._debug_cb_timer_handles_total = 0
self._debug_cb_timer_handles_count = 0
self._poll_read_events_total = 0
self._poll_read_cb_errors_total = 0
self._poll_write_events_total = 0
self._poll_write_cb_errors_total = 0
self._sock_try_write_total = 0
self._debug_exception_handler_cnt = 0
cdef _setup_or_resume_signals(self):
if not self._is_main_thread():
return
if self._listening_signals:
raise RuntimeError('signals handling has been already setup')
if self._ssock is not None:
raise RuntimeError('self-pipe exists before loop run')
# Create a self-pipe and call set_signal_wakeup_fd() with one
# of its ends. This is needed so that libuv knows that it needs
# to wakeup on ^C (no matter if the SIGINT handler is still the
# standard Python's one or or user set their own.)
self._ssock, self._csock = socket_socketpair()
try:
self._ssock.setblocking(False)
self._csock.setblocking(False)
fileno = self._csock.fileno()
self._old_signal_wakeup_id = _set_signal_wakeup_fd(fileno)
except Exception:
# Out of all statements in the try block, only the
# "_set_signal_wakeup_fd()" call can fail, but it shouldn't,
# as we ensure that the current thread is the main thread.
# Still, if something goes horribly wrong we want to clean up
# the socket pair.
self._ssock.close()
self._csock.close()
self._ssock = None
self._csock = None
raise
self._add_reader(
self._ssock,
new_MethodHandle(
self,
"Loop._read_from_self",
<method_t>self._read_from_self,
None,
self))
self._listening_signals = True
cdef _pause_signals(self):
if not self._is_main_thread():
if self._listening_signals:
raise RuntimeError(
'cannot pause signals handling; no longer running in '
'the main thread')
else:
return
if not self._listening_signals:
raise RuntimeError('signals handling has not been setup')
self._listening_signals = False
_set_signal_wakeup_fd(self._old_signal_wakeup_id)
self._remove_reader(self._ssock)
self._ssock.close()
self._csock.close()
self._ssock = None
self._csock = None
cdef _shutdown_signals(self):
if not self._is_main_thread():
if self._signal_handlers:
aio_logger.warning(
'cannot cleanup signal handlers: closing the event loop '
'in a non-main OS thread')
return
if self._listening_signals:
raise RuntimeError(
'cannot shutdown signals handling as it has not been paused')
if self._ssock:
raise RuntimeError(
'self-pipe was not cleaned up after loop was run')
for sig in list(self._signal_handlers):
self.remove_signal_handler(sig)
def __sighandler(self, signum, frame):
self._signals.add(signum)
cdef inline _ceval_process_signals(self):
# Invoke CPython eval loop to let process signals.
PyErr_CheckSignals()
# Calling a pure-Python function will invoke
# _PyEval_EvalFrameDefault which will process
# pending signal callbacks.
_noop.noop() # Might raise ^C
cdef _read_from_self(self):
cdef bytes sigdata
sigdata = b''
while True:
try:
data = self._ssock.recv(65536)
if not data:
break
sigdata += data
except InterruptedError:
continue
except BlockingIOError:
break
if sigdata:
self._invoke_signals(sigdata)
cdef _invoke_signals(self, bytes data):
cdef set sigs
self._ceval_process_signals()
sigs = self._signals.copy()
self._signals.clear()
for signum in data:
if not signum:
# ignore null bytes written by set_wakeup_fd()
continue
sigs.discard(signum)
self._handle_signal(signum)
for signum in sigs:
# Since not all signals are registered by add_signal_handler()
# (for instance, we use the default SIGINT handler) not all
# signals will trigger loop.__sighandler() callback. Therefore
# we combine two datasources: one is self-pipe, one is data
# from __sighandler; this ensures that signals shouldn't be
# lost even if set_wakeup_fd() couldn't write to the self-pipe.
self._handle_signal(signum)
cdef _handle_signal(self, sig):
cdef Handle handle
try:
handle = <Handle>(self._signal_handlers[sig])
except KeyError:
handle = None
if handle is None:
self._ceval_process_signals()
return
if handle._cancelled:
self.remove_signal_handler(sig) # Remove it properly.
else:
self._append_ready_handle(handle)
self.handler_async.send()
cdef _on_wake(self):
if ((self._ready_len > 0 or self._stopping) and
not self.handler_idle.running):
self.handler_idle.start()
cdef _on_idle(self):
cdef:
int i, ntodo
object popleft = self._ready.popleft
Handle handler
ntodo = len(self._ready)
if self._debug:
for i from 0 <= i < ntodo:
handler = <Handle> popleft()
if handler._cancelled == 0:
try:
started = time_monotonic()
handler._run()
except BaseException as ex:
self._stop(ex)
return
else:
delta = time_monotonic() - started
if delta > self.slow_callback_duration:
aio_logger.warning(
'Executing %s took %.3f seconds',
handler._format_handle(), delta)
else:
for i from 0 <= i < ntodo:
handler = <Handle> popleft()
if handler._cancelled == 0:
try:
handler._run()
except BaseException as ex:
self._stop(ex)
return
if len(self._queued_streams):
self._exec_queued_writes()
self._ready_len = len(self._ready)
if self._ready_len == 0 and self.handler_idle.running:
self.handler_idle.stop()
if self._stopping:
uv.uv_stop(self.uvloop) # void
cdef _stop(self, exc):
if exc is not None:
self._last_error = exc
if self._stopping == 1:
return
self._stopping = 1
if not self.handler_idle.running:
self.handler_idle.start()
cdef __run(self, uv.uv_run_mode mode):
# Although every UVHandle holds a reference to the loop,
# we want to do everything to ensure that the loop will
# never deallocate during the run -- so we do some
# manual refs management.
Py_INCREF(self)
with nogil:
err = uv.uv_run(self.uvloop, mode)
Py_DECREF(self)
if err < 0:
raise convert_error(err)
cdef _run(self, uv.uv_run_mode mode):
cdef int err
if self._closed == 1:
raise RuntimeError('unable to start the loop; it was closed')
if self._running == 1:
raise RuntimeError('this event loop is already running.')
if (aio_get_running_loop is not None and
aio_get_running_loop() is not None):
raise RuntimeError(
'Cannot run the event loop while another loop is running')
# reset _last_error
self._last_error = None
self._thread_id = PyThread_get_thread_ident()
self._running = 1
self.handler_check__exec_writes.start()
self.handler_idle.start()
self._setup_or_resume_signals()
if aio_set_running_loop is not None:
aio_set_running_loop(self)
try:
self.__run(mode)
finally:
if aio_set_running_loop is not None:
aio_set_running_loop(None)
self.handler_check__exec_writes.stop()
self.handler_idle.stop()
self._pause_signals()
self._thread_id = 0
self._running = 0
self._stopping = 0
if self._last_error is not None:
# The loop was stopped with an error with 'loop._stop(error)' call
raise self._last_error
cdef _close(self):
cdef int err
if self._running == 1:
raise RuntimeError("Cannot close a running event loop")
if self._closed == 1:
return
self._closed = 1
for cb_handle in self._ready:
cb_handle.cancel()
self._ready.clear()
self._ready_len = 0
if self._polls:
for poll_handle in self._polls.values():
(<UVHandle>poll_handle)._close()
self._polls.clear()
if self._timers:
for timer_cbhandle in tuple(self._timers):
timer_cbhandle.cancel()
# Close all remaining handles
self.handler_async._close()
self.handler_idle._close()
self.handler_check__exec_writes._close()
__close_all_handles(self)
self._shutdown_signals()
# During this run there should be no open handles,
# so it should finish right away
self.__run(uv.UV_RUN_DEFAULT)
if self._fd_to_writer_fileobj:
for fileobj in self._fd_to_writer_fileobj.values():
socket_dec_io_ref(fileobj)
self._fd_to_writer_fileobj.clear()
if self._fd_to_reader_fileobj:
for fileobj in self._fd_to_reader_fileobj.values():
socket_dec_io_ref(fileobj)
self._fd_to_reader_fileobj.clear()
if self._timers:
raise RuntimeError(
f"new timers were queued during loop closing: {self._timers}")
if self._polls:
raise RuntimeError(
f"new poll handles were queued during loop closing: "
f"{self._polls}")
if self._ready:
raise RuntimeError(
f"new callbacks were queued during loop closing: "
f"{self._ready}")
err = uv.uv_loop_close(self.uvloop)
if err < 0:
raise convert_error(err)
self.handler_async = None
self.handler_idle = None
self.handler_check__exec_writes = None
self._executor_shutdown_called = True
executor = self._default_executor
if executor is not None:
self._default_executor = None
executor.shutdown(wait=False)
cdef uint64_t _time(self):
# asyncio doesn't have a time cache, neither should uvloop.
uv.uv_update_time(self.uvloop) # void
return uv.uv_now(self.uvloop)
cdef inline _queue_write(self, UVStream stream):
self._queued_streams.add(stream)
if not self.handler_check__exec_writes.running:
self.handler_check__exec_writes.start()
cdef _exec_queued_writes(self):
if len(self._queued_streams) == 0:
if self.handler_check__exec_writes.running:
self.handler_check__exec_writes.stop()
return
cdef:
UVStream stream
streams = self._queued_streams
self._queued_streams = self._executing_streams
self._executing_streams = streams
try:
for pystream in streams:
stream = <UVStream>pystream
stream._exec_write()
finally:
streams.clear()
if self.handler_check__exec_writes.running:
if len(self._queued_streams) == 0:
self.handler_check__exec_writes.stop()
cdef inline _call_soon(self, object callback, object args, object context):
cdef Handle handle
handle = new_Handle(self, callback, args, context)
self._call_soon_handle(handle)
return handle
cdef inline _append_ready_handle(self, Handle handle):
self._check_closed()
self._ready.append(handle)
self._ready_len += 1
cdef inline _call_soon_handle(self, Handle handle):
self._append_ready_handle(handle)
if not self.handler_idle.running:
self.handler_idle.start()
cdef _call_later(self, uint64_t delay, object callback, object args,
object context):
return TimerHandle(self, callback, args, delay, context)
cdef void _handle_exception(self, object ex):
if isinstance(ex, Exception):
self.call_exception_handler({'exception': ex})
else:
# BaseException
self._last_error = ex
# Exit ASAP
self._stop(None)
cdef inline _check_signal(self, sig):
if not isinstance(sig, int):
raise TypeError('sig must be an int, not {!r}'.format(sig))
if not (1 <= sig < signal_NSIG):
raise ValueError(
'sig {} out of range(1, {})'.format(sig, signal_NSIG))
cdef inline _check_closed(self):
if self._closed == 1:
raise RuntimeError('Event loop is closed')
cdef inline _check_thread(self):
if self._thread_id == 0:
return
cdef uint64_t thread_id
thread_id = <uint64_t>PyThread_get_thread_ident()
if thread_id != self._thread_id:
raise RuntimeError(
"Non-thread-safe operation invoked on an event loop other "
"than the current one")
cdef inline _new_future(self):
return aio_Future(loop=self)
cdef _track_transport(self, UVBaseTransport transport):
self._transports[transport._fileno()] = transport
cdef _track_process(self, UVProcess proc):
self._processes.add(proc)
cdef _untrack_process(self, UVProcess proc):
self._processes.discard(proc)
cdef _fileobj_to_fd(self, fileobj):
"""Return a file descriptor from a file object.
Parameters:
fileobj -- file object or file descriptor
Returns:
corresponding file descriptor
Raises:
ValueError if the object is invalid
"""
# Copy of the `selectors._fileobj_to_fd()` function.
if isinstance(fileobj, int):
fd = fileobj
else:
try:
fd = int(fileobj.fileno())
except (AttributeError, TypeError, ValueError):
raise ValueError("Invalid file object: "
"{!r}".format(fileobj)) from None
if fd < 0:
raise ValueError("Invalid file descriptor: {}".format(fd))
return fd
cdef _ensure_fd_no_transport(self, fd):
cdef UVBaseTransport tr
try:
tr = <UVBaseTransport>(self._transports[fd])
except KeyError:
pass
else:
if tr._is_alive():
raise RuntimeError(
'File descriptor {!r} is used by transport {!r}'.format(
fd, tr))
cdef _add_reader(self, fileobj, Handle handle):
cdef:
UVPoll poll
self._check_closed()
fd = self._fileobj_to_fd(fileobj)
self._ensure_fd_no_transport(fd)
try:
poll = <UVPoll>(self._polls[fd])
except KeyError:
poll = UVPoll.new(self, fd)
self._polls[fd] = poll
poll.start_reading(handle)
old_fileobj = self._fd_to_reader_fileobj.pop(fd, None)
if old_fileobj is not None:
socket_dec_io_ref(old_fileobj)
self._fd_to_reader_fileobj[fd] = fileobj
socket_inc_io_ref(fileobj)
cdef _remove_reader(self, fileobj):
cdef:
UVPoll poll
fd = self._fileobj_to_fd(fileobj)
self._ensure_fd_no_transport(fd)
mapped_fileobj = self._fd_to_reader_fileobj.pop(fd, None)
if mapped_fileobj is not None:
socket_dec_io_ref(mapped_fileobj)
if self._closed == 1:
return False
try:
poll = <UVPoll>(self._polls[fd])
except KeyError:
return False
result = poll.stop_reading()
if not poll.is_active():
del self._polls[fd]
poll._close()
return result
cdef _has_reader(self, fileobj):
cdef:
UVPoll poll
self._check_closed()
fd = self._fileobj_to_fd(fileobj)
try:
poll = <UVPoll>(self._polls[fd])
except KeyError:
return False
return poll.is_reading()
cdef _add_writer(self, fileobj, Handle handle):
cdef:
UVPoll poll
self._check_closed()
fd = self._fileobj_to_fd(fileobj)
self._ensure_fd_no_transport(fd)
try:
poll = <UVPoll>(self._polls[fd])
except KeyError:
poll = UVPoll.new(self, fd)
self._polls[fd] = poll
poll.start_writing(handle)
old_fileobj = self._fd_to_writer_fileobj.pop(fd, None)
if old_fileobj is not None:
socket_dec_io_ref(old_fileobj)
self._fd_to_writer_fileobj[fd] = fileobj
socket_inc_io_ref(fileobj)
cdef _remove_writer(self, fileobj):
cdef:
UVPoll poll
fd = self._fileobj_to_fd(fileobj)
self._ensure_fd_no_transport(fd)
mapped_fileobj = self._fd_to_writer_fileobj.pop(fd, None)
if mapped_fileobj is not None:
socket_dec_io_ref(mapped_fileobj)
if self._closed == 1:
return False
try:
poll = <UVPoll>(self._polls[fd])
except KeyError:
return False
result = poll.stop_writing()
if not poll.is_active():
del self._polls[fd]
poll._close()
return result
cdef _has_writer(self, fileobj):
cdef:
UVPoll poll
self._check_closed()
fd = self._fileobj_to_fd(fileobj)
try:
poll = <UVPoll>(self._polls[fd])
except KeyError:
return False
return poll.is_writing()
cdef _getaddrinfo(self, object host, object port,
int family, int type,
int proto, int flags,
int unpack):
if isinstance(port, str):
port = port.encode()
elif isinstance(port, int):
port = str(port).encode()
if port is not None and not isinstance(port, bytes):
raise TypeError('port must be a str, bytes or int')
if isinstance(host, str):
host = host.encode('idna')
if host is not None:
if not isinstance(host, bytes):
raise TypeError('host must be a str or bytes')
fut = self._new_future()
def callback(result):
if AddrInfo.isinstance(result):
try:
if unpack == 0:
data = result
else:
data = (<AddrInfo>result).unpack()
except (KeyboardInterrupt, SystemExit):
raise
except BaseException as ex:
if not fut.cancelled():
fut.set_exception(ex)
else:
if not fut.cancelled():
fut.set_result(data)
else:
if not fut.cancelled():
fut.set_exception(result)
AddrInfoRequest(self, host, port, family, type, proto, flags, callback)
return fut
cdef _getnameinfo(self, system.sockaddr *addr, int flags):
cdef NameInfoRequest nr
fut = self._new_future()
def callback(result):
if isinstance(result, tuple):
fut.set_result(result)
else:
fut.set_exception(result)
nr = NameInfoRequest(self, callback)
nr.query(addr, flags)
return fut
cdef _sock_recv(self, fut, sock, n):
if UVLOOP_DEBUG:
if fut.cancelled():
# Shouldn't happen with _SyncSocketReaderFuture.
raise RuntimeError(
f'_sock_recv is called on a cancelled Future')
if not self._has_reader(sock):
raise RuntimeError(
f'socket {sock!r} does not have a reader '
f'in the _sock_recv callback')
try:
data = sock.recv(n)
except (BlockingIOError, InterruptedError):
# No need to re-add the reader, let's just wait until
# the poll handler calls this callback again.
pass
except (KeyboardInterrupt, SystemExit):
raise
except BaseException as exc:
fut.set_exception(exc)
self._remove_reader(sock)
else:
fut.set_result(data)
self._remove_reader(sock)
cdef _sock_recv_into(self, fut, sock, buf):
if UVLOOP_DEBUG:
if fut.cancelled():
# Shouldn't happen with _SyncSocketReaderFuture.
raise RuntimeError(
f'_sock_recv_into is called on a cancelled Future')
if not self._has_reader(sock):
raise RuntimeError(
f'socket {sock!r} does not have a reader '
f'in the _sock_recv_into callback')
try:
data = sock.recv_into(buf)
except (BlockingIOError, InterruptedError):
# No need to re-add the reader, let's just wait until
# the poll handler calls this callback again.
pass
except (KeyboardInterrupt, SystemExit):
raise
except BaseException as exc:
fut.set_exception(exc)
self._remove_reader(sock)
else:
fut.set_result(data)
self._remove_reader(sock)
cdef _sock_sendall(self, fut, sock, data):
cdef:
Handle handle
int n
if UVLOOP_DEBUG:
if fut.cancelled():
# Shouldn't happen with _SyncSocketWriterFuture.
raise RuntimeError(
f'_sock_sendall is called on a cancelled Future')
if not self._has_writer(sock):
raise RuntimeError(
f'socket {sock!r} does not have a writer '
f'in the _sock_sendall callback')
try:
n = sock.send(data)
except (BlockingIOError, InterruptedError):
# Try next time.
return
except (KeyboardInterrupt, SystemExit):
raise
except BaseException as exc:
fut.set_exception(exc)
self._remove_writer(sock)
return
self._remove_writer(sock)
if n == len(data):
fut.set_result(None)
else:
if n:
if not isinstance(data, memoryview):
data = memoryview(data)
data = data[n:]
handle = new_MethodHandle3(
self,
"Loop._sock_sendall",
<method3_t>self._sock_sendall,
None,
self,
fut, sock, data)
self._add_writer(sock, handle)
cdef _sock_accept(self, fut, sock):
try:
conn, address = sock.accept()
conn.setblocking(False)
except (BlockingIOError, InterruptedError):
# There is an active reader for _sock_accept, so
# do nothing, it will be called again.
pass
except (KeyboardInterrupt, SystemExit):
raise
except BaseException as exc:
fut.set_exception(exc)
self._remove_reader(sock)
else:
fut.set_result((conn, address))
self._remove_reader(sock)
cdef _sock_connect(self, sock, address):
cdef:
Handle handle
try:
sock.connect(address)
except (BlockingIOError, InterruptedError):
pass
else:
return
fut = _SyncSocketWriterFuture(sock, self)
handle = new_MethodHandle3(
self,
"Loop._sock_connect",
<method3_t>self._sock_connect_cb,
None,
self,
fut, sock, address)
self._add_writer(sock, handle)
return fut
cdef _sock_connect_cb(self, fut, sock, address):
if UVLOOP_DEBUG:
if fut.cancelled():
# Shouldn't happen with _SyncSocketWriterFuture.
raise RuntimeError(
f'_sock_connect_cb is called on a cancelled Future')
if not self._has_writer(sock):
raise RuntimeError(
f'socket {sock!r} does not have a writer '
f'in the _sock_connect_cb callback')
try:
err = sock.getsockopt(uv.SOL_SOCKET, uv.SO_ERROR)
if err != 0:
# Jump to any except clause below.
raise OSError(err, 'Connect call failed %s' % (address,))
except (BlockingIOError, InterruptedError):
# socket is still registered, the callback will be retried later
pass
except (KeyboardInterrupt, SystemExit):
raise
except BaseException as exc:
fut.set_exception(exc)
self._remove_writer(sock)
else:
fut.set_result(None)
self._remove_writer(sock)
cdef _sock_set_reuseport(self, int fd):
cdef:
int err
int reuseport_flag = 1
err = system.setsockopt(
fd,
uv.SOL_SOCKET,
SO_REUSEPORT,
<char*>&reuseport_flag,
sizeof(reuseport_flag))
if err < 0:
raise convert_error(-errno.errno)
cdef _set_coroutine_debug(self, bint enabled):
enabled = bool(enabled)
if self._coroutine_debug_set == enabled:
return
if enabled:
self._coroutine_origin_tracking_saved_depth = (
sys.get_coroutine_origin_tracking_depth())
sys.set_coroutine_origin_tracking_depth(
DEBUG_STACK_DEPTH)
else:
sys.set_coroutine_origin_tracking_depth(
self._coroutine_origin_tracking_saved_depth)
self._coroutine_debug_set = enabled
def _get_backend_id(self):
"""This method is used by uvloop tests and is not part of the API."""
return uv.uv_backend_fd(self.uvloop)
cdef _print_debug_info(self):
cdef:
int err
uv.uv_rusage_t rusage
err = uv.uv_getrusage(&rusage)
if err < 0:
raise convert_error(err)
# OS
print('---- Process info: -----')
print('Process memory: {}'.format(rusage.ru_maxrss))
print('Number of signals: {}'.format(rusage.ru_nsignals))
print('')
# Loop
print('--- Loop debug info: ---')
print('Loop time: {}'.format(self.time()))
print('Errors logged: {}'.format(
self._debug_exception_handler_cnt))
print()
print('Callback handles: {: <8} | {}'.format(
self._debug_cb_handles_count,
self._debug_cb_handles_total))
print('Timer handles: {: <8} | {}'.format(
self._debug_cb_timer_handles_count,
self._debug_cb_timer_handles_total))
print()
print(' alive | closed |')
print('UVHandles python | libuv | total')
print(' objs | handles |')
print('-------------------------------+---------+---------')
for name in sorted(self._debug_handles_total):
print(' {: <18} {: >7} | {: >7} | {: >7}'.format(
name,
self._debug_handles_current[name],
self._debug_handles_closed[name],
self._debug_handles_total[name]))
print()
print('uv_handle_t (current: {}; freed: {}; total: {})'.format(
self._debug_uv_handles_total - self._debug_uv_handles_freed,
self._debug_uv_handles_freed,
self._debug_uv_handles_total))
print()
print('--- Streams debug info: ---')
print('Write errors: {}'.format(
self._debug_stream_write_errors_total))
print('Write without poll: {}'.format(
self._debug_stream_write_tries))
print('Write contexts: {: <8} | {}'.format(
self._debug_stream_write_ctx_cnt,
self._debug_stream_write_ctx_total))
print('Write failed callbacks: {}'.format(
self._debug_stream_write_cb_errors_total))
print()
print('Read errors: {}'.format(
self._debug_stream_read_errors_total))
print('Read callbacks: {}'.format(
self._debug_stream_read_cb_total))
print('Read failed callbacks: {}'.format(
self._debug_stream_read_cb_errors_total))
print('Read EOFs: {}'.format(
self._debug_stream_read_eof_total))
print('Read EOF failed callbacks: {}'.format(
self._debug_stream_read_eof_cb_errors_total))
print()
print('Listen errors: {}'.format(
self._debug_stream_listen_errors_total))
print('Shutdown errors {}'.format(
self._debug_stream_shutdown_errors_total))
print()
print('--- Polls debug info: ---')
print('Read events: {}'.format(
self._poll_read_events_total))
print('Read callbacks failed: {}'.format(
self._poll_read_cb_errors_total))
print('Write events: {}'.format(
self._poll_write_events_total))
print('Write callbacks failed: {}'.format(
self._poll_write_cb_errors_total))
print()
print('--- Sock ops successful on 1st try: ---')
print('Socket try-writes: {}'.format(
self._sock_try_write_total))
print(flush=True)
property print_debug_info:
def __get__(self):
if UVLOOP_DEBUG:
return lambda: self._print_debug_info()
else:
raise AttributeError('print_debug_info')
# Public API
def __repr__(self):
return '<{}.{} running={} closed={} debug={}>'.format(
self.__class__.__module__,
self.__class__.__name__,
self.is_running(),
self.is_closed(),
self.get_debug()
)
def call_soon(self, callback, *args, context=None):
"""Arrange for a callback to be called as soon as possible.
This operates as a FIFO queue: callbacks are called in the
order in which they are registered. Each callback will be
called exactly once.
Any positional arguments after the callback will be passed to
the callback when it is called.
"""
if self._debug == 1:
self._check_thread()
if args:
return self._call_soon(callback, args, context)
else:
return self._call_soon(callback, None, context)
def call_soon_threadsafe(self, callback, *args, context=None):
"""Like call_soon(), but thread-safe."""
if not args:
args = None
cdef Handle handle = new_Handle(self, callback, args, context)
self._append_ready_handle(handle) # deque append is atomic
# libuv async handler is thread-safe while the idle handler is not -
# we only set the async handler here, which will start the idle handler
# in _on_wake() from the loop and eventually call the callback.
self.handler_async.send()
return handle
def call_later(self, delay, callback, *args, context=None):
"""Arrange for a callback to be called at a given time.
Return a Handle: an opaque object with a cancel() method that
can be used to cancel the call.
The delay can be an int or float, expressed in seconds. It is
always relative to the current time.
Each callback will be called exactly once. If two callbacks
are scheduled for exactly the same time, it undefined which
will be called first.
Any positional arguments after the callback will be passed to
the callback when it is called.
"""
cdef uint64_t when
self._check_closed()
if self._debug == 1:
self._check_thread()
if delay < 0:
delay = 0
elif delay == py_inf or delay > MAX_SLEEP:
# ~100 years sounds like a good approximation of
# infinity for a Python application.
delay = MAX_SLEEP
when = <uint64_t>round(delay * 1000)
if not args:
args = None
if when == 0:
return self._call_soon(callback, args, context)
else:
return self._call_later(when, callback, args, context)
def call_at(self, when, callback, *args, context=None):
"""Like call_later(), but uses an absolute time.
Absolute time corresponds to the event loop's time() method.
"""
return self.call_later(
when - self.time(), callback, *args, context=context)
def time(self):
"""Return the time according to the event loop's clock.
This is a float expressed in seconds since an epoch, but the
epoch, precision, accuracy and drift are unspecified and may
differ per event loop.
"""
return self._time() / 1000
def stop(self):
"""Stop running the event loop.
Every callback already scheduled will still run. This simply informs
run_forever to stop looping after a complete iteration.
"""
self._call_soon_handle(
new_MethodHandle1(
self,
"Loop._stop",
<method1_t>self._stop,
None,
self,
None))
def run_forever(self):
"""Run the event loop until stop() is called."""
self._check_closed()
mode = uv.UV_RUN_DEFAULT
if self._stopping:
# loop.stop() was called right before loop.run_forever().
# This is how asyncio loop behaves.
mode = uv.UV_RUN_NOWAIT
self._set_coroutine_debug(self._debug)
old_agen_hooks = sys.get_asyncgen_hooks()
sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook,
finalizer=self._asyncgen_finalizer_hook)
try:
self._run(mode)
finally:
self._set_coroutine_debug(False)
sys.set_asyncgen_hooks(*old_agen_hooks)
def close(self):
"""Close the event loop.
The event loop must not be running.
This is idempotent and irreversible.
No other methods should be called after this one.
"""
self._close()
def get_debug(self):
return bool(self._debug)
def set_debug(self, enabled):
self._debug = bool(enabled)
if self.is_running():
self.call_soon_threadsafe(
self._set_coroutine_debug, self, self._debug)
def is_running(self):
"""Return whether the event loop is currently running."""
return bool(self._running)
def is_closed(self):
"""Returns True if the event loop was closed."""
return bool(self._closed)
def create_future(self):
"""Create a Future object attached to the loop."""
return self._new_future()
def create_task(self, coro, *, name=None, context=None):
"""Schedule a coroutine object.
Return a task object.
If name is not None, task.set_name(name) will be called if the task
object has the set_name attribute, true for default Task in CPython.
An optional keyword-only context argument allows specifying a custom
contextvars.Context for the coro to run in. The current context copy is
created when no context is provided.
"""
self._check_closed()
if PY311:
if self._task_factory is None:
task = aio_Task(coro, loop=self, context=context)
else:
task = self._task_factory(self, coro, context=context)
else:
if context is None:
if self._task_factory is None:
task = aio_Task(coro, loop=self)
else:
task = self._task_factory(self, coro)
else:
if self._task_factory is None:
task = context.run(aio_Task, coro, self)
else:
task = context.run(self._task_factory, self, coro)
# copied from asyncio.tasks._set_task_name (bpo-34270)
if name is not None:
try:
set_name = task.set_name
except AttributeError:
pass
else:
set_name(name)
return task
def set_task_factory(self, factory):
"""Set a task factory that will be used by loop.create_task().
If factory is None the default task factory will be set.
If factory is a callable, it should have a signature matching
'(loop, coro)', where 'loop' will be a reference to the active
event loop, 'coro' will be a coroutine object. The callable
must return a Future.
"""
if factory is not None and not callable(factory):
raise TypeError('task factory must be a callable or None')
self._task_factory = factory
def get_task_factory(self):
"""Return a task factory, or None if the default one is in use."""
return self._task_factory
def run_until_complete(self, future):
"""Run until the Future is done.
If the argument is a coroutine, it is wrapped in a Task.
WARNING: It would be disastrous to call run_until_complete()
with the same coroutine twice -- it would wrap it in two
different Tasks and that can't be good.
Return the Future's result, or raise its exception.
"""
self._check_closed()
new_task = not isfuture(future)
future = aio_ensure_future(future, loop=self)
if new_task:
# An exception is raised if the future didn't complete, so there
# is no need to log the "destroy pending task" message
future._log_destroy_pending = False
def done_cb(fut):
if not fut.cancelled():
exc = fut.exception()
if isinstance(exc, (SystemExit, KeyboardInterrupt)):
# Issue #336: run_forever() already finished,
# no need to stop it.
return
self.stop()
future.add_done_callback(done_cb)
try:
self.run_forever()
except BaseException:
if new_task and future.done() and not future.cancelled():
# The coroutine raised a BaseException. Consume the exception
# to not log a warning, the caller doesn't have access to the
# local task.
future.exception()
raise
finally:
future.remove_done_callback(done_cb)
if not future.done():
raise RuntimeError('Event loop stopped before Future completed.')
return future.result()
@cython.iterable_coroutine
async def getaddrinfo(self, object host, object port, *,
int family=0, int type=0, int proto=0, int flags=0):
addr = __static_getaddrinfo_pyaddr(host, port, family,
type, proto, flags)
if addr is not None:
return [addr]
return await self._getaddrinfo(
host, port, family, type, proto, flags, 1)
@cython.iterable_coroutine
async def getnameinfo(self, sockaddr, int flags=0):
cdef:
AddrInfo ai_cnt
system.addrinfo *ai
system.sockaddr_in6 *sin6
if not isinstance(sockaddr, tuple):
raise TypeError('getnameinfo() argument 1 must be a tuple')
sl = len(sockaddr)
if sl < 2 or sl > 4:
raise ValueError('sockaddr must be a tuple of 2, 3 or 4 values')
if sl > 2:
flowinfo = sockaddr[2]
if flowinfo < 0 or flowinfo > 0xfffff:
raise OverflowError(
'getnameinfo(): flowinfo must be 0-1048575.')
else:
flowinfo = 0
if sl > 3:
scope_id = sockaddr[3]
if scope_id < 0 or scope_id > 2 ** 32:
raise OverflowError(
'getsockaddrarg: scope_id must be unsigned 32 bit integer')
else:
scope_id = 0
ai_cnt = await self._getaddrinfo(
sockaddr[0], sockaddr[1],
uv.AF_UNSPEC, # family
uv.SOCK_DGRAM, # type
0, # proto
uv.AI_NUMERICHOST, # flags
0) # unpack
ai = ai_cnt.data
if ai.ai_next:
raise OSError("sockaddr resolved to multiple addresses")
if ai.ai_family == uv.AF_INET:
if sl > 2:
raise OSError("IPv4 sockaddr must be 2 tuple")
elif ai.ai_family == uv.AF_INET6:
# Modify some fields in `ai`
sin6 = <system.sockaddr_in6*> ai.ai_addr
sin6.sin6_flowinfo = system.htonl(flowinfo)
sin6.sin6_scope_id = scope_id
return await self._getnameinfo(ai.ai_addr, flags)
@cython.iterable_coroutine
async def start_tls(self, transport, protocol, sslcontext, *,
server_side=False,
server_hostname=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
"""Upgrade transport to TLS.
Return a new transport that *protocol* should start using
immediately.
"""
if not isinstance(sslcontext, ssl_SSLContext):
raise TypeError(
f'sslcontext is expected to be an instance of ssl.SSLContext, '
f'got {sslcontext!r}')
if isinstance(transport, (TCPTransport, UnixTransport)):
context = (<UVStream>transport).context
elif isinstance(transport, _SSLProtocolTransport):
context = (<_SSLProtocolTransport>transport).context
else:
raise TypeError(
f'transport {transport!r} is not supported by start_tls()')
waiter = self._new_future()
ssl_protocol = SSLProtocol(
self, protocol, sslcontext, waiter,
server_side, server_hostname,
ssl_handshake_timeout=ssl_handshake_timeout,
ssl_shutdown_timeout=ssl_shutdown_timeout,
call_connection_made=False)
# Pause early so that "ssl_protocol.data_received()" doesn't
# have a chance to get called before "ssl_protocol.connection_made()".
transport.pause_reading()
transport.set_protocol(ssl_protocol)
conmade_cb = self.call_soon(ssl_protocol.connection_made, transport,
context=context)
# transport.resume_reading() will use the right context
# (transport.context) to call e.g. data_received()
resume_cb = self.call_soon(transport.resume_reading)
app_transport = ssl_protocol._get_app_transport(context)
try:
await waiter
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
app_transport.close()
conmade_cb.cancel()
resume_cb.cancel()
raise
return app_transport
@cython.iterable_coroutine
async def create_server(self, protocol_factory, host=None, port=None,
*,
int family=uv.AF_UNSPEC,
int flags=uv.AI_PASSIVE,
sock=None,
backlog=100,
ssl=None,
reuse_address=None,
reuse_port=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None,
start_serving=True):
"""A coroutine which creates a TCP server bound to host and port.
The return value is a Server object which can be used to stop
the service.
If host is an empty string or None all interfaces are assumed
and a list of multiple sockets will be returned (most likely
one for IPv4 and another one for IPv6). The host parameter can also be
a sequence (e.g. list) of hosts to bind to.
family can be set to either AF_INET or AF_INET6 to force the
socket to use IPv4 or IPv6. If not set it will be determined
from host (defaults to AF_UNSPEC).
flags is a bitmask for getaddrinfo().
sock can optionally be specified in order to use a preexisting
socket object.
backlog is the maximum number of queued connections passed to
listen() (defaults to 100).
ssl can be set to an SSLContext to enable SSL over the
accepted connections.
reuse_address tells the kernel to reuse a local socket in
TIME_WAIT state, without waiting for its natural timeout to
expire. If not specified will automatically be set to True on
UNIX.
reuse_port tells the kernel to allow this endpoint to be bound to
the same port as other existing endpoints are bound to, so long as
they all set this flag when being created. This option is not
supported on Windows.
ssl_handshake_timeout is the time in seconds that an SSL server
will wait for completion of the SSL handshake before aborting the
connection. Default is 60s.
ssl_shutdown_timeout is the time in seconds that an SSL server
will wait for completion of the SSL shutdown before aborting the
connection. Default is 30s.
"""
cdef:
TCPServer tcp
system.addrinfo *addrinfo
Server server
if sock is not None and sock.family == uv.AF_UNIX:
if host is not None or port is not None:
raise ValueError(
'host/port and sock can not be specified at the same time')
return await self.create_unix_server(
protocol_factory, sock=sock, backlog=backlog, ssl=ssl,
start_serving=start_serving)
server = Server(self)
if ssl is not None:
if not isinstance(ssl, ssl_SSLContext):
raise TypeError('ssl argument must be an SSLContext or None')
else:
if ssl_handshake_timeout is not None:
raise ValueError(
'ssl_handshake_timeout is only meaningful with ssl')
if ssl_shutdown_timeout is not None:
raise ValueError(
'ssl_shutdown_timeout is only meaningful with ssl')
if host is not None or port is not None:
if sock is not None:
raise ValueError(
'host/port and sock can not be specified at the same time')
if reuse_address is None:
reuse_address = os_name == 'posix' and sys_platform != 'cygwin'
reuse_port = bool(reuse_port)
if reuse_port and not has_SO_REUSEPORT:
raise ValueError(
'reuse_port not supported by socket module')
if host == '':
hosts = [None]
elif (isinstance(host, str) or not isinstance(host, col_Iterable)):
hosts = [host]
else:
hosts = host
fs = [self._getaddrinfo(host, port, family,
uv.SOCK_STREAM, 0, flags,
0) for host in hosts]
infos = await aio_gather(*fs)
completed = False
sock = None
try:
for info in infos:
addrinfo = (<AddrInfo>info).data
while addrinfo != NULL:
if addrinfo.ai_family == uv.AF_UNSPEC:
raise RuntimeError('AF_UNSPEC in DNS results')
try:
sock = socket_socket(addrinfo.ai_family,
addrinfo.ai_socktype,
addrinfo.ai_protocol)
except socket_error:
# Assume it's a bad family/type/protocol
# combination.
if self._debug:
aio_logger.warning(
'create_server() failed to create '
'socket.socket(%r, %r, %r)',
addrinfo.ai_family,
addrinfo.ai_socktype,
addrinfo.ai_protocol, exc_info=True)
addrinfo = addrinfo.ai_next
continue
if reuse_address:
sock.setsockopt(uv.SOL_SOCKET, uv.SO_REUSEADDR, 1)
if reuse_port:
sock.setsockopt(uv.SOL_SOCKET, uv.SO_REUSEPORT, 1)
# Disable IPv4/IPv6 dual stack support (enabled by
# default on Linux) which makes a single socket
# listen on both address families.
if (addrinfo.ai_family == uv.AF_INET6 and
has_IPV6_V6ONLY):
sock.setsockopt(uv.IPPROTO_IPV6, IPV6_V6ONLY, 1)
pyaddr = __convert_sockaddr_to_pyaddr(addrinfo.ai_addr)
try:
sock.bind(pyaddr)
except OSError as err:
raise OSError(
err.errno, 'error while attempting '
'to bind on address %r: %s'
% (pyaddr, err.strerror.lower())) from None
tcp = TCPServer.new(self, protocol_factory, server,
uv.AF_UNSPEC, backlog,
ssl, ssl_handshake_timeout,
ssl_shutdown_timeout)
try:
tcp._open(sock.fileno())
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
tcp._close()
raise
server._add_server(tcp)
sock.detach()
sock = None
addrinfo = addrinfo.ai_next
completed = True
finally:
if not completed:
if sock is not None:
sock.close()
server.close()
else:
if sock is None:
raise ValueError('Neither host/port nor sock were specified')
if not _is_sock_stream(sock.type):
raise ValueError(
'A Stream Socket was expected, got {!r}'.format(sock))
# libuv will set the socket to non-blocking mode, but
# we want Python socket object to notice that.
sock.setblocking(False)
tcp = TCPServer.new(self, protocol_factory, server,
uv.AF_UNSPEC, backlog,
ssl, ssl_handshake_timeout,
ssl_shutdown_timeout)
try:
tcp._open(sock.fileno())
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
tcp._close()
raise
tcp._attach_fileobj(sock)
server._add_server(tcp)
if start_serving:
server._start_serving()
server._ref()
return server
@cython.iterable_coroutine
async def create_connection(self, protocol_factory, host=None, port=None,
*,
ssl=None,
family=0, proto=0, flags=0, sock=None,
local_addr=None, server_hostname=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
"""Connect to a TCP server.
Create a streaming transport connection to a given Internet host and
port: socket family AF_INET or socket.AF_INET6 depending on host (or
family if specified), socket type SOCK_STREAM. protocol_factory must be
a callable returning a protocol instance.
This method is a coroutine which will try to establish the connection
in the background. When successful, the coroutine returns a
(transport, protocol) pair.
"""
cdef:
AddrInfo ai_local = None
AddrInfo ai_remote
TCPTransport tr
system.addrinfo *rai = NULL
system.addrinfo *lai = NULL
system.addrinfo *rai_iter = NULL
system.addrinfo *lai_iter = NULL
system.addrinfo rai_static
system.sockaddr_storage rai_addr_static
system.addrinfo lai_static
system.sockaddr_storage lai_addr_static
object app_protocol
object app_transport
object protocol
object ssl_waiter
if sock is not None and sock.family == uv.AF_UNIX:
if host is not None or port is not None:
raise ValueError(
'host/port and sock can not be specified at the same time')
return await self.create_unix_connection(
protocol_factory, None,
sock=sock, ssl=ssl, server_hostname=server_hostname)
app_protocol = protocol = protocol_factory()
ssl_waiter = None
context = Context_CopyCurrent()
if ssl:
if server_hostname is None:
if not host:
raise ValueError('You must set server_hostname '
'when using ssl without a host')
server_hostname = host
ssl_waiter = self._new_future()
sslcontext = None if isinstance(ssl, bool) else ssl
protocol = SSLProtocol(
self, app_protocol, sslcontext, ssl_waiter,
False, server_hostname,
ssl_handshake_timeout=ssl_handshake_timeout,
ssl_shutdown_timeout=ssl_shutdown_timeout)
else:
if server_hostname is not None:
raise ValueError('server_hostname is only meaningful with ssl')
if ssl_handshake_timeout is not None:
raise ValueError(
'ssl_handshake_timeout is only meaningful with ssl')
if ssl_shutdown_timeout is not None:
raise ValueError(
'ssl_shutdown_timeout is only meaningful with ssl')
if host is not None or port is not None:
if sock is not None:
raise ValueError(
'host/port and sock can not be specified at the same time')
fs = []
f1 = f2 = None
addr = __static_getaddrinfo(
host, port, family, uv.SOCK_STREAM,
proto, <system.sockaddr*>&rai_addr_static)
if addr is None:
f1 = self._getaddrinfo(
host, port, family,
uv.SOCK_STREAM, proto, flags,
0) # 0 == don't unpack
fs.append(f1)
else:
rai_static.ai_addr = <system.sockaddr*>&rai_addr_static
rai_static.ai_next = NULL
rai = &rai_static
if local_addr is not None:
if not isinstance(local_addr, (tuple, list)) or \
len(local_addr) != 2:
raise ValueError(
'local_addr must be a tuple of host and port')
addr = __static_getaddrinfo(
local_addr[0], local_addr[1],
family, uv.SOCK_STREAM,
proto, <system.sockaddr*>&lai_addr_static)
if addr is None:
f2 = self._getaddrinfo(
local_addr[0], local_addr[1], family,
uv.SOCK_STREAM, proto, flags,
0) # 0 == don't unpack
fs.append(f2)
else:
lai_static.ai_addr = <system.sockaddr*>&lai_addr_static
lai_static.ai_next = NULL
lai = &lai_static
if len(fs):
await aio_wait(fs)
if rai is NULL:
ai_remote = f1.result()
if ai_remote.data is NULL:
raise OSError('getaddrinfo() returned empty list')
rai = ai_remote.data
if lai is NULL and f2 is not None:
ai_local = f2.result()
if ai_local.data is NULL:
raise OSError(
'getaddrinfo() returned empty list for local_addr')
lai = ai_local.data
exceptions = []
rai_iter = rai
while rai_iter is not NULL:
tr = None
try:
waiter = self._new_future()
tr = TCPTransport.new(self, protocol, None, waiter,
context)
if lai is not NULL:
lai_iter = lai
while lai_iter is not NULL:
try:
tr.bind(lai_iter.ai_addr)
break
except OSError as exc:
exceptions.append(exc)
lai_iter = lai_iter.ai_next
else:
tr._close()
tr = None
rai_iter = rai_iter.ai_next
continue
tr.connect(rai_iter.ai_addr)
await waiter
except OSError as exc:
if tr is not None:
tr._close()
tr = None
exceptions.append(exc)
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
if tr is not None:
tr._close()
tr = None
raise
else:
break
rai_iter = rai_iter.ai_next
else:
# If they all have the same str(), raise one.
model = str(exceptions[0])
if all(str(exc) == model for exc in exceptions):
raise exceptions[0]
# Raise a combined exception so the user can see all
# the various error messages.
raise OSError('Multiple exceptions: {}'.format(
', '.join(str(exc) for exc in exceptions)))
else:
if sock is None:
raise ValueError(
'host and port was not specified and no sock specified')
if not _is_sock_stream(sock.type):
raise ValueError(
'A Stream Socket was expected, got {!r}'.format(sock))
# libuv will set the socket to non-blocking mode, but
# we want Python socket object to notice that.
sock.setblocking(False)
waiter = self._new_future()
tr = TCPTransport.new(self, protocol, None, waiter, context)
try:
# libuv will make socket non-blocking
tr._open(sock.fileno())
tr._init_protocol()
await waiter
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
# It's OK to call `_close()` here, as opposed to
# `_force_close()` or `close()` as we want to terminate the
# transport immediately. The `waiter` can only be waken
# up in `Transport._call_connection_made()`, and calling
# `_close()` before it is fine.
tr._close()
raise
tr._attach_fileobj(sock)
if ssl:
app_transport = protocol._get_app_transport(context)
try:
await ssl_waiter
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
app_transport.close()
raise
return app_transport, app_protocol
else:
return tr, protocol
@cython.iterable_coroutine
async def create_unix_server(self, protocol_factory, path=None,
*, backlog=100, sock=None, ssl=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None,
start_serving=True):
"""A coroutine which creates a UNIX Domain Socket server.
The return value is a Server object, which can be used to stop
the service.
path is a str, representing a file systsem path to bind the
server socket to.
sock can optionally be specified in order to use a preexisting
socket object.
backlog is the maximum number of queued connections passed to
listen() (defaults to 100).
ssl can be set to an SSLContext to enable SSL over the
accepted connections.
ssl_handshake_timeout is the time in seconds that an SSL server
will wait for completion of the SSL handshake before aborting the
connection. Default is 60s.
ssl_shutdown_timeout is the time in seconds that an SSL server
will wait for completion of the SSL shutdown before aborting the
connection. Default is 30s.
"""
cdef:
UnixServer pipe
Server server = Server(self)
if ssl is not None:
if not isinstance(ssl, ssl_SSLContext):
raise TypeError('ssl argument must be an SSLContext or None')
else:
if ssl_handshake_timeout is not None:
raise ValueError(
'ssl_handshake_timeout is only meaningful with ssl')
if ssl_shutdown_timeout is not None:
raise ValueError(
'ssl_shutdown_timeout is only meaningful with ssl')
if path is not None:
if sock is not None:
raise ValueError(
'path and sock can not be specified at the same time')
orig_path = path
path = os_fspath(path)
if isinstance(path, str):
path = PyUnicode_EncodeFSDefault(path)
# Check for abstract socket.
if path[0] != 0:
try:
if stat_S_ISSOCK(os_stat(path).st_mode):
os_remove(path)
except FileNotFoundError:
pass
except OSError as err:
# Directory may have permissions only to create socket.
aio_logger.error(
'Unable to check or remove stale UNIX socket %r: %r',
orig_path, err)
# We use Python sockets to create a UNIX server socket because
# when UNIX sockets are created by libuv, libuv removes the path
# they were bound to. This is different from asyncio, which
# doesn't cleanup the socket path.
sock = socket_socket(uv.AF_UNIX)
try:
sock.bind(path)
except OSError as exc:
sock.close()
if exc.errno == errno.EADDRINUSE:
# Let's improve the error message by adding
# with what exact address it occurs.
msg = 'Address {!r} is already in use'.format(orig_path)
raise OSError(errno.EADDRINUSE, msg) from None
else:
raise
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
sock.close()
raise
else:
if sock is None:
raise ValueError(
'path was not specified, and no sock specified')
if sock.family != uv.AF_UNIX or not _is_sock_stream(sock.type):
raise ValueError(
'A UNIX Domain Stream Socket was expected, got {!r}'
.format(sock))
# libuv will set the socket to non-blocking mode, but
# we want Python socket object to notice that.
sock.setblocking(False)
pipe = UnixServer.new(
self, protocol_factory, server, backlog,
ssl, ssl_handshake_timeout, ssl_shutdown_timeout)
try:
pipe._open(sock.fileno())
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
pipe._close()
sock.close()
raise
pipe._attach_fileobj(sock)
server._add_server(pipe)
if start_serving:
server._start_serving()
return server
@cython.iterable_coroutine
async def create_unix_connection(self, protocol_factory, path=None, *,
ssl=None, sock=None,
server_hostname=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
cdef:
UnixTransport tr
object app_protocol
object app_transport
object protocol
object ssl_waiter
app_protocol = protocol = protocol_factory()
ssl_waiter = None
context = Context_CopyCurrent()
if ssl:
if server_hostname is None:
raise ValueError('You must set server_hostname '
'when using ssl without a host')
ssl_waiter = self._new_future()
sslcontext = None if isinstance(ssl, bool) else ssl
protocol = SSLProtocol(
self, app_protocol, sslcontext, ssl_waiter,
False, server_hostname,
ssl_handshake_timeout=ssl_handshake_timeout,
ssl_shutdown_timeout=ssl_shutdown_timeout)
else:
if server_hostname is not None:
raise ValueError('server_hostname is only meaningful with ssl')
if ssl_handshake_timeout is not None:
raise ValueError(
'ssl_handshake_timeout is only meaningful with ssl')
if ssl_shutdown_timeout is not None:
raise ValueError(
'ssl_shutdown_timeout is only meaningful with ssl')
if path is not None:
if sock is not None:
raise ValueError(
'path and sock can not be specified at the same time')
path = os_fspath(path)
if isinstance(path, str):
path = PyUnicode_EncodeFSDefault(path)
waiter = self._new_future()
tr = UnixTransport.new(self, protocol, None, waiter, context)
tr.connect(path)
try:
await waiter
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
tr._close()
raise
else:
if sock is None:
raise ValueError('no path and sock were specified')
if sock.family != uv.AF_UNIX or not _is_sock_stream(sock.type):
raise ValueError(
'A UNIX Domain Stream Socket was expected, got {!r}'
.format(sock))
# libuv will set the socket to non-blocking mode, but
# we want Python socket object to notice that.
sock.setblocking(False)
waiter = self._new_future()
tr = UnixTransport.new(self, protocol, None, waiter, context)
try:
tr._open(sock.fileno())
tr._init_protocol()
await waiter
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
tr._close()
raise
tr._attach_fileobj(sock)
if ssl:
app_transport = protocol._get_app_transport(Context_CopyCurrent())
try:
await ssl_waiter
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
app_transport.close()
raise
return app_transport, app_protocol
else:
return tr, protocol
def default_exception_handler(self, context):
"""Default exception handler.
This is called when an exception occurs and no exception
handler is set, and can be called by a custom exception
handler that wants to defer to the default behavior.
The context parameter has the same meaning as in
`call_exception_handler()`.
"""
message = context.get('message')
if not message:
message = 'Unhandled exception in event loop'
exception = context.get('exception')
if exception is not None:
exc_info = (type(exception), exception, exception.__traceback__)
else:
exc_info = False
log_lines = [message]
for key in sorted(context):
if key in {'message', 'exception'}:
continue
value = context[key]
if key == 'source_traceback':
tb = ''.join(tb_format_list(value))
value = 'Object created at (most recent call last):\n'
value += tb.rstrip()
else:
try:
value = repr(value)
except (KeyboardInterrupt, SystemExit):
raise
except BaseException as ex:
value = ('Exception in __repr__ {!r}; '
'value type: {!r}'.format(ex, type(value)))
log_lines.append('{}: {}'.format(key, value))
aio_logger.error('\n'.join(log_lines), exc_info=exc_info)
def get_exception_handler(self):
"""Return an exception handler, or None if the default one is in use.
"""
return self._exception_handler
def set_exception_handler(self, handler):
"""Set handler as the new event loop exception handler.
If handler is None, the default exception handler will
be set.
If handler is a callable object, it should have a
signature matching '(loop, context)', where 'loop'
will be a reference to the active event loop, 'context'
will be a dict object (see `call_exception_handler()`
documentation for details about context).
"""
if handler is not None and not callable(handler):
raise TypeError('A callable object or None is expected, '
'got {!r}'.format(handler))
self._exception_handler = handler
def call_exception_handler(self, context):
"""Call the current event loop's exception handler.
The context argument is a dict containing the following keys:
- 'message': Error message;
- 'exception' (optional): Exception object;
- 'future' (optional): Future instance;
- 'handle' (optional): Handle instance;
- 'protocol' (optional): Protocol instance;
- 'transport' (optional): Transport instance;
- 'socket' (optional): Socket instance.
New keys maybe introduced in the future.
Note: do not overload this method in an event loop subclass.
For custom exception handling, use the
`set_exception_handler()` method.
"""
if UVLOOP_DEBUG:
self._debug_exception_handler_cnt += 1
if self._exception_handler is None:
try:
self.default_exception_handler(context)
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
# Second protection layer for unexpected errors
# in the default implementation, as well as for subclassed
# event loops with overloaded "default_exception_handler".
aio_logger.error('Exception in default exception handler',
exc_info=True)
else:
try:
self._exception_handler(self, context)
except (KeyboardInterrupt, SystemExit):
raise
except BaseException as exc:
# Exception in the user set custom exception handler.
try:
# Let's try default handler.
self.default_exception_handler({
'message': 'Unhandled error in exception handler',
'exception': exc,
'context': context,
})
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
# Guard 'default_exception_handler' in case it is
# overloaded.
aio_logger.error('Exception in default exception handler '
'while handling an unexpected error '
'in custom exception handler',
exc_info=True)
def add_reader(self, fileobj, callback, *args):
"""Add a reader callback."""
if len(args) == 0:
args = None
self._add_reader(fileobj, new_Handle(self, callback, args, None))
def remove_reader(self, fileobj):
"""Remove a reader callback."""
self._remove_reader(fileobj)
def add_writer(self, fileobj, callback, *args):
"""Add a writer callback.."""
if len(args) == 0:
args = None
self._add_writer(fileobj, new_Handle(self, callback, args, None))
def remove_writer(self, fileobj):
"""Remove a writer callback."""
self._remove_writer(fileobj)
@cython.iterable_coroutine
async def sock_recv(self, sock, n):
"""Receive data from the socket.
The return value is a bytes object representing the data received.
The maximum amount of data to be received at once is specified by
nbytes.
This method is a coroutine.
"""
cdef:
Handle handle
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
fut = _SyncSocketReaderFuture(sock, self)
handle = new_MethodHandle3(
self,
"Loop._sock_recv",
<method3_t>self._sock_recv,
None,
self,
fut, sock, n)
self._add_reader(sock, handle)
return await fut
@cython.iterable_coroutine
async def sock_recv_into(self, sock, buf):
"""Receive data from the socket.
The received data is written into *buf* (a writable buffer).
The return value is the number of bytes written.
This method is a coroutine.
"""
cdef:
Handle handle
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
fut = _SyncSocketReaderFuture(sock, self)
handle = new_MethodHandle3(
self,
"Loop._sock_recv_into",
<method3_t>self._sock_recv_into,
None,
self,
fut, sock, buf)
self._add_reader(sock, handle)
return await fut
@cython.iterable_coroutine
async def sock_sendall(self, sock, data):
"""Send data to the socket.
The socket must be connected to a remote socket. This method continues
to send data from data until either all data has been sent or an
error occurs. None is returned on success. On error, an exception is
raised, and there is no way to determine how much data, if any, was
successfully processed by the receiving end of the connection.
This method is a coroutine.
"""
cdef:
Handle handle
ssize_t n
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
if not data:
return
socket_inc_io_ref(sock)
try:
try:
n = sock.send(data)
except (BlockingIOError, InterruptedError):
pass
else:
if UVLOOP_DEBUG:
# This can be a partial success, i.e. only part
# of the data was sent
self._sock_try_write_total += 1
if n == len(data):
return
if not isinstance(data, memoryview):
data = memoryview(data)
data = data[n:]
fut = _SyncSocketWriterFuture(sock, self)
handle = new_MethodHandle3(
self,
"Loop._sock_sendall",
<method3_t>self._sock_sendall,
None,
self,
fut, sock, data)
self._add_writer(sock, handle)
return await fut
finally:
socket_dec_io_ref(sock)
@cython.iterable_coroutine
async def sock_accept(self, sock):
"""Accept a connection.
The socket must be bound to an address and listening for connections.
The return value is a pair (conn, address) where conn is a new socket
object usable to send and receive data on the connection, and address
is the address bound to the socket on the other end of the connection.
This method is a coroutine.
"""
cdef:
Handle handle
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
fut = _SyncSocketReaderFuture(sock, self)
handle = new_MethodHandle2(
self,
"Loop._sock_accept",
<method2_t>self._sock_accept,
None,
self,
fut, sock)
self._add_reader(sock, handle)
return await fut
@cython.iterable_coroutine
async def sock_connect(self, sock, address):
"""Connect to a remote socket at address.
This method is a coroutine.
"""
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
socket_inc_io_ref(sock)
try:
if sock.family == uv.AF_UNIX:
fut = self._sock_connect(sock, address)
else:
addrs = await self.getaddrinfo(
*address[:2], family=sock.family)
_, _, _, _, address = addrs[0]
fut = self._sock_connect(sock, address)
if fut is not None:
await fut
finally:
socket_dec_io_ref(sock)
@cython.iterable_coroutine
async def sock_recvfrom(self, sock, bufsize):
raise NotImplementedError
@cython.iterable_coroutine
async def sock_recvfrom_into(self, sock, buf, nbytes=0):
raise NotImplementedError
@cython.iterable_coroutine
async def sock_sendto(self, sock, data, address):
raise NotImplementedError
@cython.iterable_coroutine
async def connect_accepted_socket(self, protocol_factory, sock, *,
ssl=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
"""Handle an accepted connection.
This is used by servers that accept connections outside of
asyncio but that use asyncio to handle connections.
This method is a coroutine. When completed, the coroutine
returns a (transport, protocol) pair.
"""
cdef:
UVStream transport = None
if ssl is not None:
if not isinstance(ssl, ssl_SSLContext):
raise TypeError('ssl argument must be an SSLContext or None')
else:
if ssl_handshake_timeout is not None:
raise ValueError(
'ssl_handshake_timeout is only meaningful with ssl')
if ssl_shutdown_timeout is not None:
raise ValueError(
'ssl_shutdown_timeout is only meaningful with ssl')
if not _is_sock_stream(sock.type):
raise ValueError(
'A Stream Socket was expected, got {!r}'.format(sock))
app_protocol = protocol_factory()
waiter = self._new_future()
transport_waiter = None
context = Context_CopyCurrent()
if ssl is None:
protocol = app_protocol
transport_waiter = waiter
else:
protocol = SSLProtocol(
self, app_protocol, ssl, waiter,
server_side=True,
server_hostname=None,
ssl_handshake_timeout=ssl_handshake_timeout,
ssl_shutdown_timeout=ssl_shutdown_timeout)
transport_waiter = None
if sock.family == uv.AF_UNIX:
transport = <UVStream>UnixTransport.new(
self, protocol, None, transport_waiter, context)
elif sock.family in (uv.AF_INET, uv.AF_INET6):
transport = <UVStream>TCPTransport.new(
self, protocol, None, transport_waiter, context)
if transport is None:
raise ValueError(
'invalid socket family, expected AF_UNIX, AF_INET or AF_INET6')
transport._open(sock.fileno())
transport._init_protocol()
transport._attach_fileobj(sock)
if ssl:
app_transport = protocol._get_app_transport(context)
try:
await waiter
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
app_transport.close()
raise
return app_transport, protocol
else:
try:
await waiter
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
transport._close()
raise
return transport, protocol
def run_in_executor(self, executor, func, *args):
if aio_iscoroutine(func) or aio_iscoroutinefunction(func):
raise TypeError("coroutines cannot be used with run_in_executor()")
self._check_closed()
if executor is None:
executor = self._default_executor
# Only check when the default executor is being used
self._check_default_executor()
if executor is None:
executor = cc_ThreadPoolExecutor()
self._default_executor = executor
return aio_wrap_future(executor.submit(func, *args), loop=self)
def set_default_executor(self, executor):
self._default_executor = executor
@cython.iterable_coroutine
async def __subprocess_run(self, protocol_factory, args,
stdin=subprocess_PIPE,
stdout=subprocess_PIPE,
stderr=subprocess_PIPE,
universal_newlines=False,
shell=True,
bufsize=0,
preexec_fn=None,
close_fds=None,
cwd=None,
env=None,
startupinfo=None,
creationflags=0,
restore_signals=True,
start_new_session=False,
executable=None,
pass_fds=(),
# For tests only! Do not use in your code. Ever.
__uvloop_sleep_after_fork=False):
# TODO: Implement close_fds (might not be very important in
# Python 3.5, since all FDs aren't inheritable by default.)
cdef:
int debug_flags = 0
if universal_newlines:
raise ValueError("universal_newlines must be False")
if bufsize != 0:
raise ValueError("bufsize must be 0")
if startupinfo is not None:
raise ValueError('startupinfo is not supported')
if creationflags != 0:
raise ValueError('creationflags is not supported')
if executable is not None:
args[0] = executable
if __uvloop_sleep_after_fork:
debug_flags |= __PROCESS_DEBUG_SLEEP_AFTER_FORK
waiter = self._new_future()
protocol = protocol_factory()
proc = UVProcessTransport.new(self, protocol,
args, env, cwd, start_new_session,
stdin, stdout, stderr, pass_fds,
waiter,
debug_flags,
preexec_fn,
restore_signals)
try:
await waiter
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
proc.close()
raise
return proc, protocol
@cython.iterable_coroutine
async def subprocess_shell(self, protocol_factory, cmd, *,
shell=True,
**kwargs):
if not shell:
raise ValueError("shell must be True")
args = [cmd]
if shell:
args = [b'/bin/sh', b'-c'] + args
return await self.__subprocess_run(protocol_factory, args, shell=True,
**kwargs)
@cython.iterable_coroutine
async def subprocess_exec(self, protocol_factory, program, *args,
shell=False, **kwargs):
if shell:
raise ValueError("shell must be False")
args = list((program,) + args)
return await self.__subprocess_run(protocol_factory, args, shell=False,
**kwargs)
@cython.iterable_coroutine
async def connect_read_pipe(self, proto_factory, pipe):
"""Register read pipe in event loop. Set the pipe to non-blocking mode.
protocol_factory should instantiate object with Protocol interface.
pipe is a file-like object.
Return pair (transport, protocol), where transport supports the
ReadTransport interface."""
cdef:
ReadUnixTransport transp
waiter = self._new_future()
proto = proto_factory()
transp = ReadUnixTransport.new(self, proto, None, waiter)
transp._add_extra_info('pipe', pipe)
try:
transp._open(pipe.fileno())
transp._init_protocol()
await waiter
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
transp._close()
raise
transp._attach_fileobj(pipe)
return transp, proto
@cython.iterable_coroutine
async def connect_write_pipe(self, proto_factory, pipe):
"""Register write pipe in event loop.
protocol_factory should instantiate object with BaseProtocol interface.
Pipe is file-like object already switched to nonblocking.
Return pair (transport, protocol), where transport support
WriteTransport interface."""
cdef:
WriteUnixTransport transp
waiter = self._new_future()
proto = proto_factory()
transp = WriteUnixTransport.new(self, proto, None, waiter)
transp._add_extra_info('pipe', pipe)
try:
transp._open(pipe.fileno())
transp._init_protocol()
await waiter
except (KeyboardInterrupt, SystemExit):
raise
except BaseException:
transp._close()
raise
transp._attach_fileobj(pipe)
return transp, proto
def add_signal_handler(self, sig, callback, *args):
"""Add a handler for a signal. UNIX only.
Raise ValueError if the signal number is invalid or uncatchable.
Raise RuntimeError if there is a problem setting up the handler.
"""
cdef:
Handle h
if not self._is_main_thread():
raise ValueError(
'add_signal_handler() can only be called from '
'the main thread')
if (aio_iscoroutine(callback)
or aio_iscoroutinefunction(callback)):
raise TypeError(
"coroutines cannot be used with add_signal_handler()")
if sig == uv.SIGCHLD:
if (hasattr(callback, '__self__') and
isinstance(callback.__self__, aio_AbstractChildWatcher)):
warnings_warn(
"!!! asyncio is trying to install its ChildWatcher for "
"SIGCHLD signal !!!\n\nThis is probably because a uvloop "
"instance is used with asyncio.set_event_loop(). "
"The correct way to use uvloop is to install its policy: "
"`asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())`"
"\n\n", RuntimeWarning, source=self)
# TODO: ideally we should always raise an error here,
# but that would be a backwards incompatible change,
# because we recommended using "asyncio.set_event_loop()"
# in our README. Need to start a deprecation period
# at some point to turn this warning into an error.
return
raise RuntimeError(
'cannot add a signal handler for SIGCHLD: it is used '
'by the event loop to track subprocesses')
self._check_signal(sig)
self._check_closed()
h = new_Handle(self, callback, args or None, None)
self._signal_handlers[sig] = h
try:
# Register a dummy signal handler to ask Python to write the signal
# number in the wakeup file descriptor.
signal_signal(sig, self.__sighandler)
# Set SA_RESTART to limit EINTR occurrences.
signal_siginterrupt(sig, False)
except OSError as exc:
del self._signal_handlers[sig]
if not self._signal_handlers:
try:
signal_set_wakeup_fd(-1)
except (ValueError, OSError) as nexc:
aio_logger.info('set_wakeup_fd(-1) failed: %s', nexc)
if exc.errno == errno_EINVAL:
raise RuntimeError('sig {} cannot be caught'.format(sig))
else:
raise
def remove_signal_handler(self, sig):
"""Remove a handler for a signal. UNIX only.
Return True if a signal handler was removed, False if not.
"""
if not self._is_main_thread():
raise ValueError(
'remove_signal_handler() can only be called from '
'the main thread')
self._check_signal(sig)
if not self._listening_signals:
return False
try:
del self._signal_handlers[sig]
except KeyError:
return False
if sig == uv.SIGINT:
handler = signal_default_int_handler
else:
handler = signal_SIG_DFL
try:
signal_signal(sig, handler)
except OSError as exc:
if exc.errno == errno_EINVAL:
raise RuntimeError('sig {} cannot be caught'.format(sig))
else:
raise
return True
@cython.iterable_coroutine
async def create_datagram_endpoint(self, protocol_factory,
local_addr=None, remote_addr=None, *,
family=0, proto=0, flags=0,
reuse_address=_unset, reuse_port=None,
allow_broadcast=None, sock=None):
"""A coroutine which creates a datagram endpoint.
This method will try to establish the endpoint in the background.
When successful, the coroutine returns a (transport, protocol) pair.
protocol_factory must be a callable returning a protocol instance.
socket family AF_INET or socket.AF_INET6 depending on host (or
family if specified), socket type SOCK_DGRAM.
reuse_port tells the kernel to allow this endpoint to be bound to
the same port as other existing endpoints are bound to, so long as
they all set this flag when being created. This option is not
supported on Windows and some UNIX's. If the
:py:data:`~socket.SO_REUSEPORT` constant is not defined then this
capability is unsupported.
allow_broadcast tells the kernel to allow this endpoint to send
messages to the broadcast address.
sock can optionally be specified in order to use a preexisting
socket object.
"""
cdef:
UDPTransport udp = None
system.addrinfo * lai
system.addrinfo * rai
if sock is not None:
if not _is_sock_dgram(sock.type):
raise ValueError(
'A UDP Socket was expected, got {!r}'.format(sock))
if (local_addr or remote_addr or
family or proto or flags or
reuse_port or allow_broadcast):
# show the problematic kwargs in exception msg
opts = dict(local_addr=local_addr, remote_addr=remote_addr,
family=family, proto=proto, flags=flags,
reuse_address=reuse_address, reuse_port=reuse_port,
allow_broadcast=allow_broadcast)
problems = ', '.join(
'{}={}'.format(k, v) for k, v in opts.items() if v)
raise ValueError(
'socket modifier keyword arguments can not be used '
'when sock is specified. ({})'.format(problems))
sock.setblocking(False)
udp = UDPTransport.__new__(UDPTransport)
udp._init(self, uv.AF_UNSPEC)
udp.open(sock.family, sock.fileno())
udp._attach_fileobj(sock)
else:
if reuse_address is not _unset:
if reuse_address:
raise ValueError("Passing `reuse_address=True` is no "
"longer supported, as the usage of "
"SO_REUSEPORT in UDP poses a significant "
"security concern.")
else:
warnings_warn("The *reuse_address* parameter has been "
"deprecated as of 0.15.", DeprecationWarning,
stacklevel=2)
reuse_port = bool(reuse_port)
if reuse_port and not has_SO_REUSEPORT:
raise ValueError(
'reuse_port not supported by socket module')
lads = None
if local_addr is not None:
if (not isinstance(local_addr, (tuple, list)) or
len(local_addr) != 2):
raise TypeError(
'local_addr must be a tuple of (host, port)')
lads = await self._getaddrinfo(
local_addr[0], local_addr[1],
family, uv.SOCK_DGRAM, proto, flags,
0)
rads = None
if remote_addr is not None:
if (not isinstance(remote_addr, (tuple, list)) or
len(remote_addr) != 2):
raise TypeError(
'remote_addr must be a tuple of (host, port)')
rads = await self._getaddrinfo(
remote_addr[0], remote_addr[1],
family, uv.SOCK_DGRAM, proto, flags,
0)
excs = []
if lads is None:
if rads is not None:
udp = UDPTransport.__new__(UDPTransport)
rai = (<AddrInfo>rads).data
udp._init(self, rai.ai_family)
udp._connect(rai.ai_addr, rai.ai_addrlen)
udp._set_address(rai)
else:
if family not in (uv.AF_INET, uv.AF_INET6):
raise ValueError('unexpected address family')
udp = UDPTransport.__new__(UDPTransport)
udp._init(self, family)
if reuse_port:
self._sock_set_reuseport(udp._fileno())
else:
lai = (<AddrInfo>lads).data
while lai is not NULL:
try:
udp = UDPTransport.__new__(UDPTransport)
udp._init(self, lai.ai_family)
if reuse_port:
self._sock_set_reuseport(udp._fileno())
udp._bind(lai.ai_addr)
except (KeyboardInterrupt, SystemExit):
raise
except BaseException as ex:
lai = lai.ai_next
excs.append(ex)
continue
else:
break
else:
ctx = None
if len(excs):
ctx = excs[0]
raise OSError('could not bind to local_addr {}'.format(
local_addr)) from ctx
if rads is not None:
rai = (<AddrInfo>rads).data
while rai is not NULL:
if rai.ai_family != lai.ai_family:
rai = rai.ai_next
continue
if rai.ai_protocol != lai.ai_protocol:
rai = rai.ai_next
continue
udp._connect(rai.ai_addr, rai.ai_addrlen)
udp._set_address(rai)
break
else:
raise OSError(
'could not bind to remote_addr {}'.format(
remote_addr))
if allow_broadcast:
udp._set_broadcast(1)
protocol = protocol_factory()
waiter = self._new_future()
assert udp is not None
udp._set_protocol(protocol)
udp._set_waiter(waiter)
udp._init_protocol()
await waiter
return udp, protocol
def _monitor_fs(self, path: str, callback) -> asyncio.Handle:
cdef:
UVFSEvent fs_handle
char* c_str_path
self._check_closed()
fs_handle = UVFSEvent.new(self, callback, None)
p_bytes = path.encode('UTF-8')
c_str_path = p_bytes
flags = 0
fs_handle.start(c_str_path, flags)
return fs_handle
def _check_default_executor(self):
if self._executor_shutdown_called:
raise RuntimeError('Executor shutdown has been called')
def _asyncgen_finalizer_hook(self, agen):
self._asyncgens.discard(agen)
if not self.is_closed():
self.call_soon_threadsafe(self.create_task, agen.aclose())
def _asyncgen_firstiter_hook(self, agen):
if self._asyncgens_shutdown_called:
warnings_warn(
"asynchronous generator {!r} was scheduled after "
"loop.shutdown_asyncgens() call".format(agen),
ResourceWarning, source=self)
self._asyncgens.add(agen)
@cython.iterable_coroutine
async def shutdown_asyncgens(self):
"""Shutdown all active asynchronous generators."""
self._asyncgens_shutdown_called = True
if not len(self._asyncgens):
return
closing_agens = list(self._asyncgens)
self._asyncgens.clear()
shutdown_coro = aio_gather(
*[ag.aclose() for ag in closing_agens],
return_exceptions=True)
results = await shutdown_coro
for result, agen in zip(results, closing_agens):
if isinstance(result, Exception):
self.call_exception_handler({
'message': 'an error occurred during closing of '
'asynchronous generator {!r}'.format(agen),
'exception': result,
'asyncgen': agen
})
@cython.iterable_coroutine
async def shutdown_default_executor(self, timeout=None):
"""Schedule the shutdown of the default executor.
The timeout parameter specifies the amount of time the executor will
be given to finish joining. The default value is None, which means
that the executor will be given an unlimited amount of time.
"""
self._executor_shutdown_called = True
if self._default_executor is None:
return
future = self.create_future()
thread = threading_Thread(target=self._do_shutdown, args=(future,))
thread.start()
try:
await future
finally:
thread.join(timeout)
if thread.is_alive():
warnings_warn(
"The executor did not finishing joining "
f"its threads within {timeout} seconds.",
RuntimeWarning,
stacklevel=2
)
self._default_executor.shutdown(wait=False)
def _do_shutdown(self, future):
try:
self._default_executor.shutdown(wait=True)
self.call_soon_threadsafe(future.set_result, None)
except Exception as ex:
self.call_soon_threadsafe(future.set_exception, ex)
# Expose pointer for integration with other C-extensions
def libuv_get_loop_t_ptr(loop):
return PyCapsule_New(<void *>(<Loop>loop).uvloop, NULL, NULL)
def libuv_get_version():
return uv.uv_version()
def _testhelper_unwrap_capsuled_pointer(obj):
return <uint64_t>PyCapsule_GetPointer(obj, NULL)
cdef void __loop_alloc_buffer(
uv.uv_handle_t* uvhandle,
size_t suggested_size,
uv.uv_buf_t* buf
) noexcept with gil:
cdef:
Loop loop = (<UVHandle>uvhandle.data)._loop
if loop._recv_buffer_in_use == 1:
buf.len = 0
exc = RuntimeError('concurrent allocations')
loop._handle_exception(exc)
return
loop._recv_buffer_in_use = 1
buf.base = loop._recv_buffer
buf.len = sizeof(loop._recv_buffer)
cdef inline void __loop_free_buffer(Loop loop):
loop._recv_buffer_in_use = 0
class _SyncSocketReaderFuture(aio_Future):
def __init__(self, sock, loop):
aio_Future.__init__(self, loop=loop)
self.__sock = sock
self.__loop = loop
def __remove_reader(self):
if self.__sock is not None and self.__sock.fileno() != -1:
self.__loop.remove_reader(self.__sock)
self.__sock = None
if PY39:
def cancel(self, msg=None):
self.__remove_reader()
aio_Future.cancel(self, msg=msg)
else:
def cancel(self):
self.__remove_reader()
aio_Future.cancel(self)
class _SyncSocketWriterFuture(aio_Future):
def __init__(self, sock, loop):
aio_Future.__init__(self, loop=loop)
self.__sock = sock
self.__loop = loop
def __remove_writer(self):
if self.__sock is not None and self.__sock.fileno() != -1:
self.__loop.remove_writer(self.__sock)
self.__sock = None
if PY39:
def cancel(self, msg=None):
self.__remove_writer()
aio_Future.cancel(self, msg=msg)
else:
def cancel(self):
self.__remove_writer()
aio_Future.cancel(self)
include "cbhandles.pyx"
include "pseudosock.pyx"
include "lru.pyx"
include "handles/handle.pyx"
include "handles/async_.pyx"
include "handles/idle.pyx"
include "handles/check.pyx"
include "handles/timer.pyx"
include "handles/poll.pyx"
include "handles/basetransport.pyx"
include "handles/stream.pyx"
include "handles/streamserver.pyx"
include "handles/tcp.pyx"
include "handles/pipe.pyx"
include "handles/process.pyx"
include "handles/fsevent.pyx"
include "request.pyx"
include "dns.pyx"
include "sslproto.pyx"
include "handles/udp.pyx"
include "server.pyx"
# Used in UVProcess
cdef vint __atfork_installed = 0
cdef vint __forking = 0
cdef Loop __forking_loop = None
cdef void __get_fork_handler() noexcept nogil:
with gil:
if (__forking and __forking_loop is not None and
__forking_loop.active_process_handler is not None):
__forking_loop.active_process_handler._after_fork()
cdef __install_atfork():
global __atfork_installed
if __atfork_installed:
return
__atfork_installed = 1
cdef int err
err = system.pthread_atfork(NULL, NULL, &system.handleAtFork)
if err:
__atfork_installed = 0
raise convert_error(-err)
# Install PyMem* memory allocators
cdef vint __mem_installed = 0
cdef __install_pymem():
global __mem_installed
if __mem_installed:
return
__mem_installed = 1
cdef int err
err = uv.uv_replace_allocator(<uv.uv_malloc_func>PyMem_RawMalloc,
<uv.uv_realloc_func>PyMem_RawRealloc,
<uv.uv_calloc_func>PyMem_RawCalloc,
<uv.uv_free_func>PyMem_RawFree)
if err < 0:
__mem_installed = 0
raise convert_error(err)
cdef _set_signal_wakeup_fd(fd):
if fd >= 0:
return signal_set_wakeup_fd(fd, warn_on_full_buffer=False)
else:
return signal_set_wakeup_fd(fd)
# Helpers for tests
@cython.iterable_coroutine
async def _test_coroutine_1():
return 42