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