libcore/luni/src/main/java/java/util/concurrent/AbstractExecutorService.java

   1 /*
   2  * Written by Doug Lea with assistance from members of JCP JSR-166
   3  * Expert Group and released to the public domain, as explained at
   4  * http://creativecommons.org/licenses/publicdomain
   5  */
   6
   7 package java.util.concurrent;
   8
   9 import java.util.ArrayList;
  10 import java.util.Collection;
  11 import java.util.Iterator;
  12 import java.util.List;
  13
  14 /**
  15  * Provides default implementations of {@link ExecutorService}
  16  * execution methods. This class implements the <tt>submit</tt>,
  17  * <tt>invokeAny</tt> and <tt>invokeAll</tt> methods using a
  18  * {@link RunnableFuture} returned by <tt>newTaskFor</tt>, which defaults
  19  * to the {@link FutureTask} class provided in this package.  For example,
  20  * the implementation of <tt>submit(Runnable)</tt> creates an
  21  * associated <tt>RunnableFuture</tt> that is executed and
  22  * returned. Subclasses may override the <tt>newTaskFor</tt> methods
  23  * to return <tt>RunnableFuture</tt> implementations other than
  24  * <tt>FutureTask</tt>.
  25  *
  26  * <p> <b>Extension example</b>. Here is a sketch of a class
  27  * that customizes {@link ThreadPoolExecutor} to use
  28  * a <tt>CustomTask</tt> class instead of the default <tt>FutureTask</tt>:
  29  * <pre>
  30  * public class CustomThreadPoolExecutor extends ThreadPoolExecutor {
  31  *
  32  *   static class CustomTask&lt;V&gt; implements RunnableFuture&lt;V&gt; {...}
  33  *
  34  *   protected &lt;V&gt; RunnableFuture&lt;V&gt; newTaskFor(Callable&lt;V&gt; c) {
  35  *       return new CustomTask&lt;V&gt;(c);
  36  *   }
  37  *   protected &lt;V&gt; RunnableFuture&lt;V&gt; newTaskFor(Runnable r, V v) {
  38  *       return new CustomTask&lt;V&gt;(r, v);
  39  *   }
  40  *   // ... add constructors, etc.
  41  * }
  42  * </pre>
  43  * @since 1.5
  44  * @author Doug Lea
  45  */
  46 public abstract class AbstractExecutorService implements ExecutorService {
  47
  48     /**
  49      * Returns a <tt>RunnableFuture</tt> for the given runnable and default
  50      * value.
  51      *
  52      * @param runnable the runnable task being wrapped
  53      * @param value the default value for the returned future
  54      * @return a <tt>RunnableFuture</tt> which when run will run the
  55      * underlying runnable and which, as a <tt>Future</tt>, will yield
  56      * the given value as its result and provide for cancellation of
  57      * the underlying task.
  58      * @since 1.6
  59      */
  60     protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
  61         return new FutureTask<T>(runnable, value);
  62     }
  63
  64     /**
  65      * Returns a <tt>RunnableFuture</tt> for the given callable task.
  66      *
  67      * @param callable the callable task being wrapped
  68      * @return a <tt>RunnableFuture</tt> which when run will call the
  69      * underlying callable and which, as a <tt>Future</tt>, will yield
  70      * the callable's result as its result and provide for
  71      * cancellation of the underlying task.
  72      * @since 1.6
  73      */
  74     protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
  75         return new FutureTask<T>(callable);
  76     }
  77
  78     /**
  79      * @throws RejectedExecutionException {@inheritDoc}
  80      * @throws NullPointerException       {@inheritDoc}
  81      */
  82     public Future<?> submit(Runnable task) {
  83         if (task == null) throw new NullPointerException();
  84         RunnableFuture<Object> ftask = newTaskFor(task, null);
  85         execute(ftask);
  86         return ftask;
  87     }
  88
  89     /**
  90      * @throws RejectedExecutionException {@inheritDoc}
  91      * @throws NullPointerException       {@inheritDoc}
  92      */
  93     public <T> Future<T> submit(Runnable task, T result) {
  94         if (task == null) throw new NullPointerException();
  95         RunnableFuture<T> ftask = newTaskFor(task, result);
  96         execute(ftask);
  97         return ftask;
  98     }
  99
 100     /**
 101      * @throws RejectedExecutionException {@inheritDoc}
 102      * @throws NullPointerException       {@inheritDoc}
 103      */
 104     public <T> Future<T> submit(Callable<T> task) {
 105         if (task == null) throw new NullPointerException();
 106         RunnableFuture<T> ftask = newTaskFor(task);
 107         execute(ftask);
 108         return ftask;
 109     }
 110
 111     /**
 112      * the main mechanics of invokeAny.
 113      */
 114     private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
 115                             boolean timed, long nanos)
 116         throws InterruptedException, ExecutionException, TimeoutException {
 117         if (tasks == null)
 118             throw new NullPointerException();
 119         int ntasks = tasks.size();
 120         if (ntasks == 0)
 121             throw new IllegalArgumentException();
 122         List<Future<T>> futures= new ArrayList<Future<T>>(ntasks);
 123         ExecutorCompletionService<T> ecs =
 124             new ExecutorCompletionService<T>(this);
 125
 126         // For efficiency, especially in executors with limited
 127         // parallelism, check to see if previously submitted tasks are
 128         // done before submitting more of them. This interleaving
 129         // plus the exception mechanics account for messiness of main
 130         // loop.
 131
 132         try {
 133             // Record exceptions so that if we fail to obtain any
 134             // result, we can throw the last exception we got.
 135             ExecutionException ee = null;
 136             long lastTime = (timed)? System.nanoTime() : 0;
 137             Iterator<? extends Callable<T>> it = tasks.iterator();
 138
 139             // Start one task for sure; the rest incrementally
 140             futures.add(ecs.submit(it.next()));
 141             --ntasks;
 142             int active = 1;
 143
 144             for (;;) {
 145                 Future<T> f = ecs.poll();
 146                 if (f == null) {
 147                     if (ntasks > 0) {
 148                         --ntasks;
 149                         futures.add(ecs.submit(it.next()));
 150                         ++active;
 151                     }
 152                     else if (active == 0)
 153                         break;
 154                     else if (timed) {
 155                         f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
 156                         if (f == null)
 157                             throw new TimeoutException();
 158                         long now = System.nanoTime();
 159                         nanos -= now - lastTime;
 160                         lastTime = now;
 161                     }
 162                     else
 163                         f = ecs.take();
 164                 }
 165                 if (f != null) {
 166                     --active;
 167                     try {
 168                         return f.get();
 169                     } catch (InterruptedException ie) {
 170                         throw ie;
 171                     } catch (ExecutionException eex) {
 172                         ee = eex;
 173                     } catch (RuntimeException rex) {
 174                         ee = new ExecutionException(rex);
 175                     }
 176                 }
 177             }
 178
 179             if (ee == null)
 180                 ee = new ExecutionException();
 181             throw ee;
 182
 183         } finally {
 184             for (Future<T> f : futures)
 185                 f.cancel(true);
 186         }
 187     }
 188
 189     public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
 190         throws InterruptedException, ExecutionException {
 191         try {
 192             return doInvokeAny(tasks, false, 0);
 193         } catch (TimeoutException cannotHappen) {
 194             assert false;
 195             return null;
 196         }
 197     }
 198
 199     public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
 200                            long timeout, TimeUnit unit)
 201         throws InterruptedException, ExecutionException, TimeoutException {
 202         return doInvokeAny(tasks, true, unit.toNanos(timeout));
 203     }
 204
 205     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
 206         throws InterruptedException {
 207         if (tasks == null)
 208             throw new NullPointerException();
 209         List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
 210         boolean done = false;
 211         try {
 212             for (Callable<T> t : tasks) {
 213                 RunnableFuture<T> f = newTaskFor(t);
 214                 futures.add(f);
 215                 execute(f);
 216             }
 217             for (Future<T> f : futures) {
 218                 if (!f.isDone()) {
 219                     try {
 220                         f.get();
 221                     } catch (CancellationException ignore) {
 222                     } catch (ExecutionException ignore) {
 223                     }
 224                 }
 225             }
 226             done = true;
 227             return futures;
 228         } finally {
 229             if (!done)
 230                 for (Future<T> f : futures)
 231                     f.cancel(true);
 232         }
 233     }
 234
 235     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
 236                                          long timeout, TimeUnit unit)
 237         throws InterruptedException {
 238         if (tasks == null || unit == null)
 239             throw new NullPointerException();
 240         long nanos = unit.toNanos(timeout);
 241         List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
 242         boolean done = false;
 243         try {
 244             for (Callable<T> t : tasks)
 245                 futures.add(newTaskFor(t));
 246
 247             long lastTime = System.nanoTime();
 248
 249             // Interleave time checks and calls to execute in case
 250             // executor doesn't have any/much parallelism.
 251             Iterator<Future<T>> it = futures.iterator();
 252             while (it.hasNext()) {
 253                 execute((Runnable)(it.next()));
 254                 long now = System.nanoTime();
 255                 nanos -= now - lastTime;
 256                 lastTime = now;
 257                 if (nanos <= 0)
 258                     return futures;
 259             }
 260
 261             for (Future<T> f : futures) {
 262                 if (!f.isDone()) {
 263                     if (nanos <= 0)
 264                         return futures;
 265                     try {
 266                         f.get(nanos, TimeUnit.NANOSECONDS);
 267                     } catch (CancellationException ignore) {
 268                     } catch (ExecutionException ignore) {
 269                     } catch (TimeoutException toe) {
 270                         return futures;
 271                     }
 272                     long now = System.nanoTime();
 273                     nanos -= now - lastTime;
 274                     lastTime = now;
 275                 }
 276             }
 277             done = true;
 278             return futures;
 279         } finally {
 280             if (!done)
 281                 for (Future<T> f : futures)
 282                     f.cancel(true);
 283         }
 284     }
 285
 286 }