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
9 package org.opendaylight.yangtools.util.concurrent;
11 import static org.junit.Assert.assertEquals;
12 import static org.junit.Assert.assertNotNull;
13 import static org.junit.Assert.assertTrue;
14 import static org.opendaylight.yangtools.util.concurrent.AsyncNotifyingListeningExecutorServiceTest.testListenerCallback;
15 import static org.opendaylight.yangtools.util.concurrent.CommonTestUtils.SUBMIT_CALLABLE;
16 import static org.opendaylight.yangtools.util.concurrent.CommonTestUtils.SUBMIT_RUNNABLE;
17 import static org.opendaylight.yangtools.util.concurrent.CommonTestUtils.SUBMIT_RUNNABLE_WITH_RESULT;
19 import com.google.common.base.Supplier;
20 import com.google.common.util.concurrent.FutureCallback;
21 import com.google.common.util.concurrent.Futures;
22 import com.google.common.util.concurrent.ListenableFuture;
23 import com.google.common.util.concurrent.ListeningExecutorService;
24 import com.google.common.util.concurrent.ThreadFactoryBuilder;
25 import java.util.concurrent.CountDownLatch;
26 import java.util.concurrent.ExecutionException;
27 import java.util.concurrent.Executor;
28 import java.util.concurrent.ExecutorService;
29 import java.util.concurrent.Executors;
30 import java.util.concurrent.TimeUnit;
31 import java.util.concurrent.atomic.AtomicReference;
32 import org.junit.After;
33 import org.junit.Before;
34 import org.junit.Test;
35 import org.opendaylight.yangtools.util.concurrent.CommonTestUtils.Invoker;
38 * Unit tests for DeadlockDetectingListeningExecutorService.
40 * @author Thomas Pantelis
42 public class DeadlockDetectingListeningExecutorServiceTest {
44 interface InitialInvoker {
45 void invokeExecutor( ListeningExecutorService executor, Runnable task );
48 static final InitialInvoker SUBMIT = ListeningExecutorService::submit;
50 static final InitialInvoker EXECUTE = Executor::execute;
52 @SuppressWarnings("serial")
53 public static class TestDeadlockException extends Exception {
56 private static final Supplier<Exception> DEADLOCK_EXECUTOR_SUPPLIER = TestDeadlockException::new;
58 DeadlockDetectingListeningExecutorService executor;
65 public void tearDown() {
66 if (executor != null ) {
67 executor.shutdownNow();
71 DeadlockDetectingListeningExecutorService newExecutor() {
72 return new DeadlockDetectingListeningExecutorService( Executors.newSingleThreadExecutor(),
73 DEADLOCK_EXECUTOR_SUPPLIER );
77 public void testBlockingSubmitOffExecutor() throws Exception {
79 executor = newExecutor();
81 // Test submit with Callable.
83 ListenableFuture<String> future = executor.submit(() -> "foo");
85 assertEquals( "Future result", "foo", future.get( 5, TimeUnit.SECONDS ) );
87 // Test submit with Runnable.
89 executor.submit(() -> {
92 // Test submit with Runnable and value.
94 future = executor.submit(() -> {
97 assertEquals( "Future result", "foo", future.get( 5, TimeUnit.SECONDS ) );
101 public void testNonBlockingSubmitOnExecutorThread() throws Throwable {
103 executor = newExecutor();
105 testNonBlockingSubmitOnExecutorThread( SUBMIT, SUBMIT_CALLABLE );
106 testNonBlockingSubmitOnExecutorThread( SUBMIT, SUBMIT_RUNNABLE );
107 testNonBlockingSubmitOnExecutorThread( SUBMIT, SUBMIT_RUNNABLE_WITH_RESULT );
109 testNonBlockingSubmitOnExecutorThread( EXECUTE, SUBMIT_CALLABLE );
112 void testNonBlockingSubmitOnExecutorThread( final InitialInvoker initialInvoker,
113 final Invoker invoker ) throws Throwable {
115 final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
116 final CountDownLatch futureCompletedLatch = new CountDownLatch( 1 );
118 Runnable task = () -> Futures.addCallback( invoker.invokeExecutor( executor, null ), new FutureCallback() {
120 public void onSuccess( final Object result ) {
121 futureCompletedLatch.countDown();
125 public void onFailure( final Throwable t ) {
127 futureCompletedLatch.countDown();
131 initialInvoker.invokeExecutor( executor, task );
133 assertTrue( "Task did not complete - executor likely deadlocked",
134 futureCompletedLatch.await( 5, TimeUnit.SECONDS ) );
136 if (caughtEx.get() != null ) {
137 throw caughtEx.get();
142 public void testBlockingSubmitOnExecutorThread() throws Exception {
144 executor = newExecutor();
146 testBlockingSubmitOnExecutorThread( SUBMIT, SUBMIT_CALLABLE );
147 testBlockingSubmitOnExecutorThread( SUBMIT, SUBMIT_RUNNABLE );
148 testBlockingSubmitOnExecutorThread( SUBMIT, SUBMIT_RUNNABLE_WITH_RESULT );
150 testBlockingSubmitOnExecutorThread( EXECUTE, SUBMIT_CALLABLE );
153 void testBlockingSubmitOnExecutorThread( final InitialInvoker initialInvoker,
154 final Invoker invoker ) throws Exception {
156 final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
157 final CountDownLatch latch = new CountDownLatch( 1 );
159 Runnable task = () -> {
162 invoker.invokeExecutor( executor, null ).get();
163 } catch( ExecutionException e ) {
164 caughtEx.set( e.getCause() );
165 } catch( Throwable e ) {
172 initialInvoker.invokeExecutor( executor, task );
174 assertTrue( "Task did not complete - executor likely deadlocked",
175 latch.await( 5, TimeUnit.SECONDS ) );
177 assertNotNull( "Expected exception thrown", caughtEx.get() );
178 assertEquals( "Caught exception type", TestDeadlockException.class, caughtEx.get().getClass() );
182 public void testListenableFutureCallbackWithExecutor() throws InterruptedException {
184 String listenerThreadPrefix = "ListenerThread";
185 ExecutorService listenerExecutor = Executors.newFixedThreadPool( 1,
186 new ThreadFactoryBuilder().setNameFormat( listenerThreadPrefix + "-%d" ).build() );
188 executor = new DeadlockDetectingListeningExecutorService(
189 Executors.newSingleThreadExecutor(
190 new ThreadFactoryBuilder().setNameFormat( "SingleThread" ).build() ),
191 DEADLOCK_EXECUTOR_SUPPLIER, listenerExecutor );
194 testListenerCallback( executor, SUBMIT_CALLABLE, listenerThreadPrefix );
195 testListenerCallback( executor, SUBMIT_RUNNABLE, listenerThreadPrefix );
196 testListenerCallback( executor, SUBMIT_RUNNABLE_WITH_RESULT, listenerThreadPrefix );
198 listenerExecutor.shutdownNow();