1u0 commented on a change in pull request #8692: [FLINK-12804] Introduce
mailbox-based ExecutorService
URL: https://github.com/apache/flink/pull/8692#discussion_r292870984
##########
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();
+ }
+ } else {
+ mailbox.putFirst(mailboxPoisonLetter);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (MailboxStateException me) {
+ LOG.debug("Action context could not submit
poison letter to mailbox.", me);
+ }
}
/**
* Calling this method signals that the mailbox-thread should
continue invoking the default action, e.g. because
* new input became available for processing.
- *
- * @throws InterruptedException on interruption.
*/
- public void actionsAvailable() throws InterruptedException {
- mailbox.putMail(DEFAULT_ACTION_AVAILABLE);
+ public void actionsAvailable() {
+ putOrExecuteDirectly(DEFAULT_ACTION_AVAILABLE);
}
/**
* Calling this method signals that the mailbox-thread should
(temporarily) stop invoking the default action,
* e.g. because there is currently no input available.
- *
- * @throws InterruptedException on interruption.
*/
- public void actionsUnavailable() throws InterruptedException {
- mailbox.putMail(actionUnavailableLetter);
+ public void actionsUnavailable() {
+ putOrExecuteDirectly(actionUnavailableLetter);
+ }
+
+ private void putOrExecuteDirectly(Runnable letter) {
+ try {
+ if (Thread.currentThread() == mailboxThread) {
+ if (!mailbox.tryPutMail(letter)) {
+ letter.run();
Review comment:
Imo, this has weird semantics and may lead to some bugs if the callers of
those methods not aware that a letter can be executed directly as normal
function call.
It makes things harder to reason about.
----------------------------------------------------------------
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
