# 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)