Lesson 12 Concurrency

variables
Use a ReentrantReadWriteLock
Use the java.util.concurrent collections
Describe the synchronizer classes
Use an

ExecutorService to concurrently execute tasks
Apply the Fork-Join framework

classes that are useful in concurrent programming. Features include:
Concurrent collections
Synchronization

and locking alternatives
Thread pools
Fixed and dynamic thread count pools available
Parallel divide and conquer (Fork-Join) new in Java 7

thread-safe programming on single variables
AtomicInteger ai = new AtomicInteger(5);
if(ai.compareAndSet(5, 42))

{
System.out.println("Replaced 5 with 42");
}

An atomic operation ensures that the current value is 5 and then sets it to 42.

and waiting for conditions that is distinct from built-in synchronization

and monitors.
public class ShoppingCart {
private final ReentrantReadWriteLock rwl =
new ReentrantReadWriteLock();

public void addItem(Object o) {
rwl.writeLock().lock();
// modify shopping cart
rwl.writeLock().unlock();
}

A single writer, multi-reader lock

Write Lock

s = "";
rwl.readLock().lock();

// read cart, modify s
rwl.readLock().unlock();
return s;
}
public double getTotal() {
// another read-only method
}
}

All read-only methods can concurrently execute.

Read Lock

in a thread-safe fashion:
Use synchronized code blocks for all access

to a collection if writes are performed
Create a synchronized wrapper using library methods, such as java.util.Collections.synchronizedList(List)
Use the java.util.concurrent collections
Note: Just because a Collection is made thread-safe, this does not make its elements thread-safe.

the List.
True
False

synchronization idioms.

classes provided by java.util.concurrent.

final CyclicBarrier barrier = new CyclicBarrier(2);

new Thread()

{
public void run() {
try {
System.out.println("before await - thread 1");
barrier.await();
System.out.println("after await - thread 1");
} catch (BrokenBarrierException|InterruptedException ex) {

}
}
}.start();

Two threads must await before they can unblock.

May not be reached

use properly. Alternatives include:
java.util.concurrent.ExecutorService, a higher level mechanism used to

execute tasks
It may create and reuse Thread objects for you.
It allows you to submit work and check on the results in the future.
The Fork-Join framework, a specialized work-stealing ExecutorService new in Java 7

to manually create and manage threads.
Tasks might be executed in

parallel depending on the ExecutorService implementation.
Tasks can be:
java.lang.Runnable
java.util.concurrent.Callable
Implementing instances can be obtained with Executors.

ExecutorService es = Executors.newCachedThreadPool();

in nature to Runnable, but can:
Return a result using generics
Throw

a checked exception

package java.util.concurrent;
public interface Callable {
V call() throws Exception;
}

a Callable’s V call() method.


Future future = es.submit(callable);
//submit many callables
try

{
V result = future.get();
} catch (ExecutionException|InterruptedException ex) {

}

Gets the result of the Callable’s call method (blocks if needed).

ExecutorService controls when the work is done.

If the Callable threw an Exception

its threads are nondaemon threads and will keep your JVM

from shutting down.

es.shutdown();

try {
es.awaitTermination(5, TimeUnit.SECONDS);
} catch (InterruptedException ex) {
System.out.println("Stopped waiting early");
}

If you want to wait for the Callables to finish

Stop accepting new Callables.

available CPUs in a system.
True
False

time than concurrent blocking calls.

void main(String[] args) {
String host =

"localhost";
for (int port = 10000; port < 10010; port++) {
RequestResponse lookup =
new RequestResponse(host, port);
try (Socket sock = new Socket(lookup.host, lookup.port);
Scanner scanner = new Scanner(sock.getInputStream());){
lookup.response = scanner.next();
System.out.println(lookup.host + ":" + lookup.port + " " +
lookup.response);
} catch (NoSuchElementException|IOException ex) {
System.out.println("Error talking to " + host + ":" +
port);
}
}
}
}

public static void main(String[] args) {
//ThreadPool

used to execute Callables
ExecutorService es = Executors.newCachedThreadPool();
//A Map used to connect the request data with the result
Map> callables =
new HashMap<>();

String host = "localhost";
//loop to create and submit a bunch of Callable instances
for (int port = 10000; port < 10010; port++) {
RequestResponse lookup = new RequestResponse(host, port);
NetworkClientCallable callable =
new NetworkClientCallable(lookup);
Future future = es.submit(callable);
callables.put(lookup, future);
}

accepting new Callables
es.shutdown();

try {
//Block until all Callables have a chance to finish
es.awaitTermination(5, TimeUnit.SECONDS);
} catch (InterruptedException ex) {
System.out.println("Stopped waiting early");
}

lookup : callables.keySet()) {
Future

future = callables.get(lookup);
try {
lookup = future.get();
System.out.println(lookup.host + ":" + lookup.port + " " +
lookup.response);
} catch (ExecutionException|InterruptedException ex) {
//This is why the callables Map exists
//future.get() fails if the task failed
System.out.println("Error talking to " + lookup.host +
":" + lookup.port);
}
}
}
}

public String host; //request
public int port; //request

public String response; //response

public RequestResponse(String host, int port) {
this.host = host;
this.port = port;
}

// equals and hashCode

}

{
private RequestResponse lookup;

public NetworkClientCallable(RequestResponse

lookup) {
this.lookup = lookup;
}

@Override
public RequestResponse call() throws IOException {
try (Socket sock = new Socket(lookup.host, lookup.port);
Scanner scanner = new Scanner(sock.getInputStream());) {
lookup.response = scanner.next();
return lookup;
}
}
}

power in a system requires you to execute tasks in

parallel on multiple CPUs.
Divide and conquer: A task should be divided into subtasks. You should attempt to identify those subtasks that can be executed in parallel.
Some problems can be difficult to execute as parallel tasks.
Some problems are easier. Servers that support multiple clients can use a separate task to handle each client.
Be aware of your hardware. Scheduling too many parallel tasks can negatively impact performance.

leverage threads in some way, only a portion of your

system’s processing power will be utilized.

processed into multiple sets. One data set for each CPU

and one thread to process each data set.

subsets for each thread to process has a couple of

problems. Ideally all CPUs should be fully utilized until the task is finished but:
CPUs may run a different speeds
Non-Java tasks require CPU time and may reduce the time available for a Java thread to spend executing on a CPU

The data being analyzed may require varying amounts of time to process

into a large number of subsets
Assign the data subsets to

a thread’s processing queue

Each thread will have many subsets queued
If a thread finishes all its subsets early, it can “steal” subsets from another thread.

256]; //1G

for (int i = 0; i < data.length; i++)

{
data[i] = ThreadLocalRandom.current().nextInt();
}

int max = Integer.MIN_VALUE;
for (int value : data) {
if (value > max) {
max = value;
}
}
System.out.println("Max value found:" + max);

A very large dataset

Fill up the array with values.

Sequentially search the array for the largest value.

contains the code and data to be processed. Similar to

a Runnable or Callable.
A huge number of tasks are created and processed by a small number of threads in a Fork-Join pool.
A ForkJoinTask typically creates more ForkJoinTask instances until the data to processed has been subdivided adequately.
Developers typically use the following subclasses:
RecursiveAction: When a task does not need to return a result
RecursiveTask: When a task does need to return a result

int threshold;
private final int[] myArray;
private int

start;
private int end;

public FindMaxTask(int[] myArray, int start, int end, int threshold) {
// copy parameters to fields
}
protected Integer compute() {
// shown later
}
}

Result type of the task

The data to process

Where the work is done. Notice the generic return type.

PROCESS_DATA
return RESULT;
}

else {
SPLIT_DATA_INTO_LEFT_AND_RIGHT_PARTS
TASK t1 = new TASK(LEFT_DATA);
t1.fork();
TASK t2 = new TASK(RIGHT_DATA);
return COMBINE(t2.compute(), t1.join());
}
}

Block until done

Asynchronously execute

Process in current thread

- start < threshold) {
int max

= Integer.MIN_VALUE;
for (int i = start; i <= end; i++) {
int n = myArray[i];
if (n > max) {
max = n;
}
}
return max;
} else {
// split data and create tasks
}
}

You decide the threshold.

The range within the array

- start < threshold) {
// find

max
} else {
int midway = (end - start) / 2 + start;
FindMaxTask a1 =
new FindMaxTask(myArray, start, midway, threshold);
a1.fork();
FindMaxTask a2 =
new FindMaxTask(myArray, midway + 1, end, threshold);
return Math.max(a2.compute(), a1.join());
}
}

Task for left half of data

Task for right half of data

creates a thread for each CPU in the system by

default.

ForkJoinPool pool = new ForkJoinPool();
FindMaxTask task =
new FindMaxTask(data, 0, data.length-1, data.length/16);
Integer result = pool.invoke(task);

The task's compute method is automatically called .

CPU is created by default. Blocking operations would keep you

from utilizing all CPU resources.
Know your hardware.
A Fork-Join solution will perform slower on a one-CPU system than a standard sequential solution.
Some CPUs increase in speed when only using a single core, potentially offsetting any performance gain provided by Fork-Join.
Know your problem.
Many problems have additional overhead if executed in parallel (parallel sorting, for example).

benefit.
True
False

variables
Use a ReentrantReadWriteLock
Use the java.util.concurrent collections
Describe the synchronizer classes
Use an

ExecutorService to concurrently execute tasks
Apply the Fork-Join framework