util/concurrent/AbstractExecutorService.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain. Use, modify, and
 * redistribute this code in any way without acknowledgement.
 */

package java.util.concurrent;

import java.security.AccessControlContext;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.security.PrivilegedExceptionAction;
import java.util.*;
import java.util.concurrent.locks.*;

/**
 * Provides default implementation of {@link ExecutorService}
 * execution methods. This class implements the <tt>submit</tt> and
 * <tt>invoke</tt> methods using the default {@link FutureTask} and
 * {@link PrivilegedFutureTask} classes provided in this package.  For
 * example, the the implementation of <tt>submit(Runnable)</tt>
 * creates an associated <tt>FutureTask</tt> that is executed and
 * returned. Subclasses overriding these methods to use different
 * {@link Future} implementations should do so consistently for each
 * of these methods.
 *
 * @since 1.5
 * @author Doug Lea
 */
public abstract class AbstractExecutorService implements ExecutorService {

    public <T> Future<T> submit(Runnable task, T result) {
        FutureTask<T> ftask = new FutureTask<T>(task, result);
        execute(ftask);
        return ftask;
    }

    public <T> Future<T> submit(Callable<T> task) {
        FutureTask<T> ftask = new FutureTask<T>(task);
        execute(ftask);
        return ftask;
    }

    public void invoke(Runnable task) throws ExecutionException, InterruptedException {
        FutureTask<?> ftask = new FutureTask<Boolean>(task, Boolean.TRUE);
        execute(ftask);
        ftask.get();
    }

    public <T> T invoke(Callable<T> task) throws ExecutionException, InterruptedException {
        FutureTask<T> ftask = new FutureTask<T>(task);
        execute(ftask);
        return ftask.get();
    }

    public Future<Object> submit(PrivilegedAction action) {
        Callable<Object> task = new PrivilegedActionAdapter(action);
        FutureTask<Object> future = new PrivilegedFutureTask<Object>(task);
        execute(future);
        return future;
    }

    public Future<Object> submit(PrivilegedExceptionAction action) {
        Callable<Object> task = new PrivilegedExceptionActionAdapter(action);
        FutureTask<Object> future = new PrivilegedFutureTask<Object>(task);
        execute(future);
        return future;
    }

    private static class PrivilegedActionAdapter implements Callable<Object> {
        PrivilegedActionAdapter(PrivilegedAction action) {
            this.action = action;
        }
        public Object call () {
            return action.run();
        }
        private final PrivilegedAction action;
    }
    
    private static class PrivilegedExceptionActionAdapter implements Callable<Object> {
        PrivilegedExceptionActionAdapter(PrivilegedExceptionAction action) {
            this.action = action;
        }
        public Object call () throws Exception {
            return action.run();
        }
        private final PrivilegedExceptionAction action;
    }

    // any/all methods, each a little bit different than the other


    public <T> T invokeAny(Collection<Callable<T>> tasks)
        throws InterruptedException, ExecutionException {
        if (tasks == null)
            throw new NullPointerException();
        int n = tasks.size();
        if (n == 0)
            throw new IllegalArgumentException();
        List<Future<T>> futures= new ArrayList<Future<T>>(n);
        ExecutorCompletionService<T> ecs = 
            new ExecutorCompletionService<T>(this);
        try {
            for (Callable<T> t : tasks) 
                futures.add(ecs.submit(t));
            ExecutionException ee = null;
            RuntimeException re = null;
            while (n-- > 0) {
                Future<T> f = ecs.take();
                try {
                    return f.get();
                } catch(ExecutionException eex) {
                    ee = eex;
                } catch(RuntimeException rex) {
                    re = rex;
                }
            }    
            if (ee != null)
                throw ee;
            if (re != null)
                throw new ExecutionException(re);
            throw new ExecutionException();
        } finally {
            for (Future<T> f : futures) 
                f.cancel(true);
        }
    }

    public <T> T invokeAny(Collection<Callable<T>> tasks, 
                           long timeout, TimeUnit unit) 
        throws InterruptedException, ExecutionException, TimeoutException {
        if (tasks == null || unit == null)
            throw new NullPointerException();
        long nanos = unit.toNanos(timeout);
        int n = tasks.size();
        if (n == 0)
            throw new IllegalArgumentException();
        List<Future<T>> futures= new ArrayList<Future<T>>(n);
        ExecutorCompletionService<T> ecs = 
            new ExecutorCompletionService<T>(this);
        try {
            for (Callable<T> t : tasks) 
                futures.add(ecs.submit(t));
            ExecutionException ee = null;
            RuntimeException re = null;
            long lastTime = System.nanoTime();
            while (n-- > 0) {
                Future<T> f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
                if (f == null) {
                    if (nanos <= 0)
                        throw new TimeoutException();
                    long now = System.nanoTime();
                    nanos -= now - lastTime;
                    lastTime = now;
                }
                try {
                    return f.get();
                } catch(ExecutionException eex) {
                    ee = eex;
                } catch(RuntimeException rex) {
                    re = rex;
                }
            }    
            if (ee != null)
                throw ee;
            if (re != null)
                throw new ExecutionException(re);
            throw new ExecutionException();
        } finally {
            for (Future<T> f : futures) 
                f.cancel(true);
        }
    }


    public <T> T invokeAny(Collection<Runnable> tasks, T result)
        throws InterruptedException, ExecutionException {
        if (tasks == null)
            throw new NullPointerException();
        int n = tasks.size();
        if (n == 0)
            throw new IllegalArgumentException();
        List<Future<T>> futures= new ArrayList<Future<T>>(n);
        ExecutorCompletionService<T> ecs = 
            new ExecutorCompletionService<T>(this);
        try {
            for (Runnable t : tasks) 
                futures.add(ecs.submit(t, result));
            ExecutionException ee = null;
            RuntimeException re = null;
            while (n-- > 0) {
                Future<T> f = ecs.take();
                try {
                    return f.get();
                } catch(ExecutionException eex) {
                    ee = eex;
                } catch(RuntimeException rex) {
                    re = rex;
                }
            }    
            if (ee != null)
                throw ee;
            if (re != null)
                throw new ExecutionException(re);
            throw new ExecutionException();
        } finally {
            for (Future<T> f : futures) 
                f.cancel(true);
        }
    }

    public <T> T invokeAny(Collection<Runnable> tasks, T result,
                           long timeout, TimeUnit unit) 
        throws InterruptedException, ExecutionException, TimeoutException {
        if (tasks == null || unit == null)
            throw new NullPointerException();
        long nanos = unit.toNanos(timeout);
        int n = tasks.size();
        if (n == 0)
            throw new IllegalArgumentException();
        List<Future<T>> futures= new ArrayList<Future<T>>(n);
        ExecutorCompletionService<T> ecs = 
            new ExecutorCompletionService<T>(this);
        try {
            for (Runnable t : tasks) 
                futures.add(ecs.submit(t, result));
            ExecutionException ee = null;
            RuntimeException re = null;
            long lastTime = System.nanoTime();
            while (n-- > 0) {
                Future<T> f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
                if (f == null) {
                    if (nanos <= 0)
                        throw new TimeoutException();
                    long now = System.nanoTime();
                    nanos -= now - lastTime;
                    lastTime = now;
                }
                try {
                    return f.get();
                } catch(ExecutionException eex) {
                    ee = eex;
                } catch(RuntimeException rex) {
                    re = rex;
                }
            }    
            if (ee != null)
                throw ee;
            if (re != null)
                throw new ExecutionException(re);
            throw new ExecutionException();
        } finally {
            for (Future<T> f : futures) 
                f.cancel(true);
        }
    }


    public <T> List<Future<T>> invokeAll(List<Runnable> tasks, T result)
        throws InterruptedException {
        if (tasks == null)
            throw new NullPointerException();
        List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        boolean done = false;
        try {
            for (Runnable t : tasks) {
                FutureTask<T> f = new FutureTask<T>(t, result);
                futures.add(f);
                execute(f);
            }
            for (Future<T> f : futures) {
                if (!f.isDone()) {
                    try { 
                        f.get(); 
                    } catch(CancellationException ignore) {
                    } catch(ExecutionException ignore) {
                    }
                }
            }
            done = true;
            return futures;
        } finally {
            if (!done)
                for (Future<T> f : futures) 
                    f.cancel(true);
        }
    }

    public <T> List<Future<T>> invokeAll(List<Runnable> tasks, T result,
                                         long timeout, TimeUnit unit) 
        throws InterruptedException {
        if (tasks == null || unit == null)
            throw new NullPointerException();
        long nanos = unit.toNanos(timeout);
        List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        boolean done = false;
        try {
            for (Runnable t : tasks) {
                FutureTask<T> f = new FutureTask<T>(t, result);
                futures.add(f);
                execute(f);
            }
            long lastTime = System.nanoTime();
            for (Future<T> f : futures) {
                if (!f.isDone()) {
                    if (nanos < 0) 
                        return futures;
                    try { 
                        f.get(nanos, TimeUnit.NANOSECONDS); 
                        long now = System.nanoTime();
                        nanos -= now - lastTime;
                        lastTime = now;
                    } catch(CancellationException ignore) {
                    } catch(ExecutionException ignore) {
                    } catch(TimeoutException toe) {
                        return futures;
                    }
                }
            }
            done = true;
            return futures;
        } finally {
            if (!done)
                for (Future<T> f : futures) 
                    f.cancel(true);
        }
    }

    public <T> List<Future<T>> invokeAll(List<Callable<T>> tasks)
        throws InterruptedException {
        if (tasks == null)
            throw new NullPointerException();
        List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        boolean done = false;
        try {
            for (Callable<T> t : tasks) {
                FutureTask<T> f = new FutureTask<T>(t);
                futures.add(f);
                execute(f);
            }
            for (Future<T> f : futures) {
                if (!f.isDone()) {
                    try { 
                        f.get(); 
                    } catch(CancellationException ignore) {
                    } catch(ExecutionException ignore) {
                    }
                }
            }
            done = true;
            return futures;
        } finally {
            if (!done)
                for (Future<T> f : futures) 
                    f.cancel(true);
        }
    }

    public <T> List<Future<T>> invokeAll(List<Callable<T>> tasks, 
                                         long timeout, TimeUnit unit) 
        throws InterruptedException {
        if (tasks == null || unit == null)
            throw new NullPointerException();
        long nanos = unit.toNanos(timeout);
        List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        boolean done = false;
        try {
            for (Callable<T> t : tasks) {
                FutureTask<T> f = new FutureTask<T>(t);
                futures.add(f);
                execute(f);
            }
            long lastTime = System.nanoTime();
            for (Future<T> f : futures) {
                if (!f.isDone()) {
                    if (nanos < 0) 
                        return futures; 
                    try { 
                        f.get(nanos, TimeUnit.NANOSECONDS); 
                        long now = System.nanoTime();
                        nanos -= now - lastTime;
                        lastTime = now;
                    } catch(CancellationException ignore) {
                    } catch(ExecutionException ignore) {
                    } catch(TimeoutException toe) {
                        return futures;
                    }
                }
            }
            done = true;
            return futures;
        } finally {
            if (!done)
                for (Future<T> f : futures) 
                    f.cancel(true);
        }
    }

}
Revision:	1.7
Committed:	Tue Dec 16 12:45:06 2003 UTC (20 years, 5 months ago) by dl
Branch:	MAIN
Changes since 1.6:	+122 -72 lines
Log Message:	invokeAny returns value, not list
#	Content
1	/*
2	* Written by Doug Lea with assistance from members of JCP JSR-166
3	* Expert Group and released to the public domain. Use, modify, and
4	* redistribute this code in any way without acknowledgement.
5	*/
6
7	package java.util.concurrent;
8
9	import java.security.AccessControlContext;
10	import java.security.AccessController;
11	import java.security.PrivilegedAction;
12	import java.security.PrivilegedExceptionAction;
13	import java.util.*;
14	import java.util.concurrent.locks.*;
15
16	/**
17	* Provides default implementation of {@link ExecutorService}
18	* execution methods. This class implements the <tt>submit</tt> and
19	* <tt>invoke</tt> methods using the default {@link FutureTask} and
20	* {@link PrivilegedFutureTask} classes provided in this package. For
21	* example, the the implementation of <tt>submit(Runnable)</tt>
22	* creates an associated <tt>FutureTask</tt> that is executed and
23	* returned. Subclasses overriding these methods to use different
24	* {@link Future} implementations should do so consistently for each
25	* of these methods.
26	*
27	* @since 1.5
28	* @author Doug Lea
29	*/
30	public abstract class AbstractExecutorService implements ExecutorService {
31
32	public <T> Future<T> submit(Runnable task, T result) {
33	FutureTask<T> ftask = new FutureTask<T>(task, result);
34	execute(ftask);
35	return ftask;
36	}
37
38	public <T> Future<T> submit(Callable<T> task) {
39	FutureTask<T> ftask = new FutureTask<T>(task);
40	execute(ftask);
41	return ftask;
42	}
43
44	public void invoke(Runnable task) throws ExecutionException, InterruptedException {
45	FutureTask<?> ftask = new FutureTask<Boolean>(task, Boolean.TRUE);
46	execute(ftask);
47	ftask.get();
48	}
49
50	public <T> T invoke(Callable<T> task) throws ExecutionException, InterruptedException {
51	FutureTask<T> ftask = new FutureTask<T>(task);
52	execute(ftask);
53	return ftask.get();
54	}
55
56	public Future<Object> submit(PrivilegedAction action) {
57	Callable<Object> task = new PrivilegedActionAdapter(action);
58	FutureTask<Object> future = new PrivilegedFutureTask<Object>(task);
59	execute(future);
60	return future;
61	}
62
63	public Future<Object> submit(PrivilegedExceptionAction action) {
64	Callable<Object> task = new PrivilegedExceptionActionAdapter(action);
65	FutureTask<Object> future = new PrivilegedFutureTask<Object>(task);
66	execute(future);
67	return future;
68	}
69
70	private static class PrivilegedActionAdapter implements Callable<Object> {
71	PrivilegedActionAdapter(PrivilegedAction action) {
72	this.action = action;
73	}
74	public Object call () {
75	return action.run();
76	}
77	private final PrivilegedAction action;
78	}
79
80	private static class PrivilegedExceptionActionAdapter implements Callable<Object> {
81	PrivilegedExceptionActionAdapter(PrivilegedExceptionAction action) {
82	this.action = action;
83	}
84	public Object call () throws Exception {
85	return action.run();
86	}
87	private final PrivilegedExceptionAction action;
88	}
89
90	// any/all methods, each a little bit different than the other
91
92
93	public <T> T invokeAny(Collection<Callable<T>> tasks)
94	throws InterruptedException, ExecutionException {
95	if (tasks == null)
96	throw new NullPointerException();
97	int n = tasks.size();
98	if (n == 0)
99	throw new IllegalArgumentException();
100	List<Future<T>> futures= new ArrayList<Future<T>>(n);
101	ExecutorCompletionService<T> ecs =
102	new ExecutorCompletionService<T>(this);
103	try {
104	for (Callable<T> t : tasks)
105	futures.add(ecs.submit(t));
106	ExecutionException ee = null;
107	RuntimeException re = null;
108	while (n-- > 0) {
109	Future<T> f = ecs.take();
110	try {
111	return f.get();
112	} catch(ExecutionException eex) {
113	ee = eex;
114	} catch(RuntimeException rex) {
115	re = rex;
116	}
117	}
118	if (ee != null)
119	throw ee;
120	if (re != null)
121	throw new ExecutionException(re);
122	throw new ExecutionException();
123	} finally {
124	for (Future<T> f : futures)
125	f.cancel(true);
126	}
127	}
128
129	public <T> T invokeAny(Collection<Callable<T>> tasks,
130	long timeout, TimeUnit unit)
131	throws InterruptedException, ExecutionException, TimeoutException {
132	if (tasks == null \|\| unit == null)
133	throw new NullPointerException();
134	long nanos = unit.toNanos(timeout);
135	int n = tasks.size();
136	if (n == 0)
137	throw new IllegalArgumentException();
138	List<Future<T>> futures= new ArrayList<Future<T>>(n);
139	ExecutorCompletionService<T> ecs =
140	new ExecutorCompletionService<T>(this);
141	try {
142	for (Callable<T> t : tasks)
143	futures.add(ecs.submit(t));
144	ExecutionException ee = null;
145	RuntimeException re = null;
146	long lastTime = System.nanoTime();
147	while (n-- > 0) {
148	Future<T> f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
149	if (f == null) {
150	if (nanos <= 0)
151	throw new TimeoutException();
152	long now = System.nanoTime();
153	nanos -= now - lastTime;
154	lastTime = now;
155	}
156	try {
157	return f.get();
158	} catch(ExecutionException eex) {
159	ee = eex;
160	} catch(RuntimeException rex) {
161	re = rex;
162	}
163	}
164	if (ee != null)
165	throw ee;
166	if (re != null)
167	throw new ExecutionException(re);
168	throw new ExecutionException();
169	} finally {
170	for (Future<T> f : futures)
171	f.cancel(true);
172	}
173	}
174
175
176	public <T> T invokeAny(Collection<Runnable> tasks, T result)
177	throws InterruptedException, ExecutionException {
178	if (tasks == null)
179	throw new NullPointerException();
180	int n = tasks.size();
181	if (n == 0)
182	throw new IllegalArgumentException();
183	List<Future<T>> futures= new ArrayList<Future<T>>(n);
184	ExecutorCompletionService<T> ecs =
185	new ExecutorCompletionService<T>(this);
186	try {
187	for (Runnable t : tasks)
188	futures.add(ecs.submit(t, result));
189	ExecutionException ee = null;
190	RuntimeException re = null;
191	while (n-- > 0) {
192	Future<T> f = ecs.take();
193	try {
194	return f.get();
195	} catch(ExecutionException eex) {
196	ee = eex;
197	} catch(RuntimeException rex) {
198	re = rex;
199	}
200	}
201	if (ee != null)
202	throw ee;
203	if (re != null)
204	throw new ExecutionException(re);
205	throw new ExecutionException();
206	} finally {
207	for (Future<T> f : futures)
208	f.cancel(true);
209	}
210	}
211
212	public <T> T invokeAny(Collection<Runnable> tasks, T result,
213	long timeout, TimeUnit unit)
214	throws InterruptedException, ExecutionException, TimeoutException {
215	if (tasks == null \|\| unit == null)
216	throw new NullPointerException();
217	long nanos = unit.toNanos(timeout);
218	int n = tasks.size();
219	if (n == 0)
220	throw new IllegalArgumentException();
221	List<Future<T>> futures= new ArrayList<Future<T>>(n);
222	ExecutorCompletionService<T> ecs =
223	new ExecutorCompletionService<T>(this);
224	try {
225	for (Runnable t : tasks)
226	futures.add(ecs.submit(t, result));
227	ExecutionException ee = null;
228	RuntimeException re = null;
229	long lastTime = System.nanoTime();
230	while (n-- > 0) {
231	Future<T> f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
232	if (f == null) {
233	if (nanos <= 0)
234	throw new TimeoutException();
235	long now = System.nanoTime();
236	nanos -= now - lastTime;
237	lastTime = now;
238	}
239	try {
240	return f.get();
241	} catch(ExecutionException eex) {
242	ee = eex;
243	} catch(RuntimeException rex) {
244	re = rex;
245	}
246	}
247	if (ee != null)
248	throw ee;
249	if (re != null)
250	throw new ExecutionException(re);
251	throw new ExecutionException();
252	} finally {
253	for (Future<T> f : futures)
254	f.cancel(true);
255	}
256	}
257
258
259	public <T> List<Future<T>> invokeAll(List<Runnable> tasks, T result)
260	throws InterruptedException {
261	if (tasks == null)
262	throw new NullPointerException();
263	List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
264	boolean done = false;
265	try {
266	for (Runnable t : tasks) {
267	FutureTask<T> f = new FutureTask<T>(t, result);
268	futures.add(f);
269	execute(f);
270	}
271	for (Future<T> f : futures) {
272	if (!f.isDone()) {
273	try {
274	f.get();
275	} catch(CancellationException ignore) {
276	} catch(ExecutionException ignore) {
277	}
278	}
279	}
280	done = true;
281	return futures;
282	} finally {
283	if (!done)
284	for (Future<T> f : futures)
285	f.cancel(true);
286	}
287	}
288
289	public <T> List<Future<T>> invokeAll(List<Runnable> tasks, T result,
290	long timeout, TimeUnit unit)
291	throws InterruptedException {
292	if (tasks == null \|\| unit == null)
293	throw new NullPointerException();
294	long nanos = unit.toNanos(timeout);
295	List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
296	boolean done = false;
297	try {
298	for (Runnable t : tasks) {
299	FutureTask<T> f = new FutureTask<T>(t, result);
300	futures.add(f);
301	execute(f);
302	}
303	long lastTime = System.nanoTime();
304	for (Future<T> f : futures) {
305	if (!f.isDone()) {
306	if (nanos < 0)
307	return futures;
308	try {
309	f.get(nanos, TimeUnit.NANOSECONDS);
310	long now = System.nanoTime();
311	nanos -= now - lastTime;
312	lastTime = now;
313	} catch(CancellationException ignore) {
314	} catch(ExecutionException ignore) {
315	} catch(TimeoutException toe) {
316	return futures;
317	}
318	}
319	}
320	done = true;
321	return futures;
322	} finally {
323	if (!done)
324	for (Future<T> f : futures)
325	f.cancel(true);
326	}
327	}
328
329	public <T> List<Future<T>> invokeAll(List<Callable<T>> tasks)
330	throws InterruptedException {
331	if (tasks == null)
332	throw new NullPointerException();
333	List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
334	boolean done = false;
335	try {
336	for (Callable<T> t : tasks) {
337	FutureTask<T> f = new FutureTask<T>(t);
338	futures.add(f);
339	execute(f);
340	}
341	for (Future<T> f : futures) {
342	if (!f.isDone()) {
343	try {
344	f.get();
345	} catch(CancellationException ignore) {
346	} catch(ExecutionException ignore) {
347	}
348	}
349	}
350	done = true;
351	return futures;
352	} finally {
353	if (!done)
354	for (Future<T> f : futures)
355	f.cancel(true);
356	}
357	}
358
359	public <T> List<Future<T>> invokeAll(List<Callable<T>> tasks,
360	long timeout, TimeUnit unit)
361	throws InterruptedException {
362	if (tasks == null \|\| unit == null)
363	throw new NullPointerException();
364	long nanos = unit.toNanos(timeout);
365	List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
366	boolean done = false;
367	try {
368	for (Callable<T> t : tasks) {
369	FutureTask<T> f = new FutureTask<T>(t);
370	futures.add(f);
371	execute(f);
372	}
373	long lastTime = System.nanoTime();
374	for (Future<T> f : futures) {
375	if (!f.isDone()) {
376	if (nanos < 0)
377	return futures;
378	try {
379	f.get(nanos, TimeUnit.NANOSECONDS);
380	long now = System.nanoTime();
381	nanos -= now - lastTime;
382	lastTime = now;
383	} catch(CancellationException ignore) {
384	} catch(ExecutionException ignore) {
385	} catch(TimeoutException toe) {
386	return futures;
387	}
388	}
389	}
390	done = true;
391	return futures;
392	} finally {
393	if (!done)
394	for (Future<T> f : futures)
395	f.cancel(true);
396	}
397	}
398
399	}