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
commit 00fe29b1be59c9ee6d7d6f3f71a4759f5132e12d Author: ifndef-SleePy <[email protected]> AuthorDate: Mon Sep 30 11:17:39 2019 +0800 [FLINK-13904][checkpointing] Remove trigger lock of CheckpointCoordinator Now the checkpoint and savepoint triggering are executed in the one thread without any competition. So the triggerLock is no longer needed. --- .../runtime/checkpoint/CheckpointCoordinator.java | 228 ++++++++++----------- 1 file changed, 107 insertions(+), 121 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java index b4843e4..7517a38 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java @@ -87,13 +87,6 @@ public class CheckpointCoordinator { /** Coordinator-wide lock to safeguard the checkpoint updates. */ private final Object lock = new Object(); - /** Lock specially to make sure that trigger requests do not overtake each other. - * This is not done with the coordinator-wide lock, because as part of triggering, - * blocking operations may happen (distributed atomic counters). - * Using a dedicated lock, we avoid blocking the processing of 'acknowledge/decline' - * messages during that phase. */ - private final Object triggerLock = new Object(); - /** The job whose checkpoint this coordinator coordinates. */ private final JobID job; @@ -537,149 +530,142 @@ public class CheckpointCoordinator { // we will actually trigger this checkpoint! - // we lock with a special lock to make sure that trigger requests do not overtake each other. - // this is not done with the coordinator-wide lock, because the 'checkpointIdCounter' - // may issue blocking operations. Using a different lock than the coordinator-wide lock, - // we avoid blocking the processing of 'acknowledge/decline' messages during that time. - synchronized (triggerLock) { - - final CheckpointStorageLocation checkpointStorageLocation; - final long checkpointID; + final CheckpointStorageLocation checkpointStorageLocation; + final long checkpointID; - try { - // this must happen outside the coordinator-wide lock, because it communicates - // with external services (in HA mode) and may block for a while. - checkpointID = checkpointIdCounter.getAndIncrement(); + try { + // this must happen outside the coordinator-wide lock, because it communicates + // with external services (in HA mode) and may block for a while. + checkpointID = checkpointIdCounter.getAndIncrement(); - checkpointStorageLocation = props.isSavepoint() ? - checkpointStorage.initializeLocationForSavepoint(checkpointID, externalSavepointLocation) : - checkpointStorage.initializeLocationForCheckpoint(checkpointID); - } - catch (Throwable t) { - int numUnsuccessful = numUnsuccessfulCheckpointsTriggers.incrementAndGet(); - LOG.warn("Failed to trigger checkpoint for job {} ({} consecutive failed attempts so far).", - job, - numUnsuccessful, - t); - throw new CheckpointException(CheckpointFailureReason.EXCEPTION, t); - } + checkpointStorageLocation = props.isSavepoint() ? + checkpointStorage.initializeLocationForSavepoint(checkpointID, externalSavepointLocation) : + checkpointStorage.initializeLocationForCheckpoint(checkpointID); + } + catch (Throwable t) { + int numUnsuccessful = numUnsuccessfulCheckpointsTriggers.incrementAndGet(); + LOG.warn("Failed to trigger checkpoint for job {} ({} consecutive failed attempts so far).", + job, + numUnsuccessful, + t); + throw new CheckpointException(CheckpointFailureReason.EXCEPTION, t); + } - final PendingCheckpoint checkpoint = new PendingCheckpoint( - job, + final PendingCheckpoint checkpoint = new PendingCheckpoint( + job, + checkpointID, + timestamp, + ackTasks, + props, + checkpointStorageLocation, + executor); + + if (statsTracker != null) { + PendingCheckpointStats callback = statsTracker.reportPendingCheckpoint( checkpointID, timestamp, - ackTasks, - props, - checkpointStorageLocation, - executor); - - if (statsTracker != null) { - PendingCheckpointStats callback = statsTracker.reportPendingCheckpoint( - checkpointID, - timestamp, - props); + props); - checkpoint.setStatsCallback(callback); - } + checkpoint.setStatsCallback(callback); + } - // schedule the timer that will clean up the expired checkpoints - final Runnable canceller = () -> { - synchronized (lock) { - // only do the work if the checkpoint is not discarded anyways - // note that checkpoint completion discards the pending checkpoint object - if (!checkpoint.isDiscarded()) { - LOG.info("Checkpoint {} of job {} expired before completing.", checkpointID, job); + // schedule the timer that will clean up the expired checkpoints + final Runnable canceller = () -> { + synchronized (lock) { + // only do the work if the checkpoint is not discarded anyways + // note that checkpoint completion discards the pending checkpoint object + if (!checkpoint.isDiscarded()) { + LOG.info("Checkpoint {} of job {} expired before completing.", checkpointID, job); - failPendingCheckpoint(checkpoint, CheckpointFailureReason.CHECKPOINT_EXPIRED); - pendingCheckpoints.remove(checkpointID); - rememberRecentCheckpointId(checkpointID); + failPendingCheckpoint(checkpoint, CheckpointFailureReason.CHECKPOINT_EXPIRED); + pendingCheckpoints.remove(checkpointID); + rememberRecentCheckpointId(checkpointID); - triggerQueuedRequests(); - } + triggerQueuedRequests(); } - }; + } + }; - try { - // re-acquire the coordinator-wide lock - synchronized (lock) { - // since we released the lock in the meantime, we need to re-check - // that the conditions still hold. - if (shutdown) { - throw new CheckpointException(CheckpointFailureReason.CHECKPOINT_COORDINATOR_SHUTDOWN); + try { + // re-acquire the coordinator-wide lock + synchronized (lock) { + // since we released the lock in the meantime, we need to re-check + // that the conditions still hold. + if (shutdown) { + throw new CheckpointException(CheckpointFailureReason.CHECKPOINT_COORDINATOR_SHUTDOWN); + } + else if (!props.forceCheckpoint()) { + if (triggerRequestQueued) { + LOG.warn("Trying to trigger another checkpoint for job {} while one was queued already.", job); + throw new CheckpointException(CheckpointFailureReason.ALREADY_QUEUED); } - else if (!props.forceCheckpoint()) { - if (triggerRequestQueued) { - LOG.warn("Trying to trigger another checkpoint for job {} while one was queued already.", job); - throw new CheckpointException(CheckpointFailureReason.ALREADY_QUEUED); - } - checkConcurrentCheckpoints(); + checkConcurrentCheckpoints(); - checkMinPauseBetweenCheckpoints(); - } + checkMinPauseBetweenCheckpoints(); + } - LOG.info("Triggering checkpoint {} @ {} for job {}.", checkpointID, timestamp, job); + LOG.info("Triggering checkpoint {} @ {} for job {}.", checkpointID, timestamp, job); - pendingCheckpoints.put(checkpointID, checkpoint); + pendingCheckpoints.put(checkpointID, checkpoint); - ScheduledFuture<?> cancellerHandle = timer.schedule( - canceller, - checkpointTimeout, TimeUnit.MILLISECONDS); + ScheduledFuture<?> cancellerHandle = timer.schedule( + canceller, + checkpointTimeout, TimeUnit.MILLISECONDS); - if (!checkpoint.setCancellerHandle(cancellerHandle)) { - // checkpoint is already disposed! - cancellerHandle.cancel(false); - } + if (!checkpoint.setCancellerHandle(cancellerHandle)) { + // checkpoint is already disposed! + cancellerHandle.cancel(false); + } - // trigger the master hooks for the checkpoint - final List<MasterState> masterStates = MasterHooks.triggerMasterHooks(masterHooks.values(), - checkpointID, timestamp, executor, Time.milliseconds(checkpointTimeout)); - for (MasterState s : masterStates) { - checkpoint.addMasterState(s); - } + // trigger the master hooks for the checkpoint + final List<MasterState> masterStates = MasterHooks.triggerMasterHooks(masterHooks.values(), + checkpointID, timestamp, executor, Time.milliseconds(checkpointTimeout)); + for (MasterState s : masterStates) { + checkpoint.addMasterState(s); } - // end of lock scope + } + // end of lock scope - final CheckpointOptions checkpointOptions = new CheckpointOptions( - props.getCheckpointType(), - checkpointStorageLocation.getLocationReference()); + final CheckpointOptions checkpointOptions = new CheckpointOptions( + props.getCheckpointType(), + checkpointStorageLocation.getLocationReference()); - // send the messages to the tasks that trigger their checkpoint - for (Execution execution: executions) { - if (props.isSynchronous()) { - execution.triggerSynchronousSavepoint(checkpointID, timestamp, checkpointOptions, advanceToEndOfTime); - } else { - execution.triggerCheckpoint(checkpointID, timestamp, checkpointOptions); - } + // send the messages to the tasks that trigger their checkpoint + for (Execution execution: executions) { + if (props.isSynchronous()) { + execution.triggerSynchronousSavepoint(checkpointID, timestamp, checkpointOptions, advanceToEndOfTime); + } else { + execution.triggerCheckpoint(checkpointID, timestamp, checkpointOptions); } - - numUnsuccessfulCheckpointsTriggers.set(0); - return checkpoint.getCompletionFuture(); } - catch (Throwable t) { - // guard the map against concurrent modifications - synchronized (lock) { - pendingCheckpoints.remove(checkpointID); - } - int numUnsuccessful = numUnsuccessfulCheckpointsTriggers.incrementAndGet(); - LOG.warn("Failed to trigger checkpoint {} for job {}. ({} consecutive failed attempts so far)", - checkpointID, job, numUnsuccessful, t); + numUnsuccessfulCheckpointsTriggers.set(0); + return checkpoint.getCompletionFuture(); + } + catch (Throwable t) { + // guard the map against concurrent modifications + synchronized (lock) { + pendingCheckpoints.remove(checkpointID); + } - if (!checkpoint.isDiscarded()) { - failPendingCheckpoint(checkpoint, CheckpointFailureReason.TRIGGER_CHECKPOINT_FAILURE, t); - } + int numUnsuccessful = numUnsuccessfulCheckpointsTriggers.incrementAndGet(); + LOG.warn("Failed to trigger checkpoint {} for job {}. ({} consecutive failed attempts so far)", + checkpointID, job, numUnsuccessful, t); - try { - checkpointStorageLocation.disposeOnFailure(); - } - catch (Throwable t2) { - LOG.warn("Cannot dispose failed checkpoint storage location {}", checkpointStorageLocation, t2); - } + if (!checkpoint.isDiscarded()) { + failPendingCheckpoint(checkpoint, CheckpointFailureReason.TRIGGER_CHECKPOINT_FAILURE, t); + } - throw new CheckpointException(CheckpointFailureReason.EXCEPTION, t); + try { + checkpointStorageLocation.disposeOnFailure(); } - } // end trigger lock + catch (Throwable t2) { + LOG.warn("Cannot dispose failed checkpoint storage location {}", checkpointStorageLocation, t2); + } + + throw new CheckpointException(CheckpointFailureReason.EXCEPTION, t); + } } // --------------------------------------------------------------------------------------------
