Producers and Consumers by Mihailo Despotovic, December 2015

Producer threads put items into a shared buffer and consumer threads take them out, one lock guarding the count. A producer that finds the buffer full calls wait(), and so does a consumer that finds it empty; whoever changes the count calls notifyAll() to wake them. The count and the " F " and " E " (full, empty) printed as in the original scroll at the bottom. The original also sleeps while holding the lock, so the threads mostly take turns: nearly all of them are waiting for the lock at any moment.

Producers Consumers Buffer full at

Break it: if instead of while around wait() notify() instead of notifyAll()

With if, a woken thread doesn't check the count again, and the buffer soon overflows or goes below empty. With notify(), the one thread woken can be the wrong kind, until every thread is waiting and none is left to wake the others; that takes a while with 30 threads, so try the small setup at a high speed.

Speed 1× Time 0.0 s

Producers

Consumers

sleepingwaiting for the lock holding the lockin wait(): buffer full or empty

The original Java code: PC.java

import java.util.Random;

public class PC {
  private static volatile int counter = 0;
  private static Object mutex = new Object();
  private static final int C_POOL_SIZE = 10;
  private static final int P_POOL_SIZE = 20;

  private static class Consumer extends Thread { public void run() { while (true) { consume(); } } }
  private static class Producer extends Thread { public void run() { while (true) { produce(); } } }

  public static void main(String[] args) {
    final Consumer[] consumers = new Consumer[C_POOL_SIZE];
    for (int i=0; i<C_POOL_SIZE; i++) {
      consumers[i] = new Consumer(); consumers[i].start();
    }
    final Producer[] producers = new Producer[P_POOL_SIZE];
    for (int i=0; i<P_POOL_SIZE; i++) {
      producers[i] = new Producer(); producers[i].start();
    }
  }

  private static void produce() {
    try { Thread.sleep(new Random().nextInt(200)); } catch (InterruptedException ie) { ie.printStackTrace(); }
    synchronized (mutex) {
      while (counter == 9) {
        System.out.print(" F ");
        try { mutex.wait(); } catch (InterruptedException ie) { ie.printStackTrace(); } 
      }
      try { Thread.sleep(new Random().nextInt(200)); } catch (InterruptedException ie) { ie.printStackTrace(); }
      counter++;
      System.out.print(counter + " ");
      mutex.notifyAll();
    }
  }

  private static void consume() {
    try { Thread.sleep(new Random().nextInt(200)); } catch (InterruptedException ie) { ie.printStackTrace(); }
    synchronized (mutex) {
      while (counter == 0) {
        System.out.print(" E ");
        try { mutex.wait(); } catch (InterruptedException ie) { ie.printStackTrace(); }
      }
      try { Thread.sleep(new Random().nextInt(200)); } catch (InterruptedException ie) { ie.printStackTrace(); }
      counter--;
      System.out.print(counter + " ");
      mutex.notifyAll();
    }
  }
}