[bugfix] Do not leave stale test jobs when ReFrame receives a TERM signal

reframe/core/exceptions.py

@@ -112,6 +112,12 @@ class PipelineError(ReframeError):
     '''
 
 
+class ReframeForceExitError(ReframeError):
+    '''Raised when ReFrame execution must be forcefully ended,
+    e.g., after a SIGTERM was received.
+    '''
+
+
 class StatisticsError(ReframeError):
     '''Raised to denote an error in dealing with statistics.'''
 

reframe/frontend/cli.py

@@ -19,9 +19,11 @@
 import reframe.frontend.check_filters as filters
 import reframe.frontend.dependency as dependency
 import reframe.utility.os_ext as os_ext
-from reframe.core.exceptions import (EnvironError, ConfigError, ReframeError,
-                                     ReframeFatalError, format_exception,
-                                     SystemAutodetectionError)
+from reframe.core.exceptions import (
+    ConfigError, EnvironError, ReframeError, ReframeFatalError,
+    ReframeForceExitError, SystemAutodetectionError
+)
+from reframe.core.exceptions import format_exception
 from reframe.frontend.executors import Runner, generate_testcases
 from reframe.frontend.executors.policies import (SerialExecutionPolicy,
                                                  AsynchronousExecutionPolicy)
@@ -637,7 +639,7 @@ def main():
 
         sys.exit(0)
 
-    except KeyboardInterrupt:
+    except (KeyboardInterrupt, ReframeForceExitError):
         sys.exit(1)
     except ReframeError as e:
         printer.error(str(e))

reframe/frontend/executors/__init__.py

@@ -5,6 +5,7 @@
 
 import abc
 import copy
+import signal
 import sys
 import weakref
 
@@ -14,11 +15,11 @@
 import reframe.core.runtime as runtime
 import reframe.frontend.dependency as dependency
 from reframe.core.exceptions import (AbortTaskError, JobNotStartedError,
-                                     ReframeFatalError, TaskExit)
+                                     ReframeForceExitError, TaskExit)
 from reframe.frontend.printer import PrettyPrinter
 from reframe.frontend.statistics import TestStats
 
-ABORT_REASONS = (KeyboardInterrupt, ReframeFatalError, AssertionError)
+ABORT_REASONS = (KeyboardInterrupt, ReframeForceExitError, AssertionError)
 
 
 class TestCase:
@@ -256,6 +257,10 @@ def on_task_success(self, task):
         '''Called when a regression test has succeeded.'''
 
 
+def _handle_sigterm(signum, frame):
+    raise ReframeForceExitError('received TERM signal')
+
+
 class Runner:
     '''Responsible for executing a set of regression tests based on an
     execution policy.'''
@@ -267,6 +272,7 @@ def __init__(self, policy, printer=None, max_retries=0):
         self._stats = TestStats()
         self._policy.stats = self._stats
         self._policy.printer = self._printer
+        signal.signal(signal.SIGTERM, _handle_sigterm)
 
     def __repr__(self):
         return debug.repr(self)
@@ -376,7 +382,6 @@ def __init__(self):
 
         # Task event listeners
         self.task_listeners = []
-
         self.stats = None
 
     def __repr__(self):

unittests/test_policies.py

@@ -6,7 +6,9 @@
 import collections
 import itertools
 import os
+import multiprocessing
 import pytest
+import time
 import tempfile
 import unittest
 
@@ -19,7 +21,8 @@
 import reframe.utility.os_ext as os_ext
 from reframe.core.environments import Environment
 from reframe.core.exceptions import (
-    DependencyError, JobNotStartedError, TaskDependencyError
+    DependencyError, JobNotStartedError,
+    ReframeForceExitError, TaskDependencyError
 )
 from reframe.frontend.loader import RegressionCheckLoader
 import unittests.fixtures as fixtures
@@ -257,6 +260,39 @@ def test_dependencies(self):
             if t.ref_count == 0:
                 assert os.path.exists(os.path.join(check.outputdir, 'out.txt'))
 
+    def test_sigterm(self):
+        # Wrapper of self.runall which is used from a child process and
+        # passes any exception to the parent process
+        def _runall(checks, conn):
+            exc = None
+            try:
+                self.runall(checks)
+            except BaseException as e:
+                exc = e
+
+            conn.send((exc, len(self.runner.stats.failures())))
+            conn.close()
+
+        rd_endpoint, wr_endpoint = multiprocessing.Pipe(duplex=False)
+        p = multiprocessing.Process(target=_runall,
+                                    args=([SleepCheck(3)], wr_endpoint))
+        p.start()
+
+        # The unused write endpoint has to be closed from the parent to
+        # ensure that the `recv()` method of `rd_endpoint` returns
+        wr_endpoint.close()
+
+        # Allow some time so that the SleepCheck is submitted
+        time.sleep(1)
+        p.terminate()
+        p.join()
+        exc, num_failures = rd_endpoint.recv()
+        assert 1 == num_failures
+        with pytest.raises(ReframeForceExitError,
+                           match='received TERM signal'):
+            if exc:
+                raise exc
+
     def test_dependencies_with_retries(self):
         self.runner._max_retries = 2
         self.test_dependencies()