Skip to content

Commit 784256d

Browse files
authored
Refactor 4 different renderings of task description (#3743)
Prior to this PR, there were 4 variant ways to render a future (which is usually but not always a Parsl task) as a text string: 1. describing failure of inner future in a join_app with 1 future to join - rendered task ID if the inner future has a task record attribute, without checking it is a parsl.dataflow.taskrecord.TaskRecord or that that attribute contains a task id entry. - otherwise render as None 2. describing failure of inner future in a join_app with a list of futures to join - is meant to do the same as case 1, but is buggy and always renders as None (the wrong object is checked for having a task_record attribute) 3. describing failure of a dependency future - rendered task ID if there was a task_record, and the future is from the same DFK as is rendering the code (so that a task number from a different DFK is not rendered, in the corner case of a task from one DFK being used as a dependency for a task in another DFK) - otherwise, renders the repr of the future 4. describing which dependencies will be waited on when submitting a task to the DFK - rendered task ID if the future is an instance of AppFuture or DataFuture - otherwise renders the repr of the future This PR makes a single render method. It is a member of the DFK, because rendering is done in the context of the DFK: two parsl task AppFutures will be rendered differently if they are from this DFK or a different DFK. This PR implements render_future_description which tries to combine all of the above: - if the future is an AppFuture, and is from the same DFK, render the task ID - otherwise render the repr of the future # Changed Behaviour human readable descriptions of dependencies/join inner futures will be changed # Fixes Fixes # (issue) ## Type of change Choose which options apply, and delete the ones which do not apply. - Bug fix - New feature - Update to human readable text: Documentation/error messages/comments - Code maintenance/cleanup
1 parent c3091a0 commit 784256d

File tree

4 files changed

+43
-28
lines changed

4 files changed

+43
-28
lines changed

parsl/dataflow/dflow.py

Lines changed: 17 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -484,24 +484,18 @@ def handle_join_update(self, task_record: TaskRecord, inner_app_future: Optional
484484

485485
# now we know each joinable Future is done
486486
# so now look for any exceptions
487-
exceptions_tids: List[Tuple[BaseException, Optional[str]]]
487+
exceptions_tids: List[Tuple[BaseException, str]]
488488
exceptions_tids = []
489489
if isinstance(joinable, Future):
490490
je = joinable.exception()
491491
if je is not None:
492-
if hasattr(joinable, 'task_record'):
493-
tid = joinable.task_record['id']
494-
else:
495-
tid = None
492+
tid = self.render_future_description(joinable)
496493
exceptions_tids = [(je, tid)]
497494
elif isinstance(joinable, list):
498495
for future in joinable:
499496
je = future.exception()
500497
if je is not None:
501-
if hasattr(joinable, 'task_record'):
502-
tid = joinable.task_record['id']
503-
else:
504-
tid = None
498+
tid = self.render_future_description(future)
505499
exceptions_tids.append((je, tid))
506500
else:
507501
raise TypeError(f"Unknown joinable type {type(joinable)}")
@@ -918,13 +912,7 @@ def _unwrap_futures(self, args: Sequence[Any], kwargs: Dict[str, Any]) \
918912
dep_failures = []
919913

920914
def append_failure(e: Exception, dep: Future) -> None:
921-
# If this Future is associated with a task inside this DFK,
922-
# then refer to the task ID.
923-
# Otherwise make a repr of the Future object.
924-
if hasattr(dep, 'task_record') and dep.task_record['dfk'] == self:
925-
tid = "task " + repr(dep.task_record['id'])
926-
else:
927-
tid = repr(dep)
915+
tid = self.render_future_description(dep)
928916
dep_failures.extend([(e, tid)])
929917

930918
# Replace item in args
@@ -1076,10 +1064,7 @@ def submit(self,
10761064

10771065
depend_descs = []
10781066
for d in depends:
1079-
if isinstance(d, AppFuture) or isinstance(d, DataFuture):
1080-
depend_descs.append("task {}".format(d.tid))
1081-
else:
1082-
depend_descs.append(repr(d))
1067+
depend_descs.append(self.render_future_description(d))
10831068

10841069
if depend_descs != []:
10851070
waiting_message = "waiting on {}".format(", ".join(depend_descs))
@@ -1438,6 +1423,18 @@ def default_std_autopath(self, taskrecord, kw):
14381423
'' if label is None else '_{}'.format(label),
14391424
kw))
14401425

1426+
def render_future_description(self, dep: Future) -> str:
1427+
"""Renders a description of the future in the context of the
1428+
current DFK.
1429+
"""
1430+
if isinstance(dep, AppFuture) and dep.task_record['dfk'] == self:
1431+
tid = "task " + repr(dep.task_record['id'])
1432+
elif isinstance(dep, DataFuture):
1433+
tid = "DataFuture from task " + repr(dep.tid)
1434+
else:
1435+
tid = repr(dep)
1436+
return tid
1437+
14411438

14421439
class DataFlowKernelLoader:
14431440
"""Manage which DataFlowKernel is active.

parsl/tests/test_python_apps/test_dep_standard_futures.py

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,3 +43,6 @@ def test_future_fail_dependency():
4343
# Future, plain_fut, somewhere in its str
4444

4545
assert repr(plain_fut) in str(ex)
46+
assert len(ex.dependent_exceptions_tids) == 1
47+
assert isinstance(ex.dependent_exceptions_tids[0][0], ValueError)
48+
assert ex.dependent_exceptions_tids[0][1].startswith("<Future ")

parsl/tests/test_python_apps/test_fail.py

Lines changed: 17 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -27,17 +27,26 @@ def test_no_deps():
2727
pass
2828

2929

30-
@pytest.mark.parametrize("fail_probs", ((1, 0), (0, 1)))
31-
def test_fail_sequence(fail_probs):
32-
"""Test failure in a sequence of dependencies
30+
def test_fail_sequence_first():
31+
t1 = random_fail(fail_prob=1)
32+
t2 = random_fail(fail_prob=0, inputs=[t1])
33+
t_final = random_fail(fail_prob=0, inputs=[t2])
3334

34-
App1 -> App2 ... -> AppN
35-
"""
35+
with pytest.raises(DependencyError):
36+
t_final.result()
3637

37-
t1_fail_prob, t2_fail_prob = fail_probs
38-
t1 = random_fail(fail_prob=t1_fail_prob)
39-
t2 = random_fail(fail_prob=t2_fail_prob, inputs=[t1])
38+
assert len(t_final.exception().dependent_exceptions_tids) == 1
39+
assert isinstance(t_final.exception().dependent_exceptions_tids[0][0], DependencyError)
40+
assert t_final.exception().dependent_exceptions_tids[0][1].startswith("task ")
41+
42+
43+
def test_fail_sequence_middle():
44+
t1 = random_fail(fail_prob=0)
45+
t2 = random_fail(fail_prob=1, inputs=[t1])
4046
t_final = random_fail(fail_prob=0, inputs=[t2])
4147

4248
with pytest.raises(DependencyError):
4349
t_final.result()
50+
51+
assert len(t_final.exception().dependent_exceptions_tids) == 1
52+
assert isinstance(t_final.exception().dependent_exceptions_tids[0][0], ManufacturedTestFailure)

parsl/tests/test_python_apps/test_join.py

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,7 +97,10 @@ def test_error():
9797
f = outer_error()
9898
e = f.exception()
9999
assert isinstance(e, JoinError)
100+
101+
assert len(e.dependent_exceptions_tids) == 1
100102
assert isinstance(e.dependent_exceptions_tids[0][0], InnerError)
103+
assert e.dependent_exceptions_tids[0][1].startswith("task ")
101104

102105

103106
def test_two_errors():
@@ -109,10 +112,12 @@ def test_two_errors():
109112
de0 = e.dependent_exceptions_tids[0][0]
110113
assert isinstance(de0, InnerError)
111114
assert de0.args[0] == "Error A"
115+
assert e.dependent_exceptions_tids[0][1].startswith("task ")
112116

113117
de1 = e.dependent_exceptions_tids[1][0]
114118
assert isinstance(de1, InnerError)
115119
assert de1.args[0] == "Error B"
120+
assert e.dependent_exceptions_tids[1][1].startswith("task ")
116121

117122

118123
def test_one_error_one_result():
@@ -125,6 +130,7 @@ def test_one_error_one_result():
125130
de0 = e.dependent_exceptions_tids[0][0]
126131
assert isinstance(de0, InnerError)
127132
assert de0.args[0] == "Error A"
133+
assert e.dependent_exceptions_tids[0][1].startswith("task ")
128134

129135

130136
@join_app

0 commit comments

Comments
 (0)