Queue that connects multiple producers and multiple consumers

Question

The scenario is about processing 'Message' objects. Producer creates them and Consumer does the consumption.

class Message{...}

class MessageBatch{

    public MessageBatch(Collection<Message> messages){...}

    public List<Message> getMessages(){...};

    //more utility methods like getMessageIds(), etc.
}

The following is a pipeline object (queue) that connects multiple producer threads and multiple consumer threads. Both my producers & consumers are I/O bound (database) and that justifies this attempt.

API requirements

• [A] Ability to add & get batch of Message objects.
• [B] Consumer has a way to realize that no more elements are expected.

public class MessagePipeline {

    //[1]; to support API requirement [B]
    public static final MessageBatch SENTINEL_BATCH = new MessageBatch(Collections.<Message>emptyList());

    private BlockingQueue<Message> messageQueue = new LinkedBlockingQueue<>();

    // to support API requirement [B]
    private CountDownLatch producersWorkCompletionLatch;

    public MessagePipeline(int numOfProducerThreads) {
        this.producersWorkCompletionLatch = new CountDownLatch(numOfProducerThreads);
    }

    public boolean add(MessageBatch batch) {
        isWriteAllowed(); // UPDATE: this step is misleading and thinking of removing this.
        return messageQueue.addAll(batch.getMessages());
    }

    // returns a batch containing up to maxSize elements. 
    public MessageBatch takeBatch(int maxSize) {
        List<Message> messages = drainIfMoreElementsExpected(new ArrayList<Message>(), maxSize);

        return messages.isEmpty() ? SENTINEL_BATCH : new MessageBatch(messages);
    }

    //[2]: Producer should invoke this at the end.
    public void notifyCompletion() {
        producersWorkCompletionLatch.countDown();
    }

    public boolean isMoreElementsExpected() {
        boolean isExpected = !(producersWorkCompletionLatch.getCount() == 0 && messageQueue.size() == 0);
        return isExpected;
    }
    //[3] : This is CPU bound waiting. TODO: Make it I/O bound.
    private List<Message> drainIfMoreElementsExpected(List<Message> messages, int maxSize) {
        while (messages.isEmpty() && isMoreElementsExpected()) {
            messageQueue.drainTo(messages, maxSize);
        }
        return messages;
    }

    /**
     * Assures write-operations 'happen-before' producer-threads-completion.
     */
    private void isWriteAllowed(){
        if(producersWorkCompletionLatch.getCount() == 0){
            throw new UnsupportedOperationException("New elements are not accepted after all producer threads notify of completion");
        }
    }
}

Questions

• I'm looking forward to code review in general.
• Do approaches [1] and [2] seem right?
• [3]: Am I right that it is CPU bound? How do I make it I/O bound waiting? Introducing locks seem to make it more complicated.

I'm in the middle of writing unit tests. It's good to know if I missed anything because unit-testing multi-threaded code tells you that the code is broken only one in million times and doesn't tell you why!

Answer 1

Since you say that your whole system is based on message passing you should take a serious look at akka. You get a rock solid message passing system, and you can even spread it on multiple machines in the future.

I have not checked in detail, but you have some very serious concurrency bugs. For example, add checks if writing is allowed, then writes to the queue. The problem is that you did not synchronize add, so the check might pass, but before calling the next line, some other thread might bring the count down latch down to zero. BlockingQueue is thread-safe, but it is far from enough to make your code thread-safe.

I don't like the MessagePipeline api: it is too complex. I really don't see the advantage of writing or taking "batches" of messages. If the reader or writers want to process many messages, just let them read or write many messages themselves by processing their many messages in a loop. The api itself should only read/write one message at a time. In that case, your whole MessagePipeline is superfluous since a simple BlockingQueue is exactly what you want. You will also avoid many concurrency bugs. You would also need to keep the CountDownLatch or something similar to let the consumer know when it's over.

Answer 2

If this is a real problem, rather than an exercise, you should look at the LMAX Disruptor library. Multiple producer x multiple consumer support is built in, the ring buffer data structure is a natural fit for batching, and a lot of the problems you are going to need to think about (eventually) are already covered.

Speaking to your design, as is....

There's no reason for your pipeline to know how many producers there are -- pipeline doesn't create them, after all, so you might as well have the thing that owns responsibility for creating the producers also have responsibility for creating the CountDownLatch.

You've really got two different interfaces mixed in here -- Consumers are never going to be calling notifyCompletion(), and Producers are never going to be calling takeBatch, so I think you should tease those two bits apart.

In fact, I think the fact that Pipeline isn't already implementing an interface is a code smell -- why should the Producers need to know that they are talking to a pipeline, rather than a dedicated Consumer?

You don't seem to be addressing cancellation and shutdown at all.

Allowing a single consumer to grab most of the messages out of the queue at once seems to defeat the purpose of the parallelism I expect you want. Presumably, each message is supposed to be consumed once, and you want the latency to be as low as possible -- so what you don't want is messages backed up in the queue of worker #0 while workers #1...#n spin idly.