# Receptive map queue

I've written a class I call ReceptiveMapQueue. Its purpose is to organize the internal structure of an AFKable RPG. It only has two public methods - dumping entries into the queue, and extracting values from the head of the queue. I won't waste time describing it in paragraph form, since it is fully documented.

The expected usage of it might be:

private final Object authToken = new Object();
private final ReceptiveMapQueue<Integer, PlayerCharacter> playersByTick = new ReceptiveMapQueue<>(authToken);


Later, various methods will be updating characters

playersByTick.add(currentTick + 5, currentActor);


And every tick, you'll see something like this from the game loop:

Set<PlayerCharacter> charactersToProcess = playersByTick.extractAllOnOrBefore(currentTick);
for(PlayerCharacter pc : charactersToProcess) {
try(AutoLock pclock = pc.getLock().write()) {
simulator.bringGameStateToCurrent(pc);
} catch(Exception pokemon) {
log("error in processing " + pc.getName() + " tick " + currentTick, pokemon);
}
}


That's the basic usage in a nutshell. I've ran it through the basic tests of adding and extracting and it seems to pass. It does have a large memory footprint when I send it sky-high (a million entries) but I figure if I have a million registered users, I'll have the capital to tackle that problem.

That said, I'm open to all suggestions about how I can reduce its footprint, make it faster or more reliable, make it safer, or just things that I could generally be doing better. I'm especially interested in test cases that break. I think I've got most of them, but that's the funny thing about test cases: it's the ones you DON'T think about that usually break your code, right?

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.Set;

/**
* The Receptive Map Queue is a structure to permit batches of entities to be processed based on a given key. The
* operation was inspired by having RPG characters who would take their next automated action on a given 'game
* tick'. However, inactive users do not need to be processed as often and can be pushed back in the queue. In
* the game, they would still get to take the same amount of actions, but we can execute multiple actions at a
* time to save on load.
*
* That leads to the problem, though, of needing to process someone faster based on their user input. If an active
* user is processed every 5 game ticks, and an inactive user every 500 game ticks, a previously inactive user
* making an action on game tick 250 should be processed a little sooner.
*
* There are four structures internal to this class.
* <ul>
* <li> The map of keys to values. Because multiple values can share a key, we make keys map to a set of values.
* <li> The reverse map of values to keys. This optimization allows us to quickly find the set a given value is
* in so we can remove it from that set to overwrite their value.
* <li> The priority queue of keys. Whenever a key is added to the map, it is added to this priority queue. When
* an extraction takes place, we pop off the top of the queue until we find a value higher than our key.
* <li> The waiting list of key-value pairs to be added to the queue. This is necessary to improve concurrency and
* not have adds to the RMQ block while an extract or another add takes place.
* </ul>
*
*
* @param <K>
* @param <V>
*/
public class ReceptiveMapQueue<K extends Comparable<K>,V> {

/*
* A dummy class to hold keys and values for processing in the waiting queue
*/
private class DummyPair {
final K k;
final V v;
DummyPair(K k, V v) {
this.k = k;
this.v = v;
}

}

/* The map of keys to sets of values - this is the meat-and-bones of the RMQ */
private final Map<K, Set<V>> map = new HashMap<>();
/* A reverse lookup to quickly find values for removal from the map */
private final Map<V, K> reverseLookup = new HashMap<>();
/* An ordering of the keys that allows us to quickly find the first. Consider removing this and making
* map a NavigableMap? */
private final PriorityQueue<K> keys = new PriorityQueue<>();
/* The waiting list of queued up additions to the queue */
private final Queue<DummyPair> waiting = new ConcurrentLinkedQueue<DummyPair>();
/* The private 'permission slip' to extract values from the RMQ */
private final Object extractAuthToken;

/**
* Create a new Receptive Map Queue with a given auth token that demands only authorized threads/methods are
* extracting values from this RMQ.
* @param extractAuthToken The auth token for the RMQ.
*/
public ReceptiveMapQueue(Object extractAuthToken) {
this.extractAuthToken = extractAuthToken;
}

/**
* Add a key-value pair to the queue. The key indicates when in the sequence you would like the val to be
* available for processing.
*
* This method is thread safe.
*
* @param key A non null sequence identifier
* @param val A value to be returned when it's key's turn to be processed
*/
public void add(K key, V val) {
if(key == null) throw new NullPointerException("Null keys not permitted.");
if(val == null) throw new NullPointerException("Null values not permitted.");
}

/**
* Extract all values from the queue that map to keys equal to or less than the given key.
*
* This method is not thread safe, although the extractAuthToken will verify that either the thread that
* created the queue is the thread that will run this method or that that thread has granted permission to
* another thread to run this method (by virtue of passing the auth token to it).
*
* This thread begins by burning through the adds that have been queued up. Thus some values may be processed
* even if they began after this method was called. Once this method has finished processing the queue, it
* will begin to extract values from the internal structure to be returned for processing.
*
* After this method runs, there will be no values in the map that are not greater than the given key.
*
* @param key The highest/latest key to extract values for.
* @param extractAuthToken The 'password' ensuring that the caller is permitted to access this method.
* @return
*/
public Set<V> extractAllOnOrBefore(K key, Object extractAuthToken) {
if(this.extractAuthToken != extractAuthToken) {
throw new IllegalArgumentException("Invalid auth token");
}
clearWaitList();
Set<V> set = new HashSet<>();

while(true) {
K cur = keys.peek();
if(cur == null) return set;
if(cur.compareTo(key) > 0) return set;
cur = keys.poll();
Set<V> curSet = map.remove(cur);
for(V v : curSet) reverseLookup.remove(v);
}
}

/* creating a new empty set every Map#putIfAbsent() call is a bad idea. Thus, we cache an empty map
* and replace it when we use it. This is not thread safe. We could make it thread safe (with atomic
* reference) but if we really decide to thread safe THIS, we would just thread safe the entire extract
* procedure.
*/
private Set<V> nextEmptySet = new HashSet<>();

/*
* Clear the waiting list - this is called once per extract, which we know is externally synchronized. We could
* just add the various entries whenever they get added but that would require even more synchronization and
* we don't want external threads (like servlets) to drop their commands in quickly to release locks.
*/
private void clearWaitList() {
while(!waiting.isEmpty()) {
DummyPair p = waiting.poll();
if( p == null) return;
K old = reverseLookup.remove(p.v);
if(old != null) {
map.get(old).remove(p.v);
// leave the old set, could be others - who cares if it's empty...
// leave keys, it will get cleaned up in time
}
Set<V> set = map.putIfAbsent(p.k, nextEmptySet);
if(set == null) {
set = nextEmptySet;
nextEmptySet = new HashSet<>();
}
reverseLookup.put(p.v, p.k);
}
}
}

• your example code is broken. e is undefined. Why are you putting example code in your question? – Pimgd Jul 20 '16 at 17:57
• I don't understand your query - I think the real question is what is wrong with putting example code in the question? To be clear, the "example code" isn't from real production code. The class itself is. – corsiKa Jul 20 '16 at 18:28

The auth token is weird. Your class can't protect itself against reflection as far as I'm aware, so all this is doing is covering up a bad design. From the documentation I can sort of see it's supposed to guarantee thread safety - I don't understand why you'd go to such lengths; there are better ways of handling thread safety. Synchronizing on a private final Object lockObject = new Object() comes to mind.

The function names are also weird.

private void clearWaitList() {
while(!waiting.isEmpty()) {


This is not what I expected.

Here's what I expected:

private void clearWaitList() {
waiting.clear();
}


And done.

/*
* Clear the waiting list - this is called once per extract, which we know is externally synchronized. We could
* just add the various entries whenever they get added but that would require even more synchronization and
* we don't want external threads (like servlets) to drop their commands in quickly to release locks.
*/


It's not described in the documentation. It's not described in the function name. The purpose of the class is unclear. You've managed to write so many words and explain so little.

Perhaps I'm being rude. But there's so much here to confuse your fellow programmer!

/* creating a new empty set every Map#putIfAbsent() call is a bad idea. Thus, we cache an empty map
* and replace it when we use it. This is not thread safe. We could make it thread safe (with atomic
* reference) but if we really decide to thread safe THIS, we would just thread safe the entire extract
* procedure.
*/
private Set<V> nextEmptySet = new HashSet<>();


This is premature optimization! You are worried about ONE new instance of HashSet?! ONE?

Have you profiled your code? Do you know that constantly creating a new object is expensive in this case? If it does, you could improve this comment by stating how much the difference exactly is.

20 minutes of trying to understand this code later (That's a lot of dev time!), I sort-of get what you're trying to do here. That's not clearWaitList, it's processAddedItems. I'm still left confused:

        DummyPair p = waiting.poll();
if( p == null) return;


This code doesn't seem to make sense. The only way you can get to the return here is if the queue is not empty, but it is empty.

        K old = reverseLookup.remove(p.v);
if(old != null) {
map.get(old).remove(p.v);
// leave the old set, could be others - who cares if it's empty...
// leave keys, it will get cleaned up in time
}


This removes the old value. The comments here didn't help my understanding at first; I understand that you mean to say "don't clean up empty value sets for keys here". I don't see how it will get cleaned up in time.

        Set<V> set = map.putIfAbsent(p.k, nextEmptySet);


This... adds the key - set mapping if it's not there. Then it returns the old value, which will be null if it added and the old value if it didn't add. Pretty clear.

        if(set == null) {
set = nextEmptySet;
nextEmptySet = new HashSet<>();
}


If we just had to add a set for this key, make a new cached empty set... and add the key to the priority queue. Pretty clear.

        set.add(p.v);
reverseLookup.put(p.v, p.k);


Add key -> value and value -> key mappings. Pretty clear.

...

After looking at this in a bit more detail, you could merge these two lines:

         K old = reverseLookup.remove(p.v);
reverseLookup.put(p.v, p.k);


As put will return the old "value" (which is the "old key" in this case).

I don't think I can provide any meaningful commentary regarding the purpose and usage of this class. It seems to hide several VERY BIG assumptions, and it's very likely that you'll get bitten by them somewhere down the road. If I inherited this as-is, and there were no other pressing needs, I'd try to rewrite the whole thing from scratch in a simpler way, even sacrificing performance. I would personally prefer slower code over code that I couldn't understand.

A final comment:

/*
* A dummy class to hold keys and values for processing in the waiting queue
*/
private class DummyPair {
final K k;
final V v;
DummyPair(K k, V v) {
this.k = k;
this.v = v;
}

}


This really could have used key and value as variable names. It would have helped with reading.

• Thank you for the thorough review. Since CR is a bit different than the normal SE model these comments are based off of, I will probably have to spam you with comments to clarify your questions and to ask followups on your critique. I apologize in advance. – corsiKa Jul 20 '16 at 19:21
• chat.stackexchange.com/rooms/8595/the-2nd-monitor – Pimgd Jul 20 '16 at 19:22
• Not going to obliterate the original, as per policy, but I can say I've made the following changes: merged K old = rev... with the rev.put() from the end; renamed clearWaitList() to processPendingEntries, renaming waiting accordingly; eliminated redundant return; removed auth token in favor of implements MapQueueReceiver<K,V>; added optimization documentation to nextEmptySet. I think this is a better piece of code after our conversation. Many thanks. I'll leave it open for today and maybe a little longer to encourage other reviews. – corsiKa Jul 20 '16 at 21:52