001 /*
002 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
003 *
004 * This code is free software; you can redistribute it and/or modify it
005 * under the terms of the GNU General Public License version 2 only, as
006 * published by the Free Software Foundation. Sun designates this
007 * particular file as subject to the "Classpath" exception as provided
008 * by Sun in the LICENSE file that accompanied this code.
009 *
010 * This code is distributed in the hope that it will be useful, but WITHOUT
011 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
012 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
013 * version 2 for more details (a copy is included in the LICENSE file that
014 * accompanied this code).
015 *
016 * You should have received a copy of the GNU General Public License version
017 * 2 along with this work; if not, write to the Free Software Foundation,
018 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
019 *
020 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
021 * CA 95054 USA or visit www.sun.com if you need additional information or
022 * have any questions.
023 */
024
025 /*
026 * This file is available under and governed by the GNU General Public
027 * License version 2 only, as published by the Free Software Foundation.
028 * However, the following notice accompanied the original version of this
029 * file:
030 *
031 * Written by Doug Lea with assistance from members of JCP JSR-166
032 * Expert Group and released to the public domain, as explained at
033 * http://creativecommons.org/licenses/publicdomain
034 */
035
036 package java.util.concurrent;
037
038 /**
039 * A {@link CompletionService} that uses a supplied {@link Executor}
040 * to execute tasks. This class arranges that submitted tasks are,
041 * upon completion, placed on a queue accessible using <tt>take</tt>.
042 * The class is lightweight enough to be suitable for transient use
043 * when processing groups of tasks.
044 *
045 * <p>
046 *
047 * <b>Usage Examples.</b>
048 *
049 * Suppose you have a set of solvers for a certain problem, each
050 * returning a value of some type <tt>Result</tt>, and would like to
051 * run them concurrently, processing the results of each of them that
052 * return a non-null value, in some method <tt>use(Result r)</tt>. You
053 * could write this as:
054 *
055 * <pre>
056 * void solve(Executor e,
057 * Collection<Callable<Result>> solvers)
058 * throws InterruptedException, ExecutionException {
059 * CompletionService<Result> ecs
060 * = new ExecutorCompletionService<Result>(e);
061 * for (Callable<Result> s : solvers)
062 * ecs.submit(s);
063 * int n = solvers.size();
064 * for (int i = 0; i < n; ++i) {
065 * Result r = ecs.take().get();
066 * if (r != null)
067 * use(r);
068 * }
069 * }
070 * </pre>
071 *
072 * Suppose instead that you would like to use the first non-null result
073 * of the set of tasks, ignoring any that encounter exceptions,
074 * and cancelling all other tasks when the first one is ready:
075 *
076 * <pre>
077 * void solve(Executor e,
078 * Collection<Callable<Result>> solvers)
079 * throws InterruptedException {
080 * CompletionService<Result> ecs
081 * = new ExecutorCompletionService<Result>(e);
082 * int n = solvers.size();
083 * List<Future<Result>> futures
084 * = new ArrayList<Future<Result>>(n);
085 * Result result = null;
086 * try {
087 * for (Callable<Result> s : solvers)
088 * futures.add(ecs.submit(s));
089 * for (int i = 0; i < n; ++i) {
090 * try {
091 * Result r = ecs.take().get();
092 * if (r != null) {
093 * result = r;
094 * break;
095 * }
096 * } catch (ExecutionException ignore) {}
097 * }
098 * }
099 * finally {
100 * for (Future<Result> f : futures)
101 * f.cancel(true);
102 * }
103 *
104 * if (result != null)
105 * use(result);
106 * }
107 * </pre>
108 */
109 public class ExecutorCompletionService<V> implements
110 CompletionService<V> {
111 private final Executor executor;
112 private final AbstractExecutorService aes;
113 private final BlockingQueue<Future<V>> completionQueue;
114
115 /**
116 * FutureTask extension to enqueue upon completion
117 */
118 private class QueueingFuture extends FutureTask<Void> {
119 QueueingFuture(RunnableFuture<V> task) {
120 super (task, null);
121 this .task = task;
122 }
123
124 protected void done() {
125 completionQueue.add(task);
126 }
127
128 private final Future<V> task;
129 }
130
131 private RunnableFuture<V> newTaskFor(Callable<V> task) {
132 if (aes == null)
133 return new FutureTask<V>(task);
134 else
135 return aes.newTaskFor(task);
136 }
137
138 private RunnableFuture<V> newTaskFor(Runnable task, V result) {
139 if (aes == null)
140 return new FutureTask<V>(task, result);
141 else
142 return aes.newTaskFor(task, result);
143 }
144
145 /**
146 * Creates an ExecutorCompletionService using the supplied
147 * executor for base task execution and a
148 * {@link LinkedBlockingQueue} as a completion queue.
149 *
150 * @param executor the executor to use
151 * @throws NullPointerException if executor is <tt>null</tt>
152 */
153 public ExecutorCompletionService(Executor executor) {
154 if (executor == null)
155 throw new NullPointerException();
156 this .executor = executor;
157 this .aes = (executor instanceof AbstractExecutorService) ? (AbstractExecutorService) executor
158 : null;
159 this .completionQueue = new LinkedBlockingQueue<Future<V>>();
160 }
161
162 /**
163 * Creates an ExecutorCompletionService using the supplied
164 * executor for base task execution and the supplied queue as its
165 * completion queue.
166 *
167 * @param executor the executor to use
168 * @param completionQueue the queue to use as the completion queue
169 * normally one dedicated for use by this service
170 * @throws NullPointerException if executor or completionQueue are <tt>null</tt>
171 */
172 public ExecutorCompletionService(Executor executor,
173 BlockingQueue<Future<V>> completionQueue) {
174 if (executor == null || completionQueue == null)
175 throw new NullPointerException();
176 this .executor = executor;
177 this .aes = (executor instanceof AbstractExecutorService) ? (AbstractExecutorService) executor
178 : null;
179 this .completionQueue = completionQueue;
180 }
181
182 public Future<V> submit(Callable<V> task) {
183 if (task == null)
184 throw new NullPointerException();
185 RunnableFuture<V> f = newTaskFor(task);
186 executor.execute(new QueueingFuture(f));
187 return f;
188 }
189
190 public Future<V> submit(Runnable task, V result) {
191 if (task == null)
192 throw new NullPointerException();
193 RunnableFuture<V> f = newTaskFor(task, result);
194 executor.execute(new QueueingFuture(f));
195 return f;
196 }
197
198 public Future<V> take() throws InterruptedException {
199 return completionQueue.take();
200 }
201
202 public Future<V> poll() {
203 return completionQueue.poll();
204 }
205
206 public Future<V> poll(long timeout, TimeUnit unit)
207 throws InterruptedException {
208 return completionQueue.poll(timeout, unit);
209 }
210
211 }
|