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/stdlib.pxi"

include "errors.pyx"

cdef:

int PY39 = PY_VERSION_HEX >= 0x03090000

int PY311 = PY_VERSION_HEX >= 0x030b0000

int PY313 = PY_VERSION_HEX >= 0x030d0000

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):

Context_Enter(context)

try:

return method()

finally:

Context_Exit(context)

cdef inline run_in_context1(context, method, arg):

Context_Enter(context)

try:

return method(arg)

finally:

Context_Exit(context)

cdef inline run_in_context2(context, method, arg1, arg2):

Context_Enter(context)

try:

return method(arg1, arg2)

finally:

Context_Exit(context)

_unset = object()

cdef class Loop:

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):

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 ((len(self._ready) > 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()

if len(self._ready) == 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()

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)

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.

# 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):

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):

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