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