This is an automated email from the ASF dual-hosted git repository. not-in-ldap pushed a commit to branch tpollard/buildsubtemp in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 7870bc3bcbc958e07fc9cbdb320418209a489113 Author: Tom Pollard <[email protected]> AuthorDate: Mon Oct 21 16:24:33 2019 +0100 basic user SIG interrupt handling in Stream --- src/buildstream/_scheduler/scheduler.py | 9 +++++++-- src/buildstream/_stream.py | 21 +++++++++++++++++++-- 2 files changed, 26 insertions(+), 4 deletions(-) diff --git a/src/buildstream/_scheduler/scheduler.py b/src/buildstream/_scheduler/scheduler.py index 854921b..74a688f 100644 --- a/src/buildstream/_scheduler/scheduler.py +++ b/src/buildstream/_scheduler/scheduler.py @@ -70,6 +70,7 @@ class NotificationType(FastEnum): TASK_GROUPS = "task_groups" ELEMENT_TOTALS = "element_totals" FINISH = "finish" + SIGTSTP = "sigstp" # Notification() @@ -536,8 +537,10 @@ class Scheduler(): if self.terminated: return - notification = Notification(NotificationType.INTERRUPT) - self._notify_front(notification) + if not self._notify_front_queue: + # Not running in a subprocess, scheduler process to handle keyboard interrupt + notification = Notification(NotificationType.INTERRUPT) + self._notify_front(notification) # _terminate_event(): # @@ -629,6 +632,8 @@ class Scheduler(): self.jobs_unsuspended() elif notification.notification_type == NotificationType.RETRY: self._failure_retry(notification.job_action, notification.element) + elif notification.notification_type == NotificationType.SIGTSTP: + self._suspend_event() else: raise ValueError("Unrecognised notification type received") diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py index 8bae8ec..866cbd7 100644 --- a/src/buildstream/_stream.py +++ b/src/buildstream/_stream.py @@ -30,6 +30,7 @@ import shutil import tarfile import tempfile import queue +import signal from contextlib import contextmanager, suppress from fnmatch import fnmatch @@ -150,17 +151,17 @@ class Stream(): # We can now launch another async self.loop = asyncio.new_event_loop() asyncio.set_event_loop(self.loop) + self._connect_signals() self._start_listening() self.loop.set_exception_handler(self._handle_exception) self._watch_casd() self.loop.run_forever() - # TODO connect signal handlers with asyncio - # Scheduler has stopped running, so safe to still have async here self._stop_listening() self._stop_watching_casd() self.loop.close() + self._disconnect_signals() self.loop = None self._subprocess.join() self._subprocess = None @@ -1221,10 +1222,14 @@ class Stream(): # Send the notification to suspend jobs notification = Notification(NotificationType.SUSPEND) self._notify_back(notification) + # Disconnect signals if stream is handling them + self._disconnect_signals() yield # Unsuspend jobs on context exit notification = Notification(NotificationType.UNSUSPEND) self._notify_back(notification) + # Connect signals if stream is handling them + self._connect_signals() ############################################################# # Private Methods # @@ -1835,6 +1840,18 @@ class Stream(): self._context._subprocess_exception = exception self.loop.stop() + def _connect_signals(self): + if self.loop: + self.loop.add_signal_handler(signal.SIGINT, self._interrupt_callback) + self.loop.add_signal_handler(signal.SIGTERM, lambda: self._notify_back(Notification(NotificationType.TERMINATE))) + self.loop.add_signal_handler(signal.SIGTSTP, lambda: self._notify_back(Notification(NotificationType.SIGTSTP))) + + def _disconnect_signals(self): + if self.loop: + self.loop.remove_signal_handler(signal.SIGINT) + self.loop.remove_signal_handler(signal.SIGTSTP) + self.loop.remove_signal_handler(signal.SIGTERM) + def __getstate__(self): # The only use-cases for pickling in BuildStream at the time of writing # are enabling the 'spawn' method of starting child processes, and
