commit: d66e9ec0b10522528d62e18b83e012c1ec121787 Author: Zac Medico <zmedico <AT> gentoo <DOT> org> AuthorDate: Mon Feb 17 18:32:06 2020 +0000 Commit: Zac Medico <zmedico <AT> gentoo <DOT> org> CommitDate: Mon Feb 17 22:29:09 2020 +0000 URL: https://gitweb.gentoo.org/proj/portage.git/commit/?id=d66e9ec0
AsynchronousTask: add coroutine async_start method Add coroutine async_start coroutine method which calls an _async_start template method. Eventually, subclasses having _start implementations that need to write to a build log will be required to implement an _async_start coroutine method to replace the _start method as discussed in bug 709746. Bug: https://bugs.gentoo.org/709746 Signed-off-by: Zac Medico <zmedico <AT> gentoo.org> lib/_emerge/AsynchronousTask.py | 22 +++++++++++++++++----- lib/_emerge/CompositeTask.py | 17 ++++++++++++++--- lib/_emerge/TaskSequence.py | 7 ++++++- lib/portage/tests/ebuild/test_doebuild_fd_pipes.py | 6 ++++-- lib/portage/tests/ebuild/test_doebuild_spawn.py | 6 +++--- lib/portage/tests/ebuild/test_fetch.py | 9 ++++++--- lib/portage/tests/ebuild/test_ipc_daemon.py | 4 ++-- lib/portage/tests/ebuild/test_spawn.py | 4 ++-- .../test_lazy_import_portage_baseline.py | 6 +++--- lib/portage/tests/locks/test_asynchronous_lock.py | 18 +++++++++--------- lib/portage/tests/process/test_PopenProcess.py | 6 +++--- .../tests/process/test_PopenProcessBlockingIO.py | 4 ++-- lib/portage/tests/process/test_poll.py | 4 ++-- .../tests/util/futures/test_iter_completed.py | 16 +++++++++++----- lib/portage/tests/util/test_file_copier.py | 6 +++--- 15 files changed, 87 insertions(+), 48 deletions(-) diff --git a/lib/_emerge/AsynchronousTask.py b/lib/_emerge/AsynchronousTask.py index cf6e6dc44..280ed16da 100644 --- a/lib/_emerge/AsynchronousTask.py +++ b/lib/_emerge/AsynchronousTask.py @@ -1,10 +1,11 @@ -# Copyright 1999-2018 Gentoo Foundation +# Copyright 1999-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import signal from portage import os from portage.util.futures import asyncio +from portage.util.futures.compat_coroutine import coroutine, coroutine_return from portage.util.SlotObject import SlotObject class AsynchronousTask(SlotObject): @@ -22,6 +23,17 @@ class AsynchronousTask(SlotObject): _cancelled_returncode = - signal.SIGINT + @coroutine + def async_start(self): + self._start_hook() + yield self._async_start() + + @coroutine + def _async_start(self): + self._start() + coroutine_return() + yield None + def start(self): """ Start an asynchronous task and then return as soon as possible. @@ -29,6 +41,10 @@ class AsynchronousTask(SlotObject): self._start_hook() self._start() + def _start(self): + self.returncode = os.EX_OK + self._async_wait() + def async_wait(self): """ Wait for returncode asynchronously. Notification is available @@ -49,10 +65,6 @@ class AsynchronousTask(SlotObject): self._async_wait() return waiter - def _start(self): - self.returncode = os.EX_OK - self._async_wait() - def isAlive(self): return self.returncode is None diff --git a/lib/_emerge/CompositeTask.py b/lib/_emerge/CompositeTask.py index 1edec4a17..ce7136c3d 100644 --- a/lib/_emerge/CompositeTask.py +++ b/lib/_emerge/CompositeTask.py @@ -1,8 +1,10 @@ -# Copyright 1999-2018 Gentoo Foundation +# Copyright 1999-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 from _emerge.AsynchronousTask import AsynchronousTask from portage import os +from portage.util.futures import asyncio + class CompositeTask(AsynchronousTask): @@ -97,7 +99,7 @@ class CompositeTask(AsynchronousTask): def _start_task(self, task, exit_handler): """ Register exit handler for the given task, set it - as self._current_task, and call task.start(). + as self._current_task, and call task.async_start(). Subclasses can use this as a generic way to start a task. @@ -109,7 +111,16 @@ class CompositeTask(AsynchronousTask): pass task.addExitListener(exit_handler) self._current_task = task - task.start() + result = asyncio.ensure_future(task.async_start(), loop=self.scheduler) + result.add_done_callback(self._current_task_start_cb) + + def _current_task_start_cb(self, future): + try: + future.result() + except asyncio.CancelledError: + self.cancelled = True + self._was_cancelled() + self._async_wait() def _task_queued(self, task): task.addStartListener(self._task_queued_start_handler) diff --git a/lib/_emerge/TaskSequence.py b/lib/_emerge/TaskSequence.py index 1f2ba94c2..59ee5686d 100644 --- a/lib/_emerge/TaskSequence.py +++ b/lib/_emerge/TaskSequence.py @@ -1,4 +1,4 @@ -# Copyright 1999-2014 Gentoo Foundation +# Copyright 1999-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import sys @@ -42,6 +42,11 @@ class TaskSequence(CompositeTask): self._start_task(task, self._task_exit_handler) + def _current_task_start_cb(self, future): + CompositeTask._current_task_start_cb(self, future) + if self.cancelled: + self._task_queue.clear() + def _task_exit_handler(self, task): if self._default_exit(task) != os.EX_OK: self.wait() diff --git a/lib/portage/tests/ebuild/test_doebuild_fd_pipes.py b/lib/portage/tests/ebuild/test_doebuild_fd_pipes.py index b89421822..05ea24c4b 100644 --- a/lib/portage/tests/ebuild/test_doebuild_fd_pipes.py +++ b/lib/portage/tests/ebuild/test_doebuild_fd_pipes.py @@ -1,4 +1,4 @@ -# Copyright 2013-2016 Gentoo Foundation +# Copyright 2013-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import portage @@ -8,6 +8,7 @@ from portage.tests.resolver.ResolverPlayground import ResolverPlayground from portage.package.ebuild._ipc.QueryCommand import QueryCommand from portage.util._async.ForkProcess import ForkProcess from portage.util._async.TaskScheduler import TaskScheduler +from portage.util.futures import asyncio from _emerge.Package import Package from _emerge.PipeReader import PipeReader @@ -54,6 +55,7 @@ class DoebuildFdPipesTestCase(TestCase): self.assertEqual(true_binary is None, False, "true command not found") + loop = asyncio._wrap_loop() dev_null = open(os.devnull, 'wb') playground = ResolverPlayground(ebuilds=ebuilds) try: @@ -115,7 +117,7 @@ class DoebuildFdPipesTestCase(TestCase): max_jobs=2) try: - task_scheduler.start() + loop.run_until_complete(task_scheduler.async_start()) finally: # PipeReader closes pr os.close(pw) diff --git a/lib/portage/tests/ebuild/test_doebuild_spawn.py b/lib/portage/tests/ebuild/test_doebuild_spawn.py index 6b344658f..7750f87b3 100644 --- a/lib/portage/tests/ebuild/test_doebuild_spawn.py +++ b/lib/portage/tests/ebuild/test_doebuild_spawn.py @@ -1,4 +1,4 @@ -# Copyright 2010-2015 Gentoo Foundation +# Copyright 2010-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import textwrap @@ -90,14 +90,14 @@ class DoebuildSpawnTestCase(TestCase): ebuild_phase = EbuildPhase(background=False, phase=phase, scheduler=scheduler, settings=settings) - ebuild_phase.start() + scheduler.run_until_complete(ebuild_phase.async_start()) ebuild_phase.wait() self.assertEqual(ebuild_phase.returncode, os.EX_OK) ebuild_phase = MiscFunctionsProcess(background=False, commands=['success_hooks'], scheduler=scheduler, settings=settings) - ebuild_phase.start() + scheduler.run_until_complete(ebuild_phase.async_start()) ebuild_phase.wait() self.assertEqual(ebuild_phase.returncode, os.EX_OK) diff --git a/lib/portage/tests/ebuild/test_fetch.py b/lib/portage/tests/ebuild/test_fetch.py index 9a8a4a544..0c45230cc 100644 --- a/lib/portage/tests/ebuild/test_fetch.py +++ b/lib/portage/tests/ebuild/test_fetch.py @@ -1,4 +1,4 @@ -# Copyright 2019 Gentoo Authors +# Copyright 2019-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 from __future__ import unicode_literals @@ -15,6 +15,7 @@ from portage.tests.resolver.ResolverPlayground import ResolverPlayground from portage.tests.util.test_socks5 import AsyncHTTPServer from portage.util.configparser import ConfigParserError from portage.util.futures import asyncio +from portage.util.futures.compat_coroutine import coroutine, coroutine_return from portage.util.futures.executor.fork import ForkExecutor from portage.util._async.SchedulerInterface import SchedulerInterface from portage.util._eventloop.global_event_loop import global_event_loop @@ -193,11 +194,13 @@ class EbuildFetchTestCase(TestCase): def deallocate(settings): pass + @coroutine def async_fetch(pkg, ebuild_path): fetcher = EbuildFetcher(config_pool=config_pool, ebuild_path=ebuild_path, fetchonly=False, fetchall=True, pkg=pkg, scheduler=loop) - fetcher.start() - return fetcher.async_wait() + yield fetcher.async_start() + result = yield fetcher.async_wait() + coroutine_return(result) for cpv in ebuilds: metadata = dict(zip(Package.metadata_keys, diff --git a/lib/portage/tests/ebuild/test_ipc_daemon.py b/lib/portage/tests/ebuild/test_ipc_daemon.py index e6da51a76..6b6a12ac9 100644 --- a/lib/portage/tests/ebuild/test_ipc_daemon.py +++ b/lib/portage/tests/ebuild/test_ipc_daemon.py @@ -1,4 +1,4 @@ -# Copyright 2010-2016 Gentoo Foundation +# Copyright 2010-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import tempfile @@ -155,7 +155,7 @@ class IpcDaemonTestCase(TestCase): task_scheduler.addExitListener(self._exit_callback) try: - task_scheduler.start() + event_loop.run_until_complete(task_scheduler.async_start()) event_loop.run_until_complete(self._run_done) event_loop.run_until_complete(task_scheduler.async_wait()) finally: diff --git a/lib/portage/tests/ebuild/test_spawn.py b/lib/portage/tests/ebuild/test_spawn.py index a38e10972..89d58f98c 100644 --- a/lib/portage/tests/ebuild/test_spawn.py +++ b/lib/portage/tests/ebuild/test_spawn.py @@ -1,4 +1,4 @@ -# Copyright 1998-2013 Gentoo Foundation +# Copyright 1998-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import errno @@ -34,7 +34,7 @@ class SpawnTestCase(TestCase): }, scheduler=global_event_loop(), logfile=logfile) - proc.start() + global_event_loop().run_until_complete(proc.async_start()) os.close(null_fd) self.assertEqual(proc.wait(), os.EX_OK) f = io.open(_unicode_encode(logfile, diff --git a/lib/portage/tests/lazyimport/test_lazy_import_portage_baseline.py b/lib/portage/tests/lazyimport/test_lazy_import_portage_baseline.py index 080cf3f98..f52fb9cd2 100644 --- a/lib/portage/tests/lazyimport/test_lazy_import_portage_baseline.py +++ b/lib/portage/tests/lazyimport/test_lazy_import_portage_baseline.py @@ -1,4 +1,4 @@ -# Copyright 2010-2011 Gentoo Foundation +# Copyright 2010-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import re @@ -60,14 +60,14 @@ sys.stdout.write(" ".join(k for k in sys.modules args=self._baseline_import_cmd, env=env, fd_pipes={1:slave_fd}, scheduler=scheduler) - producer.start() + scheduler.run_until_complete(producer.async_start()) slave_file.close() consumer = PipeReader( input_files={"producer" : master_file}, scheduler=scheduler) - consumer.start() + scheduler.run_until_complete(consumer.async_start()) consumer.wait() self.assertEqual(producer.wait(), os.EX_OK) self.assertEqual(consumer.wait(), os.EX_OK) diff --git a/lib/portage/tests/locks/test_asynchronous_lock.py b/lib/portage/tests/locks/test_asynchronous_lock.py index 338d91e09..43008f038 100644 --- a/lib/portage/tests/locks/test_asynchronous_lock.py +++ b/lib/portage/tests/locks/test_asynchronous_lock.py @@ -1,4 +1,4 @@ -# Copyright 2010-2011 Gentoo Foundation +# Copyright 2010-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import signal @@ -29,7 +29,7 @@ class AsynchronousLockTestCase(TestCase): scheduler=scheduler, _force_async=force_async, _force_thread=True, _force_dummy=force_dummy) - async_lock.start() + scheduler.run_until_complete(async_lock.async_start()) self.assertEqual(async_lock.wait(), os.EX_OK) self.assertEqual(async_lock.returncode, os.EX_OK) scheduler.run_until_complete(async_lock.async_unlock()) @@ -37,7 +37,7 @@ class AsynchronousLockTestCase(TestCase): async_lock = AsynchronousLock(path=path, scheduler=scheduler, _force_async=force_async, _force_process=True) - async_lock.start() + scheduler.run_until_complete(async_lock.async_start()) self.assertEqual(async_lock.wait(), os.EX_OK) self.assertEqual(async_lock.returncode, os.EX_OK) scheduler.run_until_complete(async_lock.async_unlock()) @@ -63,7 +63,7 @@ class AsynchronousLockTestCase(TestCase): try: path = os.path.join(tempdir, 'lock_me') lock1 = AsynchronousLock(path=path, scheduler=scheduler) - lock1.start() + scheduler.run_until_complete(lock1.async_start()) self.assertEqual(lock1.wait(), os.EX_OK) self.assertEqual(lock1.returncode, os.EX_OK) @@ -73,7 +73,7 @@ class AsynchronousLockTestCase(TestCase): # one time concurrently. lock2 = AsynchronousLock(path=path, scheduler=scheduler, _force_async=True, _force_process=True) - lock2.start() + scheduler.run_until_complete(lock2.async_start()) # lock2 should be waiting for lock1 to release self.assertEqual(lock2.poll(), None) self.assertEqual(lock2.returncode, None) @@ -104,12 +104,12 @@ class AsynchronousLockTestCase(TestCase): try: path = os.path.join(tempdir, 'lock_me') lock1 = AsynchronousLock(path=path, scheduler=scheduler) - lock1.start() + scheduler.run_until_complete(lock1.async_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() + scheduler.run_until_complete(lock2.async_start()) # lock2 should be waiting for lock1 to release self.assertEqual(lock2.poll(), None) self.assertEqual(lock2.returncode, None) @@ -142,12 +142,12 @@ class AsynchronousLockTestCase(TestCase): try: path = os.path.join(tempdir, 'lock_me') lock1 = AsynchronousLock(path=path, scheduler=scheduler) - lock1.start() + scheduler.run_until_complete(lock1.async_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() + scheduler.run_until_complete(lock2.async_start()) # lock2 should be waiting for lock1 to release self.assertEqual(lock2.poll(), None) self.assertEqual(lock2.returncode, None) diff --git a/lib/portage/tests/process/test_PopenProcess.py b/lib/portage/tests/process/test_PopenProcess.py index 88da0b354..ed506b814 100644 --- a/lib/portage/tests/process/test_PopenProcess.py +++ b/lib/portage/tests/process/test_PopenProcess.py @@ -1,4 +1,4 @@ -# Copyright 2012-2013 Gentoo Foundation +# Copyright 2012-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import subprocess @@ -34,7 +34,7 @@ class PopenPipeTestCase(TestCase): consumer = producer.pipe_reader consumer.input_files = {"producer" : producer.proc.stdout} - producer.start() + global_event_loop().run_until_complete(producer.async_start()) producer.wait() self.assertEqual(producer.returncode, os.EX_OK) @@ -58,7 +58,7 @@ class PopenPipeTestCase(TestCase): producer.pipe_reader = consumer - producer.start() + global_event_loop().run_until_complete(producer.async_start()) producer.wait() self.assertEqual(producer.returncode, os.EX_OK) diff --git a/lib/portage/tests/process/test_PopenProcessBlockingIO.py b/lib/portage/tests/process/test_PopenProcessBlockingIO.py index 9ee291a39..91a272ce1 100644 --- a/lib/portage/tests/process/test_PopenProcessBlockingIO.py +++ b/lib/portage/tests/process/test_PopenProcessBlockingIO.py @@ -1,4 +1,4 @@ -# Copyright 2012 Gentoo Foundation +# Copyright 2012-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import subprocess @@ -40,7 +40,7 @@ class PopenPipeBlockingIOTestCase(TestCase): consumer = producer.pipe_reader consumer.input_files = {"producer" : producer.proc.stdout} - producer.start() + global_event_loop().run_until_complete(producer.async_start()) producer.wait() self.assertEqual(producer.returncode, os.EX_OK) diff --git a/lib/portage/tests/process/test_poll.py b/lib/portage/tests/process/test_poll.py index f505b5049..2217b6d6e 100644 --- a/lib/portage/tests/process/test_poll.py +++ b/lib/portage/tests/process/test_poll.py @@ -1,4 +1,4 @@ -# Copyright 1998-2019 Gentoo Authors +# Copyright 1998-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import functools @@ -67,7 +67,7 @@ class PipeReaderTestCase(TestCase): input_files={"producer" : master_file}, _use_array=self._use_array, scheduler=scheduler) - consumer.start() + scheduler.run_until_complete(consumer.async_start()) producer = scheduler.run_until_complete(asyncio.create_subprocess_exec( "bash", "-c", self._echo_cmd % test_string, diff --git a/lib/portage/tests/util/futures/test_iter_completed.py b/lib/portage/tests/util/futures/test_iter_completed.py index 9ab410a9e..aa24f5685 100644 --- a/lib/portage/tests/util/futures/test_iter_completed.py +++ b/lib/portage/tests/util/futures/test_iter_completed.py @@ -1,4 +1,4 @@ -# Copyright 2018 Gentoo Foundation +# Copyright 2018-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import time @@ -6,6 +6,7 @@ from portage.tests import TestCase from portage.util._async.ForkProcess import ForkProcess from portage.util._eventloop.global_event_loop import global_event_loop from portage.util.futures import asyncio +from portage.util.futures.compat_coroutine import coroutine, coroutine_return from portage.util.futures.iter_completed import ( iter_completed, async_iter_completed, @@ -43,12 +44,17 @@ class IterCompletedTestCase(TestCase): expected_order = sorted(task.seconds for task in tasks) + @coroutine + def task_coroutine(task): + task.future = loop.create_future() + task.scheduler = loop + yield task.async_start() + result = yield task.future + coroutine_return(result) + def future_generator(): for task in tasks: - task.future = loop.create_future() - task.scheduler = loop - task.start() - yield task.future + yield task_coroutine(task) for seconds, future in zip(expected_order, iter_completed(future_generator(), max_jobs=True, max_load=None, loop=loop)): diff --git a/lib/portage/tests/util/test_file_copier.py b/lib/portage/tests/util/test_file_copier.py index 3f1ec6b78..26a72b04b 100644 --- a/lib/portage/tests/util/test_file_copier.py +++ b/lib/portage/tests/util/test_file_copier.py @@ -1,4 +1,4 @@ -# Copyright 2019 Gentoo Authors +# Copyright 2019-2020 Gentoo Authors # Distributed under the terms of the GNU General Public License v2 import errno @@ -28,7 +28,7 @@ class FileCopierTestCase(TestCase): f.write(content) os.chmod(src_path, file_mode) copier = FileCopier(src_path=src_path, dest_path=dest_path, scheduler=loop) - copier.start() + loop.run_until_complete(copier.async_start()) loop.run_until_complete(copier.async_wait()) self.assertEqual(copier.returncode, 0) copier.future.result() @@ -39,7 +39,7 @@ class FileCopierTestCase(TestCase): # failure due to nonexistent src_path src_path = os.path.join(tempdir, 'does-not-exist') copier = FileCopier(src_path=src_path, dest_path=dest_path, scheduler=loop) - copier.start() + loop.run_until_complete(copier.async_start()) loop.run_until_complete(copier.async_wait()) self.assertEqual(copier.returncode, 1) self.assertEqual(copier.future.exception().errno, errno.ENOENT)
