jsr166y/forkjoin/LinkedAsyncAction.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 *
 */

package jsr166y.forkjoin;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

/**
 * Resultless ForkJoinTasks with explicit completions, that may be
 * linked in parent-child relationships.  Unlike other kinds of tasks,
 * LinkedAsyncActions do not intrinisically complete upon exit from
 * their <tt>compute</tt> methods, but instead require explicit
 * invocation of their <tt>finish</tt> methods and completion of
 * subtasks.
 *
 * <p> Upon construction, an LinkedAsyncAction may register as a
 * subtask of a given parent task. In this case, completion of this
 * task will propagate to its parent. If the parent's pending subtask
 * completion count becomes zero, it too will finish.
 * LinkedAsyncActions rarely use methods <tt>join</tt> or
 * <tt>invoke</tt> but instead propagate completion to parents
 * implicitly via <tt>finish</tt>.  While typical, it is not necessary
 * for each task to <tt>finish</tt> itself. For example, it is
 * possible to treat one subtask as a continuation of the current task
 * by not registering it on construction.  In this case, a
 * <tt>finish</tt> of the subtask will trigger <tt>finish</tt> of the
 * parent without the parent explicitly doing so.
 *
 * <p> In addition to supporting these different computation styles
 * compared to Recursive tasks, LinkedAsyncActions may have smaller
 * stack space footprints while executing, but may have greater
 * per-task overhead.
 *
 * <p> <b>Sample Usage.</b> Here is a sketch of an LinkedAsyncAction
 * that visits all of the nodes of a graph. The details of the graph's
 * Node and Edge classes are omitted, but we assume each node contains
 * an <tt>AtomicBoolean</tt> mark that starts out false. To execute
 * this, you would create a GraphVisitor for the root node with null
 * parent, and <tt>invoke</tt> in a ForkJoinPool. Upon return, all
 * reachable nodes will have been visited.
 * 
 * <pre>
 * class GraphVisitor extends LinkedAsyncAction {
 *    final Node node;
 *    GraphVisitor(GraphVistor parent, Node node) {
 *      super(parent); this.node = node;
 *    }
 *    protected void compute() {
 *      if (node.mark.compareAndSet(false, true)) {
 *         for (Edge e : node.edges()) {
 *            Node dest = e.getDestination();
 *            if (!dest.mark.get())
 *               new GraphVisitor(this, dest).fork();
 *         }
 *         visit(node);
 *      }
 *      finish();
 *   }
 * }
 * </pre>
 *
 */
public abstract class LinkedAsyncAction extends ForkJoinTask<Void> {
    /**
     * Parent to notify on completion
     */
    private LinkedAsyncAction parent;

    /**
     * Count of outstanding subtask joins
     */
    private volatile int pendingCount;

    private static final AtomicIntegerFieldUpdater<LinkedAsyncAction> pendingCountUpdater =
        AtomicIntegerFieldUpdater.newUpdater(LinkedAsyncAction.class, "pendingCount");

    /**
     * Creates a new action with no parent. (You can add a parent
     * later (but before forking) via <tt>reinitialize</tt>).
     */
    protected LinkedAsyncAction() {
    }

    /**
     * Creates a new action with the given parent. If the parent is
     * non-null, this tasks registers with the parent, in which case,
     * the parent task cannot complete until this task completes.
     * @param parent the parent task, or null if none
     */
    protected LinkedAsyncAction(LinkedAsyncAction parent) {
        this.parent = parent;
        if (parent != null) 
            pendingCountUpdater.incrementAndGet(parent);
    }

    /**
     * Creates a new action with the given parent, optionally
     * registering with the parent. If the parent is non-null and
     * <tt>register</tt> is true, this tasks registers with the
     * parent, in which case, the parent task cannot complete until
     * this task completes.
     * @param parent the parent task, or null if none
     * @param register true if parent must wait for this task
     * to complete before it completes
     */
    protected LinkedAsyncAction(LinkedAsyncAction parent, boolean register) {
        this.parent = parent;
        if (parent != null && register) 
            pendingCountUpdater.incrementAndGet(parent);
    }

    /**
     * Creates a new action with the given parent, optionally
     * registering with the parent, and setting the pending join count
     * to the given value. If the parent is non-null and
     * <tt>register</tt> is true, this tasks registers with the
     * parent, in which case, the parent task cannot complete until
     * this task completes. Setting the pending join count requires
     * care -- it is correct only if child tasks do not themselves
     * register.
     * @param parent the parent task, or null if none
     * @param register true if parent must wait for this task
     * to complete before it completes
     * @param pending the pending join count
     */
    protected LinkedAsyncAction(LinkedAsyncAction parent, 
                          boolean register,
                          int pending) {
        this.parent = parent;
        pendingCount = pending;
        if (parent != null && register) 
            pendingCountUpdater.incrementAndGet(parent);
    }

    /**
     * The asynchronous part of the computation performed by this
     * task.  While you must define this method, you should not in
     * general call it directly (although you can invoke immediately
     * via <tt>exec</tt>.) If this method throws a Throwable,
     * <tt>finishExceptionally</tt> is immediately invoked.
     */
    protected abstract void compute();

    /**
     * Overridable callback action triggered by <tt>finish</tt>.  Upon
     * invocation, all subtasks have completed.  After return, this
     * task <tt>isDone</tt> and is joinable by other tasks. The
     * default version of this method does nothing. But it may may be
     * overridden in subclasses to perform some action when this task
     * is about to complete.
     */
    protected void onCompletion() {
    }

    /**
     * Overridable callback action triggered by
     * <tt>finishExceptionally</tt>.  Upon invocation, this task has
     * aborted due to an exception (accessible via
     * <tt>getException</tt>). If this method returns <tt>true</tt>,
     * the exception propagates to the current task's
     * parent. Otherwise, normal completion is propagated.  The
     * default version of this method does nothing and returns
     * <tt>true</tt>.
     * @return true if this task's exception should be propagated to
     * this tasks parent.
     */
    protected boolean onException() {
        return true;
    }

    /**
     * Equivalent to <tt>finish(null)</tt>.
     */
    public final void finish() {
        finish(null);
    }

    /**
     * Completes this task. If the pending subtask completion count is
     * zero, invokes <tt>onCompletion</tt>, then causes this task to
     * be joinable (<tt>isDone</tt> becomes true), and then
     * recursively applies to this tasks's parent, if it exists. If an
     * exception is encountered in any <tt>onCompletion</tt>
     * invocation, that task and its ancestors
     * <tt>finishExceptionally</tt>.
     * 
     * @param result must be <tt>null</tt>.
     */
    public final void finish(Void result) {
        LinkedAsyncAction a = this;
        while (a != null && !a.isDone()) {
            int c = a.pendingCount;
            if (c == 0) {
                try {
                    a.onCompletion();
                } catch (Throwable rex) {
                    a.finishExceptionally(rex);
                    return;
                } 
                a.setDone();
                a = a.parent;
            }
            else if (pendingCountUpdater.compareAndSet(a, c, c-1))
                return;
        }
    }

    /**
     * Completes this task abnormally. Unless this task already
     * cancelled or aborted, upon invocation, this method invokes
     * <tt>onException</tt>, and then, depending on its return value,
     * finishes parent (if one exists) exceptionally or normally.  To
     * avoid unbounded exception loops, this method aborts if an
     * exception is encountered in any <tt>onException</tt>
     * invocation.
     * @param ex the exception to throw when joining this task
     * @throws NullPointerException if ex is null
     * @throws Throwable if any invocation of
     * <tt>onException</tt> does so.
     */
    public final void finishExceptionally(Throwable ex) {
        if (ex == null)
            throw new NullPointerException();
        LinkedAsyncAction a = this;
        while (a.setDoneExceptionally(ex) == ex) {
            boolean up = a.onException(); // abort if this throws
            a = a.parent;
            if (a == null)
                return;
            if (!up) {
                a.finish();
                return;
            }
        }
    }

    /**
     * Returns this task's parent, or null if none.
     * @return this task's parent, or null if none.
     */
    public final LinkedAsyncAction getParent() { 
        return parent; 
    }

    /**
     * Returns the number of subtasks that have not yet completed.
     * @return the number of subtasks that have not yet completed.
     */
    public final int getPendingSubtaskCount() { 
        return pendingCount; 
    }

    /**
     * Always returns null.
     * @return null
     */
    public final Void rawResult() { 
        return null; 
    }

    /**
     * Resets the internal bookkeeping state of this task, maintaining
     * the current parent but clearing pending joins.
     */
    public void reinitialize() {
        super.reinitialize();
        if (pendingCount != 0)
            pendingCount = 0;
    }

    /**
     * Resets the internal bookkeeping state of this task, maintaining
     * the current parent and setting pending joins to the given value.
     * @param pending the number of pending joins
     */
    public void reinitialize(int pending) {
        super.reinitialize();
        pendingCount = pending;
    }

    /**
     * Reinitialize with the given parent, optionally registering.
     * @param parent the parent task, or null if none
     * @param register true if parent must wait for this task
     * to complete before it completes
     */
    public void reinitialize(LinkedAsyncAction parent, boolean register) {
        super.reinitialize();
        if (pendingCount != 0)
            pendingCount = 0;
        this.parent = parent;
        if (parent != null && register) 
            pendingCountUpdater.incrementAndGet(parent);
    }

    /**
     * Reinitialize with the given parent, optionally registering
     * and setting pending join count.
     * @param parent the parent task, or null if none
     * @param register true if parent must wait for this task
     * to complete before it completes
     * @param pending the pending join count
     */
    public void reinitialize(LinkedAsyncAction parent, 
                             boolean register,
                             int pending) {
        super.reinitialize();
        pendingCount = pending;
        this.parent = parent;
        if (parent != null && register) 
            pendingCountUpdater.incrementAndGet(parent);
    }


    public final Void forkJoin() {
        exec();
        return join();
    }

    public final Throwable exec() {
        if (exception == null) {
            try {
                compute();
            } catch(Throwable rex) {
                finishExceptionally(rex);
            }
        }
        return exception;
    }

}
Revision:	1.4
Committed:	Tue Dec 18 21:38:32 2007 UTC (16 years, 6 months ago) by dl
Branch:	MAIN
Changes since 1.3:	+1 -1 lines
Log Message:	rename invoke, coInvoke => forkJoin
#	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, as explained at
4			*
5			*/
6
7			package jsr166y.forkjoin;
8			import java.util.*;
9			import java.util.concurrent.*;
10			import java.util.concurrent.atomic.*;
11
12			/**
13			* Resultless ForkJoinTasks with explicit completions, that may be
14			* linked in parent-child relationships. Unlike other kinds of tasks,
15			* LinkedAsyncActions do not intrinisically complete upon exit from
16			* their <tt>compute</tt> methods, but instead require explicit
17			* invocation of their <tt>finish</tt> methods and completion of
18			* subtasks.
19			*
20			* <p> Upon construction, an LinkedAsyncAction may register as a
21			* subtask of a given parent task. In this case, completion of this
22			* task will propagate to its parent. If the parent's pending subtask
23			* completion count becomes zero, it too will finish.
24			* LinkedAsyncActions rarely use methods <tt>join</tt> or
25			* <tt>invoke</tt> but instead propagate completion to parents
26			* implicitly via <tt>finish</tt>. While typical, it is not necessary
27			* for each task to <tt>finish</tt> itself. For example, it is
28			* possible to treat one subtask as a continuation of the current task
29			* by not registering it on construction. In this case, a
30			* <tt>finish</tt> of the subtask will trigger <tt>finish</tt> of the
31			* parent without the parent explicitly doing so.
32			*
33			* <p> In addition to supporting these different computation styles
34			* compared to Recursive tasks, LinkedAsyncActions may have smaller
35			* stack space footprints while executing, but may have greater
36			* per-task overhead.
37			*
38			* <p> <b>Sample Usage.</b> Here is a sketch of an LinkedAsyncAction
39			* that visits all of the nodes of a graph. The details of the graph's
40			* Node and Edge classes are omitted, but we assume each node contains
41			* an <tt>AtomicBoolean</tt> mark that starts out false. To execute
42			* this, you would create a GraphVisitor for the root node with null
43			* parent, and <tt>invoke</tt> in a ForkJoinPool. Upon return, all
44			* reachable nodes will have been visited.
45			*
46			* <pre>
47			* class GraphVisitor extends LinkedAsyncAction {
48			* final Node node;
49			* GraphVisitor(GraphVistor parent, Node node) {
50			* super(parent); this.node = node;
51			* }
52			* protected void compute() {
53			* if (node.mark.compareAndSet(false, true)) {
54			* for (Edge e : node.edges()) {
55			* Node dest = e.getDestination();
56			* if (!dest.mark.get())
57			* new GraphVisitor(this, dest).fork();
58			* }
59			* visit(node);
60			* }
61			* finish();
62			* }
63			* }
64			* </pre>
65			*
66			*/
67			public abstract class LinkedAsyncAction extends ForkJoinTask<Void> {
68			/**
69			* Parent to notify on completion
70			*/
71			private LinkedAsyncAction parent;
72
73			/**
74			* Count of outstanding subtask joins
75			*/
76			private volatile int pendingCount;
77
78			private static final AtomicIntegerFieldUpdater<LinkedAsyncAction> pendingCountUpdater =
79			AtomicIntegerFieldUpdater.newUpdater(LinkedAsyncAction.class, "pendingCount");
80
81			/**
82			* Creates a new action with no parent. (You can add a parent
83			* later (but before forking) via <tt>reinitialize</tt>).
84			*/
85			protected LinkedAsyncAction() {
86			}
87
88			/**
89			* Creates a new action with the given parent. If the parent is
90			* non-null, this tasks registers with the parent, in which case,
91			* the parent task cannot complete until this task completes.
92			* @param parent the parent task, or null if none
93			*/
94			protected LinkedAsyncAction(LinkedAsyncAction parent) {
95			this.parent = parent;
96			if (parent != null)
97			pendingCountUpdater.incrementAndGet(parent);
98			}
99
100			/**
101			* Creates a new action with the given parent, optionally
102			* registering with the parent. If the parent is non-null and
103			* <tt>register</tt> is true, this tasks registers with the
104			* parent, in which case, the parent task cannot complete until
105			* this task completes.
106			* @param parent the parent task, or null if none
107			* @param register true if parent must wait for this task
108			* to complete before it completes
109			*/
110			protected LinkedAsyncAction(LinkedAsyncAction parent, boolean register) {
111			this.parent = parent;
112			if (parent != null && register)
113			pendingCountUpdater.incrementAndGet(parent);
114			}
115
116			/**
117			* Creates a new action with the given parent, optionally
118			* registering with the parent, and setting the pending join count
119			* to the given value. If the parent is non-null and
120			* <tt>register</tt> is true, this tasks registers with the
121			* parent, in which case, the parent task cannot complete until
122			* this task completes. Setting the pending join count requires
123			* care -- it is correct only if child tasks do not themselves
124			* register.
125			* @param parent the parent task, or null if none
126			* @param register true if parent must wait for this task
127			* to complete before it completes
128			* @param pending the pending join count
129			*/
130			protected LinkedAsyncAction(LinkedAsyncAction parent,
131			boolean register,
132			int pending) {
133			this.parent = parent;
134			pendingCount = pending;
135			if (parent != null && register)
136			pendingCountUpdater.incrementAndGet(parent);
137			}
138
139			/**
140			* The asynchronous part of the computation performed by this
141			* task. While you must define this method, you should not in
142			* general call it directly (although you can invoke immediately
143	dl	1.2	* via <tt>exec</tt>.) If this method throws a Throwable,
144	dl	1.1	* <tt>finishExceptionally</tt> is immediately invoked.
145			*/
146			protected abstract void compute();
147
148			/**
149			* Overridable callback action triggered by <tt>finish</tt>. Upon
150			* invocation, all subtasks have completed. After return, this
151			* task <tt>isDone</tt> and is joinable by other tasks. The
152			* default version of this method does nothing. But it may may be
153			* overridden in subclasses to perform some action when this task
154			* is about to complete.
155			*/
156			protected void onCompletion() {
157			}
158
159			/**
160			* Overridable callback action triggered by
161			* <tt>finishExceptionally</tt>. Upon invocation, this task has
162			* aborted due to an exception (accessible via
163			* <tt>getException</tt>). If this method returns <tt>true</tt>,
164			* the exception propagates to the current task's
165			* parent. Otherwise, normal completion is propagated. The
166			* default version of this method does nothing and returns
167			* <tt>true</tt>.
168			* @return true if this task's exception should be propagated to
169			* this tasks parent.
170			*/
171			protected boolean onException() {
172			return true;
173			}
174
175			/**
176			* Equivalent to <tt>finish(null)</tt>.
177			*/
178			public final void finish() {
179			finish(null);
180			}
181
182			/**
183			* Completes this task. If the pending subtask completion count is
184			* zero, invokes <tt>onCompletion</tt>, then causes this task to
185			* be joinable (<tt>isDone</tt> becomes true), and then
186			* recursively applies to this tasks's parent, if it exists. If an
187			* exception is encountered in any <tt>onCompletion</tt>
188			* invocation, that task and its ancestors
189			* <tt>finishExceptionally</tt>.
190			*
191			* @param result must be <tt>null</tt>.
192			*/
193			public final void finish(Void result) {
194			LinkedAsyncAction a = this;
195			while (a != null && !a.isDone()) {
196			int c = a.pendingCount;
197			if (c == 0) {
198			try {
199			a.onCompletion();
200	dl	1.2	} catch (Throwable rex) {
201	dl	1.1	a.finishExceptionally(rex);
202			return;
203			}
204			a.setDone();
205			a = a.parent;
206			}
207			else if (pendingCountUpdater.compareAndSet(a, c, c-1))
208			return;
209			}
210			}
211
212			/**
213			* Completes this task abnormally. Unless this task already
214			* cancelled or aborted, upon invocation, this method invokes
215			* <tt>onException</tt>, and then, depending on its return value,
216			* finishes parent (if one exists) exceptionally or normally. To
217			* avoid unbounded exception loops, this method aborts if an
218			* exception is encountered in any <tt>onException</tt>
219			* invocation.
220			* @param ex the exception to throw when joining this task
221			* @throws NullPointerException if ex is null
222	dl	1.2	* @throws Throwable if any invocation of
223	dl	1.1	* <tt>onException</tt> does so.
224			*/
225	dl	1.2	public final void finishExceptionally(Throwable ex) {
226	dl	1.1	if (ex == null)
227			throw new NullPointerException();
228			LinkedAsyncAction a = this;
229	dl	1.2	while (a.setDoneExceptionally(ex) == ex) {
230	dl	1.1	boolean up = a.onException(); // abort if this throws
231			a = a.parent;
232			if (a == null)
233			return;
234			if (!up) {
235			a.finish();
236			return;
237			}
238			}
239			}
240
241			/**
242			* Returns this task's parent, or null if none.
243			* @return this task's parent, or null if none.
244			*/
245			public final LinkedAsyncAction getParent() {
246			return parent;
247			}
248
249			/**
250			* Returns the number of subtasks that have not yet completed.
251			* @return the number of subtasks that have not yet completed.
252			*/
253			public final int getPendingSubtaskCount() {
254			return pendingCount;
255			}
256
257			/**
258			* Always returns null.
259			* @return null
260			*/
261	dl	1.3	public final Void rawResult() {
262	dl	1.1	return null;
263			}
264
265			/**
266			* Resets the internal bookkeeping state of this task, maintaining
267			* the current parent but clearing pending joins.
268			*/
269			public void reinitialize() {
270			super.reinitialize();
271			if (pendingCount != 0)
272			pendingCount = 0;
273			}
274
275			/**
276			* Resets the internal bookkeeping state of this task, maintaining
277			* the current parent and setting pending joins to the given value.
278			* @param pending the number of pending joins
279			*/
280			public void reinitialize(int pending) {
281			super.reinitialize();
282			pendingCount = pending;
283			}
284
285			/**
286			* Reinitialize with the given parent, optionally registering.
287			* @param parent the parent task, or null if none
288			* @param register true if parent must wait for this task
289			* to complete before it completes
290			*/
291			public void reinitialize(LinkedAsyncAction parent, boolean register) {
292			super.reinitialize();
293			if (pendingCount != 0)
294			pendingCount = 0;
295			this.parent = parent;
296			if (parent != null && register)
297			pendingCountUpdater.incrementAndGet(parent);
298			}
299
300			/**
301			* Reinitialize with the given parent, optionally registering
302			* and setting pending join count.
303			* @param parent the parent task, or null if none
304			* @param register true if parent must wait for this task
305			* to complete before it completes
306			* @param pending the pending join count
307			*/
308			public void reinitialize(LinkedAsyncAction parent,
309			boolean register,
310			int pending) {
311			super.reinitialize();
312			pendingCount = pending;
313			this.parent = parent;
314			if (parent != null && register)
315			pendingCountUpdater.incrementAndGet(parent);
316			}
317
318
319	dl	1.4	public final Void forkJoin() {
320	dl	1.1	exec();
321			return join();
322			}
323
324	dl	1.2	public final Throwable exec() {
325	dl	1.1	if (exception == null) {
326			try {
327			compute();
328	dl	1.2	} catch(Throwable rex) {
329	dl	1.1	finishExceptionally(rex);
330			}
331			}
332			return exception;
333			}
334
335			}