CODEWITHSUNDEEP
javamultithreadingconcurrencythread-safetyproducer-consumer
cxyz
cxyz

Reputation: 843

Terminating out of Producer Consumer setup using wait and notify

Please find below consumer producer code:

// Producer

    public boolean busy = false;
    while (rst != null && rst.next()) {
        while (queue.size() == 10) {
            synchronized (queue) {
                try {
                    log.debug("Queue is full, waiting");
                    queue.wait();
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        }

        synchronized (queue) {
            queue.add(/* add item to queue */);
            busy = true;
            queue.notifyAll();
        }
    }

// Consumer

    try {
        while (!busy) {
            while (queue.isEmpty()) {
                synchronized (queue) {
                    try {
                        log.debug("Queue is empty, waiting");
                        queue.wait();
                    } catch (InterruptedException ex) {
                        ex.getMessage();
                    }
                }
            }

            synchronized (queue) {
                item = queue.remove();
                log.debug("Consumed item" + ++count + " :" + item);
                busy = false;
                queue.notifyAll();
            }
        }

In my producer code snippet i have synchronized the queue(linked blocking queue) which i am using to add elements and made the global boolean variable busy to true and notify consumer. Once queue size is 10 producer releases object lock and goes to wait state. For my consumer once global flag busy is true,consumer consumes elements and turn flag to false. The elements are produced and consumed appropriately.But my code is not terminating out of the producer consumer loop and final statement to be executed is "Queue is empty,waiting". Please let me know how to modify my code and termination condition to exit out of loop.

Upvotes: 1

Views: 486

Answers (1)

Holger
Holger

Reputation: 298233

Your code is broken is so many ways that it’s hard to decide, where to start. At first, you consumer is entirely enclosed in a while(!busy){ loop so once busy becomes true, it will exit the loop instead of starting to consume items. But since you are accessing the busy variable outside of any synchronization mechanism, there is no guaranty that this thread will ever see a true value written by another thread. At this point, it “helps” that your consumer not really checks the queue’s state when wait returns but just assumes that there are items (which is not guaranteed, read “spurios wakeups”, etc.) The documentation of Object.wait explains how to wait correctly.

But there are more failures to synchronize correctly. In the producer you are saying

while(queue.size() == 10){
    synchronized(queue){
        …

in other words, you are accessing the queue by calling size() without proper synchronization.

Another issue is your exception handling:

catch(InterruptedException ex)
{
    ex.getMessage();
}

calling getMessage but ignoring the result is like not calling it at all. You should add a real action here. Even a simple print statement is better than that.

Further, splitting the wait part and the produce/consume part into two distinct synchronized blocks may work as long as you have exactly one producer and one consumer but as soon as you have more threads, it’s broken since in between these two synchronized blocks, the state of the queue may change again.

Doing it correctly is not so hard, just simplify your code:

//Producer
public volatile boolean producerDone = false;
while( /* can produce more items */ ) {
    Item item=produceItem();
    synchronized(queue) {
        while(queue.size() == 10) try {
           log.debug("Queue is full, waiting");
           queue.wait();
        } catch(InterruptedException ex) {
            ex.printStackTrace();
        }
        queue.add(item);
        queue.notifyAll();
    }
}
producerDone=true;

 

//Consumer
while(!producerDone) {
    Item item;
    synchronized(queue) {
        while(queue.isEmpty()) try {
            log.debug("Queue is empty, waiting");
            queue.wait();
        } catch(InterruptedException ex) {
            ex.printStackTrace();
        }
        item = queue.remove();
        queue.notifyAll();
    }
    processItem(item);
}

Upvotes: 4

Related Questions