src/java/org/apache/cassandra/utils/concurrent/Accumulator.java

/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements.  See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership.  The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License.  You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied.  See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.utils.concurrent;

import java.util.AbstractCollection;
import java.util.Collection;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;

/**
 * A simple append-only collection supporting an unbounded number of concurrent readers/writers,
 * but a bounded number of items.
 *
 * @param <E>
 */
public class Accumulator<E>
{
    private volatile int nextIndex;
    private volatile int presentCount;
    private final Object[] values;
    private static final AtomicIntegerFieldUpdater<Accumulator> nextIndexUpdater = AtomicIntegerFieldUpdater.newUpdater(Accumulator.class, "nextIndex");
    private static final AtomicIntegerFieldUpdater<Accumulator> presentCountUpdater = AtomicIntegerFieldUpdater.newUpdater(Accumulator.class, "presentCount");

    public Accumulator(int size)
    {
        values = new Object[size];
    }

    /**
     * Adds an item to the collection.
     *
     * Note it is not guaranteed to be visible on exiting the method, if another add was happening concurrently;
     * it will be visible once all concurrent adds (which are non-blocking) complete, but it is not guaranteed
     * that any size change occurs during the execution of any specific call.
     *
     * @param item add to collection
     */
    public void add(E item)
    {
        int insertPos;
        while (true)
        {
            insertPos = nextIndex;
            if (insertPos >= values.length)
                throw new IllegalStateException();
            if (nextIndexUpdater.compareAndSet(this, insertPos, insertPos + 1))
                break;
        }
        values[insertPos] = item;
        // we then try to increase presentCount for each consecutive value that is visible after the current size;
        // this should hopefully extend past us, but if it doesn't this behaviour means the lagging write will fix up
        // our state for us.
        //
        // we piggyback off presentCountUpdater to get volatile write semantics for our update to values
        boolean volatileWrite = false;
        while (true)
        {
            int cur = presentCount;
            if (cur != insertPos && (cur == values.length || values[cur] == null))
            {
                // ensure our item has been made visible before aborting
                if (!volatileWrite && cur < insertPos && !presentCountUpdater.compareAndSet(this, cur, cur))
                {
                    // if we fail to CAS it means an older write has completed, and may have not fixed us up
                    // due to our write not being visible
                    volatileWrite = true;
                    continue;
                }
                return;
            }
            presentCountUpdater.compareAndSet(this, cur, cur + 1);
            volatileWrite = true;
        }
    }

    public boolean isEmpty()
    {
        return presentCount == 0;
    }

    /**
     * @return the size of guaranteed-to-be-visible portion of the list
     */
    public int size()
    {
        return presentCount;
    }

    public int capacity()
    {
        return values.length;
    }

    private Iterator<E> iterator(int count)
    {
        return new Iterator<E>()
        {
            int p = 0;

            public boolean hasNext()
            {
                return p < count;
            }

            public E next()
            {
                return (E) values[p++];
            }

            public void remove()
            {
                throw new UnsupportedOperationException();
            }
        };
    }

    public E get(int i)
    {
        // we read presentCount to guarantee a volatile read of values
        if (i >= presentCount)
            throw new IndexOutOfBoundsException();
        return (E) values[i];
    }

    public Collection<E> snapshot()
    {
        int count = presentCount;
        return new AbstractCollection<E>()
        {
            @Override
            public Iterator<E> iterator()
            {
                return Accumulator.this.iterator(count);
            }

            @Override
            public int size()
            {
                return count;
            }
        };
    }

    /**
     * Removes element at the speficied index from this accumulator.
     *
     * This method is not thread-safe when used concurrently with {@link #add(Object)}.
     */
    public void clearUnsafe(int i)
    {
        values[i] = null;
    }
}