# 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 = 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 = 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, self._on_wake, self) self.handler_idle = UVIdle.new( self, new_MethodHandle( self, "loop._on_idle", 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", 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 = 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", 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 = (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 = 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 = 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(): (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 = 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 = 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 = (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 = (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 = (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 = (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 = (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 = (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 = (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 = (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", 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", 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, &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 = 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", 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 = 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 = (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 = (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, &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 = &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, &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 = &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", 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", 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", 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", 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 = UnixTransport.new( self, protocol, None, transport_waiter, context) elif sock.family in (uv.AF_INET, uv.AF_INET6): transport = 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 = (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 = (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 = (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((loop).uvloop, NULL, NULL) def libuv_get_version(): return uv.uv_version() def _testhelper_unwrap_capsuled_pointer(obj): return 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.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(PyMem_RawMalloc, PyMem_RawRealloc, PyMem_RawCalloc, 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