common/util/src/test/java/org/opendaylight/yangtools/util/concurrent/DeadlockDetectingListeningExecutorServiceTest.java

   1 /*
   2  * Copyright (c) 2014 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8
   9 package org.opendaylight.yangtools.util.concurrent;
  10
  11 import static org.junit.Assert.*;
  12
  13 import java.util.concurrent.Callable;
  14 import java.util.concurrent.CountDownLatch;
  15 import java.util.concurrent.ExecutionException;
  16 import java.util.concurrent.ExecutorService;
  17 import java.util.concurrent.Executors;
  18 import java.util.concurrent.TimeUnit;
  19 import java.util.concurrent.atomic.AtomicReference;
  20
  21 import org.junit.After;
  22 import org.junit.Before;
  23 import org.junit.Test;
  24
  25 import com.google.common.base.Function;
  26 import com.google.common.util.concurrent.FutureCallback;
  27 import com.google.common.util.concurrent.Futures;
  28 import com.google.common.util.concurrent.ListenableFuture;
  29 import com.google.common.util.concurrent.ListeningExecutorService;
  30 import com.google.common.util.concurrent.ThreadFactoryBuilder;
  31
  32 import static org.opendaylight.yangtools.util.concurrent.AsyncNotifyingListeningExecutorServiceTest.testListenerCallback;
  33 import static org.opendaylight.yangtools.util.concurrent.CommonTestUtils.Invoker;
  34 import static org.opendaylight.yangtools.util.concurrent.CommonTestUtils.SUBMIT_CALLABLE;
  35 import static org.opendaylight.yangtools.util.concurrent.CommonTestUtils.SUBMIT_RUNNABLE;
  36 import static org.opendaylight.yangtools.util.concurrent.CommonTestUtils.SUBMIT_RUNNABLE_WITH_RESULT;
  37
  38 /**
  39  * Unit tests for DeadlockDetectingListeningExecutorService.
  40  *
  41  * @author Thomas Pantelis
  42  */
  43 public class DeadlockDetectingListeningExecutorServiceTest {
  44
  45     interface InitialInvoker {
  46         void invokeExecutor( ListeningExecutorService executor, Runnable task );
  47     };
  48
  49     static final InitialInvoker SUBMIT = new InitialInvoker() {
  50         @Override
  51         public void invokeExecutor( ListeningExecutorService executor, Runnable task ) {
  52             executor.submit( task );
  53         }
  54     };
  55
  56     static final InitialInvoker EXECUTE = new InitialInvoker() {
  57         @Override
  58         public void invokeExecutor( ListeningExecutorService executor, Runnable task ) {
  59             executor.execute( task );
  60         }
  61     };
  62
  63     @SuppressWarnings("serial")
  64     public static class TestDeadlockException extends Exception {
  65     }
  66
  67     public static Function<Void, Exception> DEADLOCK_EXECUTOR_FUNCTION = new Function<Void, Exception>() {
  68         @Override
  69         public Exception apply( Void notUsed ) {
  70             return new TestDeadlockException();
  71         }
  72     };
  73
  74     DeadlockDetectingListeningExecutorService executor;
  75
  76     @Before
  77     public void setup() {
  78     }
  79
  80     @After
  81     public void tearDown() {
  82         if( executor != null ) {
  83             executor.shutdownNow();
  84         }
  85     }
  86
  87     DeadlockDetectingListeningExecutorService newExecutor() {
  88         return new DeadlockDetectingListeningExecutorService( Executors.newSingleThreadExecutor(),
  89                 DEADLOCK_EXECUTOR_FUNCTION );
  90     }
  91
  92     @Test
  93     public void testBlockingSubmitOffExecutor() throws Exception {
  94
  95         executor = newExecutor();
  96
  97         // Test submit with Callable.
  98
  99         ListenableFuture<String> future = executor.submit( new Callable<String>() {
 100             @Override
 101             public String call() throws Exception{
 102                 return "foo";
 103             }
 104         } );
 105
 106         assertEquals( "Future result", "foo", future.get( 5, TimeUnit.SECONDS ) );
 107
 108         // Test submit with Runnable.
 109
 110         executor.submit( new Runnable() {
 111             @Override
 112             public void run(){
 113             }
 114         } ).get();
 115
 116         // Test submit with Runnable and value.
 117
 118         future = executor.submit( new Runnable() {
 119             @Override
 120             public void run(){
 121             }
 122         }, "foo" );
 123
 124         assertEquals( "Future result", "foo", future.get( 5, TimeUnit.SECONDS ) );
 125     }
 126
 127     @Test
 128     public void testNonBlockingSubmitOnExecutorThread() throws Throwable {
 129
 130         executor = newExecutor();
 131
 132         testNonBlockingSubmitOnExecutorThread( SUBMIT, SUBMIT_CALLABLE );
 133         testNonBlockingSubmitOnExecutorThread( SUBMIT, SUBMIT_RUNNABLE );
 134         testNonBlockingSubmitOnExecutorThread( SUBMIT, SUBMIT_RUNNABLE_WITH_RESULT );
 135
 136         testNonBlockingSubmitOnExecutorThread( EXECUTE, SUBMIT_CALLABLE );
 137     }
 138
 139     void testNonBlockingSubmitOnExecutorThread( InitialInvoker initialInvoker,
 140                                                 final Invoker invoker ) throws Throwable {
 141
 142         final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
 143         final CountDownLatch futureCompletedLatch = new CountDownLatch( 1 );
 144
 145         Runnable task = new Runnable() {
 146             @SuppressWarnings({ "unchecked", "rawtypes" })
 147             @Override
 148             public void run() {
 149
 150                 Futures.addCallback( invoker.invokeExecutor( executor, null ), new FutureCallback() {
 151                     @Override
 152                     public void onSuccess( Object result ) {
 153                         futureCompletedLatch.countDown();
 154                     }
 155
 156                     @Override
 157                     public void onFailure( Throwable t ) {
 158                         caughtEx.set( t );
 159                         futureCompletedLatch.countDown();
 160                     }
 161                 } );
 162             }
 163
 164         };
 165
 166         initialInvoker.invokeExecutor( executor, task );
 167
 168         assertTrue( "Task did not complete - executor likely deadlocked",
 169                     futureCompletedLatch.await( 5, TimeUnit.SECONDS ) );
 170
 171         if( caughtEx.get() != null ) {
 172             throw caughtEx.get();
 173         }
 174     }
 175
 176     @Test
 177     public void testBlockingSubmitOnExecutorThread() throws Exception {
 178
 179         executor = newExecutor();
 180
 181         testBlockingSubmitOnExecutorThread( SUBMIT, SUBMIT_CALLABLE );
 182         testBlockingSubmitOnExecutorThread( SUBMIT, SUBMIT_RUNNABLE );
 183         testBlockingSubmitOnExecutorThread( SUBMIT, SUBMIT_RUNNABLE_WITH_RESULT );
 184
 185         testBlockingSubmitOnExecutorThread( EXECUTE, SUBMIT_CALLABLE );
 186     }
 187
 188     void testBlockingSubmitOnExecutorThread( InitialInvoker initialInvoker,
 189                                              final Invoker invoker ) throws Exception {
 190
 191         final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
 192         final CountDownLatch latch = new CountDownLatch( 1 );
 193
 194         Runnable task = new Runnable() {
 195             @Override
 196             public void run() {
 197
 198                 try {
 199                     invoker.invokeExecutor( executor, null ).get();
 200                 } catch( ExecutionException e ) {
 201                     caughtEx.set( e.getCause() );
 202                 } catch( Throwable e ) {
 203                     caughtEx.set( e );
 204                 } finally {
 205                     latch.countDown();
 206                 }
 207             }
 208
 209         };
 210
 211         initialInvoker.invokeExecutor( executor, task );
 212
 213         assertTrue( "Task did not complete - executor likely deadlocked",
 214                     latch.await( 5, TimeUnit.SECONDS ) );
 215
 216         assertNotNull( "Expected exception thrown", caughtEx.get() );
 217         assertEquals( "Caught exception type", TestDeadlockException.class, caughtEx.get().getClass() );
 218     }
 219
 220     @Test
 221     public void testListenableFutureCallbackWithExecutor() throws InterruptedException {
 222
 223         String listenerThreadPrefix = "ListenerThread";
 224         ExecutorService listenerExecutor = Executors.newFixedThreadPool( 1,
 225                 new ThreadFactoryBuilder().setNameFormat( listenerThreadPrefix + "-%d" ).build() );
 226
 227         executor = new DeadlockDetectingListeningExecutorService(
 228                 Executors.newSingleThreadExecutor(
 229                         new ThreadFactoryBuilder().setNameFormat( "SingleThread" ).build() ),
 230                 DEADLOCK_EXECUTOR_FUNCTION, listenerExecutor );
 231
 232         try {
 233             testListenerCallback( executor, SUBMIT_CALLABLE, listenerThreadPrefix );
 234             testListenerCallback( executor, SUBMIT_RUNNABLE, listenerThreadPrefix );
 235             testListenerCallback( executor, SUBMIT_RUNNABLE_WITH_RESULT, listenerThreadPrefix );
 236         } finally {
 237             listenerExecutor.shutdownNow();
 238         }
 239     }
 240 }