INF 329 – Selectedtopics in programming theory

Java Concurrency and Java Concurrency and PracticePractice

Chapters 5 and 6Chapters 5 and 6

Speaker: Fabiano Fabiano IzzoIzzo

Chapter 5

“Building Blocks”

3

Intro

The Java platform libraries include a rich set of concurrent building blocks

•thread-safe collections

•variety of synchronizers for coordinate the control flow of cooperating threads

We will see the most useful concurrent building blocks and some patterns for using them.

5.1 - Synchronized collections

The synchronized collections classes include Vector and Hashtable but also include

•synchronized wrapper classes (Collections.synchronizedMap, Collections.synchronizedList,… )that achieve thread safety so that only one thread at time can access the collection state.

5.1.1 – Problems withsynchronized collections

The synchronized collections are thread-safe, but you may need to use additional client-side locking to protect compound actions on collections (iteration, navigation, conditional operations,… ).

When other threads can concurrently modify the collections, these compound action may not behave as you might expect.

5.1.1 – Example 1 (1)Compound actions on a Vector that may produce confusing results

5.1.1 – Example 1 (2)Interleaving of getLast and deleteLast that throws ArrayIndexOutOfBoundException

5.1.1 – Example 1 (3)Compound Actions on Vector Using Client-sideLocking. We can make methods atomic.

5.1.1 – Example 2Compound action Iteration that may throw ArrayIndexOutOfBoundException when other threads may concurrently modify the vector.

Solution

5.1.2 – Iterators and ConcurrentModificationException

The standard way to iterate a Collection is with an Iterator, but we have seen that using iterators does not avoid the need to lock the collection during iteration if other threads can concurrently modify it.

The iterators returned by the synchronized collections are fail-fast. If the collection has changed, they throw the unchecked ConcurrentModificationException

5.1.2 – … continue (1)

The fail-fast iterators are implemented (without synchronization) by modification count whit the collection. If it changes during iteration, hasNext or next throws the exception.

5.1.2 – … continue (2)

We can prevent this exception holding the collection lock for the duration of the iteration.

But sometimes this can be undesirable:•other threads could wait a long time

•deadlock risk (we will see it in Chap. 10)

•locking collection can hurts application scalability

•CPU utilization can suffer (we will see it in Chap. 11)

An alternative?

Clone the collection and iterate the copy instead.

5.1.3 – Hidden Iterators

There are situations wherein iterators are hidden. We have to remember to use locking everywhere a shared collection might be iterated.

In the next example there is no explicit iteration, but the code in evidence implicates iteration just the same.

5.1.3 – … continue (1) addTenThigs() could trowConcurrentModificationException

example code: 1

5.1.3 – … continue (2)What is the problem?

HiddenIterator is not thread-safe.

If HiddenIterator wrapped the HashSet with a synchronizedSet, encapsulating the synchronization, this sort of error would not occur.

Rule:

Encapsulating object’s state synchronizationmakes it easier to enforce its synchronizationpolicy.

5.2 – Concurrent collections Locking practice can cause throughput suffers.

In Java 5.0 the concurrent collections are designed for concurrent access from multiple threads (ex.: ConcurrentHashMap, CopyOnWriteArrayList).

Rule:

Replacing synchronized collections withconcurrent collections can offer dramaticscalability improvements with little risk.

Java 5.0 also adds two new collection type: Queue and BlockingQueue

5.2.1 – ConcurrentHashMap

ConcurrentHashMap is a replacement for synchronized hash-based Map implementations.

It uses a different locking strategy, lock striping,that offers better concurrency and scalability.

Further the iterators returned by ConcurrentHashMapare weakly consistent instead of fail-fast. The iterator can tolerate concurrent modification and may reflect modifications to the collection after the construction of the iterator without throw ConcurrentModificationException.

5.2.1 – … continue

Finally ConcurrentHashMap has so many advantages an so few disadvantages compared to Hashtable or synchronizedMap.

Replacing synchronized Map with ConcurrentHashMap generally results only better scalability.

Only if your application needs to lock the map for exclusive access, it is not an appropriate replacement.

5.2.2 – Additional atomic Map operations

Since a ConcurrentHashMap cannot be locked for exclusive access, there are common compound operations implemented as atomic operations and specified by ConcurrentMap interface.

5.2.3 – CopyOnWriteArrayList

CopyOnWriteArrayList is a concurrent replacement for synchronized List that offer better concurrency and eliminates the need to look or copy the collection during iteration.

All mutative operations (add, set, and so on) are implemented by making a fresh copy of the basilar array.

Obviously there is cost to copying the backing array every time the collection is modified. These collections are reasonable to use only when iteration is far more common than modification.

5.3 – Blocking queues and the producer-consumer pattern

A Queue that additionally supports operations that wait for the queue to become non-empty when retrieving an element, and wait for space to become available when storing an element in the queue.

One person washes the dishes and places them in the dish rack, and the other person retrieves the dishes from the rack and dries them.

Blocking queue

Producer-consumer pattern (example)

5.3 – … continue (1)

BlockingQueue simplifies the implementation of producer consumer pattern. Why?:

•producers don’t need to know anything about identity or number of consumers

•consumers don’t need to know who the producers are and where the work came from

•take() and put(obj) block the current thread (producer or consumer) indefinitely until the operation can succeed.

5.3 – … continue (2)

offer method: returns a failure status if the item cannot be enqueued.

With this method is possible to create more flexible policies for dealing with overload.

Rule:

Bounded queues are a powerful resource management tool for building reliable applications: they make your program more robust to overload by throttling activities that menace to produce more work than can be handled.

5.3 – … continue (3)

•LinkedBlockingQueue, ArrayBlockingQueue are FIFO queues.

•PriorityBlockingQueue is priority oriented queue.

•SynchronousQueue is a BlockingQueue in which each put must wait for a take, and vice versa. It maintains no storage space for queued elements.

SynchronousQueue are generally suitable only when there are enough consumers that there nearly always will be one ready to take the handoff.

Several implementation of BlockingQueue

5.3.1 – Example: desktop searchDo you know Google Desktop?

We will see an example of program that is open to decomposition into producers and consumers. It is an agent that scans local drives for documents (crawler) and indexes them for later searching (indexer).

example code: 2Pages 91 and 92

5.3.2 – Serial thread confinementA thread-confined object is owned by a single thread.

That ownership can be transferred by publishing object safely where only one other thread will gain access to it.

The new owner may modify it freely since it has exclusive access.

Producer-consumer designs and blocking queues facilitate serial thread confinement for handing off ownership of objects from producers to consumers.

5.3.3 – Deques and work stealingNew collection type in Java 6: Deque

It is a double-ended queue that allows efficient insertion and removal from both the head & the tail.

Deques lend themselves to a related work stealing pattern wherein every consumer has its own deque(the threads don’t share a queue).

If a consumer exhaust the work in its own deque, it can steal work from the tail (not from the head, further reducing contention) of someone else’s deque.

“deck”is the right

pronunciation

5.3.3 – … continueWhen can you use work stealing pattern ?

It is well suited to problems in which consumers are also producers. When performing a unit of work is likely to result in the identification of more work.

Example: processing a page in a web crawler usually results in the identification of new page to be crawled (consumer = producer).

5.4 – Blocking and interruptible methods

Threads may block! When? Waiting for I/O completion, for acquire a lock, for wake up from sleep, …

A lot of methods throw the checked InterruptedException (put, take, sleep, … ). When this happens the method is telling you that it is a blocking method, and if it is interrupted, it will make an attempt to stop blocking early.

Thread provides the interrupt method for interrupting a thread and querying whether a thread has been interrupted or not.

5.4 – … continue (1)Propagate the InterruptedException

Blocking methods

The easiest strategy is to throw InterruptedException yourself.

5.4 – … continue (2)Restore the interrupt

You must catch InterruptedException and restore the interrupted status by calling interrupt on the current thread.

5.4 – … continue (3)The worst approach ? swallow the interrupt

This deprives the opportunity of act on the interrupt.

Only when you are extending Thread and control all the code higher up on the call stack, you can swallow the interrupt.

5.5 – SyncronizersSynchronizer: Any object that coordinates the control flow of threads based on its state.

Examples: blocking queue, latches, semaphores, barriers, … and you can also create your own synchronizers.

Common properties:

•encapsulate state that determines whether threads arriving at the synchronizer should be permitted to pass or forced to wait, and provide methods to manipulate that state.

•provide methods to wait efficiently for the synchronizer to enter the desired state.

5.5.1 – LatchesThis synchronizer can delay the progress of threads until reaches its terminal state.

When does it use latch?

•ensuring that a computation does not proceed until resources it needs have been initialized.

•ensuring that a service does not start until other services on witch it depends have started.

•waiting until all the parties involved in an activity are ready to proceed.

5.5.1 – … continueFor put in practice this situations we can use the CountDownLatch implementation.

The latch state consists in a counter initialized to a positive number, representing the number of events to wait for.

We will see an example wherein there are two common uses for latches.

example code: 3Page 96

5.5.2 – FutureTaskThis synchronizer also acts like a latch.

A computation represented by a FutureTask is implemented with a Callable (the result-bearing equivalent of Runnable) and can be in one of 3 states:

•waiting to run

•running

•completed (when FutureTask is in this state, it stays here forever)

5.5.2 – … continue (1)The behavior of Future.get is linked on the state of the task. If this state is:

•completed – get returns the result immediately

•otherwise – get blocks until the task transitions to the completed states

FutureTask is used by the Executor framework to represent asynchronous tasks.

It can also be used to represent any potentially lengthy computation that can be started before the results are needed.

5.5.2 – … continue (2)In the next example we use FutureTask to perform an expansive computation by reducing the time you would have to wait later when you actually need the results.

example code: 4Pages 97 and 98

5.5.3 – SemaphoresCounting Semaphores are used:

•to control the number of activities that can access a certain resources

•perform a given action at the same time

A semaphore maintains a set of virtual permits. Activities can acquire permits and release them then they are done.

2 main methods:

•acquire: blocks until one permit is available

•release: returns a permit to the semaphore

5.5.3 – … continueThe most common use of semaphores is implementing resource pools (ex.: database connection pools).

Initializing a Semaphore to the pool size (through constructor), you can acquire a permit before trying to get a resource from the pool, and release the permit after putting a resource back in the pool.

Also, you can use semaphores to transform any collection into a blocking bounded collection.

example code: 5Page 100

5.5.4 – BarriersThese synchronizers are similar to latches in that they block a group of threads until some events has occurred.

The difference is that whit a barrier, all the threads must come together at a barrier point at the same time in order to proceed. So, barriers waiting for barriers waiting for other threads.other threads.

CyclicBarrier allows a fixed number of parties to meet repeatedly at barrier point. It is useful in parallel iterative algorithms that spit a problem in independent sub-problems.

5.5.4 – … continueRole of await method: blocks until all threads have reached the barrier point. Barrier is reset and threads are released only if all threads have reached the barrier point. In successful situation await return a unique arrival index for each thread.

BrokenBarrierException: It is arose if a thread blocked in await is interrupted or a call to await times out.

We will see an example wherein a barrier is used to compute a cellular automata simulation. (http://en.wikipedia.org/wiki/Cellular_automata)

Page 102

5.6 – Building an efficient, scalable result cache

Caching activities are a common use of server applications. They reuse the results of a previous computation for:

•reduce latency

•increase throughput

We will see the develop of an efficient and scalable result cache for a computationally expansive function.

5.6 – … continue (1)Development - Obvious approach

A simple HashMap!

5.6 – … continue (2)Memoization technique: Computable wrapper that remembers the results of previous computations and encapsulates the caching process

Makes HashMap Thread safe but only one thread at time can execute “compute” at all

- Checks whether the desired result is already cached

- If not, the result is computed and cached in the HashMap before returning

?

5.6 – … continue (3)Improvement – Memorizer2: replacing HashMap with ConcurrentHasMap.

This is thread-safe, so there is no need to synchronize when accessing the backing Map. Multiple threads can use Memorizer2 concurrently.

Still problems: two thread can call compute at the same time and computing the same value.

Thread B don’t know that the computation on f(1) is in progress

5.6 – … continue (4)Improvement – Memorizer3: use of FutureTask

- checks to see if the appropriate calculation has been started. - If not, it creates a FutureTask, registers it in the Map, and starts the computation; - otherwise it waits for the result of the existing computation

5.6 – … continue (5)

Improvement – Memorizer: use putIfAbsent atomic method of ConcurrentMap.

Defect: 2 threads can call compute on the same value, because “if block”in compute is still a non atomic check-then-act.

Both threads see that the cache does not contain the desired value, and both start the computation.

Page 108

5.6 – … continue (6)Final Example: Now we have the instruments for Factorizing servlet that caches results using Memoizer

This method execute the factorization

Chapter 6

“Task Execution”

51

IntroTask: abstract, discrete unit of work.

Dividing the work of an application into tasks:

•simplifies program organization

•facilitates error recovery by providing natural transaction boundaries

•promotes concurrency by providing a natural structure for parallelizing work

6.1 – Executing tasks in threadsTasks are independent!

Independence: This property facilitate concurrency because independent tasks can be executed in parallel.

Choosing a good task boundaries and a sensible task execution policy, can help for build applications:•that exhibit good throughput and good responsiveness under normal load

•to support many users

•with a good graceful degradation in overload conditions

6.1.1 – Executing tasks sequentiallyAn individual client request is a natural choice of task boundary that most server applications offer.

The simplest possible policy for scheduling tasks within an application is to execute tasks sequentially in a single thread.

This approach is correct but it has poor performance. It can handle only one request at a time.Except some cases, server applications don’t use sequential processing.

6.1.2 – Explicitly creating threads for tasks

The New Approach: The creation of a new thread for servicing each request.

The main loop creates a new thread to process the request instead of processing it within the main thread.

6.1.2 – … continueConsequences:•New connections can be accepted before previous requests complete, improving responsiveness.•Tasks can be processed in parallel, enabling multiple requests to be serviced simultaneously. This may improve throughput.•Task-handling code must be thread-safe, because it may be invoked concurrently for multiple tasks.

6.1.3 – Disadvantages of unbounded thread creation

The previous approach has some practical disadvantages:•Thread lifecycle overhead. If requests are frequent and lightweight creating a new thread for each request can consume significant computing resources.•Resource consumption. Active threads consume system resources, especially memory. If you have enough threads to keep all the CPUs busy, creating more threads won't help and may even hurt.•Stability. There is a limit on how many threads can be created. When you hit this limit, the most likely result is an OutOfMemoryError. It is important to structure your program to avoid hitting this limit.

6.2 – The Executor frameworkBoth previous techniques have serious limitations.

The executor framework (based on Thread pools) simplifies management of task and thread lifecycles and provides a simple and flexible means for decoupling task submission from execution policy.

java.util.concurrent provides a flexible thread pool implementation as part of Executor framework.

Executor is primary abstraction for task execution.This interface forms the basis for a flexible and powerfulframework for asynchronous task execution

6.2.1 – Example: web server using Executor

Standard Executor implementation with fixed-size thread pool.

The submission of the request-handling task is decoupled from its execution using an Executor, and its behavior can be changed merely by substituting a different Executor implementation.

6.2.1 – … continueWe can make a modification to behave like ThreadPerTaskWebServer by substituting an Executor.

Or, merely, we can write an Executor that would make TaskExecutionWebServer behave like single-threaded version.

It execute each task synchronously before returning from execute

It create new thread for each request

6.2.2 – Execution policies

An execution policy specify a lot of things about task execution, including:•In which thread will tasks be executed and in what order?

•How many tasks may execute concurrently and may be queued pending execution?

•If a task has to be rejected because the system is overloaded, which task should be selected as the victim, and how should the application be notified?

•What actions should be taken before or after executing a task?

6.2.2 – … continue Rule:

Whenever you see code of the form:new Thread(runnable).start()

and you think you might at some point want a more flexible execution policy, seriously consider replacing it with the use of an Executor.

6.2.3 – Thread pools

Thread pools manage a uniform pool of worker threads like a work queue.

Life of worker thread:•request the next task from the work queue

•execute it

•go back to waiting for another task

The main advantage of executing tasks in pool threads:•reusing an existing thread instead of creating a new one

6.2.3 – … continue (1)There are thread pool implementation with useful predefined configurations:

•newFixedThreadPool: This pool creates threads as tasks are submitted, up to the maximum pool size, and then attempts to keep the pool size constant

•newCachedThreadPool: It has more flexibility to eliminate idle threads when the current size of the pool exceeds the demand for processing, and to add new threads when demand increases, but places no bounds on the size of the pool.

6.2.3 – … continue (2)

•newSingleThreadExecutor: It creates a single worker thread to process tasks, replacing it if it dies unexpectedly. Tasks are processed sequentially in an predetermine order.

•newScheduledThreadPool: A fixed-size thread pool that supports delayed and periodic task execution.

Advantages to use Executor (pool-based policy):

•It has a big effect on application stability and it also degrades more gracefully.

•It permits additional opportunities for tuning, management, monitoring, logging, error reporting,…

6.2.4 – Executor lifecycleHow shot down an Executor?Since Executors provide a service to applications, they are able to be shutdown as abruptly or gracefully. This is addressed by the ExecutorService, which implements Executor and adds a number of methods for lifecycle management (and some utility methods).

GracefullyAbruptly

6.2.4 – … continueThe lifecycle has 3 states:

•running (Executor services are initially created in this state)

•shutting down

•terminated

Example: LifecycleWebServer extends the previous web server with lifecycle support.

Page 122

6.2.5 – Delayed and periodic tasks

The Timer facility manages the execution of deferred ("run this task in 100 ms") and periodic ("run this task every 10 ms") tasks.

Disadvantages:•A Timer creates only a single thread for executing timer tasks.•A Timer behaves poorly if a TimerTask throws an unchecked exception. The timer thread doesn’t catch the exception, so an unchecked exception, thrown from a TimerTask, terminates the timer thread that will be never run (problem called “thread leakage”).

6.2.5 – … continueScheduledThreadPoolExecutor should be thought for solve these problems and as replacement for Timer.

We will see an example that show how Timer can make confusion and to fall in “thread leakage”problem.

Page 124example code: 6

6.3 – Finding exploitable parallelism

The Executor framework makes it easy specify an execution policy, but you have to be able to describe your task as a Runnable.

Sometimes good task boundaries are not obvious to describe. There may be exploitable parallelism within a single client request in server applications, as is sometimes the case in database servers.

We will see different versions of a part of a browser application (page-rendering portion).

6.3.1 – Example: sequential page renderer

1° step ? renderText: Renders the text elements first, leaving rectangular placeholders for the images. 2° step ? go back and downloading the images and drawing them into the associated placeholder

This approach may underutilize the CPU (waiting for I/O to complete) and also make user wait long time to see the finished page.

6.3.2 – Result-bearing tasks: Callable and Future

Rannable is used by Executor framework as its basis task representation.

Sometimes Callable is a better abstraction then Runnable (when: for postponed computation; ex.: fetching a resource over the network, computing a complicated function,… ). Why?The main entry point, “call”, will return a value and anticipates that it might throw an exception.

Executors includes methods for wrapping Runnablewith a Callable.

6.3.2 – … continue (1)The lifecycle of a task executed by Executor has 4 phases:•created•submitted•started•completed

Do you remember FutureTask? (slide 36)Future represents lifecycle of a task and have methods for:•test whether the task is completed or been cancelled•retrieve its result•cancel the task

During the task’s lifecycle we want

to be able to cancel a task.

6.3.2 – … continue (2)There are several ways to create a Future to describe a task:• submit a Runnable or a Callable to an executor and get back a Future that can be used to retrieve the result or cancel the task.•explicitly instantiate a FutureTask for a given Runnable or Callable•use a default implementation of newTaskFor in ThreadPool Executor•or override this last method

Default implementation

6.3.2 – … continue (3)Conclusions

Submitting a Runnable (or Callable) to an Executorconstitutes a safe publication of the Runnable (or Callable) from the submitting thread to the thread that will execute the task.

Also, setting the result value for a Futureconstitutes a safe publication of the result from the thread in which it was computed to any thread that retrieves it via get.

6.3.3 – Example: page rendererwith Future

Improvements for page renderer:Division in two task. The 1° for renders the text (CPU-bound), the 2° for downloads all the images (I/O-bound).In this way, FutureRenderer allows the text to be rendered concurrently with downloading the image data.

We will use Callable and Future for express the interaction between these two tasks.

Page 128

6.3.4 – Limitations of parallelizing heterogeneous tasks

Why in the last example we have a face in doubt?Trying to parallelize sequential heterogeneoustasks can be tricky.

Problems:•Assigning a different type of task to each worker does not scale well if several more people show up (It is necessary restructuring the division of labor).•The tasks may have disparate sizes. If you divide tasks A and B between two workers but A takes ten times as long as B, you've only speeded up the total process by 9%.

6.3.4 – … continueComing back to FutureRenderer, generally rendering the text is much faster than downloading the images. So, the resulting performance is not much different from the sequential version, but the code is a lot more complicated

Rule:

The real performance payoff of dividing a program's workload into tasks comes when there are a large number of independent, homogeneous tasks that can be processed concurrently.

6.3.5 – CompletionService: Executor meets BlockingQueue

A good way for retrieve results as they become available of a series of computations to submit to an Executor is completion service.

It combines the functionality of an Executor and a BlockingQueue. You can submit Callable tasks to it for execution and use the queue methods take and poll to retrieve completed results, packaged as Futures, as they become available.

6.3.5 – … continueStructure of ExecutorCompletionService•It implements CompletionService, delegating the computation to an Executor.•The constructor creates a BlockingQueue to hold the completed results.•FutureTask has a done method that is called when the computation completes. When a task is submitted, it is wrapped with a QueueingFuture (subclass of FutureTask)that overrides done method to place the result on the BlockingQueue.

6.3.6 – Example: page rendererwith CompletionService

Improvements with CompletionService:•short total runtime•improved responsiveness

From sequential download to parallel oneWe can create a separate task for downloading each image and execute them in a thread pool.

More dynamic and responsive user interfaceWe can fetching results from the CompletionService and rendering each image as soon as it is available.

Page 130

6.3.7 – Placing time limits on tasks

Some applications (ex.: site portal) have a time budget. In this case, you don’t want wait longer than the time budget to get an answer.

There is a timed version of Future.get that support this requirement. It returns as soon as the result is ready, but throws TimeoutException if the result is not ready within the timeout period.

6.3.7 – … continueFuture can also help for stop timed tasks that run out of time. •If a timed get completes with a TimeoutException, you can cancel the task through the Future. •If the task is written to be cancelable, it can be terminated early so as not to consume excessive resources.

We will see a typical application of a timed Future.get. It generates a composite web page that contains the requested content plus an advertisement fetched from an ad server.

Page 132

6.3.8 – Example: a travel reservations portal

User Input: travel dates, requirements.Portal Output: bids from a number of airlines, hotels or car rental companies.

Probably, fetching a bid might involve invoking a web service, consulting a database, performing an EDI transaction, … This can require a long time. So, usually it may be preferable to present only the information available within a given time budget and omit slow providers.

6.3.8 – … continue Each action for fetching a single bid could represent a task. In fact these action are independently.Smell of concurrency?

IDEA:•to create n tasks•submit them to a thread pool•retain the Futures•use a timed get to fetch each result sequentially via its Future

Page 134

An easy way?

We can use the timed version of

invokeAll to submit multiple tasks to an

ExecutorServiceand retrieve the

results.