Multiplayer queue optimisation

Question

I currently have code that is working but I think it can be optimised. The code waits until a player joins and adds the player to a BlockingQueue if there is not already a Player waiting then it starts a game. The code also allows for multiple games to be played at the same time. The Players are two different threads that need to get the same answer about the game outcome.

PlayerQueue

import java.util.concurrent.*;

public class PlayerQueue {
    private final BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>(1);
    private final ConcurrentMap<String, String> userIdToResponse = new ConcurrentHashMap<>();

    public String battle(String user) {
        String opponent = null;
        try {
            boolean stopLoop = false;
            while (!stopLoop) {
                if (blockingQueue.remainingCapacity() > 0) {
                    stopLoop = blockingQueue.offer(user, 1, TimeUnit.SECONDS);
                } else {
                    opponent = blockingQueue.poll(500, TimeUnit.MILLISECONDS);
                    if (opponent != null) {
                        System.out.println("starting game");
                        String outcome = match(user, opponent);
                        userIdToResponse.put(opponent, outcome);
                        return outcome;
                    }
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            return "SOMETHING WENT WRONG";
        }

        if (opponent == null) {
            while (!userIdToResponse.containsKey(user)) {
            }
            return userIdToResponse.get(user);
        }

        return "ERROR";
    }

    private String match(String user1, String user2) {
        return user1 + " vs. " + user2;
    }
}

Main

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class Main {

    public static void main(String[] args) {
        PlayerQueue playerQueue = new PlayerQueue();
        Executor executor = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; ++i) {
            String user = "user" + Integer.toString(i);
            executor.execute(() -> System.out.println(playerQueue.battle(user)));
        }
    }
}

Step-by-Step Game Example:

1. Player 1 Thread calls the battle and passes the User its user instance which will be stored in the Queue
2. Player 2 Thread calls the battle since there is a player in the queue it starts a game
3. The game ends the Player 2 Thread puts the Game result into the Map
4. Player 1 Thread finds their User.id in the map gets the response and returns to the Response to the thread

Concerns:

1. Infinity while loop on the map and no possibility to stop thread 1
2. Overall performance because of the while loop

Answer 1

• Method battle is doing way too much. It is adding first player and waiting for second. It is adding the second and then playing and getting response. This all should have separate methods for clearer code.

• Infinite loop can be handled by multiple ways. The simplest would be to use wait + notify. That way the first thread just waits and the second thread notifies the first one and wakes him up. I will describe imho better ways later.

• In fact your whole infinite loop looks incorrect, because you are catching InterruptedException (for blockingQueue.offer and poll) outside your while-loop. Makes sense to put try-catch within your first loop. But I would get rid of the whole loop and try-catch completely and choose reasonable timeouts.

• You are never removing items from userIdToResponse. That leads to memory leaks and possibly bugs.

• Your code is not thread-safe and sometimes players joining can fail. Imagine your first condition succeeds, because remainingCapacity() is greater than 0, but there are 2 threads there and one thread calls offer inbetween so one thread ends up being without anything in the queue and fails on the timeout.

• The fact, that this is synchronous code smells. You call one method, that joins the queue, battle and returns results of the game. Is very likely to be bad in real life. I see getting results more like asynchronous callback called in the end for both players. It also gets rid of the second infinite loop that you have there.

• For me this seems like ideal case for producer-consumer scenario. It deals with all of the problems I described above. You can have multiple producers (game clients) and single consumer that handles game ("server"). Whenever a player wants to join, he will be added to player queue and battles will be handled by consumer, that will always require 2 players. This will also clearly separate code, that handles players waiting to join and players that are starting the game. Then you can for example do some kind of matchmaking and match players together based on their rating easily by choosing how to take player pairs from the current queue.

Answer 2

Blocking, busy-loop polling

Your code has threading issues. To see this, let's mentally walk through the code.

Imagine all 10 threads start and run in lock-step. They each will initialize stopLoop = false, start the while loop, and simultaneously check blockingQueue.remainingCapacity() and find it is empty, so they will all execute:

                    stopLoop = blockingQueue.offer(user, 1, TimeUnit.SECONDS);

Now, one thread (say "user0") will get the one and only spot on blockingQueue, and the other will start waiting for up to 1 second.

The "user0" thread will get stopLoop = true, and exit the while loop. Since opponent will still be null, it will start the while (!userIdToResponse.containsKey(user)) {} busy do-nothing loop, and peg that thread at 100% usage ... doing nothing ... wasting processing power.

One second later, all 9 other threads will time-out, and loop back to the start of the while loop. This time, when all 9 threads call blockingQueue.remainingCapacity(), they discover something is in the blockingQueue, so proceed to the else clause, where they all execute:

                    opponent = blockingQueue.poll(500, TimeUnit.MILLISECONDS);

Now, one of the 9 threads will have "user0" returned to it, the other 8 threads will start waiting (futilely) for up to 500 milliseconds for something to be added to the queue.

The thread that received "user0" from the queue calls match(), and stores outcome under the "user0" key in userIdToResponse, but not under its own key! Why the dichotomy?

Now, finally, after 1 second of busy looping doing nothing the "user0" thread sees its key in userIdToResponse, fetches the outcome and returns.

0.5 seconds later, the process repeats with the 8 remaining threads. One gets a spot on the blockingQueue and starts a busy loop, the other 9 wait for one second before the poll check.

With this "lock step" execution, we can see 5 of the 10 threads executing a time-wasting busy loop for 1 second each, and 0.5 second gaps between them, for around 7 seconds of time to get through the 10 threads!

Only half of the threads have outcomes in userIdToResponse, so you are missing information? Or, as mentioned by K.H., nothing is removed from userIdToResponse, so you have a possible memory leak.

Threaded I/O

            executor.execute(() -> System.out.println(playerQueue.battle(user)));

Your output here can get mixed up. From the println` documentation:

Prints a String and then terminate the line. This method behaves as though it invokes print(String) and then println().

This means that while the out stream itself is thread-safe, each string and newline could be interrupted by another output, and you could get (say):

user0 vs. user5user9 vs. user1user3 vs. user4


user2 vs. user6user8 vs. user7

5 "vs." strings, and 5 new-lines ... just not in the expected order.

Wrong Data Structure

This question borders on "off-topic", as it seems hypothetical. Perhaps you've reduced it too much in order to post a MCVE here. Code Review actually wants more code, so we have context to review.

In this case, you've reduced your users to strings! Your outcome, also a string. Your threads would have to parse the outcome string to determine who their opponent is, and whether they are the first or second player in the match.

It would be better to (say) pass MatchRequest objects with a player field into the queue, and the match() procedure could fill in an opponent member from pairs of MatchRequest objects. Again, we'd need to see the actual code, not a stripped down toy example.

Fixing the Code

You current code could be salvaged, and made to work better.

First, increase blockingQueue to hold enough information for a match:

    private final BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>(2);

Replace the while (!stopLoop) loop with if/else offer/poll with simply:

            blockingQueue.put(user);

No timeout. Two threads will succeed, the other 8 will block until space is made available. No looping required.

Add:

    private final CyclicBarrier barrier = new CyclicBarrier(2, barrierAction);

A CyclicBarrier will stop a thread, blocking it until the required number of threads have reached the barrier. So after blockingQueue.put(user), add:

            barrier.await();

Now, the 2 threads which get past the .put() operation will stop at the barrier.await() until both threads have reached that point, and barrierAction has been performed.

void PlayerQueue::barrierAction() is where you would remove the two players from the blockingQueue, and set the outcomes in the userIdResponse

Since barrier.await() will block until the barrierAction is complete, it is guaranteed that userIdResponse will contain outcomes for the two players, so barrier.await() can be followed directly with:

            outcome = userIdToResponse.remove(user);

to fetch the outcome for the user and ensure userIdToResponse does not grow without bound.

---

Something like:

    public String battle(String user) {
        try {
            blockingQueue.put(user);
            barrier.await();
            String outcome = userIdToResponse.remove(user);
            return outcome;
        } catch (InterruptedException e) {
            ...
        }
    }

    private void barrierAction() {
        String user1 = blockingQueue.take();
        String user2 = blockingQueue.take();
        String outcome = match(user1, user2);
        userIdToResponse.put(user1, outcome);
        userIdToResponse.put(user2, outcome);
    }