common/util/src/main/java/org/opendaylight/yangtools/util/concurrent/DeadlockDetectingListeningExecutorService.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 com.google.common.base.Preconditions.checkNotNull;
  12
  13 import com.google.common.base.Function;
  14 import com.google.common.util.concurrent.ForwardingListenableFuture;
  15 import com.google.common.util.concurrent.ListenableFuture;
  16 import java.util.concurrent.Callable;
  17 import java.util.concurrent.ExecutionException;
  18 import java.util.concurrent.Executor;
  19 import java.util.concurrent.ExecutorService;
  20 import java.util.concurrent.TimeUnit;
  21 import java.util.concurrent.TimeoutException;
  22
  23 import javax.annotation.Nullable;
  24
  25 /**
  26  * An implementation of ListeningExecutorService that attempts to detect deadlock scenarios that
  27  * could occur if clients invoke the returned Future's <ode>get</code> methods synchronously.
  28  * <p>
  29  * Deadlock scenarios are most apt to occur with a backing single-threaded executor where setting of
  30  * the Future's result is executed on the single thread. Here's a scenario:
  31  * <ul>
  32  * <li>Client code is currently executing in an executor's single thread.</li>
  33  * <li>The client submits another task to the same executor.</li>
  34  * <li>The client calls <code>get()</code> synchronously on the returned Future</li>
  35  * </ul>
  36  * The second submitted task will never execute since the single thread is currently executing
  37  * the client code which is blocked waiting for the submitted task to complete. Thus, deadlock has
  38  * occurred.
  39  * <p>
  40  * This class prevents this scenario via the use of a ThreadLocal variable. When a task is invoked,
  41  * the ThreadLocal is set and, when a task completes, the ThreadLocal is cleared. Futures returned
  42  * from this class override the <code>get</code> methods to check if the ThreadLocal is set. If it is,
  43  * an ExecutionException is thrown with a custom cause.
  44  *
  45  * @author Thomas Pantelis
  46  */
  47 public class DeadlockDetectingListeningExecutorService extends AsyncNotifyingListeningExecutorService {
  48     private final ThreadLocal<Boolean> deadlockDetector = new ThreadLocal<>();
  49     private final Function<Void, Exception> deadlockExceptionFunction;
  50
  51     /**
  52      * Constructor.
  53      *
  54      * @param delegate the backing ExecutorService.
  55      * @param deadlockExceptionFunction Function that returns an Exception instance to set as the
  56      *             cause of the ExecutionException when a deadlock is detected.
  57      */
  58     public DeadlockDetectingListeningExecutorService( ExecutorService delegate,
  59                                           Function<Void,Exception> deadlockExceptionFunction ) {
  60         this(delegate, deadlockExceptionFunction, null);
  61     }
  62
  63     /**
  64      * Constructor.
  65      *
  66      * @param delegate the backing ExecutorService.
  67      * @param deadlockExceptionFunction Function that returns an Exception instance to set as the
  68      *             cause of the ExecutionException when a deadlock is detected.
  69      * @param listenableFutureExecutor the executor used to run listener callbacks asynchronously.
  70      *             If null, no executor is used.
  71      */
  72     public DeadlockDetectingListeningExecutorService( ExecutorService delegate,
  73                                           Function<Void,Exception> deadlockExceptionFunction,
  74                                           @Nullable Executor listenableFutureExecutor ) {
  75         super(delegate, listenableFutureExecutor);
  76         this.deadlockExceptionFunction = checkNotNull(deadlockExceptionFunction);
  77     }
  78
  79     @Override
  80     public void execute( Runnable command ){
  81         getDelegate().execute(wrapRunnable(command));
  82     }
  83
  84     @Override
  85     public <T> ListenableFuture<T> submit( Callable<T> task ){
  86         return wrapListenableFuture(super.submit(wrapCallable(task)));
  87     }
  88
  89     @Override
  90     public ListenableFuture<?> submit( Runnable task ){
  91         return wrapListenableFuture(super.submit(wrapRunnable(task)));
  92     }
  93
  94     @Override
  95     public <T> ListenableFuture<T> submit( Runnable task, T result ){
  96         return wrapListenableFuture(super.submit(wrapRunnable(task), result));
  97     }
  98
  99     private Runnable wrapRunnable(final Runnable task) {
 100         return new Runnable() {
 101             @Override
 102             public void run() {
 103                 deadlockDetector.set(Boolean.TRUE);
 104                 try {
 105                     task.run();
 106                 } finally {
 107                     deadlockDetector.remove();
 108                 }
 109             }
 110         };
 111     }
 112
 113     private <T> Callable<T> wrapCallable(final Callable<T> delagate) {
 114         return new Callable<T>() {
 115             @Override
 116             public T call() throws Exception {
 117                 deadlockDetector.set(Boolean.TRUE);
 118                 try {
 119                     return delagate.call();
 120                 } finally {
 121                     deadlockDetector.remove();
 122                 }
 123             }
 124         };
 125     }
 126
 127     private <T> ListenableFuture<T> wrapListenableFuture(final ListenableFuture<T> delegate ) {
 128         /*
 129          * This creates a forwarding Future that overrides calls to get(...) to check, via the
 130          * ThreadLocal, if the caller is doing a blocking call on a thread from this executor. If
 131          * so, we detect this as a deadlock and throw an ExecutionException even though it may not
 132          * be a deadlock if there are more than 1 thread in the pool. Either way, there's bad
 133          * practice somewhere, either on the client side for doing a blocking call or in the
 134          * framework's threading model.
 135          */
 136         return new ForwardingListenableFuture.SimpleForwardingListenableFuture<T>(delegate) {
 137             @Override
 138             public T get() throws InterruptedException, ExecutionException {
 139                 checkDeadLockDetectorTL();
 140                 return super.get();
 141             }
 142
 143             @Override
 144             public T get(final long timeout, final TimeUnit unit)
 145                     throws InterruptedException, ExecutionException, TimeoutException {
 146                 checkDeadLockDetectorTL();
 147                 return super.get(timeout, unit);
 148             }
 149
 150             void checkDeadLockDetectorTL() throws ExecutionException {
 151                 if (deadlockDetector.get() != null) {
 152                     throw new ExecutionException("A potential deadlock was detected.",
 153                             deadlockExceptionFunction.apply(null));
 154                 }
 155             }
 156         };
 157     }
 158 }