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.util.*;

/**
 * Provides default implementation of {@link ExecutorService}
 * execution methods. This class implements the <tt>submit</tt>,
 * <tt>invokeAny</tt> and <tt>invokeAll</tt> methods using the default
 * {@link FutureTask} class 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 Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        FutureTask<Object> ftask = new FutureTask<Object>(task, null);
        execute(ftask);
        return ftask;
    }

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

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

    // 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)
                    throw new TimeoutException();
                try {
                    return f.get();
                } catch(ExecutionException eex) {
                    ee = eex;
                } catch(RuntimeException rex) {
                    re = rex;
                }
                long now = System.nanoTime();
                nanos -= now - lastTime;
                lastTime = now;
            }    
            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(Collection<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(Collection<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); 
                    } catch(CancellationException ignore) {
                    } catch(ExecutionException ignore) {
                    } catch(TimeoutException toe) {
                        return futures;
                    }
                    long now = System.nanoTime();
                    nanos -= now - lastTime;
                    lastTime = now;
                }
            }
            done = true;
            return futures;
        } finally {
            if (!done)
                for (Future<T> f : futures) 
                    f.cancel(true);
        }
    }

}
Revision:	1.13
Committed:	Sun Dec 21 12:24:48 2003 UTC (20 years, 5 months ago) by dl
Branch:	MAIN
Changes since 1.12:	+9 -0 lines
Log Message:	Documentation improvments; support two-arg submit
#	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.util.*;
10
11	/**
12	* Provides default implementation of {@link ExecutorService}
13	* execution methods. This class implements the <tt>submit</tt>,
14	* <tt>invokeAny</tt> and <tt>invokeAll</tt> methods using the default
15	* {@link FutureTask} class provided in this package. For example,
16	* the the implementation of <tt>submit(Runnable)</tt> creates an
17	* associated <tt>FutureTask</tt> that is executed and
18	* returned. Subclasses overriding these methods to use different
19	* {@link Future} implementations should do so consistently for each
20	* of these methods.
21	*
22	* @since 1.5
23	* @author Doug Lea
24	*/
25	public abstract class AbstractExecutorService implements ExecutorService {
26
27	public Future<?> submit(Runnable task) {
28	if (task == null) throw new NullPointerException();
29	FutureTask<Object> ftask = new FutureTask<Object>(task, null);
30	execute(ftask);
31	return ftask;
32	}
33
34	public <T> Future<T> submit(Runnable task, T result) {
35	if (task == null) throw new NullPointerException();
36	FutureTask<T> ftask = new FutureTask<T>(task, result);
37	execute(ftask);
38	return ftask;
39	}
40
41	public <T> Future<T> submit(Callable<T> task) {
42	if (task == null) throw new NullPointerException();
43	FutureTask<T> ftask = new FutureTask<T>(task);
44	execute(ftask);
45	return ftask;
46	}
47
48	// any/all methods, each a little bit different than the other
49
50
51	public <T> T invokeAny(Collection<Callable<T>> tasks)
52	throws InterruptedException, ExecutionException {
53	if (tasks == null)
54	throw new NullPointerException();
55	int n = tasks.size();
56	if (n == 0)
57	throw new IllegalArgumentException();
58	List<Future<T>> futures= new ArrayList<Future<T>>(n);
59	ExecutorCompletionService<T> ecs =
60	new ExecutorCompletionService<T>(this);
61	try {
62	for (Callable<T> t : tasks)
63	futures.add(ecs.submit(t));
64	ExecutionException ee = null;
65	RuntimeException re = null;
66	while (n-- > 0) {
67	Future<T> f = ecs.take();
68	try {
69	return f.get();
70	} catch(ExecutionException eex) {
71	ee = eex;
72	} catch(RuntimeException rex) {
73	re = rex;
74	}
75	}
76	if (ee != null)
77	throw ee;
78	if (re != null)
79	throw new ExecutionException(re);
80	throw new ExecutionException();
81	} finally {
82	for (Future<T> f : futures)
83	f.cancel(true);
84	}
85	}
86
87	public <T> T invokeAny(Collection<Callable<T>> tasks,
88	long timeout, TimeUnit unit)
89	throws InterruptedException, ExecutionException, TimeoutException {
90	if (tasks == null \|\| unit == null)
91	throw new NullPointerException();
92	long nanos = unit.toNanos(timeout);
93	int n = tasks.size();
94	if (n == 0)
95	throw new IllegalArgumentException();
96	List<Future<T>> futures= new ArrayList<Future<T>>(n);
97	ExecutorCompletionService<T> ecs =
98	new ExecutorCompletionService<T>(this);
99	try {
100	for (Callable<T> t : tasks)
101	futures.add(ecs.submit(t));
102	ExecutionException ee = null;
103	RuntimeException re = null;
104	long lastTime = System.nanoTime();
105	while (n-- > 0) {
106	Future<T> f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
107	if (f == null)
108	throw new TimeoutException();
109	try {
110	return f.get();
111	} catch(ExecutionException eex) {
112	ee = eex;
113	} catch(RuntimeException rex) {
114	re = rex;
115	}
116	long now = System.nanoTime();
117	nanos -= now - lastTime;
118	lastTime = now;
119	}
120	if (ee != null)
121	throw ee;
122	if (re != null)
123	throw new ExecutionException(re);
124	throw new ExecutionException();
125	} finally {
126	for (Future<T> f : futures)
127	f.cancel(true);
128	}
129	}
130
131	public <T> List<Future<T>> invokeAll(Collection<Callable<T>> tasks)
132	throws InterruptedException {
133	if (tasks == null)
134	throw new NullPointerException();
135	List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
136	boolean done = false;
137	try {
138	for (Callable<T> t : tasks) {
139	FutureTask<T> f = new FutureTask<T>(t);
140	futures.add(f);
141	execute(f);
142	}
143	for (Future<T> f : futures) {
144	if (!f.isDone()) {
145	try {
146	f.get();
147	} catch(CancellationException ignore) {
148	} catch(ExecutionException ignore) {
149	}
150	}
151	}
152	done = true;
153	return futures;
154	} finally {
155	if (!done)
156	for (Future<T> f : futures)
157	f.cancel(true);
158	}
159	}
160
161	public <T> List<Future<T>> invokeAll(Collection<Callable<T>> tasks,
162	long timeout, TimeUnit unit)
163	throws InterruptedException {
164	if (tasks == null \|\| unit == null)
165	throw new NullPointerException();
166	long nanos = unit.toNanos(timeout);
167	List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
168	boolean done = false;
169	try {
170	for (Callable<T> t : tasks) {
171	FutureTask<T> f = new FutureTask<T>(t);
172	futures.add(f);
173	execute(f);
174	}
175	long lastTime = System.nanoTime();
176	for (Future<T> f : futures) {
177	if (!f.isDone()) {
178	if (nanos < 0)
179	return futures;
180	try {
181	f.get(nanos, TimeUnit.NANOSECONDS);
182	} catch(CancellationException ignore) {
183	} catch(ExecutionException ignore) {
184	} catch(TimeoutException toe) {
185	return futures;
186	}
187	long now = System.nanoTime();
188	nanos -= now - lastTime;
189	lastTime = now;
190	}
191	}
192	done = true;
193	return futures;
194	} finally {
195	if (!done)
196	for (Future<T> f : futures)
197	f.cancel(true);
198	}
199	}
200
201	}