util/concurrent/CancellableTask.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.concurrent.atomic.*;
import java.util.concurrent.locks.*;

/**
 * Base class for cancellable actions running in the Executor
 * framework. In addition to serving as a standalone class, this
 * provides <tt>protected</tt> functionality that may be useful when
 * creating customized task classes.
 * @since 1.5
 * @author Doug Lea
 */

public class CancellableTask implements Cancellable, Runnable {
    /** 
     * Holds the run-state, taking on values:
     *   null              = not yet started,
     *   [some thread ref] = running,
     *   DONE              = completed normally,
     *   CANCELLED         = cancelled (may or may not have ever run).
     * Transitions use atomic updates.
     */
    private volatile Object runner;

    /**
     * Special value for "runner" indicating task is completed
     */
    private static final Object DONE = new Object();

    /**
     * Special value for "runner" indicating task is cancelled
     */
    private static final Object CANCELLED = new Object();

    private static AtomicReferenceFieldUpdater<CancellableTask, Object> 
        runnerUpdater = 
        AtomicReferenceFieldUpdater.newUpdater
        (CancellableTask.class, Object.class, "runner");

    /**
     * The runnable underlying this task
     */
    private volatile Runnable runnable;

    /**
     * Creates a new CancellableTask which invokes the given
     * <tt>Runnable</tt> when executed.
     * @param r the runnable action
     */
    public CancellableTask(Runnable r) {
        this.runnable = r;
    }

    /**
     * Creates a new CancellableTask without a runnable action, which
     * must be set using <tt>setRunnable</tt> before use.  This is
     * intended for use in subclasses that must complete superclass
     * construction beofre establishing the runnable action.
     */
    protected CancellableTask() {
    }


    public boolean cancel(boolean mayInterruptIfRunning) {
        Object r = runner;
        if (r == DONE || r == CANCELLED)
            return false;

        if (mayInterruptIfRunning &&
            r != null &&
            r instanceof Thread &&
            runnerUpdater.compareAndSet(this, r, CANCELLED)) 

            ((Thread)r).interrupt();
        else 
            runnerUpdater.set(this, CANCELLED);
        return true;
    }
    
    public boolean isCancelled() {
        return runner == CANCELLED;
    }
    
    public boolean isDone() {
        Object r = runner;
        return r == DONE || r == CANCELLED;
    }

    /**
     * Return the Runnable forming the basis of this task.
     * @return the runnable action
     * @see #setRunnable
     */
    protected Runnable getRunnable() {
        return runnable;
    }

    /**
     * Set the Runnable forming the basis of this task.
     * @param r the runnable action
     * @see #getRunnable
     */
    protected void setRunnable(Runnable r) {
        runnable = r;
    }

    /**
     * Set the state of this task to Cancelled.
     */
    protected void setCancelled() {
        runnerUpdater.set(this, CANCELLED);
    }
    
    /**
     * Set the state of this task to Done, unless already
     * in a Cancelled state, in which Cancelled status is preserved.
     *
     */
    protected void setDone() {
        for (;;) {
            Object r = runner;
            if (r == DONE || r == CANCELLED) 
                return;
            if (runnerUpdater.compareAndSet(this, r, DONE))
                return;
        }
    }

    /**
     * Attempt to set the state of this task to Running, succeeding
     * only if the state is currently NOT Done, Running, or Cancelled.
     * @return true if successful
     */ 
    protected boolean setRunning() {
        return runnerUpdater.compareAndSet(this, null, Thread.currentThread());
    }

    public void run() {
        if (setRunning()) {
            try {
                runnable.run();
            } finally {
                setDone();
            }
        }
    }

    /**
     * Implementation of Future methods under the control of a current
     * <tt>CancellableTask</tt>, which it relies on for methods
     * <tt>isDone</tt>, <tt>isCancelled</tt> and <tt>cancel</tt>. This
     * class is split into an inner class to permit Future support to
     * be mixed-in with other flavors of tasks.  Normally, such a
     * class will delegate <tt>Future</tt> <tt>get</tt> methods to the
     * <tt>InnerCancellableFuture</tt>, and internally arrange that
     * <tt>set</tt> methods be invoked when computations are ready.
     *
     * <p><b>Sample Usage</b>. Here are fragments of an example subclass.
     * <pre>
     *  class MyFutureTask&lt;V&gt; extends CancellableTask implements Future&lt;
V&gt; {
     *                
     *    MyFutureTask(Callable&lt;V&gt; callable) {
     *      setRunnable(new InnerCancellableFuture&lt;V&gt;(callable));
     *    }
     *
     *    public V get() throws InterruptedException, ExecutionException {
     *      return ((InnerCancellableFuture&lt;V&gt;)getRunnable()).get();
     *    }
     *    // (And similarly for timeout version.)
     *
     *    void action() { // whatever action causes execution
     *      try {
     *        ((InnerCancellableFuture&lt;V&gt;)getRunnable()).set(compute());
     *      } catch (Exception ex) {
     *        ((InnerCancellableFuture&lt;V&gt;)getRunnable()).setException(ex);
     *      }
     *   }
     * }
     *</pre>
     */
    protected class InnerCancellableFuture<V> implements Future<V>, Runnable {
        private final Callable<V> callable;
        private final ReentrantLock lock = new ReentrantLock();
        private final Condition accessible = lock.newCondition();
        private V result;
        private Throwable exception;

        /**
         * Create an InnerCancellableFuture that will execute the
         * given callable.
         * @param callable the function to execute
         */
        protected InnerCancellableFuture(Callable<V> callable) {
            this.callable = callable;
        }
        
        public boolean cancel(boolean mayInterruptIfRunning) {
            return CancellableTask.this.cancel(mayInterruptIfRunning);
        }
        
        public boolean isCancelled() {
            return CancellableTask.this.isCancelled();
        }


        public boolean isDone() {
            return CancellableTask.this.isDone();
        }


        /**
         * Sets this Future to the results of <tt>callable.call</tt>
         */
        public void run() {
            try {
                set(callable.call());
            } catch(Throwable ex) {
                setException(ex);
            }
        }

        /**
         * Waits if necessary for the call to <tt>callable.call</tt> to
         * complete, and then retrieves its result.
         *
         * @return computed result
         * @throws CancellationException here???
         * @throws ExecutionException if underlying computation threw an
         * exception
         * @throws InterruptedException if current thread was interrupted
         * while waiting
         */
        public V get() throws InterruptedException, ExecutionException {
            lock.lock();
            try {
                while (!isDone())
                    accessible.await();
                if (isCancelled())
                    throw new CancellationException();
                else if (exception != null)
                    throw new ExecutionException(exception);
                else
                    return result;
            } finally {
                lock.unlock();
            }
        }

        /**
         * Waits if necessary for at most the given time for the call to
         * <tt>callable.call</tt> to complete, and then retrieves its
         * result.
         *
         * @param timeout the maximum time to wait
         * @param granularity the time unit of the timeout argument
         * @return computed result
         * @throws ExecutionException if underlying computation threw an
         * exception
         * @throws InterruptedException if current thread was interrupted
         * while waiting
         * @throws TimeoutException if the wait timed out
         */
        public V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
            lock.lock();
            try {
                if (!isDone()) {
                    long nanos = unit.toNanos(timeout);
                    do {
                        if (nanos <= 0)
                            throw new TimeoutException();
                        nanos = accessible.awaitNanos(nanos);
                    } while (!isDone());
                }
                if (isCancelled())
                    throw new CancellationException();
                else if (exception != null)
                    throw new ExecutionException(exception);
                else
                    return result;
            } finally {
                lock.unlock();
            }
        }

        /**
         * Sets the result of this Future to the given value.
         * @param v the value
         */ 
        protected void set(V v) {
            lock.lock();
            try {
                result = v;
                setDone();
                accessible.signalAll();
            } finally {
                lock.unlock();
            }
        }

        /**
         * Causes this futue to report an <tt>ExecutionException</tt>
         * with the given throwable as its cause.
         * @param t the cause of failure.
         */ 
        protected void setException(Throwable t) {
            lock.lock();
            try {
                exception = t;
                setDone();
                accessible.signalAll();
            } finally {
                lock.unlock();
            }
        }
    }
   
}
Revision:	1.13
Committed:	Sun Aug 24 14:47:31 2003 UTC (20 years, 9 months ago) by dl
Branch:	MAIN
Changes since 1.12:	+69 -2 lines
Log Message:	Javadoc clarifications
#	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	import java.util.concurrent.atomic.*;
9	import java.util.concurrent.locks.*;
10
11	/**
12	* Base class for cancellable actions running in the Executor
13	* framework. In addition to serving as a standalone class, this
14	* provides <tt>protected</tt> functionality that may be useful when
15	* creating customized task classes.
16	* @since 1.5
17	* @author Doug Lea
18	*/
19
20	public class CancellableTask implements Cancellable, Runnable {
21	/**
22	* Holds the run-state, taking on values:
23	* null = not yet started,
24	* [some thread ref] = running,
25	* DONE = completed normally,
26	* CANCELLED = cancelled (may or may not have ever run).
27	* Transitions use atomic updates.
28	*/
29	private volatile Object runner;
30
31	/**
32	* Special value for "runner" indicating task is completed
33	*/
34	private static final Object DONE = new Object();
35
36	/**
37	* Special value for "runner" indicating task is cancelled
38	*/
39	private static final Object CANCELLED = new Object();
40
41	private static AtomicReferenceFieldUpdater<CancellableTask, Object>
42	runnerUpdater =
43	AtomicReferenceFieldUpdater.newUpdater
44	(CancellableTask.class, Object.class, "runner");
45
46	/**
47	* The runnable underlying this task
48	*/
49	private volatile Runnable runnable;
50
51	/**
52	* Creates a new CancellableTask which invokes the given
53	* <tt>Runnable</tt> when executed.
54	* @param r the runnable action
55	*/
56	public CancellableTask(Runnable r) {
57	this.runnable = r;
58	}
59
60	/**
61	* Creates a new CancellableTask without a runnable action, which
62	* must be set using <tt>setRunnable</tt> before use. This is
63	* intended for use in subclasses that must complete superclass
64	* construction beofre establishing the runnable action.
65	*/
66	protected CancellableTask() {
67	}
68
69
70	public boolean cancel(boolean mayInterruptIfRunning) {
71	Object r = runner;
72	if (r == DONE \|\| r == CANCELLED)
73	return false;
74
75	if (mayInterruptIfRunning &&
76	r != null &&
77	r instanceof Thread &&
78	runnerUpdater.compareAndSet(this, r, CANCELLED))
79
80	((Thread)r).interrupt();
81	else
82	runnerUpdater.set(this, CANCELLED);
83	return true;
84	}
85
86	public boolean isCancelled() {
87	return runner == CANCELLED;
88	}
89
90	public boolean isDone() {
91	Object r = runner;
92	return r == DONE \|\| r == CANCELLED;
93	}
94
95	/**
96	* Return the Runnable forming the basis of this task.
97	* @return the runnable action
98	* @see #setRunnable
99	*/
100	protected Runnable getRunnable() {
101	return runnable;
102	}
103
104	/**
105	* Set the Runnable forming the basis of this task.
106	* @param r the runnable action
107	* @see #getRunnable
108	*/
109	protected void setRunnable(Runnable r) {
110	runnable = r;
111	}
112
113	/**
114	* Set the state of this task to Cancelled.
115	*/
116	protected void setCancelled() {
117	runnerUpdater.set(this, CANCELLED);
118	}
119
120	/**
121	* Set the state of this task to Done, unless already
122	* in a Cancelled state, in which Cancelled status is preserved.
123	*
124	*/
125	protected void setDone() {
126	for (;;) {
127	Object r = runner;
128	if (r == DONE \|\| r == CANCELLED)
129	return;
130	if (runnerUpdater.compareAndSet(this, r, DONE))
131	return;
132	}
133	}
134
135	/**
136	* Attempt to set the state of this task to Running, succeeding
137	* only if the state is currently NOT Done, Running, or Cancelled.
138	* @return true if successful
139	*/
140	protected boolean setRunning() {
141	return runnerUpdater.compareAndSet(this, null, Thread.currentThread());
142	}
143
144	public void run() {
145	if (setRunning()) {
146	try {
147	runnable.run();
148	} finally {
149	setDone();
150	}
151	}
152	}
153
154	/**
155	* Implementation of Future methods under the control of a current
156	* <tt>CancellableTask</tt>, which it relies on for methods
157	* <tt>isDone</tt>, <tt>isCancelled</tt> and <tt>cancel</tt>. This
158	* class is split into an inner class to permit Future support to
159	* be mixed-in with other flavors of tasks. Normally, such a
160	* class will delegate <tt>Future</tt> <tt>get</tt> methods to the
161	* <tt>InnerCancellableFuture</tt>, and internally arrange that
162	* <tt>set</tt> methods be invoked when computations are ready.
163	*
164	* <p><b>Sample Usage</b>. Here are fragments of an example subclass.
165	* <pre>
166	* class MyFutureTask<V> extends CancellableTask implements Future<
167	V> {
168	*
169	* MyFutureTask(Callable<V> callable) {
170	* setRunnable(new InnerCancellableFuture<V>(callable));
171	* }
172	*
173	* public V get() throws InterruptedException, ExecutionException {
174	* return ((InnerCancellableFuture<V>)getRunnable()).get();
175	* }
176	* // (And similarly for timeout version.)
177	*
178	* void action() { // whatever action causes execution
179	* try {
180	* ((InnerCancellableFuture<V>)getRunnable()).set(compute());
181	* } catch (Exception ex) {
182	* ((InnerCancellableFuture<V>)getRunnable()).setException(ex);
183	* }
184	* }
185	* }
186	*</pre>
187	*/
188	protected class InnerCancellableFuture<V> implements Future<V>, Runnable {
189	private final Callable<V> callable;
190	private final ReentrantLock lock = new ReentrantLock();
191	private final Condition accessible = lock.newCondition();
192	private V result;
193	private Throwable exception;
194
195	/**
196	* Create an InnerCancellableFuture that will execute the
197	* given callable.
198	* @param callable the function to execute
199	*/
200	protected InnerCancellableFuture(Callable<V> callable) {
201	this.callable = callable;
202	}
203
204	public boolean cancel(boolean mayInterruptIfRunning) {
205	return CancellableTask.this.cancel(mayInterruptIfRunning);
206	}
207
208	public boolean isCancelled() {
209	return CancellableTask.this.isCancelled();
210	}
211
212
213	public boolean isDone() {
214	return CancellableTask.this.isDone();
215	}
216
217
218	/**
219	* Sets this Future to the results of <tt>callable.call</tt>
220	*/
221	public void run() {
222	try {
223	set(callable.call());
224	} catch(Throwable ex) {
225	setException(ex);
226	}
227	}
228
229	/**
230	* Waits if necessary for the call to <tt>callable.call</tt> to
231	* complete, and then retrieves its result.
232	*
233	* @return computed result
234	* @throws CancellationException here???
235	* @throws ExecutionException if underlying computation threw an
236	* exception
237	* @throws InterruptedException if current thread was interrupted
238	* while waiting
239	*/
240	public V get() throws InterruptedException, ExecutionException {
241	lock.lock();
242	try {
243	while (!isDone())
244	accessible.await();
245	if (isCancelled())
246	throw new CancellationException();
247	else if (exception != null)
248	throw new ExecutionException(exception);
249	else
250	return result;
251	} finally {
252	lock.unlock();
253	}
254	}
255
256	/**
257	* Waits if necessary for at most the given time for the call to
258	* <tt>callable.call</tt> to complete, and then retrieves its
259	* result.
260	*
261	* @param timeout the maximum time to wait
262	* @param granularity the time unit of the timeout argument
263	* @return computed result
264	* @throws ExecutionException if underlying computation threw an
265	* exception
266	* @throws InterruptedException if current thread was interrupted
267	* while waiting
268	* @throws TimeoutException if the wait timed out
269	*/
270	public V get(long timeout, TimeUnit unit)
271	throws InterruptedException, ExecutionException, TimeoutException {
272	lock.lock();
273	try {
274	if (!isDone()) {
275	long nanos = unit.toNanos(timeout);
276	do {
277	if (nanos <= 0)
278	throw new TimeoutException();
279	nanos = accessible.awaitNanos(nanos);
280	} while (!isDone());
281	}
282	if (isCancelled())
283	throw new CancellationException();
284	else if (exception != null)
285	throw new ExecutionException(exception);
286	else
287	return result;
288	} finally {
289	lock.unlock();
290	}
291	}
292
293	/**
294	* Sets the result of this Future to the given value.
295	* @param v the value
296	*/
297	protected void set(V v) {
298	lock.lock();
299	try {
300	result = v;
301	setDone();
302	accessible.signalAll();
303	} finally {
304	lock.unlock();
305	}
306	}
307
308	/**
309	* Causes this futue to report an <tt>ExecutionException</tt>
310	* with the given throwable as its cause.
311	* @param t the cause of failure.
312	*/
313	protected void setException(Throwable t) {
314	lock.lock();
315	try {
316	exception = t;
317	setDone();
318	accessible.signalAll();
319	} finally {
320	lock.unlock();
321	}
322	}
323	}
324
325	}