StefanRRichter commented on a change in pull request #8692: [FLINK-12804]
Introduce mailbox-based ExecutorService
URL: https://github.com/apache/flink/pull/8692#discussion_r296607259
##########
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:
This is a private method, that is only there to avoid some code duplication.
No caller uses it directly, and for the two indirect use-cases that are
exposed, they are about two special "signal" letters for which the sematics
should be perfectly fine. But maybe you can go into more detail if you think I
overlook a problem?
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
