This is an automated email from the ASF dual-hosted git repository. tvb pushed a commit to branch tpollard/buildsubtemp in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 2fe0c6a939bee0c205dcc650210c4905c07f1902 Author: Tom Pollard <[email protected]> AuthorDate: Fri Oct 11 10:45:58 2019 +0100 basic async in stream --- src/buildstream/_context.py | 3 + src/buildstream/_exceptions.py | 8 +++ src/buildstream/_frontend/app.py | 53 ++++++++------- src/buildstream/_frontend/status.py | 1 + src/buildstream/_scheduler/scheduler.py | 22 +++--- src/buildstream/_stream.py | 115 +++++++++++++++++++++----------- 6 files changed, 130 insertions(+), 72 deletions(-) diff --git a/src/buildstream/_context.py b/src/buildstream/_context.py index 8795550..54f483f 100644 --- a/src/buildstream/_context.py +++ b/src/buildstream/_context.py @@ -175,6 +175,9 @@ class Context(): self._workspace_project_cache = WorkspaceProjectCache() self._cascache = None + # An exception caught from subprocessing, used to handle run exceptions in App + self._subprocess_exception = None + # __enter__() # # Called when entering the with-statement context. diff --git a/src/buildstream/_exceptions.py b/src/buildstream/_exceptions.py index a6f726b..1468381 100644 --- a/src/buildstream/_exceptions.py +++ b/src/buildstream/_exceptions.py @@ -43,6 +43,14 @@ def get_last_exception(): _last_exception = None return le +# set_last_exception() +# +# Sets the last exception from the main process, used if Stream is running a subprocess +# +def set_last_exception(exception): + if 'BST_TEST_SUITE' in os.environ: + global _last_exception + _last_exception = exception # get_last_task_error() # diff --git a/src/buildstream/_frontend/app.py b/src/buildstream/_frontend/app.py index 4fd632d..c021602 100644 --- a/src/buildstream/_frontend/app.py +++ b/src/buildstream/_frontend/app.py @@ -285,38 +285,23 @@ class App(): try: yield except BstError as e: + self._handle_run_exception(e, session_name) - # Print a nice summary if this is a session - if session_name: - elapsed = self.stream.elapsed_time - - if isinstance(e, StreamError) and e.terminated: # pylint: disable=no-member - self._message(MessageType.WARN, session_name + ' Terminated', elapsed=elapsed) - else: - self._message(MessageType.FAIL, session_name, elapsed=elapsed) - - # Notify session failure - self._notify("{} failed".format(session_name), e) - - if self._started: - self._print_summary() - - # Exit with the error - self._error_exit(e) except RecursionError: click.echo("RecursionError: Dependency depth is too large. Maximum recursion depth exceeded.", err=True) sys.exit(-1) - else: + if self.context._subprocess_exception: + # If an exception was thrown in a Stream subprocessed asyncio method, handle it + self._handle_run_exception(self.context._subprocess_exception, session_name) + elif session_name: # No exceptions occurred, print session time and summary - if session_name: - self._message(MessageType.SUCCESS, session_name, elapsed=self.stream.elapsed_time) - if self._started: - self._print_summary() - - # Notify session success - self._notify("{} succeeded".format(session_name), "") + self._message(MessageType.SUCCESS, session_name, elapsed=self.stream.elapsed_time) + if self._started: + self._print_summary() + # Notify session success + self._notify("{} succeeded".format(session_name), "") # init_project() # @@ -891,6 +876,24 @@ class App(): return (project_name, format_version, element_path) + def _handle_run_exception(self, exception, session_name): + # Print a nice summary if this is a session + if session_name: + elapsed = self.stream.elapsed_time + + if isinstance(exception, StreamError) and exception.terminated: # pylint: disable=no-member + self._message(MessageType.WARN, session_name + ' Terminated', elapsed=elapsed) + else: + self._message(MessageType.FAIL, session_name, elapsed=elapsed) + + # Notify session failure + self._notify("{} failed".format(session_name), exception) + + if self._started: + self._print_summary() + + self._error_exit(exception) + # # Return a value processor for partial choice matching. diff --git a/src/buildstream/_frontend/status.py b/src/buildstream/_frontend/status.py index c648e93..87f35fb 100644 --- a/src/buildstream/_frontend/status.py +++ b/src/buildstream/_frontend/status.py @@ -373,6 +373,7 @@ class _StatusHeader(): # # ========= 00:00:00 project-name (143/387) ========= # + session = self._stream.len_session_elements total = self._stream.len_total_elements diff --git a/src/buildstream/_scheduler/scheduler.py b/src/buildstream/_scheduler/scheduler.py index a0a87fb..854921b 100644 --- a/src/buildstream/_scheduler/scheduler.py +++ b/src/buildstream/_scheduler/scheduler.py @@ -69,6 +69,7 @@ class NotificationType(FastEnum): START = "start" TASK_GROUPS = "task_groups" ELEMENT_TOTALS = "element_totals" + FINISH = "finish" # Notification() @@ -184,6 +185,9 @@ class Scheduler(): # Hold on to the queues to process self.queues = queues + # Check if we're subprocessed + subprocessed = bool(self._notify_front_queue) + # Ensure that we have a fresh new event loop, in case we want # to run another test in this thread. self.loop = asyncio.new_event_loop() @@ -198,10 +202,11 @@ class Scheduler(): # Handle unix signals while running self._connect_signals() - # Watch casd while running to ensure it doesn't die - self._casd_process = casd_process - _watcher = asyncio.get_child_watcher() - _watcher.add_child_handler(casd_process.pid, self._abort_on_casd_failure) + # If we're not in a subprocess, watch casd while running to ensure it doesn't die + if not subprocessed: + self._casd_process = casd_process + _watcher = asyncio.get_child_watcher() + _watcher.add_child_handler(casd_process.pid, self._abort_on_casd_failure) # Add notification listener if in subprocess self._start_listening() @@ -215,9 +220,10 @@ class Scheduler(): self._stop_listening() self.loop.close() - # Stop watching casd - _watcher.remove_child_handler(casd_process.pid) - self._casd_process = None + # Stop watching casd if not subprocessed + if self._casd_process: + _watcher.remove_child_handler(casd_process.pid) + self._casd_process = None # Stop handling unix signals self._disconnect_signals() @@ -236,7 +242,7 @@ class Scheduler(): status = SchedStatus.SUCCESS # Send the state taskgroups if we're running under the subprocess - if self._notify_front_queue: + if subprocessed: # Don't pickle state for group in self._state.task_groups.values(): group._state = None diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py index d034e77..8bae8ec 100644 --- a/src/buildstream/_stream.py +++ b/src/buildstream/_stream.py @@ -34,7 +34,7 @@ from contextlib import contextmanager, suppress from fnmatch import fnmatch from ._artifactelement import verify_artifact_ref, ArtifactElement -from ._exceptions import StreamError, ImplError, BstError, ArtifactElementError, ArtifactError, set_last_task_error, SubprocessException +from ._exceptions import StreamError, ImplError, BstError, ArtifactElementError, ArtifactError, set_last_task_error, SubprocessException, set_last_exception from ._message import Message, MessageType from ._scheduler import Scheduler, SchedStatus, TrackQueue, FetchQueue, \ SourcePushQueue, BuildQueue, PullQueue, ArtifactPushQueue, NotificationType, Notification, JobStatus @@ -72,8 +72,9 @@ class Stream(): self.session_elements = [] # List of elements being processed this session self.total_elements = [] # Total list of elements based on targets self.queues = [] # Queue objects - self.len_session_elements = None - self.len_total_elements = None + self.len_session_elements = '' + self.len_total_elements = '' + self.loop = None # # Private members @@ -101,6 +102,8 @@ class Stream(): self._scheduler_suspended = False self._notify_front_queue = None self._notify_back_queue = None + self._casd_process = None + self._watcher = None # init() # @@ -121,6 +124,7 @@ class Stream(): # Send the exceptions members dict to be reraised in main process exception_attrs = vars(e) notify.put(Notification(NotificationType.EXCEPTION, exception=exception_attrs)) + notify.put(Notification(NotificationType.FINISH)) def run_in_subprocess(self, func, *args, **kwargs): assert not self._subprocess @@ -143,24 +147,30 @@ class Stream(): self._subprocess.start() + # We can now launch another async + self.loop = asyncio.new_event_loop() + asyncio.set_event_loop(self.loop) + self._start_listening() + self.loop.set_exception_handler(self._handle_exception) + self._watch_casd() + self.loop.run_forever() + # TODO connect signal handlers with asyncio - while self._subprocess.exitcode is None: - # check every given time interval on subprocess state - self._subprocess.join(0.01) - # if no exit code, go back to checking the message queue - self._loop() + # Scheduler has stopped running, so safe to still have async here + self._stop_listening() + self._stop_watching_casd() + self.loop.close() + self.loop = None + self._subprocess.join() + self._subprocess = None # Set main process back utils._reset_main_pid() # Ensure no more notifcations to process - try: - while True: - notification = self._notify_front_queue.get_nowait() - self._notification_handler(notification) - except queue.Empty: - pass - + while not self._notify_front_queue.empty(): + notification = self._notify_front_queue.get_nowait() + self._notification_handler(notification) # cleanup() # @@ -171,7 +181,6 @@ class Stream(): for q in [self._notify_back_queue, self._notify_front_queue]: if q is not None: q.close() - #self._notification_queue.cancel_join_thread() if self._project: self._project.cleanup() @@ -334,16 +343,17 @@ class Stream(): if remote: use_config = False + elements, track_elements = \ self._load(targets, track_targets, - selection=selection, track_selection=PipelineSelection.ALL, - track_except_targets=track_except, - track_cross_junctions=track_cross_junctions, - ignore_junction_targets=ignore_junction_targets, - use_artifact_config=use_config, - artifact_remote_url=remote, - use_source_config=True, - dynamic_plan=True) + selection=selection, track_selection=PipelineSelection.ALL, + track_except_targets=track_except, + track_cross_junctions=track_cross_junctions, + ignore_junction_targets=ignore_junction_targets, + use_artifact_config=use_config, + artifact_remote_url=remote, + use_source_config=True, + dynamic_plan=True) # Remove the tracking elements from the main targets elements = self._pipeline.subtract_elements(elements, track_elements) @@ -1435,15 +1445,15 @@ class Stream(): # self.total_elements = list(self._pipeline.dependencies(self.targets, Scope.ALL)) - if self._session_start_callback is not None: - self._notify_front(Notification(NotificationType.START)) - # Also send through the session & total elements list lengths for status rendering element_totals = str(len(self.session_elements)), str(len(self.total_elements)) self._notify_front(Notification(NotificationType.ELEMENT_TOTALS, element_totals=element_totals)) + if self._session_start_callback is not None: + self._notify_front(Notification(NotificationType.START)) + status = self._scheduler.run(self.queues, self._context.get_cascache().get_casd_process()) if status == SchedStatus.ERROR: @@ -1764,12 +1774,14 @@ class Stream(): elif notification.notification_type == NotificationType.TASK_ERROR: set_last_task_error(*notification.task_error) elif notification.notification_type == NotificationType.EXCEPTION: - # Regenerate the exception here, so we don't have to pickle it raise SubprocessException(**notification.exception) elif notification.notification_type == NotificationType.START: self._session_start_callback() elif notification.notification_type == NotificationType.ELEMENT_TOTALS: self.len_session_elements, self.len_total_elements = notification.element_totals + elif notification.notification_type == NotificationType.FINISH: + if self.loop: + self.loop.stop() else: raise StreamError("Unrecognised notification type received") @@ -1785,18 +1797,43 @@ class Stream(): else: self._notification_handler(notification) - # The code to be run by the Stream's event loop while delegating - # work to a subprocess with the @subprocessed decorator def _loop(self): - assert self._notify_front_queue - # Check for and process new messages - while True: - try: - notification = self._notify_front_queue.get_nowait() - self._notification_handler(notification) - except queue.Empty: - notification = None - break + while not self._notify_front_queue.empty(): + notification = self._notify_front_queue.get_nowait() + self._notification_handler(notification) + + def _start_listening(self): + if self._notify_front_queue: + self.loop.add_reader(self._notify_front_queue._reader.fileno(), self._loop) + + def _stop_listening(self): + if self._notify_front_queue: + self.loop.remove_reader(self._notify_front_queue._reader.fileno()) + + def _watch_casd(self): + if self._context.get_cascache()._casd_process: + self._casd_process = self._context.get_cascache().get_casd_process() + self._watcher = asyncio.get_child_watcher() + self._watcher.add_child_handler(self._casd_process.pid, self._abort_on_casd_failure) + + def _abort_on_casd_failure(self, pid, returncode): + message = Message(MessageType.BUG, "buildbox-casd died while the pipeline was active.") + self._notify_front(Notification(NotificationType.MESSAGE, message=message)) + self._casd_process.returncode = returncode + notification = Notification(NotificationType.TERMINATE) + self._notify_back(notification) + + def _stop_watching_casd(self): + self._watcher.remove_child_handler(self._casd_process.pid) + self._casd_process = None + + def _handle_exception(self, loop, context): + exception = context.get('exception') + # Set the last exception for the test suite if needed + set_last_exception(exception) + # Add it to context + self._context._subprocess_exception = exception + self.loop.stop() def __getstate__(self): # The only use-cases for pickling in BuildStream at the time of writing
