[
https://issues.apache.org/jira/browse/FLINK-16174?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17039990#comment-17039990
]
Haibo Sun edited comment on FLINK-16174 at 2/19/20 12:57 PM:
-------------------------------------------------------------
CC [~pnowojski], [~roman_khachatryan]
was (Author: sunhaibotb):
CC [~pnowojski][~roman_khachatryan]
> Add a better tryYield() method to MailboxExecutor to return the lowest
> priority of the remaining mails
> ------------------------------------------------------------------------------------------------------
>
> Key: FLINK-16174
> URL: https://issues.apache.org/jira/browse/FLINK-16174
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Task
> Affects Versions: 1.10.0
> Reporter: Haibo Sun
> Priority: Major
>
> Currently, we use chainIndex as the priority to create MailboxExecutor to
> process its mails. When MailboxExecutor#tryYield is called to process mails,
> it will take the mails of this operator and all downstream operators in the
> chain. But sometimes, after calling MailboxExecutor#tryYield, we need to know
> whether there is any mail of the current operator in the mailbox, which can
> simplify some operations.
> For example, when we close a operator in runtime, after quiescing the
> processing time service and waiting for its running timers to finish, if
> there is no mail of the current operator in the mailbox, we call
> StreamOperator#close to close the operator. Then the runtime code of closing
> a operator can be simplified as follows.
> {code:java}
> quiesceProcessingTimeService().get();
> while (mailboxExecuto.betterTryYield() <= self.priority) {}
> closeOperator(actionExecutor);
> {code}
> With the existing #tryYield method, if the following simplified code is used
> to close a operator, then when a downstream operator is implemented like
> MailboxOperatorTest.ReplicatingMail, the tryyield() loop will
> be prevented from exiting, which results deadlock.
> {code:java}
> quiesceProcessingTimeService().get();
> while (mailboxExecuto.tryYield()) {}
> closeOperator(actionExecutor);
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
