Copy futures._chain_future implementation (unbreaks Python 3.5.0)

Author: 1st1

uvloop/chain_futs.pyx

@@ -0,0 +1,88 @@
+# Copied from asyncio 3.5.2
+
+cdef _set_concurrent_future_state(concurrent, source):
+    """Copy state from a future to a concurrent.futures.Future."""
+    assert source.done()
+    if source.cancelled():
+        concurrent.cancel()
+    if not concurrent.set_running_or_notify_cancel():
+        return
+    exception = source.exception()
+    if exception is not None:
+        concurrent.set_exception(exception)
+    else:
+        result = source.result()
+        concurrent.set_result(result)
+
+
+cdef _copy_future_state(source, dest):
+    """Internal helper to copy state from another Future.
+
+    The other Future may be a concurrent.futures.Future.
+    """
+    assert source.done()
+    if dest.cancelled():
+        return
+    assert not dest.done()
+    if source.cancelled():
+        dest.cancel()
+    else:
+        exception = source.exception()
+        if exception is not None:
+            dest.set_exception(exception)
+        else:
+            result = source.result()
+            dest.set_result(result)
+
+
+cdef _chain_future(source, destination):
+    """Chain two futures so that when one completes, so does the other.
+
+    The result (or exception) of source will be copied to destination.
+    If destination is cancelled, source gets cancelled too.
+    Compatible with both asyncio.Future and concurrent.futures.Future.
+    """
+    if not isinstance(source, (aio_Future, cc_Future)):
+        raise TypeError('A future is required for source argument')
+    if not isinstance(destination, (aio_Future, cc_Future)):
+        raise TypeError('A future is required for destination argument')
+
+    source_loop = None
+    dest_loop = None
+
+    source_type = type(source)
+    dest_type = type(destination)
+
+    if source_type is uvloop_Future:
+        source_loop = (<BaseFuture>source)._loop
+    elif source_type is not cc_Future and \
+            isinstance(source, aio_Future):
+        source_loop = source._loop
+
+    if dest_type is uvloop_Future:
+        dest_loop = (<BaseFuture>destination)._loop
+    elif dest_type is not cc_Future and \
+            isinstance(destination, aio_Future):
+        dest_loop = destination._loop
+
+    def _set_state(future, other):
+        if isinstance(future, aio_Future):
+            _copy_future_state(other, future)
+        else:
+            _set_concurrent_future_state(future, other)
+
+    def _call_check_cancel(destination):
+        if destination.cancelled():
+            if source_loop is None or source_loop is dest_loop:
+                source.cancel()
+            else:
+                source_loop.call_soon_threadsafe(source.cancel)
+
+    def _call_set_state(source):
+        if dest_loop is None or dest_loop is source_loop:
+            _set_state(destination, source)
+        else:
+            dest_loop.call_soon_threadsafe(_set_state, destination, source)
+
+    destination.add_done_callback(_call_check_cancel)
+    source.add_done_callback(_call_set_state)

uvloop/includes/stdlib.pxi

@@ -18,7 +18,6 @@ import time
 import warnings
 
 
-cdef aio_chain_future = asyncio.futures._chain_future
 cdef aio_get_event_loop = asyncio.get_event_loop
 cdef aio_CancelledError = asyncio.CancelledError
 cdef aio_InvalidStateError = asyncio.InvalidStateError
@@ -32,7 +31,6 @@ cdef aio_logger = asyncio.log.logger
 cdef aio__check_resolved_address = asyncio.base_events._check_resolved_address
 cdef aio_iscoroutine = asyncio.iscoroutine
 cdef aio_iscoroutinefunction = asyncio.iscoroutinefunction
-cdef aio_wrap_future = asyncio.wrap_future
 cdef aio_BaseProtocol = asyncio.BaseProtocol
 cdef aio_Protocol = asyncio.Protocol
 cdef aio_SSLProtocol = asyncio.sslproto.SSLProtocol
@@ -43,6 +41,7 @@ cdef col_Iterable = collections.Iterable
 cdef col_Counter = collections.Counter
 
 cdef cc_ThreadPoolExecutor = concurrent.futures.ThreadPoolExecutor
+cdef cc_Future = concurrent.futures.Future
 
 cdef ft_partial = functools.partial
 

uvloop/loop.pyx

@@ -1837,7 +1837,7 @@ cdef class Loop:
                 self._default_executor = executor
 
         new_fut = self._new_future()
-        aio_chain_future(executor.submit(func, *args), new_fut)
+        _chain_future(executor.submit(func, *args), new_fut)
         return new_fut
 
     def set_default_executor(self, executor):
@@ -2263,6 +2263,7 @@ include "server.pyx"
 include "os_signal.pyx"
 
 include "future.pyx"
+include "chain_futs.pyx"
 include "task.pyx"