Merge pull request binarymatt#107 from cezarsa/master

Handling crashes and unexpected exits in the child process

Author: binarymatt

pyres/exceptions.py

@@ -1,5 +1,11 @@
 class NoQueueError(Exception):
     pass
 
-class TimeoutError(RuntimeError):
+class JobError(RuntimeError):
+    pass
+
+class TimeoutError(JobError):
+    pass
+
+class CrashError(JobError):
     pass

pyres/worker.py

@@ -6,7 +6,7 @@
 import commands
 import random
 
-from pyres.exceptions import NoQueueError, TimeoutError
+from pyres.exceptions import NoQueueError, JobError, TimeoutError, CrashError
 from pyres.job import Job
 from pyres import ResQ, Stat, __version__
 
@@ -128,8 +128,6 @@ def work(self, interval=5):
         that job to make sure another worker won't run it, then *forks* itself to
         work on that job.
 
-        Finally, the ``process`` method actually processes the job by eventually calling the Job instance's ``perform`` method.
-
         """
         self._setproctitle("Starting")
         self.startup()
@@ -142,63 +140,7 @@ def work(self, interval=5):
             job = self.reserve(interval)
 
             if job:
-                logger.debug('picked up job')
-                logger.debug('job details: %s' % job)
-                self.before_fork(job)
-                self.child = os.fork() 
-                if self.child:
-                    self._setproctitle("Forked %s at %s" %
-                                       (self.child,
-                                        datetime.datetime.now()))
-                    logger.info('Forked %s at %s' % (self.child,
-                                                      datetime.datetime.now()))
-
-                    try:
-                        start = datetime.datetime.now()
-
-                        # waits for the result or times out
-                        while True:
-                            result = os.waitpid(self.child, os.WNOHANG)
-                            if result != (0, 0):
-                                break
-                            time.sleep(0.5)
-
-                            now = datetime.datetime.now()
-                            if self.timeout and ((now - start).seconds > self.timeout):
-                                os.kill(self.child, signal.SIGKILL)
-                                os.waitpid(-1, os.WNOHANG)
-                                raise TimeoutError("Timed out after %d seconds" % self.timeout)
-
-                    except OSError as ose:
-                        import errno
-
-                        if ose.errno != errno.EINTR:
-                            raise ose
-
-                    except TimeoutError as e:
-                        exceptionType, exceptionValue, exceptionTraceback = sys.exc_info()
-                        logger.exception("%s timed out: %s" % (job, e))
-                        job.fail(exceptionTraceback)
-                        self.failed()
-                        self.done_working()
-
-                    logger.debug('done waiting')
-                else:
-                    self._setproctitle("Processing %s since %s" %
-                                       (job._queue,
-                                        datetime.datetime.now()))
-                    logger.info('Processing %s since %s' %
-                                 (job._queue, datetime.datetime.now()))
-                    self.after_fork(job)
-
-                    # re-seed the Python PRNG after forking, otherwise
-                    # all job process will share the same sequence of
-                    # random numbers
-                    random.seed()
-
-                    self.process(job)
-                    os._exit(0)
-                self.child = None
+                self.fork_worker(job)
             else:
                 if interval == 0:
                     break
@@ -207,6 +149,81 @@ def work(self, interval=5):
                 #time.sleep(interval)
         self.unregister_worker()
 
+    def fork_worker(self, job):
+        """Invoked by ``work`` method. ``fork_worker`` does the actual forking to create the child
+        process that will process the job. It's also responsible for monitoring the child process
+        and handling hangs and crashes.
+
+        Finally, the ``process`` method actually processes the job by eventually calling the Job
+        instance's ``perform`` method.
+
+        """
+        logger.debug('picked up job')
+        logger.debug('job details: %s' % job)
+        self.before_fork(job)
+        self.child = os.fork()
+        if self.child:
+            self._setproctitle("Forked %s at %s" %
+                               (self.child,
+                                datetime.datetime.now()))
+            logger.info('Forked %s at %s' % (self.child,
+                                              datetime.datetime.now()))
+
+            try:
+                start = datetime.datetime.now()
+
+                # waits for the result or times out
+                while True:
+                    pid, status = os.waitpid(self.child, os.WNOHANG)
+                    if pid != 0:
+                        if os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0:
+                            break
+                        if os.WIFSTOPPED(status):
+                            logger.warning("Process stopped by signal %d" % os.WSTOPSIG(status))
+                        else:
+                            if os.WIFSIGNALED(status):
+                                raise CrashError("Unexpected exit by signal %d" % os.WTERMSIG(status))
+                            raise CrashError("Unexpected exit status %d" % os.WEXITSTATUS(status))
+
+                    time.sleep(0.5)
+
+                    now = datetime.datetime.now()
+                    if self.timeout and ((now - start).seconds > self.timeout):
+                        os.kill(self.child, signal.SIGKILL)
+                        os.waitpid(-1, os.WNOHANG)
+                        raise TimeoutError("Timed out after %d seconds" % self.timeout)
+
+            except OSError as ose:
+                import errno
+
+                if ose.errno != errno.EINTR:
+                    raise ose
+            except JobError:
+                self._handle_job_exception(job)
+            finally:
+                # If the child process' job called os._exit manually we need to
+                # finish the clean up here.
+                if self.job():
+                    self.done_working()
+
+            logger.debug('done waiting')
+        else:
+            self._setproctitle("Processing %s since %s" %
+                               (job._queue,
+                                datetime.datetime.now()))
+            logger.info('Processing %s since %s' %
+                         (job._queue, datetime.datetime.now()))
+            self.after_fork(job)
+
+            # re-seed the Python PRNG after forking, otherwise
+            # all job process will share the same sequence of
+            # random numbers
+            random.seed()
+
+            self.process(job)
+            os._exit(0)
+        self.child = None
+
     def before_fork(self, job):
         """
         hook for making changes immediately before forking to process
@@ -227,21 +244,33 @@ def before_process(self, job):
     def process(self, job=None):
         if not job:
             job = self.reserve()
+
+        job_failed = False
         try:
-            self.working_on(job)
-            job = self.before_process(job)
-            return job.perform()
-        except Exception, e:
-            exceptionType, exceptionValue, exceptionTraceback = sys.exc_info()
-            logger.exception("%s failed: %s" % (job, e))
-            job.fail(exceptionTraceback)
-            self.failed()
-        else:
-            logger.info('completed job')
-            logger.debug('job details: %s' % job)
+            try:
+                self.working_on(job)
+                job = self.before_process(job)
+                return job.perform()
+            except Exception:
+                job_failed = True
+                self._handle_job_exception(job)
+            except SystemExit, e:
+                if e.code != 0:
+                    job_failed = True
+                    self._handle_job_exception(job)
+
+            if not job_failed:
+                logger.info('completed job')
+                logger.debug('job details: %s' % job)
         finally:
             self.done_working()
 
+    def _handle_job_exception(self, job):
+        exceptionType, exceptionValue, exceptionTraceback = sys.exc_info()
+        logger.exception("%s failed: %s" % (job, exceptionValue))
+        job.fail(exceptionTraceback)
+        self.failed()
+
     def reserve(self, timeout=10):
         logger.debug('checking queues %s' % self.queues)
         job = self.job_class.reserve(self.queues, self.resq, self.__str__(), timeout=timeout)

tests/__init__.py

@@ -46,6 +46,33 @@ def perform(wait_for):
         time.sleep(wait_for)
         return "Done Sleeping"
 
+class CrashJob(object):
+    queue = 'basic'
+
+    @staticmethod
+    def perform():
+        # Dangerous, this will cause a hard crash of the python process
+        import ctypes
+        ctypes.string_at(1)
+        return "Never got here"
+
+class PrematureExitJob(object):
+    queue = 'basic'
+
+    @staticmethod
+    def perform(exit_code):
+        import sys
+        sys.exit(exit_code)
+        return "Never got here"
+
+class PrematureHardExitJob(object):
+    queue = 'basic'
+
+    @staticmethod
+    def perform(exit_code):
+        os._exit(exit_code)
+        return "Never got here"
+
 class TestProcess(object):
     queue = 'high'
 

tests/test_worker.py

@@ -1,4 +1,4 @@
-from tests import PyResTests, Basic, TestProcess, ErrorObject, RetryOnExceptionJob, TimeoutJob
+from tests import PyResTests, Basic, TestProcess, ErrorObject, RetryOnExceptionJob, TimeoutJob, CrashJob, PrematureExitJob, PrematureHardExitJob
 from pyres import ResQ
 from pyres.job import Job
 from pyres.scheduler import Scheduler
@@ -203,21 +203,74 @@ def test_retry_on_exception(self):
         assert worker.get_failed() == 0
 
     def test_kills_stale_workers_after_timeout(self):
-        import signal
         timeout = 1
 
         worker = Worker(['basic'], timeout=timeout)
         self.resq.enqueue(TimeoutJob, timeout + 1)
 
-        child = os.fork()
-        if child:
-            assert worker.get_failed() == 0
-            time.sleep(timeout + 2)
-            os.kill(child, signal.SIGKILL)
-            os.waitpid(-1, os.WNOHANG)
-            assert worker.get_failed() == 1
-        else:
-            worker.work()
+        assert worker.get_failed() == 0
+        worker.fork_worker(worker.reserve())
+        assert worker.get_failed() == 1
+
+    def test_detect_crashed_workers_as_failures(self):
+        worker = Worker(['basic'])
+        self.resq.enqueue(CrashJob)
+
+        assert worker.job() == {}
+        assert worker.get_failed() == 0
+
+        worker.fork_worker(worker.reserve())
+
+        assert worker.job() == {}
+        assert worker.get_failed() == 1
+
+    def test_detect_non_0_sys_exit_as_failure(self):
+        worker = Worker(['basic'])
+        self.resq.enqueue(PrematureExitJob, 9)
+
+        assert worker.job() == {}
+        assert worker.get_failed() == 0
+
+        worker.fork_worker(worker.reserve())
+
+        assert worker.job() == {}
+        assert worker.get_failed() == 1
+
+    def test_detect_code_0_sys_exit_as_success(self):
+        worker = Worker(['basic'])
+        self.resq.enqueue(PrematureExitJob, 0)
+
+        assert worker.job() == {}
+        assert worker.get_failed() == 0
+
+        worker.fork_worker(worker.reserve())
+
+        assert worker.job() == {}
+        assert worker.get_failed() == 0
+
+    def test_detect_non_0_os_exit_as_failure(self):
+        worker = Worker(['basic'])
+        self.resq.enqueue(PrematureHardExitJob, 9)
+
+        assert worker.job() == {}
+        assert worker.get_failed() == 0
+
+        worker.fork_worker(worker.reserve())
+
+        assert worker.job() == {}
+        assert worker.get_failed() == 1
+
+    def test_detect_code_0_os_exit_as_success(self):
+        worker = Worker(['basic'])
+        self.resq.enqueue(PrematureHardExitJob, 0)
+
+        assert worker.job() == {}
+        assert worker.get_failed() == 0
+
+        worker.fork_worker(worker.reserve())
+
+        assert worker.job() == {}
+        assert worker.get_failed() == 0
 
     def test_retries_give_up_eventually(self):
         now = datetime.datetime.now()