forked from wupeixuan/JDKSourceCode1.8
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathAbstractTask.java
More file actions
352 lines (327 loc) · 11.4 KB
/
AbstractTask.java
File metadata and controls
352 lines (327 loc) · 11.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
/*
* Copyright (c) 2012, 2013, Oracle and/or its affiliates. All rights reserved.
* ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*/
package java.util.stream;
import java.util.Spliterator;
import java.util.concurrent.CountedCompleter;
import java.util.concurrent.ForkJoinPool;
/**
* Abstract base class for most fork-join tasks used to implement stream ops.
* Manages splitting logic, tracking of child tasks, and intermediate results.
* Each task is associated with a {@link Spliterator} that describes the portion
* of the input associated with the subtree rooted at this task.
* Tasks may be leaf nodes (which will traverse the elements of
* the {@code Spliterator}) or internal nodes (which split the
* {@code Spliterator} into multiple child tasks).
*
* @implNote
* <p>This class is based on {@link CountedCompleter}, a form of fork-join task
* where each task has a semaphore-like count of uncompleted children, and the
* task is implicitly completed and notified when its last child completes.
* Internal node tasks will likely override the {@code onCompletion} method from
* {@code CountedCompleter} to merge the results from child tasks into the
* current task's result.
*
* <p>Splitting and setting up the child task links is done by {@code compute()}
* for internal nodes. At {@code compute()} time for leaf nodes, it is
* guaranteed that the parent's child-related fields (including sibling links
* for the parent's children) will be set up for all children.
*
* <p>For example, a task that performs a reduce would override {@code doLeaf()}
* to perform a reduction on that leaf node's chunk using the
* {@code Spliterator}, and override {@code onCompletion()} to merge the results
* of the child tasks for internal nodes:
*
* <pre>{@code
* protected S doLeaf() {
* spliterator.forEach(...);
* return localReductionResult;
* }
*
* public void onCompletion(CountedCompleter caller) {
* if (!isLeaf()) {
* ReduceTask<P_IN, P_OUT, T, R> child = children;
* R result = child.getLocalResult();
* child = child.nextSibling;
* for (; child != null; child = child.nextSibling)
* result = combine(result, child.getLocalResult());
* setLocalResult(result);
* }
* }
* }</pre>
*
* <p>Serialization is not supported as there is no intention to serialize
* tasks managed by stream ops.
*
* @param <P_IN> Type of elements input to the pipeline
* @param <P_OUT> Type of elements output from the pipeline
* @param <R> Type of intermediate result, which may be different from operation
* result type
* @param <K> Type of parent, child and sibling tasks
* @since 1.8
*/
@SuppressWarnings("serial")
abstract class AbstractTask<P_IN, P_OUT, R,
K extends AbstractTask<P_IN, P_OUT, R, K>>
extends CountedCompleter<R> {
/**
* Default target factor of leaf tasks for parallel decomposition.
* To allow load balancing, we over-partition, currently to approximately
* four tasks per processor, which enables others to help out
* if leaf tasks are uneven or some processors are otherwise busy.
*/
static final int LEAF_TARGET = ForkJoinPool.getCommonPoolParallelism() << 2;
/** The pipeline helper, common to all tasks in a computation */
protected final PipelineHelper<P_OUT> helper;
/**
* The spliterator for the portion of the input associated with the subtree
* rooted at this task
*/
protected Spliterator<P_IN> spliterator;
/** Target leaf size, common to all tasks in a computation */
protected long targetSize; // may be laziliy initialized
/**
* The left child.
* null if no children
* if non-null rightChild is non-null
*/
protected K leftChild;
/**
* The right child.
* null if no children
* if non-null leftChild is non-null
*/
protected K rightChild;
/** The result of this node, if completed */
private R localResult;
/**
* Constructor for root nodes.
*
* @param helper The {@code PipelineHelper} describing the stream pipeline
* up to this operation
* @param spliterator The {@code Spliterator} describing the source for this
* pipeline
*/
protected AbstractTask(PipelineHelper<P_OUT> helper,
Spliterator<P_IN> spliterator) {
super(null);
this.helper = helper;
this.spliterator = spliterator;
this.targetSize = 0L;
}
/**
* Constructor for non-root nodes.
*
* @param parent this node's parent task
* @param spliterator {@code Spliterator} describing the subtree rooted at
* this node, obtained by splitting the parent {@code Spliterator}
*/
protected AbstractTask(K parent,
Spliterator<P_IN> spliterator) {
super(parent);
this.spliterator = spliterator;
this.helper = parent.helper;
this.targetSize = parent.targetSize;
}
/**
* Constructs a new node of type T whose parent is the receiver; must call
* the AbstractTask(T, Spliterator) constructor with the receiver and the
* provided Spliterator.
*
* @param spliterator {@code Spliterator} describing the subtree rooted at
* this node, obtained by splitting the parent {@code Spliterator}
* @return newly constructed child node
*/
protected abstract K makeChild(Spliterator<P_IN> spliterator);
/**
* Computes the result associated with a leaf node. Will be called by
* {@code compute()} and the result passed to @{code setLocalResult()}
*
* @return the computed result of a leaf node
*/
protected abstract R doLeaf();
/**
* Returns a suggested target leaf size based on the initial size estimate.
*
* @return suggested target leaf size
*/
public static long suggestTargetSize(long sizeEstimate) {
long est = sizeEstimate / LEAF_TARGET;
return est > 0L ? est : 1L;
}
/**
* Returns the targetSize, initializing it via the supplied
* size estimate if not already initialized.
*/
protected final long getTargetSize(long sizeEstimate) {
long s;
return ((s = targetSize) != 0 ? s :
(targetSize = suggestTargetSize(sizeEstimate)));
}
/**
* Returns the local result, if any. Subclasses should use
* {@link #setLocalResult(Object)} and {@link #getLocalResult()} to manage
* results. This returns the local result so that calls from within the
* fork-join framework will return the correct result.
*
* @return local result for this node previously stored with
* {@link #setLocalResult}
*/
@Override
public R getRawResult() {
return localResult;
}
/**
* Does nothing; instead, subclasses should use
* {@link #setLocalResult(Object)}} to manage results.
*
* @param result must be null, or an exception is thrown (this is a safety
* tripwire to detect when {@code setRawResult()} is being used
* instead of {@code setLocalResult()}
*/
@Override
protected void setRawResult(R result) {
if (result != null)
throw new IllegalStateException();
}
/**
* Retrieves a result previously stored with {@link #setLocalResult}
*
* @return local result for this node previously stored with
* {@link #setLocalResult}
*/
protected R getLocalResult() {
return localResult;
}
/**
* Associates the result with the task, can be retrieved with
* {@link #getLocalResult}
*
* @param localResult local result for this node
*/
protected void setLocalResult(R localResult) {
this.localResult = localResult;
}
/**
* Indicates whether this task is a leaf node. (Only valid after
* {@link #compute} has been called on this node). If the node is not a
* leaf node, then children will be non-null and numChildren will be
* positive.
*
* @return {@code true} if this task is a leaf node
*/
protected boolean isLeaf() {
return leftChild == null;
}
/**
* Indicates whether this task is the root node
*
* @return {@code true} if this task is the root node.
*/
protected boolean isRoot() {
return getParent() == null;
}
/**
* Returns the parent of this task, or null if this task is the root
*
* @return the parent of this task, or null if this task is the root
*/
@SuppressWarnings("unchecked")
protected K getParent() {
return (K) getCompleter();
}
/**
* Decides whether or not to split a task further or compute it
* directly. If computing directly, calls {@code doLeaf} and pass
* the result to {@code setRawResult}. Otherwise splits off
* subtasks, forking one and continuing as the other.
*
* <p> The method is structured to conserve resources across a
* range of uses. The loop continues with one of the child tasks
* when split, to avoid deep recursion. To cope with spliterators
* that may be systematically biased toward left-heavy or
* right-heavy splits, we alternate which child is forked versus
* continued in the loop.
*/
@Override
public void compute() {
Spliterator<P_IN> rs = spliterator, ls; // right, left spliterators
long sizeEstimate = rs.estimateSize();
long sizeThreshold = getTargetSize(sizeEstimate);
boolean forkRight = false;
@SuppressWarnings("unchecked") K task = (K) this;
while (sizeEstimate > sizeThreshold && (ls = rs.trySplit()) != null) {
K leftChild, rightChild, taskToFork;
task.leftChild = leftChild = task.makeChild(ls);
task.rightChild = rightChild = task.makeChild(rs);
task.setPendingCount(1);
if (forkRight) {
forkRight = false;
rs = ls;
task = leftChild;
taskToFork = rightChild;
}
else {
forkRight = true;
task = rightChild;
taskToFork = leftChild;
}
taskToFork.fork();
sizeEstimate = rs.estimateSize();
}
task.setLocalResult(task.doLeaf());
task.tryComplete();
}
/**
* {@inheritDoc}
*
* @implNote
* Clears spliterator and children fields. Overriders MUST call
* {@code super.onCompletion} as the last thing they do if they want these
* cleared.
*/
@Override
public void onCompletion(CountedCompleter<?> caller) {
spliterator = null;
leftChild = rightChild = null;
}
/**
* Returns whether this node is a "leftmost" node -- whether the path from
* the root to this node involves only traversing leftmost child links. For
* a leaf node, this means it is the first leaf node in the encounter order.
*
* @return {@code true} if this node is a "leftmost" node
*/
protected boolean isLeftmostNode() {
@SuppressWarnings("unchecked")
K node = (K) this;
while (node != null) {
K parent = node.getParent();
if (parent != null && parent.leftChild != node)
return false;
node = parent;
}
return true;
}
}