From b7670e81eac2996109756ba098cf964f6888a5e9 Mon Sep 17 00:00:00 2001 From: Yury Selivanov Date: Thu, 12 Nov 2015 12:17:30 -0500 Subject: [PATCH] Add loop.create_future() and start using it --- asyncio/base_events.py | 18 +++++++++++------- asyncio/base_subprocess.py | 2 +- asyncio/events.py | 3 +++ asyncio/futures.py | 4 +++- asyncio/locks.py | 8 ++++---- asyncio/proactor_events.py | 2 +- asyncio/queues.py | 4 ++-- asyncio/selector_events.py | 10 +++++----- asyncio/streams.py | 4 ++-- asyncio/tasks.py | 14 +++++++++----- asyncio/unix_events.py | 2 +- asyncio/windows_events.py | 4 ++-- tests/test_futures.py | 17 +++++++++-------- 13 files changed, 53 insertions(+), 39 deletions(-) diff --git a/asyncio/base_events.py b/asyncio/base_events.py index ada178f3..313cc316 100644 --- a/asyncio/base_events.py +++ b/asyncio/base_events.py @@ -209,7 +209,7 @@ def _wakeup(self): 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 @@ def __repr__(self): % (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 @@ def run_in_executor(self, executor, func, *args): 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 @@ def getaddrinfo(self, host, port, *, 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 @@ def create_connection(self, protocol_factory, host=None, port=None, *, 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 @@ def create_datagram_endpoint(self, protocol_factory, 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 @@ def create_server(self, protocol_factory, host=None, port=None, @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 @@ def connect_read_pipe(self, protocol_factory, pipe): @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/asyncio/base_subprocess.py b/asyncio/base_subprocess.py index 73425d9b..08080bd7 100644 --- a/asyncio/base_subprocess.py +++ b/asyncio/base_subprocess.py @@ -227,7 +227,7 @@ def _wait(self): 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/asyncio/events.py b/asyncio/events.py index 8358ebfd..c48c5bed 100644 --- a/asyncio/events.py +++ b/asyncio/events.py @@ -266,6 +266,9 @@ def call_at(self, when, callback, *args): 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/asyncio/futures.py b/asyncio/futures.py index ddb9cde1..1feba4d3 100644 --- a/asyncio/futures.py +++ b/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/asyncio/locks.py b/asyncio/locks.py index 34f6bc16..1804d7b8 100644 --- a/asyncio/locks.py +++ b/asyncio/locks.py @@ -170,7 +170,7 @@ def acquire(self): 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 @@ def wait(self): 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 @@ def wait(self): 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 @@ def acquire(self): 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/asyncio/proactor_events.py b/asyncio/proactor_events.py index b2ddee40..eb92458a 100644 --- a/asyncio/proactor_events.py +++ b/asyncio/proactor_events.py @@ -443,7 +443,7 @@ def sock_connect(self, sock, address): 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/asyncio/queues.py b/asyncio/queues.py index e3a1d5ed..c453f02d 100644 --- a/asyncio/queues.py +++ b/asyncio/queues.py @@ -128,7 +128,7 @@ def put(self, item): 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 @@ def get(self): 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/asyncio/selector_events.py b/asyncio/selector_events.py index 7b5a084d..b34fee34 100644 --- a/asyncio/selector_events.py +++ b/asyncio/selector_events.py @@ -196,7 +196,7 @@ def _accept_connection2(self, protocol_factory, conn, extra, 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 @@ def sock_recv(self, sock, n): """ 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 @@ def sock_sendall(self, sock, data): """ 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 @@ def sock_connect(self, sock, address): """ 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 @@ def sock_accept(self, sock): """ 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/asyncio/streams.py b/asyncio/streams.py index b7b0485a..da3d5263 100644 --- a/asyncio/streams.py +++ b/asyncio/streams.py @@ -210,7 +210,7 @@ def _drain_helper(self): 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 @@ def _wait_for_data(self, func_name): 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/asyncio/tasks.py b/asyncio/tasks.py index cab4998e..81510ba8 100644 --- a/asyncio/tasks.py +++ b/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/asyncio/unix_events.py b/asyncio/unix_events.py index b62dd389..d712749e 100644 --- a/asyncio/unix_events.py +++ b/asyncio/unix_events.py @@ -177,7 +177,7 @@ def _make_subprocess_transport(self, protocol, args, shell, 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/asyncio/windows_events.py b/asyncio/windows_events.py index 7be3e022..668fe145 100644 --- a/asyncio/windows_events.py +++ b/asyncio/windows_events.py @@ -366,7 +366,7 @@ def loop_accept_pipe(f=None): 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 @@ def select(self, timeout=None): 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/tests/test_futures.py b/tests/test_futures.py index e8001062..c38c1f29 100644 --- a/tests/test_futures.py +++ b/tests/test_futures.py @@ -278,14 +278,15 @@ def test_wrap_future_future(self): 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()