How to use java notify correctly in blocking queue implementation

Question

I am trying to understand Java multi-threading constructs, and I am trying to write a simple implementation of blocking queue. Here is the code I have written:

  class BlockingBoundedQueue<E>
    {
        @SuppressWarnings("unchecked")
        BlockingBoundedQueue(int size)
        {
            fSize = size;
            fArray = (E[]) new Object[size];
//            fBlockingQueue = new ArrayBlockingQueue<E>(size);
        }

        BlockingQueue<E> fBlockingQueue;

        public synchronized void put(E elem)
        {
            if(fCnt==fSize-1)
            {
                try
                {
                    // Should I be waiting/locking on the shared array instead ? how ?
                    wait();
                }
                catch (InterruptedException e)
                {
                    throw new RuntimeException("Waiting thread was interrupted during put with msg:",e);
                }
            }
            else
            {
                fArray[fCnt++]=elem;
                //How to notify threads waiting during take()
            }
        }


        public synchronized E take()
        {
            if(fCnt==0)
            {
                try
                {
                    // Should I be waiting/locking on the shared array instead ? how ?
                    wait();
                }
                catch (InterruptedException e)
                {
                    throw new RuntimeException("Waiting thread was interrupted during take with msg:",e);
                }
            }

            return fArray[fCnt--];
            //How to notify threads waiting during put()
        }
        private int fCnt;
        private int fSize;
        private E[] fArray; 

    }

I want to notify threads waiting in Take() from put() and vice versa. Can someone please help me with the correct way of doing this.

I checked the java.utils implementation and it uses Condition and ReentrantLocks which are a little complex for me at this stage. I am okay of not being completely robust[but correct] for the sake of simplicity for now.

Thanks !

Answer 1

The short answer is, call notifyAll() where you have the comments //How to notify threads waiting during take()

Now for the more complete answer...

The reference to read is : Java Concurrency in Practice. The answer to your question is in there.

However, to briefly answer your question: in Java, threads synchronize by locking on the same object and using wait() and notify() to safely change state. The typical simplified flow is:

1. Thread A obtains a lock by entering a synchronized block on a lock object
2. Thread A checks some condition in a loop, if not "OK to go" call thread.wait(), which is a blocking call that "releases" the lock so other code synchronized on the same lock object can proceed
3. Thread B obtains the same lock and may do something that changes the condition thread A is waiting for. When it calls notifyAll(), thread A will wake up and recheck the condition and (may) proceed

Some things to remember about synchronization are:

• it is about keeping state of objects consistent by making changes to state atomic. "Atomic" means the entire change (e.g. to multiple fields) is guaranteed to complete (no partial, and therefore inconsistent, changes)
• it is cooperative - code synchronized on a given lock object has in common the state that is being changed and the conditions that allow that state change - you wait and notify about the same "subject". Each part of state should be guarded by its own lock object - usually a private field, e.g. private Object lock = new Object(); would be fine
• methods that are synchronized use this as the lock object - doing this is easy but potentially expensive, because you are locking for every call, instead of just when you need to
• static methods that are synchronized use the Class object as the lock object