Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
openanolis
dragonwell8_jdk
提交
b7456c03
D
dragonwell8_jdk
项目概览
openanolis
/
dragonwell8_jdk
通知
3
Star
2
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
D
dragonwell8_jdk
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
b7456c03
编写于
11月 08, 2017
作者:
P
psandoz
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
8190974: Parallel stream execution within a custom ForkJoinPool should obey the parallelism
Reviewed-by: martin, tvaleev
上级
3bbef61d
变更
4
显示空白变更内容
内联
并排
Showing
4 changed file
with
177 addition
and
15 deletion
+177
-15
src/share/classes/java/util/stream/AbstractTask.java
src/share/classes/java/util/stream/AbstractTask.java
+20
-9
src/share/classes/java/util/stream/ForEachOps.java
src/share/classes/java/util/stream/ForEachOps.java
+2
-3
src/share/classes/java/util/stream/StreamSpliterators.java
src/share/classes/java/util/stream/StreamSpliterators.java
+3
-3
test/java/util/stream/test/org/openjdk/tests/java/util/stream/CustomFJPoolTest.java
.../org/openjdk/tests/java/util/stream/CustomFJPoolTest.java
+152
-0
未找到文件。
src/share/classes/java/util/stream/AbstractTask.java
浏览文件 @
b7456c03
/*
* Copyright (c) 2012, 201
3
, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2012, 201
7
, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
...
...
@@ -27,6 +27,7 @@ package java.util.stream;
import
java.util.Spliterator
;
import
java.util.concurrent.CountedCompleter
;
import
java.util.concurrent.ForkJoinPool
;
import
java.util.concurrent.ForkJoinWorkerThread
;
/**
* Abstract base class for most fork-join tasks used to implement stream ops.
...
...
@@ -88,13 +89,7 @@ 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
;
private
static
final
int
LEAF_TARGET
=
ForkJoinPool
.
getCommonPoolParallelism
()
<<
2
;
/** The pipeline helper, common to all tasks in a computation */
protected
final
PipelineHelper
<
P_OUT
>
helper
;
...
...
@@ -156,6 +151,22 @@ abstract class AbstractTask<P_IN, P_OUT, R,
this
.
targetSize
=
parent
.
targetSize
;
}
/**
* Default target 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.
*/
public
static
int
getLeafTarget
()
{
Thread
t
=
Thread
.
currentThread
();
if
(
t
instanceof
ForkJoinWorkerThread
)
{
return
((
ForkJoinWorkerThread
)
t
).
getPool
().
getParallelism
()
<<
2
;
}
else
{
return
LEAF_TARGET
;
}
}
/**
* Constructs a new node of type T whose parent is the receiver; must call
* the AbstractTask(T, Spliterator) constructor with the receiver and the
...
...
@@ -181,7 +192,7 @@ abstract class AbstractTask<P_IN, P_OUT, R,
* @return suggested target leaf size
*/
public
static
long
suggestTargetSize
(
long
sizeEstimate
)
{
long
est
=
sizeEstimate
/
LEAF_TARGET
;
long
est
=
sizeEstimate
/
getLeafTarget
()
;
return
est
>
0L
?
est
:
1L
;
}
...
...
src/share/classes/java/util/stream/ForEachOps.java
浏览文件 @
b7456c03
/*
* Copyright (c) 2012, 201
3
, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2012, 201
7
, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
...
...
@@ -28,7 +28,6 @@ import java.util.Objects;
import
java.util.Spliterator
;
import
java.util.concurrent.ConcurrentHashMap
;
import
java.util.concurrent.CountedCompleter
;
import
java.util.concurrent.ForkJoinTask
;
import
java.util.function.Consumer
;
import
java.util.function.DoubleConsumer
;
import
java.util.function.IntConsumer
;
...
...
@@ -378,7 +377,7 @@ final class ForEachOps {
this
.
spliterator
=
spliterator
;
this
.
targetSize
=
AbstractTask
.
suggestTargetSize
(
spliterator
.
estimateSize
());
// Size map to avoid concurrent re-sizes
this
.
completionMap
=
new
ConcurrentHashMap
<>(
Math
.
max
(
16
,
AbstractTask
.
LEAF_TARGET
<<
1
));
this
.
completionMap
=
new
ConcurrentHashMap
<>(
Math
.
max
(
16
,
AbstractTask
.
getLeafTarget
()
<<
1
));
this
.
action
=
action
;
this
.
leftPredecessor
=
null
;
}
...
...
src/share/classes/java/util/stream/StreamSpliterators.java
浏览文件 @
b7456c03
/*
* Copyright (c) 2012, 201
3
, Oracle and/or its affiliates. All rights reserved.
* Copyright (c) 2012, 201
7
, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
...
...
@@ -898,7 +898,7 @@ class StreamSpliterators {
* Note: The source spliterator may report {@code ORDERED} since that
* spliterator be the result of a previous pipeline stage that was
* collected to a {@code Node}. It is the order of the pipeline stage
* that governs whether th
e th
is slice spliterator is to be used or not.
* that governs whether this slice spliterator is to be used or not.
*/
static
abstract
class
UnorderedSliceSpliterator
<
T
,
T_SPLITR
extends
Spliterator
<
T
>>
{
static
final
int
CHUNK_SIZE
=
1
<<
7
;
...
...
@@ -915,7 +915,7 @@ class StreamSpliterators {
this
.
unlimited
=
limit
<
0
;
this
.
skipThreshold
=
limit
>=
0
?
limit
:
0
;
this
.
chunkSize
=
limit
>=
0
?
(
int
)
Math
.
min
(
CHUNK_SIZE
,
((
skip
+
limit
)
/
AbstractTask
.
LEAF_TARGET
)
+
1
)
:
CHUNK_SIZE
;
((
skip
+
limit
)
/
AbstractTask
.
getLeafTarget
()
)
+
1
)
:
CHUNK_SIZE
;
this
.
permits
=
new
AtomicLong
(
limit
>=
0
?
skip
+
limit
:
skip
);
}
...
...
test/java/util/stream/test/org/openjdk/tests/java/util/stream/CustomFJPoolTest.java
0 → 100644
浏览文件 @
b7456c03
/*
* Copyright (c) 2017, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation. Oracle designates this
* particular file as subject to the "Classpath" exception as provided
* by Oracle in the LICENSE file that accompanied this code.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
package
org.openjdk.tests.java.util.stream
;
import
org.testng.annotations.Test
;
import
java.util.Comparator
;
import
java.util.Spliterator
;
import
java.util.concurrent.ForkJoinPool
;
import
java.util.concurrent.ForkJoinTask
;
import
java.util.concurrent.atomic.AtomicInteger
;
import
java.util.function.Consumer
;
import
java.util.stream.IntStream
;
import
java.util.stream.StreamSupport
;
import
static
org
.
testng
.
Assert
.
assertEquals
;
import
static
org
.
testng
.
Assert
.
assertTrue
;
/**
* Tests stream execution in a custom ForkJoinPool. See JDK-8190974.
*/
@Test
public
class
CustomFJPoolTest
{
// A Spliterator that counts the number of spliterators created
// including itself, thus the count starts at 1
static
class
SplitCountingSpliterator
<
T
>
implements
Spliterator
<
T
>
{
final
Spliterator
<
T
>
s
;
final
AtomicInteger
nsplits
;
// Top-level constructor
public
SplitCountingSpliterator
(
Spliterator
<
T
>
s
)
{
this
.
s
=
s
;
nsplits
=
new
AtomicInteger
(
1
);
}
// Splitting constructor
SplitCountingSpliterator
(
Spliterator
<
T
>
s
,
AtomicInteger
nsplits
)
{
this
.
s
=
s
;
this
.
nsplits
=
nsplits
;
}
int
splits
()
{
return
nsplits
.
get
();
}
@Override
public
boolean
tryAdvance
(
Consumer
<?
super
T
>
action
)
{
return
s
.
tryAdvance
(
action
);
}
@Override
public
void
forEachRemaining
(
Consumer
<?
super
T
>
action
)
{
s
.
forEachRemaining
(
action
);
}
@Override
public
Spliterator
<
T
>
trySplit
()
{
Spliterator
<
T
>
split
=
s
.
trySplit
();
if
(
split
!=
null
)
{
nsplits
.
incrementAndGet
();
return
new
SplitCountingSpliterator
<>(
split
,
nsplits
);
}
else
{
return
null
;
}
}
@Override
public
long
estimateSize
()
{
return
s
.
estimateSize
();
}
@Override
public
long
getExactSizeIfKnown
()
{
return
s
.
getExactSizeIfKnown
();
}
@Override
public
int
characteristics
()
{
return
s
.
characteristics
();
}
@Override
public
boolean
hasCharacteristics
(
int
characteristics
)
{
return
s
.
hasCharacteristics
(
characteristics
);
}
@Override
public
Comparator
<?
super
T
>
getComparator
()
{
return
s
.
getComparator
();
}
}
public
void
testCustomPools
()
throws
Exception
{
int
splitsForP1
=
countSplits
(
new
ForkJoinPool
(
1
));
int
splitsForP2
=
countSplits
(
new
ForkJoinPool
(
2
));
assertEquals
(
splitsForP2
,
splitsForP1
*
2
);
int
commonParallelism
=
ForkJoinPool
.
getCommonPoolParallelism
();
if
(
commonParallelism
>
1
&&
commonParallelism
<
128
)
{
int
splitsForPHalfC
=
countSplits
(
new
ForkJoinPool
(
commonParallelism
/
2
));
int
splitsForPC
=
countSplits
(
ForkJoinPool
.
commonPool
());
assertTrue
(
splitsForPHalfC
<
splitsForPC
);
assertEquals
(
splitsForPC
/
splitsForPHalfC
,
nearestPowerOfTwo
(
commonParallelism
)
/
nearestPowerOfTwo
(
commonParallelism
/
2
));
}
}
static
int
countSplits
(
ForkJoinPool
fjp
)
throws
Exception
{
// The number of splits will be equivalent to the number of leaf nodes
// and will be a power of 2
ForkJoinTask
<
Integer
>
fInteger
=
fjp
.
submit
(()
->
{
Spliterator
<
Integer
>
s
=
IntStream
.
range
(
0
,
1024
).
boxed
().
parallel
().
spliterator
();
SplitCountingSpliterator
<
Integer
>
cs
=
new
SplitCountingSpliterator
<>(
s
);
StreamSupport
.
stream
(
cs
,
true
).
forEach
(
e
->
{});
return
cs
.
splits
();
});
return
fInteger
.
get
();
}
static
int
nearestPowerOfTwo
(
int
i
)
{
return
(
i
&
(
i
-
1
))
==
0
?
i
:
1
<<
(
32
-
Integer
.
numberOfLeadingZeros
(
i
));
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录