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

    /**
     * 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 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 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 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.16
Committed:	Fri Sep 12 15:40:10 2003 UTC (20 years, 9 months ago) by dl
Branch:	MAIN
Changes since 1.15:	+15 -3 lines
Log Message:	Adapt AbstractQueue changes; Conditionalize CancellableTask.reset; new TimeUnit methods
#	User	Rev	Content
1	dl	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			import java.util.concurrent.atomic.*;
9	dl	1.6	import java.util.concurrent.locks.*;
10	dl	1.1
11			/**
12	dl	1.14	* 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	dl	1.5	* @since 1.5
18			* @author Doug Lea
19	dl	1.1	*/
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	dl	1.5	/**
33	dl	1.1	* Special value for "runner" indicating task is completed
34			*/
35			private static final Object DONE = new Object();
36
37	dl	1.5	/**
38	dl	1.1	* 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	dl	1.9	runnerUpdater =
44			AtomicReferenceFieldUpdater.newUpdater
45			(CancellableTask.class, Object.class, "runner");
46	dl	1.1
47			/**
48			* The runnable underlying this task
49			*/
50			private volatile Runnable runnable;
51
52			/**
53	brian	1.4	* Creates a new CancellableTask which invokes the given
54			* <tt>Runnable</tt> when executed.
55	dl	1.5	* @param r the runnable action
56	dl	1.16	* @throws NullPointerException if runnable is null
57	dl	1.1	*/
58			public CancellableTask(Runnable r) {
59	dl	1.16	if (r == null)
60			throw new NullPointerException();
61	dl	1.1	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	dl	1.14	* construction before establishing the runnable action.
69	dl	1.1	*/
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	dl	1.3	runnerUpdater.compareAndSet(this, r, CANCELLED))
83	dl	1.1
84			((Thread)r).interrupt();
85	dl	1.3	else
86	dl	1.1	runnerUpdater.set(this, CANCELLED);
87			return true;
88			}
89
90			public boolean isCancelled() {
91			return runner == CANCELLED;
92			}
93
94			public boolean isDone() {
95			Object r = runner;
96			return r == DONE \|\| r == CANCELLED;
97			}
98
99			/**
100			* Return the Runnable forming the basis of this task.
101	dl	1.5	* @return the runnable action
102	tim	1.10	* @see #setRunnable
103	dl	1.1	*/
104			protected Runnable getRunnable() {
105			return runnable;
106			}
107
108			/**
109			* Set the Runnable forming the basis of this task.
110	dl	1.5	* @param r the runnable action
111	tim	1.10	* @see #getRunnable
112	dl	1.1	*/
113			protected void setRunnable(Runnable r) {
114			runnable = r;
115			}
116
117			/**
118	tim	1.10	* Set the state of this task to Cancelled.
119	dl	1.1	*/
120			protected void setCancelled() {
121			runnerUpdater.set(this, CANCELLED);
122			}
123
124			/**
125			* Set the state of this task to Done, unless already
126			* in a Cancelled state, in which Cancelled status is preserved.
127			*
128			*/
129			protected void setDone() {
130			for (;;) {
131			Object r = runner;
132			if (r == DONE \|\| r == CANCELLED)
133			return;
134			if (runnerUpdater.compareAndSet(this, r, DONE))
135			return;
136			}
137			}
138
139			/**
140			* Attempt to set the state of this task to Running, succeeding
141			* only if the state is currently NOT Done, Running, or Cancelled.
142	dl	1.2	* @return true if successful
143	dl	1.1	*/
144			protected boolean setRunning() {
145			return runnerUpdater.compareAndSet(this, null, Thread.currentThread());
146			}
147
148			public void run() {
149			if (setRunning()) {
150			try {
151			runnable.run();
152	tim	1.11	} finally {
153	dl	1.1	setDone();
154			}
155			}
156			}
157
158			/**
159	dl	1.16	* Reset the run state of this task to its initial state unless
160			* it has been cancelled. (Note that a cancelled task cannot be
161			* reset.)
162			* @return true if successful
163	dl	1.15	*/
164	dl	1.16	protected boolean reset() {
165			for (;;) {
166			Object r = runner;
167			if (r == CANCELLED)
168			return false;
169			if (runnerUpdater.compareAndSet(this, r, null))
170			return true;
171			}
172	dl	1.15	}
173
174			/**
175	dl	1.1	* Implementation of Future methods under the control of a current
176	dl	1.13	* <tt>CancellableTask</tt>, which it relies on for methods
177			* <tt>isDone</tt>, <tt>isCancelled</tt> and <tt>cancel</tt>. This
178			* class is split into an inner class to permit Future support to
179			* be mixed-in with other flavors of tasks. Normally, such a
180			* class will delegate <tt>Future</tt> <tt>get</tt> methods to the
181			* <tt>InnerCancellableFuture</tt>, and internally arrange that
182			* <tt>set</tt> methods be invoked when computations are ready.
183			*
184			* <p><b>Sample Usage</b>. Here are fragments of an example subclass.
185			* <pre>
186			* class MyFutureTask<V> extends CancellableTask implements Future<
187			V> {
188			*
189			* MyFutureTask(Callable<V> callable) {
190			* setRunnable(new InnerCancellableFuture<V>(callable));
191			* }
192			*
193			* public V get() throws InterruptedException, ExecutionException {
194			* return ((InnerCancellableFuture<V>)getRunnable()).get();
195			* }
196			* // (And similarly for timeout version.)
197			*
198			* void action() { // whatever action causes execution
199			* try {
200			* ((InnerCancellableFuture<V>)getRunnable()).set(compute());
201			* } catch (Exception ex) {
202			* ((InnerCancellableFuture<V>)getRunnable()).setException(ex);
203			* }
204			* }
205			* }
206			*</pre>
207	dl	1.1	*/
208			protected class InnerCancellableFuture<V> implements Future<V>, Runnable {
209			private final Callable<V> callable;
210			private final ReentrantLock lock = new ReentrantLock();
211			private final Condition accessible = lock.newCondition();
212			private V result;
213			private Throwable exception;
214	dl	1.5
215			/**
216			* Create an InnerCancellableFuture that will execute the
217			* given callable.
218			* @param callable the function to execute
219			*/
220	dl	1.1	protected InnerCancellableFuture(Callable<V> callable) {
221			this.callable = callable;
222			}
223	tim	1.12
224			public boolean cancel(boolean mayInterruptIfRunning) {
225			return CancellableTask.this.cancel(mayInterruptIfRunning);
226			}
227
228			public boolean isCancelled() {
229			return CancellableTask.this.isCancelled();
230			}
231
232	dl	1.1
233			public boolean isDone() {
234			return CancellableTask.this.isDone();
235			}
236
237	dl	1.13
238			/**
239			* Sets this Future to the results of <tt>callable.call</tt>
240			*/
241	dl	1.1	public void run() {
242			try {
243			set(callable.call());
244	tim	1.11	} catch(Throwable ex) {
245	dl	1.1	setException(ex);
246			}
247			}
248
249	dl	1.13	/**
250			* Waits if necessary for the call to <tt>callable.call</tt> to
251			* complete, and then retrieves its result.
252			*
253			* @return computed result
254			* @throws CancellationException here???
255			* @throws ExecutionException if underlying computation threw an
256			* exception
257			* @throws InterruptedException if current thread was interrupted
258			* while waiting
259			*/
260	dl	1.1	public V get() throws InterruptedException, ExecutionException {
261			lock.lock();
262			try {
263			while (!isDone())
264			accessible.await();
265			if (isCancelled())
266			throw new CancellationException();
267			else if (exception != null)
268			throw new ExecutionException(exception);
269			else
270			return result;
271	tim	1.11	} finally {
272	dl	1.1	lock.unlock();
273			}
274			}
275
276	dl	1.13	/**
277			* Waits if necessary for at most the given time for the call to
278			* <tt>callable.call</tt> to complete, and then retrieves its
279			* result.
280			*
281			* @param timeout the maximum time to wait
282	dl	1.14	* @param unit the time unit of the timeout argument
283	dl	1.13	* @return computed result
284			* @throws ExecutionException if underlying computation threw an
285			* exception
286			* @throws InterruptedException if current thread was interrupted
287			* while waiting
288			* @throws TimeoutException if the wait timed out
289			*/
290	dl	1.1	public V get(long timeout, TimeUnit unit)
291			throws InterruptedException, ExecutionException, TimeoutException {
292			lock.lock();
293			try {
294			if (!isDone()) {
295			long nanos = unit.toNanos(timeout);
296			do {
297			if (nanos <= 0)
298			throw new TimeoutException();
299			nanos = accessible.awaitNanos(nanos);
300			} while (!isDone());
301			}
302			if (isCancelled())
303			throw new CancellationException();
304			else if (exception != null)
305			throw new ExecutionException(exception);
306			else
307			return result;
308	tim	1.11	} finally {
309	dl	1.1	lock.unlock();
310			}
311			}
312
313	dl	1.13	/**
314			* Sets the result of this Future to the given value.
315			* @param v the value
316			*/
317	dl	1.1	protected void set(V v) {
318			lock.lock();
319			try {
320			result = v;
321			setDone();
322			accessible.signalAll();
323	tim	1.11	} finally {
324	dl	1.1	lock.unlock();
325			}
326			}
327
328	dl	1.13	/**
329			* Causes this futue to report an <tt>ExecutionException</tt>
330			* with the given throwable as its cause.
331			* @param t the cause of failure.
332			*/
333	dl	1.1	protected void setException(Throwable t) {
334			lock.lock();
335			try {
336			exception = t;
337			setDone();
338			accessible.signalAll();
339	tim	1.11	} finally {
340	dl	1.1	lock.unlock();
341			}
342			}
343			}
344
345			}