0N/A/*
0N/A * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
0N/A *
0N/A * This code is free software; you can redistribute it and/or modify it
0N/A * under the terms of the GNU General Public License version 2 only, as
2362N/A * published by the Free Software Foundation. Oracle designates this
0N/A * particular file as subject to the "Classpath" exception as provided
2362N/A * by Oracle in the LICENSE file that accompanied this code.
0N/A *
0N/A * This code is distributed in the hope that it will be useful, but WITHOUT
0N/A * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
0N/A * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
0N/A * version 2 for more details (a copy is included in the LICENSE file that
0N/A * accompanied this code).
0N/A *
0N/A * You should have received a copy of the GNU General Public License version
0N/A * 2 along with this work; if not, write to the Free Software Foundation,
0N/A * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
0N/A *
2362N/A * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
2362N/A * or visit www.oracle.com if you need additional information or have any
2362N/A * questions.
0N/A */
0N/A
0N/A/*
0N/A * This file is available under and governed by the GNU General Public
0N/A * License version 2 only, as published by the Free Software Foundation.
0N/A * However, the following notice accompanied the original version of this
0N/A * file:
0N/A *
0N/A * Written by Doug Lea with assistance from members of JCP JSR-166
0N/A * Expert Group and released to the public domain, as explained at
3984N/A * http://creativecommons.org/publicdomain/zero/1.0/
0N/A */
0N/A
0N/Apackage java.util.concurrent;
0N/A
0N/A/**
0N/A * A {@link CompletionService} that uses a supplied {@link Executor}
0N/A * to execute tasks. This class arranges that submitted tasks are,
0N/A * upon completion, placed on a queue accessible using {@code take}.
0N/A * The class is lightweight enough to be suitable for transient use
0N/A * when processing groups of tasks.
0N/A *
0N/A * <p>
0N/A *
0N/A * <b>Usage Examples.</b>
0N/A *
0N/A * Suppose you have a set of solvers for a certain problem, each
0N/A * returning a value of some type {@code Result}, and would like to
0N/A * run them concurrently, processing the results of each of them that
0N/A * return a non-null value, in some method {@code use(Result r)}. You
0N/A * could write this as:
0N/A *
0N/A * <pre> {@code
0N/A * void solve(Executor e,
0N/A * Collection<Callable<Result>> solvers)
0N/A * throws InterruptedException, ExecutionException {
0N/A * CompletionService<Result> ecs
0N/A * = new ExecutorCompletionService<Result>(e);
0N/A * for (Callable<Result> s : solvers)
0N/A * ecs.submit(s);
0N/A * int n = solvers.size();
0N/A * for (int i = 0; i < n; ++i) {
0N/A * Result r = ecs.take().get();
0N/A * if (r != null)
0N/A * use(r);
0N/A * }
0N/A * }}</pre>
0N/A *
0N/A * Suppose instead that you would like to use the first non-null result
0N/A * of the set of tasks, ignoring any that encounter exceptions,
0N/A * and cancelling all other tasks when the first one is ready:
0N/A *
0N/A * <pre> {@code
0N/A * void solve(Executor e,
0N/A * Collection<Callable<Result>> solvers)
0N/A * throws InterruptedException {
0N/A * CompletionService<Result> ecs
0N/A * = new ExecutorCompletionService<Result>(e);
0N/A * int n = solvers.size();
0N/A * List<Future<Result>> futures
0N/A * = new ArrayList<Future<Result>>(n);
0N/A * Result result = null;
0N/A * try {
0N/A * for (Callable<Result> s : solvers)
0N/A * futures.add(ecs.submit(s));
0N/A * for (int i = 0; i < n; ++i) {
0N/A * try {
0N/A * Result r = ecs.take().get();
0N/A * if (r != null) {
0N/A * result = r;
0N/A * break;
0N/A * }
0N/A * } catch (ExecutionException ignore) {}
0N/A * }
0N/A * }
0N/A * finally {
0N/A * for (Future<Result> f : futures)
0N/A * f.cancel(true);
0N/A * }
0N/A *
0N/A * if (result != null)
0N/A * use(result);
0N/A * }}</pre>
0N/A */
0N/Apublic class ExecutorCompletionService<V> implements CompletionService<V> {
0N/A private final Executor executor;
0N/A private final AbstractExecutorService aes;
0N/A private final BlockingQueue<Future<V>> completionQueue;
0N/A
0N/A /**
0N/A * FutureTask extension to enqueue upon completion
0N/A */
0N/A private class QueueingFuture extends FutureTask<Void> {
0N/A QueueingFuture(RunnableFuture<V> task) {
0N/A super(task, null);
0N/A this.task = task;
0N/A }
0N/A protected void done() { completionQueue.add(task); }
0N/A private final Future<V> task;
0N/A }
0N/A
0N/A private RunnableFuture<V> newTaskFor(Callable<V> task) {
0N/A if (aes == null)
0N/A return new FutureTask<V>(task);
0N/A else
0N/A return aes.newTaskFor(task);
0N/A }
0N/A
0N/A private RunnableFuture<V> newTaskFor(Runnable task, V result) {
0N/A if (aes == null)
0N/A return new FutureTask<V>(task, result);
0N/A else
0N/A return aes.newTaskFor(task, result);
0N/A }
0N/A
0N/A /**
0N/A * Creates an ExecutorCompletionService using the supplied
0N/A * executor for base task execution and a
0N/A * {@link LinkedBlockingQueue} as a completion queue.
0N/A *
0N/A * @param executor the executor to use
0N/A * @throws NullPointerException if executor is {@code null}
0N/A */
0N/A public ExecutorCompletionService(Executor executor) {
0N/A if (executor == null)
0N/A throw new NullPointerException();
0N/A this.executor = executor;
0N/A this.aes = (executor instanceof AbstractExecutorService) ?
0N/A (AbstractExecutorService) executor : null;
0N/A this.completionQueue = new LinkedBlockingQueue<Future<V>>();
0N/A }
0N/A
0N/A /**
0N/A * Creates an ExecutorCompletionService using the supplied
0N/A * executor for base task execution and the supplied queue as its
0N/A * completion queue.
0N/A *
0N/A * @param executor the executor to use
0N/A * @param completionQueue the queue to use as the completion queue
0N/A * normally one dedicated for use by this service. This
0N/A * queue is treated as unbounded -- failed attempted
0N/A * {@code Queue.add} operations for completed taskes cause
0N/A * them not to be retrievable.
0N/A * @throws NullPointerException if executor or completionQueue are {@code null}
0N/A */
0N/A public ExecutorCompletionService(Executor executor,
0N/A BlockingQueue<Future<V>> completionQueue) {
0N/A if (executor == null || completionQueue == null)
0N/A throw new NullPointerException();
0N/A this.executor = executor;
0N/A this.aes = (executor instanceof AbstractExecutorService) ?
0N/A (AbstractExecutorService) executor : null;
0N/A this.completionQueue = completionQueue;
0N/A }
0N/A
0N/A public Future<V> submit(Callable<V> task) {
0N/A if (task == null) throw new NullPointerException();
0N/A RunnableFuture<V> f = newTaskFor(task);
0N/A executor.execute(new QueueingFuture(f));
0N/A return f;
0N/A }
0N/A
0N/A public Future<V> submit(Runnable task, V result) {
0N/A if (task == null) throw new NullPointerException();
0N/A RunnableFuture<V> f = newTaskFor(task, result);
0N/A executor.execute(new QueueingFuture(f));
0N/A return f;
0N/A }
0N/A
0N/A public Future<V> take() throws InterruptedException {
0N/A return completionQueue.take();
0N/A }
0N/A
0N/A public Future<V> poll() {
0N/A return completionQueue.poll();
0N/A }
0N/A
3203N/A public Future<V> poll(long timeout, TimeUnit unit)
3203N/A throws InterruptedException {
0N/A return completionQueue.poll(timeout, unit);
0N/A }
0N/A
0N/A}