View Javadoc
1   /*
2    * Copyright (C) 2006 The Guava Authors
3    *
4    * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
5    * in compliance with the License. You may obtain a copy of the License at
6    *
7    * http://www.apache.org/licenses/LICENSE-2.0
8    *
9    * Unless required by applicable law or agreed to in writing, software distributed under the License
10   * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
11   * or implied. See the License for the specific language governing permissions and limitations under
12   * the License.
13   */
14  
15  package com.google.common.util.concurrent;
16  
17  import static com.google.common.base.Preconditions.checkArgument;
18  import static com.google.common.base.Preconditions.checkNotNull;
19  
20  import com.google.common.annotations.Beta;
21  import com.google.common.annotations.GwtIncompatible;
22  import com.google.common.collect.ObjectArrays;
23  import com.google.common.collect.Sets;
24  import java.lang.reflect.InvocationHandler;
25  import java.lang.reflect.InvocationTargetException;
26  import java.lang.reflect.Method;
27  import java.lang.reflect.Proxy;
28  import java.util.Set;
29  import java.util.concurrent.Callable;
30  import java.util.concurrent.ExecutionException;
31  import java.util.concurrent.ExecutorService;
32  import java.util.concurrent.Executors;
33  import java.util.concurrent.Future;
34  import java.util.concurrent.TimeUnit;
35  import java.util.concurrent.TimeoutException;
36  
37  /**
38   * A TimeLimiter that runs method calls in the background using an {@link ExecutorService}. If the
39   * time limit expires for a given method call, the thread running the call will be interrupted.
40   *
41   * @author Kevin Bourrillion
42   * @author Jens Nyman
43   * @since 1.0
44   */
45  @Beta
46  @GwtIncompatible
47  public final class SimpleTimeLimiter implements TimeLimiter {
48  
49    private final ExecutorService executor;
50  
51    private
52    SimpleTimeLimiter(ExecutorService executor) {
53      this.executor = checkNotNull(executor);
54    }
55  
56    /**
57     * Creates a TimeLimiter instance using the given executor service to execute method calls.
58     *
59     * <p><b>Warning:</b> using a bounded executor may be counterproductive! If the thread pool fills
60     * up, any time callers spend waiting for a thread may count toward their time limit, and in this
61     * case the call may even time out before the target method is ever invoked.
62     *
63     * @param executor the ExecutorService that will execute the method calls on the target objects;
64     *     for example, a {@link Executors#newCachedThreadPool()}.
65     * @since 22.0
66     */
67    public static SimpleTimeLimiter create(ExecutorService executor) {
68      return new SimpleTimeLimiter(executor);
69    }
70  
71    @Override
72    public <T> T newProxy(
73        final T target,
74        Class<T> interfaceType,
75        final long timeoutDuration,
76        final TimeUnit timeoutUnit) {
77      checkNotNull(target);
78      checkNotNull(interfaceType);
79      checkNotNull(timeoutUnit);
80      checkPositiveTimeout(timeoutDuration);
81      checkArgument(interfaceType.isInterface(), "interfaceType must be an interface type");
82  
83      final Set<Method> interruptibleMethods = findInterruptibleMethods(interfaceType);
84  
85      InvocationHandler handler =
86          new InvocationHandler() {
87            @Override
88            public Object invoke(Object obj, final Method method, final Object[] args)
89                throws Throwable {
90              Callable<Object> callable =
91                  new Callable<Object>() {
92                    @Override
93                    public Object call() throws Exception {
94                      try {
95                        return method.invoke(target, args);
96                      } catch (InvocationTargetException e) {
97                        throw throwCause(e, false /* combineStackTraces */);
98                      }
99                    }
100                 };
101             return callWithTimeout(
102                 callable, timeoutDuration, timeoutUnit, interruptibleMethods.contains(method));
103           }
104         };
105     return newProxy(interfaceType, handler);
106   }
107 
108   private
109   <T> T callWithTimeout(
110       Callable<T> callable, long timeoutDuration, TimeUnit timeoutUnit, boolean amInterruptible)
111       throws Exception {
112     checkNotNull(callable);
113     checkNotNull(timeoutUnit);
114     checkPositiveTimeout(timeoutDuration);
115 
116     Future<T> future = executor.submit(callable);
117 
118     try {
119       if (amInterruptible) {
120         try {
121           return future.get(timeoutDuration, timeoutUnit);
122         } catch (InterruptedException e) {
123           future.cancel(true);
124           throw e;
125         }
126       } else {
127         return Uninterruptibles.getUninterruptibly(future, timeoutDuration, timeoutUnit);
128       }
129     } catch (ExecutionException e) {
130       throw throwCause(e, true /* combineStackTraces */);
131     } catch (TimeoutException e) {
132       future.cancel(true);
133       throw new UncheckedTimeoutException(e);
134     }
135   }
136 
137   @Override
138   public <T> T callWithTimeout(Callable<T> callable, long timeoutDuration, TimeUnit timeoutUnit)
139       throws TimeoutException, InterruptedException, ExecutionException {
140     checkNotNull(callable);
141     checkNotNull(timeoutUnit);
142     checkPositiveTimeout(timeoutDuration);
143 
144     Future<T> future = executor.submit(callable);
145 
146     try {
147       return future.get(timeoutDuration, timeoutUnit);
148     } catch (InterruptedException | TimeoutException e) {
149       future.cancel(true /* mayInterruptIfRunning */);
150       throw e;
151     } catch (ExecutionException e) {
152       wrapAndThrowExecutionExceptionOrError(e.getCause());
153       throw new AssertionError();
154     }
155   }
156 
157   @Override
158   public <T> T callUninterruptiblyWithTimeout(
159       Callable<T> callable, long timeoutDuration, TimeUnit timeoutUnit)
160       throws TimeoutException, ExecutionException {
161     checkNotNull(callable);
162     checkNotNull(timeoutUnit);
163     checkPositiveTimeout(timeoutDuration);
164 
165     Future<T> future = executor.submit(callable);
166 
167     try {
168       return Uninterruptibles.getUninterruptibly(future, timeoutDuration, timeoutUnit);
169     } catch (TimeoutException e) {
170       future.cancel(true /* mayInterruptIfRunning */);
171       throw e;
172     } catch (ExecutionException e) {
173       wrapAndThrowExecutionExceptionOrError(e.getCause());
174       throw new AssertionError();
175     }
176   }
177 
178   @Override
179   public void runWithTimeout(Runnable runnable, long timeoutDuration, TimeUnit timeoutUnit)
180       throws TimeoutException, InterruptedException {
181     checkNotNull(runnable);
182     checkNotNull(timeoutUnit);
183     checkPositiveTimeout(timeoutDuration);
184 
185     Future<?> future = executor.submit(runnable);
186 
187     try {
188       future.get(timeoutDuration, timeoutUnit);
189     } catch (InterruptedException | TimeoutException e) {
190       future.cancel(true /* mayInterruptIfRunning */);
191       throw e;
192     } catch (ExecutionException e) {
193       wrapAndThrowRuntimeExecutionExceptionOrError(e.getCause());
194       throw new AssertionError();
195     }
196   }
197 
198   @Override
199   public void runUninterruptiblyWithTimeout(
200       Runnable runnable, long timeoutDuration, TimeUnit timeoutUnit) throws TimeoutException {
201     checkNotNull(runnable);
202     checkNotNull(timeoutUnit);
203     checkPositiveTimeout(timeoutDuration);
204 
205     Future<?> future = executor.submit(runnable);
206 
207     try {
208       Uninterruptibles.getUninterruptibly(future, timeoutDuration, timeoutUnit);
209     } catch (TimeoutException e) {
210       future.cancel(true /* mayInterruptIfRunning */);
211       throw e;
212     } catch (ExecutionException e) {
213       wrapAndThrowRuntimeExecutionExceptionOrError(e.getCause());
214       throw new AssertionError();
215     }
216   }
217 
218   private static Exception throwCause(Exception e, boolean combineStackTraces) throws Exception {
219     Throwable cause = e.getCause();
220     if (cause == null) {
221       throw e;
222     }
223     if (combineStackTraces) {
224       StackTraceElement[] combined =
225           ObjectArrays.concat(cause.getStackTrace(), e.getStackTrace(), StackTraceElement.class);
226       cause.setStackTrace(combined);
227     }
228     if (cause instanceof Exception) {
229       throw (Exception) cause;
230     }
231     if (cause instanceof Error) {
232       throw (Error) cause;
233     }
234     // The cause is a weird kind of Throwable, so throw the outer exception.
235     throw e;
236   }
237 
238   private static Set<Method> findInterruptibleMethods(Class<?> interfaceType) {
239     Set<Method> set = Sets.newHashSet();
240     for (Method m : interfaceType.getMethods()) {
241       if (declaresInterruptedEx(m)) {
242         set.add(m);
243       }
244     }
245     return set;
246   }
247 
248   private static boolean declaresInterruptedEx(Method method) {
249     for (Class<?> exType : method.getExceptionTypes()) {
250       // debate: == or isAssignableFrom?
251       if (exType == InterruptedException.class) {
252         return true;
253       }
254     }
255     return false;
256   }
257 
258   // TODO: replace with version in common.reflect if and when it's open-sourced
259   private static <T> T newProxy(Class<T> interfaceType, InvocationHandler handler) {
260     Object object =
261         Proxy.newProxyInstance(
262             interfaceType.getClassLoader(), new Class<?>[] {interfaceType}, handler);
263     return interfaceType.cast(object);
264   }
265 
266   private void wrapAndThrowExecutionExceptionOrError(Throwable cause) throws ExecutionException {
267     if (cause instanceof Error) {
268       throw new ExecutionError((Error) cause);
269     } else if (cause instanceof RuntimeException) {
270       throw new UncheckedExecutionException(cause);
271     } else {
272       throw new ExecutionException(cause);
273     }
274   }
275 
276   private void wrapAndThrowRuntimeExecutionExceptionOrError(Throwable cause) {
277     if (cause instanceof Error) {
278       throw new ExecutionError((Error) cause);
279     } else {
280       throw new UncheckedExecutionException(cause);
281     }
282   }
283 
284   private static void checkPositiveTimeout(long timeoutDuration) {
285     checkArgument(timeoutDuration > 0, "timeout must be positive: %s", timeoutDuration);
286   }
287 }