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.*;

/**
 * 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 = new AtomicReferenceFieldUpdater<CancellableTask, Object>(new CancellableTask[0], new Object[0], "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 &&
            r != Thread.currentThread() &&
            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
     */
    protected Runnable getRunnable() {
        return runnable;
    }

    /**
     * Set the Runnable forming the basis of this task.
     * @param r the runnable action
     */
    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
     * CancellableTask. This is split into an inner class to permit
     * Future support to be mixed-in with other flavors of tasks.
     */
    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 isDone() {
            return CancellableTask.this.isDone();
        }

        public void run() {
            try {
                set(callable.call());
            }
            catch(Throwable ex) {
                setException(ex);
            }
        }

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

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

        protected void set(V v) {
            lock.lock();
            try {
                result = v;
                setDone();
                accessible.signalAll();
            }
            finally {
                lock.unlock();
            }
        }

        protected void setException(Throwable t) {
            lock.lock();
            try {
                exception = t;
                setDone();
                accessible.signalAll();
            }
            finally {
                lock.unlock();
            }
        }
    }
   
}
Revision:	1.5
Committed:	Tue Jun 24 14:34:47 2003 UTC (20 years, 11 months ago) by dl
Branch:	MAIN
Changes since 1.4:	+13 -2 lines
Log Message:	Added missing javadoc tags; minor reformatting
#	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
10			/**
11			* Base class for cancellable actions running in the Executor
12			* framework. In addition to serving as a standalone class, this
13			* provides <tt>protected</tt> functionality that may be useful when
14			* creating customized task classes.
15	dl	1.5	* @since 1.5
16			* @author Doug Lea
17	dl	1.1	*/
18
19			public class CancellableTask implements Cancellable, Runnable {
20			/**
21			* Holds the run-state, taking on values:
22			* null = not yet started,
23			* [some thread ref] = running,
24			* DONE = completed normally,
25			* CANCELLED = cancelled (may or may not have ever run).
26			* Transitions use atomic updates.
27			*/
28			private volatile Object runner;
29
30	dl	1.5	/**
31	dl	1.1	* Special value for "runner" indicating task is completed
32			*/
33			private static final Object DONE = new Object();
34
35	dl	1.5	/**
36	dl	1.1	* Special value for "runner" indicating task is cancelled
37			*/
38			private static final Object CANCELLED = new Object();
39
40			private static AtomicReferenceFieldUpdater<CancellableTask, Object>
41			runnerUpdater = new AtomicReferenceFieldUpdater<CancellableTask, Object>(new CancellableTask[0], new Object[0], "runner");
42
43			/**
44			* The runnable underlying this task
45			*/
46			private volatile Runnable runnable;
47
48			/**
49	brian	1.4	* Creates a new CancellableTask which invokes the given
50			* <tt>Runnable</tt> when executed.
51	dl	1.5	* @param r the runnable action
52	dl	1.1	*/
53			public CancellableTask(Runnable r) {
54			this.runnable = r;
55			}
56
57			/**
58			* Creates a new CancellableTask without a runnable action, which
59			* must be set using <tt>setRunnable</tt> before use. This is
60			* intended for use in subclasses that must complete superclass
61			* construction beofre establishing the runnable action.
62			*/
63			protected CancellableTask() {
64			}
65
66
67			public boolean cancel(boolean mayInterruptIfRunning) {
68			Object r = runner;
69			if (r == DONE \|\| r == CANCELLED)
70			return false;
71
72			if (mayInterruptIfRunning &&
73			r != null &&
74			r instanceof Thread &&
75			r != Thread.currentThread() &&
76	dl	1.3	runnerUpdater.compareAndSet(this, r, CANCELLED))
77	dl	1.1
78			((Thread)r).interrupt();
79	dl	1.3	else
80	dl	1.1	runnerUpdater.set(this, CANCELLED);
81			return true;
82			}
83
84			public boolean isCancelled() {
85			return runner == CANCELLED;
86			}
87
88			public boolean isDone() {
89			Object r = runner;
90			return r == DONE \|\| r == CANCELLED;
91			}
92
93			/**
94			* Return the Runnable forming the basis of this task.
95	dl	1.5	* @return the runnable action
96	dl	1.1	*/
97			protected Runnable getRunnable() {
98			return runnable;
99			}
100
101			/**
102			* Set the Runnable forming the basis of this task.
103	dl	1.5	* @param r the runnable action
104	dl	1.1	*/
105			protected void setRunnable(Runnable r) {
106			runnable = r;
107			}
108
109			/**
110			* Set the state of this task to Cancelled
111			*/
112			protected void setCancelled() {
113			runnerUpdater.set(this, CANCELLED);
114			}
115
116			/**
117			* Set the state of this task to Done, unless already
118			* in a Cancelled state, in which Cancelled status is preserved.
119			*
120			*/
121			protected void setDone() {
122			for (;;) {
123			Object r = runner;
124			if (r == DONE \|\| r == CANCELLED)
125			return;
126			if (runnerUpdater.compareAndSet(this, r, DONE))
127			return;
128			}
129			}
130
131			/**
132			* Attempt to set the state of this task to Running, succeeding
133			* only if the state is currently NOT Done, Running, or Cancelled.
134	dl	1.2	* @return true if successful
135	dl	1.1	*/
136			protected boolean setRunning() {
137			return runnerUpdater.compareAndSet(this, null, Thread.currentThread());
138			}
139
140			public void run() {
141			if (setRunning()) {
142			try {
143			runnable.run();
144			}
145			finally {
146			setDone();
147			}
148			}
149			}
150
151			/**
152			* Implementation of Future methods under the control of a current
153			* CancellableTask. This is split into an inner class to permit
154			* Future support to be mixed-in with other flavors of tasks.
155			*/
156			protected class InnerCancellableFuture<V> implements Future<V>, Runnable {
157			private final Callable<V> callable;
158			private final ReentrantLock lock = new ReentrantLock();
159			private final Condition accessible = lock.newCondition();
160			private V result;
161			private Throwable exception;
162	dl	1.5
163			/**
164			* Create an InnerCancellableFuture that will execute the
165			* given callable.
166			* @param callable the function to execute
167			*/
168	dl	1.1	protected InnerCancellableFuture(Callable<V> callable) {
169			this.callable = callable;
170			}
171
172			public boolean isDone() {
173			return CancellableTask.this.isDone();
174			}
175
176			public void run() {
177			try {
178			set(callable.call());
179			}
180			catch(Throwable ex) {
181			setException(ex);
182			}
183			}
184
185			public V get() throws InterruptedException, ExecutionException {
186			lock.lock();
187			try {
188			while (!isDone())
189			accessible.await();
190			if (isCancelled())
191			throw new CancellationException();
192			else if (exception != null)
193			throw new ExecutionException(exception);
194			else
195			return result;
196			}
197			finally {
198			lock.unlock();
199			}
200			}
201
202			public V get(long timeout, TimeUnit unit)
203			throws InterruptedException, ExecutionException, TimeoutException {
204			lock.lock();
205			try {
206			if (!isDone()) {
207			long nanos = unit.toNanos(timeout);
208			do {
209			if (nanos <= 0)
210			throw new TimeoutException();
211			nanos = accessible.awaitNanos(nanos);
212			} while (!isDone());
213			}
214			if (isCancelled())
215			throw new CancellationException();
216			else if (exception != null)
217			throw new ExecutionException(exception);
218			else
219			return result;
220			}
221			finally {
222			lock.unlock();
223			}
224			}
225
226			protected void set(V v) {
227			lock.lock();
228			try {
229			result = v;
230			setDone();
231			accessible.signalAll();
232			}
233			finally {
234			lock.unlock();
235			}
236			}
237
238			protected void setException(Throwable t) {
239			lock.lock();
240			try {
241			exception = t;
242			setDone();
243			accessible.signalAll();
244			}
245			finally {
246			lock.unlock();
247			}
248			}
249			}
250
251			}