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