From fa844327fbbe01e6ea93fbae401c80cce6455508 Mon Sep 17 00:00:00 2001 From: Mathieu Benoit Date: Sat, 18 Feb 2023 23:20:42 -0500 Subject: [PATCH] [UPD] script test parallel: move async to lib --- script/lib_asyncio.py | 152 +++++++++++++++++++++++++++++ script/test/run_parallel_test.py | 160 ++++++------------------------- 2 files changed, 181 insertions(+), 131 deletions(-) create mode 100644 script/lib_asyncio.py diff --git a/script/lib_asyncio.py b/script/lib_asyncio.py new file mode 100644 index 0000000..7b64e90 --- /dev/null +++ b/script/lib_asyncio.py @@ -0,0 +1,152 @@ +#!./.venv/bin/python +# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl) +import asyncio +from collections import deque + +import uvloop + + +async def run_in_serial(task_list): + q = asyncio.Queue() + for task in task_list: + await q.put(task) + lst_result = [] + for i in range(len(task_list)): + co = await q.get() + result = await co + lst_result.append(result) + return lst_result + + +async def run_command_get_output(*args, cwd=None): + if cwd is not None: + process = await asyncio.create_subprocess_exec( + *args, + # stdout must a pipe to be accessible as process.stdout + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=cwd, + ) + else: + process = await asyncio.create_subprocess_exec( + *args, + # stdout must a pipe to be accessible as process.stdout + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + # Wait for the subprocess to finish + stdout, stderr = await process.communicate() + return stdout.decode() + + +async def run_command_get_output_and_status(*args, cwd=None): + if cwd is not None: + process = await asyncio.create_subprocess_exec( + *args, + # stdout must a pipe to be accessible as process.stdout + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=cwd, + ) + else: + process = await asyncio.create_subprocess_exec( + *args, + # stdout must a pipe to be accessible as process.stdout + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + # Wait for the subprocess to finish + stdout, stderr = await process.communicate() + return stdout.decode(), stderr.decode(), process.returncode + + +def print_summary_task(task_list): + for task in task_list: + print(task.cr_code.co_name) + + +def execute(config, lst_task, use_uvloop=False): + + if not config.no_parallel and asyncio.get_event_loop().is_closed(): + asyncio.set_event_loop(asyncio.new_event_loop()) + + if config.no_parallel: + return_value = asyncio.run(run_in_serial(lst_task)) + elif config.max_process: + pool = AsyncioPool(config.max_process) + for task in lst_task: + pool.add_coro(task) + try: + return_value = pool.run_until_complete() + finally: + pool.close() + else: + # Use maximal resource + if use_uvloop: + uvloop.install() + loop = asyncio.get_event_loop() + if config.debug: + loop.set_debug(True) + try: + commands = asyncio.gather(*lst_task) + return_value = loop.run_until_complete(commands) + finally: + loop.close() + return return_value + + +class AsyncioPool: + # TODO check to replace this pool by https://github.com/dano/aioprocessing + def __init__(self, concurrency, loop=None): + """ + @param loop: asyncio loop + @param concurrency: Maximum number of concurrently running tasks + """ + self._loop = loop or asyncio.get_event_loop() + self._concurrency = concurrency + self._coros = deque([]) # All coroutines queued for execution + self._futures = [] # All currently running coroutines + self._lst_result = [] + + def close(self): + self._loop.close() + + def add_coro(self, coro): + """ + @param coro: coroutine to add + """ + self._coros.append(coro) + self.print_status() + + def run_until_complete(self): + self._loop.run_until_complete(self._wait_for_futures()) + return self._lst_result + + def print_status(self): + print( + " Status: coros:%s - futures:%s" + % (len(self._coros), len(self._futures)) + ) + + def _start_futures(self): + num_to_start = self._concurrency - len(self._futures) + num_to_start = min(num_to_start, len(self._coros)) + for _ in range(num_to_start): + coro = self._coros.popleft() + future = asyncio.ensure_future(coro, loop=self._loop) + self._futures.append(future) + self.print_status() + + async def _wait_for_futures(self): + while len(self._coros) > 0 or len(self._futures) > 0: + self._start_futures() + futures_completed, futures_pending = await asyncio.wait( + self._futures, + loop=self._loop, + return_when=asyncio.FIRST_COMPLETED, + ) + + for future in futures_completed: + self._lst_result.append(future.result()) + self._futures.remove(future) + self._start_futures() diff --git a/script/test/run_parallel_test.py b/script/test/run_parallel_test.py index de1b041..f36bcb0 100755 --- a/script/test/run_parallel_test.py +++ b/script/test/run_parallel_test.py @@ -9,13 +9,19 @@ import sys import tempfile import time import uuid -from collections import deque from typing import Tuple import aioshutil import git from colorama import Fore +new_path = os.path.normpath( + os.path.join(os.path.dirname(__file__), "..", "..") +) +sys.path.append(new_path) + +from script import lib_asyncio + logging.basicConfig(level=logging.DEBUG) _logger = logging.getLogger(__name__) @@ -172,27 +178,6 @@ def print_log(lst_task, tpl_result): f.write("\n") -async def run_command_get_output(*args, cwd=None): - if cwd is not None: - process = await asyncio.create_subprocess_exec( - *args, - # stdout must a pipe to be accessible as process.stdout - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - cwd=cwd, - ) - else: - process = await asyncio.create_subprocess_exec( - *args, - # stdout must a pipe to be accessible as process.stdout - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - # Wait for the subprocess to finish - stdout, stderr = await process.communicate() - return stdout.decode() - - async def run_command(*args, test_name=None): # Create subprocess start_time = time.time() @@ -378,8 +363,10 @@ async def test_exec( lst_module_to_test = tested_module.split(",") for module_name in lst_module_to_test: # Update path to change new emplacement - s_lst_path_tested_module = await run_command_get_output( - "find", "./addons/", "-name", module_name + s_lst_path_tested_module = ( + await lib_asyncio.run_command_get_output( + "find", "./addons/", "-name", module_name + ) ) if not s_lst_path_tested_module: return ( @@ -420,12 +407,14 @@ async def test_exec( os.path.join(temp_dir_name, s_first_path) ) - s_lst_path_generated_module = await run_command_get_output( - "find", - "./addons/", - "-name", - generated_module, - cwd=temp_dir_name, + s_lst_path_generated_module = ( + await lib_asyncio.run_command_get_output( + "find", + "./addons/", + "-name", + generated_module, + cwd=temp_dir_name, + ) ) if s_lst_path_generated_module: lst_path_generated_module = ( @@ -507,15 +496,19 @@ async def test_exec( hook.write(new_hook_line) # Format editing code before commit - await run_command_get_output( + await lib_asyncio.run_command_get_output( "./script/maintenance/black.sh", temp_dir_name ) # init repo with git for dir_to_git in lst_path_to_add_config: - await run_command_get_output("git", "init", ".", cwd=dir_to_git) - await run_command_get_output("git", "add", ".", cwd=dir_to_git) - await run_command_get_output( + await lib_asyncio.run_command_get_output( + "git", "init", ".", cwd=dir_to_git + ) + await lib_asyncio.run_command_get_output( + "git", "add", ".", cwd=dir_to_git + ) + await lib_asyncio.run_command_get_output( "git", "commit", "-am", "'first commit'", cwd=dir_to_git ) @@ -714,11 +707,6 @@ def check_git_change(): return not status -def print_summary_task(task_list): - for task in task_list: - print(task.cr_code.co_name) - - # START TEST @@ -1173,75 +1161,6 @@ async def run_helloworld_test(config) -> Tuple[str, int]: return res, status -async def run_in_serial(task_list): - q = asyncio.Queue() - for task in task_list: - await q.put(task) - lst_result = [] - for i in range(len(task_list)): - co = await q.get() - result = await co - lst_result.append(result) - return lst_result - - -class AsyncioPool: - # TODO check to replace this pool by https://github.com/dano/aioprocessing - def __init__(self, concurrency, loop=None): - """ - @param loop: asyncio loop - @param concurrency: Maximum number of concurrently running tasks - """ - self._loop = loop or asyncio.get_event_loop() - self._concurrency = concurrency - self._coros = deque([]) # All coroutines queued for execution - self._futures = [] # All currently running coroutines - self._lst_result = [] - - def close(self): - self._loop.close() - - def add_coro(self, coro): - """ - @param coro: coroutine to add - """ - self._coros.append(coro) - self.print_status() - - def run_until_complete(self): - self._loop.run_until_complete(self._wait_for_futures()) - return self._lst_result - - def print_status(self): - print( - " Status: coros:%s - futures:%s" - % (len(self._coros), len(self._futures)) - ) - - def _start_futures(self): - num_to_start = self._concurrency - len(self._futures) - num_to_start = min(num_to_start, len(self._coros)) - for _ in range(num_to_start): - coro = self._coros.popleft() - future = asyncio.ensure_future(coro, loop=self._loop) - self._futures.append(future) - self.print_status() - - async def _wait_for_futures(self): - while len(self._coros) > 0 or len(self._futures) > 0: - self._start_futures() - futures_completed, futures_pending = await asyncio.wait( - self._futures, - loop=self._loop, - return_when=asyncio.FIRST_COMPLETED, - ) - - for future in futures_completed: - self._lst_result.append(future.result()) - self._futures.remove(future) - self._start_futures() - - def run_all_test(config) -> None: # low in time, at the end for more speed task_list = [ @@ -1268,31 +1187,10 @@ def run_all_test(config) -> None: run_demo_test(config), ] - print_summary_task(task_list) + lib_asyncio.print_summary_task(task_list) - if asyncio.get_event_loop().is_closed(): - asyncio.set_event_loop(asyncio.new_event_loop()) + tpl_result = lib_asyncio.execute(config, task_list) - if config.no_parallel: - tpl_result = asyncio.run(run_in_serial(task_list)) - elif config.max_process: - pool = AsyncioPool(config.max_process) - for task in task_list: - pool.add_coro(task) - try: - tpl_result = pool.run_until_complete() - finally: - pool.close() - else: - # Use maximal resource - loop = asyncio.get_event_loop() - if config.debug: - loop.set_debug(True) - try: - commands = asyncio.gather(*task_list) - tpl_result = loop.run_until_complete(commands) - finally: - loop.close() print_log(task_list, tpl_result) status = check_result(task_list, tpl_result) if status: