1

UPDATE: Please note.
The question I have asked was answered. Unfortunately for me, the issue is quite bigger than question in the Title. Apart from adding new entries to the map I had to handle updates and removals at the same time. The scenario I have in mind seems not possible to implement without one or the other:
a. deadlocks b. complex & time consuming checks and locks
Check the bottom of the Question for final thoughts.

ORIGINAL POST:

Hi,

I've got a spring bean with a Map.

Here's what I want to use it for:

few concurrent JMS Listeners will receive messages with actions. Each action consist of two users: long userA and long userB. Message will have it's own String replyTo queue which will be used to identify the action. Because I cannot allow to execute an action when one of the users participates in another action which is executed I am going to use this map as a registry of what is going on and in order to control execution of actions. So let's say I receive three actions:
1. userA, userB
2. userB, userC
3. userC, userA

When first action is received the map is empty so I am going to record info about the action in it and start executing the action.
When second action is received I can see that userB is 'busy' with first action so I simply record information about the action.
Same thing for third action.

Map is going to look like this:
[userA:[action1, action3],
userB:[action1, action2],
userC:[action2, action3]]

Once first action is complete I will remove information about it from the registry and get info about next actions for userA and userB [action3, action2]. Then I will try to restart them.

I think by now you get what I want to do with this map.

Because map is going to be accessed from several threads at the same time I have to handle synchronization somehow.

I will have methods to add new information to the map and to remove info from the map when action is done. The remove method will return next actions [if there are any] for the two users for whom the action just finished.

Because there could be hundreds of actions executed at the same time and the percentage of actions with busy users is supposed to be low I don't want to block access to the map for every add/remove operation.

I thought about making synchronized access only to each of the Lists within the Map to allow concurrent access to several user entries at the same time. However... because when there are no actions left for the user I want to remove entry for this user from the map. Also... when user has no entry in the map I will have to create one. I am a little bit afraid there could be clashes in there somewhere.

What would be the best way to handle this scenario? Is making both methods - add and remove - synchronized (which I consider the worst case scenario) the only proper [safe] way to do it?

Additionally I will have another map which will contain action id as keys and user ids as values so it's easier to identify/remove user pairs. I believe I can skip synchronization on this one since there's no scenario where one action would be executed twice at the same time.

Although code is in Groovy I believe no Java programmer will find it difficult to read. It is Java behind it. Please consider following as pseudo code as I am just prototyping.

class UserRegistry {

    // ['actionA':[userA, userB]]
    // ['actionB':[userC, userA]]
    // ['actionC':[userB, userC]]
    private Map<String, List<Long>> messages = [:]
    /**
     * ['userA':['actionA', 'actionB'],
     * ['userB':['actionA', 'actionC'],
     * ['userC':['actionB', 'actionC']
     */
    private Map<long, List<String>> users = [:].asSynchronized()

    /**
     * Function will add entries for users and action to the registry.
     * @param userA
     * @param userB
     * @param action
     * @return true if a new entry was added, false if entries for at least one user already existed
     */
    public boolean add(long userA, long userB, String action) {
        boolean userABusy = users.containsKey(userA)
        boolean userBBusy = users.containsKey(userB)
        boolean retValue
        if (userABusy || userBBusy)  {
            if (userABusy) {
                users.get(userA).add(action)
            } else {
                users.put(userA, [action].asSynchronized())
            }
            if (userBBusy) {
                users.get(userB).add(action)
            } else {
                users.put(userB, [action].asSynchronized())
            }
            messages.put(action, [userA, userB])
            retValue = false
        } else {
            users.put(userA, [action].asSynchronized())
            users.put(userB, [action].asSynchronized())
            messages.put(action, [userA, userB])
            retValue = true
        }
        return retValue
    }

    public List remove(String action) {
        if(!messages.containsKey(action)) throw new Exception("we're screwed, I'll figure this out later")
        List nextActions = []
        long userA = messages.get(action).get(0)
        long userB = messages.get(action).get(1)
        if (users.get(userA).size() > 1) {
            users.get(userA).remove(0)
            nextActions.add(users.get(userA).get(0))
        } else {
            users.remove(userA)
        }
        if (users.get(userB).size() > 1) {
            users.get(userB).remove(0)
            nextActions.add(users.get(userB).get(0))
        } else {
            users.remove(userB)
        }
        messages.remove(action)
        return nextActions
    }
}

EDIT I thought about this solution last night and it seems that messages map could go away and users Map would be:

Map users<String, List<UserRegistryEntry>>

where
UserRegistryEntry:
String actionId
boolean waiting

now let's assume I get these actions:

action1: userA, userC
action2: userA, userD
action3: userB, userC
action4: userB, userD

This means that action1 and action4 can be executed simultaneously and action2 and action3 are blocked. Map would look like this:

[
[userAId: [actionId: action1, waiting: false],[actionId: action2, waiting: true]],  
[userBId: [actionId: action3, waiting: true], [actionId: action4, waiting: false]],  
[userCId: [actionId: action1, waiting: false],[actionId: action3, waiting: true]],
[userDId: [actionId: action2, waiting: true], [actionId: action4, waiting: false]]
]

This way, when action execution is finished I remove entry from the map using:
userAId, userBId, actionId
And take details about first non blocked waiting action on userA and userB [if there are any] and pass them for execution.

So now the two methods I will need, which are going to write data to the Map and remove it from the map.

public boolean add(long userA, long userB, String action) {
    boolean userAEntryExists = users.containsKey(userA)
    boolean userBEntryExists = users.containsKey(userB)
    boolean actionWaiting = true
    UserRegistryEntry userAEntry = new UserRegistryEntry(actionId: action, waiting: false)
    UserRegistryEntry userBEntry = new UserRegistryEntry(actionId: action, waiting: false)
    if (userAEntryExists || userBEntryExists) {
        if (userAEntryExists) {
            for (entry in users.get(userA)) {
                if (!entry.waiting) {
                    userAEntry.waiting = true
                    userBEntry.waiting = true
                    actionWaiting = true
                    break;
                }
            }
        }

        if (!actionWaiting && userBEntryExists) {
            for (entry in users.get(userB)) {
                if (!entry.waiting) {
                    userAEntry.waiting = true
                    userBEntry.waiting = true
                    actionWaiting = true
                    break;
                }
            }
        }
    }

    if (userBEntryExists) {
        users.get(userA).add(userAEntry)
    } else {
        users.put(userA, [userAEntry])
    }

    if (userAEntryExists) {
        users.get(userB).add(userBEntry)
    } else {
        users.put(userB, [userBEntry])
    }

    return actionWaiting
}

And for removes:

public List remove(long userA, long userB, String action) {
        List<String> nextActions = []
        finishActionAndReturnNew(userA, action, nextActions)
        finishActionAndReturnNew(userB, action, nextActions)

        return nextActions;
    }

    private def finishActionAndReturnNew(long userA, String action, List<String> nextActions) {
        boolean userRemoved = false
        boolean actionFound = false
        Iterator itA = users.get(userA).iterator()
        while (itA.hasNext()) {
            UserRegistryEntry entry = itA.next()
            if (!userRemoved && entry.actionId == action) {
                itA.remove()
            } else {
                if (!actionFound && isUserFree(entry.otherUser)) {
                    nextActions.add(entry.actionId)
                }
            }
            if (userRemoved && actionFound) break
        }
    }

    public boolean isUserFree(long userId) {
        boolean userFree = true
        if (!users.containsKey(userId)) return true
        for (entry in users.get(userId)) {
            if (!entry.waiting) userFree = false
        }
        return userFree
    }

FINAL THOUGHT:
This scenario is a killer:
[ActionID, userA,userB]
[a, 1,2]
[b, 1,3]
[c, 3,4]
[d, 3,1]
Action a and c are executed simultaneously, b and d are waiting.
When a and c are done, entries for users 1,2,3,4 will have to be removed, thus one thread will have 1 and 2 locked, the other thread will have 3 and 4 locked. When these users are locked a check for next action for each of them has to be performed. When code determines that for user 1 next action is with user 3 and for user 3 next action is with user 1, whey will try to lock them. This is when the deadlock happens. I know I could code around that, but it seems it will take a lot of time to execute and it will block two workers.
For now I will ask another question on SO, more on the subject of my issue and try to prototype the solution using JMS in the meantime.

2
  • Primitives can't be in generic types. Use Long instead of long.
    – Leo Izen
    Nov 8, 2010 at 23:13
  • Thanks Leo, as I I wrote in the question, consider this to be a pseudo code. Just wanted to visualise more or less what I want to do with the map.
    – Krystian
    Nov 8, 2010 at 23:26

3 Answers 3

3

You may need to review how synchronized (collections) work again:

This (as a non-exclusive example) is not thread-safe:

if (users.get(userA).size() > 1) {
    users.get(userA).remove(0)

Remember that only individual "synchronized" methods are guaranteed atomic without a larger lock scope.

Happy coding.

Edit - per-user synchronization locks (updated for comment):

Just by using the standard data-structures you can achieve per-key locks by using ConcurrentHashMap -- in particular by using the 'putIfAbsent' method. (This is significantly different than just using get/put of a 'synchronized HashMap', see above.)

Below is some pseudo-code and notes:

public boolean add(long userA, long userB, String action) {
    // The put-if-absent ensures the *the same* object but may be violated when:
    //   -users is re-assigned
    //   -following approach is violated
    // A new list is created if needed and the current list is returned if
    // it already exists (as per the method name).
    // Since we have synchronized manually here, these lists
    // themselves do not need to be synchronized, provided:
    // Access should consistently be protected across the "higher"
    // structure (per user-entry in the map) when using this approach.
    List listA = users.putIfAbsent(userA, new List)
    List listB = users.putIfAbsent(userB, new List)
    // The locks must be ordered consistently so that
    // a A B/B A deadlock does not occur.
    Object lock1, lock2
    if (userA < userB) {
        lock1 = listA, lock2 = listB
    } else {
        lock1 = listB, lock2 = listA
    }
    synchronized (lock1) { synchronized (lock2) {{ // start locks

    // The rest of the code can be simplified, since the
    // list items are already *guaranteed* to exist there is no
    // need to alternate between add and creating a new list.
    bool eitherUserBusy = listA.length > 0 || listB.length > 0
    listA.add(action)
    listB.add(action)
    // make sure messages allows thread-safe access as well
    messages.put(action, [userA, userB])
    return !eitherUserBusy

    }} // end locks
}

I have no how this fairs under your usage scenario vs. a single common lock object. It is often advisable to go with "simpler" unless there is a clear advantage to do otherwise.

HTH and Happy coding.

6
  • Thanks pst. What I have to do is perform some actions synchronised and some not. I would like to know if it is possible to make insert into the Map synchronised so when a new key is created no other thread can create a new key. Then if a Thread modifies a List which is in the Map under key A I would like no other thread to access the same list, but allow access for lists under other keys. For example, would it work if I would check whether a key in map exists in one method call and if it does not have a separate synchronized method to recheck and add entry if needed? Would this be ok?
    – Krystian
    Nov 9, 2010 at 18:26
  • @Krystian The updated answer may answer some of your questions. I think you will find the putIfAbsent method one suitable approach.
    – user166390
    Nov 10, 2010 at 10:02
  • I was trying to implement ReentrantReadWriteLock on two levels: user Map and every List inside map entry. This seems to be a better approach. I'll try to implement it with appropriate remove command and see how it goes. In the remove command I have to check and lock at the same time several lists inside the map [if they exist] because I have to check availability of users for next actions with users who just finished execution. [Scenario is described in my last edit of the question]
    – Krystian
    Nov 11, 2010 at 10:12
  • I read about ConcurrentHashMap and it seems even though it's great for inserting new entries, it might not be as good when it comes to operation during remove. During remove I have to update both entries, have the entries locked and go through [and I mean lock] other entries and if needed update them. In this case I think it would mean a loop with a synchronized block inside. sounds weird. I'll try to prototype later today. Thanks again!
    – Krystian
    Nov 11, 2010 at 10:33
  • 1
    Hi again. I am getting NPE between putIfAbsent and synchronised block due to other threads removing map entries before I get to synchronized. Also I am getting deadlocks between different remove calls. I'm beginning to think it won't be possible without locking the map.
    – Krystian
    Nov 13, 2010 at 1:56
2

You might want to check out Collections.synchronizedMap() or Collections.synchronizedList()

2
  • [:].asSynchronized() means Collections.synchronizedMap(new HashMap()); but I'm not sure if what I'm doing in here is wise. From performance and data-integrity point of view.
    – Krystian
    Nov 8, 2010 at 23:27
  • But neither of these (by themselves) address the issues above. In particular, without an explicit lock encompassing the synchronized map (which is perfectly feasible), there is no way to get an atomic test-and-set semantic or similar.
    – user166390
    Nov 10, 2010 at 22:53
1

You have two global state-holders in the class and compound-actions in each of the two methods that modify both of them. So, even if we changed the Map's to be ConcurrentHashMap's and the List to something like CopyOnWriteArrayList, it would still not guarantee a consistent state.

I see that you will be writing to the List often, so, CopyOnWriteArrayList might be too expensive anyway. ConcurrentHashMap is only 16-way striped. If you have better hardware, an alternative would be Cliff Click's highscalelib (after appropriate locking in the methods).

Back to the consistency question, how about use a ReentrantLock instead of synchronizing and see if you can exclude some statements out of the lock()-to-unlock() sequence. If you went with a ConcurrentMap, the first two statements in the add() that do containsKey() can be optimistic and you may be able to exclude them from the lock block.

Do you really need the messages map? It is kind of like an inverse index of users. One other option would be to have another watch() method that periodically updates the messages map based on a signal from add() after a change to users. The refresh could alternatively be completely async. In doing that, you might be able to use a ReadWriteLock with the readLock() on users while you update messages. In this situation, add() can safely acquire a writeLock() on users. It is just some more work to get this reasonably correct.

1
  • Thanks gshx. I thought about it last night and yep, the messages map is not needed but this will mean a change in how the other map looks like. Take a look at the edit. Still not sure what I would have to make synchronized to make it work without blocking other threads.
    – Krystian
    Nov 9, 2010 at 13:12

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.