Specific Number of Parallel Streams

[Johnathan Davis]

I'm going through the book "Java 8 for the really impatient" by Cay Hortsmann and I'm trying to do the practice exercises at the end of each chapter.

On page 44, question 1 in chapter 2 asks the reader to run a number of parallel streams equal to the number of cores in the system. From looking on stackoverflow, I found the following code to determine the number of cores:

int cores = Runtime.getRuntime().availableProcessors();

Now, I'm really stuck on trying to figure out how to tell the stream api how many cores to use. The book provides the following single-stream version (note that words is an arraylist):

long parallelCount = words.stream().filter(w -> w.length() > 12).count();

I know that I can change words.stream to words.parallelStream(), but that will make an indeterminate number of streams. Any hints about how to tell the stream api - make 8 streams and run them in parallel? Was the author intending me to use pre-java 8 methodology to create threads and run the streams manually? Any hints would be appreciated. Thanks!

[Campbell Ritchie]

Welcome to the Ranch

Does a parallel stream allow you to specify the number of threads/streams used at all? Let's look at the parallel method. It doesn't take any arguments and doesn't appear to be overloaded. I am not sure how many Streams that method sets off in parallel; it might be optimised to the number of cores anyway.

[Jeanne Boyarsky]

Johnathan,
When I read the book, I glossed over that part of the question and just used parallelStream(). The book was written with open JDK very early on so it is possible this functionality used to exist. If you actually need that control, you'd use an ExecutorService or ForkJoinPool rather than parallelStream(). It turns out that needing that control is an advanced feature so most people don't have to worry about it. Since this is question #1, I assumed it was a simple question and suspect it was when the book was written.

As a point of interest, the Stream API sets off two times the number of threads in parallel than you have cores. On my machine, I have 8 cores and get 16 parallel operations. The following shows this. IF I remove the +1 from the code, it runs to completion. All 16 threads hit the cyclic carrier and then the barrier releases them all at once. However with 17 threads, the last one gets stuck and the program hangs.

int cores = Runtime.getRuntime().availableProcessors();
System.out.println(cores); // 8
CyclicBarrier b = new CyclicBarrier(cores);
IntStream.rangeClosed(1, cores * 2 +1 ).parallel().forEach(i -> {
	System.out.println("run");
	try {
		b.await();
	} catch (Exception e) {
		e.printStackTrace();
	}
});

[Johnathan Davis]

Hi Jeanne and Campbell,

Thanks for the helpful answers. I don't know if there is a real-world scenario for wanting to limit the number of parallel streams. This was just one of the questions in the book that I couldn't figure out and I was wondering if someone smarter than me knew how to do it. It does make sense that they may have dropped this feature from the beta to the final version since it doesn't seem to be super-useful.

[Jeanne Boyarsky]

Johnathan,
For people who do expert level work with streams, that control is useful. I'll be honest, I'm not at that level. My friend and co-author Scott is. Which means he feels this missing API a lot. Whereas for me, it would be rope to hang myself with. (I'll let you guess who wrote the concurrency chapter in our upcoming book!)

[Stefan Zobel]

Jeanne Boyarsky wrote:
As a point of interest, the Stream API sets off two times the number of threads in parallel than you have cores. On my machine, I have 8 cores and get 16 parallel operations. The following shows this. IF I remove the +1 from the code, it runs to completion. All 16 threads hit the cyclic carrier and then the barrier releases them all at once. However with 17 threads, the last one gets stuck and the program hangs.

int cores = Runtime.getRuntime().availableProcessors();
System.out.println(cores); // 8
CyclicBarrier b = new CyclicBarrier(cores);
IntStream.rangeClosed(1, cores * 2 +1 ).parallel().forEach(i -> {
	System.out.println("run");
	try {
		b.await();
	} catch (Exception e) {
		e.printStackTrace();
	}
});

@Jeanne

No, you don't have 2 * Runtime.getRuntime().availableProcessors() threads (at least not by default).

You have Runtime.getRuntime().availableProcessors() - 1 ForkJoinPool.commonPool worker threads plus the main thread (which makes 8 threads in total in your case).

Try "cores * 3" instead of "cores * 2 + 1" and you should see that it works, whereas "cores + 1" should hang the program. That's the behavior on my hexa-core machine at least.

What seems to be really happening when the number of ints is "n * cores + 1" (where n = 1, 2, 3, ...) is that the commonPool worker thread that gets assigned the task of processing the additional int is blocked at the CyclicBarrier while the main thread which has forked (directly or indirectly) the now blocked commonPool task is waiting for that task to signal completion (which won't happen, of course).