1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
|
.. include:: ../../global.inc
.. include:: manual_chapter_numbers.inc
.. index::
pair: exceptions; Tutorial
.. _new_manual.exceptions:
###################################################################################################
|new_manual.exceptions.chapter_num|: 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:
.. code-block:: python
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)
.. code-block:: pycon
:emphasize-lines: 5, 21
>>> 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
.. _new_manual.exceptions.multiple_errors:
.. index:: signalling, interrupts, break, errors, exceptions, multiple errors
****************************************************************
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 :ref:`pipeline_run(multiprocess = NN) <pipeline_functions.pipeline_run>`
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 :ref:`pipeline_run(exceptions_terminate_immediately = True) <pipeline_functions.pipeline_run>`
==============================================================================================================================
To have all exceptions interrupt the pipeline immediately, invoke:
.. code-block:: python
pipeline_run(exceptions_terminate_immediately = True)
For example, with this change, only a single exception will be thrown before the pipeline is interrupted:
.. code-block:: python
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)
.. code-block:: pycon
:emphasize-lines: 5, 21
>>> 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.
.. code-block:: python
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
:ref:`pipeline_run(logger = XXX) <pipeline_functions.pipeline_run>`
.. code-block:: python
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)
|
