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 {@link Cancellable} {@link java.lang.Runnable} 
 * actions within the {@link 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
     * @throws NullPointerException if runnable is null
     */
    public CancellableTask(Runnable r) {
        if (r == null)
            throw new NullPointerException();
        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 before 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);

        done();
        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)) {
                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());
    }

    /**
     * Runs the runnable if not cancelled, maintaining run status.
     */
    public void run() {
        if (setRunning()) {
            try {
                runnable.run();
            } finally {
                setDone();
            }
        }
    }

    /**
     * Protected method invoked when this task transitions to state
     * <tt>isDone</tt> (whether normally or via cancellation). The
     * default implementation does nothing.  Subclasses may override
     * this method to invoke completion callbacks or perform
     * bookkeeping. Note that you can query status inside the
     * implementation of this method to determine whether this task
     * has been cancelled.
     */
    protected void done() { }

    /**
     * Reset the run state of this task to its initial state unless
     * it has been cancelled. (Note that a cancelled task cannot be
     * reset.)
     * @return true if successful
     */
    protected boolean reset() {
        for (;;) {
            Object r = runner;
            if (r == CANCELLED) 
                return false;
            if (runnerUpdater.compareAndSet(this, r, null))
                return true;
        }
    }

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