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