Introduce separate channel for trigger workloads#48835
Closed
gopidesupavan wants to merge 5 commits into
Closed
Conversation
gopidesupavan
requested review from
amoghrajesh,
ashb,
dstandish,
hussein-awala and
kaxil
as code owners
April 5, 2025 14:31
Member
Author
|
Does this approach okay? |
gopidesupavan
commented
Apr 5, 2025
| child_comms, read_msgs = mkpipe() | ||
| child_logs, read_logs = mkpipe() | ||
|
|
||
| if "TriggerRunnerSupervisor" in cls.__name__: |
Member
Author
There was a problem hiding this comment.
Open these sockets only when in triggerer, for others we dont need it.
Member
|
I'll take a closer look at this on Monday |
Member
Author
cool thanks :) |
pierrejeambrun
left a comment
Member
There was a problem hiding this comment.
I just made a couple of nits
Makes sense to me, I'll let Ash have the final word on this.
Comment on lines
+543
to
+547
| """" | ||
| It require special case to handle the workloads and api calls to api-server, due to mixing up messages | ||
| a separate channel is used to send the workloads from parent process to child process the child process. | ||
| connect_stdin will use this channel to read the workloads in read_workload method. | ||
| """ |
Member
There was a problem hiding this comment.
If I remember correctly those should field docstring should go bellow the attr they describe, not above.
| ) | ||
|
|
||
| def _send(self, msg: BaseModel): | ||
| self.trigger_stdin.write(msg.model_dump_json().encode("utf-8") + b"\n") # type: ignore[union-attr] |
Member
There was a problem hiding this comment.
That type ignore seems of. It wasn't necessary before and it seems really similar now:
Member
Author
|
Thanks @pierrejeambrun for review, will be closing this in favour of #48880 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
closes: #48820
realted: #48819 #48747
Why
Introducing a separate channel for trigger workloads.
Currently when multiple tasks are entering into trigger, the messages are mixing up with Trigger Workloads. this is due to the read side when we are trying to read same sys.stdin, causing this problem IMHO. When this mixing happens we are loosing the Trigger Workloads as its read in different place.
What
Created two new sockets to handle Trigger Workloads, when main process writes with
trigger_stdinand in child process this Workloads will be reading from thetrigger_requests_fdHave observed this behaviour while testing DagStateTrigger/WorkflowTrigger
Connections are failing to retrieve in Trigger, when multiple tasks running in tirgger.
After
Retrieved connections properly for all the task and their own connection id, have created a separate connection for each task to verify whether messages are mixing up or not. Looking good.
^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named
{pr_number}.significant.rstor{issue_number}.significant.rst, in airflow-core/newsfragments.