ConcurrentModificationException is thrown (multiple for loops inside)

Question

I'm sorry this is not a SSCCE, but I've attempted to adequately describe the Objects in this problem.

Data Structures

Everything nested inside is a member variable

PerformanceData (Object)
|- process (String): The name of the process this PerformanceData has data for (ie: firefox.exe)
|- metrics (Map<String, List<DataValue>>): The keys are metrics whose usage we are monitoring ("cpu", "disk", "memory")

DataValue (Object)
|- value (double): The observed value (ie: 0.43 which means 43% cpu usage)
|- time (Date) : The time the value was observed

ModelEnsemble (Thread)
|- data (List<DataValue>): The DataValues available to this ModelEnsemble
|- models (Map<String, IEnsembleModel>): maps a name to each IEnsembleModel.
||-This is used because the user can choose which IEnsembleModels to run via a myApp.properties file.
||-In the constructor, we parse this properties file for enabled IEnsembleModels

IEnsembleModel (Object)
|- window (int): The max size input should be. When input.size() > window, we remove (older elements) from the front of the queue
|- input (ArrayDeque<DataValue>): The queue of DataValues we are modelling
|- addInput (DataValue): adds a DataValue to the end of the list
|- getLastInput (DataValue): The last inserted DataValue (from addInput)
|- getLastPrediction (double): The second latest predicted outcome from the current input data (minus the last input)
|- getError (double): Percent error, how much the last prediction (getLastPrediction) differed from the last input (getLastInput)
|- getNextPrediction (double): The latest predicted outcome from the current input data (including the last input)
|- model (double): Computes the next prediction, returns getNextPrediction().

The Problem

The following code throws a ConcurrentModificationException. I believe it is caused by the anonymous PropertChangeListener, as well as addInput(). Could some generous SO user look at my code and point out where I'm allowing multiple edits or race conditions?

Outer (main) loop:

for (final PerformanceData perfData : myPerformanceData) {
    for (Map.Entry<String, List<DataValue>> entry : perfData.getMetrics().entrySet()) {

        final String metric             = entry.getKey();
        final List<DataValue> values    = entry.getValue();

        ensemble = new ModelEnsemble(values);

        // Pretty sure this causes the Concurrent Modification Exception
        ensemble.addPropertyChangeListener(new PropertyChangeListener() {

            @Override
            public void propertyChange(PropertyChangeEvent evt) {

                IEnsembleModel model = (IEnsembleModel) evt.getNewValue();
                ModelPanel.this.firePropertyChange("result", null, new ModelResult(
                        perfData.getProcess(),
                        metric,
                        model.getLastInput().getValue(),
                        model.getLastPrediction(),
                        model.getError()
                ));
            }
        });

        ensemble.start();
    }
}

ModelEnsemble's run() loop

int offset = 0; // array index we are currently modelling
while (!this.isInterrupted()) {


    if (offset > data.size() - 1) {

        // we've exhausted the input input, close the model solver
        System.out.println("Input input exhausted, ending ensemble");
        this.interrupt();

    } else {

        // get the latest input value
        DataValue value = data.get(offset);

        for (Map.Entry<String, IEnsembleModel> entry : models.entrySet()) {
            String name         = entry.getKey();
            IEnsembleModel model= entry.getValue();

            model.addInput(value);
            model.model();
            this.notifyListeners(name, model); // triggers that anonymous PropertyChangeListener in the above piece of code
        }
    }

    offset++; // so we can model the next element of the DataValue List
}

IEnsembleModel example methods

double model() {
    double sum = 0;

    for (DataValue value : input) {
        sum += value.getValue();
    }

    setNextPrediction(sum / ((double) input.size()));
    return getNextPrediction();
}

void addInput(DataValue input) {
    this.input.addLast(input); // add to back of queue

    this.maintainWindow();
}

private void maintainWindow() {
    int size = this.input.size(); 
    while (size > window) { // by default window = 10
        this.input.pop();   // error here
    }
}

Stack Trace

This happens in every thread:

Exception in thread "Thread-13" java.util.ConcurrentModificationException
    at java.util.ArrayDeque$DeqIterator.next(ArrayDeque.java:632)
    at ca.yorku.cirillom.ensemble.models.MovingAverageModel.maintainWindow(MovingAverageModel.java:93)
    at ca.yorku.cirillom.ensemble.models.MovingAverageModel.addInput(MovingAverageModel.java:53)
    at ca.yorku.cirillom.ensemble.models.ModelEnsemble.run(ModelEnsemble.java:123)

This also happens in every thread:

Exception in thread "Thread-7" java.util.NoSuchElementException
    at java.util.ArrayDeque.removeFirst(ArrayDeque.java:278)
    at java.util.ArrayDeque.pop(ArrayDeque.java:507)
    at ca.yorku.cirillom.ensemble.models.MovingAverageModel.maintainWindow(MovingAverageModel.java:85)
    at ca.yorku.cirillom.ensemble.models.MovingAverageModel.addInput(MovingAverageModel.java:49)
    at ca.yorku.cirillom.ensemble.models.ModelEnsemble.run(ModelEnsemble.java:123)

I Feel like there must be a race condition somewhere because maintainWindow() checks if input.size() is larger than window, which is 10 by default. A NoSuchElementException can't happen if the size() is greater than 10

Answer 1

Concurrent modification exceptions can only be thrown if:

1. You are iterating over a loop using an iterator or foreach
2. You add or remove something from the list while the iterator is still running.

These two lines are doing the iteration so either myPerformanceData or perfData.getMetrics are being modified before these for loops exit.

for (final PerformanceData perfData : myPerformanceData) {
     for (Map.Entry<String, List<DataValue>> entry : perfData.getMetrics().entrySet()) {

or alternatively this line here:

    for (Map.Entry<String, IEnsembleModel> entry : models.entrySet()) {

The full stack trace of the error should give you the location where the concurrent modification happens and let you identify the collection being modified. You then just need to modify your logic so that changes to the collections happen outside of iterating over them.