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