This is an automated email from the ASF dual-hosted git repository. pnowojski pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push: new 77cd1cd72fb [FLINK-35528][task] Skip execution of interruptible mails when yielding 77cd1cd72fb is described below commit 77cd1cd72fbfa369d6f298d09b72dea8613774aa Author: Piotr Nowojski <piotr.nowoj...@gmail.com> AuthorDate: Thu May 23 13:04:44 2024 +0200 [FLINK-35528][task] Skip execution of interruptible mails when yielding When operators are yielding, for example waiting for async state access to complete before a checkpoint, it would be beneficial to not execute interruptible mails. Otherwise continuation mail for firing timers would be continuously re-enqeueed. To achieve that MailboxExecutor must be aware which mails are interruptible. The easiest way to achieve this is to set MIN_PRIORITY for interruptible mails. --- .../api/common/operators/MailOptionsImpl.java | 37 +++++++++++++ .../api/common/operators/MailboxExecutor.java | 64 ++++++++++++++++++++++ .../AsyncExecutionControllerTest.java | 1 + .../flink/runtime/mailbox/SyncMailboxExecutor.java | 1 + .../api/operators/MailboxWatermarkProcessor.java | 1 + .../streaming/runtime/tasks/mailbox/Mail.java | 13 ++++- .../runtime/tasks/mailbox/MailboxExecutorImpl.java | 8 ++- .../streaming/api/functions/PrintSinkTest.java | 1 + .../operators/MailboxWatermarkProcessorTest.java | 9 ++- .../tasks/mailbox/MailboxExecutorImplTest.java | 20 +++++++ 10 files changed, 152 insertions(+), 3 deletions(-) diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/MailOptionsImpl.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/MailOptionsImpl.java new file mode 100644 index 00000000000..3bb5b771309 --- /dev/null +++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/MailOptionsImpl.java @@ -0,0 +1,37 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.api.common.operators; + +import org.apache.flink.annotation.Internal; + +/** Options to configure behaviour of executing mailbox mails. */ +@Internal +public class MailOptionsImpl implements MailboxExecutor.MailOptions { + static final MailboxExecutor.MailOptions DEFAULT = new MailOptionsImpl(false); + static final MailboxExecutor.MailOptions DEFERRABLE = new MailOptionsImpl(true); + + private final boolean deferrable; + + private MailOptionsImpl(boolean deferrable) { + this.deferrable = deferrable; + } + + public boolean isDeferrable() { + return deferrable; + } +} diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/MailboxExecutor.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/MailboxExecutor.java index 639ed18b673..79e17931a3d 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/operators/MailboxExecutor.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/MailboxExecutor.java @@ -86,6 +86,26 @@ public interface MailboxExecutor { /** A constant for empty args to save on object allocation. */ Object[] EMPTY_ARGS = new Object[0]; + /** Extra options to configure enqueued mails. */ + @PublicEvolving + interface MailOptions { + static MailOptions options() { + return MailOptionsImpl.DEFAULT; + } + + /** + * Mark this mail as deferrable. + * + * <p>Runtime can decide to defer execution of deferrable mails. For example, to unblock + * subtask thread as quickly as possible, deferrable mails are not executed during {@link + * #yield()} or {@link #tryYield()}. This is done to speed up checkpointing, by skipping + * execution of potentially long-running mails. + */ + static MailOptions deferrable() { + return MailOptionsImpl.DEFERRABLE; + } + } + /** * Executes the given command at some time in the future in the mailbox thread. * @@ -110,6 +130,49 @@ public interface MailboxExecutor { * The description may contain placeholder that refer to the provided description arguments * using {@link java.util.Formatter} syntax. The actual description is only formatted on demand. * + * @param mailOptions additional options to configure behaviour of the {@code command} + * @param command the runnable task to add to the mailbox for execution. + * @param description the optional description for the command that is used for debugging and + * error-reporting. + * @throws RejectedExecutionException if this task cannot be accepted for execution, e.g. + * because the mailbox is quiesced or closed. + */ + default void execute( + MailOptions mailOptions, + ThrowingRunnable<? extends Exception> command, + String description) { + execute(mailOptions, command, description, EMPTY_ARGS); + } + + /** + * Executes the given command at some time in the future in the mailbox thread. + * + * <p>An optional description can (and should) be added to ease debugging and error-reporting. + * The description may contain placeholder that refer to the provided description arguments + * using {@link java.util.Formatter} syntax. The actual description is only formatted on demand. + * + * @param command the runnable task to add to the mailbox for execution. + * @param descriptionFormat the optional description for the command that is used for debugging + * and error-reporting. + * @param descriptionArgs the parameters used to format the final description string. + * @throws RejectedExecutionException if this task cannot be accepted for execution, e.g. + * because the mailbox is quiesced or closed. + */ + default void execute( + ThrowingRunnable<? extends Exception> command, + String descriptionFormat, + Object... descriptionArgs) { + execute(MailOptions.options(), command, descriptionFormat, descriptionArgs); + } + + /** + * Executes the given command at some time in the future in the mailbox thread. + * + * <p>An optional description can (and should) be added to ease debugging and error-reporting. + * The description may contain placeholder that refer to the provided description arguments + * using {@link java.util.Formatter} syntax. The actual description is only formatted on demand. + * + * @param mailOptions additional options to configure behaviour of the {@code command} * @param command the runnable task to add to the mailbox for execution. * @param descriptionFormat the optional description for the command that is used for debugging * and error-reporting. @@ -118,6 +181,7 @@ public interface MailboxExecutor { * because the mailbox is quiesced or closed. */ void execute( + MailOptions mailOptions, ThrowingRunnable<? extends Exception> command, String descriptionFormat, Object... descriptionArgs); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java index f99dd877cf7..e98a66b82a0 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java @@ -824,6 +824,7 @@ class AsyncExecutionControllerTest { @Override public void execute( + MailOptions mailOptions, ThrowingRunnable<? extends Exception> command, String descriptionFormat, Object... descriptionArgs) { diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/mailbox/SyncMailboxExecutor.java b/flink-runtime/src/test/java/org/apache/flink/runtime/mailbox/SyncMailboxExecutor.java index 13ab803a385..5b862849938 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/mailbox/SyncMailboxExecutor.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/mailbox/SyncMailboxExecutor.java @@ -25,6 +25,7 @@ import org.apache.flink.util.function.ThrowingRunnable; public class SyncMailboxExecutor implements MailboxExecutor { @Override public void execute( + MailOptions mailOptions, ThrowingRunnable<? extends Exception> command, String descriptionFormat, Object... descriptionArgs) { diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java index 02cfd966c1c..76eabe6dec1 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java @@ -77,6 +77,7 @@ public class MailboxWatermarkProcessor<OUT> { progressWatermarkScheduled = true; // We still have work to do, but we need to let other mails to be processed first. mailboxExecutor.execute( + MailboxExecutor.MailOptions.deferrable(), () -> { progressWatermarkScheduled = false; emitWatermarkInsideMailbox(); diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java index bed96ae559a..6afd8918e3d 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java @@ -18,6 +18,8 @@ package org.apache.flink.streaming.runtime.tasks.mailbox; import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.operators.MailOptionsImpl; +import org.apache.flink.api.common.operators.MailboxExecutor; import org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor; import org.apache.flink.util.Preconditions; import org.apache.flink.util.function.ThrowingRunnable; @@ -30,6 +32,7 @@ import java.util.concurrent.Future; */ @Internal public class Mail { + private final MailOptionsImpl mailOptions; /** The action to execute. */ private final ThrowingRunnable<? extends Exception> runnable; /** @@ -50,6 +53,7 @@ public class Mail { String descriptionFormat, Object... descriptionArgs) { this( + MailboxExecutor.MailOptions.options(), runnable, priority, StreamTaskActionExecutor.IMMEDIATE, @@ -58,11 +62,13 @@ public class Mail { } public Mail( + MailboxExecutor.MailOptions mailOptions, ThrowingRunnable<? extends Exception> runnable, int priority, StreamTaskActionExecutor actionExecutor, String descriptionFormat, Object... descriptionArgs) { + this.mailOptions = (MailOptionsImpl) mailOptions; this.runnable = Preconditions.checkNotNull(runnable); this.priority = priority; this.descriptionFormat = @@ -71,8 +77,13 @@ public class Mail { this.actionExecutor = actionExecutor; } + public MailboxExecutor.MailOptions getMailOptions() { + return mailOptions; + } + public int getPriority() { - return priority; + /** See {@link MailboxExecutor.MailOptions#deferrable()} ()}. */ + return mailOptions.isDeferrable() ? TaskMailbox.MIN_PRIORITY : priority; } public void tryCancel(boolean mayInterruptIfRunning) { 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 b2c8eac935a..2987ee21cca 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 @@ -67,13 +67,19 @@ public final class MailboxExecutorImpl implements MailboxExecutor { @Override public void execute( + MailOptions mailOptions, final ThrowingRunnable<? extends Exception> command, final String descriptionFormat, final Object... descriptionArgs) { try { mailbox.put( new Mail( - command, priority, actionExecutor, descriptionFormat, descriptionArgs)); + mailOptions, + command, + priority, + actionExecutor, + descriptionFormat, + descriptionArgs)); } catch (MailboxClosedException mbex) { throw new RejectedExecutionException(mbex); } diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkTest.java index 88e0bf3bf46..bc92ac15184 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkTest.java @@ -235,6 +235,7 @@ class PrintSinkTest { @Override public void execute( + MailOptions mailOptions, ThrowingRunnable<? extends Exception> command, String descriptionFormat, Object... descriptionArgs) {} diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessorTest.java index 9fb9d348525..87f19f4032d 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessorTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessorTest.java @@ -35,6 +35,7 @@ import org.junit.jupiter.api.Test; import java.util.ArrayList; import java.util.List; +import java.util.Optional; import static org.assertj.core.api.Assertions.assertThat; @@ -43,6 +44,7 @@ class MailboxWatermarkProcessorTest { @Test void testEmitWatermarkInsideMailbox() throws Exception { + int priority = 42; final List<StreamElement> emittedElements = new ArrayList<>(); final TaskMailboxImpl mailbox = new TaskMailboxImpl(); final InternalTimeServiceManager<?> timerService = new NoOpInternalTimeServiceManager(); @@ -50,7 +52,8 @@ class MailboxWatermarkProcessorTest { final MailboxWatermarkProcessor<StreamRecord<String>> watermarkProcessor = new MailboxWatermarkProcessor<>( new CollectorOutput<>(emittedElements), - new MailboxExecutorImpl(mailbox, 0, StreamTaskActionExecutor.IMMEDIATE), + new MailboxExecutorImpl( + mailbox, priority, StreamTaskActionExecutor.IMMEDIATE), timerService); final List<Watermark> expectedOutput = new ArrayList<>(); watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(1)); @@ -69,6 +72,10 @@ class MailboxWatermarkProcessorTest { assertThat(emittedElements).containsExactlyElementsOf(expectedOutput); + // FLINK-35528: do not allow yielding to continuation mails + assertThat(mailbox.tryTake(priority)).isEqualTo(Optional.empty()); + assertThat(emittedElements).containsExactlyElementsOf(expectedOutput); + while (mailbox.hasMail()) { mailbox.take(TaskMailbox.MIN_PRIORITY).run(); } 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 cfeb92b2aa1..1b6fddc7d91 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 @@ -110,6 +110,26 @@ class MailboxExecutorImplTest { assertThat(wasExecuted).isTrue(); } + @Test + void testDeferrable() throws Exception { + int priority = 42; + MailboxExecutor localExecutor = mailboxProcessor.getMailboxExecutor(priority); + + AtomicBoolean deferrableMailExecuted = new AtomicBoolean(); + + localExecutor.execute( + MailboxExecutor.MailOptions.deferrable(), + () -> deferrableMailExecuted.set(true), + "deferrable mail"); + assertThat(localExecutor.tryYield()).isFalse(); + assertThat(deferrableMailExecuted.get()).isFalse(); + assertThat(mailboxExecutor.tryYield()).isFalse(); + assertThat(deferrableMailExecuted.get()).isFalse(); + + assertThat(mailboxProcessor.runMailboxStep()).isTrue(); + assertThat(deferrableMailExecuted.get()).isTrue(); + } + @Test void testClose() throws Exception { final TestRunnable yieldRun = new TestRunnable();