From 1cbffa3c4afab605d6fe9a6b6b35e8690efba9aa Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Thu, 15 Feb 2018 11:59:12 -0500 Subject: [PATCH] Handle throws on tasks submitted to thread pools When we submit a task to a thread pool for asynchronous execution, we are returned a future. Since we submitted to go asynchronous, these futures are not inspected for failure (we would have to block a thread to do that). While we have on failure handlers for exceptions that are thrown during execution, we do not handle throwables that are not exceptions and these end up silently lost. This commit adds a check after the runnable returns that inspects the status of the future. If an unhandled throwable occurred during execution, this throwable is propogated out where it will land in the uncaught exception handler. Relates #28667 --- .../threadpool/EvilThreadPoolTests.java | 105 ++++++++++++++++++ .../common/util/concurrent/ThreadContext.java | 37 ++++++ 2 files changed, 142 insertions(+) create mode 100644 qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java diff --git a/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java b/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java new file mode 100644 index 0000000000000..c7848267ff17f --- /dev/null +++ b/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java @@ -0,0 +1,105 @@ +/* + * 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.threadpool; + +import org.elasticsearch.test.ESTestCase; +import org.junit.After; +import org.junit.Before; + +import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; +import java.util.function.Supplier; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.hasToString; +import static org.hamcrest.Matchers.instanceOf; + +public class EvilThreadPoolTests extends ESTestCase { + + private ThreadPool threadPool; + + @Before + public void setUpThreadPool() { + threadPool = new TestThreadPool(EvilThreadPoolTests.class.getName()); + } + + @After + public void tearDownThreadPool() throws InterruptedException { + terminate(threadPool); + } + + public void testExecutionException() throws InterruptedException { + runExecutionExceptionTest( + () -> { + throw new Error("future error"); + }, + true, + o -> { + assertTrue(o.isPresent()); + assertThat(o.get(), instanceOf(Error.class)); + assertThat(o.get(), hasToString(containsString("future error"))); + }); + runExecutionExceptionTest( + () -> { + throw new IllegalStateException("future exception"); + }, + false, + o -> assertFalse(o.isPresent())); + } + + private void runExecutionExceptionTest( + final Runnable runnable, + final boolean expectThrowable, + final Consumer> consumer) throws InterruptedException { + final AtomicReference throwableReference = new AtomicReference<>(); + final Thread.UncaughtExceptionHandler uncaughtExceptionHandler = Thread.getDefaultUncaughtExceptionHandler(); + final CountDownLatch uncaughtExceptionHandlerLatch = new CountDownLatch(1); + + try { + Thread.setDefaultUncaughtExceptionHandler((t, e) -> { + assertTrue(expectThrowable); + throwableReference.set(e); + uncaughtExceptionHandlerLatch.countDown(); + }); + + final CountDownLatch supplierLatch = new CountDownLatch(1); + + threadPool.generic().submit(() -> { + try { + runnable.run(); + } finally { + supplierLatch.countDown(); + } + }); + + supplierLatch.await(); + + if (expectThrowable) { + uncaughtExceptionHandlerLatch.await(); + } + consumer.accept(Optional.ofNullable(throwableReference.get())); + } finally { + Thread.setDefaultUncaughtExceptionHandler(uncaughtExceptionHandler); + } + } + +} diff --git a/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java b/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java index 6427368c4b915..8f950c5434bd7 100644 --- a/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java +++ b/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java @@ -19,9 +19,11 @@ package org.elasticsearch.common.util.concurrent; import org.apache.lucene.util.CloseableThreadLocal; +import org.elasticsearch.ExceptionsHelper; import org.elasticsearch.common.io.stream.StreamInput; import org.elasticsearch.common.io.stream.StreamOutput; import org.elasticsearch.common.io.stream.Writeable; +import org.elasticsearch.common.logging.ESLoggerFactory; import org.elasticsearch.common.settings.Setting; import org.elasticsearch.common.settings.Setting.Property; import org.elasticsearch.common.settings.Settings; @@ -33,7 +35,12 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.FutureTask; +import java.util.concurrent.RunnableFuture; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; import java.util.function.Supplier; @@ -564,6 +571,36 @@ public void run() { ctx.restore(); whileRunning = true; in.run(); + if (in instanceof RunnableFuture) { + /* + * The wrapped runnable arose from asynchronous submission of a task to an executor. If an uncaught exception was thrown + * during the execution of this task, we need to inspect this runnable and see if it is an error that should be + * propagated to the uncaught exception handler. + */ + try { + ((RunnableFuture) in).get(); + } catch (final Exception e) { + /* + * In theory, Future#get can only throw a cancellation exception, an interrupted exception, or an execution + * exception. We want to ignore cancellation exceptions, restore the interrupt status on interrupted exceptions, and + * inspect the cause of an execution. We are going to be extra paranoid here though and completely unwrap the + * exception to ensure that there is not a buried error anywhere. We assume that a general exception has been + * handled by the executed task or the task submitter. + */ + assert e instanceof CancellationException + || e instanceof InterruptedException + || e instanceof ExecutionException : e; + final Optional maybeError = ExceptionsHelper.maybeError(e, ESLoggerFactory.getLogger(ThreadContext.class)); + if (maybeError.isPresent()) { + // throw this error where it will propagate to the uncaught exception handler + throw maybeError.get(); + } + if (e instanceof InterruptedException) { + // restore the interrupt status + Thread.currentThread().interrupt(); + } + } + } whileRunning = false; } catch (IllegalStateException ex) { if (whileRunning || threadLocal.closed.get() == false) {