2 * Copyright (c) 2014 Brocade Communications Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.yangtools.util.concurrent;
10 import static org.junit.jupiter.api.Assertions.assertEquals;
11 import static org.junit.jupiter.api.Assertions.assertNotNull;
12 import static org.junit.jupiter.api.Assertions.assertTrue;
13 import static org.opendaylight.yangtools.util.concurrent.AsyncNotifyingListeningExecutorServiceTest.testListenerCallback;
14 import static org.opendaylight.yangtools.util.concurrent.CommonTestUtils.SUBMIT_CALLABLE;
15 import static org.opendaylight.yangtools.util.concurrent.CommonTestUtils.SUBMIT_RUNNABLE;
16 import static org.opendaylight.yangtools.util.concurrent.CommonTestUtils.SUBMIT_RUNNABLE_WITH_RESULT;
18 import com.google.common.util.concurrent.FutureCallback;
19 import com.google.common.util.concurrent.Futures;
20 import com.google.common.util.concurrent.ListeningExecutorService;
21 import com.google.common.util.concurrent.MoreExecutors;
22 import com.google.common.util.concurrent.ThreadFactoryBuilder;
23 import java.util.concurrent.CountDownLatch;
24 import java.util.concurrent.ExecutionException;
25 import java.util.concurrent.Executor;
26 import java.util.concurrent.ExecutorService;
27 import java.util.concurrent.Executors;
28 import java.util.concurrent.TimeUnit;
29 import java.util.concurrent.atomic.AtomicReference;
30 import java.util.function.Supplier;
31 import org.junit.jupiter.api.AfterEach;
32 import org.junit.jupiter.api.Test;
33 import org.opendaylight.yangtools.util.concurrent.CommonTestUtils.Invoker;
36 * Unit tests for DeadlockDetectingListeningExecutorService.
38 * @author Thomas Pantelis
40 class DeadlockDetectingListeningExecutorServiceTest {
42 interface InitialInvoker {
43 void invokeExecutor(ListeningExecutorService executor, Runnable task);
46 static final InitialInvoker SUBMIT = ListeningExecutorService::submit;
48 static final InitialInvoker EXECUTE = Executor::execute;
50 public static class TestDeadlockException extends Exception {
52 private static final long serialVersionUID = 1L;
55 private static final Supplier<Exception> DEADLOCK_EXECUTOR_SUPPLIER = TestDeadlockException::new;
57 DeadlockDetectingListeningExecutorService executor;
61 if (executor != null) {
62 executor.shutdownNow();
66 DeadlockDetectingListeningExecutorService newExecutor() {
67 return new DeadlockDetectingListeningExecutorService(Executors.newSingleThreadExecutor(),
68 DEADLOCK_EXECUTOR_SUPPLIER);
72 void testBlockingSubmitOffExecutor() throws Exception {
74 executor = newExecutor();
76 // Test submit with Callable.
78 var future = executor.submit(() -> "foo");
80 assertEquals("foo", future.get(5, TimeUnit.SECONDS), "Future result");
82 // Test submit with Runnable.
84 executor.submit(() -> { }).get();
86 // Test submit with Runnable and value.
88 future = executor.submit(() -> { }, "foo");
90 assertEquals("foo", future.get(5, TimeUnit.SECONDS), "Future result");
94 @SuppressWarnings("checkstyle:illegalThrows")
95 void testNonBlockingSubmitOnExecutorThread() throws Throwable {
97 executor = newExecutor();
99 testNonBlockingSubmitOnExecutorThread(SUBMIT, SUBMIT_CALLABLE);
100 testNonBlockingSubmitOnExecutorThread(SUBMIT, SUBMIT_RUNNABLE);
101 testNonBlockingSubmitOnExecutorThread(SUBMIT, SUBMIT_RUNNABLE_WITH_RESULT);
103 testNonBlockingSubmitOnExecutorThread(EXECUTE, SUBMIT_CALLABLE);
106 @SuppressWarnings("checkstyle:illegalThrows")
107 void testNonBlockingSubmitOnExecutorThread(final InitialInvoker initialInvoker, final Invoker invoker)
110 final var caughtEx = new AtomicReference<Throwable>();
111 final var futureCompletedLatch = new CountDownLatch(1);
113 final var task = (Runnable) () -> Futures.addCallback(invoker.invokeExecutor(executor, null),
114 new FutureCallback<Object>() {
116 public void onSuccess(final Object result) {
117 futureCompletedLatch.countDown();
121 @SuppressWarnings("checkstyle:parameterName")
122 public void onFailure(final Throwable t) {
124 futureCompletedLatch.countDown();
126 }, MoreExecutors.directExecutor());
128 initialInvoker.invokeExecutor(executor, task);
130 assertTrue(futureCompletedLatch.await(5, TimeUnit.SECONDS),
131 "Task did not complete - executor likely deadlocked");
133 if (caughtEx.get() != null) {
134 throw caughtEx.get();
139 void testBlockingSubmitOnExecutorThread() throws InterruptedException {
141 executor = newExecutor();
143 testBlockingSubmitOnExecutorThread(SUBMIT, SUBMIT_CALLABLE);
144 testBlockingSubmitOnExecutorThread(SUBMIT, SUBMIT_RUNNABLE);
145 testBlockingSubmitOnExecutorThread(SUBMIT, SUBMIT_RUNNABLE_WITH_RESULT);
147 testBlockingSubmitOnExecutorThread(EXECUTE, SUBMIT_CALLABLE);
150 @SuppressWarnings("checkstyle:illegalCatch")
151 void testBlockingSubmitOnExecutorThread(final InitialInvoker initialInvoker, final Invoker invoker)
152 throws InterruptedException {
154 final var caughtEx = new AtomicReference<Throwable>();
155 final var latch = new CountDownLatch(1);
157 final var task = (Runnable) () -> {
160 invoker.invokeExecutor(executor, null).get();
161 } catch (ExecutionException e) {
162 caughtEx.set(e.getCause());
163 } catch (Throwable e) {
170 initialInvoker.invokeExecutor(executor, task);
172 assertTrue(latch.await(5, TimeUnit.SECONDS), "Task did not complete - executor likely deadlocked");
173 assertNotNull(caughtEx.get(), "Expected exception thrown");
174 assertEquals(TestDeadlockException.class, caughtEx.get().getClass(), "Caught exception type");
178 void testListenableFutureCallbackWithExecutor() throws InterruptedException {
180 final var listenerThreadPrefix = "ListenerThread";
181 ExecutorService listenerExecutor = Executors.newFixedThreadPool(1,
182 new ThreadFactoryBuilder().setNameFormat(listenerThreadPrefix + "-%d").build());
184 executor = new DeadlockDetectingListeningExecutorService(
185 Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("SingleThread").build()),
186 DEADLOCK_EXECUTOR_SUPPLIER, listenerExecutor);
189 testListenerCallback(executor, SUBMIT_CALLABLE, listenerThreadPrefix);
190 testListenerCallback(executor, SUBMIT_RUNNABLE, listenerThreadPrefix);
191 testListenerCallback(executor, SUBMIT_RUNNABLE_WITH_RESULT, listenerThreadPrefix);
193 listenerExecutor.shutdownNow();