Fix attribute forwarding for tasks with dynamic dependencies #2478

riga · 2018-07-30T12:34:30Z

Description

This PR fixes the forwarding of attributes (mostly callbacks such as set_status_message) from TaskProcesses to running Tasks that yield dynamic dependencies within their run() method.

Motivation and Context

Currently, the attributes are forwarded and reset like this:

luigi/luigi/worker.py

Lines 137 to 146 in 7d2c557

    
           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)

However, when the run method returns a generator, attributes are reset again before the yielded dependencies are handled (right below the referenced code). Actually, this should happen right before _run_get_new_deps returns. I added a context-managed method that handles the attributes so I could place the (only) call to _run_get_new_deps in that context.

Have you tested this?

I added a new test case that checks if attributes are forwarded for tasks that don't yield dependencies and for those who do. The TaskForwardedAttributesTest.test_yielding_task test would actually fail in the current implementation, but passes with the proposed change.

dlstadther · 2018-07-30T13:45:02Z

test/task_forwarded_attributes_test.py

+    def test_non_yielding_task(self):
+        task = self.run_task(NonYieldingTask())
+
+        self.assertEqual(task.attributes_while_running, forwarded_attributes)


Am i correct that this test passes before AND after the code changes?

Yes, that's correct.

dlstadther · 2018-07-30T13:45:15Z

test/task_forwarded_attributes_test.py

+        task = self.run_task(YieldingTask())
+
+        self.assertEqual(task.attributes_before_yield, forwarded_attributes)
+        self.assertEqual(task.attributes_after_yield, forwarded_attributes)


But this one would fail before this code change?

dlstadther

I think this makes sense to me.

Could we get at least one more person to sign off on this?

Tarrasch · 2018-07-30T21:17:20Z

test/task_forwarded_attributes_test.py

+
+        self.attributes_after_yield = self.gather_forwarded_attributes()
+
+        RunOnceTask.run(self)


This is quite unintuitive, maybe add a comment?

Tarrasch · 2018-07-30T21:18:12Z

test/task_forwarded_attributes_test.py

+import luigi.worker
+
+
+forwarded_attributes = set(luigi.worker.TaskProcess.forward_reporter_attributes.values())


make this constant in capital letters?

Tarrasch · 2018-07-30T21:18:48Z

luigi/worker.py

+
+        try:
+            yield self
+


You can remove the white-space here and the love above I think.

…rwarded_task_attributes.

riga · 2018-08-08T13:44:28Z

Anything else I can do here? =)

dlstadther · 2018-08-08T13:50:06Z

Nah, LGTM. All reviewer comments have been addressed.

* upstream-master: (82 commits) S3 client refactor (spotify#2482) Rename to rpc_log_retries, and make it apply to all the logging involved Factor log_exceptions into a configuration parameter Fix attribute forwarding for tasks with dynamic dependencies (spotify#2478) Add a visiblity level for luigi.Parameters (spotify#2278) Add support for multiple requires and inherits arguments (spotify#2475) Add metadata columns to the RDBMS contrib (spotify#2440) Fix race condition in luigi.lock.acquire_for (spotify#2357) (spotify#2477) tests: Use RunOnceTask where possible (spotify#2476) Optional TOML configs support (spotify#2457) Added default port behaviour for Redshift (spotify#2474) Add codeowners file with default and specific example (spotify#2465) Add Data Revenue to the `blogged` list (spotify#2472) Fix Scheduler.add_task to overwrite accepts_messages attribute. (spotify#2469) Use task_id comparison in Task.__eq__. (spotify#2462) Add stale config Move github templates to .github dir Fix transfer config import (spotify#2458) Additions to provide support for the Load Sharing Facility (LSF) job scheduler (spotify#2373) Version 2.7.6 ...

* upstream-master: S3 client refactor (spotify#2482) Rename to rpc_log_retries, and make it apply to all the logging involved Factor log_exceptions into a configuration parameter Fix attribute forwarding for tasks with dynamic dependencies (spotify#2478) Add a visiblity level for luigi.Parameters (spotify#2278) Add support for multiple requires and inherits arguments (spotify#2475) Add metadata columns to the RDBMS contrib (spotify#2440) Fix race condition in luigi.lock.acquire_for (spotify#2357) (spotify#2477) tests: Use RunOnceTask where possible (spotify#2476) Optional TOML configs support (spotify#2457) Added default port behaviour for Redshift (spotify#2474) Add codeowners file with default and specific example (spotify#2465) Add Data Revenue to the `blogged` list (spotify#2472)

* upstream-master: Remove long-deprecated scheduler config variable alternatives (spotify#2491) Bump tornado milestone version (spotify#2490) Update moto to 1.x milestone version (spotify#2471) Use passed password when create a redis connection (spotify#2489) S3 client refactor (spotify#2482) Rename to rpc_log_retries, and make it apply to all the logging involved Factor log_exceptions into a configuration parameter Fix attribute forwarding for tasks with dynamic dependencies (spotify#2478) Add a visiblity level for luigi.Parameters (spotify#2278) Add support for multiple requires and inherits arguments (spotify#2475) Add metadata columns to the RDBMS contrib (spotify#2440) Fix race condition in luigi.lock.acquire_for (spotify#2357) (spotify#2477) tests: Use RunOnceTask where possible (spotify#2476) Optional TOML configs support (spotify#2457) Added default port behaviour for Redshift (spotify#2474) Add codeowners file with default and specific example (spotify#2465) Add Data Revenue to the `blogged` list (spotify#2472) Fix Scheduler.add_task to overwrite accepts_messages attribute. (spotify#2469) Use task_id comparison in Task.__eq__. (spotify#2462)

riga added 2 commits July 30, 2018 14:16

Add test case to check fowarded task attributes.

d5c9765

Fix attribute forwarding for tasks yielding deps.

05158a7

riga requested review from dlstadther, Tarrasch and ulzha as code owners July 30, 2018 12:34

dlstadther reviewed Jul 30, 2018

View reviewed changes

dlstadther approved these changes Jul 30, 2018

View reviewed changes

Tarrasch approved these changes Jul 30, 2018

View reviewed changes

riga added 2 commits July 31, 2018 11:23

Address review comments on code style and comments.

e70c98d

Merge branch 'master' of https://github.com/spotify/luigi into fix/fo…

a3ac5d1

…rwarded_task_attributes.

dlstadther approved these changes Jul 31, 2018

View reviewed changes

dlstadther merged commit de4629a into spotify:master Aug 8, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix attribute forwarding for tasks with dynamic dependencies #2478

Fix attribute forwarding for tasks with dynamic dependencies #2478

riga commented Jul 30, 2018 •

edited

Loading

dlstadther Jul 30, 2018

riga Jul 30, 2018

dlstadther Jul 30, 2018

riga Jul 30, 2018

dlstadther left a comment

Tarrasch Jul 30, 2018

riga Jul 31, 2018

Tarrasch Jul 30, 2018

riga Jul 31, 2018

Tarrasch Jul 30, 2018

riga Jul 31, 2018

riga commented Aug 8, 2018

dlstadther commented Aug 8, 2018

	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)


		self.attributes_after_yield = self.gather_forwarded_attributes()

		RunOnceTask.run(self)

		import luigi.worker


		forwarded_attributes = set(luigi.worker.TaskProcess.forward_reporter_attributes.values())

Fix attribute forwarding for tasks with dynamic dependencies #2478

Fix attribute forwarding for tasks with dynamic dependencies #2478

Conversation

riga commented Jul 30, 2018 • edited Loading

Description

Motivation and Context

Have you tested this?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlstadther left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

riga commented Aug 8, 2018

dlstadther commented Aug 8, 2018

riga commented Jul 30, 2018 •

edited

Loading