dtonhofer

dtonhofer

Functional Programming in Java, Second Edition: Chapter 9, p.164 addendum to "parallel stream"

In Chapter 9, p.164 the stream is parallelized in 1 step.

This inspired my to write some test some code to call a task “in parallel”:

  • Not in parallel, inside a loop
  • Using ThreadGroups (essentially ‘temporary worker pools’) to run “slices” of the list the tasks in parallel
  • Using a Java 8 ForkJoinPool
  • Using a Java 8 parallel stream.

Not sure whether this is of interest, the most interesting part is the handling of checked and unchecked exceptions.

package chapter9;

import org.junit.jupiter.api.Test;

import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class RunningInParallel {

    // This could also implement Runnable instead.

    public static class DoSomething {

        private final int index;

        public DoSomething(int index) {
            this.index = index;
        }

        // This method should not throw any checked exceptions

        public void doSomething() {
            try {
                // Math.random is synchronized, so we can use it here
                final long sleep_ms = (long) (Math.random() * 1000.0);
                System.out.println("Thread " + index + " starts on thread '" + Thread.currentThread().getName() + "', sleeping for " + sleep_ms + " ms");
                Thread.sleep(sleep_ms);
            } catch (InterruptedException ex) {
                // Someone told us to stop sleeping, so we do!
                // Set the "interrupted" bit again and get out.
                Thread.currentThread().interrupt();
            }
        }

    }

    private List<DoSomething> createElements() {
        return IntStream.rangeClosed(1, 20)
                .mapToObj(DoSomething::new)
                .collect(Collectors.toList());
    }

    private static ThreadGroup startThreads(final int sliceStart, final int sliceEndIncl, final List<DoSomething> elements) {
        final int slizeSize = sliceEndIncl - sliceStart + 1;
        ThreadGroup tGroup = new ThreadGroup("slice [" + sliceStart + "," + sliceEndIncl + "] of size " + slizeSize);
        for (int threadIndex = 0; threadIndex < slizeSize; threadIndex++) {
            final int elementIndex = sliceStart + threadIndex;
            // no need to retain reference to the Thread, we will get it back from the ThreadGroup
            new Thread(tGroup, new Runnable() {
                @Override
                public void run() {
                    // If doSomething() threw a checked exception, we would have to catch it here
                    // If doSomething() throws a RuntimeException, the Exception is left up the stack here,
                    // terminating the worker thread.
                    elements.get(elementIndex).doSomething();
                }
            }).start();
        }
        return tGroup;
    }

    private static void waitForThreadEnd(final ThreadGroup tGroup, final int slizeSize) throws InterruptedException {
        final Thread[] threads = new Thread[slizeSize];
        final int count = tGroup.enumerate(threads);
        assert count <= slizeSize; // some threads may already have finished
        for (int i = 0; i < count; i++) {
            try {
                // Dangerous, as infinite waiting may follow, there should be a timeout value!!
                System.out.println("Joining thread " + (i + 1) + " of " + count);
                threads[i].join();
            } catch (InterruptedException ex) {
                // What should we do here? Just set the interrupt flag and throw...
                Thread.currentThread().interrupt();
                throw ex;
            }
        }
    }

    @Test
    void preJava8_singleThread() {
        final List<DoSomething> elements = Collections.unmodifiableList(createElements());
        for (DoSomething elem : elements) {
            // If doSomething() threw a checked exception, we would have to catch it here.
            // If doSomething() throws a RuntimeException, the Exception is left up the stack here.
            elem.doSomething();
        }
        System.out.println("DONE!");
    }

    @Test
    void preJava8_rollYourOwnMultipleThreads() throws InterruptedException {
        final List<DoSomething> elements = Collections.unmodifiableList(createElements());
        final int threadCount = 7;
        // Iterate over "slices" of "threadCount" threads.
        int sliceIndex = 0;
        while (sliceIndex * threadCount < elements.size()) {
            final int sliceStart = sliceIndex * threadCount;
            final int sliceEndIncl = Math.min(sliceStart + threadCount - 1, elements.size() - 1);
            final int slizeSize = sliceEndIncl - sliceStart + 1;
            ThreadGroup tGroup = startThreads(sliceStart, sliceEndIncl, elements);
            waitForThreadEnd(tGroup, slizeSize);
            System.out.println("Done with ThreadGroup '" + tGroup.getName() + "' running " + slizeSize + " threads");
            sliceIndex++;
        }
        System.out.println("DONE!");
    }

    @Test
    void java8_multipleThreadsWithForkJoinPool() throws InterruptedException {
        final List<DoSomething> elements = Collections.unmodifiableList(createElements());
        final List<ForkJoinTask<?>> tasks = new LinkedList<>();
        for (DoSomething elem : elements) {
            // If doSomething() threw a checked exception, we COULD NOT use it as argument to submit()
            // We would need to wrap doSomething().
            // Note that we use the "common pool" provided by the runtime environment.
            // We could also create our own pool instead, but why bother?
            // Note that some of the tasks will actaully be run by the main thread instead
            // of by a thread from the pool.
            tasks.add(ForkJoinPool.commonPool().submit(elem::doSomething));
        }
        for (ForkJoinTask<?> task : tasks) {
            try {
                task.get();
            } catch (ExecutionException ex) {
                // If doSomething() throws a RuntimeException it will be rethrown as an ExecutionException.
                // The thrown RuntimeException will appear as the "cause".
                System.err.println("Task failed to finish properly, got ExecutionException: '" + ex.getMessage()
                        + "' caused by: '" + ex.getCause() + "'");
            } catch (CancellationException ex) {
                System.err.println("Task was cancelled, hot CancellationException: " + ex.getMessage());
            } catch (InterruptedException ex) {
                // What should we do here? Just set the interrupt flag and throw.
                // Note the ,ethod "doSomething()" does not actually throw it.
                Thread.currentThread().interrupt();
                throw ex;
            }
        }
        System.out.println("DONE!");
    }

    @Test
    void java8_multipleThreadsWithStream() throws InterruptedException {
        final List<DoSomething> elements = Collections.unmodifiableList(createElements());
        // If doSomething() threw a checked exception (a subclass of Exception),
        // we COULD NOT use it as argument to forEach().
        // If doSomething() throws a RuntimeException, the stream pipeline would terminate
        // arbitrarily with any of the exceptions thrown in any of the threads.
        elements.stream().parallel().forEach(DoSomething::doSomething);
        System.out.println("DONE!");
    }
}

Where Next?

Popular Pragmatic Bookshelf topics Top

jimschubert
In Chapter 3, the source for index introduces Config on page 31, followed by more code including tests; Config isn’t introduced until pag...
New
telemachus
Python Testing With Pytest - Chapter 2, warnings for “unregistered custom marks” While running the smoke tests in Chapter 2, I get these...
New
jesse050717
Title: Web Development with Clojure, Third Edition, pg 116 Hi - I just started chapter 5 and I am stuck on page 116 while trying to star...
New
sdmoralesma
Title: Web Development with Clojure, Third Edition - migrations/create not working: p159 When I execute the command: user=&gt; (create-...
New
patoncrispy
I’m new to Rust and am using this book to learn more as well as to feed my interest in game dev. I’ve just finished the flappy dragon exa...
New
hgkjshegfskef
The test is as follows: Scenario: Intersecting a scaled sphere with a ray Given r ← ray(point(0, 0, -5), vector(0, 0, 1)) And s ← sphere...
New
akraut
The markup used to display the uploaded image results in a Phoenix.LiveView.HTMLTokenizer.ParseError error. lib/pento_web/live/product_l...
New
jonmac
The allprojects block listed on page 245 produces the following error when syncing gradle: “org.gradle.api.GradleScriptException: A prob...
New
dtonhofer
@parrt In the context of Chapter 4.3, the grammar Java.g4, meant to parse Java 6 compilation units, no longer passes ANTLR (currently 4....
New
mcpierce
@mfazio23 I’ve applied the changes from Chapter 5 of the book and everything builds correctly and runs. But, when I try to start a game,...
New

Other popular topics Top

ohm
Which, if any, games do you play? On what platform? I just bought (and completed) Minecraft Dungeons for my Nintendo Switch. Other than ...
New
dasdom
No chair. I have a standing desk. This post was split into a dedicated thread from our thread about chairs :slight_smile:
New
AstonJ
Curious to know which languages and frameworks you’re all thinking about learning next :upside_down_face: Perhaps if there’s enough peop...
New
PragmaticBookshelf
Rust is an exciting new programming language combining the power of C with memory safety, fearless concurrency, and productivity boosters...
New
New
hilfordjames
There appears to have been an update that has changed the terminology for what has previously been known as the Taskbar Overflow - this h...
New
PragmaticBookshelf
Author Spotlight: Peter Ullrich @PJUllrich Data is at the core of every business, but it is useless if nobody can access and analyze ...
New
sir.laksmana_wenk
I’m able to do the “artistic” part of game-development; character designing/modeling, music, environment modeling, etc. However, I don’t...
New
PragmaticBookshelf
A concise guide to MySQL 9 database administration, covering fundamental concepts, techniques, and best practices. Neil Smyth MySQL...
New
PragmaticBookshelf
Use advanced functional programming principles, practical Domain-Driven Design techniques, and production-ready Elixir code to build scal...
New

Latest in Functional Programming in Java, Second Edition

Functional Programming in Java, Second Edition Portal

Sub Categories: