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