In problem is a classical example of a multi-process

Process (computing)

In computing, a process is an instance of a computer program that is being executed. It contains the program code and its current activity. Depending on the operating system , a process may be made up of multiple threads of execution that execute instructions concurrently.A computer program is a...

 synchronization

Synchronization (computer science)

In computer science, synchronization refers to one of two distinct but related concepts: synchronization of processes, and synchronization of data. Process synchronization refers to the idea that multiple processes are to join up or handshake at a certain point, so as to reach an agreement or...

 problem. The problem describes two processes, the producer and the consumer, who share a common, fixed-size buffer

Buffer (computer science)

In computer science, a buffer is a region of a physical memory storage used to temporarily hold data while it is being moved from one place to another. Typically, the data is stored in a buffer as it is retrieved from an input device or just before it is sent to an output device...

 used as a queue. The producer's job is to generate a piece of data, put it into the buffer and start again. At the same time the consumer is consuming the data (i.e., removing it from the buffer) one piece at a time. The problem is to make sure that the producer won't try to add data into the buffer if it's full and that the consumer won't try to remove data from an empty buffer.

The solution for the producer is to either go to sleep or discard data if the buffer is full. The next time the consumer removes an item from the buffer, it notifies the producer who starts to fill the buffer again. In the same way, the consumer can go to sleep if it finds the buffer to be empty. The next time the producer puts data into the buffer, it wakes up the sleeping consumer. The solution can be reached by means of inter-process communication

Inter-process communication

In computing, Inter-process communication is a set of methods for the exchange of data among multiple threads in one or more processes. Processes may be running on one or more computers connected by a network. IPC methods are divided into methods for message passing, synchronization, shared...

, typically using semaphores

Semaphore (programming)

In computer science, a semaphore is a variable or abstract data type that provides a simple but useful abstraction for controlling access by multiple processes to a common resource in a parallel programming environment....

. An inadequate solution could result in a deadlock

Deadlock

A deadlock is a situation where in two or more competing actions are each waiting for the other to finish, and thus neither ever does. It is often seen in a paradox like the "chicken or the egg"...

 where both processes are waiting to be awakened. The problem can also be generalized to have multiple producers and consumers.

Inadequate implementation

This solution has a race condition

Race condition

A race condition or race hazard is a flaw in an electronic system or process whereby the output or result of the process is unexpectedly and critically dependent on the sequence or timing of other events...

. To solve the problem, a careless programmer might come up with a solution shown below. In the solution two library routines are used, sleep and wakeup. When sleep is called, the caller is blocked until another process wakes it up by using the wakeup routine. itemCount is the number of items in the buffer.


int itemCount = 0;

procedure producer {
while (true) {
item = produceItem;

if (itemCount

BUFFER_SIZE) {
sleep;
}

putItemIntoBuffer(item);
itemCount = itemCount + 1;

if (itemCount

1) {
wakeup(consumer);
}
}
}

procedure consumer {
while (true) {

if (itemCount

0) {
sleep;
}

item = removeItemFromBuffer;
itemCount = itemCount - 1;

if (itemCount

BUFFER_SIZE - 1) {
wakeup(producer);
}

consumeItem(item);
}
}


The problem with this solution is that it contains a race condition

Race condition

A race condition or race hazard is a flaw in an electronic system or process whereby the output or result of the process is unexpectedly and critically dependent on the sequence or timing of other events...

 that can lead into a deadlock. Consider the following scenario:

1. The consumer has just read the variable itemCount, noticed it's zero and is just about to move inside the if-block.
2. Just before calling sleep, the consumer is interrupted and the producer is resumed.
3. The producer creates an item, puts it into the buffer, and increases itemCount.
4. Because the buffer was empty prior to the last addition, the producer tries to wake up the consumer.
5. Unfortunately the consumer wasn't yet sleeping, and the wakeup call is lost. When the consumer resumes, it goes to sleep and will never be awakened again. This is because the consumer is only awakened by the producer when itemCount is equal to 1.
6. The producer will loop until the buffer is full, after which it will also go to sleep.

Since both processes will sleep forever, we have run into a deadlock. This solution therefore is unsatisfactory.

An alternative analysis is that if the programming language does not define the semantics of concurrent accesses to shared
variables (in this case itemCount) without use of synchronization, then the solution is unsatisfactory for that reason, without needing to explicitly demonstrate a race condition.

Using semaphores

Semaphores

Semaphore (programming)

In computer science, a semaphore is a variable or abstract data type that provides a simple but useful abstraction for controlling access by multiple processes to a common resource in a parallel programming environment....

 solve the problem of lost wakeup calls. In the solution below we use two semaphores, fillCount and emptyCount, to solve the problem. fillCount is the number of items to be read in the buffer, and emptyCount is the number of available spaces in the buffer where items could be written. fillCount is incremented and emptyCount decremented when a new item has been put into the buffer. If the producer tries to decrement emptyCount while its value is zero, the producer is put to sleep. The next time an item is consumed, emptyCount is incremented and the producer wakes up. The consumer works analogously.


semaphore fillCount = 0; // items produced
semaphore emptyCount = BUFFER_SIZE; // remaining space

procedure producer {
while (true) {
item = produceItem;
down(emptyCount);
putItemIntoBuffer(item);
up(fillCount);
}
}

procedure consumer {
while (true) {
down(fillCount);
item = removeItemFromBuffer;
up(emptyCount);
consumeItem(item);
}
}


The solution above works fine when there is only one producer and consumer. Unfortunately, with multiple producers or consumers this solution contains a serious race condition that could result in two or more processes reading or writing into the same slot at the same time. To understand how this is possible, imagine how the procedure putItemIntoBuffer can be implemented. It could contain two actions, one determining the next available slot and the other writing into it. If the procedure can be executed concurrently by multiple producers, then the following scenario is possible:

1. Two producers decrement emptyCount
2. One of the producers determines the next empty slot in the buffer
3. Second producer determines the next empty slot and gets the same result as the first producer
4. Both producers write into the same slot

To overcome this problem, we need a way to make sure that only one producer is executing putItemIntoBuffer at a time. In other words we need a way to execute a critical section

Critical section

In concurrent programming a critical section is a piece of code that accesses a shared resource that must not be concurrently accessed by more than one thread of execution. A critical section will usually terminate in fixed time, and a thread, task or process will have to wait a fixed time to...

 with mutual exclusion

Mutual exclusion

Mutual exclusion algorithms are used in concurrent programming to avoid the simultaneous use of a common resource, such as a global variable, by pieces of computer code called critical sections. A critical section is a piece of code in which a process or thread accesses a common resource...

. To accomplish this we use a binary semaphore called mutex. Since the value of a binary semaphore can be only either one or zero, only one process can be executing between down(mutex) and up(mutex). The solution for multiple producers and consumers is shown below.


semaphore mutex = 1;
semaphore fillCount = 0;
semaphore emptyCount = BUFFER_SIZE;

procedure producer {
while (true) {
item = produceItem;
down(emptyCount);
down(mutex);
putItemIntoBuffer(item);
up(mutex);
up(fillCount);
}
up(fillCount); //the consumer may not finish before the producer.
}

procedure consumer {
while (true) {
down(fillCount);
down(mutex);
item = removeItemFromBuffer;
up(mutex);
up(emptyCount);
consumeItem(item);
}
}


Notice that the order in which different semaphores are incremented or decremented is essential: changing the order might result in a deadlock.

Using monitors

The following pseudo code shows a solution to the producer-consumer problem using monitors

Monitor (synchronization)

In concurrent programming, a monitor is an object or module intended to be used safely by more than one thread. The defining characteristic of a monitor is that its methods are executed with mutual exclusion. That is, at each point in time, at most one thread may be executing any of its methods...

. Since mutual exclusion is implicit with monitors, no extra effort is necessary to protect the critical section. In other words, the solution shown below works with any number of producers and consumers without any modifications. It is also noteworthy that using monitors makes race conditions much less likely than when using semaphores.


monitor ProducerConsumer {
int itemCount
condition full;
condition empty;

procedure add(item) {
while (itemCount

BUFFER_SIZE) {
wait(full);
}

putItemIntoBuffer(item);
itemCount = itemCount + 1;

if (itemCount

1) {
notify(empty);
}
}
procedure remove {
while (itemCount

0) {
wait(empty);
}

item = removeItemFromBuffer;
itemCount = itemCount - 1;

if (itemCount

BUFFER_SIZE - 1) {
notify(full);
}

return item;
}
}

procedure producer {
while (true) {
item = produceItem
ProducerConsumer.add(item)
}
}

procedure consumer {
while (true) {
item = ProducerConsumer.remove
consumeItem
}
}


Note the use of while statements in the above code, both when testing if the buffer is full or empty. With multiple consumers, there is a race condition

Race condition

A race condition or race hazard is a flaw in an electronic system or process whereby the output or result of the process is unexpectedly and critically dependent on the sequence or timing of other events...

 where one consumer gets notified that an item has been put into the buffer but another consumer is already waiting on the monitor so removes it from the buffer instead. If the while was instead an if, too many items might be put into the buffer or a remove might be attempted on an empty buffer.

Without semaphores or monitors

The producer-consumer problem, particularly in the case of a single producer and single consumer, strongly relates to implementing a FIFO

FIFO

FIFO is an acronym for First In, First Out, an abstraction related to ways of organizing and manipulation of data relative to time and prioritization...

 or a communication channel. The producer-consumer pattern can provide highly efficient data communication without relying on semaphores, mutexes, or monitors for data transfer. Use of those primitives can give performance issues as they are expensive to implement. Channels and Fifo's are popular just because they avoid the need for end-to-end atomic synchronization. A basic example coded in C is shown below. Note that:

• Atomic read-modify-write access to shared variables is avoided: each of the two Count variables is updated by a single thread only.
• This example does not put threads to sleep which might be OK depending on system context. The sched_yield is just to behave nice and could be removed. Thread libraries typically require semaphores or condition variables to control the sleep/wakeup of threads. In a multi-processor environment, thread sleep/wakeup would occur much less frequently than passing of data tokens, so avoiding atomic operations on data passing is beneficial.

volatile unsigned int produceCount, consumeCount;
TokenType buffer[BUFFER_SIZE];

void producer(void) {
while (1) {
while (produceCount - consumeCount

BUFFER_SIZE)
sched_yield; // buffer is full

buffer[produceCount % BUFFER_SIZE] = produceToken;
produceCount += 1;
}
}

void consumer(void) {
while (1) {
while (produceCount - consumeCount

0)
sched_yield; // buffer is empty

consumeToken( buffer[consumeCount % BUFFER_SIZE]);
consumeCount += 1;
}
}

Example in Java

import java.util.Stack;
import java.util.concurrent.atomic.AtomicInteger;

/**
* 1 producer and 3 consumers producing/consuming 10 items
*
* @author pt
*
*/
public class ProducerConsumer {

Stack items = new Stack;
final static int NO_ITEMS = 10;

public static void main(String args[]) {
ProducerConsumer pc = new ProducerConsumer;
Thread t1 = new Thread(pc.new Producer);
Consumer consumer = pc.new Consumer;
Thread t2 = new Thread(consumer);
Thread t3 = new Thread(consumer);
Thread t4 = new Thread(consumer);
t1.start;
try {
Thread.sleep(100);
} catch (InterruptedException e1) {
e1.printStackTrace;
}
t2.start;
t3.start;
t4.start;
try {
t2.join;
t3.join;
t4.join;
} catch (InterruptedException e) {
e.printStackTrace;
}
}

class Producer implements Runnable {

public void produce(int i) {
System.out.println("Producing " + i);
items.push(new Integer(i));
}

@Override
public void run {
int i = 0;
// produce 10 items
while (i++ < NO_ITEMS) {
synchronized (items) {
produce(i);
items.notifyAll;
}
try {
// sleep for some time,
Thread.sleep(10);
} catch (InterruptedException e) {
}
}
}
}

class Consumer implements Runnable {
//consumed counter to allow the thread to stop
AtomicInteger consumed = new AtomicInteger;

public void consume {
if (!items.isEmpty) {
System.out.println("Consuming " + items.pop);
consumed.incrementAndGet;
}
}

private boolean theEnd {
return consumed.get >= NO_ITEMS;
}

@Override
public void run {
while (!theEnd) {
synchronized (items) {
while (items.isEmpty && (!theEnd)) {
try {
items.wait(10);
} catch (InterruptedException e) {
Thread.interrupted;
}
}
consume;

}
}
}
}
}

See also

• Design pattern
  Design pattern (computer science)
  In software engineering, a design pattern is a general reusable solution to a commonly occurring problem within a given context in software design. A design pattern is not a finished design that can be transformed directly into code. It is a description or template for how to solve a problem that...
  
  
• Pipeline
  Pipeline (software)
  In software engineering, a pipeline consists of a chain of processing elements , arranged so that the output of each element is the input of the next. Usually some amount of buffering is provided between consecutive elements...
  
  
• Atomic operation
• FIFO
  FIFO
  FIFO is an acronym for First In, First Out, an abstraction related to ways of organizing and manipulation of data relative to time and prioritization...
  
  

Reference

• Mark Grand Patterns in Java, Volume 1, A Catalog of Reusable Design Patterns Illustrated with UML http://www.mindspring.com/~mgrand/pattern_synopses.htm

The source of this article is wikipedia, the free encyclopedia.  The text of this article is licensed under the GFDL.