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