Optimize `_SyncSocketReaderFuture` and with a Cythonized version of the `Future` Object

Nobody has claimed this yet.

Something that has always annoyed me with both this library and the winloop version is the lack of a C Extension on these classes that need to subclassed to the Future object. I am aware that we could extern it but wrapping that much code from the `_asyncio` C Module would just break ABI Changes and begging Python to wrap `_asynciomodule.c` as a C-API capsule would be a waste of time, so I came up with a second plan that involves writing a new Future Object as a cython cdef extension class to resolve this issue. Thank goodness I was able to successfully get this to work because it took a very long time for me to get it to be successful. My idea involves making another submodule for it and possibly naming it as a `UVFuture` to not be confused with `aio_Future` which will still be used elsewhere and then we can turn the remaining objects that rely on sub classing `asyncio.Future` to be replaced with this new object.

```cython

# Will Hypothetically say the Loop cdef extension is in here.
from cpython.contextvars cimport PyContext_CopyCurrent
from cpython.list cimport PyList_GET_SIZE
import sys
from asyncio import base_futures, format_helpers
from asyncio import (
CancelledError,
InvalidStateError,
get_event_loop
)
from collections.abc import Awaitable
from types import GenericAlias, coroutine

cdef enum fut_state:
_PENDING = 0
_CANCELLED = 1
_FINISHED = 2

# This is mostly being done as an attempt to show that uvloop and winloop are infact
# still optimizable if they had a cdef Future extension instead of a normal class to
# go off of...

cdef class UVFuture:
cdef:
fut_state state
object _result
object _exception
Loop _loop
object _source_traceback
object _cancel_message
object _cancelled_exc
list _callbacks
public bint _asyncio_future_blocking
bint __log_traceback

def __init__(self, *, loop=None):
self.state = _PENDING
self._result = None
self._exception = None
self._source_traceback = None
self._cancel_message = None
self._cancelled_exc = None
self.__log_traceback = False

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

self._callbacks = []

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

@property
def _state(self):
# compatability with python futures
if self.state == _PENDING:
return "PENDING"
elif self.state == _CANCELLED:
return "CANCELLED"
return "FINISHED"

def __repr__(self):
return base_futures._future_repr(self)

def __del__(self):
cdef dict context
cdef object exc
if not self.__log_traceback:
# set_exception() was not called, or result() or exception()
# has consumed the exception
return
exc = self._exception
context = {
'message':
f'{self.__class__.__name__} exception was never retrieved',
'exception': exc,
'future': self,
}
if self._source_traceback:
context['source_traceback'] = self._source_traceback
self._loop.call_exception_handler(context)

__class_getitem__ = classmethod(GenericAlias)

@property
def _log_traceback(self):
return self.__log_traceback

@_log_traceback.setter
def _log_traceback(self, val):
if val:
raise ValueError('_log_traceback can only be set to False')
self.__log_traceback = False

def get_loop(self):
"""Return the event loop the Future is bound to."""
loop = self._loop
if loop is None:
raise RuntimeError("Future object is not initialized.")
return loop

def _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._cancel_message is None:
exc = CancelledError()
else:
exc = CancelledError(self._cancel_message)
exc.__context__ = self._cancelled_exc
# Remove the reference since we don't need this anymore.
self._cancelled_exc = None
return exc

cpdef bint 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.__log_traceback = False
if self.state != _PENDING:
return False
self.state = _CANCELLED
self._cancel_message = msg
self.__schedule_callbacks()
return True

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.
"""
callbacks = self._callbacks[:]
if not callbacks:
return

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

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

# Don't implement running(); see http://bugs.python.org/issue18699

cpdef bint done(self):
"""Return True if the future is done.

Done means either that a result / exception are available, or that the
future was cancelled.
"""
return self.state != _PENDING

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.state == _CANCELLED:
exc = self._make_cancelled_error()
raise exc
if self.state != _FINISHED:
raise InvalidStateError('Result is not ready.')
self.__log_traceback = False
if self._exception is not None:
raise self._exception.with_traceback(self._exception_tb)
return self._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.state == _CANCELLED:
exc = self._make_cancelled_error()
raise exc
if self.state != _FINISHED:
raise InvalidStateError('Exception is not set.')
self.__log_traceback = False
return self._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.state != _PENDING:
self._loop.call_soon(fn, self, context=context)
else:
self._callbacks.append((fn, PyContext_CopyCurrent() if context is None else context))

# New method not in PEP 3148.

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.
"""
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

# So-called internal methods (note: no set_running_or_notify_cancel()).

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.
"""
if self.state != _PENDING:
raise InvalidStateError(f'{self._state}: {self!r}')
self._result = result
self.state = _FINISHED
self.__schedule_callbacks()

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

If the future is already done when this method is called, raises
InvalidStateError.
"""
if self.state != _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.state = _FINISHED
self.__schedule_callbacks()
self.__log_traceback = True

@coroutine
def __await__(self):
while not self.done():
self._asyncio_future_blocking = True
yield # This tells Task to wait for completion.
if not self.done():
raise RuntimeError("await wasn't used with future")
return self.result() # May raise too.

__iter__ = __await__ # make compatible with 'yield from'.

# flag as awaitable incase needed...
Awaitable.register(Future)
```

After that the following changes can be made. Same can later be done to `_SyncSocketWriterFuture` No more costly loading functions on `__remove_reader`
```cython
cdef class _SyncSocketReaderFuture(UVFuture):
cdef:
object __sock
Loop __loop
def __init__(self, sock, loop):
UVFuture.__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()
UVFuture.cancel(self, msg=msg)

else:
def cancel(self):
self.__remove_reader()
UVFuture.cancel(self)
```

Start by reviewing the existing _SyncSocketReaderFuture and the related _SyncSocketWriterFuture mentioned in the issue, then compare their asyncio.Future subclassing with the proposed UVFuture Cython extension class. Done would mean determining whether the new Future can replace those implementations while preserving the shown Future behavior and avoiding the costly __remove_reader path.

Written by the indexing model from the issue text.