UVLoop Performance Enhancement ideas & Notes. This Brings in a custom Coroutine and Awaitable Container object which could vastly eliminate a lot of python heavy rendering with the main task as well as other unique stuff

uvloop_and_winloop_enhancements.pyx

cimport cython
from cpython.exc cimport PyErr_SetObject
from cpython.contextvars cimport PyContext_CopyCurrent
from asyncio import get_event_loop, format_helpers, CancelledError, InvalidStateError
import sys
from cpython.list cimport PyList_GET_SIZE

# Concept work for winloop & uvloop speedups in theory.

cdef extern from "Python.h":
    int PyException_SetTraceback(object exc, object tb) except -1


DEF UV_COROUTINE_FREELIST = 300

cdef enum UVState:
    UV_PENDING = 0
    UV_CANCELLED = 1
    UV_FINISHED = 2

# Basic Object for making awaitable containers with...
# NOTE: uv_* prefixes refer to non-public objects

cdef class UVAwaitable:
    """Primarly used as a baseplate object for making different 
    types of awaitable objects..."""
    cdef:
        public bint _asyncio_future_blocking
        UVState uv_state
        object uv_loop # uvloop.Loop in this case...

    def __cinit__(self, *, object loop = None):
        self.uv_loop = loop or get_event_loop()
        self._asyncio_future_blocking = False
    
    cpdef bint done(self):
        return self.uv_state != UV_PENDING
    
    cpdef object result(self):
        raise NotImplementedError("result not implemented.")

    def __await__(self):
        yield from UVCoroutine(self)
    
    def __iter__(self):
        return UVCoroutine(self).__iter__()
    
@cython.freelist(300)
cdef class UVCoroutine:
    cdef:
        UVAwaitable uv_fut

    def __cinit__(self, UVAwaitable uv_fut):
        self.uv_fut = uv_fut

    def __iter__(self):
        cdef UVAwaitable fut = self.uv_fut
        while True:
            if not fut.uv_state != UV_PENDING:
                if not fut._asyncio_future_blocking:
                    fut._asyncio_future_blocking = True
                    yield fut
                    continue
                raise RuntimeError("await wasn't used with future")
            return fut.result()

    cpdef object send(self, object value):
        if self.uv_fut.uv_state != UV_PENDING:
            raise StopIteration(self.uv_fut.result())
        return next(self)

    cpdef object throw(
        self, 
        object typ, 
        object val=None, 
        object tb=None
    ):
        """Raise an exception in the coroutine.
        Return next yielded value or raise StopIteration.
        """
        if val is None:
            if tb is None:
                raise typ
            val = typ()
        if tb is not None:
            # Same as with_traceback
            PyException_SetTraceback(val, tb)
        raise val

    cpdef object close(self):
        """Raise GeneratorExit inside coroutine."""
        try:
            self.throw(GeneratorExit)
        except (GeneratorExit, StopIteration):
            pass
        else:
            raise RuntimeError("coroutine ignored GeneratorExit")


# From there we could implement a Future Object. 
# NOTE: This may or may not have coverage on everything asyncio.Future does.
# But it does work correctly!

# This UVFuture Object could then be utilized to speedup the classes
# such as 

# - _SyncSocketReaderFuture
# - _SyncSocketWriterFuture 

# Both of which are very costly on performance because they do not use a
# C Class extension.


cdef class UVFuture(UVAwaitable):
    cdef:
        list uv_callbacks
        object uv_result
        object uv_source_traceback
        object uv_exception
        object uv_exception_tb
        object uv_cancelled_exc

        bint uv__log_traceback
    
    
    def __cinit__(self, *, loop=None):
        self.uv_state = UV_PENDING
        self.uv_result = None
        self.uv_exception = None
        self.uv_source_traceback = None
        self.uv_cancel_message = None
        self.uv_cancelled_exc = None
        self.uv__log_traceback = False

        self.uv_callbacks = []

        if loop is None:
            self.uv_loop = get_event_loop()
        else:
            self.uv_loop = loop

        if self.uv_loop.get_debug():
            self._source_traceback = format_helpers.extract_stack(
                sys._getframe(1))

    # it's a little faster to error handle using an integer rather 
    # than an actual python object.
    cdef int ensure_alive(self) except -1:
        if self.uv_loop is None:
            PyErr_SetObject(
                RuntimeError,
                "Future object is not initialized."
            )
            return -1 
        return 0

    @property
    def _state(self):
        # compatability with python futures
        if self.uv_state == UV_PENDING:
            return "PENDING"
        elif self.uv_state == UV_CANCELLED:                
            return "CANCELLED"
        return "FINISHED"

    @property
    def _log_traceback(self):
        self.ensure_alive()
        return self.uv__log_traceback

    @_log_traceback.setter
    def _log_traceback(self, val):
        if val:
            raise ValueError('_log_traceback can only be set to False')
        self.uv__log_traceback = False
    
    cpdef object get_loop(self):
        """Return the event loop the Future is bound to."""
        self.ensure_alive()
        return self._loop

    cpdef object _make_cancelled_error(self):
        """Create the CancelledError to raise if the Future is cancelled.

        This should only be called once when handling a cancellation since
        it erases the saved context exception value.
        """
        if self._cancelled_exc is not None:
            exc = self._cancelled_exc
            self._cancelled_exc = None
            return exc

        if self.uv_cancel_message is None:
            exc = CancelledError()
        else:
            exc = CancelledError(self.uv_cancel_message)
        exc.__context__ = self._cancelled_exc
        # Remove the reference since we don't need this anymore.
        self._cancelled_exc = None
        return exc

    cpdef object cancel(self, object msg=None):
        """Cancel the future and schedule callbacks.

        If the future is already done or cancelled, return False.  Otherwise,
        change the future's state to cancelled, schedule the callbacks and
        return True.
        """
        self.ensure_alive()
        self.uv__log_traceback = False
        if self.uv_state != UV_PENDING:
            return False
        self.uv_state = UV_CANCELLED
        self.uv_cancel_message = msg
        self.__schedule_callbacks()
        return True
    
    @property
    def _callbacks(self):
        self.ensure_alive()
        return self.uv_callbacks
    
    cpdef object __schedule_callbacks(self):
        """Internal: Ask the event loop to call all callbacks.

        The callbacks are scheduled to be called as soon as possible. Also
        clears the callback list.
        """
        
        cdef list callbacks = self.uv_callbacks[:]
  
        if not callbacks:
            return

        self.uv_callbacks[:] = []
        for callback, ctx in callbacks:
            self.uv_loop.call_soon(callback, self, context=ctx)

    cpdef bint cancelled(self):
        """Return True if the future was cancelled."""
        return self.uv_state == UV_CANCELLED

    cpdef object result(self):
        """Return the result this future represents.

        If the future has been cancelled, raises CancelledError.  If the
        future's result isn't yet available, raises InvalidStateError.  If
        the future is done and has an exception set, this exception is raised.
        """
        if self.uv_state == UV_CANCELLED:
            exc = self._make_cancelled_error()
            raise exc
        if self.uv_state != UV_FINISHED:
            raise InvalidStateError('Result is not ready.')
        self.uv__log_traceback = False
        if self.uv_exception is not None:
            raise self.uv_exception.with_traceback(self.uv_exception_tb)
        return self.uv_result

    cpdef object exception(self):
        """Return the exception that was set on this future.

        The exception (or None if no exception was set) is returned only if
        the future is done.  If the future has been cancelled, raises
        CancelledError.  If the future isn't done yet, raises
        InvalidStateError.
        """
        if self.uv_state == UV_CANCELLED:
            exc = self._make_cancelled_error()
            raise exc
        if self.uv_state != UV_FINISHED:
            raise InvalidStateError('Exception is not set.')
        self.uv__log_traceback = False
        return self.uv_exception

    def add_done_callback(self, fn, *, context=None):
        """Add a callback to be run when the future becomes done.

        The callback is called with a single argument - the future object. If
        the future is already done when this is called, the callback is
        scheduled with call_soon.
        """
        if self.uv_state != UV_PENDING:
            self._loop.call_soon(fn, self, context=context)
        else:
            self.__callbacks.append((fn, PyContext_CopyCurrent() if context is None else context))


    cpdef Py_ssize_t remove_done_callback(self, object fn):
        """Remove all instances of a callback from the "call when done" list.

        Returns the number of callbacks removed.
        """
        self.ensure_alive()
        cdef Py_ssize_t removed_count
        cdef list filtered_callbacks = [(f, ctx)
                              for (f, ctx) in self.__callbacks
                              if f != fn]
        removed_count = PyList_GET_SIZE(self.__callbacks) - PyList_GET_SIZE(filtered_callbacks)
        if removed_count:
            self.__callbacks[:] = filtered_callbacks
        return removed_count

 
    cpdef set_result(self, object result):
        """Mark the future done and set its result.

        If the future is already done when this method is called, raises
        InvalidStateError.
        """
        self.ensure_alive()
        if self.uv_state != UV_PENDING:
            raise InvalidStateError(f'{self._state}: {self!r}')
        self._result = result
        self.uv_state = UV_FINISHED
        self.__schedule_callbacks()

    cpdef set_exception(self, object exception):
        """Mark the future done and set an exception.

        If the future is already done when this method is called, raises
        InvalidStateError.
        """
        self.ensure_alive()
        if self.uv_state != UV_PENDING:
            raise InvalidStateError(f'{self._state}: {self!r}')
        if isinstance(exception, type):
            exception = exception()
        if type(exception) is StopIteration:
            raise TypeError("StopIteration interacts badly with generators "
                            "and cannot be raised into a Future")
        self._exception = exception
        self._exception_tb = exception.__traceback__
        self.uv_state = UV_FINISHED
        self.__schedule_callbacks()
        self.uv__log_traceback = True
    
    def __await__(self):
        yield from UVCoroutine(self)

    def __iter__(self):
        return UVCoroutine(self).__iter__()


# From there the things we could implement would be very huge but also
# exteremely rewarding. Examples include a MainTask that doesn't
# need to use the python interpreter on it's own which could be
# A very good performance enhancement...

cdef class UVMainTask(UVFuture):
    ...