test/loops/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
 *
 */

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

/**
 * AsyncActions that may be linked in parent-child relationships.
 *
 * <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 complete.
 * LinkedAsyncActions rarely use methods <tt>join</tt> or
 * <tt>invoke</tt> but instead propagate completion to parents
 * implicitly via <tt>complete</tt>.  While typical, it is not necessary
 * for each task to <tt>complete</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>complete</tt> of the subtask will trigger <tt>complete</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);
 *      }
 *      complete();
 *   }
 * }
 * </pre>
 *
 */
public abstract class LinkedAsyncAction extends ForkJoinTask<Void> {

    /**
     * The maximum number of pending children
     */
    private static int maxPending = 0xffff;

    private volatile int controlState;

    static final AtomicIntegerFieldUpdater<LinkedAsyncAction> controlStateUpdater =
        AtomicIntegerFieldUpdater.newUpdater(LinkedAsyncAction.class, "controlState");


    /**
     * Parent to notify on completion
     */
    private LinkedAsyncAction parent;

    /**
     * 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)
            parent.incrementControlState();
    }

    /**
     * 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)
            parent.incrementControlState();
    }

    /**
     * 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;
        setControlState(pending);
        if (parent != null && register)
            parent.incrementControlState();
    }

    public final Void getRawResult() { return null; }
    protected final void setRawResult(Void mustBeNull) { }

    /**
     * Overridable callback action triggered by <tt>complete</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>completeExceptionally</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;
    }

    /** Basic per-task complete */
    private void completeThis() {
        complete(null);
    }

    /** Basic per-task completeExceptionally */
    private void completeThisExceptionally(Throwable ex) {
        super.completeExceptionally(ex);
    }

    /**
     * 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>completeExceptionally</tt>.
     */
    public final void complete() {
        LinkedAsyncAction a = this;
        while (a != null && !a.isDone()) {
            int c = a.getControlState();
            if (c <= 0) {
                try {
                    a.onCompletion();
                    a.completeThis();
                } catch (Throwable rex) {
                    a.completeExceptionally(rex);
                    return;
                }
                a = a.parent;
            }
            else if (a.compareAndSetControlState(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,
     * completees 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 completeExceptionally(Throwable ex) {
        LinkedAsyncAction a = this;
        while (!a.isCompletedAbnormally()) {
            a.completeThisExceptionally(ex);
            boolean up = a.onException(); // abort if this throws
            a = a.parent;
            if (a == null)
                break;
            if (!up) {
                a.complete();
                break;
            }
        }
    }

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

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

    /**
     * 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();
        setControlState(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) {
        this.parent = parent;
        super.reinitialize();
        if (parent != null && register)
            parent.incrementControlState();
    }

    /**
     * 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) {
        this.parent = parent;
        super.reinitialize();
        setControlState(pending);
        if (parent != null && register)
            parent.incrementControlState();
    }

    /**
     * Gets the control state, which is initially zero, or negative if
     * this task has completed or cancelled. Once negative, the value
     * cannot be changed.
     * @return control state
     */
    protected final int getControlState() {
        return controlState;
    }

    /**
     * Atomically sets the control state to the given updated value if
     * the current value is and equal to the expected value.
     * @param expect the expected value
     * @param update the new value
     * @return true if successful
     */
    protected final boolean compareAndSetControlState(int expect,
                                                      int update) {
        return controlStateUpdater.compareAndSet(this, expect, update);
    }

    /**
     * Sets the control state to the given value,
     * @param value the new value
     */
    protected final void setControlState(int value) {
        controlState = value;
    }

    /**
     * Increment the control state
     * @return true if successful
     */
    protected final void incrementControlState() {
        controlStateUpdater.incrementAndGet(this);
    }

    /**
     * Decrement the control state
     * @return true if successful
     */
    protected final void decrementControlState() {
        controlStateUpdater.decrementAndGet(this);
    }


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