Merge branch 'master' into enhancement/valid_sysenv-logging

Author: vkarak

reframe/core/schedulers/__init__.py

@@ -625,7 +625,7 @@ def wait(self):
             raise JobNotStartedError('cannot wait an unstarted job')
 
         self.scheduler.wait(self)
-        self._completion_time = self._completion_time or time.time()
+        self.finished()
 
     def cancel(self):
         if self.jobid is None:
@@ -640,6 +640,10 @@ def finished(self):
         done = self.scheduler.finished(self)
         if done:
             self._completion_time = self._completion_time or time.time()
+            if self._exception:
+                exc = self._exception
+                self._exception = None
+                raise exc
 
         return done
 

reframe/core/schedulers/flux.py

@@ -150,7 +150,4 @@ def wait(self, job):
             time.sleep(next(intervals))
 
     def finished(self, job):
-        if job.exception:
-            raise job.exception
-
         return job.completed

reframe/core/schedulers/local.py

@@ -3,6 +3,7 @@
 #
 # SPDX-License-Identifier: BSD-3-Clause
 
+import contextlib
 import errno
 import os
 import signal
@@ -67,6 +68,7 @@ def submit(self, job):
             stderr=f_stderr,
             start_new_session=True
         )
+        self.log(f'spawned local process: {proc.pid}')
 
         # Update job info
         job._jobid = proc.pid
@@ -92,10 +94,11 @@ def filternodes(self, job, nodes):
         return [sched.AlwaysIdleNode(socket.gethostname())]
 
     def _kill_all(self, job):
-        '''Send SIGKILL to all the processes of the spawned job.'''
+        '''Send SIGKILL to all the processes of the spawned job and wait for
+        any children to finish'''
         try:
+            self.log(f'sending SIGKILL to process group {job._jobid}')
             os.killpg(job._jobid, signal.SIGKILL)
-            job._signal = signal.SIGKILL
         except (ProcessLookupError, PermissionError):
             # The process group may already be dead or assigned to a different
             # group, so ignore this error
@@ -104,19 +107,19 @@ def _kill_all(self, job):
             # Close file handles
             job.f_stdout.close()
             job.f_stderr.close()
-            job._state = 'FAILURE'
+            with contextlib.suppress(ChildProcessError):
+                os.waitpid(0, 0)
 
     def _term_all(self, job):
         '''Send SIGTERM to all the processes of the spawned job.'''
         try:
+            self.log(f'sending SIGTERM to process group {job._jobid}')
             os.killpg(job._jobid, signal.SIGTERM)
-            job._signal = signal.SIGTERM
         except (ProcessLookupError, PermissionError):
             # Job has finished already, close file handles
             self.log(f'pid {job.jobid} already dead')
             job.f_stdout.close()
             job.f_stderr.close()
-            job._state = 'FAILURE'
 
     def cancel(self, job):
         '''Cancel job.
@@ -126,6 +129,7 @@ def cancel(self, job):
 
         This function waits for the spawned process tree to finish.
         '''
+        self.log(f'cancelling job {job._jobid}')
         self._term_all(job)
         job._cancel_time = time.time()
 
@@ -150,10 +154,7 @@ def finished(self, job):
         the process has finished, you *must* call wait() to properly cleanup
         after it.
         '''
-        if job.exception:
-            raise job.exception
-
-        return job.state in ['SUCCESS', 'FAILURE', 'TIMEOUT']
+        return job.exitcode is not None or job.signal is not None
 
     def poll(self, *jobs):
         for job in jobs:
@@ -173,37 +174,45 @@ def _poll_job(self, job):
             else:
                 raise e
 
-        if job.cancel_time:
-            # Job has been cancelled; give it a grace period and kill it
-            self.log(f'Job {job.jobid} has been cancelled; '
-                     f'giving it a grace period')
-            t_rem = self.CANCEL_GRACE_PERIOD - (time.time() - job.cancel_time)
-            if t_rem > 0:
-                time.sleep(t_rem)
+        if pid:
+            # Job has finished
+            self.log(f'spawned process {job._jobid} has finished')
 
+            # Forcefully kill the whole session once the parent process exits
             self._kill_all(job)
-            return
 
-        if not pid:
-            # Job has not finished; check if we have reached a timeout
-            t_elapsed = time.time() - job.submit_time
-            if job.time_limit and t_elapsed > job.time_limit:
-                self._kill_all(job)
+            # Call wait() in the underlying Popen object to avoid false
+            # positive warnings
+            job._proc.wait()
+
+            # Retrieve the status of the job and return
+            if os.WIFEXITED(status):
+                job._exitcode = os.WEXITSTATUS(status)
+                if job._state == 'RUNNING':
+                    job._state = 'FAILURE' if job._exitcode != 0 else 'SUCCESS'
+            elif os.WIFSIGNALED(status):
+                if job._state == 'RUNNING':
+                    job._state = 'FAILURE'
+
+                job._signal = os.WTERMSIG(status)
+                self.log(f'job killed by signal: {job._signal}')
+
+            self.log(f'job state: {job._state}')
+        else:
+            # Job has not finished; check for timeouts
+            now = time.time()
+            t_elapsed = now - job.submit_time
+            if job.cancel_time:
+                t_rem = self.CANCEL_GRACE_PERIOD - (now - job.cancel_time)
+                self.log(f'job {job.jobid} has been cancelled; '
+                         f'giving it a grace period of {t_rem} seconds')
+                if t_rem <= 0:
+                    self._kill_all(job)
+            elif job.time_limit and t_elapsed > job.time_limit:
+                self.log(f'job {job._jobid} timed out; cancelling it')
+                self.cancel(job)
                 job._state = 'TIMEOUT'
                 job._exception = JobError(
                     f'job timed out ({t_elapsed:.6f}s > {job.time_limit}s)',
                     job.jobid
                 )
-
-            return
-
-        # Job has finished; kill the whole session
-        self._kill_all(job)
-
-        # Retrieve the status of the job and return
-        if os.WIFEXITED(status):
-            job._exitcode = os.WEXITSTATUS(status)
-            job._state = 'FAILURE' if job.exitcode != 0 else 'SUCCESS'
-        elif os.WIFSIGNALED(status):
-            job._state = 'FAILURE'
-            job._signal = os.WTERMSIG(status)

reframe/core/schedulers/lsf.py

@@ -147,7 +147,4 @@ def poll(self, *jobs):
                 job._completed = True
 
     def finished(self, job):
-        if job.exception:
-            raise job.exception
-
         return job.state == 'COMPLETED'

reframe/core/schedulers/pbs.py

@@ -190,9 +190,6 @@ def cancel(self, job):
         job._cancelled = True
 
     def finished(self, job):
-        if job.exception:
-            raise job.exception
-
         return job.completed
 
     def _update_nodelist(self, job, nodespec):

reframe/core/schedulers/sge.py

@@ -137,7 +137,4 @@ def poll(self, *jobs):
             job._state = 'COMPLETED'
 
     def finished(self, job):
-        if job.exception:
-            raise job.exception
-
         return job.state == 'COMPLETED'

reframe/core/schedulers/slurm.py

@@ -628,9 +628,6 @@ def cancel(self, job):
         job._is_cancelling = True
 
     def finished(self, job):
-        if job.exception:
-            raise job.exception
-
         return slurm_state_completed(job.state)
 
 

reframe/core/schedulers/ssh.py

@@ -174,9 +174,6 @@ def cancel(self, job):
                 step.cancel()
 
     def finished(self, job):
-        if job.exception:
-            raise job.exception
-
         return job.state is not None
 
     def poll(self, *jobs):

reframe/frontend/cli.py

@@ -1028,7 +1028,7 @@ def restrict_logging():
             sys.exit(0)
 
     if options.list_stored_testcases:
-        namepatt = '|'.join(options.names)
+        namepatt = '|'.join(n.replace('%', ' %') for n in options.names)
         with exit_gracefully_on_error('failed to retrieve test case data',
                                       printer):
             filt = options.filter_expr[-1] if options.filter_expr else None
@@ -1050,7 +1050,7 @@ def restrict_logging():
     if options.describe_stored_testcases:
         # Restore logging level
         printer.setLevel(logging.INFO)
-        namepatt = '|'.join(options.names)
+        namepatt = '|'.join(n.replace('%', ' %') for n in options.names)
         with exit_gracefully_on_error('failed to retrieve test case data',
                                       printer):
             filt = options.filter_expr[-1] if options.filter_expr else None
@@ -1068,7 +1068,7 @@ def restrict_logging():
             sys.exit(0)
 
     if options.performance_compare:
-        namepatt = '|'.join(options.names)
+        namepatt = '|'.join(n.replace('%', ' %') for n in options.names)
         with exit_gracefully_on_error('failed to generate performance report',
                                       printer):
             filt = [None, None]

reframe/frontend/executors/__init__.py

@@ -215,9 +215,6 @@ def __init__(self, case, listeners=None, timeout=None):
         # if it is zero
         self.ref_count = case.num_dependents
 
-        # Test case has finished, but has not been waited for yet
-        self.zombie = False
-
         # Timestamps for the start and finish phases of the pipeline
         self._timestamps = {}
 
@@ -445,7 +442,6 @@ def run(self):
     def run_complete(self):
         done = self._safe_call(self.check.run_complete)
         if done:
-            self.zombie = True
             self._notify_listeners('on_task_exit')
 
         return done
@@ -461,7 +457,6 @@ def compile_complete(self):
     @logging.time_function
     def run_wait(self):
         self._safe_call(self.check.run_wait)
-        self.zombie = False
 
     @logging.time_function
     def sanity(self):
@@ -501,6 +496,15 @@ def cleanup(self, *args, **kwargs):
         self._safe_call(self.check.cleanup, *args, **kwargs)
 
     def fail(self, exc_info=None, callback='on_task_failure'):
+        def _wait_job(job):
+            if job:
+                with contextlib.suppress(JobNotStartedError):
+                    job.wait()
+
+        # Make sure to properly wait/reap any spawned job in case of failures
+        _wait_job(self.check.build_job)
+        _wait_job(self.check.job)
+
         self._failed_stage = self._current_stage
         self._exc_info = exc_info or sys.exc_info()
         self._notify_listeners(callback)
@@ -527,8 +531,9 @@ def abort(self, cause=None):
         exc.__cause__ = cause
         self._aborted = True
         try:
-            if not self.zombie and self.check.job:
+            if self.check.job:
                 self.check.job.cancel()
+                self.check.job.wait()
         except JobNotStartedError:
             self.fail((type(exc), exc, None), 'on_task_abort')
         except BaseException: