1 Threads and Locks
1.1 Java Threads
concept of threads as lower-level building blocks for concurrent programs. A unique aspect of Java compared to prior mainstream programming languages is that Java included the notions of threads (as instances of the java.lang.Thread class) in its language definition right from the start.
When an instance of Thread is created (via a new operation), it does not start executing right away; instead, it can only start executing when its start() method is invoked. The statement or computation to be executed by the thread is specified as a parameter to the constructor.
The Thread class also includes a wait operation in the form of a join() method. If thread t0 performs a t1.join() call, thread t0 will be forced to wait until thread \mathtt {t1}t1 completes, after which point it can safely access any values computed by thread t1. Since there is no restriction on which thread can perform a join on which other thread, it is possible for a programmer to erroneously create a deadlock cycle with join operations. (A deadlock occurs when two threads wait for each other indefinitely, so that neither can make any progress.)
1.2 Structured Locks
Structured locks can be used to enforce mutual exclusion and avoid data races, as illustrated by the incr() method in the A.count example, and the insert() and remove() methods in the the Buffer example. A major benefit of structured locks is that their acquire and release operations are implicit, since these operations are automatically performed by the Java runtime environment when entering and exiting the scope of a synchronized statement or method, even if an exception is thrown in the middle.
We also learned about wait() and notify() operations that can be used to block and resume threads that need to wait for specific conditions. For example, a producer thread performing an insert() operation on a bounded buffer can call wait() when the buffer is full, so that it is only unblocked when a consumer thread performing a remove() operation calls notify(). Likewise, a consumer thread performing a remove() operation on a bounded buffer can call wait() when the buffer is empty, so that it is only unblocked when a producer thread performing an insert() operation calls notify(). Structured locks are also referred to as intrinsic locks or monitors.
1.3 Unstructured Locks
In this lecture, we introduced unstructured locks (which can be obtained in Java by creating instances of ReentrantLock()), and used three examples to demonstrate their generality relative to structured locks. The first example showed how explicit lock() and unlock() operations on unstructured locks can be used to support a hand-over-hand locking pattern that implements a non-nested pairing of lock/unlock operations which cannot be achieved with synchronized statements/methods. The second example showed how the tryLock() operations in unstructured locks can enable a thread to check the availability of a lock, and thereby acquire it if it is available or do something else if it is not. The third example illustrated the value of read-write locks (which can be obtained in Java by creating instances of ReentrantReadWriteLock()), whereby multiple threads are permitted to acquire a lock L in “read mode”, L.readLock().lock(), but only one thread is permitted to acquire the lock in “write mode”, L.writeLock().lock().
However, it is also important to remember that the generality and power of unstructured locks is accompanied by an extra responsibility on the part of the programmer, e.g., ensuring that calls to unlock() are not forgotten, even in the presence of exceptions.
1.4 Liveness and Progress Guarantees
In this lecture, we studied three ways in which a parallel program may enter a state in which it stops making forward progress. For sequential programs, an “infinite loop” is a common way for a program to stop making forward progress, but there are other ways to obtain an absence of progress in a parallel program. The first is deadlock, in which all threads are blocked indefinitely, thereby preventing any forward progress. The second is livelock, in which all threads repeatedly perform an interaction that prevents forward progress, e.g., an infinite “loop” of repeating lock acquire/release patterns. The third is starvation, in which at least one thread is prevented from making any forward progress.
The term “liveness” refers to a progress guarantee. The three progress guarantees that correspond to the absence of the conditions listed above are deadlock freedom, livelock freedom, and starvation freedom.
1.5 Dining Philosophers Problem
In this lecture, we studied a classical concurrent programming example that is referred to as the Dining Philosophers Problem. In this problem, there are five threads, each of which models a “philosopher” that repeatedly performs a sequence of actions which include think, pick up chopsticks, eat, and put down chopsticks.
First, we examined a solution to this problem using structured locks, and demonstrated how this solution could lead to a deadlock scenario (but not livelock). Second, we examined a solution using unstructured locks with tryLock() and unlock() operations that never block, and demonstrated how this solution could lead to a livelock scenario (but not deadlock). Finally, we observed how a simple modification to the first solution with structured locks, in which one philosopher picks up their right chopstick and their left, while the others pick up their left chopstick first and then their right, can guarantee an absence of deadlock.
1.6 Mini Project
Your main goals for this assignment are as follows:
- Inside CoarseLists.java, implement the CoarseList class using a ReentrantLock object to protect the add, remove, and contains methods from concurrent accesses. You should refer to SyncList.java as a guide for placing synchronization and understanding the list management logic.
- Inside CoarseLists.java, implement the RWCoarseList class using a ReentrantReadWriteLock object to protect the add, remove, and contains methods from concurrent accesses. You should refer to SyncList.java as a guide for placing synchronization and understanding the list management logic.
package edu.coursera.concurrent;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* Wrapper class for two lock-based concurrent list implementations.
*/
public final class CoarseLists {
/**
* An implementation of the ListSet interface that uses Java locks to
* protect against concurrent accesses.
*
* TODO Implement the add, remove, and contains methods below to support
* correct, concurrent access to this list. Use a Java ReentrantLock object
* to protect against those concurrent accesses. You may refer to
* SyncList.java for help understanding the list management logic, and for
* guidance in understanding where to place lock-based synchronization.
*/
public static final class CoarseList extends ListSet {
/*
* TODO Declare a lock for this class to be used in implementing the
* concurrent add, remove, and contains methods below.
*/
private final ReentrantLock lock = new ReentrantLock();
/**
* Default constructor.
*/
public CoarseList() {
super();
}
/**
* {@inheritDoc}
*
* TODO Use a lock to protect against concurrent access.
*/
@Override
boolean add(final Integer object) {
try {
lock.lock();
Entry pred = this.head;
Entry curr = pred.next;
while (curr.object.compareTo(object) < 0) {
pred = curr;
curr = curr.next;
}
if (object.equals(curr.object)) {
return false;
} else {
final Entry entry = new Entry(object);
entry.next = curr;
pred.next = entry;
return true;
}
} finally {
lock.unlock();
}
}
/**
* {@inheritDoc}
*
* TODO Use a lock to protect against concurrent access.
*/
@Override
boolean remove(final Integer object) {
try {
lock.lock();
Entry pred = this.head;
Entry curr = pred.next;
while (curr.object.compareTo(object) < 0) {
pred = curr;
curr = curr.next;
}
if (object.equals(curr.object)) {
pred.next = curr.next;
return true;
} else {
return false;
}
} finally {
lock.unlock();
}
}
/**
* {@inheritDoc}
*
* TODO Use a lock to protect against concurrent access.
*/
@Override
boolean contains(final Integer object) {
try {
lock.lock();
Entry pred = this.head;
Entry curr = pred.next;
while (curr.object.compareTo(object) < 0) {
pred = curr;
curr = curr.next;
}
return object.equals(curr.object);
} finally {
lock.unlock();
}
}
}
/**
* An implementation of the ListSet interface that uses Java read-write
* locks to protect against concurrent accesses.
*
* TODO Implement the add, remove, and contains methods below to support
* correct, concurrent access to this list. Use a Java
* ReentrantReadWriteLock object to protect against those concurrent
* accesses. You may refer to SyncList.java for help understanding the list
* management logic, and for guidance in understanding where to place
* lock-based synchronization.
*/
public static final class RWCoarseList extends ListSet {
/*
* TODO Declare a read-write lock for this class to be used in
* implementing the concurrent add, remove, and contains methods below.
*/
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
/**
* Default constructor.
*/
public RWCoarseList() {
super();
}
/**
* {@inheritDoc}
*
* TODO Use a read-write lock to protect against concurrent access.
*/
@Override
boolean add(final Integer object) {
try {
lock.writeLock().lock();
Entry pred = this.head;
Entry curr = pred.next;
while (curr.object.compareTo(object) < 0) {
pred = curr;
curr = curr.next;
}
if (object.equals(curr.object)) {
return false;
} else {
final Entry entry = new Entry(object);
entry.next = curr;
pred.next = entry;
return true;
}
} finally {
lock.writeLock().unlock();
}
}
/**
* {@inheritDoc}
*
* TODO Use a read-write lock to protect against concurrent access.
*/
@Override
boolean remove(final Integer object) {
try {
lock.writeLock().lock();
Entry pred = this.head;
Entry curr = pred.next;
while (curr.object.compareTo(object) < 0) {
pred = curr;
curr = curr.next;
}
if (object.equals(curr.object)) {
pred.next = curr.next;
return true;
} else {
return false;
}
} finally {
lock.writeLock().unlock();
}
}
/**
* {@inheritDoc}
*
* TODO Use a read-write lock to protect against concurrent access.
*/
@Override
boolean contains(final Integer object) {
try {
lock.readLock().lock();
Entry pred = this.head;
Entry curr = pred.next;
while (curr.object.compareTo(object) < 0) {
pred = curr;
curr = curr.next;
}
return object.equals(curr.object);
} finally {
lock.readLock().unlock();
}
}
}
}
2 Critical Sections and Isolation
2.1 Critical Sections
In this lecture, we learned how critical sections and the isolated construct can help concurrent threads manage their accesses to shared resources, at a higher level than just using locks. When programming with threads, it is well known that the following situation is defined to be a data race error — when two accesses on the same shared location can potentially execute in parallel, with least one access being a write. However, there are many cases in practice when two tasks may legitimately need to perform concurrent accesses to shared locations, as in the bank transfer example.
With critical sections, two blocks of code that are marked as isolated, say A and B, are guaranteed to be executed in mutual exclusion with A executing before B or vice versa. With the use of isolated constructs, it is impossible for the bank transfer example to end up in an inconsistent state because all the reads and writes for one isolated section must complete before the start of another isolated construct. Thus, the parallel program will see the effect of one isolated section completely before another isolated section can start.
2.2 Object-Based Isolation
In this lecture, we studied object-based isolation, which generalizes the isolated construct and relates to the classical concept of monitors. The fundamental idea behind object-based isolation is that an isolated construct can be extended with a set of objects that indicate the scope of isolation, by using the following rules: if two isolated constructs have an empty intersection in their object sets they can execute in parallel, otherwise they must execute in mutual exclusion. We observed that implementing this capability can be very challenging with locks because a correct implementation must enforce the correct levels of mutual exclusion without entering into deadlock or livelock states. The linked-list example showed how the object set for a delete() method can be defined as consisting of three objects — the current, previous, and next objects in the list, and that this object set is sufficient to safely enable parallelism across multiple calls to delete(). The Java code sketch to achieve this object-based isolation using the PCDP library is as follows:
isolated(cur, cur.prev, cur.next, () -> {
. . . // Body of object-based isolated construct
});
The relationship between object-based isolation and monitors is that all methods in a monitor object, M1, are executed as object-based isolated constructs with a singleton object set, {M1}. Similarly, all methods in a monitor object, M2, are executed as object-based isolated constructs with a singleton object set, {M2} which has an empty intersection with {M1}.
2.3 Spanning Tree Example
In this lecture, we learned how to use object-based isolation to create a parallel algorithm to compute spanning trees for an undirected graph. Recall that a spanning tree specifies a subset of edges in the graph that form a tree (no cycles), and connect all vertices in the graph. A standard recursive method for creating a spanning tree is to perform a depth-first traversal of the graph (the Compute(v) function in our example), making the current vertex a parent of all its neighbors that don’t already have a parent assigned in the tree (the MakeParent(v, c) function in the example).
The approach described in this lecture to parallelize the spanning tree computation executes recursive Compute(c) method calls in parallel for all neighbors, c, of the current vertex, v. Object-based isolation helps avoid a data race in the MakeParent(v,c) method, when two parallel threads might attempt to call MakeParent(v1, c) and MakeParent(v2, c) on the same vertex c at the same time. In this example, the role of object-based isolation is to ensure that all calls to MakeParent(v,c) with the same c value must execute the object-based isolated statement in mutual exclusion, whereas calls with different values of c can proceed in parallel.
2.4 Atomic Variables
In this lecture, we studied Atomic Variables, an important special case of object-based isolation which can be very efficiently implemented on modern computer systems. In the example given in the lecture, we have multiple threads processing an array, each using object-based isolation to safely increment a shared object, cur, to compute an index j which can then be used by the thread to access a thread-specific element of the array.
However, instead of using object-based isolation, we can declare the index cur to be an Atomic Integer variable and use an atomic operation called getAndAdd() to atomically read the current value of cur and increment its value by 1. Thus, j=cur.getAndAdd(1) has the same semantics as isolated (cur) {j=cur;cur=cur+1; } but is implemented much more efficiently using hardware support on today’s machines.
Another example that we studied in the lecture concerns Atomic Reference variables, which are reference variables that can be atomically read and modified using methods such as compareAndSet(). If we have an atomic reference ref, then the call to ref.compareAndSet(expected, new) will compare the value of ref to expected, and if they are the same, set the value of ref to new and return true. This all occurs in one atomic operation that cannot be interrupted by any other methods invoked on the ref object. If ref and expected have different values, compareAndSet() will not modify anything and will simply return false.
2.5 Read-Write Isolation
In this lecture we discussed Read-Write Isolation, which is a refinement of object-based isolation, and is a higher-level abstraction of the read-write locks studied earlier as part of Unstructured Locks. The main idea behind read-write isolation is to separate read accesses to shared objects from write accesses. This approach enables two threads that only read shared objects to freely execute in parallel since they are not modifying any shared objects. The need for mutual exclusion only arises when one or more threads attempt to enter an isolated section with write access to a shared object.
This approach exposes more concurrency than object-based isolation since it allows read accesses to be executed in parallel. In the doubly-linked list example from our lecture, when deleting an object cur from the list by calling delete(cur), we can replace object-based isolation on cur with read-only isolation, since deleting an object does not modify the object being deleted; only the previous and next objects in the list need to be modified.
2.6 Mini Project
Your main goals for this assignment are as follows:
- Inside BankTransactionsUsingObjectIsolation.java, implement the issueTransfer method using object-based isolation to protect against concurrent accesses to the source and destination bank accounts.
package edu.coursera.concurrent;
import static edu.rice.pcdp.PCDP.isolated;
/**
* A thread-safe transaction implementation using object-based isolation.
*/
public final class BankTransactionsUsingObjectIsolation
extends ThreadSafeBankTransaction {
/**
* {@inheritDoc}
*/
@Override
public void issueTransfer(final int amount, final Account src,
final Account dst) {
isolated(src, dst, () -> src.performTransfer(amount, dst));
}
}
3 Actors
3.1 Actor Model
In this lecture, we introduced the Actor Model as an even higher level of concurrency control than locks or isolated sections. One limitation of locks, and even isolated sections, is that, while many threads might correctly control the access to a shared object (e.g., by using object-based isolation) it only takes one thread that accesses the object directly to create subtle and hard-to-discover concurrency errors. The Actor model avoids this problem by forcing all accesses to an object to be isolated by default. The object is part of the local state of an actor, and cannot be accessed directly by any other actor.
An Actor consists of a Mailbox, a set of Methods, and Local State. The Actor model is reactive, in that actors can only execute methods in response to messages; these methods can read/write local state and/or send messages to other actors. Thus, the only way to modify an object in a pure actor model is to send messages to the actor that owns that object as part of its local state. In general, messages sent to actors from different actors can be arbitrarily reordered in the system. However, in many actor models, messages sent between the same pair of actors preserve the order in which they are sent.
3.2 Actor Examples
In this lecture, we further studied the Actor Model through two simple examples of using actors to implement well-known concurrent programming patterns. The PrintActor in our first example processes simple String messages by printing them. If an EXIT message is sent, then the PrintActor completes its current computation and exits. As a reminder, we assume that messages sent between the same pair of actors preserve the order in which they are sent.
In the second example, we created an actor pipeline, in which one actor checks the incoming messages and only forwards the ones that are in lower case. The second actor processes the lowercase messages and only forwards the ones that are of even length. This example illustrates the power of the actor model, as this concurrent system would be much more difficult to implement using threads, for example, since much care would have to be taken on how to implement a shared mailbox for correct and efficient processing by parallel threads.
3.3 Sieve of Eratosthenes
Lecture Summary: In this lecture, we studied how to use actors to implement a pipelined variant of the Sieve of Eratosthenes algorithm for generating prime numbers. This example illustrates the power of the Actor Model, including dynamic creation of new actors during a computation.
To implement the Sieve of Eratosthenes, we first create an actor, Non-Mul-2, that receives (positive) natural numbers as input (up to some limit), and then filters out the numbers that are multiples of 2. After receiving a number that is not a multiple of 2 (in our case, the first would be 3), the Non-Mul-2 actor creates the next actor in the pipeline, Non-Mul-3, with the goal of discarding all the numbers that are multiples of 3. The Non-Mul-2 actor then forwards all non-multiples of 2 to the Non-Mul-3 actor. Similarly, this new actor will create the next actor in the pipeline, Non-Mul-5, with the goal of discarding all the numbers that are multiples of 5. The power of the Actor Model is reflected in the dynamic nature of this problem, where pieces of the computation (new actors) are created dynamically as needed.
A Java code sketch for the process() method for an actor responsible for filtering out multiples of the actor’s “local prime” in the Sieve of Eratosthenes is as follows:
public void process(final Object msg) {
int candidate = (Integer) msg;
// Check if the candidate is a non-multiple of the "local prime".
// For example, localPrime = 2 in the Non-Mul-2 actor
boolean nonMul = ((candidate % localPrime) != 0);
// nothing needs to be done if nonMul = false
if (nonMul) {
if (nextActor == null) {
. . . // create & start new actor with candidate as its local prime
}
else nextActor.send(msg); // forward message to next actor
}
} // process
3.4 Producer-Consumer Problem with Unbounded Buffer
In this lecture, we studied the producer-consumer pattern in concurrent programming which is used to solve the following classical problem: how can we safely coordinate accesses by multiple producer tasks, P1,P2,P3, … and multiple consumer tasks, C1,C2,C3, … to a shared buffer of unbounded size without giving up any concurrency? Part of the reason that this problem can be challenging is that we cannot assume any a priori knowledge about the rate at which different tasks produce and consume items in the buffer. While it is possible to solve this problem by using locks with wait-notify operations or by using object-based isolation, both approaches will require low-level concurrent programming techniques to ensure correctness and maximum performance. Instead, a more elegant solution can be achieved by using actors as follows.
The key idea behind any actor-based solution is to think of all objects involved in the concurrent program as actors, which in this case implies that producer tasks, consumer tasks, and the shared buffer should all be implemented as actors. The next step is to establish the communication protocols among the actors. A producer actor can simply send a message to the buffer actor whenever it has an item to produce. The protocol for consumer actors is a bit more complicated. Our solution requires a consumer actor to send a message to the buffer actor whenever it is ready to process an item. Thus, whenever the buffer actor receives a message from a producer, it knows which consumers are ready to process items and can forward the produced item to any one of them. Thus, with the actor model, all concurrent interactions involving the buffer can be encoded in messages, instead of using locks or isolated statements.
3.5 Producer-Consumer Problem with Bounded Buffer
A major simplification made in the previous lecture was to assume that the shared buffer used by producer and consumer tasks can be unbounded in size. However, in practice, it is also important to consider a more realistic version of the the producer-consumer problem in which the buffer has a bounded size. In fact, the classical producer-consumer problem statement usually assumes a bounded buffer by default. In this lecture, we studied how the actor-based solution to the unbounded buffer case can be extended to support a bounded buffer.
The main new challenge with bounding the size of the shared buffer is to ensure that producer tasks are not permitted to send items to the buffer when the buffer is full. Thus, the buffer actor needs to play a master role in the protocol by informing producer actors when they are permitted to send data. This is akin to the role played by the buffer/master actor with respect to consumer actors, even in the unbounded buffer case (in which the consumer actor informed the buffer actor when it is ready to consume an item). Now, the producer actor will only send data when requested to do so by the buffer actor. Though, this actor-based solution appears to be quite simple, it actually solves a classical problem that has been studied in advanced operating system classes for decades.
3.6 Mini Project
TODO: Inside SieveActor .java, implement the countPrimes method by using actor parallelism to implement a parallel version of the Sieve of Eratosthenes based on the descriptions in the lecture and demo videos from this week. A sample actor declaration is provided in SieveActor.java for you to fill in. Note that the Actor API shown in the demo video and lectures is slightly different to the one used in this assignment. In particular, it is a simplification that does not require the start and exit methods. For example, a start operation is implicitly performed when an actor is created with a new operation. A full description of the Actor APIs can be found in the PCDP Javadocs at https://habanero-rice.github.io/PCDP/.
package edu.coursera.concurrent;
import edu.rice.pcdp.Actor;
import edu.rice.pcdp.PCDP;
import edu.rice.pcdp.config.SystemProperty;
import edu.rice.pcdp.runtime.ActorMessageWrapper;
import edu.rice.pcdp.runtime.Runtime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
/**
* An actor-based implementation of the Sieve of Eratosthenes.
* <p>
* TODO Fill in the empty SieveActorActor actor class below and use it from
* countPrimes to determin the number of primes <= limit.
*/
public final class SieveActor extends Sieve {
/**
* {@inheritDoc}
* <p>
* TODO Use the SieveActorActor class to calculate the number of primes <=
* limit in parallel. You might consider how you can model the Sieve of
* Eratosthenes as a pipeline of actors, each corresponding to a single
* prime number.
*/
@Override
public int countPrimes(final int limit) {
int[] numPrimes = new int[1];
final SieveActorActor sieveActor = new SieveActorActor(2);
PCDP.finish(() -> {
for (int i = 3; i <= limit; i += 2) {
sieveActor.send(i);
}
sieveActor.send(0);
});
SieveActorActor loopActor = sieveActor;
while (loopActor != null) {
numPrimes[0] += loopActor.numLocalPrimes();
loopActor = loopActor.nextActor();
}
return numPrimes[0];
}
/**
* An actor class that helps implement the Sieve of Eratosthenes in
* parallel.
*/
public static final class SieveActorActor extends Actor {
private static int MAX_LOCAL_PRIMES = 1000;
private int[] localPrimes;
private int numLocalPrimes;
private SieveActorActor nextActor;
public SieveActorActor(int localPrime) {
this.localPrimes = new int[MAX_LOCAL_PRIMES];
this.localPrimes[0] = localPrime;
this.numLocalPrimes = 1;
this.nextActor = null;
}
public SieveActorActor nextActor() {
return nextActor;
}
public int numLocalPrimes() {
return numLocalPrimes;
}
/**
* Process a single message sent to this actor.
* <p>
* TODO complete this method.
*
* @param msg Received message
*/
@Override
public void process(final Object msg) {
final int candidate = (Integer) msg;
if (candidate <= 0) {
if (nextActor != null) {
nextActor.send(msg);
}
} else {
final boolean locallyPrime = isLocallyPrime(candidate);
if (locallyPrime) {
if (numLocalPrimes < MAX_LOCAL_PRIMES) {
localPrimes[numLocalPrimes] = candidate;
numLocalPrimes += 1;
} else if (nextActor == null) {
nextActor = new SieveActorActor(candidate);
} else {
nextActor.send(msg);
}
}
}
}
private boolean isLocallyPrime(final int candidate) {
for (int i = 0; i < numLocalPrimes; i++) {
if (candidate % localPrimes[i] == 0) {
return false;
}
}
return true;
}
}
}
4 Concurrent Data Structures
4.1 Optimistic Concurrency
In this lecture, we studied the optimistic concurrency pattern, which can be used to improve the performance of concurrent data structures. In practice, this pattern is most often used by experts who implement components of concurrent libraries, such as AtomicInteger and ConcurrentHashMap, but it is useful for all programmers to understand the underpinnings of this approach. As an example, we considered how the getAndAdd() method is typically implemented for a shared AtomicInteger object. The basic idea is to allow multiple threads to read the existing value of the shared object (curVal) without any synchronization, and also to compute its new value after the addition (newVal) without synchronization. These computations are performed optimistically under the assumption that no interference will occur with other threads during the period between reading curVal and computing newVal. However, it is necessary for each thread to confirm this assumption by using the compareAndSet() method as follows. (compareAndSet() is used as an important building block for optimistic concurrency because it is implemented very efficiently on many hardware platforms.)
The method call A.compareAndSet(curVal, newVal) invoked on AtomicInteger A checks that the value in A still equals curVal, and, if so, updates A’s value to newVal before returning true; otherwise, the method simply returns false without updating A. Further, the compareAndSet() method is guaranteed to be performed atomically, as if it was in an object-based isolated statement with respect to object A. Thus, if two threads, T1 and T2 call compareAndSet() with the same curVal that matches A’s current value, only one of them will succeed in updating A with their newVal. Furthermore, each thread will invoke an operation like compareAndSet() repeatedly in a loop until the operation succeeds. This approach is guaranteed to never result in a deadlock since there are no blocking operations. Also, since each call compareAndSet() is guaranteed to eventually succeed, there cannot be a livelock either. In general, so long as the contention on a single shared object like A is not high, the number of calls to compareAndSet() that return false will be very small, and the optimistic concurrency approach can perform much better in practice (but at the cost of more complex code logic) than using locks, isolation, or actors.
4.2 Concurrent Queues
In this lecture, we studied concurrent queues, an extension of the popular queue data structure to support concurrent accesses. The most common operations on a queue are enq(x), which enqueues object x at the end (tail) of the queue, and deq() which removes and returns the item at the start (head) of the queue. A correct implementation of a concurrent queue must ensure that calls to enq() and deq() maintain the correct semantics, even if the calls are invoked concurrently from different threads. While it is always possible to use locks, isolation, or actors to obtain correct but less efficient implementations of a concurrent queue, this lecture illustrated how an expert might implement a more efficient concurrent queue using the optimistic concurrency pattern.
A common approach for such an implementation is to replace an object reference like tail by an AtomicReference. Since the compareAndSet() method can also be invoked on AtomicReference objects, we can use it to support (for example) concurrent calls to enq() by identifying which calls to compareAndSet() succeeded, and repeating the calls that failed. This provides the basic recipe for more efficient implementations of enq() and deq(), as are typically developed by concurrency experts. A popular implementation of concurrent queues available in Java is java.util.concurent.ConcurrentLinkedQueue.
4.3 Linearizability
In this lecture, we studied an important correctness property of concurrent objects that is called Linearizability. A concurrent object is a data structure that is designed to support operations in parallel by multiple threads. The key question answered by linearizability is what return values are permissible when multiple threads perform these operations in parallel, taking into account what we know about the expected return values from those operations when they are performed sequentially. As an example, we considered two threads, T1 and T2, performing enq(x) and enq(y) operations in parallel on a shared concurrent queue data structure, and considered what values can be returned by a deq() operation performed by T2 after the call to enq(y). From the viewpoint of linearizability, it is possible for the deq() operation to return item x or item y.
One way to look at the definition of linearizability is as though you are a lawyer attempting to “defend” a friend who implemented a concurrent data structure, and that all you need to do to prove that your friend is “not guilty” (did not write a buggy implementation) is to show one scenario in which all the operations return values that would be consistent with a sequential execution by identifying logical moments of time at which the operations can be claimed to have taken effect. Thus, if deq() returned item x or item y you can claim that either scenario is plausible because we can reasonably assume that enq(x) took effect before enq(y), or vice versa. However, there is absolutely no plausible scenario in which the call to deq() can correctly return a code/exception to indicate that the queue is empty since at least enq(y) must have taken effect before the call to deq(). Thus, a goal for any implementation of a concurrent data structure is to ensure that all its executions are linearizable by using whatever combination of constructs (e.g., locks, isolated, actors, optimistic concurrency) is deemed appropriate to ensure correctness while giving the maximum performance.
4.4 Concurrent HashMap
In this lecture, we studied the ConcurrentHashMap data structure, which is available as part of the java.util.concurrent standard library in Java. A ConcurrentHashMap instance, chm, implements the Map interface, including the get(key) and put(key, value) operations. It also implements additional operations specified in the ConcurrentMap interface (which in turn extends the Map interface); one such operation is putIfAbsent(key, value). The motivation for using putIfAbsent() is to ensure that only one instance of key is inserted in chm, even if multiple threads attempt to insert the same key in parallel. Thus, the semantics of calls to get(), put(), and putIfAbsent() can all be specified by the theory of linearizability studied earlier. However, it is worth noting that there are also some aggregate operations, such as clear() and putAll(), that cannot safely be performed in parallel with put(), get() and putIfAbsent().
Motivated by the large number of concurrent data structures available in the java.util.concurrent library, this lecture advocates that, when possible, you use libraries such as ConcurrentHashMap rather than try to implement your own version.
4.5 Minimum Spanning Tree Example
In this lecture, we discussed how to apply concepts learned in this course to design a concurrent algorithm that solves the problem of finding a minimum-cost spanning tree (MST) for an undirected graph. It is well known that undirected graphs can be used to represent all kinds of networks, including roadways, train routes, and air routes. A spanning tree is a data structure that contains a subset of edges from the graph which connect all nodes in the graph without including a cycle. The cost of a spanning tree is computed as the sum of the weights of all edges in the tree.
The concurrent algorithm studied in this lecture builds on a well-known sequential algorithm that iteratively performs edge contraction operations, such that given a node N1 in the graph, GetMinEdge(N1) returns an edge adjacent to N1 with minimum cost for inclusion in the MST. If the minimum-cost edge is (N1,N2), the algorithm will attempt to combine nodes N1 and N2 in the graph and replace the pair by a single node, N3. To perform edge contractions in parallel, we have to look out for the case when two threads may collide on the same vertex. For example, even if two threads started with vertices A and D, they may both end up with C as the neighbor with the minimum cost edge. We must avoid a situation in which the algorithm tries to combine both A and C and D and C. One possible approach is to use unstructured locks with calls to tryLock() to perform the combining safely, but without creating the possibility of deadlock or livelock situations. A key challenge with calling tryLock() is that some fix-up is required if the call returns false. Finally, it also helps to use a concurrent queue data structure to keep track of nodes that are available for processing.
4.6 Mini Project
Your main goal for this assignment is to complete the computeBoruvka method at the top of ParBoruvka. The testing infrastructure will call computeBoruvka from a certain number of Java threads. You do not need to create any additional threads if you do not wish to (indeed, it is recommended that you do not). computeBoruvka is passed two objects:
-
nodesLoaded: A ConcurrentLinkedQueue object containing a list of all nodes in the input graph for which you are to compute a minimum spanning tree using Boruvka’s algorithm. Because this Queue
object is a ConcurrentLinkedQueue, it is safe to access nodesLoaded from multiple threads concurrently without additional synchronization. -
solution: A BoruvkaSolution object on which you will call BoruvkaSolution.setSolution once your parallel Boruvka implementation has collapsed the input graph down to a single component. You must only call setSolution once, and the testing infrastructure will use the provided result to verify your output.
Inside of ParBoruvka, there are two additional inner classes: ParComponent and ParEdge. ParComponent represents a single component in the graph. A component may be a singleton node, or it may be formed from collapsing multiple nodes into each other. A ParEdge represents an edge between two ParComponents. You may not change the signatures for the existing methods in ParComponent or ParEdge. However, you are free to modify their method bodies, to add new methods, or to add new fields. An efficient implementation will likely require modifications to ParComponent and/or ParEdge.
package edu.coursera.concurrent;
import com.sun.corba.se.impl.orbutil.concurrent.ReentrantMutex;
import edu.coursera.concurrent.AbstractBoruvka;
import edu.coursera.concurrent.SolutionToBoruvka;
import edu.coursera.concurrent.boruvka.Edge;
import edu.coursera.concurrent.boruvka.Component;
import edu.coursera.concurrent.boruvka.sequential.SeqComponent;
import java.util.Queue;
import java.util.List;
import java.util.ArrayList;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.StampedLock;
/**
* A parallel implementation of Boruvka's algorithm to compute a Minimum
* Spanning Tree.
*/
public final class ParBoruvka extends AbstractBoruvka<ParBoruvka.ParComponent> {
/**
* Constructor.
*/
public ParBoruvka() {
super();
}
/**
* {@inheritDoc}
*/
@Override
public void computeBoruvka(final Queue<ParComponent> nodesLoaded,
final SolutionToBoruvka<ParComponent> solution) {
final int numberThreads = 4;
final Thread[] threads = new Thread[numberThreads];
final ParComponent[] result = new ParComponent[1];
for (int i = 0; i < numberThreads; i++) {
threads[i] = new Thread(() -> {
ParComponent n;
while ((n = nodesLoaded.poll()) != null) {
if(!n.lock.tryLock()) {
continue;
}
if(n.isDead) {
n.lock.unlock();
continue;
}
final Edge<ParComponent> e = n.getMinEdge();
if(e == null) {
result[0] = n;
break;
}
final ParComponent other = e.getOther(n);
if(!other.lock.tryLock()) {
n.lock.unlock();
nodesLoaded.add(n);
continue;
}
if (other.isDead) {
other.lock.unlock();
n.lock.unlock();
nodesLoaded.add(n);
continue;
}
other.isDead = true;
n.merge(other, e.weight());
n.lock.unlock();
other.lock.unlock();
nodesLoaded.add(n);
}
});
threads[i].start();
}
for (int i = 0; i < numberThreads; i++) {
while (true) {
try {
threads[i].join();
break;
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
final ParComponent resultNode = result[0];
if(resultNode != null) {
solution.setSolution(resultNode);
}
}
/**
* ParComponent represents a single component in the graph. A component may
* be a singleton representing a single node in the graph, or may be the
* result of collapsing edges to form a component from multiple nodes.
*/
public static final class ParComponent extends Component<ParComponent> {
public final ReentrantLock lock = new ReentrantLock();
/**
* A unique identifier for this component in the graph that contains
* it.
*/
public final int nodeId;
/**
* List of edges attached to this component, sorted by weight from least
* to greatest.
*/
private List<Edge<ParComponent>> edges = new ArrayList<>();
/**
* The weight this component accounts for. A component gains weight when
* it is merged with another component across an edge with a certain
* weight.
*/
private double totalWeight = 0;
/**
* Number of edges that have been collapsed to create this component.
*/
private long totalEdges = 0;
/**
* Whether this component has already been collapsed into another
* component.
*/
public boolean isDead = false;
/**
* Constructor.
*
* @param setNodeId ID for this node.
*/
public ParComponent(final int setNodeId) {
super();
this.nodeId = setNodeId;
}
/**
* {@inheritDoc}
*/
@Override
public int nodeId() {
return nodeId;
}
/**
* {@inheritDoc}
*/
@Override
public double totalWeight() {
return totalWeight;
}
/**
* {@inheritDoc}
*/
@Override
public long totalEdges() {
return totalEdges;
}
/**
* {@inheritDoc}
*
* Edge is inserted in weight order, from least to greatest.
*/
public void addEdge(final Edge<ParComponent> e) {
int i = 0;
while (i < edges.size()) {
if (e.weight() < edges.get(i).weight()) {
break;
}
i++;
}
edges.add(i, e);
}
/**
* Get the edge with minimum weight from the sorted edge list.
*
* @return Edge with the smallest weight attached to this component.
*/
public Edge<ParComponent> getMinEdge() {
if (edges.size() == 0) {
return null;
}
return edges.get(0);
}
/**
* Merge two components together, connected by an edge with weight
* edgeWeight.
*
* @param other The other component to merge into this component.
* @param edgeWeight Weight of the edge connecting these components.
*/
public void merge(final ParComponent other, final double edgeWeight) {
totalWeight += other.totalWeight + edgeWeight;
totalEdges += other.totalEdges + 1;
final List<Edge<ParComponent>> newEdges = new ArrayList<>();
int i = 0;
int j = 0;
while (i + j < edges.size() + other.edges.size()) {
// Get rid of inter-component edges
while (i < edges.size()) {
final Edge<ParComponent> e = edges.get(i);
if ((e.fromComponent() != this
&& e.fromComponent() != other)
|| (e.toComponent() != this
&& e.toComponent() != other)) {
break;
}
i++;
}
while (j < other.edges.size()) {
final Edge<ParComponent> e = other.edges.get(j);
if ((e.fromComponent() != this
&& e.fromComponent() != other)
|| (e.toComponent() != this
&& e.toComponent() != other)) {
break;
}
j++;
}
if (j < other.edges.size() && (i >= edges.size()
|| edges.get(i).weight()
> other.edges.get(j).weight())) {
newEdges.add(other.edges.get(j++).replaceComponent(other,
this));
} else if (i < edges.size()) {
newEdges.add(edges.get(i++).replaceComponent(other, this));
}
}
other.edges.clear();
edges.clear();
edges = newEdges;
}
/**
* Test for equality based on node ID.
*
* @param o Object to compare against.
* @return true if they are the same component in the graph.
*/
@Override
public boolean equals(final Object o) {
if (this == o) {
return true;
}
if (!(o instanceof Component)) {
return false;
}
final Component component = (Component) o;
return component.nodeId() == nodeId;
}
/**
* Hash based on component node ID.
*
* @return Hash code.
*/
@Override
public int hashCode() {
return nodeId;
}
}
/* End ParComponent */
/**
* A ParEdge represents a weighted edge between two ParComponents.
*/
public static final class ParEdge extends Edge<ParComponent>
implements Comparable<Edge> {
/**
* Source component.
*/
protected ParComponent fromComponent;
/**
* Destination component.
*/
protected ParComponent toComponent;
/**
* Weight of this edge.
*/
public double weight;
/**
* Constructor.
*
* @param from From edge.
* @param to To edges.
* @param w Weight of this edge.
*/
public ParEdge(final ParComponent from, final ParComponent to,
final double w) {
fromComponent = from;
toComponent = to;
weight = w;
}
/**
* {@inheritDoc}
*/
@Override
public ParComponent fromComponent() {
return fromComponent;
}
/**
* {@inheritDoc}
*/
@Override
public ParComponent toComponent() {
return toComponent;
}
/**
* {@inheritDoc}
*/
@Override
public double weight() {
return weight;
}
/**
* {@inheritDoc}
*/
public ParComponent getOther(final ParComponent from) {
if (fromComponent == from) {
assert (toComponent != from);
return toComponent;
}
if (toComponent == from) {
assert (fromComponent != from);
return fromComponent;
}
assert (false);
return null;
}
/**
* {@inheritDoc}
*/
@Override
public int compareTo(final Edge e) {
if (e.weight() == weight) {
return 0;
} else if (weight < e.weight()) {
return -1;
} else {
return 1;
}
}
/**
* {@inheritDoc}
*/
public ParEdge replaceComponent(final ParComponent from,
final ParComponent to) {
if (fromComponent == from) {
fromComponent = to;
}
if (toComponent == from) {
toComponent = to;
}
return this;
}
}
}