ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/ExecutorCompletionService.java
Revision: 1.13
Committed: Fri Jul 8 19:59:32 2005 UTC (18 years, 10 months ago) by dl
Branch: MAIN
Changes since 1.12: +32 -8 lines
Log Message: 
Use newTaskFor

File Contents

# Content
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
10 /**
11 * A {@link CompletionService} that uses a supplied {@link Executor}
12 * to execute tasks. This class arranges that submitted tasks are,
13 * upon completion, placed on a queue accessible using <tt>take</tt>.
14 * The class is lightweight enough to be suitable for transient use
15 * when processing groups of tasks.
16 *
17 * <p>
18 *
19 * <b>Usage Examples.</b>
20 *
21 * Suppose you have a set of solvers for a certain problem, each
22 * returning a value of some type <tt>Result</tt>, and would like to
23 * run them concurrently, processing the results of each of them that
24 * return a non-null value, in some method <tt>use(Result r)</tt>. You
25 * could write this as:
26 *
27 * <pre>
28 * void solve(Executor e, Collection&lt;Callable&lt;Result&gt;&gt; solvers)
29 * throws InterruptedException, ExecutionException {
30 * CompletionService&lt;Result&gt; ecs = new ExecutorCompletionService&lt;Result&gt;(e);
31 * for (Callable&lt;Result&gt; s : solvers)
32 * ecs.submit(s);
33 * int n = solvers.size();
34 * for (int i = 0; i &lt; n; ++i) {
35 * Result r = ecs.take().get();
36 * if (r != null)
37 * use(r);
38 * }
39 * }
40 * </pre>
41 *
42 * Suppose instead that you would like to use the first non-null result
43 * of the set of tasks, ignoring any that encounter exceptions,
44 * and cancelling all other tasks when the first one is ready:
45 *
46 * <pre>
47 * void solve(Executor e, Collection&lt;Callable&lt;Result&gt;&gt; solvers)
48 * throws InterruptedException {
49 * CompletionService&lt;Result&gt; ecs = new ExecutorCompletionService&lt;Result&gt;(e);
50 * int n = solvers.size();
51 * List&lt;Future&lt;Result&gt;&gt; futures = new ArrayList&lt;Future&lt;Result&gt;&gt;(n);
52 * Result result = null;
53 * try {
54 * for (Callable&lt;Result&gt; s : solvers)
55 * futures.add(ecs.submit(s));
56 * for (int i = 0; i &lt; n; ++i) {
57 * try {
58 * Result r = ecs.take().get();
59 * if (r != null) {
60 * result = r;
61 * break;
62 * }
63 * } catch (ExecutionException ignore) {}
64 * }
65 * }
66 * finally {
67 * for (Future&lt;Result&gt; f : futures)
68 * f.cancel(true);
69 * }
70 *
71 * if (result != null)
72 * use(result);
73 * }
74 * </pre>
75 */
76 public class ExecutorCompletionService<V> implements CompletionService<V> {
77 private final Executor executor;
78 private final AbstractExecutorService aes;
79 private final BlockingQueue<Future<V>> completionQueue;
80
81 /**
82 * FutureTask extension to enqueue upon completion
83 */
84 private class QueueingFuture extends FutureTask<Void> {
85 QueueingFuture(RunnableFuture<V> task) {
86 super(task, null);
87 this.task = task;
88 }
89 protected void done() { completionQueue.add(task); }
90 private final Future<V> task;
91 }
92
93 private RunnableFuture<V> newTaskFor(Callable<V> task) {
94 if (aes == null)
95 return new FutureTask<V>(task);
96 else
97 return aes.newTaskFor(task);
98 }
99
100 private RunnableFuture<V> newTaskFor(Runnable task, V result) {
101 if (aes == null)
102 return new FutureTask<V>(task, result);
103 else
104 return aes.newTaskFor(task, result);
105 }
106
107 /**
108 * Creates an ExecutorCompletionService using the supplied
109 * executor for base task execution and a
110 * {@link LinkedBlockingQueue} as a completion queue.
111 * @param executor the executor to use
112 * @throws NullPointerException if executor is <tt>null</tt>
113 */
114 public ExecutorCompletionService(Executor executor) {
115 if (executor == null)
116 throw new NullPointerException();
117 this.executor = executor;
118 this.aes = (executor instanceof AbstractExecutorService) ?
119 (AbstractExecutorService) executor : null;
120 this.completionQueue = new LinkedBlockingQueue<Future<V>>();
121 }
122
123 /**
124 * Creates an ExecutorCompletionService using the supplied
125 * executor for base task execution and the supplied queue as its
126 * completion queue.
127 * @param executor the executor to use
128 * @param completionQueue the queue to use as the completion queue
129 * normally one dedicated for use by this service
130 * @throws NullPointerException if executor or completionQueue are <tt>null</tt>
131 */
132 public ExecutorCompletionService(Executor executor,
133 BlockingQueue<Future<V>> completionQueue) {
134 if (executor == null || completionQueue == null)
135 throw new NullPointerException();
136 this.executor = executor;
137 this.aes = (executor instanceof AbstractExecutorService) ?
138 (AbstractExecutorService) executor : null;
139 this.completionQueue = completionQueue;
140 }
141
142 public Future<V> submit(Callable<V> task) {
143 if (task == null) throw new NullPointerException();
144 RunnableFuture<V> f = newTaskFor(task);
145 executor.execute(new QueueingFuture(f));
146 return f;
147 }
148
149 public Future<V> submit(Runnable task, V result) {
150 if (task == null) throw new NullPointerException();
151 RunnableFuture<V> f = newTaskFor(task, result);
152 executor.execute(new QueueingFuture(f));
153 return f;
154 }
155
156 public Future<V> take() throws InterruptedException {
157 return completionQueue.take();
158 }
159
160 public Future<V> poll() {
161 return completionQueue.poll();
162 }
163
164 public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException {
165 return completionQueue.poll(timeout, unit);
166 }
167
168 }
169
170
171
172