Appendix 3: Exceptions thrown inside pipelines

Overview

The goal for Ruffus is that exceptions should just work out-of-the-box without any fuss. This is especially important for exceptions that come from your code which may be raised in a different process. Often multiple parallel operations (jobs or tasks) fail at the same time. Ruffus will forward each of these exceptions with the tracebacks so you can jump straight to the offending line.

This example shows separate exceptions from two jobs running in parallel:

from ruffus import *

@originate(["a.start", "b.start", "c.start", "d.start", "e.start"])
def throw_exceptions_here(output_file):
    raise Exception("OOPS")

pipeline_run(multiprocess = 2)

    >>> pipeline_run(multiprocess = 2)

    ruffus.ruffus_exceptions.RethrownJobError:

    Original exceptions:

        Exception #1
          'exceptions.Exception(OOPS)' raised in ...
           Task = def throw_exceptions_here(...):
           Job  = [None -> b.start]

        Traceback (most recent call last):
          File "/usr/local/lib/python2.7/dist-packages/ruffus/task.py", line 685, in run_pooled_job_without_exceptions
            return_value =  job_wrapper(param, user_defined_work_func, register_cleanup, touch_files_only)
          File "/usr/local/lib/python2.7/dist-packages/ruffus/task.py", line 549, in job_wrapper_output_files
            job_wrapper_io_files(param, user_defined_work_func, register_cleanup, touch_files_only, output_files_only = True)
          File "/usr/local/lib/python2.7/dist-packages/ruffus/task.py", line 504, in job_wrapper_io_files
            ret_val = user_defined_work_func(*(param[1:]))
          File "<stdin>", line 3, in throw_exceptions_here
        Exception: OOPS


        Exception #2
          'exceptions.Exception(OOPS)' raised in ...
           Task = def throw_exceptions_here(...):
           Job  = [None -> a.start]

        Traceback (most recent call last):
          File "/usr/local/lib/python2.7/dist-packages/ruffus/task.py", line 685, in run_pooled_job_without_exceptions
            return_value =  job_wrapper(param, user_defined_work_func, register_cleanup, touch_files_only)
          File "/usr/local/lib/python2.7/dist-packages/ruffus/task.py", line 549, in job_wrapper_output_files
            job_wrapper_io_files(param, user_defined_work_func, register_cleanup, touch_files_only, output_files_only = True)
          File "/usr/local/lib/python2.7/dist-packages/ruffus/task.py", line 504, in job_wrapper_io_files
            ret_val = user_defined_work_func(*(param[1:]))
          File "<stdin>", line 3, in throw_exceptions_here
        Exception: OOPS


.. image:: ../../images/manual_exceptions.png

Pipelines running in parallel accumulate Exceptions

As show above, by default Ruffus accumulates NN exceptions before interrupting the pipeline prematurely where NN is the specified parallelism for pipeline_run(multiprocess = NN)

This seems a fair tradeoff between being able to gather detailed error information for running jobs, and not wasting too much time for a task that is going to fail anyway.

Terminate pipeline immediately upon Exceptions

Set pipeline_run(exceptions_terminate_immediately = True)

To have all exceptions interrupt the pipeline immediately, invoke:

pipeline_run(exceptions_terminate_immediately = True)

For example, with this change, only a single exception will be thrown before the pipeline is interrupted:

from ruffus import *

@originate(["a.start", "b.start", "c.start", "d.start", "e.start"])
def throw_exceptions_here(output_file):
    raise Exception("OOPS")

pipeline_run(multiprocess = 2, exceptions_terminate_immediately = True)

    >>> pipeline_run(multiprocess = 2)

    ruffus.ruffus_exceptions.RethrownJobError:

    Original exception:

        Exception #1
          'exceptions.Exception(OOPS)' raised in ...
           Task = def throw_exceptions_here(...):
           Job  = [None -> a.start]

        Traceback (most recent call last):
          [Tedious traceback snipped out!!!....]
        Exception: OOPS

raise Ruffus.JobSignalledBreak

The same can be accomplished on a finer scale by throwing the Ruffus.JobSignalledBreak Exception. Unlike other exceptions, this causes an immediate halt in pipeline execution. If there are other exceptions in play at that point, they will be rethrown in the main process but no new exceptions will be added.

from ruffus import *

@originate(["a.start", "b.start", "c.start", "d.start", "e.start"])
def throw_exceptions_here(output_file):
    raise JobSignalledBreak("OOPS")

pipeline_run(multiprocess = 2)

Display exceptions as they occur

In the following example, the jobs throw exceptions at two second staggered intervals into the job. With log_exceptions = True, the exceptions are displayed as they occur even though the pipeline continues running.

logger.error(...) will be invoked with the string representation of the each exception, and associated stack trace.

The default logger prints to sys.stderr, but as usual can be changed to any class from the logging module or compatible object via pipeline_run(logger = XXX)

from ruffus import *
import time, os

@originate(["1.start", "2.start", "3.start", "4.start", "5.start"])
def throw_exceptions_here(output_file):
    delay = int(os.path.splitext(output_file)[0])
    time.sleep(delay * 2)
    raise JobSignalledBreak("OOPS")

pipeline_run(log_exceptions = True, multiprocess = 5)