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
 * http://creativecommons.org/publicdomain/zero/1.0/
 */

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

/**
 * AsyncActions that may be linked in parent-child relationships.
 *
 * <p>Upon construction, a 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 {@code join} or
 * {@code invoke} but instead propagate completion to parents
 * implicitly via {@code complete}.  While typical, it is not necessary
 * for each task to {@code complete} 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
 * {@code complete} of the subtask will trigger {@code complete} 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 a 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 {@code AtomicBoolean} mark that starts out false. To execute
 * this, you would create a GraphVisitor for the root node with null
 * parent, and {@code invoke} 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 {@code reinitialize}).
     */
    protected LinkedAsyncAction() {
    }

    /**
     * Creates a new action with the given parent. If the parent is
     * non-null, this task 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
     * {@code register} is true, this task 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
     * {@code register} is true, this task 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 {@code complete}.  Upon
     * invocation, all subtasks have completed.  After return, this
     * task {@code isDone} and is joinable by other tasks. The
     * default version of this method does nothing. But it may be
     * overridden in subclasses to perform some action when this task
     * is about to complete.
     */
    protected void onCompletion() {
    }

    /**
     * Overridable callback action triggered by
     * {@code completeExceptionally}.  Upon invocation, this task has
     * aborted due to an exception (accessible via
     * {@code getException}). If this method returns {@code true},
     * the exception propagates to the current task's
     * parent. Otherwise, normal completion is propagated.  The
     * default version of this method does nothing and returns
     * {@code true}.
     * @return true if this task's exception should be propagated to
     * this task's 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 {@code onCompletion}, then causes this task to
     * be joinable ({@code isDone} becomes true), and then
     * recursively applies to this task's parent, if it exists. If an
     * exception is encountered in any {@code onCompletion}
     * invocation, that task and its ancestors
     * {@code completeExceptionally}.
     */
    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
     * {@code onException}, and then, depending on its return value,
     * completes parent (if one exists) exceptionally or normally.  To
     * avoid unbounded exception loops, this method aborts if an
     * exception is encountered in any {@code onException}
     * invocation.
     * @param ex the exception to throw when joining this task
     * @throws NullPointerException if ex is null
     * @throws Throwable if any invocation of
     * {@code onException} 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;
    }

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

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

}
Revision:	1.12
Committed:	Mon Nov 26 11:47:41 2012 UTC (11 years, 5 months ago) by jsr166
Branch:	MAIN
Changes since 1.11:	+3 -4 lines
Log Message:	javadoc style
#	Content
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	* http://creativecommons.org/publicdomain/zero/1.0/
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, a 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 {@code join} or
19	* {@code invoke} but instead propagate completion to parents
20	* implicitly via {@code complete}. While typical, it is not necessary
21	* for each task to {@code complete} 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	* {@code complete} of the subtask will trigger {@code complete} 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 a 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 {@code AtomicBoolean} mark that starts out false. To execute
36	* this, you would create a GraphVisitor for the root node with null
37	* parent, and {@code invoke} 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	public abstract class LinkedAsyncAction extends ForkJoinTask<Void> {
61
62	/**
63	* The maximum number of pending children
64	*/
65	private static int maxPending = 0xffff;
66
67	private volatile int controlState;
68
69	static final AtomicIntegerFieldUpdater<LinkedAsyncAction> controlStateUpdater =
70	AtomicIntegerFieldUpdater.newUpdater(LinkedAsyncAction.class, "controlState");
71
72
73	/**
74	* Parent to notify on completion
75	*/
76	private LinkedAsyncAction parent;
77
78	/**
79	* Creates a new action with no parent. (You can add a parent
80	* later (but before forking) via {@code reinitialize}).
81	*/
82	protected LinkedAsyncAction() {
83	}
84
85	/**
86	* Creates a new action with the given parent. If the parent is
87	* non-null, this task registers with the parent, in which case,
88	* the parent task cannot complete until this task completes.
89	* @param parent the parent task, or null if none
90	*/
91	protected LinkedAsyncAction(LinkedAsyncAction parent) {
92	this.parent = parent;
93	if (parent != null)
94	parent.incrementControlState();
95	}
96
97	/**
98	* Creates a new action with the given parent, optionally
99	* registering with the parent. If the parent is non-null and
100	* {@code register} is true, this task registers with the
101	* parent, in which case, the parent task cannot complete until
102	* this task completes.
103	* @param parent the parent task, or null if none
104	* @param register true if parent must wait for this task
105	* to complete before it completes
106	*/
107	protected LinkedAsyncAction(LinkedAsyncAction parent, boolean register) {
108	this.parent = parent;
109	if (parent != null && register)
110	parent.incrementControlState();
111	}
112
113	/**
114	* Creates a new action with the given parent, optionally
115	* registering with the parent, and setting the pending join count
116	* to the given value. If the parent is non-null and
117	* {@code register} is true, this task registers with the
118	* parent, in which case, the parent task cannot complete until
119	* this task completes. Setting the pending join count requires
120	* care -- it is correct only if child tasks do not themselves
121	* register.
122	* @param parent the parent task, or null if none
123	* @param register true if parent must wait for this task
124	* to complete before it completes
125	* @param pending the pending join count
126	*/
127	protected LinkedAsyncAction(LinkedAsyncAction parent,
128	boolean register,
129	int pending) {
130	this.parent = parent;
131	setControlState(pending);
132	if (parent != null && register)
133	parent.incrementControlState();
134	}
135
136	public final Void getRawResult() { return null; }
137	protected final void setRawResult(Void mustBeNull) { }
138
139	/**
140	* Overridable callback action triggered by {@code complete}. Upon
141	* invocation, all subtasks have completed. After return, this
142	* task {@code isDone} and is joinable by other tasks. The
143	* default version of this method does nothing. But it may be
144	* overridden in subclasses to perform some action when this task
145	* is about to complete.
146	*/
147	protected void onCompletion() {
148	}
149
150	/**
151	* Overridable callback action triggered by
152	* {@code completeExceptionally}. Upon invocation, this task has
153	* aborted due to an exception (accessible via
154	* {@code getException}). If this method returns {@code true},
155	* the exception propagates to the current task's
156	* parent. Otherwise, normal completion is propagated. The
157	* default version of this method does nothing and returns
158	* {@code true}.
159	* @return true if this task's exception should be propagated to
160	* this task's parent.
161	*/
162	protected boolean onException() {
163	return true;
164	}
165
166	/** Basic per-task complete */
167	private void completeThis() {
168	complete(null);
169	}
170
171	/** Basic per-task completeExceptionally */
172	private void completeThisExceptionally(Throwable ex) {
173	super.completeExceptionally(ex);
174	}
175
176	/**
177	* Completes this task. If the pending subtask completion count is
178	* zero, invokes {@code onCompletion}, then causes this task to
179	* be joinable ({@code isDone} becomes true), and then
180	* recursively applies to this task's parent, if it exists. If an
181	* exception is encountered in any {@code onCompletion}
182	* invocation, that task and its ancestors
183	* {@code completeExceptionally}.
184	*/
185	public final void complete() {
186	LinkedAsyncAction a = this;
187	while (a != null && !a.isDone()) {
188	int c = a.getControlState();
189	if (c <= 0) {
190	try {
191	a.onCompletion();
192	a.completeThis();
193	} catch (Throwable rex) {
194	a.completeExceptionally(rex);
195	return;
196	}
197	a = a.parent;
198	}
199	else if (a.compareAndSetControlState(c, c-1))
200	return;
201	}
202	}
203
204	/**
205	* Completes this task abnormally. Unless this task already
206	* cancelled or aborted, upon invocation, this method invokes
207	* {@code onException}, and then, depending on its return value,
208	* completes parent (if one exists) exceptionally or normally. To
209	* avoid unbounded exception loops, this method aborts if an
210	* exception is encountered in any {@code onException}
211	* invocation.
212	* @param ex the exception to throw when joining this task
213	* @throws NullPointerException if ex is null
214	* @throws Throwable if any invocation of
215	* {@code onException} does so
216	*/
217	public final void completeExceptionally(Throwable ex) {
218	LinkedAsyncAction a = this;
219	while (!a.isCompletedAbnormally()) {
220	a.completeThisExceptionally(ex);
221	boolean up = a.onException(); // abort if this throws
222	a = a.parent;
223	if (a == null)
224	break;
225	if (!up) {
226	a.complete();
227	break;
228	}
229	}
230	}
231
232	/**
233	* Returns this task's parent, or null if none.
234	* @return this task's parent, or null if none
235	*/
236	public final LinkedAsyncAction getParent() {
237	return parent;
238	}
239
240	/**
241	* Returns the number of subtasks that have not yet completed.
242	* @return the number of subtasks that have not yet completed
243	*/
244	public final int getPendingSubtaskCount() {
245	return getControlState();
246	}
247
248	/**
249	* Resets the internal bookkeeping state of this task, maintaining
250	* the current parent but clearing pending joins.
251	*/
252	public void reinitialize() {
253	super.reinitialize();
254	}
255
256	/**
257	* Resets the internal bookkeeping state of this task, maintaining
258	* the current parent and setting pending joins to the given value.
259	* @param pending the number of pending joins
260	*/
261	public void reinitialize(int pending) {
262	super.reinitialize();
263	setControlState(pending);
264	}
265
266	/**
267	* Reinitialize with the given parent, optionally registering.
268	* @param parent the parent task, or null if none
269	* @param register true if parent must wait for this task
270	* to complete before it completes
271	*/
272	public void reinitialize(LinkedAsyncAction parent, boolean register) {
273	this.parent = parent;
274	super.reinitialize();
275	if (parent != null && register)
276	parent.incrementControlState();
277	}
278
279	/**
280	* Reinitialize with the given parent, optionally registering
281	* and setting pending join count.
282	* @param parent the parent task, or null if none
283	* @param register true if parent must wait for this task
284	* to complete before it completes
285	* @param pending the pending join count
286	*/
287	public void reinitialize(LinkedAsyncAction parent,
288	boolean register,
289	int pending) {
290	this.parent = parent;
291	super.reinitialize();
292	setControlState(pending);
293	if (parent != null && register)
294	parent.incrementControlState();
295	}
296
297	/**
298	* Gets the control state, which is initially zero, or negative if
299	* this task has completed or cancelled. Once negative, the value
300	* cannot be changed.
301	* @return control state
302	*/
303	protected final int getControlState() {
304	return controlState;
305	}
306
307	/**
308	* Atomically sets the control state to the given updated value if
309	* the current value is and equal to the expected value.
310	* @param expect the expected value
311	* @param update the new value
312	* @return true if successful
313	*/
314	protected final boolean compareAndSetControlState(int expect,
315	int update) {
316	return controlStateUpdater.compareAndSet(this, expect, update);
317	}
318
319	/**
320	* Sets the control state to the given value.
321	* @param value the new value
322	*/
323	protected final void setControlState(int value) {
324	controlState = value;
325	}
326
327	/**
328	* Increments the control state.
329	* @return true if successful
330	*/
331	protected final void incrementControlState() {
332	controlStateUpdater.incrementAndGet(this);
333	}
334
335	/**
336	* Decrements the control state.
337	* @return true if successful
338	*/
339	protected final void decrementControlState() {
340	controlStateUpdater.decrementAndGet(this);
341	}
342
343	}