util/concurrent/Executors.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.*;
import java.security.AccessControlContext;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.security.PrivilegedExceptionAction;

/**
 * Factory and utility methods for {@link Executor}, {@link
 * ExecutorService}, {@link Future}, and {@link Cancellable} classes
 * defined in this package.
 *
 * @since 1.5
 * @author Doug Lea
 */
public class Executors {

    /**
     * A wrapper class that exposes only the ExecutorService methods
     * of an implementation.
     */
    private static class DelegatedExecutorService implements ExecutorService {
        private final ExecutorService e;
        DelegatedExecutorService(ExecutorService executor) { e = executor; }
        public void execute(Runnable command) { e.execute(command); }
        public void shutdown() { e.shutdown(); }
        public List shutdownNow() { return e.shutdownNow(); }
        public boolean isShutdown() { return e.isShutdown(); }
        public boolean isTerminated() { return e.isTerminated(); }
        public boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException {
            return e.awaitTermination(timeout, unit);
        }
    }

    /**
     * Creates a thread pool that reuses a fixed set of threads
     * operating off a shared unbounded queue. If any thread
     * terminates due to a failure during execution prior to shutdown,
     * a new one will take its place if needed to execute subsequent
     * tasks.
     *
     * @param nThreads the number of threads in the pool
     * @return the newly created thread pool
     */
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new DelegatedExecutorService
            (new ThreadPoolExecutor(nThreads, nThreads,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

    /**
     * Creates a thread pool that reuses a fixed set of threads
     * operating off a shared unbounded queue, using the provided
     * ThreadFactory to create new threads when needed.
     *
     * @param nThreads the number of threads in the pool
     * @param threadFactory the factory to use when creating new threads
     * @return the newly created thread pool
     */
    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new DelegatedExecutorService
            (new ThreadPoolExecutor(nThreads, nThreads,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }

    /**
     * Creates an Executor that uses a single worker thread operating
     * off an unbounded queue. (Note however that if this single
     * thread terminates due to a failure during execution prior to
     * shutdown, a new one will take its place if needed to execute
     * subsequent tasks.)  Tasks are guaranteed to execute
     * sequentially, and no more than one task will be active at any
     * given time. This method is equivalent in effect to
     *<tt>new FixedThreadPool(1)</tt>.
     *
     * @return the newly-created single-threaded Executor
     */
    public static ExecutorService newSingleThreadExecutor() {
        return new DelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

    /**
     * Creates an Executor that uses a single worker thread operating
     * off an unbounded queue, and uses the provided ThreadFactory to
     * create new threads when needed.
     * @param threadFactory the factory to use when creating new
     * threads
     *
     * @return the newly-created single-threaded Executor
     */
    public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
        return new DelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }

    /**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available.  These pools will typically improve the performance
     * of programs that execute many short-lived asynchronous tasks.
     * Calls to <tt>execute</tt> will reuse previously constructed
     * threads if available. If no existing thread is available, a new
     * thread will be created and added to the pool. Threads that have
     * not been used for sixty seconds are terminated and removed from
     * the cache. Thus, a pool that remains idle for long enough will
     * not consume any resources. Note that pools with similar
     * properties but different details (for example, timeout parameters)
     * may be created using {@link ThreadPoolExecutor} constructors.
     *
     * @return the newly created thread pool
     */
    public static ExecutorService newCachedThreadPool() {
        return new DelegatedExecutorService
            (new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                    60, TimeUnit.SECONDS,
                                    new SynchronousQueue<Runnable>()));
    }

    /**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available, and uses the provided
     * ThreadFactory to create new threads when needed.
     * @param threadFactory the factory to use when creating new threads
     * @return the newly created thread pool
     */
    public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
        return new DelegatedExecutorService
            (new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                    60, TimeUnit.SECONDS,
                                    new SynchronousQueue<Runnable>(),
                                    threadFactory));
    }

    /**
     * Executes a Runnable task and returns a Cancellable representing that
     * task.
     *
     * @param executor the Executor to which the task will be submitted
     * @param task the task to submit
     * @return a Cancellable representing pending completion of the task
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution
     */
    public static Cancellable execute(Executor executor, Runnable task) {
        FutureTask<Boolean> ftask = new FutureTask<Boolean>(task, Boolean.TRUE);
        executor.execute(ftask);
        return ftask;
    }

    /**
     * Executes a Runnable task and returns a Future representing that
     * task.
     *
     * @param executor the Executor to which the task will be submitted
     * @param task the task to submit
     * @param value the value which will become the return value of
     * the task upon task completion
     * @return a Future representing pending completion of the task
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution
     */
    public static <T> Future<T> execute(Executor executor, Runnable task, T value) {
        FutureTask<T> ftask = new FutureTask<T>(task, value);
        executor.execute(ftask);
        return ftask;
    }

    /**
     * Executes a value-returning task and returns a Future
     * representing the pending results of the task.
     *
     * @param executor the Executor to which the task will be submitted
     * @param task the task to submit
     * @return a Future representing pending completion of the task
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution
     */
    public static <T> Future<T> execute(Executor executor, Callable<T> task) {
        FutureTask<T> ftask = new FutureTask<T>(task);
        executor.execute(ftask);
        return ftask;
    }

    /**
     * Executes a Runnable task and blocks until it completes normally
     * or throws an exception.
     *
     * @param executor the Executor to which the task will be submitted
     * @param task the task to submit
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution
     * @throws ExecutionException if the task encountered an exception
     * while executing
     */
    public static void invoke(Executor executor, Runnable task)
            throws ExecutionException, InterruptedException {
        FutureTask<Boolean> ftask = new FutureTask<Boolean>(task, Boolean.TRUE);
        executor.execute(ftask);
        ftask.get();
    }

    /**
     * Executes a value-returning task and blocks until it returns a
     * value or throws an exception.
     *
     * @param executor the Executor to which the task will be submitted
     * @param task the task to submit
     * @return a Future representing pending completion of the task
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution
     * @throws InterruptedException if interrupted while waiting for
     * completion
     * @throws ExecutionException if the task encountered an exception
     * while executing
     */
    public static <T> T invoke(Executor executor, Callable<T> task)
            throws ExecutionException, InterruptedException {
        FutureTask<T> ftask = new FutureTask<T>(task);
        executor.execute(ftask);
        return ftask.get();
    }


    /**
     * Executes a privileged action under the current access control 
     * context and returns a Future representing the pending result 
     * object of that action.
     *
     * @param executor the Executor to which the task will be submitted
     * @param action the action to submit
     * @return a Future representing pending completion of the action
     * @throws RejectedExecutionException if action cannot be scheduled
     * for execution
     */
    public static Future<Object> execute(Executor executor, PrivilegedAction action) {
        return execute(executor, action, AccessController.getContext());
    }
    
    /**
     * Executes a privileged action under the given access control 
     * context and returns a Future representing the pending result 
     * object of that action.
     *
     * @param executor the Executor to which the task will be submitted
     * @param action the action to submit
     * @param acc the access control context under which action should run
     * @return a Future representing pending completion of the action
     * @throws RejectedExecutionException if action cannot be scheduled
     * for execution
     */
    public static Future<Object> execute(Executor executor, PrivilegedAction action,
                                         AccessControlContext acc) {
        Callable<Object> task = new PrivilegedActionAdapter(action);
        FutureTask<Object> future = new PrivilegedFutureTask<Object>(task, acc);
        executor.execute(future);
        return future;
    }

    /**
     * Executes a privileged exception action under the current access control 
     * context and returns a Future representing the pending result 
     * object of that action.
     *
     * @param executor the Executor to which the task will be submitted
     * @param action the action to submit
     * @return a Future representing pending completion of the action
     * @throws RejectedExecutionException if action cannot be scheduled
     * for execution
     */
    public static Future<Object> execute(Executor executor, PrivilegedExceptionAction action) {
        return execute(executor, action, AccessController.getContext());
    }
    
    /**
     * Executes a privileged exception action under the given access control 
     * context and returns a Future representing the pending result 
     * object of that action.
     *
     * @param executor the Executor to which the task will be submitted
     * @param action the action to submit
     * @param acc the access control context under which action should run
     * @return a Future representing pending completion of the action
     * @throws RejectedExecutionException if action cannot be scheduled
     * for execution
     */
    public static Future<Object> execute(Executor executor, PrivilegedExceptionAction action,
                                         AccessControlContext acc) {
        Callable<Object> task = new PrivilegedExceptionActionAdapter(action);
        FutureTask<Object> future = new PrivilegedFutureTask<Object>(task, acc);
        executor.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;
    }
        
        
    /** Cannot instantiate. */
    private Executors() {}
}
Revision:	1.20
Committed:	Sat Oct 25 04:23:25 2003 UTC (20 years, 7 months ago) by tim
Branch:	MAIN
Changes since 1.19:	+96 -0 lines
Log Message:	Add PrivilegedFutureTask along with related methods in Executors.
#	User	Rev	Content
1	tim	1.1	/*
2	dl	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	tim	1.1	*/
6
7			package java.util.concurrent;
8	dl	1.2	import java.util.*;
9	tim	1.20	import java.security.AccessControlContext;
10			import java.security.AccessController;
11			import java.security.PrivilegedAction;
12			import java.security.PrivilegedExceptionAction;
13	tim	1.1
14			/**
15	dl	1.18	* Factory and utility methods for {@link Executor}, {@link
16	dl	1.17	* ExecutorService}, {@link Future}, and {@link Cancellable} classes
17			* defined in this package.
18	tim	1.1	*
19			* @since 1.5
20	dl	1.12	* @author Doug Lea
21	tim	1.1	*/
22			public class Executors {
23
24			/**
25	dl	1.2	* A wrapper class that exposes only the ExecutorService methods
26			* of an implementation.
27	tim	1.6	*/
28	jozart	1.11	private static class DelegatedExecutorService implements ExecutorService {
29	dl	1.2	private final ExecutorService e;
30			DelegatedExecutorService(ExecutorService executor) { e = executor; }
31			public void execute(Runnable command) { e.execute(command); }
32			public void shutdown() { e.shutdown(); }
33			public List shutdownNow() { return e.shutdownNow(); }
34			public boolean isShutdown() { return e.isShutdown(); }
35			public boolean isTerminated() { return e.isTerminated(); }
36			public boolean awaitTermination(long timeout, TimeUnit unit)
37			throws InterruptedException {
38			return e.awaitTermination(timeout, unit);
39			}
40			}
41
42			/**
43	tim	1.1	* Creates a thread pool that reuses a fixed set of threads
44	dl	1.16	* operating off a shared unbounded queue. If any thread
45			* terminates due to a failure during execution prior to shutdown,
46			* a new one will take its place if needed to execute subsequent
47			* tasks.
48	tim	1.1	*
49			* @param nThreads the number of threads in the pool
50			* @return the newly created thread pool
51			*/
52	dl	1.2	public static ExecutorService newFixedThreadPool(int nThreads) {
53			return new DelegatedExecutorService
54			(new ThreadPoolExecutor(nThreads, nThreads,
55			0L, TimeUnit.MILLISECONDS,
56	dl	1.3	new LinkedBlockingQueue<Runnable>()));
57	dl	1.2	}
58
59			/**
60			* Creates a thread pool that reuses a fixed set of threads
61			* operating off a shared unbounded queue, using the provided
62			* ThreadFactory to create new threads when needed.
63			*
64			* @param nThreads the number of threads in the pool
65	dl	1.12	* @param threadFactory the factory to use when creating new threads
66	dl	1.2	* @return the newly created thread pool
67			*/
68			public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
69			return new DelegatedExecutorService
70			(new ThreadPoolExecutor(nThreads, nThreads,
71			0L, TimeUnit.MILLISECONDS,
72	dl	1.3	new LinkedBlockingQueue<Runnable>(),
73	dl	1.14	threadFactory));
74	dl	1.2	}
75
76			/**
77			* Creates an Executor that uses a single worker thread operating
78			* off an unbounded queue. (Note however that if this single
79			* thread terminates due to a failure during execution prior to
80			* shutdown, a new one will take its place if needed to execute
81			* subsequent tasks.) Tasks are guaranteed to execute
82			* sequentially, and no more than one task will be active at any
83	dl	1.16	* given time. This method is equivalent in effect to
84			*<tt>new FixedThreadPool(1)</tt>.
85	dl	1.2	*
86			* @return the newly-created single-threaded Executor
87			*/
88			public static ExecutorService newSingleThreadExecutor() {
89			return new DelegatedExecutorService
90	tim	1.6	(new ThreadPoolExecutor(1, 1,
91	dl	1.2	0L, TimeUnit.MILLISECONDS,
92	dl	1.3	new LinkedBlockingQueue<Runnable>()));
93	dl	1.2	}
94
95			/**
96			* Creates an Executor that uses a single worker thread operating
97			* off an unbounded queue, and uses the provided ThreadFactory to
98			* create new threads when needed.
99	dl	1.12	* @param threadFactory the factory to use when creating new
100	dl	1.2	* threads
101			*
102			* @return the newly-created single-threaded Executor
103			*/
104			public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
105			return new DelegatedExecutorService
106	tim	1.6	(new ThreadPoolExecutor(1, 1,
107	dl	1.2	0L, TimeUnit.MILLISECONDS,
108	dl	1.3	new LinkedBlockingQueue<Runnable>(),
109	dl	1.14	threadFactory));
110	tim	1.1	}
111
112			/**
113			* Creates a thread pool that creates new threads as needed, but
114			* will reuse previously constructed threads when they are
115			* available. These pools will typically improve the performance
116			* of programs that execute many short-lived asynchronous tasks.
117			* Calls to <tt>execute</tt> will reuse previously constructed
118			* threads if available. If no existing thread is available, a new
119			* thread will be created and added to the pool. Threads that have
120			* not been used for sixty seconds are terminated and removed from
121			* the cache. Thus, a pool that remains idle for long enough will
122	dl	1.16	* not consume any resources. Note that pools with similar
123			* properties but different details (for example, timeout parameters)
124			* may be created using {@link ThreadPoolExecutor} constructors.
125	tim	1.1	*
126			* @return the newly created thread pool
127			*/
128	dl	1.2	public static ExecutorService newCachedThreadPool() {
129			return new DelegatedExecutorService
130			(new ThreadPoolExecutor(0, Integer.MAX_VALUE,
131			60, TimeUnit.SECONDS,
132	dl	1.3	new SynchronousQueue<Runnable>()));
133	tim	1.1	}
134
135			/**
136	dl	1.2	* Creates a thread pool that creates new threads as needed, but
137			* will reuse previously constructed threads when they are
138	tim	1.6	* available, and uses the provided
139	dl	1.2	* ThreadFactory to create new threads when needed.
140	dl	1.12	* @param threadFactory the factory to use when creating new threads
141	tim	1.1	* @return the newly created thread pool
142			*/
143	dl	1.2	public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
144			return new DelegatedExecutorService
145			(new ThreadPoolExecutor(0, Integer.MAX_VALUE,
146			60, TimeUnit.SECONDS,
147	dl	1.3	new SynchronousQueue<Runnable>(),
148	dl	1.14	threadFactory));
149	tim	1.1	}
150
151			/**
152	tim	1.15	* Executes a Runnable task and returns a Cancellable representing that
153			* task.
154			*
155			* @param executor the Executor to which the task will be submitted
156			* @param task the task to submit
157			* @return a Cancellable representing pending completion of the task
158			* @throws RejectedExecutionException if task cannot be scheduled
159			* for execution
160			*/
161			public static Cancellable execute(Executor executor, Runnable task) {
162			FutureTask<Boolean> ftask = new FutureTask<Boolean>(task, Boolean.TRUE);
163			executor.execute(ftask);
164			return ftask;
165			}
166
167			/**
168	tim	1.1	* Executes a Runnable task and returns a Future representing that
169			* task.
170			*
171			* @param executor the Executor to which the task will be submitted
172			* @param task the task to submit
173			* @param value the value which will become the return value of
174			* the task upon task completion
175	dl	1.5	* @return a Future representing pending completion of the task
176	jozart	1.9	* @throws RejectedExecutionException if task cannot be scheduled
177	tim	1.1	* for execution
178			*/
179	dl	1.5	public static <T> Future<T> execute(Executor executor, Runnable task, T value) {
180	dl	1.10	FutureTask<T> ftask = new FutureTask<T>(task, value);
181	tim	1.1	executor.execute(ftask);
182			return ftask;
183			}
184
185			/**
186			* Executes a value-returning task and returns a Future
187			* representing the pending results of the task.
188			*
189			* @param executor the Executor to which the task will be submitted
190			* @param task the task to submit
191			* @return a Future representing pending completion of the task
192	jozart	1.9	* @throws RejectedExecutionException if task cannot be scheduled
193			* for execution
194	tim	1.1	*/
195	tim	1.15	public static <T> Future<T> execute(Executor executor, Callable<T> task) {
196	dl	1.10	FutureTask<T> ftask = new FutureTask<T>(task);
197	tim	1.1	executor.execute(ftask);
198			return ftask;
199			}
200
201			/**
202			* Executes a Runnable task and blocks until it completes normally
203			* or throws an exception.
204			*
205			* @param executor the Executor to which the task will be submitted
206			* @param task the task to submit
207	jozart	1.9	* @throws RejectedExecutionException if task cannot be scheduled
208			* for execution
209	dl	1.19	* @throws ExecutionException if the task encountered an exception
210			* while executing
211	tim	1.1	*/
212			public static void invoke(Executor executor, Runnable task)
213			throws ExecutionException, InterruptedException {
214	tim	1.13	FutureTask<Boolean> ftask = new FutureTask<Boolean>(task, Boolean.TRUE);
215	tim	1.1	executor.execute(ftask);
216			ftask.get();
217			}
218
219			/**
220			* Executes a value-returning task and blocks until it returns a
221			* value or throws an exception.
222			*
223			* @param executor the Executor to which the task will be submitted
224			* @param task the task to submit
225			* @return a Future representing pending completion of the task
226	jozart	1.9	* @throws RejectedExecutionException if task cannot be scheduled
227			* for execution
228	dl	1.12	* @throws InterruptedException if interrupted while waiting for
229			* completion
230	dl	1.19	* @throws ExecutionException if the task encountered an exception
231			* while executing
232	tim	1.1	*/
233			public static <T> T invoke(Executor executor, Callable<T> task)
234			throws ExecutionException, InterruptedException {
235			FutureTask<T> ftask = new FutureTask<T>(task);
236			executor.execute(ftask);
237			return ftask.get();
238	tim	1.6	}
239
240	tim	1.20
241			/**
242			* Executes a privileged action under the current access control
243			* context and returns a Future representing the pending result
244			* object of that action.
245			*
246			* @param executor the Executor to which the task will be submitted
247			* @param action the action to submit
248			* @return a Future representing pending completion of the action
249			* @throws RejectedExecutionException if action cannot be scheduled
250			* for execution
251			*/
252			public static Future<Object> execute(Executor executor, PrivilegedAction action) {
253			return execute(executor, action, AccessController.getContext());
254			}
255
256			/**
257			* Executes a privileged action under the given access control
258			* context and returns a Future representing the pending result
259			* object of that action.
260			*
261			* @param executor the Executor to which the task will be submitted
262			* @param action the action to submit
263			* @param acc the access control context under which action should run
264			* @return a Future representing pending completion of the action
265			* @throws RejectedExecutionException if action cannot be scheduled
266			* for execution
267			*/
268			public static Future<Object> execute(Executor executor, PrivilegedAction action,
269			AccessControlContext acc) {
270			Callable<Object> task = new PrivilegedActionAdapter(action);
271			FutureTask<Object> future = new PrivilegedFutureTask<Object>(task, acc);
272			executor.execute(future);
273			return future;
274			}
275
276			/**
277			* Executes a privileged exception action under the current access control
278			* context and returns a Future representing the pending result
279			* object of that action.
280			*
281			* @param executor the Executor to which the task will be submitted
282			* @param action the action to submit
283			* @return a Future representing pending completion of the action
284			* @throws RejectedExecutionException if action cannot be scheduled
285			* for execution
286			*/
287			public static Future<Object> execute(Executor executor, PrivilegedExceptionAction action) {
288			return execute(executor, action, AccessController.getContext());
289			}
290
291			/**
292			* Executes a privileged exception action under the given access control
293			* context and returns a Future representing the pending result
294			* object of that action.
295			*
296			* @param executor the Executor to which the task will be submitted
297			* @param action the action to submit
298			* @param acc the access control context under which action should run
299			* @return a Future representing pending completion of the action
300			* @throws RejectedExecutionException if action cannot be scheduled
301			* for execution
302			*/
303			public static Future<Object> execute(Executor executor, PrivilegedExceptionAction action,
304			AccessControlContext acc) {
305			Callable<Object> task = new PrivilegedExceptionActionAdapter(action);
306			FutureTask<Object> future = new PrivilegedFutureTask<Object>(task, acc);
307			executor.execute(future);
308			return future;
309			}
310	tim	1.15
311	tim	1.20
312			private static class PrivilegedActionAdapter implements Callable<Object> {
313			PrivilegedActionAdapter(PrivilegedAction action) {
314			this.action = action;
315			}
316			public Object call () {
317			return action.run();
318			}
319			private final PrivilegedAction action;
320			}
321
322			private static class PrivilegedExceptionActionAdapter implements Callable<Object> {
323			PrivilegedExceptionActionAdapter(PrivilegedExceptionAction action) {
324			this.action = action;
325			}
326			public Object call () throws Exception {
327			return action.run();
328			}
329			private final PrivilegedExceptionAction action;
330			}
331
332
333	tim	1.15	/** Cannot instantiate. */
334			private Executors() {}
335	tim	1.1	}