Learn about parallel and concurrent programming in Java.

Parallel and Concurrent Programming: Lesson

Congratulations! You completed the lesson on parallel and concurrent programming! 🙌

Key concepts covered in this lesson:
 - What Concurrency and Parallelism are and what their differences are
 - Thread pools and how they work
 - The Executor framework and how it uses thread pooling
 - The Fork-Join framework and how it uses ForkJoinPool and ForkJoinExecutor
 - How to implement true parallelism through the use of Fork-Join and Parallel Streams
 - What Streams are and how they can be used in Parallel Streams


Review

Now that we know what Java Streams are and how they work, we can start to understand why they’re good for parallel processing. They ultimately don’t alter the original object passed to them, which helps with the thread-safe problem.

Additionally, Java normally has one stream for processing code, which uses sequential execution and is the default when performing normal threading.

_Java Parallel Streams_ divide a running process into multiple streams that execute in parallel on separate cores, returning a combined result of all the individual outcomes. However, the order of execution is not under our control.

This means we want to try to only use parallel streams when the order of execution doesn’t matter.

To create a parallel stream, we need to invoke the operation `Collection.parallelStream`. Implementing this may look something like this:

```java
File file = new File(filePath);
List<String> textLines = Files.readAllLines(file.toPath());
textLines.parallelStream().forEach(System.out::println);
```

Alternatively, we can also invoke the operation `BaseStream.parallel`. Implementing this may look something like this:

```java
File file = new File(filePath);
Stream<String> textLines = Files.lines(file.toPath());
textLines.parallel().forEach(System.out::println);
textLines.close();
```


Java Parallel Streams

So far, we’ve seen that parallelism is hard to come by. The biggest reason for this is the innate issue of making something “thread-safe.” If something is thread-safe, then multiple threads can interact with it without causing problems.

Unfortunately, a big problem exists when trying to enforce this concept of thread-safe. Collections are not thread-safe, which include `List`s, `Map`s, and `Set`s.

Multiple threads can’t manipulate a collection without introducing thread interference or memory consistency errors. The former occurs when one thread overwrites the results of another thread in an unpredictable way, which can cause unexpected results when reading the altered data. The latter occurs when different threads have inconsistent views of what should be the same data.

And even if we introduce synchronization to make these collections thread-safe, doing so then also introduces thread contention. And thread contention directly contradicts parallelism due to it preventing threads from interacting with a resource while it’s being used by another thread.

So… how did Fork-Join get around this?

As you saw in the previous exercise, the Fork-Join framework was the first framework we’ve worked with so far that legitimately implements parallelism into its processing. The way it does this is through something called aggregate operations, which performs all the partitioning and combining of solutions during Java runtime.

We won’t get into exactly what aggregate operations are or how they work, but just know they’re one key way to implement parallelism. Another key way, which we will get into, is something called parallel streams.

Note: remember that parallelism doesn’t automatically result in faster performance than serial processing. while aggregate operations and parallel streams more easily implement parallelism, it is still the coder’s responsibility to determine if their application is even suitable for parallelism.


Parallelism

To explain what parallel streams are, we need to first talk about what exactly a Java Stream is. _Java Streams_ were introduced in Java 8, and are used to process a collection of objects which can be pipelined to produce a desired result.

A stream itself is not a data structure, it instead takes its input from collections, arrays, or other I/O channels. Additionally, streams don’t change the original data structure and only return results based on pipelined methods.

In a stream, intermediate operations are lazily executed and return results in the form of additional streams, which allow for the pipelining of those results. Terminal operations mark the end of the stream and return the final result.

An operation is _lazy_ if its value is evaluated only when it is required. The reason why it does this is to save the processing of actual stream contents until the terminal operation commences. This enables the Java compiler and runtime to optimize how they process streams.

Let’s go over some different types of operations.

Intermediate Operations:
- `map`: returns a stream consisting of the results after applying a given function to the elements of this stream

```java
List numbers = Arrays.asList(1,2,3,4);
List square = numbers.stream().map(x->x*x).collect(Collectors.toList());
```

- `filter`: selects elements as per a Predicate passed as an argument

```java
List names = Arrays.asList(“Alpha”,”Beta”,”Omega”);
List find = names.stream().filter(s->s.startsWith(“O”)).collect(Collectors.toList());
```

- `sorted`: sorts the stream

```java
List names = Arrays.asList(“Omega”,”Alpha”,”Beta”);
List sorted = names.stream().sorted().collect(Collectors.toList());
```

Terminal Operations:
- `collect`: returns the result of the intermediate operations performed on the stream

```java
List numbers = Arrays.asList(1,2,3,4,2);
Set squareSet = numbers.stream().map(x->x*x).collect(Collectors.toSet());
```

- `forEach`: iterates through every element of the stream

```java
List numbers = Arrays.asList(1,2,3,4);
numbers.stream().map(x->x*x).forEach(y->System.out.println(y));
```

- `reduce`: reduces the elements of a stream to a single value (this takes a BinaryOperator as a parameter)

```java
List numbers = Arrays.asList(1,2,3,4);
int even = numbers.stream().filter(x->x%2==0).reduce(0,(ans,i)->ans+i);
```

In the above example, `ans` is a variable that is assigned the value 0 initially, then `i` is added to it.

Remember, everything a stream does has no effect on the original value of the object initially passed to it.


Java Streams

In Intermediate Java, we learned how to implement concurrent threading by using Threads and Synchronization. We’ll cover a few more implementations for that, however, under normal circumstances, most multithreading performed in Java is processed on a single stream.

Java 7 provides a Fork-Join framework to enable parallel computing, but it requires you to specify exactly how the problems are subdivided. To truly incorporate parallelism into our programs, we need to utilize multiple streams, which is a Java 8+ feature. We’ll discuss exactly what that means later in this lesson.

For now, let’s quickly recap the definitions for concurrency and parallelism:
 - **Concurrency** is the act of processing more than one task at seemingly the same time on the same CPU, requiring the ability to switch between tasks.
 - **Parallelism** is the act of splitting tasks into smaller subtasks and processing those subtasks in parallel, for instance across multiple CPUs at the exact same time.

Java is very particular about how it allocates its resources for threading. For instance, the JVM understands your static `main` runner as a `main thread` whenever you run your projects. You can see a demonstration of this by implementing the following print statement into any project’s static `main` method:

```java
System.out.println(“Thread name: “ + Thread.currentThread().getName());
```

The following will be printed out:
```
Thread name: main
```

Whenever we create additional threads, either through extending `Thread` or implementing `Runnable`, they will all run _concurrently_ (including the main thread). This is because the JVM likes to keep a process’s running threads in the same place by default.

It’s important to note that concurrent threading isn’t a bad thing, and parallelism should not be considered “better.” Each of these concepts has its strengths and levels of efficiency, and there will always be a tradeoff of overhead in both.

Programs that don’t need parallelism should not be forced into it, but do know that sometimes real parallelism will show very real benefits. But before we see how we can actually implement parallelism, let’s cover some open-source frameworks that make concurrent threading much easier to use and keep organized.


Introduction to Parallel and Concurrent Programming

The `Executor` framework implements thread pooling through an `Executor` interface. Using it is fairly intuitive and draws on already pre-existing functionality found in the Thread class, and the executor framework provides additional interfaces for various implementation strategies.

`Executor` interfaces include:
 - `Executor`: launch a `Runnable` object task
 - `ExecutorService`: manages the lifecycle of tasks in a sub-interface of `Executor`
 - `ScheduledExecutor`: schedules the execution of tasks in a sub-interface of `ExecutorService`

Let’s take a look at how we’ll be implementing an executor into our code. We’ll need the following imports:
```java
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
```

Additionally, since we need the tasks we enqueue into the thread pool to be `Runnable` objects, we’ll be creating a `Runnable` class accordingly. To use an executor, we need to create an `ExecutorService` object and pass it the number of threads we’ll be allotting it:
```java
private static final int N = 10;
ExecutorService executor = Executors.newFixedThreadPool(N);
```

Since the thread pool is responsible for those threads, it will automatically handle creating those and managing how they run. We just need to tell it how many to create and handle. At this point, all we need to do now is pass tasks to the executor, which we do like this (as usual, we’ll probably be putting this into a loop much like how we loop the creation and assignment of manual threads):
```java
Runnable task = new RunnableTask();
executor.execute(task);
```

In the above, `RunnableTask` is the custom class you’ll be creating that implements `Runnable`. Calling `executor.execute(task)` will enqueue the newly created `Runnable` object into the thread pool, which will be processed by one of the waiting threads. Now, thanks to the `ExecutorService` interface, we have some useful methods we can call to interact with the working threads.

If we want to prevent any new tasks from being added to our executor, we can call `executor.shutdown()`. In this case, the threads will still work and clear out the queue, but this will ensure nothing new gets added after this point.

If we want to wait for the thread pool to finish executing everything in its queue, we can call `executor.awaitTermination()`.

There are plenty more useful methods in `ExecutorService`, but we’ll be using just these in our basic executor. Let’s get some practice with this!


The Executor Framework

A similar interface to the executor service was added in Java 7 that included functionality to split a task into smaller subtasks and re-enqueue them into the thread pool. This is particularly useful for more intensive tasks and actually implements parallelism to do it (we’ll touch on this more in the next exercise).

This new interface is part of a Fork-Join framework and is called `ForkJoinPool`, which distributes a task across several worker threads and waits for a result.

Another big difference is the type of task object we’ll be passing to this pool. Instead of implementing the `Runnable` class, we’ll actually be extending either `RecursiveAction` or `RecursiveTask` depending on the implementation we want.

Here is the difference between the two:
- `RecursiveAction`: does not return any result
- `RecursiveTask`: returns a result

As the namesake suggests, the Fork-Join framework utilizes recursion to perform its subtask splitting, which we’ll limit by defining a set number of threads. This controls how many times the task is allowed to split itself since each subtask will be assigned to an individual thread.

Additionally, instead of calling `execute()` as we did before to add new tasks to the thread pool queue, we’ll be calling `invoke()`. Since we’re splitting a singular task across the worker threads, we only have to call `invoke()` once, which means we can invoke the task in the same place we create the pool itself:
```java
private static final int N = 4;
ForkJoinExecutor pool = new ForkJoinPool(N);
pool.invoke(task);
```


The Fork-Join Framework

an image of a blocking queue with five tasks and an additional sixth task being added

As your programs grow, it gets harder to manage all your threads, especially if you’re manually creating them as needed and joining them on an, even more, as-needed basis. However, this feels tedious and processor intensive.

Let’s ask a few vital questions regarding how the basic implementation of threading goes.

If we’re trying to save time, why do we waste so much overhead recreating threads? Wouldn’t it be more efficient to “reuse” threads? How do we best control how many threads are running at any given time? Is there even a way to do that, or better yet, automate this so we don’t have to?

What we need is a way to create all the threads we’ll need early and just pool them together in an easily accessible place. That way, whenever our program wants to offload work onto the threads, it can just find this pool and toss whatever processing it needs threaded into it.

No need for worrying about how many threads to create for the task. No need to create any threads as it goes. No need to assign each thread what they’ll be working on.

Conveniently, Java has a way to do exactly this. It’s called a thread pool.

A _thread pool_ manages a pool of worker threads that connect to a work queue of `Runnable` tasks waiting to be executed.

In this thread pool, the added threads idle until given work. A Blocking Queue is used to manage this work, which takes a given `Runnable` object as a task to be processed and enqueues it. The idling threads, which are all watching this queue, will process the enqueued tasks on a first come first serve basis and will sort out who gets what task amongst themselves.

Once a thread finishes its task, it simply returns to watching this Blocking Queue and processes whatever tasks get enqueued next.

In other words, you enqueue a task you want processed into a thread pool’s queue, and one of the idling threads will dequeue that task and process it. This keeps your threads busy and helps with reusability, plus it adds some much-needed constraints to your memory usage when creating and using threads.

Now, while you can create and use a custom thread pool yourself, we’re going to look at Java’s built-in ways to utilize thread pooling. It does this through something called an `Executor`, which we’ll talk about next!
