Skip to content

Commit

Permalink
Do not close threadpool if termination fails
Browse files Browse the repository at this point in the history
This commit changes the code so that the threadpool is not closed
unless termination succeeds. Otherwise there can still be running tasks
that rely on resources that are closed by closing the threadpool.

Additionally, there is a test fix included for the NodeTests that
ensures the submitted task is actually running prior to closing the
node in the test.

Closes elastic#42577
  • Loading branch information
jaymode committed May 31, 2019
1 parent 56c1ed5 commit f6eead7
Show file tree
Hide file tree
Showing 4 changed files with 37 additions and 11 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -374,7 +374,14 @@ public void close() {
for (LifecycleComponent plugin : pluginLifecycleComponents) {
closeables.add(plugin);
}
closeables.add(() -> ThreadPool.terminate(injector.getInstance(ThreadPool.class), 10, TimeUnit.SECONDS));
closeables.add(() -> {
final ThreadPool pool = injector.getInstance(ThreadPool.class);
final boolean terminated = ThreadPool.terminate(pool, 10, TimeUnit.SECONDS);
if (terminated == false) {
// the pool is only closed if termination succeeds, just close even if termination failed
pool.close();
}
});
IOUtils.closeWhileHandlingException(closeables);
}

Expand Down
23 changes: 15 additions & 8 deletions server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java
Original file line number Diff line number Diff line change
Expand Up @@ -412,6 +412,7 @@ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedE
}
}
cachedTimeThread.join(unit.toMillis(timeout));
result &= cachedTimeThread.isAlive() == false;
return result;
}

Expand Down Expand Up @@ -699,22 +700,28 @@ private static boolean awaitTermination(

/**
* Returns <code>true</code> if the given pool was terminated successfully. If the termination timed out,
* the service is <code>null</code> this method will return <code>false</code>.
* the service is <code>null</code> this method will return <code>false</code>. The pool is only closed if
* the termination was successful.
*/
public static boolean terminate(ThreadPool pool, long timeout, TimeUnit timeUnit) {
boolean terminated = false;
if (pool != null) {
// Leverage try-with-resources to close the threadpool
try (ThreadPool c = pool) {
try {
pool.shutdown();
if (awaitTermination(pool, timeout, timeUnit)) {
return true;
terminated = true;
} else {
// last resort
pool.shutdownNow();
terminated = awaitTermination(pool, timeout, timeUnit);
}
} finally {
if (terminated) {
pool.close();
}
// last resort
pool.shutdownNow();
return awaitTermination(pool, timeout, timeUnit);
}
}
return false;
return terminated;
}

private static boolean awaitTermination(
Expand Down
9 changes: 8 additions & 1 deletion server/src/test/java/org/elasticsearch/node/NodeTests.java
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,6 @@
import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_SHARDS;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;

@LuceneTestCase.AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/42577")
@LuceneTestCase.SuppressFileSystems(value = "ExtrasFS")
public class NodeTests extends ESTestCase {

Expand Down Expand Up @@ -154,9 +153,12 @@ public void testCloseOnOutstandingTask() throws Exception {
node.start();
ThreadPool threadpool = node.injector().getInstance(ThreadPool.class);
AtomicBoolean shouldRun = new AtomicBoolean(true);
CountDownLatch threadRunning = new CountDownLatch(1);
threadpool.executor(ThreadPool.Names.SEARCH).execute(() -> {
threadRunning.countDown();
while (shouldRun.get());
});
threadRunning.await();
node.close();
shouldRun.set(false);
assertTrue(node.awaitClose(1, TimeUnit.DAYS));
Expand All @@ -167,12 +169,17 @@ public void testAwaitCloseTimeoutsOnNonInterruptibleTask() throws Exception {
node.start();
ThreadPool threadpool = node.injector().getInstance(ThreadPool.class);
AtomicBoolean shouldRun = new AtomicBoolean(true);
CountDownLatch threadRunning = new CountDownLatch(1);
threadpool.executor(ThreadPool.Names.SEARCH).execute(() -> {
threadRunning.countDown();
while (shouldRun.get());
});
threadRunning.await();
node.close();
assertFalse(node.awaitClose(0, TimeUnit.MILLISECONDS));
shouldRun.set(false);
// call this again to ensure we terminate and close the threadpool
assertTrue(node.awaitClose(1, TimeUnit.DAYS));
}

public void testCloseOnInterruptibleTask() throws Exception {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -57,10 +57,15 @@ void doExecute(Action<Response> action, Request request, ActionListener<Response

@Override
public void close() {
boolean terminated = false;
try {
ThreadPool.terminate(threadPool(), 10, TimeUnit.SECONDS);
terminated = ThreadPool.terminate(threadPool(), 10, TimeUnit.SECONDS);
} catch (Exception e) {
throw new ElasticsearchException(e.getMessage(), e);
}

if (terminated == false) {
throw new IllegalStateException("threadpool was not terminated after waiting for 10 seconds");
}
}
}

0 comments on commit f6eead7

Please sign in to comment.