0

I have created a class that attempts to process some messages in multiple threads, where each message belongs to a particular group. Each message is being added to the ConcurrentHashMap which has a key as the group number and is filling up while these threads are "processing". I have noticed that they sometimes run in parallel and sometimes not. And to make matters worse when there are more than 2 process threads running always they deadlock entirely.

[EDIT] Iteration of the ConcurrentHashMap seemed at the time to be a good way of going through all the elements as the numbered message groups (keys) are not known and it could change over time. The task specified that all messages be grouped together for processing but when when there is only one message in a group it should still process. So I thought this was a way to sort the elements as they arrive without knowing at the beginning which groups exist. [\EDIT]

public class GroupPriorityProcess implements Runnable {

private static final Object lock = new Object();
private static final Object counterLock = new Object();

private static int threadCounter = 0;
private final int currentThreadNumber;

private static Iterator<Integer> groupIterator;
private ConcurrentHashMap<Integer, LinkedBlockingQueue<Message>> groupMsgQueues;

public GroupPriorityProcess(ConcurrentHashMap<Integer, LinkedBlockingQueue<Message>> groupedMsgQueues) {
    groupMsgQueues = groupedMsgQueues;

    synchronized(lock){
        if (groupIterator == null)
        groupIterator = groupedMsgQueues.keySet().iterator();
    }
    synchronized (counterLock) {
        currentThreadNumber = (threadCounter++);
    }
}

// Main while loop for threads to process messages
public void run() {
    while (true) {
        LinkedBlockingQueue<Message> queue = chooseGroup();
        synchronized (queue) {
            process(queue);
        }
    }
}

// Loops till finds a message group available for processing.
private LinkedBlockingQueue<Message>  chooseGroup() {
    synchronized (lock) {
        while (!groupIterator.hasNext()) {
            groupIterator = groupMsgQueues.keySet().iterator();
        }
        LinkedBlockingQueue<Message> queue = groupMsgQueues.get(groupIterator.next());
        return queue;
    }
}

// takes messages from the a particular message group queue to completes the
// send process
private void process(LinkedBlockingQueue<Message> queue) {
    try {
        while (!queue.isEmpty()) {
            Message msg = queue.take();
            msg.appendMessage("Thread: " + currentThreadNumber);
            msg.completed();
        }
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

}

[EDIT] The messages are added here in another class GatewayImp.

public void send(Message msg) {
    int groupID = msg.getGroupID();

    if (groupedMsgQueues.containsKey(groupID)) {
        LinkedBlockingQueue<Message> queue = groupedMsgQueues.get(groupID);
        queue.add(msg);
    } else {
        LinkedBlockingQueue<Message> queue = new LinkedBlockingQueue<Message>();
        try {
            queue.put(msg);

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        groupedMsgQueues.put(groupID, queue);
    }
}
}

I have a few times when I use 'check then act' which I thought needs to be in a synchronized block to make atomic but I would like to know there is some better way. Please any help with this is greatly appreciated as I am only just starting to learn about concurrency and I'm finding it hard to get my head around locking in particular.

5
  • By looking at your code, I do not understand your design. If you're doing your own internal synchronization with locks, then there's no benefit to using ConcurrentHashMap. Having to iterate through the whole KeySet each time you use the Map seems suboptimal as you're not taking advantage of the best features of ConcurrentHashMap. If you use AtomicInteger you won't need do your own locking on your counter variable (which is better for concurrency).
    – scottb
    Commented Jul 24, 2015 at 18:26
  • One thread will go into a busy loop when groupMsgQueues is empty. This thread holds a global lock that will block other threads from calling chooseGroup() or from constructing additional GroupPriorityProcess instances. How that might cause your whole process to lock up depends on how messages are produced and how queues are added to groupMsgQueues. Please provide more information.
    – erickson
    Commented Jul 24, 2015 at 18:47
  • @scottb I have changed to AtomicInteger for keeping track of threads but other than readability in this instance are there any other advantages? Commented Jul 27, 2015 at 12:49
  • @lifesoordinary: AtomicInteger requires no external locking. For this reason, it avoids lock contention in a highly concurrent environment and can offer better performance. There is also no chance of liveness failures when using classes that require no locking.
    – scottb
    Commented Jul 27, 2015 at 15:41
  • " Iteration of the ConcurrentHashMap seemed at the time to be a good way of going through all the elements as the numbered message groups (keys) are not known and it could change over time." ... it sounds like the most common operation you are performing on your collection is a directory lookup (i.e. searching your collection for particular entries). If this is true and lookups are more common than additions to the collection, a more performant (and concurrent) data structure to consider is CopyOnWriteArrayList. No synchronization is necessary, even during iteration.
    – scottb
    Commented Jul 27, 2015 at 19:52

1 Answer 1

0

My first guess would be that you should not synchronize on the queue in your run method.

It can collide with the internal synchronization of LinkedBlockingQueue, when you call queue.take() later in the process method (while still holding the mutex of queue).

To help you debug your code, it is often useful to add verbose logging (e.g., some System.out.println statements). The good news is that you seem to be handle to reproduce the deadlock. Often, this is easier said than done...

3
  • Can you provide more explanation? While it's completely unnecessary to acquire this lock, the lock is re-entrant, and I don't see how it could "collide" with the call to take().
    – erickson
    Commented Jul 24, 2015 at 18:24
  • @erickson You are right. Though it is a good idea to get rid of the synchronize block, it is unlikely to be the reason for the deadlock. Not sure what cause the deadlock. Maybe it has to do with other parts of the application that we don't see. Commented Jul 25, 2015 at 14:57
  • I didn't want to reinitialize the iterator in the constructor many times and although this behaviour is probably unnecessary it was to make sure only one thread accessed a particular group. I assumed I should use the same lock each time I used or initialized the iterator anywhere (check if any more elements, if not reinitialize, etc.), however I have changed the the first synchronizeto lock with this and everything works. Commented Jul 27, 2015 at 12:47

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.