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

iPaul
page 37 ANTLRInputStream input = new ANTLRInputStream(is); as of ANTLR 4 .8 should be: CharStream stream = CharStreams.fromStream(i...
New
edruder
I thought that there might be interest in using the book with Rails 6.1 and Ruby 2.7.2. I’ll note what I needed to do differently here. ...
New
Mmm
Hi, build fails on: bracket-lib = “~0.8.1” when running on Mac Mini M1 Rust version 1.5.0: Compiling winit v0.22.2 error[E0308]: mi...
New
HarryDeveloper
Hi @venkats, It has been mentioned in the description of ‘Supervisory Job’ title that 2 things as mentioned below result in the same eff...
New
New
Henrai
Hi, I’m working on the Chapter 8 of the book. After I add add the point_offset, I’m still able to see acne: In the image above, I re...
New
tkhobbes
After some hassle, I was able to finally run bin/setup, now I have started the rails server but I get this error message right when I vis...
New
EdBorn
Title: Agile Web Development with Rails 7: (page 70) I am running windows 11 pro with rails 7.0.3 and ruby 3.1.2p20 (2022-04-12 revision...
New
a.zampa
@mfazio23 I’m following the indications of the book and arriver ad chapter 10, but the app cannot be compiled due to an error in the Bas...
New
dachristenson
I’ve got to the end of Ch. 11, and the app runs, with all tabs displaying what they should – at first. After switching around between St...
New

Other popular topics Top

Devtalk
Hello Devtalk World! Please let us know a little about who you are and where you’re from :nerd_face:
New
PragmaticBookshelf
Free and open source software is the default choice for the technologies that run our world, and it’s built and maintained by people like...
New
siddhant3030
I’m thinking of buying a monitor that I can rotate to use as a vertical monitor? Also, I want to know if someone is using it for program...
New
New
PragmaticBookshelf
Use WebRTC to build web applications that stream media and data in real time directly from one user to another, all in the browser. ...
New
PragmaticBookshelf
Author Spotlight Rebecca Skinner @RebeccaSkinner Welcome to our latest author spotlight, where we sit down with Rebecca Skinner, auth...
New
PragmaticBookshelf
Programming Ruby is the most complete book on Ruby, covering both the language itself and the standard library as well as commonly used t...
New
First poster: bot
zig/http.zig at 7cf2cbb33ef34c1d211135f56d30fe23b6cacd42 · ziglang/zig. General-purpose programming language and toolchain for maintaini...
New
PragmaticBookshelf
Develop, deploy, and debug BEAM applications using BEAMOps: a new paradigm that focuses on scalability, fault tolerance, and owning each ...
New
xiji2646-netizen
There’s a GitHub repo at forrestchang/andrej-karpathy-skills that’s sitting at 97.8k stars. It’s a single CLAUDE.md file with four behavi...
New

Latest in Functional Programming in Java, Second Edition

Functional Programming in Java, Second Edition Portal

Sub Categories: