Part 4 - Synchronization problems (1)
In this part we’ll cover how to solve some classical synchronization problems using threads and semaphores.
Dining Philosophers
To refresh our memory on the problem let’s cover it again:
The dining philosophers problem describes how to avoid deadlock (circular conditions).
We have five philosophers (threads) sitting at a dining table. A fork is between each adjacent pair of philosophers.
Each philosopher alternates between thinking (non-critical section) and eating (critical section). In order to eat, a philosopher needs to pick up both forks that lie to the right and left of the philosopher.
Since the forks are shared, there is a synchronization problem.
Solution
A good solution would allow:
- Having an arbitrary number of philosophers
- Deadlock freedom
- Starvation freedom
- Reasonable efficiency, it is possible to eat in parallel.
Let’s create an interface for the table.
interface Table {
// Philosopher at idx picks up both forks
void getForks(int idx);
// Philosopher at idx lies down both forks
void putForks(int k);
}
Let’s also represent each philosopher
class Philosopher {
int idx;
Table table;
void think() {
.
.
.
}
void eat() {
.
.
.
}
void run() {
while(true) {
think();
table.getForks();
eat();
table.putForks();
}
}
}
One possible solution is to number each fork and philosopher. That we can easily know what is right and left of each philosopher.
For example:
public int left(int k) {
return k;
}
public int right(int k) {
return (k + 1) % N;
}
To ensure mutual exclusion of the forks we can use semaphores.
We can even begin with using simple locks:
Lock[] forks = new Lock[N];
// To pick up fork i we do
forks[i].lock();
// To put down fork i we do
forks[i].unlock();
If we now implement a solution which makes all philosophers take up the left fork then the right fork.
We have a potential for deadlock, since if everyone picks up their left fork first. We have a circular waiting condition.
Instead, we implement a solution which ensures one philosopher picks up their right fork first, then left. By breaking the symmetry we break the deadlock.
public void getForks(int k) {
if (k == N) {
forks[right(k)].lock();
forks[left(k)].lock();
}
else {
forks[left(k)].lock();
forks[right(k)].lock();
}
}
Another strategy could be limiting the amount of philosophers that can sit at a table. If we pick M < N, we also ensure deadlock freedom.
We make each seat a semaphores like:
// M < N
Semaphore seats = new Semaphore(M);
public void getForks(int k) {
seats.down()
forks[left(k)].lock();
forks[right(k)].lock();
}
public void putForks(int k) {
forks[left(k)].lock();
forks[right(k)].lock();
seats.down()
}
One thing that might not be obvious at first glance is that these solutions also ensure starvation freedom. This is under the assumption that the locks/semaphores (and scheduling) are fair.
Producer-Consumer
Producers and consumers exchange items through a so called shared (asynchronous) buffer.
The producer (asynchronously) produces items and places them on the buffer. While the consumer (asynchronously) consumes (removes) the items from the buffer.
So it’s clear that the buffer needs to be shared. Therefore, we need a buffer that:
- Producers and consumers access the buffer in mutual exclusion
- Consumers are blocked when the buffer is empty (can’t consume emptiness)
- Producers are blocked when the buffer is full (can’t overproduce)
interface Buffer<T> {
// Add item onto buffer; If full block
void put(T item);
// Extract item from buffer; If empty do nothing
T get();
// Get current number of items in buffer
int count();
}
Other properties that we would like:
- Support an arbitrary number of producers and consumers
- Deadlock freedom
- Starvation freedom
Unbounded shared buffer
One possible solution is using one lock and one semaphore:
public class UnboundedBuffer<T> implements Buffer<T> {
// Any collection (list, set, ...)
Collection storage = ...;
// For exclusive access to buffer
Lock lock = new Lock();
// Number of items in buffer
Semaphore nItems = new Semaphore(0);
invariant {
storage.count() == nItems.count() +
at(nItems.up() in put, lock.lock and T item = storage.remove() in get);
}
public void put(T item) {
// Lock to gain access to buffer
lock.lock();
/* Critical Section */
storage.add(item);
// update nItems
nItems.up();
/* End of Critical Section */
// Release access to buffer to others
lock.unlock();
}
public T get() {
// Wait until there is items on the buffer
nItems.down();
lock.lock();
/* Critical Section */
// Retrieve item
T item =storage.remove();
/* End of Critical Section */
lock.unlock();
return item;
}
public int count() {
return nItems.count();
}
}
This solution ensures that all requirements are met. There are some LOC that you can swap around.
But this will do. We can write a solution that is a so-called, bounded shared buffer, that uses two semaphores.
One for keeping track of how many items there are in buffer. The other to know how many slots are left.
It’s essentially the same code.
Barriers
A barrier in a concurrent program, is a form of synchronization where there is a point in a program’s execution, where all threads in a group, need to reach before any of them are allowed to continue.
A simple example is when we only have 2 threads. We use two binary semaphores:
Semaphore[] done = {new Semaphore(0), new Semaphore(0)};
$t_0$’s code:
// Indicate that t0 is done
done[t0].up();
// Wait until t1 is done
done[t1].down();
$t_1$’s code:
// Indicate that t1 is done
done[t1].up();
// Wait until t0 is done
done[t0].down();
We could also do:
$t_0$’s code:
// Wait until t1 is done
done[t1].down();
// Indicate that t0 is done
done[t0].up();
$t_1$’s code:
// Indicate that t1 is done
done[t1].up();
// Wait until t0 is done
done[t0].down();
This works if $t_0$ does down before up, or symmetrically, if $t_1$ does the same.
This is however, less efficient, since, the last thread to reach the barrier has to yield for the other.
One thing that might not be obvious but, if we do:
$t_0$’s code:
// Wait until t1 is done
done[t1].down();
// Indicate that t0 is done
done[t0].up();
$t_1$’s code:
// Wait until t0 is done
done[t0].down();
// Indicate that t1 is done
done[t1].up();
We will have a deadlock if $t_0$ and $t_1$ both perform down before any up calls.
Let’s try to create a reusable, general barrier
Reusable Barriers
interface Barrier {
// Block until expect() threads have reached barrier
void wait();
// Number of threads expected at the barrier
int expect();
}
Let’s try a naive approach and see what happens:
public class NonBarrier1 implements Barrier {
int nDone = 0; // number of done threads
Semaphore open = new Semaphore(0);
final int n;
// initialize barrier for `n' threads
NonBarrier1(int n) {
this.n = n;
}
// number of threads expected at the barrier
int expect() {
return n;
}
public void wait() {
synchronized(this) {
nDone += 1;
}
if (nDone == n) {
// I'm the last arrived: All can go!
open.up();
}
// Proceed when possible
open.down()
// Let the next one go
open.up()
synchronized(this) {
nDone -= 1;
}
if (nDone == 0) {
// I'm the last through: Close barrier!
open.down();
}
}
}
This solution doesn’t work perfectly though. If the n threads wait at if(nDone == 0)
then more than one thread may try to close the barrier, which results in a deadlock.
The same goes for opening the barrier, if more than one thread tries to open the barrier,
it’s possible that some threads may be executing wait again before the barrier is closed again.
public class NonBarrier2 implements Barrier {
int nDone = 0; // number of done threads
Semaphore open = new Semaphore(0);
final int n;
// initialize barrier for `n' threads
NonBarrier1(int n) {
this.n = n;
}
// number of threads expected at the barrier
int expect() {
return n;
}
public void wait() {
synchronized(this) {
nDone += 1;
if (nDone == n) {
// I'm the last arrived: All can go!
open.up();
}
}
// Proceed when possible
open.down()
// Let the next one go
open.up()
synchronized(this) {
nDone -= 1;
if (nDone == 0) {
// I'm the last through: Close barrier!
open.down();
}
}
}
}
Now this solution fixes the past issues listed. This solution has a problem of its own though.
If we have a thread which is incredibly fast, it may now get ahead of the other (slower) threads. Even if we use strong semaphores, we cannot prevent this. This is because this happens due to the last thread leaves the gate open.
To fix this we use a kind of two gate mechanic.
public class SemaphoreBarrier implements Barrier {
int nDone = 0; // number of done threads
Semaphore gate1 = new Semaphore(0);// first gate
Semaphore gate2 = new Semaphore(1);// second gate
final int n;
// initialize barrier for `n' threads
SemaphoreBarrier(int n) {
this.n = n;
}
// number of threads expected at the barrier
int expect() {
return n;
}
public void wait() {
approach(); leave();
}
void approach() {
synchronized (this) {
nDone += 1; // arrived
if (nDone == n) { // if last in:
gate1.up(); // open gate1
gate2.down(); // close gate2
}
}
gate1.down(); // pass gate1
gate1.up(); // let next pass
}
void leave() {
synchronized (this) {
nDone -= 1; // going out
if (nDone == 0) { // if last out:
gate2.up(); // open gate2
gate1.down(); // close gate1
}
}
gate2.down(); // pass gate2
gate2.up(); // let next pass
}
This solution is what a good, reusable barrier should look like.
Readers-Writers
A problem which we can encounter every day, say we have a board displaying data. Readers and writers need to access this board to, read and write.
We need a board which does:
-
Multiple readers can operate concurrently
-
Each writer has exclusive access
- Meaning that we have the invariant: # of writers = 0 OR (# of writers == 1 AND # of Readers == 0)
A naive approach would be:
public class SyncBoard<T> implements Board<T> {
int nReaders = 0;
// For exclusive access to nReaders
Lock lock = new Lock();
// 1 if and only if no active threads
Semaphore empty = new Semaphore(1);
T message;
public T read() {
// Lock to update nReaders
lock.lock();
if(nReaders == 0) {
// If first reader, set not empty
empty.down();
}
// Update nReaders
nReaders += 1
// Release nReaders
lock.unlock();
// Read (critical section)
T msg = message;
// Aquire nReaders
lock.lock();
nReaders -= 1;
if(nReaders == 0) {
// If # of readers empty, set empty
empty.up();
}
// Releaes nReaders
lock.unlock();
return msg;
}
public void write(T msg) {
// Get exclusive access
empty.down();
message = msg;
// Release board
empty.up();
}
}
This solution ensures almost everything we want, the only thing missing is that writers can now starve.
If there is always one reader active, writers will not be able to access the board at all.
public class FairBoard<T> extends SyncBoard<T> {
Semaphore baton = new Semaphore(1, true);
public T read() {
// Wait until you get baton
baton.down();
// Release a waiting thread
baton.up();
// Read as in SyncBoard
return super.read();
}
public void write(T msg) {
// Wait until you get baton
baton.down();
// Write as in SyncBoard
super.write(msg);
// Release
baton.up();
}
}
This will ensure that readers and writers have the same priority.
Summary
A looong part, but as we can see, all solution uses locks and semaphores. When writing solutions to concurrent programs, we will always need to use these. To lock and gain exclusive access.