Print this page
Split |
Close |
Expand all |
Collapse all |
--- old/src/share/classes/java/util/concurrent/ExecutorCompletionService.java
+++ new/src/share/classes/java/util/concurrent/ExecutorCompletionService.java
1 1 /*
2 2 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
3 3 *
4 4 * This code is free software; you can redistribute it and/or modify it
5 5 * under the terms of the GNU General Public License version 2 only, as
6 6 * published by the Free Software Foundation. Oracle designates this
7 7 * particular file as subject to the "Classpath" exception as provided
8 8 * by Oracle in the LICENSE file that accompanied this code.
9 9 *
10 10 * This code is distributed in the hope that it will be useful, but WITHOUT
11 11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12 12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
13 13 * version 2 for more details (a copy is included in the LICENSE file that
14 14 * accompanied this code).
15 15 *
16 16 * You should have received a copy of the GNU General Public License version
17 17 * 2 along with this work; if not, write to the Free Software Foundation,
18 18 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
19 19 *
20 20 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
21 21 * or visit www.oracle.com if you need additional information or have any
22 22 * questions.
23 23 */
24 24
25 25 /*
26 26 * This file is available under and governed by the GNU General Public
27 27 * License version 2 only, as published by the Free Software Foundation.
28 28 * However, the following notice accompanied the original version of this
29 29 * file:
30 30 *
31 31 * Written by Doug Lea with assistance from members of JCP JSR-166
32 32 * Expert Group and released to the public domain, as explained at
33 33 * http://creativecommons.org/licenses/publicdomain
34 34 */
35 35
36 36 package java.util.concurrent;
37 37
38 38 /**
39 39 * A {@link CompletionService} that uses a supplied {@link Executor}
40 40 * to execute tasks. This class arranges that submitted tasks are,
41 41 * upon completion, placed on a queue accessible using {@code take}.
42 42 * The class is lightweight enough to be suitable for transient use
43 43 * when processing groups of tasks.
44 44 *
45 45 * <p>
46 46 *
47 47 * <b>Usage Examples.</b>
48 48 *
49 49 * Suppose you have a set of solvers for a certain problem, each
50 50 * returning a value of some type {@code Result}, and would like to
51 51 * run them concurrently, processing the results of each of them that
52 52 * return a non-null value, in some method {@code use(Result r)}. You
53 53 * could write this as:
54 54 *
55 55 * <pre> {@code
56 56 * void solve(Executor e,
57 57 * Collection<Callable<Result>> solvers)
58 58 * throws InterruptedException, ExecutionException {
59 59 * CompletionService<Result> ecs
60 60 * = new ExecutorCompletionService<Result>(e);
61 61 * for (Callable<Result> s : solvers)
62 62 * ecs.submit(s);
63 63 * int n = solvers.size();
64 64 * for (int i = 0; i < n; ++i) {
65 65 * Result r = ecs.take().get();
66 66 * if (r != null)
67 67 * use(r);
68 68 * }
69 69 * }}</pre>
70 70 *
71 71 * Suppose instead that you would like to use the first non-null result
72 72 * of the set of tasks, ignoring any that encounter exceptions,
73 73 * and cancelling all other tasks when the first one is ready:
74 74 *
75 75 * <pre> {@code
76 76 * void solve(Executor e,
77 77 * Collection<Callable<Result>> solvers)
78 78 * throws InterruptedException {
79 79 * CompletionService<Result> ecs
80 80 * = new ExecutorCompletionService<Result>(e);
81 81 * int n = solvers.size();
82 82 * List<Future<Result>> futures
83 83 * = new ArrayList<Future<Result>>(n);
84 84 * Result result = null;
85 85 * try {
86 86 * for (Callable<Result> s : solvers)
87 87 * futures.add(ecs.submit(s));
88 88 * for (int i = 0; i < n; ++i) {
89 89 * try {
90 90 * Result r = ecs.take().get();
91 91 * if (r != null) {
92 92 * result = r;
93 93 * break;
94 94 * }
95 95 * } catch (ExecutionException ignore) {}
96 96 * }
97 97 * }
98 98 * finally {
99 99 * for (Future<Result> f : futures)
100 100 * f.cancel(true);
101 101 * }
102 102 *
103 103 * if (result != null)
104 104 * use(result);
105 105 * }}</pre>
106 106 */
107 107 public class ExecutorCompletionService<V> implements CompletionService<V> {
108 108 private final Executor executor;
109 109 private final AbstractExecutorService aes;
110 110 private final BlockingQueue<Future<V>> completionQueue;
111 111
112 112 /**
113 113 * FutureTask extension to enqueue upon completion
114 114 */
115 115 private class QueueingFuture extends FutureTask<Void> {
116 116 QueueingFuture(RunnableFuture<V> task) {
117 117 super(task, null);
118 118 this.task = task;
119 119 }
120 120 protected void done() { completionQueue.add(task); }
121 121 private final Future<V> task;
122 122 }
123 123
124 124 private RunnableFuture<V> newTaskFor(Callable<V> task) {
125 125 if (aes == null)
126 126 return new FutureTask<V>(task);
127 127 else
128 128 return aes.newTaskFor(task);
129 129 }
130 130
131 131 private RunnableFuture<V> newTaskFor(Runnable task, V result) {
132 132 if (aes == null)
133 133 return new FutureTask<V>(task, result);
134 134 else
135 135 return aes.newTaskFor(task, result);
136 136 }
137 137
138 138 /**
139 139 * Creates an ExecutorCompletionService using the supplied
140 140 * executor for base task execution and a
141 141 * {@link LinkedBlockingQueue} as a completion queue.
142 142 *
143 143 * @param executor the executor to use
144 144 * @throws NullPointerException if executor is {@code null}
145 145 */
146 146 public ExecutorCompletionService(Executor executor) {
147 147 if (executor == null)
148 148 throw new NullPointerException();
149 149 this.executor = executor;
150 150 this.aes = (executor instanceof AbstractExecutorService) ?
151 151 (AbstractExecutorService) executor : null;
152 152 this.completionQueue = new LinkedBlockingQueue<Future<V>>();
153 153 }
154 154
155 155 /**
156 156 * Creates an ExecutorCompletionService using the supplied
157 157 * executor for base task execution and the supplied queue as its
158 158 * completion queue.
159 159 *
160 160 * @param executor the executor to use
161 161 * @param completionQueue the queue to use as the completion queue
162 162 * normally one dedicated for use by this service. This
163 163 * queue is treated as unbounded -- failed attempted
164 164 * {@code Queue.add} operations for completed taskes cause
165 165 * them not to be retrievable.
166 166 * @throws NullPointerException if executor or completionQueue are {@code null}
167 167 */
168 168 public ExecutorCompletionService(Executor executor,
169 169 BlockingQueue<Future<V>> completionQueue) {
170 170 if (executor == null || completionQueue == null)
171 171 throw new NullPointerException();
172 172 this.executor = executor;
173 173 this.aes = (executor instanceof AbstractExecutorService) ?
174 174 (AbstractExecutorService) executor : null;
175 175 this.completionQueue = completionQueue;
176 176 }
177 177
178 178 public Future<V> submit(Callable<V> task) {
179 179 if (task == null) throw new NullPointerException();
180 180 RunnableFuture<V> f = newTaskFor(task);
181 181 executor.execute(new QueueingFuture(f));
182 182 return f;
183 183 }
184 184
185 185 public Future<V> submit(Runnable task, V result) {
186 186 if (task == null) throw new NullPointerException();
187 187 RunnableFuture<V> f = newTaskFor(task, result);
188 188 executor.execute(new QueueingFuture(f));
189 189 return f;
↓ open down ↓ |
189 lines elided |
↑ open up ↑ |
190 190 }
191 191
192 192 public Future<V> take() throws InterruptedException {
193 193 return completionQueue.take();
194 194 }
195 195
196 196 public Future<V> poll() {
197 197 return completionQueue.poll();
198 198 }
199 199
200 - public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException {
200 + public Future<V> poll(long timeout, TimeUnit unit)
201 + throws InterruptedException {
201 202 return completionQueue.poll(timeout, unit);
202 203 }
203 204
204 205 }
XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX