def _do_test(self, read_end, write_end):
		initial_policy = asyncio.get_event_loop_policy()
		if not isinstance(initial_policy, DefaultEventLoopPolicy):
			asyncio.set_event_loop_policy(DefaultEventLoopPolicy())

		loop = asyncio._wrap_loop()
		read_end = os.fdopen(read_end, 'rb', 0)
		write_end = os.fdopen(write_end, 'wb', 0)
		try:
			def reader_callback():
				if not reader_callback.called.done():
					reader_callback.called.set_result(None)

			reader_callback.called = loop.create_future()
			loop.add_reader(read_end.fileno(), reader_callback)

			# Allow the loop to check for IO events, and assert
			# that our future is still not done.
			loop.run_until_complete(asyncio.sleep(0, loop=loop))
			self.assertFalse(reader_callback.called.done())

			# Demonstrate that the callback is called afer the
			# other end of the pipe has been closed.
			write_end.close()
			loop.run_until_complete(reader_callback.called)
		finally:
			loop.remove_reader(read_end.fileno())
			write_end.close()
			read_end.close()
			asyncio.set_event_loop_policy(initial_policy)
			if loop not in (None, global_event_loop()):
				loop.close()
				self.assertFalse(global_event_loop().is_closed())

	def _run_test(self, test):
		initial_policy = asyncio.get_event_loop_policy()
		if not isinstance(initial_policy, DefaultEventLoopPolicy):
			asyncio.set_event_loop_policy(DefaultEventLoopPolicy())

		loop = asyncio._wrap_loop()
		try:
			test(loop)
		finally:
			asyncio.set_event_loop_policy(initial_policy)
			if loop not in (None, global_event_loop()):
				loop.close()
				self.assertFalse(global_event_loop().is_closed())

	def _testAsynchronousLock(self):
		scheduler = global_event_loop()
		tempdir = tempfile.mkdtemp()
		try:
			path = os.path.join(tempdir, 'lock_me')
			for force_async, async_unlock in itertools.product(
				(True, False), repeat=2):
				for force_dummy in (True, False):
					async_lock = AsynchronousLock(path=path,
						scheduler=scheduler, _force_async=force_async,
						_force_thread=True,
						_force_dummy=force_dummy)
					async_lock.start()
					self.assertEqual(async_lock.wait(), os.EX_OK)
					self.assertEqual(async_lock.returncode, os.EX_OK)
					if async_unlock:
						scheduler.run_until_complete(async_lock.async_unlock())
					else:
						async_lock.unlock()

				async_lock = AsynchronousLock(path=path,
					scheduler=scheduler, _force_async=force_async,
					_force_process=True)
				async_lock.start()
				self.assertEqual(async_lock.wait(), os.EX_OK)
				self.assertEqual(async_lock.returncode, os.EX_OK)
				if async_unlock:
					scheduler.run_until_complete(async_lock.async_unlock())
				else:
					async_lock.unlock()
		finally:
			shutil.rmtree(tempdir)

	def test_socks5_proxy(self):

		loop = global_event_loop()

		host = '127.0.0.1'
		content = b'Hello World!'
		path = '/index.html'
		proxy = None
		tempdir = tempfile.mkdtemp()

		try:
			with AsyncHTTPServer(host, {path: content}, loop) as server:

				settings = {
					'PORTAGE_TMPDIR': tempdir,
					'PORTAGE_BIN_PATH': PORTAGE_BIN_PATH,
				}

				try:
					proxy = socks5.get_socks5_proxy(settings)
				except NotImplementedError:
					# bug 658172 for python2.7
					self.skipTest('get_socks5_proxy not implemented for {} {}.{}'.format(
						platform.python_implementation(), *sys.version_info[:2]))
				else:
					loop.run_until_complete(socks5.proxy.ready())

					result = loop.run_until_complete(loop.run_in_executor(None,
						self._fetch_via_proxy, proxy, host, server.server_port, path))

					self.assertEqual(result, content)
		finally:
			socks5.proxy.stop()
			shutil.rmtree(tempdir)

	def testAsyncCancel(self):

		loop = global_event_loop()
		input_futures = set()
		future_count = 3

		def future_generator():
			for i in range(future_count):
				future = loop.create_future()
				loop.call_soon(lambda future: None if future.done()
					else future.set_result(None), future)
				input_futures.add(future)
				yield future

		for future_done_set in async_iter_completed(future_generator(),
			max_jobs=True, max_load=True, loop=loop):
			future_done_set.cancel()
			break

		# With max_jobs=True, async_iter_completed should have executed
		# the generator until it raised StopIteration.
		self.assertEqual(future_count, len(input_futures))

		loop.run_until_complete(asyncio.wait(input_futures, loop=loop))

		# The futures may have results or they may have been cancelled
		# by TaskScheduler, and behavior varies depending on the python
		# interpreter.
		for future in input_futures:
			future.cancelled() or future.result()

	def _testPipeReader(self, master_fd, slave_fd, test_string):
		"""
		Use a poll loop to read data from a pipe and assert that
		the data written to the pipe is identical to the data
		read from the pipe.
		"""

		# WARNING: It is very important to use unbuffered mode here,
		# in order to avoid issue 5380 with python3.
		master_file = os.fdopen(master_fd, 'rb', 0)
		scheduler = global_event_loop()

		consumer = PipeReader(
			input_files={"producer" : master_file},
			_use_array=self._use_array,
			scheduler=scheduler)

		producer = PopenProcess(
			pipe_reader=consumer,
			proc=subprocess.Popen(["bash", "-c", self._echo_cmd % test_string],
				stdout=slave_fd),
			scheduler=scheduler)

		producer.start()
		os.close(slave_fd)
		producer.wait()
		consumer.wait()

		self.assertEqual(producer.returncode, os.EX_OK)
		self.assertEqual(consumer.returncode, os.EX_OK)

		return consumer.getvalue().decode('ascii', 'replace')

	def _testAsynchronousLock(self):
		scheduler = global_event_loop()
		tempdir = tempfile.mkdtemp()
		try:
			path = os.path.join(tempdir, 'lock_me')
			for force_async in (True, False):
				for force_dummy in (True, False):
					async_lock = AsynchronousLock(path=path,
						scheduler=scheduler, _force_async=force_async,
						_force_thread=True,
						_force_dummy=force_dummy)
					async_lock.start()
					self.assertEqual(async_lock.wait(), os.EX_OK)
					self.assertEqual(async_lock.returncode, os.EX_OK)
					async_lock.unlock()

				async_lock = AsynchronousLock(path=path,
					scheduler=scheduler, _force_async=force_async,
					_force_process=True)
				async_lock.start()
				self.assertEqual(async_lock.wait(), os.EX_OK)
				self.assertEqual(async_lock.returncode, os.EX_OK)
				async_lock.unlock()

		finally:
			shutil.rmtree(tempdir)

	def _testPipeLogger(self, test_string):

		producer = PopenProcess(proc=subprocess.Popen(
			["bash", "-c", self._echo_cmd % test_string],
			stdout=subprocess.PIPE, stderr=subprocess.STDOUT),
			scheduler=global_event_loop())

		fd, log_file_path = tempfile.mkstemp()
		try:

			consumer = PipeLogger(background=True,
				input_fd=os.dup(producer.proc.stdout.fileno()),
				log_file_path=log_file_path)

			# Close the stdout pipe, since we duplicated it, and it
			# must be closed in order to avoid a ResourceWarning.
			producer.proc.stdout.close()
			producer.pipe_reader = consumer

			producer.start()
			producer.wait()

			self.assertEqual(producer.returncode, os.EX_OK)
			self.assertEqual(consumer.returncode, os.EX_OK)

			with open(log_file_path, 'rb') as f:
				content = f.read()

		finally:
			os.close(fd)
			os.unlink(log_file_path)

		return content.decode('ascii', 'replace')

	def _testAsynchronousLockWaitKill(self):
		scheduler = global_event_loop()
		tempdir = tempfile.mkdtemp()
		try:
			path = os.path.join(tempdir, 'lock_me')
			lock1 = AsynchronousLock(path=path, scheduler=scheduler)
			lock1.start()
			self.assertEqual(lock1.wait(), os.EX_OK)
			self.assertEqual(lock1.returncode, os.EX_OK)
			lock2 = AsynchronousLock(path=path, scheduler=scheduler,
				_force_async=True, _force_process=True)
			lock2.start()
			# lock2 should be waiting for lock1 to release
			self.assertEqual(lock2.poll(), None)
			self.assertEqual(lock2.returncode, None)

			# Kill lock2's process and then check wait() and
			# returncode results. This is intended to simulate
			# a SIGINT sent via the controlling tty.
			self.assertEqual(lock2._imp is not None, True)
			self.assertEqual(lock2._imp._proc is not None, True)
			self.assertEqual(lock2._imp._proc.pid is not None, True)
			lock2._imp._kill_test = True
			os.kill(lock2._imp._proc.pid, signal.SIGTERM)
			self.assertEqual(lock2.wait() == os.EX_OK, False)
			self.assertEqual(lock2.returncode == os.EX_OK, False)
			self.assertEqual(lock2.returncode is None, False)
			lock1.unlock()
		finally:
			shutil.rmtree(tempdir)

	def testIterCompleted(self):

		# Mark this as todo, since we don't want to fail if heavy system
		# load causes the tasks to finish in an unexpected order.
		self.todo = True

		loop = global_event_loop()
		tasks = [
			SleepProcess(seconds=0.200),
			SleepProcess(seconds=0.100),
			SleepProcess(seconds=0.001),
		]

		expected_order = sorted(task.seconds for task in tasks)

		def future_generator():
			for task in tasks:
				task.future = loop.create_future()
				task.scheduler = loop
				task.start()
				yield task.future

		for seconds, future in zip(expected_order, iter_completed(future_generator(),
			max_jobs=True, max_load=None, loop=loop)):
			self.assertEqual(seconds, future.result())

	def _sync(self, selected_repos, return_messages,
		emaint_opts=None):

		if emaint_opts is not None:
			for k, v in emaint_opts.items():
				if v is not None:
					k = "--" + k.replace("_", "-")
					self.emerge_config.opts[k] = v

		selected_repos = [repo for repo in selected_repos if repo.sync_type is not None]
		msgs = []
		if not selected_repos:
			msgs.append("Emaint sync, nothing to sync... returning")
			if return_messages:
				msgs.extend(self.rmessage([('None', os.EX_OK)], 'sync'))
				return msgs
			return
		# Portage needs to ensure a sane umask for the files it creates.
		os.umask(0o22)

		sync_manager = SyncManager(
			self.emerge_config.target_config.settings, emergelog)

		max_jobs = (self.emerge_config.opts.get('--jobs', 1)
			if 'parallel-fetch' in self.emerge_config.
			target_config.settings.features else 1)
		sync_scheduler = SyncScheduler(emerge_config=self.emerge_config,
			selected_repos=selected_repos, sync_manager=sync_manager,
			max_jobs=max_jobs,
			event_loop=global_event_loop() if portage._internal_caller else
				EventLoop(main=False))

		sync_scheduler.start()
		sync_scheduler.wait()
		retvals = sync_scheduler.retvals
		msgs.extend(sync_scheduler.msgs)

		if retvals:
			msgs.extend(self.rmessage(retvals, 'sync'))
		else:
			msgs.extend(self.rmessage([('None', os.EX_OK)], 'sync'))

		# run the post_sync_hook one last time for
		# run only at sync completion hooks
		if sync_scheduler.global_hooks_enabled:
			rcode = sync_manager.perform_post_sync_hook('')
			if rcode:
				msgs.extend(self.rmessage([('None', rcode)], 'post-sync'))

		# Reload the whole config.
		portage._sync_mode = False
		self._reload_config()
		self._do_pkg_moves()
		msgs.extend(self._check_updates())
		display_news_notification(self.emerge_config.target_config,
			self.emerge_config.opts)

		if return_messages:
			return msgs
		return

	def testFutureDoneCallback(self):

		event_loop = global_event_loop()

		def done_callback(finished):
			done_callback_called.set_result(True)

		done_callback_called = event_loop.create_future()
		finished = event_loop.create_future()
		finished.add_done_callback(done_callback)
		event_loop.call_soon(finished.set_result, True)
		event_loop.run_until_complete(done_callback_called)

		def done_callback2(finished):
			done_callback2_called.set_result(True)

		done_callback_called = event_loop.create_future()
		done_callback2_called = event_loop.create_future()
		finished = event_loop.create_future()
		finished.add_done_callback(done_callback)
		finished.add_done_callback(done_callback2)
		finished.remove_done_callback(done_callback)
		event_loop.call_soon(finished.set_result, True)
		event_loop.run_until_complete(done_callback2_called)

		self.assertFalse(done_callback_called.done())

	def _testAsynchronousLockWait(self):
		scheduler = global_event_loop()
		tempdir = tempfile.mkdtemp()
		try:
			path = os.path.join(tempdir, 'lock_me')
			lock1 = AsynchronousLock(path=path, scheduler=scheduler)
			lock1.start()
			self.assertEqual(lock1.wait(), os.EX_OK)
			self.assertEqual(lock1.returncode, os.EX_OK)

			# lock2 requires _force_async=True since the portage.locks
			# module is not designed to work as intended here if the
			# same process tries to lock the same file more than
			# one time concurrently.
			lock2 = AsynchronousLock(path=path, scheduler=scheduler,
				_force_async=True, _force_process=True)
			lock2.start()
			# lock2 should be waiting for lock1 to release
			self.assertEqual(lock2.poll(), None)
			self.assertEqual(lock2.returncode, None)

			lock1.unlock()
			self.assertEqual(lock2.wait(), os.EX_OK)
			self.assertEqual(lock2.returncode, os.EX_OK)
			lock2.unlock()
		finally:
			shutil.rmtree(tempdir)

	def _testPipeLogger(self, test_string):

		producer = PopenProcess(proc=subprocess.Popen(
			["bash", "-c", self._echo_cmd % test_string],
			stdout=subprocess.PIPE, stderr=subprocess.STDOUT),
			scheduler=global_event_loop())

		fd, log_file_path = tempfile.mkstemp()
		try:

			consumer = PipeLogger(background=True,
				input_fd=producer.proc.stdout,
				log_file_path=log_file_path)

			producer.pipe_reader = consumer

			producer.start()
			producer.wait()

			self.assertEqual(producer.returncode, os.EX_OK)
			self.assertEqual(consumer.returncode, os.EX_OK)

			with open(log_file_path, 'rb') as f:
				content = f.read()

		finally:
			os.close(fd)
			os.unlink(log_file_path)

		return content.decode('ascii', 'replace')

	def __init__(self):
		self._terminated = threading.Event()
		self._terminated_tasks = False
		self._max_jobs = 1
		self._max_load = None
		self._jobs = 0
		self._scheduling = False
		self._background = False
		self._event_loop = global_event_loop()
		self.sched_iface = self._sched_iface_class(
			IO_ERR=self._event_loop.IO_ERR,
			IO_HUP=self._event_loop.IO_HUP,
			IO_IN=self._event_loop.IO_IN,
			IO_NVAL=self._event_loop.IO_NVAL,
			IO_OUT=self._event_loop.IO_OUT,
			IO_PRI=self._event_loop.IO_PRI,
			child_watch_add=self._event_loop.child_watch_add,
			idle_add=self._event_loop.idle_add,
			io_add_watch=self._event_loop.io_add_watch,
			iteration=self._event_loop.iteration,
			output=self._task_output,
			register=self._event_loop.io_add_watch,
			source_remove=self._event_loop.source_remove,
			timeout_add=self._event_loop.timeout_add,
			unregister=self._event_loop.source_remove)

	def __init__(self, max_workers=None, loop=None):
		self._max_workers = max_workers or multiprocessing.cpu_count()
		self._loop = loop or global_event_loop()
		self._submit_queue = collections.deque()
		self._running_tasks = {}
		self._shutdown = False
		self._shutdown_future = self._loop.create_future()

    def _communicate(self, args):

        if not self._daemon_is_alive():
            self._no_daemon_msg()
            return 2

            # Open the input fifo before the output fifo, in order to make it
            # possible for the daemon to send a reply without blocking. This
            # improves performance, and also makes it possible for the daemon
            # to do a non-blocking write without a race condition.
        input_fd = os.open(self.ipc_out_fifo, os.O_RDONLY | os.O_NONBLOCK)

        # Use forks so that the child process can handle blocking IO
        # un-interrupted, while the parent handles all timeout
        # considerations. This helps to avoid possible race conditions
        # from interference between timeouts and blocking IO operations.
        msg = portage.localization._("during write")
        retval = self._run_writer(
            FifoWriter(buf=pickle.dumps(args), fifo=self.ipc_in_fifo, scheduler=global_event_loop()), msg
        )

        if retval != os.EX_OK:
            portage.util.writemsg_level(
                "ebuild-ipc: %s: %s\n" % (msg, portage.localization._("subprocess failure: %s") % retval),
                level=logging.ERROR,
                noiselevel=-1,
            )
            return retval

        if not self._daemon_is_alive():
            self._no_daemon_msg()
            return 2

        return self._receive_reply(input_fd)

	def __init__(self, main=False):
		"""
		@param main: If True then use global_event_loop(), otherwise use
			a local EventLoop instance (default is False, for safe use in
			a non-main thread)
		@type main: bool
		"""
		self._terminated = threading.Event()
		self._terminated_tasks = False
		self._max_jobs = 1
		self._max_load = None
		self._jobs = 0
		self._scheduling = False
		self._background = False
		if main:
			self._event_loop = global_event_loop()
		else:
			self._event_loop = EventLoop(main=False)
		self.sched_iface = self._sched_iface_class(
			IO_ERR=self._event_loop.IO_ERR,
			IO_HUP=self._event_loop.IO_HUP,
			IO_IN=self._event_loop.IO_IN,
			IO_NVAL=self._event_loop.IO_NVAL,
			IO_OUT=self._event_loop.IO_OUT,
			IO_PRI=self._event_loop.IO_PRI,
			child_watch_add=self._event_loop.child_watch_add,
			idle_add=self._event_loop.idle_add,
			io_add_watch=self._event_loop.io_add_watch,
			iteration=self._event_loop.iteration,
			output=self._task_output,
			register=self._event_loop.io_add_watch,
			source_remove=self._event_loop.source_remove,
			timeout_add=self._event_loop.timeout_add,
			unregister=self._event_loop.source_remove)

	def testSucceedLater(self):
		loop = global_event_loop()
		func_coroutine = self._wrap_coroutine_func(SucceedLater(1))
		decorator = retry(try_max=9999,
			delay_func=RandomExponentialBackoff(multiplier=0.1, base=2))
		decorated_func = decorator(func_coroutine, loop=loop)
		result = loop.run_until_complete(decorated_func())
		self.assertEqual(result, 'success')

	def _receive_reply(self, input_fd):

		start_time = time.time()

		pipe_reader = PipeReader(input_files={"input_fd":input_fd},
			scheduler=global_event_loop())
		pipe_reader.start()

		eof = pipe_reader.poll() is not None

		while not eof:
			pipe_reader._wait_loop(timeout=self._COMMUNICATE_RETRY_TIMEOUT)
			eof = pipe_reader.poll() is not None
			if not eof:
				if self._daemon_is_alive():
					self._timeout_retry_msg(start_time,
						portage.localization._('during read'))
				else:
					pipe_reader.cancel()
					self._no_daemon_msg()
					return 2

		buf = pipe_reader.getvalue()

		retval = 2

		if not buf:

			portage.util.writemsg_level(
				"ebuild-ipc: %s\n" % \
				(portage.localization._('read failed'),),
				level=logging.ERROR, noiselevel=-1)

		else:

			try:
				reply = pickle.loads(buf)
			except SystemExit:
				raise
			except Exception as e:
				# The pickle module can raise practically
				# any exception when given corrupt data.
				portage.util.writemsg_level(
					"ebuild-ipc: %s\n" % (e,),
					level=logging.ERROR, noiselevel=-1)

			else:

				(out, err, retval) = reply

				if out:
					portage.util.writemsg_stdout(out, noiselevel=-1)

				if err:
					portage.util.writemsg(err, noiselevel=-1)

		return retval