[ONOS-6267] Detect and complete blocked futures on I/O threads.
Change-Id: I0488dc5096f9e610b97405ad05c02d0ff3854b5f
diff --git a/utils/misc/src/main/java/org/onlab/util/BestEffortSerialExecutor.java b/utils/misc/src/main/java/org/onlab/util/BestEffortSerialExecutor.java
deleted file mode 100644
index 936a33f..0000000
--- a/utils/misc/src/main/java/org/onlab/util/BestEffortSerialExecutor.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Copyright 2017-present Open Networking Laboratory
- *
- * Licensed 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.onlab.util;
-
-import java.util.LinkedList;
-import java.util.concurrent.Executor;
-
-/**
- * Executor that executes tasks in serial on a shared thread pool, falling back to parallel execution when threads
- * are blocked.
- * <p>
- * This executor attempts to execute tasks in serial as if they occur on a single thread. However, in the event tasks
- * are blocking a thread (a thread is in the {@link Thread.State#WAITING} or {@link Thread.State#TIMED_WAITING} state)
- * the executor will execute tasks on parallel on the underlying {@link Executor}. This is useful for ensuring blocked
- * threads cannot block events, but mimics a single-threaded model otherwise.
- */
-public class BestEffortSerialExecutor implements Executor {
- private final Executor parent;
- private final LinkedList<Runnable> tasks = new LinkedList<>();
- private volatile Thread thread;
-
- public BestEffortSerialExecutor(Executor parent) {
- this.parent = parent;
- }
-
- private void run() {
- synchronized (tasks) {
- thread = Thread.currentThread();
- }
- for (;;) {
- if (!runTask()) {
- synchronized (tasks) {
- thread = null;
- }
- return;
- }
- }
- }
-
- private boolean runTask() {
- final Runnable task;
- synchronized (tasks) {
- task = tasks.poll();
- if (task == null) {
- return false;
- }
- }
- task.run();
- return true;
- }
-
- @Override
- public void execute(Runnable command) {
- synchronized (tasks) {
- tasks.add(command);
- if (thread == null) {
- parent.execute(this::run);
- } else if (thread.getState() == Thread.State.WAITING || thread.getState() == Thread.State.TIMED_WAITING) {
- parent.execute(this::runTask);
- }
- }
- }
-}
diff --git a/utils/misc/src/main/java/org/onlab/util/BlockingAwareFuture.java b/utils/misc/src/main/java/org/onlab/util/BlockingAwareFuture.java
new file mode 100644
index 0000000..51f1809
--- /dev/null
+++ b/utils/misc/src/main/java/org/onlab/util/BlockingAwareFuture.java
@@ -0,0 +1,291 @@
+/*
+ * Copyright 2017-present Open Networking Laboratory
+ *
+ * Licensed 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.onlab.util;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+/**
+ * A {@link CompletableFuture} that tracks whether the future or one of its descendants has been blocked on
+ * a {@link CompletableFuture#get()} or {@link CompletableFuture#join()} call.
+ */
+public class BlockingAwareFuture<T> extends CompletableFuture<T> {
+ private final AtomicBoolean blocked;
+
+ public BlockingAwareFuture() {
+ this(new AtomicBoolean());
+ }
+
+ private BlockingAwareFuture(AtomicBoolean blocked) {
+ this.blocked = blocked;
+ }
+
+ /**
+ * Returns a boolean indicating whether the future is blocked.
+ *
+ * @return indicates whether the future is blocked
+ */
+ public boolean isBlocked() {
+ return blocked.get();
+ }
+
+ @Override
+ public T get() throws InterruptedException, ExecutionException {
+ blocked.set(true);
+ try {
+ return super.get();
+ } finally {
+ blocked.set(false);
+ }
+ }
+
+ @Override
+ public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
+ blocked.set(true);
+ try {
+ return super.get(timeout, unit);
+ } finally {
+ blocked.set(false);
+ }
+ }
+
+ @Override
+ public synchronized T join() {
+ blocked.set(true);
+ try {
+ return super.join();
+ } finally {
+ blocked.set(false);
+ }
+ }
+
+ /**
+ * Wraps the given future in a new blockable future.
+ *
+ * @param future the future to wrap
+ * @param <U> the future value type
+ * @return a new blockable future
+ */
+ private <U> CompletableFuture<U> wrap(CompletableFuture<U> future) {
+ BlockingAwareFuture<U> blockingFuture = new BlockingAwareFuture<U>(blocked);
+ future.whenComplete((result, error) -> {
+ if (error == null) {
+ blockingFuture.complete(result);
+ } else {
+ blockingFuture.completeExceptionally(error);
+ }
+ });
+ return blockingFuture;
+ }
+
+ @Override
+ public <U> CompletableFuture<U> thenApply(Function<? super T, ? extends U> fn) {
+ return wrap(super.thenApply(fn));
+ }
+
+ @Override
+ public <U> CompletableFuture<U> thenApplyAsync(Function<? super T, ? extends U> fn) {
+ return wrap(super.thenApplyAsync(fn));
+ }
+
+ @Override
+ public <U> CompletableFuture<U> thenApplyAsync(Function<? super T, ? extends U> fn, Executor executor) {
+ return wrap(super.thenApplyAsync(fn, executor));
+ }
+
+ @Override
+ public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
+ return wrap(super.thenAccept(action));
+ }
+
+ @Override
+ public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
+ return wrap(super.thenAcceptAsync(action));
+ }
+
+ @Override
+ public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor) {
+ return wrap(super.thenAcceptAsync(action, executor));
+ }
+
+ @Override
+ public CompletableFuture<Void> thenRun(Runnable action) {
+ return wrap(super.thenRun(action));
+ }
+
+ @Override
+ public CompletableFuture<Void> thenRunAsync(Runnable action) {
+ return wrap(super.thenRunAsync(action));
+ }
+
+ @Override
+ public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor) {
+ return wrap(super.thenRunAsync(action, executor));
+ }
+
+ @Override
+ public <U, V> CompletableFuture<V> thenCombine(
+ CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn) {
+ return wrap(super.thenCombine(other, fn));
+ }
+
+ @Override
+ public <U, V> CompletableFuture<V> thenCombineAsync(
+ CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn) {
+ return wrap(super.thenCombineAsync(other, fn));
+ }
+
+ @Override
+ public <U, V> CompletableFuture<V> thenCombineAsync(
+ CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn, Executor executor) {
+ return wrap(super.thenCombineAsync(other, fn, executor));
+ }
+
+ @Override
+ public <U> CompletableFuture<Void> thenAcceptBoth(
+ CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) {
+ return wrap(super.thenAcceptBoth(other, action));
+ }
+
+ @Override
+ public <U> CompletableFuture<Void> thenAcceptBothAsync(
+ CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) {
+ return wrap(super.thenAcceptBothAsync(other, action));
+ }
+
+ @Override
+ public <U> CompletableFuture<Void> thenAcceptBothAsync(
+ CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor) {
+ return wrap(super.thenAcceptBothAsync(other, action, executor));
+ }
+
+ @Override
+ public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action) {
+ return wrap(super.runAfterBoth(other, action));
+ }
+
+ @Override
+ public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action) {
+ return wrap(super.runAfterBothAsync(other, action));
+ }
+
+ @Override
+ public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor) {
+ return wrap(super.runAfterBothAsync(other, action, executor));
+ }
+
+ @Override
+ public <U> CompletableFuture<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn) {
+ return wrap(super.applyToEither(other, fn));
+ }
+
+ @Override
+ public <U> CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn) {
+ return wrap(super.applyToEitherAsync(other, fn));
+ }
+
+ @Override
+ public <U> CompletableFuture<U> applyToEitherAsync(
+ CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor) {
+ return wrap(super.applyToEitherAsync(other, fn, executor));
+ }
+
+ @Override
+ public CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) {
+ return wrap(super.acceptEither(other, action));
+ }
+
+ @Override
+ public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) {
+ return wrap(super.acceptEitherAsync(other, action));
+ }
+
+ @Override
+ public CompletableFuture<Void> acceptEitherAsync(
+ CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor) {
+ return wrap(super.acceptEitherAsync(other, action, executor));
+ }
+
+ @Override
+ public CompletableFuture<Void> runAfterEither(CompletionStage<?> other, Runnable action) {
+ return wrap(super.runAfterEither(other, action));
+ }
+
+ @Override
+ public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action) {
+ return wrap(super.runAfterEitherAsync(other, action));
+ }
+
+ @Override
+ public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor) {
+ return wrap(super.runAfterEitherAsync(other, action, executor));
+ }
+
+ @Override
+ public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) {
+ return wrap(super.thenCompose(fn));
+ }
+
+ @Override
+ public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) {
+ return wrap(super.thenComposeAsync(fn));
+ }
+
+ @Override
+ public <U> CompletableFuture<U> thenComposeAsync(
+ Function<? super T, ? extends CompletionStage<U>> fn, Executor executor) {
+ return wrap(super.thenComposeAsync(fn, executor));
+ }
+
+ @Override
+ public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) {
+ return wrap(super.whenComplete(action));
+ }
+
+ @Override
+ public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) {
+ return wrap(super.whenCompleteAsync(action));
+ }
+
+ @Override
+ public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) {
+ return wrap(super.whenCompleteAsync(action, executor));
+ }
+
+ @Override
+ public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) {
+ return wrap(super.handle(fn));
+ }
+
+ @Override
+ public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) {
+ return wrap(super.handleAsync(fn));
+ }
+
+ @Override
+ public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
+ return wrap(super.handleAsync(fn, executor));
+ }
+}
diff --git a/utils/misc/src/main/java/org/onlab/util/SerialExecutor.java b/utils/misc/src/main/java/org/onlab/util/OrderedExecutor.java
similarity index 82%
rename from utils/misc/src/main/java/org/onlab/util/SerialExecutor.java
rename to utils/misc/src/main/java/org/onlab/util/OrderedExecutor.java
index 9e54ac2..8fe2cc2 100644
--- a/utils/misc/src/main/java/org/onlab/util/SerialExecutor.java
+++ b/utils/misc/src/main/java/org/onlab/util/OrderedExecutor.java
@@ -19,17 +19,17 @@
import java.util.concurrent.Executor;
/**
- * Executor that executes tasks in serial on a shared thread pool.
+ * Executor that executes tasks in order on a shared thread pool.
* <p>
- * The serial executor behaves semantically like a single-threaded executor, but multiplexes tasks on a shared thread
- * pool, ensuring blocked threads in the shared thread pool don't block individual serial executors.
+ * The ordered executor behaves semantically like a single-threaded executor, but multiplexes tasks on a shared thread
+ * pool, ensuring blocked threads in the shared thread pool don't block individual ordered executors.
*/
-public class SerialExecutor implements Executor {
+public class OrderedExecutor implements Executor {
private final Executor parent;
private final LinkedList<Runnable> tasks = new LinkedList<>();
private boolean running;
- public SerialExecutor(Executor parent) {
+ public OrderedExecutor(Executor parent) {
this.parent = parent;
}
diff --git a/utils/misc/src/main/java/org/onlab/util/Tools.java b/utils/misc/src/main/java/org/onlab/util/Tools.java
index 883a7c8..0c76a91 100644
--- a/utils/misc/src/main/java/org/onlab/util/Tools.java
+++ b/utils/misc/src/main/java/org/onlab/util/Tools.java
@@ -643,28 +643,41 @@
}
/**
- * Returns a future that's completed using the given {@link Executor} once the given {@code future} is completed.
+ * Returns a future that's completed using the given {@code orderedExecutor} if the future is not blocked or the
+ * given {@code threadPoolExecutor} if the future is blocked.
* <p>
- * {@link CompletableFuture}'s async methods cannot be relied upon to complete futures on an executor thread. If a
- * future is completed synchronously, {@code CompletableFuture} async methods will often complete the future on the
- * current thread, ignoring the provided {@code Executor}. This method ensures a more reliable and consistent thread
- * model by ensuring that futures are always completed using the provided {@code Executor}.
+ * This method allows futures to maintain single-thread semantics via the provided {@code orderedExecutor} while
+ * ensuring user code can block without blocking completion of futures. When the returned future or any of its
+ * descendants is blocked on a {@link CompletableFuture#get()} or {@link CompletableFuture#join()} call, completion
+ * of the returned future will be done using the provided {@code threadPoolExecutor}.
*
* @param future the future to convert into an asynchronous future
- * @param executor the executor with which to complete the returned future
+ * @param orderedExecutor the ordered executor with which to attempt to complete the future
+ * @param threadPoolExecutor the backup executor with which to complete blocked futures
* @param <T> future value type
* @return a new completable future to be completed using the provided {@code executor} once the provided
* {@code future} is complete
*/
- public static <T> CompletableFuture<T> asyncFuture(CompletableFuture<T> future, Executor executor) {
- CompletableFuture<T> newFuture = new CompletableFuture<T>();
- future.whenComplete((result, error) -> executor.execute(() -> {
- if (future.isCompletedExceptionally()) {
- newFuture.completeExceptionally(error);
+ public static <T> CompletableFuture<T> orderedFuture(
+ CompletableFuture<T> future,
+ Executor orderedExecutor,
+ Executor threadPoolExecutor) {
+ BlockingAwareFuture<T> newFuture = new BlockingAwareFuture<T>();
+ future.whenComplete((result, error) -> {
+ Runnable completer = () -> {
+ if (future.isCompletedExceptionally()) {
+ newFuture.completeExceptionally(error);
+ } else {
+ newFuture.complete(result);
+ }
+ };
+
+ if (newFuture.isBlocked()) {
+ threadPoolExecutor.execute(completer);
} else {
- newFuture.complete(result);
+ orderedExecutor.execute(completer);
}
- }));
+ });
return newFuture;
}
diff --git a/utils/misc/src/test/java/org/onlab/util/BestEffortSerialExecutorTest.java b/utils/misc/src/test/java/org/onlab/util/BestEffortSerialExecutorTest.java
deleted file mode 100644
index 39a1f87..0000000
--- a/utils/misc/src/test/java/org/onlab/util/BestEffortSerialExecutorTest.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Copyright 2017-present Open Networking Laboratory
- *
- * Licensed 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.onlab.util;
-
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
-/**
- * Best effort serial executor test.
- */
-public class BestEffortSerialExecutorTest {
-
- @Test
- public void testSerialExecution() throws Throwable {
- Executor executor = new BestEffortSerialExecutor(SharedExecutors.getPoolThreadExecutor());
- CountDownLatch latch = new CountDownLatch(2);
- executor.execute(latch::countDown);
- executor.execute(latch::countDown);
- latch.await();
- assertEquals(0, latch.getCount());
- }
-
- @Test
- public void testBlockedExecution() throws Throwable {
- Executor executor = new BestEffortSerialExecutor(SharedExecutors.getPoolThreadExecutor());
- CountDownLatch latch = new CountDownLatch(3);
- executor.execute(() -> {
- try {
- Thread.sleep(2000);
- latch.countDown();
- } catch (InterruptedException e) {
- }
- });
- Thread.sleep(10);
- executor.execute(() -> {
- try {
- new CompletableFuture<>().get(2, TimeUnit.SECONDS);
- } catch (InterruptedException | ExecutionException | TimeoutException e) {
- latch.countDown();
- }
- });
- Thread.sleep(10);
- executor.execute(latch::countDown);
- latch.await(1, TimeUnit.SECONDS);
- assertEquals(2, latch.getCount());
- latch.await(3, TimeUnit.SECONDS);
- assertEquals(0, latch.getCount());
- }
-
-}
diff --git a/utils/misc/src/test/java/org/onlab/util/BlockingAwareFutureTest.java b/utils/misc/src/test/java/org/onlab/util/BlockingAwareFutureTest.java
new file mode 100644
index 0000000..6a72b58
--- /dev/null
+++ b/utils/misc/src/test/java/org/onlab/util/BlockingAwareFutureTest.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright 2017-present Open Networking Laboratory
+ *
+ * Licensed 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.onlab.util;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Blocking-aware future test.
+ */
+public class BlockingAwareFutureTest {
+
+ /**
+ * Tests normal callback execution.
+ */
+ @Test
+ public void testNonBlockingThread() throws Exception {
+ CompletableFuture<String> future = new CompletableFuture<>();
+ Executor executor = SharedExecutors.getPoolThreadExecutor();
+ BlockingAwareFuture<String> blockingFuture =
+ (BlockingAwareFuture<String>) Tools.orderedFuture(future, new OrderedExecutor(executor), executor);
+ CountDownLatch latch = new CountDownLatch(1);
+ blockingFuture.thenRun(() -> latch.countDown());
+ executor.execute(() -> future.complete("foo"));
+ latch.await(5, TimeUnit.SECONDS);
+ assertEquals(0, latch.getCount());
+ assertEquals("foo", blockingFuture.join());
+ assertFalse(blockingFuture.isBlocked());
+ }
+
+ /**
+ * Tests blocking an ordered thread.
+ */
+ @Test
+ public void testBlockingThread() throws Exception {
+ CompletableFuture<String> future = new CompletableFuture<>();
+ Executor executor = SharedExecutors.getPoolThreadExecutor();
+ BlockingAwareFuture<String> blockingFuture =
+ (BlockingAwareFuture<String>) Tools.orderedFuture(future, new OrderedExecutor(executor), executor);
+ CountDownLatch latch = new CountDownLatch(2);
+ CompletableFuture<String> wrappedFuture = blockingFuture.thenApply(v -> {
+ assertEquals("foo", v);
+ latch.countDown();
+ return v;
+ });
+ wrappedFuture.thenRun(() -> latch.countDown());
+ executor.execute(() -> wrappedFuture.join());
+ Thread.sleep(100);
+ assertTrue(blockingFuture.isBlocked());
+ future.complete("foo");
+ latch.await(5, TimeUnit.SECONDS);
+ assertEquals(0, latch.getCount());
+ assertEquals("foo", blockingFuture.join());
+ assertEquals("foo", wrappedFuture.join());
+ assertFalse(blockingFuture.isBlocked());
+ }
+
+}
diff --git a/utils/misc/src/test/java/org/onlab/util/SerialExecutorTest.java b/utils/misc/src/test/java/org/onlab/util/OrderedExecutorTest.java
similarity index 87%
rename from utils/misc/src/test/java/org/onlab/util/SerialExecutorTest.java
rename to utils/misc/src/test/java/org/onlab/util/OrderedExecutorTest.java
index 140ce1f..b73fc56 100644
--- a/utils/misc/src/test/java/org/onlab/util/SerialExecutorTest.java
+++ b/utils/misc/src/test/java/org/onlab/util/OrderedExecutorTest.java
@@ -23,13 +23,13 @@
import static org.junit.Assert.assertEquals;
/**
- * Serial executor test.
+ * Ordered executor test.
*/
-public class SerialExecutorTest {
+public class OrderedExecutorTest {
@Test
public void testSerialExecution() throws Throwable {
- Executor executor = new SerialExecutor(SharedExecutors.getPoolThreadExecutor());
+ Executor executor = new OrderedExecutor(SharedExecutors.getPoolThreadExecutor());
CountDownLatch latch = new CountDownLatch(2);
executor.execute(latch::countDown);
executor.execute(latch::countDown);