12
\$\begingroup\$

I am currently writing a well.. manual testing site for a chat-based bot. The full code can be found on github.

For that purpose I had to keep track of the messages currently in the "system". The stored message format is that of a simple and immutable class with non-complex fields. The code for this format is coming from the bot's implementation.

I wanted to get into thread safety anyways, since it's a personally percieved shortcoming of mine, and thought this to be a good possibility, since the collection will be accessed from multiple threads.

For simplicity I decided to implement the class as a singleton. As of now, there's no contributors aside from myself, so I have to admit the project overall is underdocumented. Any comments about that are "breaking through open doors" as we Germans put it.

MessageTracker.java:

package de.vogel612.testclient_javabot.core;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntSupplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import com.gmail.inverseconduit.datatype.ChatMessage;

public final class MessageTracker {

    private static final int               LIMIT         = 200;

    private static final MessageTracker INSTANCE      = new MessageTracker();

    private final ChatMessage[]            messages      = new ChatMessage[LIMIT];

    private final AtomicInteger            lastQueryTime = new AtomicInteger(0);

    private final AtomicInteger            currentItem   = new AtomicInteger(0);

    private MessageTracker() {}

    private long getLimit(final int from) {
        return (LIMIT - from + currentItem.get()) % LIMIT;
    }

    public List<ChatMessage> newMessages() {
        return newMessages(lastQueryTime.get());
    }

    public List<ChatMessage> newMessages(final int since) {
        final List<ChatMessage> newMessages;
        final long limit = getLimit(since);
        synchronized (messages) {
            newMessages =
                    IntStream.generate(new Counter(since)).limit(limit).mapToObj(i -> messages[i]).filter(e -> e != null && !e.getMessage().isEmpty()).collect(Collectors.toList());
        }
        lastQueryTime.lazySet(currentItem.get());
        return newMessages;
    }

    public boolean newBotMessage(String message) {
        ChatMessage botMessage;
        botMessage = ChatMessageUtils.createFromString(message, "Junior");
        incrementAndWrap();
        messages[currentItem.get()] = botMessage;
        return true;
    }

    public boolean newUserMessage(String message) {
        ChatMessage userMessage;
        userMessage = ChatMessageUtils.createFromString(message, "You");
        incrementAndWrap();
        messages[currentItem.get()] = userMessage;
        return true;
    }

    private final class Counter implements IntSupplier {

        private int current;

        Counter(int start) {
            current = start;
        }

        @Override
        public int getAsInt() {
            current = incrementAndWrap(current);
            return current;
        }

        private final int incrementAndWrap(int item) {
            item++ ;
            if (item == LIMIT) {
                item = 0;
            }
            return item;
        }
    }

    /**
     * Allows resetting the TestingChatClient. All currently stored messages
     * will be lost.
     */
    public void reset() {
        currentItem.lazySet(0);
        lastQueryTime.lazySet(0);
        // no need to reset messages ;)
    }

    private final void incrementAndWrap() {
        currentItem.incrementAndGet();
        currentItem.compareAndSet(LIMIT, 0);
    }

    public List<ChatMessage> getMessages() {
        return Arrays.asList(messages);
    }

    public static MessageTracker getInstance() {
        return INSTANCE;
    }
}

Concerns:

First and foremost: threadsafety. Up to now I have not noticed any errors in multiple test sessions.

Secondly: Storage. Is it a good idea to use a fixed size array for this??

Thirdly: Duplication. I was not sure how I should resolve the problem of wrapping around at the limit of array storage, and thusly the incrementAndWrap functionality is duplicated in the Counter (I know the name is less than good, suggestions are very welcome)

Final notes:

There is a Test-class for this. It's not up for review here, but since this functionality is critical, I assumed you should know ;)

\$\endgroup\$
7
\$\begingroup\$

General Notes

I had to look at your code quite hard before I figured out what it does. Despite having variable names and method names that are descriptive, I still found it hard to 'get it'. Comments describing that the messages array is essentially a ring-buffer, and that the Counter is only used in the Stream, etc.

Threading and concurrency - don't mix systems

Your class uses two distinct strategies for thread management, synchronization, and atomics.

In general, whenever you see a mix of strategies in a single class, it is a warning that something is not right. The Java memory management required for thread-safe concurrency is relatively complicated (though not as complicated as other systems). The two strategies you use affect the model in ways that may not be obvious, and mixing them makes it even harder.

Additionally, the two models use different lock points, so you seldom get the exclusive thread accesses you need when you mix them.

Use just one system in any one class.

In this case, synchronization is the right choice. The atomic classes are useful when you do not have multi-stage operations that need to be thread-safe. Your code needs to increment counters, and add messages at the same time, and that requires at least two related operations in a single locked sequence, so synchronization is your friend for this.

Concurrency Bugs

Your code is not thread safe. Consider this method:

public boolean newUserMessage(String message) {
    ChatMessage userMessage;
    userMessage = ChatMessageUtils.createFromString(message, "You");
    incrementAndWrap();
    messages[currentItem.get()] = userMessage;
    return true;
}

That method does three things:

  1. creates a message
  2. increments (perhaps wraps) the index
  3. inserts the message in to the array.

Steps 2 and 3 should happen as one safe thing. We want each message to go in to its own index.

If two messages are created at the same time, and just before each message is created, the current index is at 2.

one possible order-of-execution will be:

Thread 1            Thread 2

                    create message

create message     
increment/wrap to 3
add to array at 3

                    increment/wrap to 4
                    add to array at 4

That's what would be a successful insert. Even though thread 2 started first, it added after.

But, what if the order was:

Thread 1            Thread 2

                    create message

create message     
increment/wrap to 3

                    increment/wrap to 4

add to array at 4

                    add to array at 4

Notice how the two increments can happen before the add-to-the-array.

Your add-to-array code is:

messages[currentItem.get()] = userMessage;

and that atomic get() may not get the value that was previously set.

So, you have a potential race condition in your atomics. That's why atomics are not the best choice here.

Now, you have a second bug in here. If your computer has 2 CPU's, the threads doing the work may be running on different cores, and have different memory caches. Consider the one thread does the add at position 4, and the other thread does it at position 3.

Since the add to position 4 is in one thread, and that access and assignment to the messages array is not in a memory-controlled block, there is no reason for anything happening on the other CPU to get the updated 'copy' of that array, and you may lose that setting at position 4. Any access to a variable/array that happens in one thread may not get the same value as what's in another thread unless the accesses are all done via the same memory control/lock/monitor.

In your 'read' side (getMessages()) you synchronize on the messages array, but, since the writes were not synchronized, there's no reason to expect the reads to get the written values.

In other words, what you put in to the messages array may not be there when you read from the messages array.

Other Issues

  • When you reset the Tracker, you should null out the values in the messages to allow for garbage collection.
  • when you return all messages, you are leaking encapsulation. The Arrays.asList(messages) call you do, will allow the person who calls your code to change the values in the array. You should return an unrelated copy of the data.

Consider this...

I put this code together to illustrate what I would consider doing. Use it as a guide. I am not thrilled with the int-stream generator, I think there's a better way, but this review really focuses on the concurrency and other bugs:

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.IntSupplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

//import com.gmail.inverseconduit.datatype.ChatMessage;

public final class MessageTracker {

    public static MessageTracker getInstance() {
        return INSTANCE;
    }

    private static final MessageTracker INSTANCE = new MessageTracker();

    private static final int CAPACITY = 200;
    private final ChatMessage[] messages = new ChatMessage[CAPACITY];

    private long messageId = 0;
    private long lastMessageReported = 0;
    private int size = 0;


    private MessageTracker() {
    }

    public List<ChatMessage> getMessages() {
        synchronized (messages) {
            // let the system figure it out.
            return newMessages(messageId - CAPACITY);
        }
    }

    public List<ChatMessage> newMessages() {
        synchronized (messages) {
            return newMessages(lastMessageReported);
        }
    }

    public List<ChatMessage> newMessages(final long since) {
        if (since < 0) {
            // correct broken input with cheeky recursion
            return newMessages(0);
        }
        synchronized (messages) {
            // cast to int is safe because size is int.
            final int reportCount = (int)Math.min(size, messageId - since);
            final List<ChatMessage> result = new ArrayList<>(reportCount);

            int offset = (int)((messageId - reportCount + 1) % CAPACITY);
            for (int i = 0; i < reportCount; i++) {
                result.add(messages[offset]);
                offset = (offset + 1) % CAPACITY;
            }

            lastMessageReported = messageId;
            return result;
        }
    }

    private boolean addMessage(ChatMessage message) {
        synchronized(messages) {
            messageId++;
            messages[(int)(messageId % CAPACITY)] = message;
            if (size < CAPACITY) {
                size++;
            }
            return true;
        }
    }

    public boolean newBotMessage(String message) {
        return addMessage(ChatMessageUtils.createFromString(message, "Junior"));
    }

    public boolean newUserMessage(String message) {
        return addMessage(ChatMessageUtils.createFromString(message, "You"));
    }

    /**
     * Allows resetting the TestingChatClient. All currently stored messages
     * will be lost.
     */
    public void reset() {
        synchronized (messages) {
            messageId = 0;
            lastMessageReported = 0;
            Arrays.fill(messages, null);
        }
    }
}
\$\endgroup\$
  • 1
    \$\begingroup\$ Reading this for the second time now I am amazed how easily you pointed out crucial flaws in my concurrency handling. But still I am sure that the code you present in 'newMessages', or more specifically the limit calculation there is flawed. Considering a call to this method with since as 0, the currentItem as 4 and lastQueryTime as 2, your limit is 0. This is undesirable, since we actually want the stream to go from 0 to 4 (iow from since to currentItem) and that won't happen with the code you put there. \$\endgroup\$ – Vogel612 Jan 10 '15 at 0:11
  • 1
    \$\begingroup\$ Oh, Yes, you are right. Your understanding of these constraints is better than mine. You want to choose the smallest 'limit' that is valid, and backtrack that amount from 'currentItem`. \$\endgroup\$ – rolfl Jan 10 '15 at 0:13
  • \$\begingroup\$ Actually it may be easier to see it as a rangeClosed. The limit is actually the count of items I want to advance with wrapping taken into account. But I can't stream rangeClosed because I'd get a IndexOutOfRange quite fast \$\endgroup\$ – Vogel612 Jan 10 '15 at 0:16
  • \$\begingroup\$ The issue, as I see it, is that you only have at most LIMIT valid messages, and your current code has the potential to loop more than that many times, thus repeating messages in the 'wrapping'. I know what the bug is, but the best solution is something you would probably identify before me. \$\endgroup\$ – rolfl Jan 10 '15 at 0:17
  • \$\begingroup\$ I don't know where exactly you're looping right now... but my original limit calculation has a value range of 0..LIMIT - 1 (thanks to modulo), which makes repetition of buffer access in the newMessages method impossible \$\endgroup\$ – Vogel612 Jan 10 '15 at 0:20

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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