You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2020/06/02 17:42:43 UTC
[flink] branch release-1.11 updated (6551b6b -> e514c71)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 6551b6b [FLINK-18012] Deactivate slot timeout when calling TaskSlotTable.tryMarkSlotActive
new f5ad2a3 [FLINK-16057][hotfix][task] Remove MailboxProcessor.mainMailboxExecutor field
new ad998f8 [FLINK-16057][task] Optimize ContinuousFileReaderOperator
new e514c71 [FLINK-16057][task] Optimize TaskMailbox state retrieval
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../source/ContinuousFileReaderOperator.java | 25 ++++++++++++----------
.../flink/streaming/runtime/tasks/StreamTask.java | 11 ++++++----
.../runtime/tasks/mailbox/MailboxExecutorImpl.java | 13 +++++++++++
.../runtime/tasks/mailbox/MailboxProcessor.java | 19 ++--------------
.../runtime/tasks/mailbox/TaskMailbox.java | 11 +++++++++-
.../runtime/tasks/mailbox/TaskMailboxImpl.java | 3 +++
.../tasks/mailbox/MailboxExecutorImplTest.java | 2 +-
7 files changed, 50 insertions(+), 34 deletions(-)
[flink] 01/03: [FLINK-16057][hotfix][task] Remove
MailboxProcessor.mainMailboxExecutor field
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f5ad2a3eb6005bcac17594c522dd68868e025468
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Tue Jun 2 09:19:14 2020 +0200
[FLINK-16057][hotfix][task] Remove MailboxProcessor.mainMailboxExecutor field
Currently, the field is unnecessary; further adding MailboxProcessor
field to MailboxExecutor creates cyclic dependency
---
.../flink/streaming/runtime/tasks/StreamTask.java | 11 +++++++----
.../runtime/tasks/mailbox/MailboxProcessor.java | 17 +----------------
.../runtime/tasks/mailbox/MailboxExecutorImplTest.java | 2 +-
3 files changed, 9 insertions(+), 21 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index 7f0011f..7ebb9e5 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -207,6 +207,8 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
protected final MailboxProcessor mailboxProcessor;
+ final MailboxExecutor mainMailboxExecutor;
+
/**
* TODO it might be replaced by the global IO executor on TaskManager level future.
*/
@@ -276,6 +278,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
this.actionExecutor = Preconditions.checkNotNull(actionExecutor);
this.mailboxProcessor = new MailboxProcessor(this::processInput, mailbox, actionExecutor);
this.mailboxProcessor.initMetric(environment.getMetricGroup());
+ this.mainMailboxExecutor = mailboxProcessor.getMainMailboxExecutor();
this.asyncExceptionHandler = new StreamTaskAsyncExceptionHandler(environment);
this.asyncOperationsThreadPool = Executors.newCachedThreadPool(
new ExecutorThreadFactory("AsyncOperations", uncaughtExceptionHandler));
@@ -498,7 +501,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
}
// Note that we must request partition after all the single gates finished recovery.
- CompletableFuture.allOf(futures).thenRun(() -> mailboxProcessor.getMainMailboxExecutor().execute(
+ CompletableFuture.allOf(futures).thenRun(() -> mainMailboxExecutor.execute(
this::requestPartitions, "Input gates request partitions"));
}
}
@@ -779,7 +782,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
boolean advanceToEndOfEventTime) {
CompletableFuture<Boolean> result = new CompletableFuture<>();
- mailboxProcessor.getMainMailboxExecutor().execute(
+ mainMailboxExecutor.execute(
() -> {
try {
result.complete(triggerCheckpoint(checkpointMetaData, checkpointOptions, advanceToEndOfEventTime));
@@ -831,7 +834,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
if (mailboxProcessor.isMailboxThread()) {
runnable.run();
} else {
- mailboxProcessor.getMainMailboxExecutor().execute(runnable, descriptionFormat, descriptionArgs);
+ mainMailboxExecutor.execute(runnable, descriptionFormat, descriptionArgs);
}
}
@@ -967,7 +970,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
@Override
public void dispatchOperatorEvent(OperatorID operator, SerializedValue<OperatorEvent> event) throws FlinkException {
try {
- mailboxProcessor.getMainMailboxExecutor().execute(
+ mainMailboxExecutor.execute(
() -> {
try {
operatorChain.dispatchOperatorEvent(operator, event);
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java
index b315e4c..a955c2b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java
@@ -71,9 +71,6 @@ public class MailboxProcessor implements Closeable {
/** Action that is repeatedly executed if no action request is in the mailbox. Typically record processing. */
protected final MailboxDefaultAction mailboxDefaultAction;
- /** A pre-created instance of mailbox executor that executes all mails. */
- private final MailboxExecutor mainMailboxExecutor;
-
/** Control flag to terminate the mailbox loop. Must only be accessed from mailbox thread. */
private boolean mailboxLoopRunning;
@@ -102,27 +99,15 @@ public class MailboxProcessor implements Closeable {
MailboxDefaultAction mailboxDefaultAction,
TaskMailbox mailbox,
StreamTaskActionExecutor actionExecutor) {
- this(mailboxDefaultAction, actionExecutor, mailbox, new MailboxExecutorImpl(mailbox, MIN_PRIORITY, actionExecutor));
- }
-
- public MailboxProcessor(
- MailboxDefaultAction mailboxDefaultAction,
- StreamTaskActionExecutor actionExecutor,
- TaskMailbox mailbox,
- MailboxExecutor mainMailboxExecutor) {
this.mailboxDefaultAction = Preconditions.checkNotNull(mailboxDefaultAction);
this.actionExecutor = Preconditions.checkNotNull(actionExecutor);
this.mailbox = Preconditions.checkNotNull(mailbox);
- this.mainMailboxExecutor = Preconditions.checkNotNull(mainMailboxExecutor);
this.mailboxLoopRunning = true;
this.suspendedDefaultAction = null;
}
- /**
- * Returns a pre-created executor service that executes all mails.
- */
public MailboxExecutor getMainMailboxExecutor() {
- return mainMailboxExecutor;
+ return new MailboxExecutorImpl(mailbox, MIN_PRIORITY, actionExecutor);
}
/**
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java
index efd70d9..ab40979 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java
@@ -60,7 +60,7 @@ public class MailboxExecutorImplTest {
this.mailbox = new TaskMailboxImpl();
this.mailboxExecutor = new MailboxExecutorImpl(mailbox, DEFAULT_PRIORITY, StreamTaskActionExecutor.IMMEDIATE);
this.otherThreadExecutor = Executors.newSingleThreadScheduledExecutor();
- this.mailboxProcessor = new MailboxProcessor(c -> { }, StreamTaskActionExecutor.IMMEDIATE, mailbox, mailboxExecutor);
+ this.mailboxProcessor = new MailboxProcessor(c -> { }, mailbox, StreamTaskActionExecutor.IMMEDIATE);
}
@After
[flink] 03/03: [FLINK-16057][task] Optimize TaskMailbox state
retrieval
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e514c71b09550e0335bebf0cca3abac26b5d79c7
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Thu May 28 10:41:56 2020 +0200
[FLINK-16057][task] Optimize TaskMailbox state retrieval
Don't lock in TaskMailboxImpl.getState for task thread.
This makes ContinuousFileReaderOperator about 29% faster.
---
.../apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java | 3 +++
1 file changed, 3 insertions(+)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java
index 99a63a4..adeb356 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxImpl.java
@@ -311,6 +311,9 @@ public class TaskMailboxImpl implements TaskMailbox {
@Nonnull
@Override
public State getState() {
+ if (isMailboxThread()) {
+ return state;
+ }
final ReentrantLock lock = this.lock;
lock.lock();
try {
[flink] 02/03: [FLINK-16057][task] Optimize
ContinuousFileReaderOperator
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ad998f8759932d33a19a621e943a09e420ce8828
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Fri May 29 18:10:22 2020 +0200
[FLINK-16057][task] Optimize ContinuousFileReaderOperator
Current approach of re-enqueuing mails creates an overhead
visible in benchmarks. This change eliminates unnecessary
re-enqueueing of mails by checking mailboxExecutor.isIdle.
---
.../source/ContinuousFileReaderOperator.java | 25 ++++++++++++----------
.../runtime/tasks/mailbox/MailboxExecutorImpl.java | 13 +++++++++++
.../runtime/tasks/mailbox/MailboxProcessor.java | 2 +-
.../runtime/tasks/mailbox/TaskMailbox.java | 11 +++++++++-
4 files changed, 38 insertions(+), 13 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java
index c53049a..87a028c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java
@@ -37,6 +37,7 @@ import org.apache.flink.streaming.api.operators.StreamSourceContexts;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.function.RunnableWithException;
@@ -205,7 +206,7 @@ public class ContinuousFileReaderOperator<OUT, T extends TimestampedInputSplit>
private transient InputFormat<OUT, ? super T> format;
private TypeSerializer<OUT> serializer;
- private transient MailboxExecutor executor;
+ private transient MailboxExecutorImpl executor;
private transient OUT reusedRecord;
private transient SourceFunction.SourceContext<OUT> sourceContext;
private transient ListState<T> checkpointedState;
@@ -233,7 +234,7 @@ public class ContinuousFileReaderOperator<OUT, T extends TimestampedInputSplit>
this.format = checkNotNull(format);
this.processingTimeService = checkNotNull(processingTimeService);
- this.executor = checkNotNull(mailboxExecutor);
+ this.executor = (MailboxExecutorImpl) checkNotNull(mailboxExecutor);
}
@Override
@@ -311,17 +312,19 @@ public class ContinuousFileReaderOperator<OUT, T extends TimestampedInputSplit>
}
private void processRecord() throws IOException {
- if (!state.prepareToProcessRecord(this)) {
- return;
- }
+ do {
+ if (!state.prepareToProcessRecord(this)) {
+ return;
+ }
- readAndCollectRecord();
+ readAndCollectRecord();
- if (format.reachedEnd()) {
- onSplitProcessed();
- } else {
- enqueueProcessRecord();
- }
+ if (format.reachedEnd()) {
+ onSplitProcessed();
+ return;
+ }
+ } while (executor.isIdle()); // todo: consider moving this loop into MailboxProcessor (return boolean "re-execute" from enqueued action)
+ enqueueProcessRecord();
}
private void onSplitProcessed() throws IOException {
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImpl.java
index 2f99bec..48b57d8 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImpl.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImpl.java
@@ -43,10 +43,22 @@ public final class MailboxExecutorImpl implements MailboxExecutor {
private final StreamTaskActionExecutor actionExecutor;
+ private final MailboxProcessor mailboxProcessor;
+
public MailboxExecutorImpl(@Nonnull TaskMailbox mailbox, int priority, StreamTaskActionExecutor actionExecutor) {
+ this(mailbox, priority, actionExecutor, null);
+ }
+
+ public MailboxExecutorImpl(@Nonnull TaskMailbox mailbox, int priority, StreamTaskActionExecutor actionExecutor, MailboxProcessor mailboxProcessor) {
this.mailbox = mailbox;
this.priority = priority;
this.actionExecutor = Preconditions.checkNotNull(actionExecutor);
+ this.mailboxProcessor = mailboxProcessor;
+ }
+
+ public boolean isIdle() {
+ return !mailboxProcessor.isMailboxLoopRunning() ||
+ (mailboxProcessor.isDefaultActionUnavailable() && !mailbox.hasMail() && mailbox.getState().isAcceptingMails());
}
@Override
@@ -85,4 +97,5 @@ public final class MailboxExecutorImpl implements MailboxExecutor {
return false;
}
}
+
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java
index a955c2b..2180971 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java
@@ -116,7 +116,7 @@ public class MailboxProcessor implements Closeable {
* @param priority the priority of the {@link MailboxExecutor}.
*/
public MailboxExecutor getMailboxExecutor(int priority) {
- return new MailboxExecutorImpl(mailbox, priority, actionExecutor);
+ return new MailboxExecutorImpl(mailbox, priority, actionExecutor, this);
}
public void initMetric(TaskMetricGroup metricGroup) {
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailbox.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailbox.java
index 5e9fb85..8928525 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailbox.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailbox.java
@@ -173,7 +173,16 @@ public interface TaskMailbox {
* This enum represents the states of the mailbox lifecycle.
*/
enum State {
- OPEN, QUIESCED, CLOSED
+ OPEN(true), QUIESCED(false), CLOSED(false);
+ private final boolean acceptingMails;
+
+ State(boolean acceptingMails) {
+ this.acceptingMails = acceptingMails;
+ }
+
+ public boolean isAcceptingMails() {
+ return acceptingMails;
+ }
}
/**