You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2018/03/09 10:48:23 UTC
flink git commit: [FLINK-8856] [TaskManager] Move all cancellation
interrupt calls to TaskCanceller thread
Repository: flink
Updated Branches:
refs/heads/release-1.4 72b3ddad1 -> d57d4e0af
[FLINK-8856] [TaskManager] Move all cancellation interrupt calls to TaskCanceller thread
This cleans up the code and guards against a JVM bug where 'interrupt()' calls
block/deadlock if the thread is engaged in certain I/O operations.
In addition, this makes sure that the process really goes away when the cancellation
timeout expires, rather than relying on the TaskManager to be able to properly handle
the fatal error notification.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d57d4e0a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d57d4e0a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d57d4e0a
Branch: refs/heads/release-1.4
Commit: d57d4e0af0c28c85d880a8261aac90e4a07eda32
Parents: 72b3dda
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Mar 6 15:18:33 2018 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Mar 8 19:07:47 2018 +0100
----------------------------------------------------------------------
.../apache/flink/runtime/taskmanager/Task.java | 282 ++++++++++++-------
1 file changed, 180 insertions(+), 102 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d57d4e0a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
index 2cb356c..ae8cb47 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
@@ -69,6 +69,7 @@ import org.apache.flink.runtime.jobmanager.PartitionProducerDisposedException;
import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.runtime.metrics.groups.TaskMetricGroup;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
+import org.apache.flink.runtime.util.FatalExitExceptionHandler;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedValue;
@@ -78,7 +79,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
import java.io.IOException;
import java.net.URL;
@@ -93,11 +93,11 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
@@ -1080,15 +1080,51 @@ public class Task implements Runnable, TaskActions {
invokable,
executingThread,
taskNameWithSubtask,
- taskCancellationInterval,
- taskCancellationTimeout,
- taskManagerActions,
producedPartitions,
inputGates);
- Thread cancelThread = new Thread(executingThread.getThreadGroup(), canceler,
+
+ Thread cancelThread = new Thread(
+ executingThread.getThreadGroup(),
+ canceler,
String.format("Canceler for %s (%s).", taskNameWithSubtask, executionId));
cancelThread.setDaemon(true);
+ cancelThread.setUncaughtExceptionHandler(FatalExitExceptionHandler.INSTANCE);
cancelThread.start();
+
+ // the periodic interrupting thread - a different thread than the canceller, in case
+ // the application code does blocking stuff in its cancellation paths.
+ Runnable interrupter = new TaskInterrupter(
+ LOG,
+ executingThread,
+ taskNameWithSubtask,
+ taskCancellationInterval);
+
+ Thread interruptingThread = new Thread(
+ executingThread.getThreadGroup(),
+ interrupter,
+ String.format("Canceler/Interrupts for %s (%s).", taskNameWithSubtask, executionId));
+ interruptingThread.setDaemon(true);
+ interruptingThread.setUncaughtExceptionHandler(FatalExitExceptionHandler.INSTANCE);
+ interruptingThread.start();
+
+ // if a cancellation timeout is set, the watchdog thread kills the process
+ // if graceful cancellation does not succeed
+ if (taskCancellationTimeout > 0) {
+ Runnable cancelWatchdog = new TaskCancelerWatchDog(
+ executingThread,
+ taskManagerActions,
+ taskCancellationTimeout,
+ LOG);
+
+ Thread watchDogThread = new Thread(
+ executingThread.getThreadGroup(),
+ cancelWatchdog,
+ String.format("Cancellation Watchdog for %s (%s).",
+ taskNameWithSubtask, executionId));
+ watchDogThread.setDaemon(true);
+ watchDogThread.setUncaughtExceptionHandler(FatalExitExceptionHandler.INSTANCE);
+ watchDogThread.start();
+ }
}
return;
}
@@ -1406,9 +1442,29 @@ public class Task implements Runnable, TaskActions {
return String.format("%s (%s) [%s]", taskNameWithSubtask, executionId, executionState);
}
+ // ------------------------------------------------------------------------
+ // Task cancellation
+ //
+ // The task cancellation uses in total three threads, as a safety net
+ // against various forms of user- and JVM bugs.
+ //
+ // - The first thread calls 'cancel()' on the invokable and closes
+ // the input and output connections, for fast thread termination
+ // - The second thread periodically interrupts the invokable in order
+ // to pull the thread out of blocking wait and I/O operations
+ // - The third thread (watchdog thread) waits until the cancellation
+ // timeout and then performs a hard cancel (kill process, or let
+ // the TaskManager know)
+ //
+ // Previously, thread two and three were in one thread, but we needed
+ // to separate this to make sure the watchdog thread does not call
+ // 'interrupt()'. This is a workaround for the following JVM bug
+ // https://bugs.java.com/bugdatabase/view_bug.do?bug_id=8138622
+ // ------------------------------------------------------------------------
+
/**
- * This runner calls cancel() on the invokable and periodically interrupts the
- * thread until it has terminated.
+ * This runner calls cancel() on the invokable, closes input-/output resources,
+ * and initially interrupts the task thread.
*/
private static class TaskCanceler implements Runnable {
@@ -1419,27 +1475,11 @@ public class Task implements Runnable, TaskActions {
private final ResultPartition[] producedPartitions;
private final SingleInputGate[] inputGates;
- /** Interrupt interval. */
- private final long interruptInterval;
-
- /** Timeout after which a fatal error notification happens. */
- private final long interruptTimeout;
-
- /** TaskManager to notify about a timeout */
- private final TaskManagerActions taskManager;
-
- /** Watch Dog thread */
- @Nullable
- private final Thread watchDogThread;
-
public TaskCanceler(
Logger logger,
AbstractInvokable invokable,
Thread executer,
String taskName,
- long cancellationInterval,
- long cancellationTimeout,
- TaskManagerActions taskManager,
ResultPartition[] producedPartitions,
SingleInputGate[] inputGates) {
@@ -1447,39 +1487,19 @@ public class Task implements Runnable, TaskActions {
this.invokable = invokable;
this.executer = executer;
this.taskName = taskName;
- this.interruptInterval = cancellationInterval;
- this.interruptTimeout = cancellationTimeout;
- this.taskManager = taskManager;
this.producedPartitions = producedPartitions;
this.inputGates = inputGates;
-
- if (cancellationTimeout > 0) {
- // The watch dog repeatedly interrupts the executor until
- // the cancellation timeout kicks in (at which point the
- // task manager is notified about a fatal error) or the
- // executor has terminated.
- this.watchDogThread = new Thread(
- executer.getThreadGroup(),
- new TaskCancelerWatchDog(),
- "WatchDog for " + taskName + " cancellation");
- this.watchDogThread.setDaemon(true);
- } else {
- this.watchDogThread = null;
- }
}
@Override
public void run() {
try {
- if (watchDogThread != null) {
- watchDogThread.start();
- }
-
// the user-defined cancel method may throw errors.
// we need do continue despite that
try {
invokable.cancel();
} catch (Throwable t) {
+ ExceptionUtils.rethrowIfFatalError(t);
logger.error("Error while canceling the task {}.", taskName, t);
}
@@ -1494,6 +1514,7 @@ public class Task implements Runnable, TaskActions {
try {
partition.destroyBufferPool();
} catch (Throwable t) {
+ ExceptionUtils.rethrowIfFatalError(t);
LOG.error("Failed to release result partition buffer pool for task {}.", taskName, t);
}
}
@@ -1502,89 +1523,146 @@ public class Task implements Runnable, TaskActions {
try {
inputGate.releaseAllResources();
} catch (Throwable t) {
+ ExceptionUtils.rethrowIfFatalError(t);
LOG.error("Failed to release input gate for task {}.", taskName, t);
}
}
- // interrupt the running thread initially
+ // send the initial interruption signal
executer.interrupt();
- try {
- executer.join(interruptInterval);
- }
- catch (InterruptedException e) {
- // we can ignore this
- }
-
- if (watchDogThread != null) {
- watchDogThread.interrupt();
- watchDogThread.join();
- }
- } catch (Throwable t) {
+ }
+ catch (Throwable t) {
+ ExceptionUtils.rethrowIfFatalError(t);
logger.error("Error in the task canceler for task {}.", taskName, t);
}
}
+ }
- /**
- * Watchdog for the cancellation. If the task is stuck in cancellation,
- * we notify the task manager about a fatal error.
- */
- private class TaskCancelerWatchDog implements Runnable {
+ /**
+ * This thread sends the delayed, periodic interrupt calls to the executing thread.
+ */
+ private static final class TaskInterrupter implements Runnable {
- @Override
- public void run() {
- long intervalNanos = TimeUnit.NANOSECONDS.convert(interruptInterval, TimeUnit.MILLISECONDS);
- long timeoutNanos = TimeUnit.NANOSECONDS.convert(interruptTimeout, TimeUnit.MILLISECONDS);
- long deadline = System.nanoTime() + timeoutNanos;
+ /** The logger to report on the fatal condition. */
+ private final Logger log;
- try {
- // Initial wait before interrupting periodically
- Thread.sleep(interruptInterval);
- } catch (InterruptedException ignored) {
- }
+ /** The executing task thread that we wait for to terminate. */
+ private final Thread executerThread;
+
+ /** The name of the task, for logging purposes. */
+ private final String taskName;
+
+ /** The interval in which we interrupt. */
+ private final long interruptIntervalMillis;
+
+ TaskInterrupter(
+ Logger log,
+ Thread executerThread,
+ String taskName,
+ long interruptIntervalMillis) {
- // It is possible that the user code does not react to the task canceller.
- // for that reason, we spawn this separate thread that repeatedly interrupts
- // the user code until it exits. If the suer user code does not exit within
- // the timeout, we notify the job manager about a fatal error.
- while (executer.isAlive()) {
- long now = System.nanoTime();
+ this.log = log;
+ this.executerThread = executerThread;
+ this.taskName = taskName;
+ this.interruptIntervalMillis = interruptIntervalMillis;
+ }
+ @Override
+ public void run() {
+ try {
+ // we initially wait for one interval
+ // in most cases, the threads go away immediately (by the cancellation thread)
+ // and we need not actually do anything
+ executerThread.join(interruptIntervalMillis);
+
+ // log stack trace where the executing thread is stuck and
+ // interrupt the running thread periodically while it is still alive
+ while (executerThread.isAlive()) {
// build the stack trace of where the thread is stuck, for the log
+ StackTraceElement[] stack = executerThread.getStackTrace();
StringBuilder bld = new StringBuilder();
- StackTraceElement[] stack = executer.getStackTrace();
for (StackTraceElement e : stack) {
bld.append(e).append('\n');
}
- if (now >= deadline) {
- long duration = TimeUnit.SECONDS.convert(interruptInterval, TimeUnit.MILLISECONDS);
- String msg = String.format("Task '%s' did not react to cancelling signal in " +
- "the last %d seconds, but is stuck in method:\n %s",
- taskName,
- duration,
- bld.toString());
+ log.warn("Task '{}' did not react to cancelling signal for {} seconds, but is stuck in method:\n {}",
+ taskName, (interruptIntervalMillis / 1000), bld);
- logger.info("Notifying TaskManager about fatal error. {}.", msg);
+ executerThread.interrupt();
+ try {
+ executerThread.join(interruptIntervalMillis);
+ }
+ catch (InterruptedException e) {
+ // we ignore this and fall through the loop
+ }
+ }
+ } catch (Throwable t) {
+ ExceptionUtils.rethrowIfFatalError(t);
+ log.error("Error in the task canceler for task {}.", taskName, t);
+ }
+ }
+ }
- taskManager.notifyFatalError(msg, null);
+ /**
+ * Watchdog for the cancellation.
+ * If the task thread does not go away gracefully within a certain time, we
+ * trigger a hard cancel action (notify TaskManager of fatal error, which in
+ * turn kills the process).
+ */
+ private static class TaskCancelerWatchDog implements Runnable {
- return; // done, don't forget to leave the loop
- } else {
- logger.warn("Task '{}' did not react to cancelling signal, but is stuck in method:\n {}",
- taskName, bld.toString());
+ /** The logger to report on the fatal condition. */
+ private final Logger log;
- executer.interrupt();
- try {
- long timeLeftNanos = Math.min(intervalNanos, deadline - now);
- long timeLeftMillis = TimeUnit.MILLISECONDS.convert(timeLeftNanos, TimeUnit.NANOSECONDS);
+ /** The executing task thread that we wait for to terminate. */
+ private final Thread executerThread;
- if (timeLeftMillis > 0) {
- executer.join(timeLeftMillis);
- }
- } catch (InterruptedException ignored) {
- }
+ /** The TaskManager to notify if cancellation does not happen in time. */
+ private final TaskManagerActions taskManager;
+
+ /** The timeout for cancellation. */
+ private final long timeoutMillis;
+
+ TaskCancelerWatchDog(
+ Thread executerThread,
+ TaskManagerActions taskManager,
+ long timeoutMillis,
+ Logger log) {
+
+ checkArgument(timeoutMillis > 0);
+
+ this.log = log;
+ this.executerThread = executerThread;
+ this.taskManager = taskManager;
+ this.timeoutMillis = timeoutMillis;
+ }
+
+ @Override
+ public void run() {
+ try {
+ final long hardKillDeadline = System.nanoTime() + timeoutMillis * 1_000_000;
+
+ long millisLeft;
+ while (executerThread.isAlive()
+ && (millisLeft = (hardKillDeadline - System.nanoTime()) / 1_000_000) > 0) {
+
+ try {
+ executerThread.join(millisLeft);
+ }
+ catch (InterruptedException ignored) {
+ // we don't react to interrupted exceptions, simply fall through the loop
}
}
+
+ if (executerThread.isAlive()) {
+ String msg = "Task did not exit gracefully within " + (timeoutMillis / 1000) + " + seconds.";
+ log.error(msg);
+ taskManager.notifyFatalError(msg, null);
+ }
+ }
+ catch (Throwable t) {
+ ExceptionUtils.rethrowIfFatalError(t);
+ log.error("Error in Task Cancellation Watch Dog", t);
}
}
}