From 1213d0b8c049e6236a8918a3af6eb21838bfb7fe Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Tue, 13 Feb 2018 12:01:03 -0500 Subject: [PATCH 1/9] 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. --- .../threadpool/EvilThreadPoolTests.java | 73 +++++++++++++++++++ .../common/util/concurrent/ThreadContext.java | 22 ++++++ 2 files changed, 95 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..409b6bdc80b59 --- /dev/null +++ b/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java @@ -0,0 +1,73 @@ +package org.elasticsearch.threadpool; + +import org.elasticsearch.test.ESTestCase; +import org.junit.After; +import org.junit.Before; + +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"); + }, + t -> { + assertThat(t, instanceOf(Error.class)); + assertThat(t, hasToString(containsString("future error"))); + }); + runExecutionExceptionTest( + () -> { + throw new IllegalStateException("future exception"); + }, + t -> { + assertThat(t, instanceOf(RuntimeException.class)); + assertNotNull(t.getCause()); + assertThat(t.getCause(), instanceOf(IllegalStateException.class)); + assertThat(t.getCause(), hasToString(containsString("future exception"))); + } + ); + } + + private void runExecutionExceptionTest(final Supplier supplier, final Consumer consumer) throws InterruptedException { + final AtomicReference maybeThrowable = new AtomicReference<>(); + final Thread.UncaughtExceptionHandler uncaughtExceptionHandler = Thread.getDefaultUncaughtExceptionHandler(); + final CountDownLatch latch = new CountDownLatch(1); + + try { + Thread.setDefaultUncaughtExceptionHandler((t, e) -> { + maybeThrowable.set(e); + latch.countDown(); + }); + + threadPool.generic().submit(supplier::get); + + latch.await(); + assertNotNull(maybeThrowable.get()); + consumer.accept(maybeThrowable.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..2a41110904ac1 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 @@ -34,6 +34,9 @@ import java.util.List; import java.util.Map; import java.util.Set; +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 +567,25 @@ 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 InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (final ExecutionException e) { + if (e.getCause() instanceof Error) { + // rethrow this as an error where it will propagate to the uncaught exception handler + throw (Error) e.getCause(); + } else { + throw new RuntimeException(e.getCause()); + } + } + } whileRunning = false; } catch (IllegalStateException ex) { if (whileRunning || threadLocal.closed.get() == false) { From de855abf86a761d9ec31d4b80db1539a592248b7 Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Tue, 13 Feb 2018 13:30:10 -0500 Subject: [PATCH 2/9] Checkstyle --- .../java/org/elasticsearch/threadpool/EvilThreadPoolTests.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 index 409b6bdc80b59..8b035a5c8983f 100644 --- a/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java +++ b/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java @@ -49,7 +49,8 @@ public void testExecutionException() throws InterruptedException { ); } - private void runExecutionExceptionTest(final Supplier supplier, final Consumer consumer) throws InterruptedException { + private void runExecutionExceptionTest( + final Supplier supplier, final Consumer consumer) throws InterruptedException { final AtomicReference maybeThrowable = new AtomicReference<>(); final Thread.UncaughtExceptionHandler uncaughtExceptionHandler = Thread.getDefaultUncaughtExceptionHandler(); final CountDownLatch latch = new CountDownLatch(1); From c725ed89e28e602c8198e0fb7d02e9518ca03230 Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Tue, 13 Feb 2018 13:55:08 -0500 Subject: [PATCH 3/9] Licenese header --- .../threadpool/EvilThreadPoolTests.java | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) 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 index 8b035a5c8983f..843ec77708a6a 100644 --- a/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java +++ b/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java @@ -1,3 +1,22 @@ +/* + * 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; From 778a3b47bf2f6868c50da65fcca9f0d391e052bd Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Tue, 13 Feb 2018 16:41:08 -0500 Subject: [PATCH 4/9] Careful now --- .../elasticsearch/common/util/concurrent/ThreadContext.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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 2a41110904ac1..5dd45022719a8 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 @@ -34,6 +34,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CancellationException; import java.util.concurrent.ExecutionException; import java.util.concurrent.FutureTask; import java.util.concurrent.RunnableFuture; @@ -575,15 +576,16 @@ public void run() { */ try { ((RunnableFuture) in).get(); + } catch (final CancellationException e) { + // task was cancelled, ignore } catch (final InterruptedException e) { Thread.currentThread().interrupt(); } catch (final ExecutionException e) { if (e.getCause() instanceof Error) { // rethrow this as an error where it will propagate to the uncaught exception handler throw (Error) e.getCause(); - } else { - throw new RuntimeException(e.getCause()); } + // we assume that a general exception has been handled by the executed task or the task submitter } } whileRunning = false; From 5aad3880a943e365e599bd44d97c49fc61b915ed Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Tue, 13 Feb 2018 17:14:50 -0500 Subject: [PATCH 5/9] Rework test --- .../threadpool/EvilThreadPoolTests.java | 36 ++++++++++--------- 1 file changed, 20 insertions(+), 16 deletions(-) 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 index 843ec77708a6a..8d1b3c440e2e6 100644 --- a/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java +++ b/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java @@ -23,6 +23,7 @@ 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; @@ -51,40 +52,43 @@ public void testExecutionException() throws InterruptedException { () -> { throw new Error("future error"); }, - t -> { - assertThat(t, instanceOf(Error.class)); - assertThat(t, hasToString(containsString("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"); }, - t -> { - assertThat(t, instanceOf(RuntimeException.class)); - assertNotNull(t.getCause()); - assertThat(t.getCause(), instanceOf(IllegalStateException.class)); - assertThat(t.getCause(), hasToString(containsString("future exception"))); - } - ); + false, + o -> assertFalse(o.isPresent())); } private void runExecutionExceptionTest( - final Supplier supplier, final Consumer consumer) throws InterruptedException { - final AtomicReference maybeThrowable = new AtomicReference<>(); + final Supplier supplier, + final boolean expectThrowable, + final Consumer> consumer) throws InterruptedException { + final AtomicReference throwableReference = new AtomicReference<>(); final Thread.UncaughtExceptionHandler uncaughtExceptionHandler = Thread.getDefaultUncaughtExceptionHandler(); final CountDownLatch latch = new CountDownLatch(1); try { Thread.setDefaultUncaughtExceptionHandler((t, e) -> { - maybeThrowable.set(e); + assertTrue(expectThrowable); + throwableReference.set(e); latch.countDown(); }); threadPool.generic().submit(supplier::get); - latch.await(); - assertNotNull(maybeThrowable.get()); - consumer.accept(maybeThrowable.get()); + if (expectThrowable) { + latch.await(); + consumer.accept(Optional.of(throwableReference.get())); + } else { + consumer.accept(Optional.empty()); + } } finally { Thread.setDefaultUncaughtExceptionHandler(uncaughtExceptionHandler); } From 80dca649c47a5053c9bc9753ca71ebbec23b0d4b Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Tue, 13 Feb 2018 17:18:23 -0500 Subject: [PATCH 6/9] More simplification --- .../org/elasticsearch/threadpool/EvilThreadPoolTests.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) 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 index 8d1b3c440e2e6..c01adeaee0780 100644 --- a/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java +++ b/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java @@ -85,10 +85,8 @@ private void runExecutionExceptionTest( if (expectThrowable) { latch.await(); - consumer.accept(Optional.of(throwableReference.get())); - } else { - consumer.accept(Optional.empty()); } + consumer.accept(Optional.ofNullable(throwableReference.get())); } finally { Thread.setDefaultUncaughtExceptionHandler(uncaughtExceptionHandler); } From f73323ed886899e40ffd0c712d3d01dc7e567cdd Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Wed, 14 Feb 2018 13:48:06 -0500 Subject: [PATCH 7/9] Safer test --- .../threadpool/EvilThreadPoolTests.java | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) 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 index c01adeaee0780..c7848267ff17f 100644 --- a/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java +++ b/qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java @@ -67,24 +67,34 @@ public void testExecutionException() throws InterruptedException { } private void runExecutionExceptionTest( - final Supplier supplier, + final Runnable runnable, final boolean expectThrowable, final Consumer> consumer) throws InterruptedException { final AtomicReference throwableReference = new AtomicReference<>(); final Thread.UncaughtExceptionHandler uncaughtExceptionHandler = Thread.getDefaultUncaughtExceptionHandler(); - final CountDownLatch latch = new CountDownLatch(1); + final CountDownLatch uncaughtExceptionHandlerLatch = new CountDownLatch(1); try { Thread.setDefaultUncaughtExceptionHandler((t, e) -> { assertTrue(expectThrowable); throwableReference.set(e); - latch.countDown(); + uncaughtExceptionHandlerLatch.countDown(); }); - threadPool.generic().submit(supplier::get); + final CountDownLatch supplierLatch = new CountDownLatch(1); + + threadPool.generic().submit(() -> { + try { + runnable.run(); + } finally { + supplierLatch.countDown(); + } + }); + + supplierLatch.await(); if (expectThrowable) { - latch.await(); + uncaughtExceptionHandlerLatch.await(); } consumer.accept(Optional.ofNullable(throwableReference.get())); } finally { From 03216950c7b72268b5b62e3065f9de551f9b3eee Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Thu, 15 Feb 2018 10:08:15 -0500 Subject: [PATCH 8/9] Refactor --- .../common/util/concurrent/ThreadContext.java | 28 +++++++++++++------ 1 file changed, 20 insertions(+), 8 deletions(-) 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 5dd45022719a8..573874d3e5eae 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,6 +35,7 @@ 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; @@ -576,14 +579,23 @@ public void run() { */ try { ((RunnableFuture) in).get(); - } catch (final CancellationException e) { - // task was cancelled, ignore - } catch (final InterruptedException e) { - Thread.currentThread().interrupt(); - } catch (final ExecutionException e) { - if (e.getCause() instanceof Error) { - // rethrow this as an error where it will propagate to the uncaught exception handler - throw (Error) e.getCause(); + } 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. + */ + assert e instanceof CancellationException + || e instanceof InterruptedException + || e instanceof ExecutionException : e; + final Optional maybeError = ExceptionsHelper.maybeError(e, ESLoggerFactory.getLogger(ThreadContext.class)); + if (maybeError.isPresent()) { + // rethrow this error where it will propagate to the uncaught exception handler + throw maybeError.get(); + } + if (e instanceof InterruptedException) { + Thread.currentThread().interrupt(); } // we assume that a general exception has been handled by the executed task or the task submitter } From 08375c29637c17b8e998cb35a42aebac062c0168 Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Thu, 15 Feb 2018 10:34:15 -0500 Subject: [PATCH 9/9] Fix comments --- .../common/util/concurrent/ThreadContext.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) 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 573874d3e5eae..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 @@ -584,20 +584,21 @@ public void run() { * 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. + * 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()) { - // rethrow this error where it will propagate to the uncaught exception handler + // 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(); } - // we assume that a general exception has been handled by the executed task or the task submitter } } whileRunning = false;