[gentoo-commits] proj/portage:master commit in: lib/portage/util/futures/

[gentoo-commits] proj/portage:master commit in: lib/portage/util/futures/

[Zac Medico]

commit:     ed685eb659fccf6e4031d12fa8a59c3829ef1155
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Sun Mar  7 06:56:36 2021 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Sun Mar  7 07:01:44 2021 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=ed685eb6

Removed unused portage.util.futures.transports

Signed-off-by: Zac Medico <zmedico <AT> gentoo.org>

 lib/portage/util/futures/transports.py  |  87 ------
 lib/portage/util/futures/unix_events.py | 492 +-------------------------------
 2 files changed, 2 insertions(+), 577 deletions(-)

diff --git a/lib/portage/util/futures/transports.py b/lib/portage/util/futures/transports.py
deleted file mode 100644
index 016ecbef8..000000000
--- a/lib/portage/util/futures/transports.py
+++ /dev/null
@@ -1,87 +0,0 @@
-# Copyright 2018 Gentoo Foundation
-# Distributed under the terms of the GNU General Public License v2
-
-from asyncio.transports import Transport as _Transport
-
-
-class _FlowControlMixin(_Transport):
-	"""
-	This is identical to the standard library's private
-	asyncio.transports._FlowControlMixin class.
-
-	All the logic for (write) flow control in a mix-in base class.
-
-	The subclass must implement get_write_buffer_size().  It must call
-	_maybe_pause_protocol() whenever the write buffer size increases,
-	and _maybe_resume_protocol() whenever it decreases.  It may also
-	override set_write_buffer_limits() (e.g. to specify different
-	defaults).
-
-	The subclass constructor must call super().__init__(extra).  This
-	will call set_write_buffer_limits().
-
-	The user may call set_write_buffer_limits() and
-	get_write_buffer_size(), and their protocol's pause_writing() and
-	resume_writing() may be called.
-	"""
-
-	def __init__(self, extra=None, loop=None):
-		super().__init__(extra)
-		assert loop is not None
-		self._loop = loop
-		self._protocol_paused = False
-		self._set_write_buffer_limits()
-
-	def _maybe_pause_protocol(self):
-		size = self.get_write_buffer_size()
-		if size <= self._high_water:
-			return
-		if not self._protocol_paused:
-			self._protocol_paused = True
-			try:
-				self._protocol.pause_writing()
-			except Exception as exc:
-				self._loop.call_exception_handler({
-					'message': 'protocol.pause_writing() failed',
-					'exception': exc,
-					'transport': self,
-					'protocol': self._protocol,
-				})
-
-	def _maybe_resume_protocol(self):
-		if (self._protocol_paused and
-			self.get_write_buffer_size() <= self._low_water):
-			self._protocol_paused = False
-			try:
-				self._protocol.resume_writing()
-			except Exception as exc:
-				self._loop.call_exception_handler({
-					'message': 'protocol.resume_writing() failed',
-					'exception': exc,
-					'transport': self,
-					'protocol': self._protocol,
-				})
-
-	def get_write_buffer_limits(self):
-		return (self._low_water, self._high_water)
-
-	def _set_write_buffer_limits(self, high=None, low=None):
-		if high is None:
-			if low is None:
-				high = 64*1024
-			else:
-				high = 4*low
-		if low is None:
-			low = high // 4
-		if not high >= low >= 0:
-			raise ValueError('high (%r) must be >= low (%r) must be >= 0' %
-							 (high, low))
-		self._high_water = high
-		self._low_water = low
-
-	def set_write_buffer_limits(self, high=None, low=None):
-		self._set_write_buffer_limits(high=high, low=low)
-		self._maybe_pause_protocol()
-
-	def get_write_buffer_size(self):
-		raise NotImplementedError

diff --git a/lib/portage/util/futures/unix_events.py b/lib/portage/util/futures/unix_events.py
index 16a9e12b7..9d5445943 100644
--- a/lib/portage/util/futures/unix_events.py
+++ b/lib/portage/util/futures/unix_events.py
@@ -1,4 +1,4 @@
-# Copyright 2018 Gentoo Foundation
+# Copyright 2018-2021 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
 __all__ = (
@@ -7,32 +7,15 @@ __all__ = (
 )
 
 import asyncio as _real_asyncio
-from asyncio.base_subprocess import BaseSubprocessTransport as _BaseSubprocessTransport
 from asyncio.unix_events import AbstractChildWatcher as _AbstractChildWatcher
-from asyncio.transports import (
-	ReadTransport as _ReadTransport,
-	WriteTransport as _WriteTransport,
-)
 
-import errno
 import fcntl
-import functools
-import logging
 import os
-import socket
-import stat
-import subprocess
-import sys
 
 from portage.util._eventloop.global_event_loop import (
 	global_event_loop as _global_event_loop,
 )
-from portage.util.futures import (
-	asyncio,
-	events,
-)
-
-from portage.util.futures.transports import _FlowControlMixin
+from portage.util.futures import events
 
 
 class _PortageEventLoop(events.AbstractEventLoop):
@@ -89,158 +72,6 @@ class _PortageEventLoop(events.AbstractEventLoop):
 		"""
 		return self
 
-	def create_task(self, coro):
-		"""
-		Schedule a coroutine object.
-
-		@type coro: coroutine
-		@param coro: a coroutine to schedule
-		@rtype: asyncio.Task
-		@return: a task object
-		"""
-		return asyncio.Task(coro, loop=self)
-
-	def connect_read_pipe(self, protocol_factory, pipe):
-		"""
-		Register read pipe in event loop. Set the pipe to non-blocking mode.
-
-		@type protocol_factory: callable
-		@param protocol_factory: must instantiate object with Protocol interface
-		@type pipe: file
-		@param pipe: a pipe to read from
-		@rtype: asyncio.Future
-		@return: Return pair (transport, protocol), where transport supports the
-			ReadTransport interface.
-		"""
-		protocol = protocol_factory()
-		result = self.create_future()
-		waiter = self.create_future()
-		transport = self._make_read_pipe_transport(pipe, protocol, waiter=waiter)
-
-		def waiter_callback(waiter):
-			try:
-				waiter.result()
-			except Exception as e:
-				transport.close()
-				result.set_exception(e)
-			else:
-				result.set_result((transport, protocol))
-
-		waiter.add_done_callback(waiter_callback)
-		return result
-
-	def connect_write_pipe(self, protocol_factory, pipe):
-		"""
-		Register write pipe in event loop. Set the pipe to non-blocking mode.
-
-		@type protocol_factory: callable
-		@param protocol_factory: must instantiate object with Protocol interface
-		@type pipe: file
-		@param pipe: a pipe to write to
-		@rtype: asyncio.Future
-		@return: Return pair (transport, protocol), where transport supports the
-			WriteTransport interface.
-		"""
-		protocol = protocol_factory()
-		result = self.create_future()
-		waiter = self.create_future()
-		transport = self._make_write_pipe_transport(pipe, protocol, waiter)
-
-		def waiter_callback(waiter):
-			try:
-				waiter.result()
-			except Exception as e:
-				transport.close()
-				result.set_exception(e)
-			else:
-				result.set_result((transport, protocol))
-
-		waiter.add_done_callback(waiter_callback)
-		return result
-
-	def subprocess_exec(self, protocol_factory, program, *args, **kwargs):
-		"""
-		Run subprocesses asynchronously using the subprocess module.
-
-		@type protocol_factory: callable
-		@param protocol_factory: must instantiate a subclass of the
-			asyncio.SubprocessProtocol class
-		@type program: str or bytes
-		@param program: the program to execute
-		@type args: str or bytes
-		@param args: program's arguments
-		@type kwargs: varies
-		@param kwargs: subprocess.Popen parameters
-		@rtype: asyncio.Future
-		@return: Returns a pair of (transport, protocol), where transport
-			is an instance of BaseSubprocessTransport
-		"""
-
-		# python2.7 does not allow arguments with defaults after *args
-		stdin = kwargs.pop('stdin', subprocess.PIPE)
-		stdout = kwargs.pop('stdout', subprocess.PIPE)
-		stderr = kwargs.pop('stderr', subprocess.PIPE)
-
-		universal_newlines = kwargs.pop('universal_newlines', False)
-		shell = kwargs.pop('shell', False)
-		bufsize = kwargs.pop('bufsize', 0)
-
-		if universal_newlines:
-			raise ValueError("universal_newlines must be False")
-		if shell:
-			raise ValueError("shell must be False")
-		if bufsize != 0:
-			raise ValueError("bufsize must be 0")
-		popen_args = (program,) + args
-		for arg in popen_args:
-			if not isinstance(arg, (str, bytes)):
-				raise TypeError("program arguments must be "
-								"a bytes or text string, not %s"
-								% type(arg).__name__)
-		result = self.create_future()
-		self._make_subprocess_transport(
-			result, protocol_factory(), popen_args, False, stdin, stdout, stderr,
-			bufsize, **kwargs)
-		return result
-
-	def _make_read_pipe_transport(self, pipe, protocol, waiter=None,
-								  extra=None):
-		return _UnixReadPipeTransport(self, pipe, protocol, waiter, extra)
-
-	def _make_write_pipe_transport(self, pipe, protocol, waiter=None,
-								   extra=None):
-		return _UnixWritePipeTransport(self, pipe, protocol, waiter, extra)
-
-	def _make_subprocess_transport(self, result, protocol, args, shell,
-		stdin, stdout, stderr, bufsize, extra=None, **kwargs):
-		waiter = self.create_future()
-		transp = _UnixSubprocessTransport(self,
-			protocol, args, shell, stdin, stdout, stderr, bufsize,
-			waiter=waiter, extra=extra,
-			**kwargs)
-
-		self._loop._asyncio_child_watcher.add_child_handler(
-			transp.get_pid(), self._child_watcher_callback, transp)
-
-		waiter.add_done_callback(functools.partial(
-			self._subprocess_transport_callback, transp, protocol, result))
-
-	def _subprocess_transport_callback(self, transp, protocol, result, waiter):
-		if waiter.exception() is None:
-			result.set_result((transp, protocol))
-		else:
-			transp.close()
-			wait_transp = asyncio.ensure_future(transp._wait(), loop=self)
-			wait_transp.add_done_callback(
-				functools.partial(self._subprocess_transport_failure,
-				result, waiter.exception()))
-
-	def _child_watcher_callback(self, pid, returncode, transp):
-		self.call_soon_threadsafe(transp._process_exited, returncode)
-
-	def _subprocess_transport_failure(self, result, exception, wait_transp):
-		result.set_exception(wait_transp.exception() or exception)
-
 
 if hasattr(os, 'set_blocking'):
 	def _set_nonblocking(fd):
@@ -252,325 +83,6 @@ else:
 		fcntl.fcntl(fd, fcntl.F_SETFL, flags)
 
 
-class _UnixReadPipeTransport(_ReadTransport):
-	"""
-	This is identical to the standard library's private
-	asyncio.unix_events._UnixReadPipeTransport class, except that it
-	only calls public AbstractEventLoop methods.
-	"""
-
-	max_size = 256 * 1024  # max bytes we read in one event loop iteration
-
-	def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
-		super().__init__(extra)
-		self._extra['pipe'] = pipe
-		self._loop = loop
-		self._pipe = pipe
-		self._fileno = pipe.fileno()
-		self._protocol = protocol
-		self._closing = False
-
-		mode = os.fstat(self._fileno).st_mode
-		if not (stat.S_ISFIFO(mode) or
-				stat.S_ISSOCK(mode) or
-				stat.S_ISCHR(mode)):
-			self._pipe = None
-			self._fileno = None
-			self._protocol = None
-			raise ValueError("Pipe transport is for pipes/sockets only.")
-
-		_set_nonblocking(self._fileno)
-
-		self._loop.call_soon(self._protocol.connection_made, self)
-		# only start reading when connection_made() has been called
-		self._loop.call_soon(self._loop.add_reader,
-							 self._fileno, self._read_ready)
-		if waiter is not None:
-			# only wake up the waiter when connection_made() has been called
-			self._loop.call_soon(
-				lambda: None if waiter.cancelled() else waiter.set_result(None))
-
-	def _read_ready(self):
-		try:
-			data = os.read(self._fileno, self.max_size)
-		except (BlockingIOError, InterruptedError):
-			pass
-		except OSError as exc:
-			self._fatal_error(exc, 'Fatal read error on pipe transport')
-		else:
-			if data:
-				self._protocol.data_received(data)
-			else:
-				self._closing = True
-				self._loop.remove_reader(self._fileno)
-				self._loop.call_soon(self._protocol.eof_received)
-				self._loop.call_soon(self._call_connection_lost, None)
-
-	def pause_reading(self):
-		self._loop.remove_reader(self._fileno)
-
-	def resume_reading(self):
-		self._loop.add_reader(self._fileno, self._read_ready)
-
-	def set_protocol(self, protocol):
-		self._protocol = protocol
-
-	def get_protocol(self):
-		return self._protocol
-
-	def is_closing(self):
-		return self._closing
-
-	def close(self):
-		if not self._closing:
-			self._close(None)
-
-	def _fatal_error(self, exc, message='Fatal error on pipe transport'):
-		# should be called by exception handler only
-		if (isinstance(exc, OSError) and exc.errno == errno.EIO):
-			if self._loop.get_debug():
-				logging.debug("%r: %s", self, message, exc_info=True)
-		else:
-			self._loop.call_exception_handler({
-				'message': message,
-				'exception': exc,
-				'transport': self,
-				'protocol': self._protocol,
-			})
-		self._close(exc)
-
-	def _close(self, exc):
-		self._closing = True
-		self._loop.remove_reader(self._fileno)
-		self._loop.call_soon(self._call_connection_lost, exc)
-
-	def _call_connection_lost(self, exc):
-		try:
-			self._protocol.connection_lost(exc)
-		finally:
-			self._pipe.close()
-			self._pipe = None
-			self._protocol = None
-			self._loop = None
-
-
-class _UnixWritePipeTransport(_FlowControlMixin, _WriteTransport):
-	"""
-	This is identical to the standard library's private
-	asyncio.unix_events._UnixWritePipeTransport class, except that it
-	only calls public AbstractEventLoop methods.
-	"""
-
-	def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
-		super().__init__(extra, loop)
-		self._extra['pipe'] = pipe
-		self._pipe = pipe
-		self._fileno = pipe.fileno()
-		self._protocol = protocol
-		self._buffer = bytearray()
-		self._conn_lost = 0
-		self._closing = False  # Set when close() or write_eof() called.
-
-		mode = os.fstat(self._fileno).st_mode
-		is_char = stat.S_ISCHR(mode)
-		is_fifo = stat.S_ISFIFO(mode)
-		is_socket = stat.S_ISSOCK(mode)
-		if not (is_char or is_fifo or is_socket):
-			self._pipe = None
-			self._fileno = None
-			self._protocol = None
-			raise ValueError("Pipe transport is only for "
-							 "pipes, sockets and character devices")
-
-		_set_nonblocking(self._fileno)
-		self._loop.call_soon(self._protocol.connection_made, self)
-
-		# On AIX, the reader trick (to be notified when the read end of the
-		# socket is closed) only works for sockets. On other platforms it
-		# works for pipes and sockets. (Exception: OS X 10.4?  Issue #19294.)
-		if is_socket or (is_fifo and not sys.platform.startswith("aix")):
-			# only start reading when connection_made() has been called
-			self._loop.call_soon(self._loop.add_reader,
-								 self._fileno, self._read_ready)
-
-		if waiter is not None:
-			# only wake up the waiter when connection_made() has been called
-			self._loop.call_soon(
-				lambda: None if waiter.cancelled() else waiter.set_result(None))
-
-	def get_write_buffer_size(self):
-		return len(self._buffer)
-
-	def _read_ready(self):
-		# Pipe was closed by peer.
-		if self._loop.get_debug():
-			logging.info("%r was closed by peer", self)
-		if self._buffer:
-			self._close(BrokenPipeError())
-		else:
-			self._close()
-
-	def write(self, data):
-		assert isinstance(data, (bytes, bytearray, memoryview)), repr(data)
-		if isinstance(data, bytearray):
-			data = memoryview(data)
-		if not data:
-			return
-
-		if self._conn_lost or self._closing:
-			self._conn_lost += 1
-			return
-
-		if not self._buffer:
-			# Attempt to send it right away first.
-			try:
-				n = os.write(self._fileno, data)
-			except (BlockingIOError, InterruptedError):
-				n = 0
-			except Exception as exc:
-				self._conn_lost += 1
-				self._fatal_error(exc, 'Fatal write error on pipe transport')
-				return
-			if n == len(data):
-				return
-			if n > 0:
-				data = memoryview(data)[n:]
-			self._loop.add_writer(self._fileno, self._write_ready)
-
-		self._buffer += data
-		self._maybe_pause_protocol()
-
-	def _write_ready(self):
-		assert self._buffer, 'Data should not be empty'
-
-		try:
-			n = os.write(self._fileno, self._buffer)
-		except (BlockingIOError, InterruptedError):
-			pass
-		except Exception as exc:
-			self._buffer.clear()
-			self._conn_lost += 1
-			# Remove writer here, _fatal_error() doesn't it
-			# because _buffer is empty.
-			self._loop.remove_writer(self._fileno)
-			self._fatal_error(exc, 'Fatal write error on pipe transport')
-		else:
-			if n == len(self._buffer):
-				self._buffer.clear()
-				self._loop.remove_writer(self._fileno)
-				self._maybe_resume_protocol()  # May append to buffer.
-				if self._closing:
-					self._loop.remove_reader(self._fileno)
-					self._call_connection_lost(None)
-				return
-			if n > 0:
-				del self._buffer[:n]
-
-	def can_write_eof(self):
-		return True
-
-	def write_eof(self):
-		if self._closing:
-			return
-		assert self._pipe
-		self._closing = True
-		if not self._buffer:
-			self._loop.remove_reader(self._fileno)
-			self._loop.call_soon(self._call_connection_lost, None)
-
-	def set_protocol(self, protocol):
-		self._protocol = protocol
-
-	def get_protocol(self):
-		return self._protocol
-
-	def is_closing(self):
-		return self._closing
-
-	def close(self):
-		if self._pipe is not None and not self._closing:
-			# write_eof is all what we needed to close the write pipe
-			self.write_eof()
-
-	def abort(self):
-		self._close(None)
-
-	def _fatal_error(self, exc, message='Fatal error on pipe transport'):
-		# should be called by exception handler only
-		if isinstance(exc,
-			(BrokenPipeError, ConnectionResetError, ConnectionAbortedError)):
-			if self._loop.get_debug():
-				logging.debug("%r: %s", self, message, exc_info=True)
-		else:
-			self._loop.call_exception_handler({
-				'message': message,
-				'exception': exc,
-				'transport': self,
-				'protocol': self._protocol,
-			})
-		self._close(exc)
-
-	def _close(self, exc=None):
-		self._closing = True
-		if self._buffer:
-			self._loop.remove_writer(self._fileno)
-		self._buffer.clear()
-		self._loop.remove_reader(self._fileno)
-		self._loop.call_soon(self._call_connection_lost, exc)
-
-	def _call_connection_lost(self, exc):
-		try:
-			self._protocol.connection_lost(exc)
-		finally:
-			self._pipe.close()
-			self._pipe = None
-			self._protocol = None
-			self._loop = None
-
-
-if hasattr(os, 'set_inheritable'):
-	# Python 3.4 and newer
-	_set_inheritable = os.set_inheritable
-else:
-	def _set_inheritable(fd, inheritable):
-		cloexec_flag = getattr(fcntl, 'FD_CLOEXEC', 1)
-
-		old = fcntl.fcntl(fd, fcntl.F_GETFD)
-		if not inheritable:
-			fcntl.fcntl(fd, fcntl.F_SETFD, old | cloexec_flag)
-		else:
-			fcntl.fcntl(fd, fcntl.F_SETFD, old & ~cloexec_flag)
-
-
-class _UnixSubprocessTransport(_BaseSubprocessTransport):
-	"""
-	This is identical to the standard library's private
-	asyncio.unix_events._UnixSubprocessTransport class, except that it
-	only calls public AbstractEventLoop methods.
-	"""
-	def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
-		stdin_w = None
-		if stdin == subprocess.PIPE:
-			# Use a socket pair for stdin, since not all platforms
-			# support selecting read events on the write end of a
-			# socket (which we use in order to detect closing of the
-			# other end).  Notably this is needed on AIX, and works
-			# just fine on other platforms.
-			stdin, stdin_w = socket.socketpair()
-
-			# Mark the write end of the stdin pipe as non-inheritable,
-			# needed by close_fds=False on Python 3.3 and older
-			# (Python 3.4 implements the PEP 446, socketpair returns
-			# non-inheritable sockets)
-			_set_inheritable(stdin_w.fileno(), False)
-		self._proc = subprocess.Popen(
-			args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
-			universal_newlines=False, bufsize=bufsize, **kwargs)
-		if stdin_w is not None:
-			stdin.close()
-			self._proc.stdin = os.fdopen(stdin_w.detach(), 'wb', bufsize)
-
-
 class AbstractChildWatcher(_AbstractChildWatcher):
 	def add_child_handler(self, pid, callback, *args):
 		raise NotImplementedError()

[gentoo-commits] proj/portage:master commit in: lib/portage/util/futures/

[Zac Medico]

commit:     2f474767eaf3b042bea31bc524281e0841cb271b
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Mon Sep 20 06:44:36 2021 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Mon Sep 20 07:07:22 2021 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=2f474767

ExtendedFuture: Fix python3.10 DeprecationWarning

DeprecationWarning: There is no current event loop

Signed-off-by: Zac Medico <zmedico <AT> gentoo.org>

 lib/portage/util/futures/extendedfutures.py | 24 +++++++++++++++++++++---
 1 file changed, 21 insertions(+), 3 deletions(-)

diff --git a/lib/portage/util/futures/extendedfutures.py b/lib/portage/util/futures/extendedfutures.py
index 6aa3ebbe8..c23feafb5 100644
--- a/lib/portage/util/futures/extendedfutures.py
+++ b/lib/portage/util/futures/extendedfutures.py
@@ -10,7 +10,13 @@ __all__ = (
     "InvalidStateError",
 )
 
-from portage.util.futures.futures import Future, InvalidStateError, CancelledError
+import concurrent.futures
+from concurrent.futures import Future, CancelledError
+
+try:
+    from concurrent.futures import InvalidStateError
+except ImportError:
+    from portage.util.futures.futures import InvalidStateError
 
 # Create our one time settable unset constant
 UNSET_CONST = Future()
@@ -62,9 +68,21 @@ class ExtendedFuture(Future):
             default = self.default_result
         if default is not UNSET_CONST.result():
             try:
-                data = super(ExtendedFuture, self).result()
+                data = self.result()
             except InvalidStateError:
                 data = default
         else:
-            data = super(ExtendedFuture, self).result()
+            data = self.result()
         return data
+
+    def exception(self):
+        try:
+            return super(ExtendedFuture, self).exception(timeout=0)
+        except concurrent.futures.TimeoutError:
+            raise InvalidStateError
+
+    def result(self):
+        try:
+            return super(ExtendedFuture, self).result(timeout=0)
+        except concurrent.futures.TimeoutError:
+            raise InvalidStateError

[gentoo-commits] proj/portage:master commit in: lib/portage/util/futures/

[Zac Medico]

commit:     20204fd8c29f3060da9891879721a54486247b0c
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Sun Mar  7 07:44:49 2021 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Sun Mar  7 07:45:11 2021 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=20204fd8

Remove unused _PortageEventLoop and _PortageChildWatcher

Signed-off-by: Zac Medico <zmedico <AT> gentoo.org>

 lib/portage/util/futures/unix_events.py | 112 --------------------------------
 1 file changed, 112 deletions(-)

diff --git a/lib/portage/util/futures/unix_events.py b/lib/portage/util/futures/unix_events.py
index 030070b1b..4feee0a3b 100644
--- a/lib/portage/util/futures/unix_events.py
+++ b/lib/portage/util/futures/unix_events.py
@@ -18,61 +18,6 @@ from portage.util._eventloop.global_event_loop import (
 )
 
 
-class _PortageEventLoop(events.AbstractEventLoop):
-	"""
-	Implementation of asyncio.AbstractEventLoop which wraps portage's
-	internal event loop.
-	"""
-
-	def __init__(self, loop):
-		"""
-		@type loop: EventLoop
-		@param loop: an instance of portage's internal event loop
-		"""
-		self._loop = loop
-		self.run_until_complete = loop.run_until_complete
-		self.call_soon = loop.call_soon
-		self.call_soon_threadsafe = loop.call_soon_threadsafe
-		self.call_later = loop.call_later
-		self.call_at = loop.call_at
-		self.is_running = loop.is_running
-		self.is_closed = loop.is_closed
-		self.close = loop.close
-		self.create_future = loop.create_future
-		self.add_reader = loop.add_reader
-		self.remove_reader = loop.remove_reader
-		self.add_writer = loop.add_writer
-		self.remove_writer = loop.remove_writer
-		self.run_in_executor = loop.run_in_executor
-		self.time = loop.time
-		self.default_exception_handler = loop.default_exception_handler
-		self.call_exception_handler = loop.call_exception_handler
-		self.set_debug = loop.set_debug
-		self.get_debug = loop.get_debug
-
-	@property
-	def _asyncio_child_watcher(self):
-		"""
-		In order to avoid accessing the internal _loop attribute, portage
-		internals should use this property when possible.
-
-		@rtype: asyncio.AbstractChildWatcher
-		@return: the internal event loop's AbstractChildWatcher interface
-		"""
-		return self._loop._asyncio_child_watcher
-
-	@property
-	def _asyncio_wrapper(self):
-		"""
-		In order to avoid accessing the internal _loop attribute, portage
-		internals should use this property when possible.
-
-		@rtype: asyncio.AbstractEventLoop
-		@return: the internal event loop's AbstractEventLoop interface
-		"""
-		return self
-
-
 if hasattr(os, 'set_blocking'):
 	def _set_nonblocking(fd):
 		os.set_blocking(fd, False)
@@ -83,63 +28,6 @@ else:
 		fcntl.fcntl(fd, fcntl.F_SETFL, flags)
 
 
-class _PortageChildWatcher(AbstractChildWatcher):
-	def __init__(self, loop):
-		"""
-		@type loop: EventLoop
-		@param loop: an instance of portage's internal event loop
-		"""
-		self._loop = loop
-		self._callbacks = {}
-
-	def close(self):
-		pass
-
-	def __enter__(self):
-		return self
-
-	def __exit__(self, a, b, c):
-		pass
-
-	def _child_exit(self, pid, status, data):
-		self._callbacks.pop(pid)
-		callback, args = data
-		callback(pid, self._compute_returncode(status), *args)
-
-	def _compute_returncode(self, status):
-		if os.WIFSIGNALED(status):
-			return -os.WTERMSIG(status)
-		if os.WIFEXITED(status):
-			return os.WEXITSTATUS(status)
-		return status
-
-	def add_child_handler(self, pid, callback, *args):
-		"""
-		Register a new child handler.
-
-		Arrange for callback(pid, returncode, *args) to be called when
-		process 'pid' terminates. Specifying another callback for the same
-		process replaces the previous handler.
-		"""
-		source_id = self._callbacks.get(pid)
-		if source_id is not None:
-			self._loop.source_remove(source_id)
-		self._callbacks[pid] = self._loop.child_watch_add(
-			pid, self._child_exit, data=(callback, args))
-
-	def remove_child_handler(self, pid):
-		"""
-		Removes the handler for process 'pid'.
-
-		The function returns True if the handler was successfully removed,
-		False if there was nothing to remove.
-		"""
-		source_id = self._callbacks.pop(pid, None)
-		if source_id is not None:
-			return self._loop.source_remove(source_id)
-		return False
-
-
 class _PortageEventLoopPolicy(events.AbstractEventLoopPolicy):
 	"""
 	Implementation of asyncio.AbstractEventLoopPolicy based on portage's

[gentoo-commits] proj/portage:master commit in: lib/portage/util/futures/

[Zac Medico]

commit:     b4f6a5126437afc5bb07ed0a42a0eb8a5745ef90
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Sun Mar  7 07:18:58 2021 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Sun Mar  7 07:32:04 2021 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=b4f6a512

Removed unused portage.util.futures.events

Signed-off-by: Zac Medico <zmedico <AT> gentoo.org>

 lib/portage/util/futures/events.py      | 186 --------------------------------
 lib/portage/util/futures/unix_events.py |  26 +----
 2 files changed, 3 insertions(+), 209 deletions(-)

diff --git a/lib/portage/util/futures/events.py b/lib/portage/util/futures/events.py
deleted file mode 100644
index 85032fcdf..000000000
--- a/lib/portage/util/futures/events.py
+++ /dev/null
@@ -1,186 +0,0 @@
-# Copyright 2018 Gentoo Foundation
-# Distributed under the terms of the GNU General Public License v2
-
-__all__ = (
-	'AbstractEventLoopPolicy',
-	'AbstractEventLoop',
-)
-
-import socket
-import subprocess
-
-from asyncio.events import (
-	AbstractEventLoop as _AbstractEventLoop,
-	AbstractEventLoopPolicy as _AbstractEventLoopPolicy,
-)
-
-
-class AbstractEventLoopPolicy(_AbstractEventLoopPolicy):
-	"""Abstract policy for accessing the event loop."""
-
-	def get_event_loop(self):
-		raise NotImplementedError
-
-	def set_event_loop(self, loop):
-		raise NotImplementedError
-
-	def new_event_loop(self):
-		raise NotImplementedError
-
-	def get_child_watcher(self):
-		raise NotImplementedError
-
-	def set_child_watcher(self, watcher):
-		raise NotImplementedError
-
-
-class AbstractEventLoop(_AbstractEventLoop):
-	"""Abstract event loop."""
-
-	def run_forever(self):
-		raise NotImplementedError
-
-	def run_until_complete(self, future):
-		raise NotImplementedError
-
-	def stop(self):
-		raise NotImplementedError
-
-	def is_running(self):
-		raise NotImplementedError
-
-	def is_closed(self):
-		raise NotImplementedError
-
-	def close(self):
-		raise NotImplementedError
-
-	def shutdown_asyncgens(self):
-		raise NotImplementedError
-
-	def _timer_handle_cancelled(self, handle):
-		raise NotImplementedError
-
-	def call_soon(self, callback, *args):
-		return self.call_later(0, callback, *args)
-
-	def call_later(self, delay, callback, *args):
-		raise NotImplementedError
-
-	def call_at(self, when, callback, *args):
-		raise NotImplementedError
-
-	def time(self):
-		raise NotImplementedError
-
-	def create_future(self):
-		raise NotImplementedError
-
-	def create_task(self, coro):
-		raise NotImplementedError
-
-	def call_soon_threadsafe(self, callback, *args):
-		raise NotImplementedError
-
-	def run_in_executor(self, executor, func, *args):
-		raise NotImplementedError
-
-	def set_default_executor(self, executor):
-		raise NotImplementedError
-
-	def getaddrinfo(self, host, port, family=0, type=0, proto=0, flags=0): # pylint: disable=redefined-builtin
-		raise NotImplementedError
-
-	def getnameinfo(self, sockaddr, flags=0):
-		raise NotImplementedError
-
-	def create_connection(self, protocol_factory, host=None, port=None,
-						  ssl=None, family=0, proto=0, flags=0, sock=None,
-						  local_addr=None, server_hostname=None):
-		raise NotImplementedError
-
-	def create_server(self, protocol_factory, host=None, port=None,
-					  family=socket.AF_UNSPEC, flags=socket.AI_PASSIVE,
-					  sock=None, backlog=100, ssl=None, reuse_address=None,
-					  reuse_port=None):
-		raise NotImplementedError
-
-	def create_unix_connection(self, protocol_factory, path,
-							   ssl=None, sock=None,
-							   server_hostname=None):
-		raise NotImplementedError
-
-	def create_unix_server(self, protocol_factory, path,
-						   sock=None, backlog=100, ssl=None):
-		raise NotImplementedError
-
-	def create_datagram_endpoint(self, protocol_factory,
-								 local_addr=None, remote_addr=None,
-								 family=0, proto=0, flags=0,
-								 reuse_address=None, reuse_port=None,
-								 allow_broadcast=None, sock=None):
-		raise NotImplementedError
-
-	def connect_read_pipe(self, protocol_factory, pipe):
-		raise NotImplementedError
-
-	def connect_write_pipe(self, protocol_factory, pipe):
-		raise NotImplementedError
-
-	def subprocess_shell(self, protocol_factory, cmd, stdin=subprocess.PIPE,
-						 stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-						 **kwargs):
-		raise NotImplementedError
-
-	def subprocess_exec(self, protocol_factory, *args, **kwargs):
-		for k in ('stdin', 'stdout', 'stderr'):
-			kwargs.setdefault(k, subprocess.PIPE)
-		raise NotImplementedError
-
-	def add_writer(self, fd, callback, *args):
-		raise NotImplementedError
-
-	def remove_writer(self, fd):
-		raise NotImplementedError
-
-	def sock_recv(self, sock, nbytes):
-		raise NotImplementedError
-
-	def sock_sendall(self, sock, data):
-		raise NotImplementedError
-
-	def sock_connect(self, sock, address):
-		raise NotImplementedError
-
-	def sock_accept(self, sock):
-		raise NotImplementedError
-
-	def add_signal_handler(self, sig, callback, *args):
-		raise NotImplementedError
-
-	def remove_signal_handler(self, sig):
-		raise NotImplementedError
-
-	def set_task_factory(self, factory):
-		raise NotImplementedError
-
-	def get_task_factory(self):
-		raise NotImplementedError
-
-	def get_exception_handler(self):
-		raise NotImplementedError
-
-	def set_exception_handler(self, handler):
-		raise NotImplementedError
-
-	def default_exception_handler(self, context):
-		raise NotImplementedError
-
-	def call_exception_handler(self, context):
-		raise NotImplementedError
-
-	def get_debug(self):
-		raise NotImplementedError
-
-	def set_debug(self, enabled):
-		raise NotImplementedError

diff --git a/lib/portage/util/futures/unix_events.py b/lib/portage/util/futures/unix_events.py
index 9d5445943..030070b1b 100644
--- a/lib/portage/util/futures/unix_events.py
+++ b/lib/portage/util/futures/unix_events.py
@@ -7,7 +7,8 @@ __all__ = (
 )
 
 import asyncio as _real_asyncio
-from asyncio.unix_events import AbstractChildWatcher as _AbstractChildWatcher
+from asyncio import events
+from asyncio.unix_events import AbstractChildWatcher
 
 import fcntl
 import os
@@ -15,7 +16,6 @@ import os
 from portage.util._eventloop.global_event_loop import (
 	global_event_loop as _global_event_loop,
 )
-from portage.util.futures import events
 
 
 class _PortageEventLoop(events.AbstractEventLoop):
@@ -83,27 +83,7 @@ else:
 		fcntl.fcntl(fd, fcntl.F_SETFL, flags)
 
 
-class AbstractChildWatcher(_AbstractChildWatcher):
-	def add_child_handler(self, pid, callback, *args):
-		raise NotImplementedError()
-
-	def remove_child_handler(self, pid):
-		raise NotImplementedError()
-
-	def attach_loop(self, loop):
-		raise NotImplementedError()
-
-	def close(self):
-		raise NotImplementedError()
-
-	def __enter__(self):
-		raise NotImplementedError()
-
-	def __exit__(self, a, b, c):
-		raise NotImplementedError()
-
-
-class _PortageChildWatcher(_AbstractChildWatcher):
+class _PortageChildWatcher(AbstractChildWatcher):
 	def __init__(self, loop):
 		"""
 		@type loop: EventLoop

[gentoo-commits] proj/portage:master commit in: lib/portage/util/futures/

[Zac Medico]

commit:     a8e0ddccf90a4f6dd0b4c0ae0832b4abb1f35b04
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Sun Mar  7 07:29:35 2021 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Sun Mar  7 07:31:11 2021 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=a8e0ddcc

Remove unused _EventLoopFuture class

Signed-off-by: Zac Medico <zmedico <AT> gentoo.org>

 lib/portage/util/futures/futures.py | 156 +-----------------------------------
 1 file changed, 1 insertion(+), 155 deletions(-)

diff --git a/lib/portage/util/futures/futures.py b/lib/portage/util/futures/futures.py
index 839c767a7..3f239890a 100644
--- a/lib/portage/util/futures/futures.py
+++ b/lib/portage/util/futures/futures.py
@@ -1,4 +1,4 @@
-# Copyright 2016-2018 Gentoo Foundation
+# Copyright 2016-2021 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 #
 # For compatibility with python versions which do not have the
@@ -19,157 +19,3 @@ from asyncio import (
 	InvalidStateError,
 	TimeoutError,
 )
-# pylint: enable=redefined-builtin
-
-import portage
-portage.proxy.lazyimport.lazyimport(globals(),
-	'portage.util._eventloop.global_event_loop:global_event_loop@_global_event_loop',
-)
-
-_PENDING = 'PENDING'
-_CANCELLED = 'CANCELLED'
-_FINISHED = 'FINISHED'
-
-class _EventLoopFuture:
-	"""
-	This class provides (a subset of) the asyncio.Future interface, for
-	use with the EventLoop class, because EventLoop is currently
-	missing some of the asyncio.AbstractEventLoop methods that
-	asyncio.Future requires.
-	"""
-
-	# Class variables serving as defaults for instance variables.
-	_state = _PENDING
-	_result = None
-	_exception = None
-	_loop = None
-
-	def __init__(self, loop=None):
-		"""Initialize the future.
-
-		The optional loop argument allows explicitly setting the event
-		loop object used by the future. If it's not provided, the future uses
-		the default event loop.
-		"""
-		if loop is None:
-			self._loop = _global_event_loop()
-		else:
-			self._loop = loop
-		self._callbacks = []
-
-	def cancel(self):
-		"""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.
-		"""
-		if self._state != _PENDING:
-			return False
-		self._state = _CANCELLED
-		self._schedule_callbacks()
-		return True
-
-	def _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 in callbacks:
-			self._loop.call_soon(callback, self)
-
-	def cancelled(self):
-		"""Return True if the future was cancelled."""
-		return self._state == _CANCELLED
-
-	def 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
-
-	def 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:
-			raise CancelledError()
-		if self._state != _FINISHED:
-			raise InvalidStateError('Result is not ready.')
-		if self._exception is not None:
-			raise self._exception
-		return self._result
-
-	def 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:
-			raise CancelledError
-		if self._state != _FINISHED:
-			raise InvalidStateError('Exception is not set.')
-		return self._exception
-
-	def add_done_callback(self, fn):
-		"""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)
-		else:
-			self._callbacks.append(fn)
-
-	def remove_done_callback(self, fn):
-		"""Remove all instances of a callback from the "call when done" list.
-
-		Returns the number of callbacks removed.
-		"""
-		filtered_callbacks = [f for f in self._callbacks if f != fn]
-		removed_count = len(self._callbacks) - len(filtered_callbacks)
-		if removed_count:
-			self._callbacks[:] = filtered_callbacks
-		return removed_count
-
-	def set_result(self, 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('{}: {!r}'.format(self._state, self))
-		self._result = result
-		self._state = _FINISHED
-		self._schedule_callbacks()
-
-	def set_exception(self, 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('{}: {!r}'.format(self._state, self))
-		if isinstance(exception, type):
-			exception = exception()
-		self._exception = exception
-		self._state = _FINISHED
-		self._schedule_callbacks()

[gentoo-commits] proj/portage:master commit in: lib/portage/util/futures/

[Zac Medico]

commit:     8d962cb5cc97a5092ff45446c0f8da55b27d2434
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Mon Jan 18 10:09:05 2021 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Mon Jan 18 11:18:43 2021 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=8d962cb5

coroutine: do not require loop parameter

The loop parameter is not needed since global_event_loop now
returns the running event loop for the current thread.

Bug: https://bugs.gentoo.org/737698
Bug: https://bugs.gentoo.org/763339
Signed-off-by: Zac Medico <zmedico <AT> gentoo.org>

 lib/portage/util/futures/compat_coroutine.py | 6 +-----
 1 file changed, 1 insertion(+), 5 deletions(-)

diff --git a/lib/portage/util/futures/compat_coroutine.py b/lib/portage/util/futures/compat_coroutine.py
index 9a0c5c1c8..3e8dcec02 100644
--- a/lib/portage/util/futures/compat_coroutine.py
+++ b/lib/portage/util/futures/compat_coroutine.py
@@ -1,4 +1,4 @@
-# Copyright 2018 Gentoo Foundation
+# Copyright 2018-2021 Gentoo Foundation
 # Distributed under the terms of the GNU General Public License v2
 
 import functools
@@ -68,10 +68,6 @@ def _generator_future(generator_func, *args, **kwargs):
 	the default event loop.
 	"""
 	loop = kwargs.get('loop')
-	if loop is None and portage._internal_caller:
-		# Require an explicit loop parameter, in order to support
-		# local event loops (bug 737698).
-		raise AssertionError("Missing required argument 'loop'")
 	loop = asyncio._wrap_loop(loop)
 	result = loop.create_future()
 	_GeneratorTask(generator_func(*args, **kwargs), result, loop=loop)

[gentoo-commits] proj/portage:master commit in: lib/portage/util/futures/

[Zac Medico]

commit:     7efa7ecfe07737239be593b2c32e497cc1d2f154
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Mon Jan  4 08:54:52 2021 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Mon Jan  4 08:57:09 2021 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=7efa7ecf

_Retry: Use ensure_future for self._current_task

Use ensure_future for compatibility with PEP 492 coroutines
with async and await syntax.

Signed-off-by: Zac Medico <zmedico <AT> gentoo.org>

 lib/portage/util/futures/retry.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/portage/util/futures/retry.py b/lib/portage/util/futures/retry.py
index 4092f60d6..31cc161da 100644
--- a/lib/portage/util/futures/retry.py
+++ b/lib/portage/util/futures/retry.py
@@ -1,4 +1,4 @@
-# Copyright 2018 Gentoo Foundation
+# Copyright 2018-2021 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
 __all__ = (
@@ -113,7 +113,7 @@ class _Retry:
 
 	def _begin_try(self):
 		self._tries += 1
-		self._current_task = self._func()
+		self._current_task = asyncio.ensure_future(self._func(), loop=self._loop)
 		self._current_task.add_done_callback(self._try_done)
 		if self._try_timeout is not None:
 			self._try_timeout_handle = self._loop.call_later(

[gentoo-commits] proj/portage:master commit in: lib/portage/util/futures/

[Zac Medico]

commit:     b0ed587308eb3cbfafe9abcb1c59f24f48b97cdc
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Mon Aug 24 03:06:03 2020 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Mon Aug 24 03:21:10 2020 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=b0ed5873

async_iter_completed: fix InvalidStateError in finally clause (bug 738766)

Do not attempt to wait for the TaskScheduler instance in the finally
clause, since it will always raise InvalidStateError if its status
is not available yet (which is normal if it has remaining tasks with
done callbacks that have not been scheduled yet).

Bug: https://bugs.gentoo.org/738766
Signed-off-by: Zac Medico <zmedico <AT> gentoo.org>

 lib/portage/util/futures/iter_completed.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/lib/portage/util/futures/iter_completed.py b/lib/portage/util/futures/iter_completed.py
index 3c1a58e7e..43672171b 100644
--- a/lib/portage/util/futures/iter_completed.py
+++ b/lib/portage/util/futures/iter_completed.py
@@ -109,7 +109,6 @@ def async_iter_completed(futures, max_jobs=None, max_load=None, loop=None):
 		# cleanup in case of interruption by SIGINT, etc
 		if not loop.is_closed():
 			scheduler.cancel()
-			scheduler.wait()
 
 
 def iter_gather(futures, max_jobs=None, max_load=None, loop=None):

[gentoo-commits] proj/portage:master commit in: lib/portage/util/futures/

[Zac Medico]

commit:     4db210cf165f426f695ecf8307942a2610b352fc
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Mon Aug 17 04:20:22 2020 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Mon Aug 17 04:23:49 2020 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=4db210cf

async_iter_completed: check if loop is closed in finally clause

This avoids RuntimeError('Event loop is closed') errors triggered
by SIGINT.

Signed-off-by: Zac Medico <zmedico <AT> gentoo.org>

 lib/portage/util/futures/iter_completed.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/lib/portage/util/futures/iter_completed.py b/lib/portage/util/futures/iter_completed.py
index 9554b4338..3c1a58e7e 100644
--- a/lib/portage/util/futures/iter_completed.py
+++ b/lib/portage/util/futures/iter_completed.py
@@ -107,8 +107,9 @@ def async_iter_completed(futures, max_jobs=None, max_load=None, loop=None):
 			yield future_done_set
 	finally:
 		# cleanup in case of interruption by SIGINT, etc
-		scheduler.cancel()
-		scheduler.wait()
+		if not loop.is_closed():
+			scheduler.cancel()
+			scheduler.wait()
 
 
 def iter_gather(futures, max_jobs=None, max_load=None, loop=None):

[gentoo-commits] proj/portage:master commit in: lib/portage/util/futures/

[Robin H. Johnson]

commit:     bde66462659a1418f0d88e71e0f61b741e2c109a
Author:     Robin H. Johnson <robbat2 <AT> gentoo <DOT> org>
AuthorDate: Sat Feb 16 17:36:51 2019 +0000
Commit:     Robin H. Johnson <robbat2 <AT> gentoo <DOT> org>
CommitDate: Sat Feb 16 17:36:51 2019 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=bde66462

util: fix thinko introduced in cleanup of get_cpu_count work

Signed-off-by: Robin H. Johnson <robbat2 <AT> gentoo.org>

 lib/portage/util/futures/iter_completed.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/portage/util/futures/iter_completed.py b/lib/portage/util/futures/iter_completed.py
index 4c48ea0fe..9554b4338 100644
--- a/lib/portage/util/futures/iter_completed.py
+++ b/lib/portage/util/futures/iter_completed.py
@@ -61,8 +61,8 @@ def async_iter_completed(futures, max_jobs=None, max_load=None, loop=None):
 	"""
 	loop = asyncio._wrap_loop(loop)
 
-	max_jobs = max_jobs or portage.util.cpuinfo.get_cpu_count()
-	max_load = max_load or portage.util.cpuinfo.get_cpu_count()
+	max_jobs = max_jobs or get_cpu_count()
+	max_load = max_load or get_cpu_count()
 
 	future_map = {}
 	def task_generator():

[gentoo-commits] proj/portage:master commit in: lib/portage/util/futures/

[Zac Medico]

commit:     74d2509c99fbcb43e018ead4950b938e41e524e5
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Tue Nov 20 10:06:57 2018 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Tue Nov 20 10:24:12 2018 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=74d2509c

compat_corouting._GeneratorTask: save throw return (bug 671472)

According to PEP 342, the generator.throw() method returns a value if
the exception is caught. The return value must be sent to the generator
in order fufill the generator protocol. This is relevant in the
portdbapi.async_xmatch() method, since it catches an exception thrown
with the generator.throw() method.

Bug: https://bugs.gentoo.org/671472
Signed-off-by: Zac Medico <zmedico <AT> gentoo.org>

 lib/portage/util/futures/compat_coroutine.py | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/lib/portage/util/futures/compat_coroutine.py b/lib/portage/util/futures/compat_coroutine.py
index b5ff92faf..b745fd845 100644
--- a/lib/portage/util/futures/compat_coroutine.py
+++ b/lib/portage/util/futures/compat_coroutine.py
@@ -106,13 +106,11 @@ class _GeneratorTask(object):
 			if previous is None:
 				future = next(self._generator)
 			elif previous.cancelled():
-				self._generator.throw(asyncio.CancelledError())
-				future = next(self._generator)
+				future = self._generator.throw(asyncio.CancelledError())
 			elif previous.exception() is None:
 				future = self._generator.send(previous.result())
 			else:
-				self._generator.throw(previous.exception())
-				future = next(self._generator)
+				future = self._generator.throw(previous.exception())
 
 		except asyncio.CancelledError:
 			self._result.cancel()