1u0 commented on a change in pull request #8692: [FLINK-12804] Introduce
mailbox-based ExecutorService
URL: https://github.com/apache/flink/pull/8692#discussion_r296621778
##########
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:
I agree for this being used only for signal letters and hopefully that
`ActionContet` is used only internally.
By "weird" I mean that there can be situations at runtime, where the caller,
for example that uses `ActionContext.actionsUnavailable()` may be blocked (from
time-to-time) + it may execute a letter.
Imaginable example:
```java
// In some sub class of StreamTask
protected void performDefaultAction(ActionConext actionContext) {
acquireResourceA();
actionContext.actionsUnavailable();
releaseA();
}
```
In this case, an execution can happen in different orders:
```
acquireResourceA
releaseA
// continue in the mailbox loop
```
OR
```
acquireResourceA
// execute some letter from the mailbox
releaseA
```
----------------------------------------------------------------
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
