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

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

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

    /**
     * Sets the state of this task to Cancelled.
     */
    protected void setCancelled() {
        runnerUpdater.set(this, CANCELLED);
    }
    
    /**
     * Sets the state of this task to Done, unless already in a
     * Cancelled state, in which case 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;
            }
        }
    }

    /**
     * Attempts 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() { }

    /**
     * Resets 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>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;

        /**
         * Creates 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 if underlying computation was
         * cancelled
         * @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 CancellationException if underlying computation was
         * cancelled
         * @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.25
Committed:	Thu Dec 4 20:54:28 2003 UTC (20 years, 6 months ago) by dl
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.24:	+0 -0 lines
State:	*FILE REMOVED*
Log Message:	Revised tests for revised Future classes
#	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	jozart	1.22
9	dl	1.1	import java.util.concurrent.atomic.*;
10	dl	1.6	import java.util.concurrent.locks.*;
11	dl	1.1
12			/**
13	dl	1.14	* Base class for {@link Cancellable} {@link java.lang.Runnable}
14			* actions within the {@link Executor} framework. In addition to
15			* serving as a standalone class, this provides <tt>protected</tt>
16			* functionality that may be useful when creating customized task
17			* classes.
18	dl	1.5	* @since 1.5
19			* @author Doug Lea
20	dl	1.1	*/
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	jozart	1.22	runnerUpdater =
44			AtomicReferenceFieldUpdater.newUpdater(
45			CancellableTask.class,
46			Object.class,
47			"runner");
48	dl	1.1
49			/**
50			* The runnable underlying this task
51			*/
52			private volatile Runnable runnable;
53
54			/**
55	brian	1.4	* Creates a new CancellableTask which invokes the given
56			* <tt>Runnable</tt> when executed.
57	dl	1.5	* @param r the runnable action
58	dl	1.16	* @throws NullPointerException if runnable is null
59	dl	1.1	*/
60			public CancellableTask(Runnable r) {
61	dl	1.16	if (r == null)
62			throw new NullPointerException();
63	dl	1.1	this.runnable = r;
64			}
65
66			/**
67			* Creates a new CancellableTask without a runnable action, which
68			* must be set using <tt>setRunnable</tt> before use. This is
69			* intended for use in subclasses that must complete superclass
70	dl	1.14	* construction before establishing the runnable action.
71	dl	1.1	*/
72			protected CancellableTask() {
73			}
74
75			public boolean cancel(boolean mayInterruptIfRunning) {
76			Object r = runner;
77			if (r == DONE \|\| r == CANCELLED)
78			return false;
79
80			if (mayInterruptIfRunning &&
81			r != null &&
82			r instanceof Thread &&
83	dl	1.3	runnerUpdater.compareAndSet(this, r, CANCELLED))
84	dl	1.1
85			((Thread)r).interrupt();
86	dl	1.3	else
87	dl	1.1	runnerUpdater.set(this, CANCELLED);
88	dl	1.21
89			done();
90	dl	1.1	return true;
91			}
92
93			public boolean isCancelled() {
94			return runner == CANCELLED;
95			}
96
97			public boolean isDone() {
98			Object r = runner;
99			return r == DONE \|\| r == CANCELLED;
100			}
101
102			/**
103	jozart	1.22	* Returns the Runnable forming the basis of this task.
104	dl	1.5	* @return the runnable action
105	tim	1.10	* @see #setRunnable
106	dl	1.1	*/
107			protected Runnable getRunnable() {
108			return runnable;
109			}
110
111			/**
112	jozart	1.22	* Sets the Runnable forming the basis of this task.
113	dl	1.5	* @param r the runnable action
114	tim	1.10	* @see #getRunnable
115	dl	1.1	*/
116			protected void setRunnable(Runnable r) {
117			runnable = r;
118			}
119
120			/**
121	jozart	1.22	* Sets the state of this task to Cancelled.
122	dl	1.1	*/
123			protected void setCancelled() {
124			runnerUpdater.set(this, CANCELLED);
125			}
126
127			/**
128	jozart	1.22	* Sets the state of this task to Done, unless already in a
129			* Cancelled state, in which case Cancelled status is preserved.
130	dl	1.1	*/
131			protected void setDone() {
132			for (;;) {
133			Object r = runner;
134			if (r == DONE \|\| r == CANCELLED)
135			return;
136	dl	1.21	if (runnerUpdater.compareAndSet(this, r, DONE)) {
137			done();
138	dl	1.1	return;
139	dl	1.21	}
140	dl	1.1	}
141			}
142
143			/**
144	jozart	1.22	* Attempts to set the state of this task to Running, succeeding
145	dl	1.1	* only if the state is currently NOT Done, Running, or Cancelled.
146	dl	1.2	* @return true if successful
147	dl	1.1	*/
148			protected boolean setRunning() {
149			return runnerUpdater.compareAndSet(this, null, Thread.currentThread());
150			}
151
152	dl	1.17	/**
153			* Runs the runnable if not cancelled, maintaining run status.
154			*/
155	dl	1.1	public void run() {
156			if (setRunning()) {
157			try {
158			runnable.run();
159	tim	1.11	} finally {
160	dl	1.1	setDone();
161			}
162			}
163			}
164	dl	1.21
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	dl	1.1
176			/**
177	jozart	1.22	* Resets the run state of this task to its initial state unless
178	dl	1.16	* it has been cancelled. (Note that a cancelled task cannot be
179			* reset.)
180			* @return true if successful
181	dl	1.15	*/
182	dl	1.16	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	dl	1.15	}
191
192			/**
193	dl	1.1	* Implementation of Future methods under the control of a current
194	dl	1.13	* <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	jozart	1.23	* class will delegate <tt>get</tt> methods to the
199	dl	1.13	* <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	dl	1.19	* class MyFutureTask<V> extends CancellableTask implements Future<V> {
205	dl	1.13	*
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	dl	1.1	*/
225			protected class InnerCancellableFuture<V> implements Future<V>, Runnable {
226			private final Callable<V> callable;
227			private final ReentrantLock lock = new ReentrantLock();
228	dl	1.20	private final ReentrantLock.ConditionObject accessible = lock.newCondition();
229	dl	1.1	private V result;
230			private Throwable exception;
231	dl	1.5
232			/**
233	jozart	1.22	* Creates an InnerCancellableFuture that will execute the
234	dl	1.5	* given callable.
235			* @param callable the function to execute
236			*/
237	dl	1.1	protected InnerCancellableFuture(Callable<V> callable) {
238			this.callable = callable;
239			}
240	tim	1.12
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	dl	1.1
250			public boolean isDone() {
251			return CancellableTask.this.isDone();
252			}
253
254	dl	1.13
255			/**
256			* Sets this Future to the results of <tt>callable.call</tt>
257			*/
258	dl	1.1	public void run() {
259			try {
260			set(callable.call());
261	tim	1.11	} catch(Throwable ex) {
262	dl	1.1	setException(ex);
263			}
264			}
265
266	dl	1.13	/**
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	jozart	1.23	* @throws CancellationException if underlying computation was
272			* cancelled
273	dl	1.13	* @throws ExecutionException if underlying computation threw an
274			* exception
275			* @throws InterruptedException if current thread was interrupted
276			* while waiting
277			*/
278	dl	1.1	public V get() throws InterruptedException, ExecutionException {
279			lock.lock();
280			try {
281			while (!isDone())
282			accessible.await();
283			if (isCancelled())
284			throw new CancellationException();
285			else if (exception != null)
286			throw new ExecutionException(exception);
287			else
288			return result;
289	tim	1.11	} finally {
290	dl	1.1	lock.unlock();
291			}
292			}
293
294	dl	1.13	/**
295			* Waits if necessary for at most the given time for the call to
296			* <tt>callable.call</tt> to complete, and then retrieves its
297			* result.
298			*
299			* @param timeout the maximum time to wait
300	dl	1.14	* @param unit the time unit of the timeout argument
301	dl	1.13	* @return computed result
302	jozart	1.24	* @throws CancellationException if underlying computation was
303			* cancelled
304	dl	1.13	* @throws ExecutionException if underlying computation threw an
305			* exception
306			* @throws InterruptedException if current thread was interrupted
307			* while waiting
308			* @throws TimeoutException if the wait timed out
309			*/
310	dl	1.1	public V get(long timeout, TimeUnit unit)
311			throws InterruptedException, ExecutionException, TimeoutException {
312			lock.lock();
313			try {
314			if (!isDone()) {
315			long nanos = unit.toNanos(timeout);
316			do {
317			if (nanos <= 0)
318			throw new TimeoutException();
319			nanos = accessible.awaitNanos(nanos);
320			} while (!isDone());
321			}
322			if (isCancelled())
323			throw new CancellationException();
324			else if (exception != null)
325			throw new ExecutionException(exception);
326			else
327			return result;
328	tim	1.11	} finally {
329	dl	1.1	lock.unlock();
330			}
331			}
332
333	dl	1.13	/**
334			* Sets the result of this Future to the given value.
335			* @param v the value
336			*/
337	dl	1.1	protected void set(V v) {
338			lock.lock();
339			try {
340			result = v;
341			setDone();
342			accessible.signalAll();
343	tim	1.11	} finally {
344	dl	1.1	lock.unlock();
345			}
346			}
347
348	dl	1.13	/**
349	dl	1.18	* Causes this future to report an <tt>ExecutionException</tt>
350	dl	1.13	* with the given throwable as its cause.
351			* @param t the cause of failure.
352			*/
353	dl	1.1	protected void setException(Throwable t) {
354			lock.lock();
355			try {
356			exception = t;
357			setDone();
358			accessible.signalAll();
359	tim	1.11	} finally {
360	dl	1.1	lock.unlock();
361			}
362			}
363			}
364
365			}