Current Stream API implementation uses collector combiner to combine the intermediate results in exactly the same way as they were previously split. Also the splitting strategy depends on the source and common pool parallelism level, but does not depend on exact reduction operation used (the same for reduce, collect, forEach, count, etc.). Relying on this it’s not very difficult to create the visualizing collector:
public static Collector<Object, ?, List<String>> parallelVisualize() {
class Range {
private String first, last;
private Range left, right;
void accept(Object obj) {
if (first == null)
first = obj.toString();
else
last = obj.toString();
}
Range combine(Range that) {
Range p = new Range();
p.first = first == null ? that.first : first;
p.last = Stream
.of(that.last, that.first, this.last, this.first)
.filter(Objects::nonNull).findFirst().orElse(null);
p.left = this;
p.right = that;
return p;
}
String pad(String s, int left, int len) {
if (len == s.length())
return s;
char[] result = new char[len];
Arrays.fill(result, ' ');
s.getChars(0, s.length(), result, left);
return new String(result);
}
public List<String> finish() {
String cur = toString();
if (left == null) {
return Collections.singletonList(cur);
}
List<String> l = left.finish();
List<String> r = right.finish();
int len1 = l.get(0).length();
int len2 = r.get(0).length();
int totalLen = len1 + len2 + 1;
int leftAdd = 0;
if (cur.length() < totalLen) {
cur = pad(cur, (totalLen - cur.length()) / 2, totalLen);
} else {
leftAdd = (cur.length() - totalLen) / 2;
totalLen = cur.length();
}
List<String> result = new ArrayList<>();
result.add(cur);
char[] dashes = new char[totalLen];
Arrays.fill(dashes, ' ');
Arrays.fill(dashes, len1 / 2 + leftAdd + 1, len1 + len2 / 2 + 1
+ leftAdd, '_');
int mid = totalLen / 2;
dashes[mid] = "https://stackoverflow.com/";
dashes[mid + 1] = '\\';
result.add(new String(dashes));
Arrays.fill(dashes, ' ');
dashes[len1 / 2 + leftAdd] = '|';
dashes[len1 + len2 / 2 + 1 + leftAdd] = '|';
result.add(new String(dashes));
int maxSize = Math.max(l.size(), r.size());
for (int i = 0; i < maxSize; i++) {
String lstr = l.size() > i ? l.get(i) : String.format("%"
+ len1 + "s", "");
String rstr = r.size() > i ? r.get(i) : String.format("%"
+ len2 + "s", "");
result.add(pad(lstr + " " + rstr, leftAdd, totalLen));
}
return result;
}
public String toString() {
if (first == null)
return "(empty)";
else if (last == null)
return "[" + first + "]";
return "[" + first + ".." + last + "]";
}
}
return Collector.of(Range::new, Range::accept, Range::combine,
Range::finish);
}
Here’s some interesting results obtained with this collector using 4-core machine (results will differ on machine with different number of availableProcessors()).
Splitting of simple range:
IntStream.range(0, 100)
.boxed().parallel().collect(parallelVisualize())
.forEach(System.out::println);
Even split to 16 tasks:
[0..99]
___________________________________/\________________________________
| |
[0..49] [50..99]
_________________/\______________ _________________/\________________
| | | |
[0..24] [25..49] [50..74] [75..99]
________/\_____ ________/\_______ ________/\_______ ________/\_______
| | | | | | | |
[0..11] [12..24] [25..36] [37..49] [50..61] [62..74] [75..86] [87..99]
___/\_ ___/\___ ___/\___ ___/\___ ___/\___ ___/\___ ___/\___ ___/\___
| | | | | | | | | | | | | | | |
[0..5] [6..11] [12..17] [18..24] [25..30] [31..36] [37..42] [43..49] [50..55] [56..61] [62..67] [68..74] [75..80] [81..86] [87..92] [93..99]
Split of two streams concatenation:
IntStream
.concat(IntStream.range(0, 10), IntStream.range(10, 100))
.boxed().parallel().collect(parallelVisualize())
.forEach(System.out::println);
As you can see, first split un-concatenates the streams:
[0..99]
_______________________________________________________________________/\_____
| |
[0..9] [10..99]
__/\__ ___________________________________/\__________________________________
| | | |
[0..4] [5..9] [10..54] [55..99]
_________________/\________________ _________________/\________________
| | | |
[10..31] [32..54] [55..76] [77..99]
________/\_______ ________/\_______ ________/\_______ ________/\_______
| | | | | | | |
[10..20] [21..31] [32..42] [43..54] [55..65] [66..76] [77..87] [88..99]
___/\___ ___/\___ ___/\___ ___/\___ ___/\___ ___/\___ ___/\___ ___/\___
| | | | | | | | | | | | | | | |
[10..14] [15..20] [21..25] [26..31] [32..36] [37..42] [43..48] [49..54] [55..59] [60..65] [66..70] [71..76] [77..81] [82..87] [88..93] [94..99]
Split of two stream concatenation where intermediate operation (boxed()) was performed before concatenation:
Stream.concat(IntStream.range(0, 50).boxed().parallel(), IntStream.range(50, 100).boxed())
.collect(parallelVisualize())
.forEach(System.out::println);
If one of input streams was not turned into parallel mode before concatenation, it refuses to split at all:
[0..99]
___/\_________________________________
| |
[0..49] [50..99]
_________________/\______________
| |
[0..24] [25..49]
________/\_____ ________/\_______
| | | |
[0..11] [12..24] [25..36] [37..49]
___/\_ ___/\___ ___/\___ ___/\___
| | | | | | | |
[0..5] [6..11] [12..17] [18..24] [25..30] [31..36] [37..42] [43..49]
Split of flatmapping:
Stream.of(0, 50)
.flatMap(start -> IntStream.range(start, start+50).boxed().parallel())
.parallel().collect(parallelVisualize())
.forEach(System.out::println);
Flat-map never parallelizes inside nested streams:
[0..99]
____/\__
| |
[0..49] [50..99]
Stream from unknown-sized iterator of 7000 elements (see this answer for context):
StreamSupport
.stream(Spliterators.spliteratorUnknownSize(
IntStream.range(0, 7000).iterator(),
Spliterator.ORDERED), true)
.collect(parallelVisualize()).forEach(System.out::println);
The splitting is really bad, everybody waits for biggest part [3072..6143]:
[0..6999]
_______________________/\___
| |
[0..1023] [1024..6999]
________________/\____
| |
[1024..3071] [3072..6999]
_________/\_____
| |
[3072..6143] [6144..6999]
___/\____
| |
[6144..6999] (empty)
Iterator source with known size:
StreamSupport
.stream(Spliterators.spliterator(IntStream.range(0, 7000)
.iterator(), 7000, Spliterator.ORDERED), true)
.collect(parallelVisualize()).forEach(System.out::println);
Supplying the size makes things much better unlocking the further splitting:
[0..6999]
______________________________________________________________________________________________/\________
| |
[0..1023] [1024..6999]
_____/\__ ____________________________________________________________________/\________________________
| | | |
[0..511] [512..1023] [1024..3071] [3072..6999]
____________/\___________ ________________/\__________________________________________________
| | | |
[1024..2047] [2048..3071] [3072..6143] [6144..6999]
_____/\_____ _____/\_____ _________________________/\________________________ ___/\___________
| | | | | | | |
[1024..1535] [1536..2047] [2048..2559] [2560..3071] [3072..4607] [4608..6143] [6144..6999] (empty)
____________/\___________ ____________/\___________ _____/\_____
| | | | | |
[3072..3839] [3840..4607] [4608..5375] [5376..6143] [6144..6571] [6572..6999]
_____/\_____ _____/\_____ _____/\_____ _____/\_____
| | | | | | | |
[3072..3455] [3456..3839] [3840..4223] [4224..4607] [4608..4991] [4992..5375] [5376..5759] [5760..6143]
Further improvements of such collector is possible to generate graphical image (like svg), track the threads where each node is processed, display number of elements per each group and so on. Use it if you like.