Detect if the GridEngine worker thread has crashed to prevent hanging the workflow

src/toil/batchSystems/abstractGridEngineBatchSystem.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
+import queue
 import time
 from abc import ABCMeta, abstractmethod
 from datetime import datetime
@@ -45,20 +46,21 @@ class AbstractGridEngineBatchSystem(BatchSystemCleanupSupport):
  A partial implementation of BatchSystemSupport for batch systems run on a
  standard HPC cluster. By default auto-deployment is not implemented.
  """
+ class GridEngineThreadException(Exception):
+ pass
 
- class Worker(Thread, metaclass=ABCMeta):
-
+ class GridEngineThread(Thread, metaclass=ABCMeta):
  def __init__(self, newJobsQueue: Queue, updatedJobsQueue: Queue, killQueue: Queue, killedJobsQueue: Queue, boss: 'AbstractGridEngineBatchSystem') -> None:
  """
- Abstract worker interface class. All instances are created with five
+ Abstract thread interface class. All instances are created with five
  initial arguments (below). Note the Queue instances passed are empty.
 
  :param newJobsQueue: a Queue of new (unsubmitted) jobs
  :param updatedJobsQueue: a Queue of jobs that have been updated
  :param killQueue: a Queue of active jobs that need to be killed
- :param killedJobsQueue: Queue of killed jobs for this worker
+ :param killedJobsQueue: Queue of killed jobs for this thread
  :param boss: the AbstractGridEngineBatchSystem instance that
- controls this AbstractGridEngineWorker
+ controls this GridEngineThread
 
  """
  Thread.__init__(self)
@@ -77,6 +79,7 @@ def __init__(self, newJobsQueue: Queue, updatedJobsQueue: Queue, killQueue: Queu
  self.batchJobIDs: Dict[int, str] = dict()
  self._checkOnJobsCache = None
  self._checkOnJobsTimestamp = None
+ self.exception = None
 
  def getBatchSystemID(self, jobID: int) -> str:
  """
@@ -110,7 +113,7 @@ def createJobs(self, newJob: JobTuple) -> bool:
  """
  Create a new job with the given attributes.
 
- Implementation-specific; called by AbstractGridEngineWorker.run()
+ Implementation-specific; called by GridEngineThread.run()
  """
  activity = False
  # Load new job id if present:
@@ -146,7 +149,7 @@ def createJobs(self, newJob: JobTuple) -> bool:
 
  def killJobs(self):
  """
- Kill any running jobs within worker
+ Kill any running jobs within thread
  """
  killList = list()
  while True:
@@ -277,14 +280,17 @@ def run(self):
  while self._runStep():
  pass
  except Exception as ex:
- logger.error("GridEngine like batch system failure", exc_info=ex)
- raise
+ self.exception = ex
+ logger.error("GridEngine like batch system failure: %s", ex)
+ # dont raise exception as is_alive will still be set to false,
+ # signalling exception in the thread as we expect the thread to
+ # always be running for the duration of the workflow
 
  def coalesce_job_exit_codes(self, batch_job_id_list: list) -> List[Union[int, Tuple[int, Optional[BatchJobExitReason]], None]]:
  """
  Returns exit codes and possibly exit reasons for a list of jobs, or None if they are running.
 
- Called by AbstractGridEngineWorker.checkOnJobs().
+ Called by GridEngineThread.checkOnJobs().
 
  This is an optional part of the interface. It should raise
  NotImplementedError if not actually implemented for a particular
@@ -345,7 +351,7 @@ def getRunningJobIDs(self):
  def killJob(self, jobID):
  """
  Kill specific job with the Toil job ID. Implementation-specific; called
- by AbstractGridEngineWorker.killJobs()
+ by GridEngineThread.killJobs()
 
  :param string jobID: Toil job ID
  """
@@ -360,7 +366,7 @@ def getJobExitCode(self, batchJobID) -> Union[int, Tuple[int, Optional[BatchJobE
 
  If the job is not running but the exit code is not available, it
  will be EXIT_STATUS_UNAVAILABLE_VALUE. Implementation-specific;
- called by AbstractGridEngineWorker.checkOnJobs().
+ called by GridEngineThread.checkOnJobs().
 
  The exit code will only be 0 if the job affirmatively succeeded.
 
@@ -379,10 +385,10 @@ def __init__(self, config, maxCores, maxMemory, maxDisk):
  self.updatedJobsQueue = Queue()
  self.killQueue = Queue()
  self.killedJobsQueue = Queue()
- # get the associated worker class here
- self.worker = self.Worker(self.newJobsQueue, self.updatedJobsQueue,
- self.killQueue, self.killedJobsQueue, self)
- self.worker.start()
+ # get the associated thread class here
+ self.background_thread = self.GridEngineThread(self.newJobsQueue, self.updatedJobsQueue,
+  self.killQueue, self.killedJobsQueue, self)
+ self.background_thread.start()
  self._getRunningBatchJobIDsTimestamp = None
  self._getRunningBatchJobIDsCache = {}
 
@@ -428,7 +434,12 @@ def killBatchJobs(self, jobIDs):
  for jobID in jobIDs:
  self.killQueue.put(jobID)
  while jobIDs:
- killedJobId = self.killedJobsQueue.get()
+ try:
+ killedJobId = self.killedJobsQueue.get(timeout=10)
+ except queue.Empty:
+ if not self.background_thread.is_alive():
+ raise self.GridEngineThreadException("Grid engine thread failed unexpectedly") from self.background_thread.exception
+ continue
  if killedJobId is None:
  break
  jobIDs.remove(killedJobId)
@@ -460,14 +471,19 @@ def getRunningBatchJobIDs(self):
  self.config.statePollingWait):
  batchIds = self._getRunningBatchJobIDsCache
  else:
- batchIds = self.with_retries(self.worker.getRunningJobIDs)
+ batchIds = self.with_retries(self.background_thread.getRunningJobIDs)
  self._getRunningBatchJobIDsCache = batchIds
  self._getRunningBatchJobIDsTimestamp = datetime.now()
  batchIds.update(self.getRunningLocalJobIDs())
  return batchIds
 
  def getUpdatedBatchJob(self, maxWait):
  local_tuple = self.getUpdatedLocalJob(0)
+
+ if not self.background_thread.is_alive():
+ # kill remaining jobs on the thread
+ self.background_thread.killJobs()
+ raise self.GridEngineThreadException("Unexpected GridEngineThread failure") from self.background_thread.exception
  if local_tuple:
  return local_tuple
  else:
@@ -481,14 +497,14 @@ def getUpdatedBatchJob(self, maxWait):
 
  def shutdown(self) -> None:
  """
- Signals worker to shutdown (via sentinel) then cleanly joins the thread
+ Signals thread to shutdown (via sentinel) then cleanly joins the thread
  """
  self.shutdownLocal()
  newJobsQueue = self.newJobsQueue
  self.newJobsQueue = None
 
  newJobsQueue.put(None)
- self.worker.join()
+ self.background_thread.join()
 
  def setEnv(self, name, value=None):
  if value and ',' in value:

src/toil/batchSystems/gridengine.py

@@ -28,7 +28,7 @@
 
 class GridEngineBatchSystem(AbstractGridEngineBatchSystem):
 
- class Worker(AbstractGridEngineBatchSystem.Worker):
+ class GridEngineThread(AbstractGridEngineBatchSystem.GridEngineThread):
  """
  Grid Engine-specific AbstractGridEngineWorker methods
  """

src/toil/batchSystems/htcondor.py

@@ -48,7 +48,7 @@
 class HTCondorBatchSystem(AbstractGridEngineBatchSystem):
  # When using HTCondor, the Schedd handles scheduling
 
- class Worker(AbstractGridEngineBatchSystem.Worker):
+ class GridEngineThread(AbstractGridEngineBatchSystem.GridEngineThread):
 
  # Override the createJobs method so that we can use htcondor.Submit objects
  # and so that we can get disk allocation requests and ceil the CPU request.

src/toil/batchSystems/lsf.py

@@ -44,8 +44,8 @@
 
 class LSFBatchSystem(AbstractGridEngineBatchSystem):
 
- class Worker(AbstractGridEngineBatchSystem.Worker):
- """LSF specific AbstractGridEngineWorker methods."""
+ class GridEngineThread(AbstractGridEngineBatchSystem.GridEngineThread):
+ """LSF specific GridEngineThread methods."""
 
  def getRunningJobIDs(self):
  times = {}

src/toil/batchSystems/slurm.py

@@ -70,7 +70,7 @@
 
 class SlurmBatchSystem(AbstractGridEngineBatchSystem):
 
- class Worker(AbstractGridEngineBatchSystem.Worker):
+ class GridEngineThread(AbstractGridEngineBatchSystem.GridEngineThread):
 
  def getRunningJobIDs(self):
  # Should return a dictionary of Job IDs and number of seconds
@@ -135,7 +135,7 @@ def submitJob(self, subLine):
  logger.debug("sbatch submitted job %d", result)
  return result
  except OSError as e:
- logger.error("sbatch command failed")
+ logger.error(f"sbatch command failed with error: {e}")
  raise e
 
  def coalesce_job_exit_codes(self, batch_job_id_list: list) -> List[Union[int, Tuple[int, Optional[BatchJobExitReason]], None]]:

src/toil/batchSystems/torque.py

@@ -31,7 +31,7 @@
 class TorqueBatchSystem(AbstractGridEngineBatchSystem):
 
  # class-specific Worker
- class Worker(AbstractGridEngineBatchSystem.Worker):
+ class GridEngineThread(AbstractGridEngineBatchSystem.GridEngineThread):
  def __init__(
  self, newJobsQueue, updatedJobsQueue, killQueue, killedJobsQueue, boss
  ):

src/toil/lib/retry.py

@@ -572,6 +572,7 @@ def old_retry(
  >>> i
  1
  """
+ timeout = timeout if timeout else DEFAULT_TIMEOUT
  if timeout > 0:
  go = [ None ]
 

src/toil/test/batchSystems/test_slurm.py

@@ -193,7 +193,7 @@ class SlurmTest(ToilTest):
 
  def setUp(self):
  self.monkeypatch = pytest.MonkeyPatch()
- self.worker = toil.batchSystems.slurm.SlurmBatchSystem.Worker(
+ self.worker = toil.batchSystems.slurm.SlurmBatchSystem.GridEngineThread(
  newJobsQueue=Queue(),
  updatedJobsQueue=Queue(),
  killQueue=Queue(),