Producer Consumer Thread Race Conditions

Background

Recently, on StackOverFlow.com, a question was asked about why java.util.concurrent.ArrayBlockingQueue use while loops instead of if around calls to await(). The accepted answer was that it was due to spurious wakeups. I strongly disagree with this answer and instead believe that it is due to bad programming around race conditions. This is an all too common problem with producer/consumer models from threaded/reentrant applications.

The Problem

In producer/comsumer applications you have threads which are producing some items and threads which are consuming the items and working on them. This can be a website with servlet threads queueing up aynchronous database requests, a thread reading lines from a file and queueing up work that needs to be done on each line, etc..

Typically you have some sort of blocking queue so that when a producer has a item, it adds it to the end of the queue, possible blocking if the queue has too many items in it already. The consumer, when it needs a new item, gets from the front of the queue, possibly blocking if the queue is empty. It is the fact that both sides are blocking and need to signal the other side is where the complexity and confusion originate.

The Race

Here is a more specific enumeration of the race condition:

1. getter A is waiting for there to be items in the queue (line #31)
2. putter B locks the lock (line #54)
3. getter C finishes the last item, calls get(), tries to lock the lock, waits for B to unlock (line #25)
4. putter B adds an item into the queue and signals that there is an item there (line #72)
5. getter A is awoken and tries to lock the lock, has to wait for B to unlock (after line #31)
6. putter B unlocks (line #74)
7. getter C is ahead of getter A waiting for lock so it is given the lock first (line #25)
8. getter C dequeues the item that getter A was notified for (!!), and then unlocks (line #48)
9. getter A is finally able to lock, goes forward and calls remove on an empty list (line #40)

The race is between getter A and C on who is able to get the lock first to dequeue the new item. The race condition would be solved by changing lines 26 and 51 from if statements to while statements. Once getter A gets the lock it would check again to see if items is empty and would go back to waiting if C had already dequeued. The producer would also check again to see if the list is at capacity.

This is the real problem with many producer/comsumer models instead of spurious wakeups which I'm sure do happen but not nearly as frequently. As an interesting historical note, I had submit corrections to the original O'Reilly Pthreads book which suffered from the same problems -- errata had to be issued and a new version published.

Sample Code

Click to download sample code.

package com.j256;

import java.util.Vector;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Little class which demonstrates the perils of using if statements instead of while
 * in producer/consumer loops.
 */
public class Foo {

	// how many maximum items should the list store
	private static final int CAPACITY = 10;
	private static final int NUM_PUTTERS = 10;
	private static final int NUM_GETTERS = 10;

	private Vector<String> items = new Vector<String>(CAPACITY);
	private ReentrantLock lock = new ReentrantLock();
	private Condition hasSpace = lock.newCondition();
	private Condition hasItems = lock.newCondition();

	// consumer method
	public String get() {
		lock.lock();
		try {
			// see if the list is empty, !!DANGER!! should be a while loop
			if (items.isEmpty()) {
				try {
					// wait for the list to have items
					hasItems.await();
				} catch (InterruptedException e) {
					e.printStackTrace();
					return null;
				}
			}
			String item = null;
			try {
				// remove the item from the front of the list
				item = items.remove(0);
				// signal a waiting producer (if any) that the list has space
				hasSpace.signal();
			} catch (ArrayIndexOutOfBoundsException e) {
				System.err.println("Under capacity: " + items.size());
			}
			return item;
		} finally {
			lock.unlock();
		}
	}

	// producer method
	public void put(String item) {
		lock.lock();
		try {
			// see if the list is full, !!DANGER!! this should be a while loop
			if (items.size() >= CAPACITY) {
				try {
					// wait for the list to have space
					hasSpace.await();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			// add an item to the end of the list
			items.add(item);
			int size = items.size();
			if (size > CAPACITY) {
				System.err.println("Over capacity: " + size);
			}
			// signal a waiting consumer (if any) that the list has an item
			hasItems.signal();
		} finally {
			lock.unlock();
		}
	}

	public static void main(String[] args) {
		new Foo().doMain(args);
	}

	private void doMain(String[] args) {
		for (int i = 0; i < NUM_GETTERS; i++) {
			new Thread(new Getter()).start();
		}
		for (int i = 0; i < NUM_PUTTERS; i++) {
			new Thread(new Putter()).start();
		}
		new Putter();
	}

	private class Getter implements Runnable {
		public void run() {
			while (true) {
				get();
			}
		}
	}

	private class Putter implements Runnable {
		public void run() {
			while (true) {
				put("foo");
			}
		}
	}
}

Free Spam Protection   ORMLite Java ORM   Android ORM   Eggnog Recipe