Java 1 producer - multiple consumer buffer deadlock

Hi,
I'm writing a java app and need to have a buffer with multiple consumers and 1 producer. The producer can only write to a buffercell if it has been read by every consumer.

I set up a test with 4 buffer places and 3 consumers + some sleep time to check things out. Sleeptimes are set so that at some point the producer tries to overtake the consumers, and thus has to wait.

Now the problem is that once it tries to overtake a deadlocks occurs, it somehow doesn't receive the signal to try again once cells are read and thus become available.

In this example that happens at 12. That's the last value it writes. Then the other cells are read and the buffer becomes empty, but not written to anymore though I call notifyAll();.

Buffer

Spoiler

Code:

package stannieman.integrityChecker;

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

/**
 * Author: Stannieman
 * Date: 27/10/13
 * Time: 23:49
 *
 */
public class Buffer {
private int[] buffer;
private int writeIdx;
private int[] readIndexes;
private boolean[][] occupied;

public Buffer(int nBuffers, int nReaders){
buffer = new int[nBuffers];
writeIdx = 0;

readIndexes = new int[nReaders];
for (int i=0; i<nReaders; i++)
readIndexes = 0;

occupied = new boolean[nBuffers][nReaders];
for (int i=0; i<nBuffers; i++)
for (int j=0; j<nReaders; j++)
occupied[j] = false;
}

public synchronized void write(int writeBytes) throws InterruptedException {
boolean isCellOccupied = false;
for (boolean occ : occupied[writeIdx])
if (occ)
isCellOccupied = true;

while(isCellOccupied)
wait();

buffer[writeIdx] = writeBytes;
for (int i=0; i<occupied[writeIdx].length; i++)
occupied[writeIdx] = true;

if (writeIdx < buffer.length - 1)
writeIdx++;
else
writeIdx = 0;

notifyAll();
}

public synchronized int read(int readerId) throws InterruptedException {
int returnBytes;

while (!occupied[readIndexes[readerId]][readerId])
wait();

returnBytes = buffer[readIndexes[readerId]];

occupied[readIndexes[readerId]][readerId] = false;

if (readIndexes[readerId] < buffer.length - 1)
readIndexes[readerId]++;
else
readIndexes[readerId] = 0;

notifyAll();

return returnBytes;
}
}

Filler (fills buffer)

Spoiler

Code:

package stannieman.integrityChecker;

/**
 * Author: Stannieman
 * Date: 28/10/13
 * Time: 16:47
 */
public class Filler implements Runnable{
private Buffer buff;

public Filler(Buffer buff)
{
this.buff = buff;
}

@Override
public void run() {
for (int i=0; i<30; i++)
{
try {
buff.write(i);
System.out.println("Write: " + i);
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

Extractor (reads from buffer)

Spoiler

Code:

package stannieman.integrityChecker;

/**
 * Author: Stannieman
 * Date: 28/10/13
 * Time: 16:52
 */
public class Extractor implements Runnable {
private int id;
private Buffer buff;

public Extractor(int id, Buffer buff){
this.id = id;
this.buff = buff;
}

@Override
public void run() {
for (int i=0; i<30; i++)
{
int read;
try {
read = buff.read(id);
System.out.println("Read" + id + ": " + read);
Thread.sleep(700);
} catch (InterruptedException e) {
e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
}
}
}
}

Main (to start stuff)

Spoiler

Code:

package stannieman.integrityChecker;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Author: Stannieman
 * Date: 28/10/13
 * Time: 16:54
 */
public class Main {
public static void main(String[] args) throws InterruptedException {
ExecutorService exServ = Executors.newCachedThreadPool();

Buffer buff = new Buffer(4, 3);

Filler fill = new Filler(buff);

Extractor ext1 = new Extractor(0, buff);
Extractor ext2 = new Extractor(1, buff);
Extractor ext3 = new Extractor(2, buff);

exServ.execute(fill);
Thread.sleep(200);
exServ.execute(ext1);
exServ.execute(ext2);
exServ.execute(ext3);

exServ.shutdown();

}
}

Thanks,
Stan

 

The buffer now stores int, but should store byte arrays. So the buffer variable would than be a 2d array. I only changed it to int for testing, but the principle and the bug remain the same.
That's why I sometimes use byte in the names of variables.

EDIT: I also tried it manually with a reentrantlock and conditions. Same issue but happens at 13, so seems doing it manual improves performance a bit.


EDIT: I implemented an ArrayBlockingQueue to fix it in different way (not optimal though), so it doesn't really matter anymore. Though it still would be nice to get this fixed.