Support chords with empty headers #4443
Conversation
AlexHill
force-pushed
the
empty_header_chords
branch
2 times, most recently
from
6588867
to
13578e3
Compare
AlexHill
force-pushed
the
empty_header_chords
branch
from
13578e3
to
dd66d85
Compare
|
Quite the radical change we have here. I'd have to review this after coffee :) |
|
Yep, fair enough - the kernel of the changes is that the header is called in |
There was a problem hiding this comment.
I'd rather avoid changing public APIs as this can result in code breakage. So I need you to be 100% sure this won't affect anyone before merging this.
Furthermore, other result backends might be affected by this change so I'm quite surprised why this works with the DynamoDB backend for example.
If we can answer yes to both questions then LGTM.
| def fallback_chord_unlock(self, group_id, body, result=None, | ||
| countdown=1, **kwargs): | ||
| kwargs['result'] = [r.as_tuple() for r in result] | ||
| def fallback_chord_unlock(self, header_result, body, countdown=1, |
There was a problem hiding this comment.
Are we sure this won't break anyone's code?
| # Overrides this to avoid calling GroupResult.save | ||
| # pylint: disable=method-hidden | ||
| # Note that KeyValueStoreBackend.__init__ sets self.apply_chord | ||
| # if the implements_incr attr is set. Redis backend doesn't set | ||
| # this flag. | ||
| options['task_id'] = group_id | ||
| return header(*partial_args, **options or {}) | ||
| pass |
There was a problem hiding this comment.
Are we sure this applies for all cases? That code is there for a reason.
There was a problem hiding this comment.
Yep - apply_chord() is only called from chord.run(), and we're unconditionally calling the header (with the replaced task ID) there now: https://github.com/celery/celery/pull/4443/files#diff-e27381cb5c64031dd28aa6855c78cfddR1277
|
Yeah, I wasn't entirely sure about what's considered a public. If there are third-party backends that out there that override If we can get away with it's a nice simplification - we were calling the header (and replacing the group ID) in three separate places, and having to pass along the arguments everywhere and so on. Now we just call the header in the one place the same way, and the backends use That said if you want to be strict about maintaining these signatures (and the behaviour of calling the header inside each backend's |
|
This fixes a real problem so I'm going to risk it. |
When a chord's header group contains no tasks, its body is never executed. The reason for this is pretty straightforward: execution of the body is triggered by completion of header tasks. If there are no header tasks to be completed, the body is never executed.
An empty
GroupResultis considered completed and successful, so I think it's reasonable to expect a chord with an empty group header to complete.The fix is to run the body "manually" in
chord.run()when there are no header tasks.In addition, I've pulled out the parts of the various
apply_chord()s that actually call the header and changed its signature accordingly, instead calling the header inchord.run()before callingapply_chord()only if there are header tasks.Fixes #4333