forked from elastic/elasticsearch
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Recovery: cleaner interrupt handling during cancellation
RecoveryTarget initiates the recovery by sending a start recovery request to the source node and then waits for the recovery to complete. During recovery cancellation, we interrupt the thread so it will wake up and clean the recovery. Depending on timing, this can leave an unneeded interrupted thread status causing future IO commands to fail unneeded. RecoverySource already had a handy utility called CancellableThreads. This extracts it to a top level class, and uses it in RecoveryTarget as well. Closes elastic#9000
- Loading branch information
Showing
6 changed files
with
366 additions
and
128 deletions.
There are no files selected for viewing
145 changes: 145 additions & 0 deletions
145
src/main/java/org/elasticsearch/common/util/CancellableThreads.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,145 @@ | ||
| /* | ||
| * Licensed to Elasticsearch under one or more contributor | ||
| * license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright | ||
| * ownership. Elasticsearch 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.elasticsearch.common.util; | ||
|
|
||
| import org.elasticsearch.ElasticsearchException; | ||
| import org.elasticsearch.common.Nullable; | ||
|
|
||
| import java.util.HashSet; | ||
| import java.util.Set; | ||
|
|
||
| /** | ||
| * A utility class for multi threaded operation that needs to be cancellable via interrupts. Every cancellable operation should be | ||
| * executed via {@link #execute(Interruptable)}, which will capture the executing thread and make sure it is interrupted in the case | ||
| * cancellation. | ||
| */ | ||
| public class CancellableThreads { | ||
| private final Set<Thread> threads = new HashSet<>(); | ||
| private boolean cancelled = false; | ||
| private String reason; | ||
|
|
||
| public synchronized boolean isCancelled() { | ||
| return cancelled; | ||
| } | ||
|
|
||
|
|
||
| /** call this will throw an exception if operation was cancelled. Override {@link #onCancel(String, java.lang.Throwable)} for custom failure logic */ | ||
| public synchronized void checkForCancel() { | ||
| if (isCancelled()) { | ||
| onCancel(reason, null); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * called if {@link #checkForCancel()} was invoked after the operation was cancelled. | ||
| * the default implementation always throws an {@link ExecutionCancelledException}, suppressing | ||
| * any other exception that occurred before cancellation | ||
| * | ||
| * @param reason reason for failure supplied by the caller of {@link @cancel} | ||
| * @param suppressedException any error that was encountered during the execution before the operation was cancelled. | ||
| */ | ||
| protected void onCancel(String reason, @Nullable Throwable suppressedException) { | ||
| RuntimeException e = new ExecutionCancelledException("operation was cancelled reason [" + reason + "]"); | ||
| if (suppressedException != null) { | ||
| e.addSuppressed(suppressedException); | ||
| } | ||
| throw e; | ||
| } | ||
|
|
||
| private synchronized boolean add() { | ||
| checkForCancel(); | ||
| threads.add(Thread.currentThread()); | ||
| // capture and clean the interrupted thread before we start, so we can identify | ||
| // our own interrupt. we do so under lock so we know we don't clear our own. | ||
| return Thread.interrupted(); | ||
| } | ||
|
|
||
| /** | ||
| * run the Interruptable, capturing the executing thread. Concurrent calls to {@link #cancel(String)} will interrupt this thread | ||
| * causing the call to prematurely return. | ||
| * | ||
| * @param interruptable code to run | ||
| */ | ||
| public void execute(Interruptable interruptable) { | ||
| boolean wasInterrupted = add(); | ||
| RuntimeException throwable = null; | ||
| try { | ||
| interruptable.run(); | ||
| } catch (InterruptedException e) { | ||
| // assume this is us and ignore | ||
| } catch (RuntimeException t) { | ||
| throwable = t; | ||
| } finally { | ||
| remove(); | ||
| } | ||
| // we are now out of threads collection so we can't be interrupted any more by this class | ||
| // restore old flag and see if we need to fail | ||
| if (wasInterrupted) { | ||
| Thread.currentThread().interrupt(); | ||
| } else { | ||
| // clear the flag interrupted flag as we are checking for failure.. | ||
| Thread.interrupted(); | ||
| } | ||
| synchronized (this) { | ||
| if (isCancelled()) { | ||
| onCancel(reason, throwable); | ||
| } else if (throwable != null) { | ||
| // if we're not canceling, we throw the original exception | ||
| throw throwable; | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
||
| private synchronized void remove() { | ||
| threads.remove(Thread.currentThread()); | ||
| } | ||
|
|
||
| /** cancel all current running operations. Future calls to {@link #checkForCancel()} will be failed with the given reason */ | ||
| public synchronized void cancel(String reason) { | ||
| if (cancelled) { | ||
| // we were already cancelled, make sure we don't interrupt threads twice | ||
| // this is important in order to make sure that we don't mark | ||
| // Thread.interrupted without handling it | ||
| return; | ||
| } | ||
| cancelled = true; | ||
| this.reason = reason; | ||
| for (Thread thread : threads) { | ||
| thread.interrupt(); | ||
| } | ||
| threads.clear(); | ||
| } | ||
|
|
||
|
|
||
| public interface Interruptable { | ||
| public void run() throws InterruptedException; | ||
| } | ||
|
|
||
| public class ExecutionCancelledException extends ElasticsearchException { | ||
|
|
||
| public ExecutionCancelledException(String msg) { | ||
| super(msg); | ||
| } | ||
|
|
||
| public ExecutionCancelledException(String msg, Throwable cause) { | ||
| super(msg, cause); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.