Merge pull request #3512 from vkarak/bugfix/properly-cleanup-buildjobs

[bugfix] Properly cleanup build jobs on abort

Author: vkarak

reframe/core/pipeline.py

@@ -1762,12 +1762,14 @@ def _create_job(self, job_type, force_local=False, **job_opts):
 
         if job_type == 'build':
             script_name = 'rfm_build.sh'
+            job_name = f'rfm_build_{self.short_name}'
         elif job_type == 'run':
             script_name = 'rfm_job.sh'
+            job_name = f'rfm_{self.short_name}'
 
         return Job.create(scheduler,
                           launcher,
-                          name=f'rfm_{self.short_name}',
+                          name=job_name,
                           script_filename=script_name,
                           workdir=self._stagedir,
                           sched_access=self._current_partition.access,

reframe/frontend/executors/__init__.py

@@ -391,8 +391,8 @@ def __exit__(this, exc_type, exc_value, traceback):
                 # This practically ignores skipping during the cleanup phase
                 self.skip()
                 raise TaskExit from e
-        except ABORT_REASONS:
-            self.fail()
+        except ABORT_REASONS as e:
+            self.abort(e)
             raise
         except BaseException as e:
             self.fail()
@@ -523,6 +523,11 @@ def skip(self, exc_info=None):
         self._notify_listeners('on_task_skip')
 
     def abort(self, cause=None):
+        def _cancel_job(job):
+            if job:
+                with contextlib.suppress(JobNotStartedError):
+                    job.cancel()
+
         if self.failed or self._aborted:
             return
 
@@ -531,11 +536,8 @@ def abort(self, cause=None):
         exc.__cause__ = cause
         self._aborted = True
         try:
-            if self.check.job:
-                self.check.job.cancel()
-                self.check.job.wait()
-        except JobNotStartedError:
-            self.fail((type(exc), exc, None), 'on_task_abort')
+            _cancel_job(self.check.build_job)
+            _cancel_job(self.check.job)
         except BaseException:
             self.fail()
         else:

unittests/test_cli.py

@@ -303,7 +303,7 @@ def test_check_kbd_interrupt(run_reframe):
     )
     assert 'Traceback' not in stdout
     assert 'Traceback' not in stderr
-    assert 'FAILED' in stdout
+    assert 'ABORTED' in stdout
     assert returncode != 0
 
 

unittests/test_policies.py

@@ -253,7 +253,7 @@ def test_kbd_interrupt_within_test(make_runner, make_cases, common_exec_ctx):
         runner.runall(make_cases([make_kbd_check()]))
 
     stats = runner.stats
-    assert 1 == len(stats.failed())
+    assert 1 == len(stats.aborted())
     assert_all_dead(runner)
 
 
@@ -318,7 +318,7 @@ def test_sigterm_handling(make_runner, make_cases, common_exec_ctx):
 
     assert_all_dead(runner)
     assert runner.stats.num_cases() == 1
-    assert len(runner.stats.failed()) == 1
+    assert len(runner.stats.aborted()) == 1
 
 
 def test_reruns(make_runner, make_cases, common_exec_ctx):
@@ -576,16 +576,13 @@ def test_concurrency_none(make_async_runner, make_cases,
 def assert_interrupted_run(runner):
     assert 4 == runner.stats.num_cases()
     assert_runall(runner)
-    assert 1 == len(runner.stats.failed())
-    assert 3 == len(runner.stats.aborted())
+    assert 4 == len(runner.stats.aborted())
     assert_all_dead(runner)
 
     # Verify that failure reasons for the different tasks are correct
     for t in runner.stats.tasks():
-        if isinstance(t.check, KeyboardInterruptCheck):
-            assert t.exc_info[0] == KeyboardInterrupt
-        else:
-            assert t.exc_info[0] == AbortTaskError
+        assert t.exc_info[0] == AbortTaskError
+        assert isinstance(t.exc_info[1].__cause__, KeyboardInterrupt)
 
 
 def test_kbd_interrupt_in_wait_with_concurrency(

unittests/test_schedulers.py

@@ -777,36 +777,39 @@ def _read_pid(job, attempts=3):
                 f'{attempts} attempts')
 
 
-@pytest.mark.flaky(reruns=3)
 def test_cancel_with_grace(minimal_job, scheduler, local_only):
     # This test emulates a spawned process that ignores the SIGTERM signal
     # and also spawns another process:
     #
     #   reframe --- local job script --- sleep 5
     #                  (TERM IGN)
     #
-    # We expect the job not to be cancelled immediately, since it ignores
-    # the gracious signal we are sending it. However, we expect it to be
-    # killed immediately after the grace period of 2 seconds expires.
+    # We expect the job not to be cancelled immediately, since it ignores the
+    # gracious signal we are sending it. However, we expect it to be killed
+    # immediately after the grace period of 2 seconds expires. There is a
+    # little tweak though. Since we do not know if the shell will be killed
+    # first or the `sleep 5` process, we add an additional `sleep 1` at the
+    # end to stall the script and make sure that it also get the `TERM`
+    # signal. Otherwise,if the `sleep 5` is killed first, the script will
+    # continue and may be fast enough to not get the signal as well.
     #
     # We also check that the additional spawned process is also killed.
     minimal_job.time_limit = '1m'
     minimal_job.scheduler.CANCEL_GRACE_PERIOD = 2
     prepare_job(minimal_job,
                 command='sleep 5 &',
                 pre_run=['trap -- "" TERM'],
-                post_run=['echo $!', 'wait'],
+                post_run=['echo $!', 'wait', 'sleep 1'],
                 prepare_cmds=[''])
     submit_job(minimal_job)
 
     # Stall a bit here to let the the spawned process start and install its
     # signal handler for SIGTERM
-    time.sleep(1)
+    time.sleep(.2)
 
     sleep_pid = _read_pid(minimal_job)
     t_grace = time.time()
     minimal_job.cancel()
-    time.sleep(0.1)
     minimal_job.wait()
     t_grace = time.time() - t_grace
 
@@ -824,7 +827,6 @@ def test_cancel_with_grace(minimal_job, scheduler, local_only):
     assert_process_died(sleep_pid)
 
 
-# @pytest.mark.flaky(reruns=3)
 def test_cancel_term_ignore(minimal_job, scheduler, local_only):
     # This test emulates a descendant process of the spawned job that
     # ignores the SIGTERM signal:
@@ -855,7 +857,6 @@ def test_cancel_term_ignore(minimal_job, scheduler, local_only):
     minimal_job.wait()
     t_grace = time.time() - t_grace
 
-    # assert t_grace >= 2 and t_grace < 5
     assert minimal_job.state == 'FAILURE'
     assert minimal_job.signal == signal.SIGTERM
     assert_process_died(sleep_pid)