Producer-consumer problem in Java with a circular buffer of N positions.
- There is a single thread representing the producer that will produce consecutive integer elements
- There is a dimensioned circular buffer shared by the producer and the consumer to produce and consume the elements. The elements will have the following attributes:
- elem: array with the elements of the buffer.
- p: indicates the position by which it is produced.
- c: indicates the position by which it is consumed.
- nelem: number of valid elements contained in the buffer
- There is a single consumer that will display the elements deposited in the buffer.
Edited code:
import java.util.concurrent.atomic.AtomicBoolean;
public class ProducerConsumer
{
public static final int bufferSize = 6;
private static CircularBuffer buffer = new CircularBuffer (bufferSize);
public static void main (String[] args)
{
if ( args.length != 1 ) {
String className = ProductorConsumidor.class.getSimpleName();
throw new RuntimeException ("Usage: java " + className + " number");
}
int numElems = Integer.parseInt (args[0]);
Consumer consumer = new Consumer (buffer, numElems);
Producer producer = new Producer (buffer, numElems);
consumer.start(); producer.start();
}
public static class Peterson
{
private static volatile int turn;
private static AtomicBoolean[] flag = { new AtomicBoolean(),
new AtomicBoolean() };
private int id;
public Peterson (int i)
{
id = i;
}
private int other ()
{
return id == 0 ? 1 : 0;
}
public void lock ()
{
flag[id].set (true);
turn = other ();
while ( turn == other() && flag[other()].get() )
Thread.yield ();
}
public void unlock ()
{
flag[id].set (false);
turn = other ();
}
}
public static class CircularBuffer
{
private volatile int[] elem;
private volatile int nElems;
private int producerPosition;
private int consumerPosition;
private Peterson pt0 = new Peterson (0);
private Peterson pt1 = new Peterson (1);
public CircularBuffer (int bufferSize)
{
elem = new int[bufferSize];
producerPosition = 0;
consumerPosition = 0;
nElems = 0;
}
public void produce (int element)
{
while ( nElems == bufferSize )
; // wait while full
produceNewElement (element);
}
public void produceNewElement (int element)
{
pt0.lock ();
elem[producerPosition] = element;
producerPosition = (producerPosition + 1) % bufferSize;
++nElems;
pt0.unlock ();
}
public int consume ()
{
while ( nElems == 0 )
; // wait while empty
return consumeNewElement ();
}
public int consumeNewElement ()
{
pt1.lock ();
int ret;
ret = elem[consumerPosition];
consumerPosition = (consumerPosition + 1) % bufferSize;
--nElems;
pt1.unlock ();
return ret;
}
}
public static class Producer extends Thread
{
private CircularBuffer buffer;
private int numElems;
public Producer (CircularBuffer b, int m)
{
buffer = b;
numElems = m;
}
@Override
public void run ()
{
for ( int i = 0; i < numElems; ++i ) {
buffer.produce (i);
}
}
}
public static class Consumer extends Thread
{
private CircularBuffer buffer;
private int numElems;
public Consumer (CircularBuffer b, int m)
{
buffer = b;
numElems = m;
}
@Override
public void run ()
{
int data, previousData;
data = buffer.consume ();
for ( int i = 0; i < numElems - 1; ++i ) {
System.out.printf (data + " ");
previousData = data;
data = buffer.consume ();
if ( previousData > data )
throw new RuntimeException ("Incorrect data order");
}
System.out.printf (data + " ");
}
}
}
Original code:
import java.util.concurrent.atomic.AtomicBoolean;
public class ProducerConsumer
{
public static void main (String[] args)
{
if ( args.length != 1 ) {
throw new RuntimeException ("Usage: fileName number");
}
int N = 6; // Size of circular buffer
int max = Integer.parseInt (args[0]); // number of elements to produce
CircularBuffer buf = new CircularBuffer (N);
Consumer cons = new Consumer (buf, max);
Producer produ = new Producer (buf, max);
cons.start(); produ.start();
}
public static class Peterson
{
private volatile int turn;
private AtomicBoolean[] flag = new AtomicBoolean[2];
private int id;
public Peterson (int i)
{
id = i;
}
private int other ()
{
return id == 0 ? 1 : 0;
}
public void lock ()
{
flag[id].set (true);
turn = other ();
while ( turn == other() && flag[id].get() )
Thread.yield ();
}
public void unlock ()
{
flag[id].set (false);
turn = other ();
}
}
public static class CircularBuffer
{
private volatile int[] elem;
private volatile int nelem; // control elements in buffer
private int p; // consumer position
private int c; // producer position
private Peterson pt0 = new Peterson (0);
private Peterson pt1 = new Peterson (1);
// Shared circular buffer
public CircularBuffer (int N)
{
elem = new int[N];
p = 0;
c = 0;
nelem = 0;
}
public void produce (int e)
{
while ( nelem == elem.length ); // while full
pt0.lock (); // lock
elem[p] = e; // produce new element
p = (p + 1) % elem.length; // new position in the circular array
++nelem; // increment number of elements in the shared buffer
pt0.unlock (); // unlock
}
public int consume ()
{
while ( nelem == 0 ); // while empty
pt1.lock (); // lock
int ret; // variable to return
ret = elem[c]; // assignment
c = (c + 1) % elem.length; // new position in the circular array
--nelem; // decrement number of elements in the shared buffer
pt1.unlock (); // unlock
return ret;
}
}
public static class Producer extends Thread
{
private CircularBuffer buf;
private int max;
public Producer (CircularBuffer b, int m)
{
buf = b;
max = m;
}
public void run ()
{
for ( int i = 0; i < max; ++i ) {
buf.produce (i);
}
}
}
public static class Consumer extends Thread
{
private CircularBuffer buf;
private int max;
public Consumer (CircularBuffer b, int m)
{
buf = b;
max = m;
}
public void run ()
{
int data, previousData;
data = buf.consume ();
for ( int i = 0; i < max - 1; ++i ) {
System.out.printf (data + " ");
previousData = data;
data = buf.consume ();
if ( previousData > data )
throw new RuntimeException ("Incorrect data order");
}
System.out.printf (data + " ");
}
}
}
othermethod? \$\endgroup\$