Semaphore in Java. producer-consumer problem

Question

I am testing the use of semaphores with the typical producer-consumer problem where I only have one producer and one consumer. The producer adds products one at a time and the consumer can withdraw several simultaneously.

To perform the test, the producer and the consumer store and remove numbers from a array of 10 elements where 0 represents that there are no products and any other number represents a product. Access to store and retrieve items is centralized in a class called Data. I use a mutex to make an orderly use of the vector in case we have more than one thread working simultaneously.

When executing it, I observe that the number of permissions is not correct according to the operations performed by the threads. The application shows an error because the semaphore of the producer says that it has permission, but the data vector is full.

package producer.consumer;

import java.io.IOException;

public class ProducerConsumer {

    public static void main(String[] args) throws IOException {
        final int MAX = 10;

        Data data = new Data(MAX);
        Consumer consumer = new Consumer(data);
        Producer producer = new Producer(data);

        consumer.start();
        producer.start();
    }   
}

package producer.consumer;

public class Producer extends Thread{
    private final Data data;

    public Producer(Data data) {
         this.data = data;
    }

    @Override
    public void run() {
        while (true) {
            try {
                data.add((int) (Math.random() * data.getLength()) + 1);
            } catch (InterruptedException ex) {
                System.out.println(ex.getMessage());
            }
        }
    }  
}

package producer.consumer;

import java.util.logging.Level;
import java.util.logging.Logger;

public class Consumer extends Thread{
    private final Data data;

    public Consumer(Data data) {
        this.data = data;
    }

    @Override
    public void run() {
        while (true) {
            try {
                data.remove((int) (Math.random() * data.getLength()) + 1);
            } catch (InterruptedException ex) {
               Logger.getLogger(Consumer.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }
}

package producer.consumer;

import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.Semaphore;

public class Data {
    private final int[] data;
    private final Semaphore mutex = new Semaphore(1);
    private final Semaphore semProducer, semConsumer;

    public Data(int MAX) throws IOException {
        data = new int[MAX];
        semProducer = new Semaphore(MAX);
        semConsumer = new Semaphore(0);
    }

    public int getLength() {
        return data.length;
    }

    public void add(int number) throws InterruptedException {

        semProducer.acquire();

        mutex.acquire();
        System.out.println("trying to add a product");

        int i = 0;
        while (data[i] != 0) {
            i++;
        }
        data[i] = number;

        int permits = semConsumer.availablePermits() + 1;

        System.out.println("data added in " + i + " " + Arrays.toString(data)
            + " Resources consumer " + permits
            + " Resources producer  " + semProducer.availablePermits());
        mutex.release();

        semConsumer.release();
    }

    public void remove(int numberElements) throws InterruptedException {

        semConsumer.acquire(numberElements);

        mutex.acquire();
        System.out.println("trying to withdraw " + numberElements);

        for (int i = 0; i < numberElements; i++) {
            if (data[i] != 0) {
                data[i] = 0;
            }
        }

        int permisos = semProducer.availablePermits() + 1;
        System.out.println(" Retired " + numberElements + " " + Arrays.toString(data)
            + " Resources consumer " + semConsumer.availablePermits()
            + " Resources producer " + permisos);
        mutex.release();

        semProducer.release(numberElements);
    }
}

Thank you very much for the help.

Answer 1

EDIT Acquiring and releasing permits seems correct; but you need to make sure that the consumer will actually clear the correct number of elements.

In example, edit the Data class with

public void remove(int numberElements) throws InterruptedException {
    semConsumer.acquire(numberElements);
    mutex.acquire();
    System.out.println("remove: num-elem=" + numberElements);

    int consumed=0;
    for (int i = 0; consumed<numberElements; i++) {
        if (data[i] != 0) {
            data[i] = 0;
            consumed++;
        }
    }
    System.out.println(
            " Retired " + numberElements + " " + Arrays.toString(data)  );
    mutex.release();
    semProducer.release(numberElements);
}

---

Note also that this implementation is not very efficient (you'll need to iterate over the whole array both when inserting and deleting items, which can be expensive when MAX is large..)

Answer 2

Your consumer does not always consume what it claims to consume.

    for (int i = 0; i < numberElements; i++) {
        if (data[i] != 0) {
            data[i] = 0;
        }
    }

Suppose numberElements is 3, and that we have exactly 3 available elements in data[7], data[8], data[9].

The loop terminates with i == 3, nothing has been removed, but the producer semaphore will still be 'upped' by 3.

In the consumer, if you use i as the array index, it needs to cover the whole array, and you need a separate counter for 'elements removed'.

It is not the case that available elements will always be in the lowest-numbered data slots even though the producer fills those in first. Consider the time sequence that the producer manages to produce at least 5 elements, then the consumer runs to consume 2, and then immediately runs again to consume 3, before any more have been produced. data[0] and data[1] will be empty on the second run of the consumer and we run into the scenario I describe.