Issue #27041: asyncio: Add loop.create_future method

13 files changed, 53 insertions, 39 deletions

diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index ada178f..313cc31 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -209,7 +209,7 @@ class Server(events.AbstractServer):
     def wait_closed(self):
         if self.sockets is None or self._waiters is None:
             return
-        waiter = futures.Future(loop=self._loop)
+        waiter = self._loop.create_future()
         self._waiters.append(waiter)
         yield from waiter
 
@@ -243,6 +243,10 @@ class BaseEventLoop(events.AbstractEventLoop):
                 % (self.__class__.__name__, self.is_running(),
                    self.is_closed(), self.get_debug()))
 
+    def create_future(self):
+        """Create a Future object attached to the loop."""
+        return futures.Future(loop=self)
+
     def create_task(self, coro):
         """Schedule a coroutine object.
 
@@ -536,7 +540,7 @@ class BaseEventLoop(events.AbstractEventLoop):
             assert not args
             assert not isinstance(func, events.TimerHandle)
             if func._cancelled:
-                f = futures.Future(loop=self)
+                f = self.create_future()
                 f.set_result(None)
                 return f
             func, args = func._callback, func._args
@@ -579,7 +583,7 @@ class BaseEventLoop(events.AbstractEventLoop):
                     family=0, type=0, proto=0, flags=0):
         info = _ipaddr_info(host, port, family, type, proto)
         if info is not None:
-            fut = futures.Future(loop=self)
+            fut = self.create_future()
             fut.set_result([info])
             return fut
         elif self._debug:
@@ -720,7 +724,7 @@ class BaseEventLoop(events.AbstractEventLoop):
     def _create_connection_transport(self, sock, protocol_factory, ssl,
                                      server_hostname):
         protocol = protocol_factory()
-        waiter = futures.Future(loop=self)
+        waiter = self.create_future()
         if ssl:
             sslcontext = None if isinstance(ssl, bool) else ssl
             transport = self._make_ssl_transport(
@@ -840,7 +844,7 @@ class BaseEventLoop(events.AbstractEventLoop):
                 raise exceptions[0]
 
         protocol = protocol_factory()
-        waiter = futures.Future(loop=self)
+        waiter = self.create_future()
         transport = self._make_datagram_transport(
             sock, protocol, r_addr, waiter)
         if self._debug:
@@ -979,7 +983,7 @@ class BaseEventLoop(events.AbstractEventLoop):
     @coroutine
     def connect_read_pipe(self, protocol_factory, pipe):
         protocol = protocol_factory()
-        waiter = futures.Future(loop=self)
+        waiter = self.create_future()
         transport = self._make_read_pipe_transport(pipe, protocol, waiter)
 
         try:
@@ -996,7 +1000,7 @@ class BaseEventLoop(events.AbstractEventLoop):
     @coroutine
     def connect_write_pipe(self, protocol_factory, pipe):
         protocol = protocol_factory()
-        waiter = futures.Future(loop=self)
+        waiter = self.create_future()
         transport = self._make_write_pipe_transport(pipe, protocol, waiter)
 
         try:
diff --git a/Lib/asyncio/base_subprocess.py b/Lib/asyncio/base_subprocess.py
index 73425d9..08080bd 100644
--- a/Lib/asyncio/base_subprocess.py
+++ b/Lib/asyncio/base_subprocess.py
@@ -227,7 +227,7 @@ class BaseSubprocessTransport(transports.SubprocessTransport):
         if self._returncode is not None:
             return self._returncode
 
-        waiter = futures.Future(loop=self._loop)
+        waiter = self._loop.create_future()
         self._exit_waiters.append(waiter)
         return (yield from waiter)
 
diff --git a/Lib/asyncio/events.py b/Lib/asyncio/events.py
index 8358ebf..c48c5be 100644
--- a/Lib/asyncio/events.py
+++ b/Lib/asyncio/events.py
@@ -266,6 +266,9 @@ class AbstractEventLoop:
     def time(self):
         raise NotImplementedError
 
+    def create_future(self):
+        raise NotImplementedError
+
     # Method scheduling a coroutine object: create a task.
 
     def create_task(self, coro):
diff --git a/Lib/asyncio/futures.py b/Lib/asyncio/futures.py
index ddb9cde..1feba4d 100644
--- a/Lib/asyncio/futures.py
+++ b/Lib/asyncio/futures.py
@@ -451,6 +451,8 @@ def wrap_future(future, *, loop=None):
         return future
     assert isinstance(future, concurrent.futures.Future), \
         'concurrent.futures.Future is expected, got {!r}'.format(future)
-    new_future = Future(loop=loop)
+    if loop is None:
+        loop = events.get_event_loop()
+    new_future = loop.create_future()
     _chain_future(future, new_future)
     return new_future
diff --git a/Lib/asyncio/locks.py b/Lib/asyncio/locks.py
index 34f6bc1..1804d7b 100644
--- a/Lib/asyncio/locks.py
+++ b/Lib/asyncio/locks.py
@@ -170,7 +170,7 @@ class Lock(_ContextManagerMixin):
             self._locked = True
             return True
 
-        fut = futures.Future(loop=self._loop)
+        fut = self._loop.create_future()
         self._waiters.append(fut)
         try:
             yield from fut
@@ -258,7 +258,7 @@ class Event:
         if self._value:
             return True
 
-        fut = futures.Future(loop=self._loop)
+        fut = self._loop.create_future()
         self._waiters.append(fut)
         try:
             yield from fut
@@ -320,7 +320,7 @@ class Condition(_ContextManagerMixin):
 
         self.release()
         try:
-            fut = futures.Future(loop=self._loop)
+            fut = self._loop.create_future()
             self._waiters.append(fut)
             try:
                 yield from fut
@@ -433,7 +433,7 @@ class Semaphore(_ContextManagerMixin):
         True.
         """
         while self._value <= 0:
-            fut = futures.Future(loop=self._loop)
+            fut = self._loop.create_future()
             self._waiters.append(fut)
             try:
                 yield from fut
diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py
index b2ddee4..eb92458 100644
--- a/Lib/asyncio/proactor_events.py
+++ b/Lib/asyncio/proactor_events.py
@@ -443,7 +443,7 @@ class BaseProactorEventLoop(base_events.BaseEventLoop):
         try:
             base_events._check_resolved_address(sock, address)
         except ValueError as err:
-            fut = futures.Future(loop=self)
+            fut = self.create_future()
             fut.set_exception(err)
             return fut
         else:
diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py
index e3a1d5e..c453f02 100644
--- a/Lib/asyncio/queues.py
+++ b/Lib/asyncio/queues.py
@@ -128,7 +128,7 @@ class Queue:
         This method is a coroutine.
         """
         while self.full():
-            putter = futures.Future(loop=self._loop)
+            putter = self._loop.create_future()
             self._putters.append(putter)
             try:
                 yield from putter
@@ -162,7 +162,7 @@ class Queue:
         This method is a coroutine.
         """
         while self.empty():
-            getter = futures.Future(loop=self._loop)
+            getter = self._loop.create_future()
             self._getters.append(getter)
             try:
                 yield from getter
diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py
index 7b5a084..b34fee3 100644
--- a/Lib/asyncio/selector_events.py
+++ b/Lib/asyncio/selector_events.py
@@ -196,7 +196,7 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
         transport = None
         try:
             protocol = protocol_factory()
-            waiter = futures.Future(loop=self)
+            waiter = self.create_future()
             if sslcontext:
                 transport = self._make_ssl_transport(
                     conn, protocol, sslcontext, waiter=waiter,
@@ -314,7 +314,7 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
         """
         if self._debug and sock.gettimeout() != 0:
             raise ValueError("the socket must be non-blocking")
-        fut = futures.Future(loop=self)
+        fut = self.create_future()
         self._sock_recv(fut, False, sock, n)
         return fut
 
@@ -352,7 +352,7 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
         """
         if self._debug and sock.gettimeout() != 0:
             raise ValueError("the socket must be non-blocking")
-        fut = futures.Future(loop=self)
+        fut = self.create_future()
         if data:
             self._sock_sendall(fut, False, sock, data)
         else:
@@ -395,7 +395,7 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
         """
         if self._debug and sock.gettimeout() != 0:
             raise ValueError("the socket must be non-blocking")
-        fut = futures.Future(loop=self)
+        fut = self.create_future()
         try:
             base_events._check_resolved_address(sock, address)
         except ValueError as err:
@@ -453,7 +453,7 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
         """
         if self._debug and sock.gettimeout() != 0:
             raise ValueError("the socket must be non-blocking")
-        fut = futures.Future(loop=self)
+        fut = self.create_future()
         self._sock_accept(fut, False, sock)
         return fut
 
diff --git a/Lib/asyncio/streams.py b/Lib/asyncio/streams.py
index b7b0485..da3d526 100644
--- a/Lib/asyncio/streams.py
+++ b/Lib/asyncio/streams.py
@@ -210,7 +210,7 @@ class FlowControlMixin(protocols.Protocol):
             return
         waiter = self._drain_waiter
         assert waiter is None or waiter.cancelled()
-        waiter = futures.Future(loop=self._loop)
+        waiter = self._loop.create_future()
         self._drain_waiter = waiter
         yield from waiter
 
@@ -449,7 +449,7 @@ class StreamReader:
             self._paused = False
             self._transport.resume_reading()
 
-        self._waiter = futures.Future(loop=self._loop)
+        self._waiter = self._loop.create_future()
         try:
             yield from self._waiter
         finally:
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index cab4998..81510ba 100644
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -373,7 +373,7 @@ def wait_for(fut, timeout, *, loop=None):
     if timeout is None:
         return (yield from fut)
 
-    waiter = futures.Future(loop=loop)
+    waiter = loop.create_future()
     timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
     cb = functools.partial(_release_waiter, waiter)
 
@@ -406,7 +406,7 @@ def _wait(fs, timeout, return_when, loop):
     The fs argument must be a collection of Futures.
     """
     assert fs, 'Set of Futures is empty.'
-    waiter = futures.Future(loop=loop)
+    waiter = loop.create_future()
     timeout_handle = None
     if timeout is not None:
         timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
@@ -507,7 +507,9 @@ def sleep(delay, result=None, *, loop=None):
         yield
         return result
 
-    future = futures.Future(loop=loop)
+    if loop is None:
+        loop = events.get_event_loop()
+    future = loop.create_future()
     h = future._loop.call_later(delay,
                                 futures._set_result_unless_cancelled,
                                 future, result)
@@ -604,7 +606,9 @@ def gather(*coros_or_futures, loop=None, return_exceptions=False):
     be cancelled.)
     """
     if not coros_or_futures:
-        outer = futures.Future(loop=loop)
+        if loop is None:
+            loop = events.get_event_loop()
+        outer = loop.create_future()
         outer.set_result([])
         return outer
 
@@ -692,7 +696,7 @@ def shield(arg, *, loop=None):
         # Shortcut.
         return inner
     loop = inner._loop
-    outer = futures.Future(loop=loop)
+    outer = loop.create_future()
 
     def _done_callback(inner):
         if outer.cancelled():
diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
index b62dd38..d712749 100644
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -177,7 +177,7 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
                                    stdin, stdout, stderr, bufsize,
                                    extra=None, **kwargs):
         with events.get_child_watcher() as watcher:
-            waiter = futures.Future(loop=self)
+            waiter = self.create_future()
             transp = _UnixSubprocessTransport(self, protocol, args, shell,
                                               stdin, stdout, stderr, bufsize,
                                               waiter=waiter, extra=extra,
diff --git a/Lib/asyncio/windows_events.py b/Lib/asyncio/windows_events.py
index 7be3e02..668fe14 100644
--- a/Lib/asyncio/windows_events.py
+++ b/Lib/asyncio/windows_events.py
@@ -366,7 +366,7 @@ class ProactorEventLoop(proactor_events.BaseProactorEventLoop):
     def _make_subprocess_transport(self, protocol, args, shell,
                                    stdin, stdout, stderr, bufsize,
                                    extra=None, **kwargs):
-        waiter = futures.Future(loop=self)
+        waiter = self.create_future()
         transp = _WindowsSubprocessTransport(self, protocol, args, shell,
                                              stdin, stdout, stderr, bufsize,
                                              waiter=waiter, extra=extra,
@@ -417,7 +417,7 @@ class IocpProactor:
         return tmp
 
     def _result(self, value):
-        fut = futures.Future(loop=self._loop)
+        fut = self._loop.create_future()
         fut.set_result(value)
         return fut
 
diff --git a/Lib/test/test_asyncio/test_futures.py b/Lib/test/test_asyncio/test_futures.py
index e800106..c38c1f2 100644
--- a/Lib/test/test_asyncio/test_futures.py
+++ b/Lib/test/test_asyncio/test_futures.py
@@ -278,14 +278,15 @@ class FutureTests(test_utils.TestCase):
         f2 = asyncio.wrap_future(f1)
         self.assertIs(f1, f2)
 
-    @mock.patch('asyncio.futures.events')
-    def test_wrap_future_use_global_loop(self, m_events):
-        def run(arg):
-            return (arg, threading.get_ident())
-        ex = concurrent.futures.ThreadPoolExecutor(1)
-        f1 = ex.submit(run, 'oi')
-        f2 = asyncio.wrap_future(f1)
-        self.assertIs(m_events.get_event_loop.return_value, f2._loop)
+    def test_wrap_future_use_global_loop(self):
+        with mock.patch('asyncio.futures.events') as events:
+            events.get_event_loop = lambda: self.loop
+            def run(arg):
+                return (arg, threading.get_ident())
+            ex = concurrent.futures.ThreadPoolExecutor(1)
+            f1 = ex.submit(run, 'oi')
+            f2 = asyncio.wrap_future(f1)
+            self.assertIs(self.loop, f2._loop)
 
     def test_wrap_future_cancel(self):
         f1 = concurrent.futures.Future()