Fix attribute forwarding for tasks with dynamic dependencies (#2478)

* Add test case to check fowarded task attributes.

* Fix attribute forwarding for tasks yielding deps.

* Address review comments on code style and comments.

luigi/worker.py

@@ -37,6 +37,7 @@
 import signal
 import subprocess
 import sys
+import contextlib
 
 try:
     import Queue
@@ -135,16 +136,8 @@ def __init__(self, task, worker_id, result_queue, status_reporter,
         self.check_unfulfilled_deps = check_unfulfilled_deps
 
     def _run_get_new_deps(self):
-        # forward some attributes before running
-        for reporter_attr, task_attr in six.iteritems(self.forward_reporter_attributes):
-            setattr(self.task, task_attr, getattr(self.status_reporter, reporter_attr))
-
         task_gen = self.task.run()
 
-        # reset attributes again
-        for reporter_attr, task_attr in six.iteritems(self.forward_reporter_attributes):
-            setattr(self.task, task_attr, None)
-
         if not isinstance(task_gen, types.GeneratorType):
             return None
 
@@ -202,7 +195,8 @@ def run(self):
                     expl = 'Task is an external data dependency ' \
                         'and data does not exist (yet?).'
             else:
-                new_deps = self._run_get_new_deps()
+                with self._forward_attributes():
+                    new_deps = self._run_get_new_deps()
                 status = DONE if not new_deps else PENDING
 
             if new_deps:
@@ -258,6 +252,18 @@ def terminate(self):
         except ImportError:
             return super(TaskProcess, self).terminate()
 
+    @contextlib.contextmanager
+    def _forward_attributes(self):
+        # forward configured attributes to the task
+        for reporter_attr, task_attr in six.iteritems(self.forward_reporter_attributes):
+            setattr(self.task, task_attr, getattr(self.status_reporter, reporter_attr))
+        try:
+            yield self
+        finally:
+            # reset attributes again
+            for reporter_attr, task_attr in six.iteritems(self.forward_reporter_attributes):
+                setattr(self.task, task_attr, None)
+
 
 # This code and the task_process_context config key currently feels a bit ad-hoc.
 # Discussion on generalizing it into a plugin system: https://github.com/spotify/luigi/issues/1897

test/task_forwarded_attributes_test.py

@@ -0,0 +1,85 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright 2012-2015 Spotify AB
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from helpers import LuigiTestCase, RunOnceTask
+
+import luigi
+import luigi.scheduler
+import luigi.worker
+
+
+FORWARDED_ATTRIBUTES = set(luigi.worker.TaskProcess.forward_reporter_attributes.values())
+
+
+class NonYieldingTask(RunOnceTask):
+
+    # need to accept messages in order for the "scheduler_message" attribute to be not None
+    accepts_messages = True
+
+    def gather_forwarded_attributes(self):
+        """
+        Returns a set of names of attributes that are forwarded by the TaskProcess and that are not
+        *None*. The tests in this file check if and which attributes are present at different times,
+        e.g. while running, or before and after a dynamic dependency was yielded.
+        """
+        attrs = set()
+        for attr in FORWARDED_ATTRIBUTES:
+            if getattr(self, attr, None) is not None:
+                attrs.add(attr)
+        return attrs
+
+    def run(self):
+        # store names of forwarded attributes which are only available within the run method
+        self.attributes_while_running = self.gather_forwarded_attributes()
+
+        # invoke the run method of the RunOnceTask which marks this task as complete
+        RunOnceTask.run(self)
+
+
+class YieldingTask(NonYieldingTask):
+
+    def run(self):
+        # as TaskProcess._run_get_new_deps handles generators in a specific way, store names of
+        # forwarded attributes before and after yielding a dynamic dependency, so we can explicitely
+        # validate the attribute forwarding implementation
+        self.attributes_before_yield = self.gather_forwarded_attributes()
+        yield RunOnceTask()
+        self.attributes_after_yield = self.gather_forwarded_attributes()
+
+        # invoke the run method of the RunOnceTask which marks this task as complete
+        RunOnceTask.run(self)
+
+
+class TaskForwardedAttributesTest(LuigiTestCase):
+
+    def run_task(self, task):
+        sch = luigi.scheduler.Scheduler()
+        with luigi.worker.Worker(scheduler=sch) as w:
+            w.add(task)
+            w.run()
+        return task
+
+    def test_non_yielding_task(self):
+        task = self.run_task(NonYieldingTask())
+
+        self.assertEqual(task.attributes_while_running, FORWARDED_ATTRIBUTES)
+
+    def test_yielding_task(self):
+        task = self.run_task(YieldingTask())
+
+        self.assertEqual(task.attributes_before_yield, FORWARDED_ATTRIBUTES)
+        self.assertEqual(task.attributes_after_yield, FORWARDED_ATTRIBUTES)