[buildstream] 11/16: basic user SIG interrupt handling in Stream

[no...@apache.org]

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 <to...@codethink.co.uk>
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