1u0 commented on a change in pull request #8692: [FLINK-12804] Introduce
mailbox-based ExecutorService
URL: https://github.com/apache/flink/pull/8692#discussion_r296615421
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -1329,33 +1364,71 @@ public void
tryHandleCheckpointException(CheckpointMetaData checkpointMetaData,
*/
public final class ActionContext {
- private final Runnable actionUnavailableLetter =
ThrowingRunnable.unchecked(mailbox::waitUntilHasMail);
+ private final Runnable actionUnavailableLetter;
+ private final Mailbox mailbox;
+ private final Thread mailboxThread;
+
+ public ActionContext(Mailbox mailbox) {
+ this(mailbox, Thread.currentThread());
+ }
+
+ public ActionContext(Mailbox mailbox, Thread mailboxThread) {
+ this.actionUnavailableLetter =
ThrowingRunnable.unchecked(mailbox::waitUntilHasMail);
+ this.mailbox = mailbox;
+ this.mailboxThread = mailboxThread;
+ }
/**
* This method must be called to end the stream task when all
actions for the tasks have been performed.
*/
public void allActionsCompleted() {
- mailbox.clearAndPut(POISON_LETTER);
+ try {
+ if (Thread.currentThread() == mailboxThread) {
+ if
(!mailbox.tryPutFirst(mailboxPoisonLetter)) {
+ // mailbox is full - in this
particular case we know for sure that we will still run through the
+ // break condition check inside
the mailbox loop and so we can just run directly.
+ mailboxPoisonLetter.run();
Review comment:
I'm reading commit version `d235a0d8b3ca2d60c91165fe3c3625437063392e`.
What I mean in my comment, that why wouldn't you just use:
```java
if (mailboxExecutorService.isMailboxThread()) {
mailboxPoisonLetter.run();
} else {
taskMailboxExecutor.getMailbox().putFirst(mailboxPoisonLetter);
}
```
?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
