java.util.concurrent.atomic.AtomicReferenceArray#compareAndSet()源码实例Demo

/**
 * @param maps Maps.
 * @param idx Index.
 * @return Map.
 */
private HadoopMultimap getOrCreateMap(AtomicReferenceArray<HadoopMultimap> maps, int idx) {
    HadoopMultimap map = maps.get(idx);

    if (map == null) { // Create new map.
        map = get(job.info(), SHUFFLE_REDUCER_NO_SORTING, false) ?
            new HadoopConcurrentHashMultimap(job.info(), mem, get(job.info(), PARTITION_HASHMAP_SIZE, 8 * 1024)) :
            new HadoopSkipList(job.info(), mem);

        if (!maps.compareAndSet(idx, null, map)) {
            map.close();

            return maps.get(idx);
        }
    }

    return map;
}
 

public void put(WeakOffHeapReference key)
{
    int hash = this.hash(key);
    AtomicReferenceArray currentArray = this.table;
    int length = currentArray.length();
    int index = ConcurrentOffHeapWeakHolder.indexFor(hash, length);
    Object o = currentArray.get(index);
    if (o == null)
    {
        Entry newEntry = new Entry(key, null);
        if (currentArray.compareAndSet(index, null, newEntry))
        {
            this.addToSize(1);
            return;
        }
    }
    this.slowPut(key, hash, currentArray);
}
 

/**
 * compareAndSet in one thread enables another waiting for value
 * to succeed
 */
public void testCompareAndSetInMultipleThreads() throws InterruptedException {
    final AtomicReferenceArray a = new AtomicReferenceArray(1);
    a.set(0, one);
    Thread t = new Thread(new CheckedRunnable() {
        public void realRun() {
            while (!a.compareAndSet(0, two, three))
                Thread.yield();
        }});

    t.start();
    assertTrue(a.compareAndSet(0, one, two));
    t.join(LONG_DELAY_MS);
    assertFalse(t.isAlive());
    assertSame(three, a.get(0));
}
 

public V put(int key, V value)
{
    int hash = this.hash(key);
    AtomicReferenceArray currentArray = this.table;
    int length = currentArray.length();
    int index = ConcurrentIntObjectHashMap.indexFor(hash, length);
    Object o = currentArray.get(index);
    if (o == null)
    {
        Entry<V> newEntry = new Entry<V>(key, value, null);
        if (currentArray.compareAndSet(index, null, newEntry))
        {
            this.addToSize(1);
            return null;
        }
    }
    return this.slowPut(key, value, hash, currentArray);
}
 

public static void main(String[] args) {
    System.out.println(31 - Integer.numberOfLeadingZeros(8));
    System.out.println(32 - Integer.numberOfLeadingZeros(8 - 1));
    int[] array = {1, 2, 3, 4};
    long[] longArray = {1, 2, 3, 4};
    AtomicIntegerArray atomicIntegerArray = new AtomicIntegerArray(array);
    atomicIntegerArray.getAndIncrement(1);

    AtomicLongArray atomicLongArray = new AtomicLongArray(longArray);
    atomicLongArray.getAndIncrement(1);

    Object[] objects = {new Object(), new Object()};
    AtomicReferenceArray atomicReferenceArray = new AtomicReferenceArray(objects);
    atomicReferenceArray.compareAndSet(1, new Object(), new Object());
}
 

@Nonnull
private <TRef extends PsiReference, TResult> Map<TRef, TResult> getMap(boolean physical, int index) {
  AtomicReferenceArray<Map> array = physical ? myPhysicalMaps : myNonPhysicalMaps;
  Map map = array.get(index);
  while (map == null) {
    Map newMap = createWeakMap();
    map = array.compareAndSet(index, null, newMap) ? newMap : array.get(index);
  }
  //noinspection unchecked
  return map;
}
 

private void basicFree(long addr, int idx, AtomicReferenceArray<SyncChunkStack> freeLists) {
  SyncChunkStack clq = freeLists.get(idx);
  if (clq != null) {
    clq.offer(addr);
  } else {
    clq = new SyncChunkStack();
    clq.offer(addr);
    if (!freeLists.compareAndSet(idx, null, clq)) {
      clq = freeLists.get(idx);
      clq.offer(addr);
    }
  }
  
}
 

@Override
public final T pollAndClear(long ptr)
{
  final AtomicReferenceArray<T> ring = _ring;
  final int index = getIndex(ptr);
  
  T value = ring.get(index);
  
  if (value != null && ring.compareAndSet(index, value, null)) {
    return value;
  }
  else {
    return null;
  }
}
 

private void transfer(AtomicReferenceArray src, ResizeContainer resizeContainer)
{
    AtomicReferenceArray dest = resizeContainer.nextArray;

    for (int j = 0; j < src.length() - 1; )
    {
        Object o = src.get(j);
        if (o == null)
        {
            if (src.compareAndSet(j, null, RESIZED))
            {
                j++;
            }
        }
        else if (o == RESIZED || o == RESIZING)
        {
            j = (j & ~(ResizeContainer.QUEUE_INCREMENT - 1)) + ResizeContainer.QUEUE_INCREMENT;
            if (resizeContainer.resizers.get() == 1)
            {
                break;
            }
        }
        else
        {
            Entry e = (Entry) o;
            if (src.compareAndSet(j, o, RESIZING))
            {
                while (e != null)
                {
                    this.unconditionalCopy(dest, e);
                    e = e.getNext();
                }
                src.set(j, RESIZED);
                j++;
            }
        }
    }
    resizeContainer.decrementResizerAndNotify();
    resizeContainer.waitForAllResizers();
}
 

private V slowRemove(int key, int hash, AtomicReferenceArray currentArray)
{
    //noinspection LabeledStatement
    outer:
    while (true)
    {
        int length = currentArray.length();
        int index = ConcurrentIntObjectHashMap.indexFor(hash, length);
        Object o = currentArray.get(index);
        if (o == RESIZED || o == RESIZING)
        {
            currentArray = this.helpWithResizeWhileCurrentIndex(currentArray, index);
        }
        else
        {
            Entry<V> e = (Entry<V>) o;
            while (e != null)
            {
                int candidate = e.getKey();
                if (candidate == key)
                {
                    Entry<V> replacement = this.createReplacementChainForRemoval((Entry<V>) o, e);
                    if (currentArray.compareAndSet(index, o, replacement))
                    {
                        this.addToSize(-1);
                        return e.getValue();
                    }
                    //noinspection ContinueStatementWithLabel
                    continue outer;
                }
                e = e.getNext();
            }
            return null;
        }
    }
}
 

private void unconditionalCopy(AtomicReferenceArray dest, Entry toCopyEntry)
{
    int hash = this.hash(toCopyEntry.getKey());
    AtomicReferenceArray currentArray = dest;
    while (true)
    {
        int length = currentArray.length();
        int index = ConcurrentOffHeapWeakHolder.indexFor(hash, length);
        Object o = currentArray.get(index);
        if (o == RESIZED || o == RESIZING)
        {
            currentArray = ((ResizeContainer) currentArray.get(length - 1)).nextArray;
        }
        else
        {
            Entry newEntry;
            if (o == null)
            {
                if (toCopyEntry.getNext() == null)
                {
                    newEntry = toCopyEntry; // no need to duplicate
                }
                else
                {
                    newEntry = new Entry(toCopyEntry.getKey());
                }
            }
            else
            {
                newEntry = new Entry(toCopyEntry.getKey(), (Entry) o);
            }
            if (currentArray.compareAndSet(index, o, newEntry))
            {
                return;
            }
        }
    }
}
 

public void remove(WeakOffHeapReference key)
{
    int hash = this.hash(key);
    AtomicReferenceArray currentArray = this.table;
    int length = currentArray.length();
    int index = ConcurrentOffHeapWeakHolder.indexFor(hash, length);
    Object o = currentArray.get(index);
    if (o == RESIZED || o == RESIZING)
    {
        this.slowRemove(key, hash, currentArray);
        return;
    }
    Entry e = (Entry) o;
    while (e != null)
    {
        WeakOffHeapReference candidate = e.getKey();
        if (candidate == key)
        {
            Entry replacement = this.createReplacementChainForRemoval((Entry) o, e);
            if (currentArray.compareAndSet(index, o, replacement))
            {
                this.addToSize(-1);
                return;
            }
            this.slowRemove(key, hash, currentArray);
            return;
        }
        e = e.getNext();
    }
}
 

private void slowRemove(WeakOffHeapReference key, int hash, AtomicReferenceArray currentArray)
{
    //noinspection LabeledStatement
    outer:
    while (true)
    {
        int length = currentArray.length();
        int index = ConcurrentOffHeapWeakHolder.indexFor(hash, length);
        Object o = currentArray.get(index);
        if (o == RESIZED || o == RESIZING)
        {
            currentArray = this.helpWithResizeWhileCurrentIndex(currentArray, index);
        }
        else
        {
            Entry e = (Entry) o;
            while (e != null)
            {
                WeakOffHeapReference candidate = e.getKey();
                if (candidate == key)
                {
                    Entry replacement = this.createReplacementChainForRemoval((Entry) o, e);
                    if (currentArray.compareAndSet(index, o, replacement))
                    {
                        this.addToSize(-1);
                        return;
                    }
                    //noinspection ContinueStatementWithLabel
                    continue outer;
                }
                e = e.getNext();
            }
            return;
        }
    }
}
 

private void transfer(AtomicReferenceArray src, ResizeContainer resizeContainer)
{
    AtomicReferenceArray dest = resizeContainer.nextArray;

    for (int j = 0; j < src.length() - 1; )
    {
        Object o = src.get(j);
        if (o == null)
        {
            if (src.compareAndSet(j, null, RESIZED))
            {
                j++;
            }
        }
        else if (o == RESIZED || o == RESIZING)
        {
            j = (j & ~(ResizeContainer.QUEUE_INCREMENT - 1)) + ResizeContainer.QUEUE_INCREMENT;
            if (resizeContainer.resizers.get() == 1)
            {
                break;
            }
        }
        else
        {
            Entry<V> e = (Entry<V>) o;
            if (src.compareAndSet(j, o, RESIZING))
            {
                while (e != null)
                {
                    this.unconditionalCopy(dest, e);
                    e = e.getNext();
                }
                src.set(j, RESIZED);
                j++;
            }
        }
    }
    resizeContainer.decrementResizerAndNotify();
    resizeContainer.waitForAllResizers();
}
 

private void unconditionalCopy(AtomicReferenceArray dest, Entry<V> toCopyEntry)
{
    int hash = this.hash(toCopyEntry.getKey());
    AtomicReferenceArray currentArray = dest;
    while (true)
    {
        int length = currentArray.length();
        int index = ConcurrentIntObjectHashMap.indexFor(hash, length);
        Object o = currentArray.get(index);
        if (o == RESIZED || o == RESIZING)
        {
            currentArray = ((ResizeContainer) currentArray.get(length - 1)).nextArray;
        }
        else
        {
            Entry<V> newEntry;
            if (o == null)
            {
                if (toCopyEntry.getNext() == null)
                {
                    newEntry = toCopyEntry; // no need to duplicate
                }
                else
                {
                    newEntry = new Entry<V>(toCopyEntry.getKey(), toCopyEntry.getValue());
                }
            }
            else
            {
                newEntry = new Entry<V>(toCopyEntry.getKey(), toCopyEntry.getValue(), (Entry<V>) o);
            }
            if (currentArray.compareAndSet(index, o, newEntry))
            {
                return;
            }
        }
    }
}
 

private boolean slowReplace(int key, V oldValue, V newValue, int hash, AtomicReferenceArray currentArray)
{
    //noinspection LabeledStatement
    outer:
    while (true)
    {
        int length = currentArray.length();
        int index = ConcurrentIntObjectHashMap.indexFor(hash, length);
        Object o = currentArray.get(index);
        if (o == RESIZED || o == RESIZING)
        {
            currentArray = this.helpWithResizeWhileCurrentIndex(currentArray, index);
        }
        else
        {
            Entry<V> e = (Entry<V>) o;
            while (e != null)
            {
                int candidate = e.getKey();
                if (candidate == key || candidate == key)
                {
                    if (oldValue == e.getValue() || (oldValue != null && oldValue.equals(e.getValue())))
                    {
                        Entry<V> replacement = this.createReplacementChainForRemoval((Entry<V>) o, e);
                        Entry<V> newEntry = new Entry<V>(key, newValue, replacement);
                        if (currentArray.compareAndSet(index, o, newEntry))
                        {
                            return true;
                        }
                        //noinspection ContinueStatementWithLabel
                        continue outer;
                    }
                    return false;
                }
                e = e.getNext();
            }
            return false;
        }
    }
}
 

public V remove(int key)
{
    int hash = this.hash(key);
    AtomicReferenceArray currentArray = this.table;
    int length = currentArray.length();
    int index = ConcurrentIntObjectHashMap.indexFor(hash, length);
    Object o = currentArray.get(index);
    if (o == RESIZED || o == RESIZING)
    {
        return this.slowRemove(key, hash, currentArray);
    }
    Entry<V> e = (Entry<V>) o;
    while (e != null)
    {
        int candidate = e.getKey();
        if (candidate == key)
        {
            Entry<V> replacement = this.createReplacementChainForRemoval((Entry<V>) o, e);
            if (currentArray.compareAndSet(index, o, replacement))
            {
                this.addToSize(-1);
                return e.getValue();
            }
            return this.slowRemove(key, hash, currentArray);
        }
        e = e.getNext();
    }
    return null;
}
 

public void clear()
{
    AtomicReferenceArray currentArray = this.table;
    ResizeContainer resizeContainer;
    do
    {
        resizeContainer = null;
        for (int i = 0; i < currentArray.length() - 1; i++)
        {
            Object o = currentArray.get(i);
            if (o == RESIZED || o == RESIZING)
            {
                resizeContainer = (ResizeContainer) currentArray.get(currentArray.length() - 1);
            }
            else if (o != null)
            {
                Entry e = (Entry) o;
                if (currentArray.compareAndSet(i, o, null))
                {
                    int removedEntries = 0;
                    while (e != null)
                    {
                        removedEntries++;
                        e = e.getNext();
                    }
                    this.addToSize(-removedEntries);
                }
            }
        }
        if (resizeContainer != null)
        {
            if (resizeContainer.isNotDone())
            {
                this.helpWithResize(currentArray);
                resizeContainer.waitForAllResizers();
            }
            currentArray = resizeContainer.nextArray;
        }
    }
    while (resizeContainer != null);
}
 

private void reverseTransfer(AtomicReferenceArray src, ResizeContainer resizeContainer)
{
    AtomicReferenceArray dest = resizeContainer.nextArray;
    while (resizeContainer.getQueuePosition() > 0)
    {
        int start = resizeContainer.subtractAndGetQueuePosition();
        int end = start + ResizeContainer.QUEUE_INCREMENT;
        if (end > 0)
        {
            if (start < 0)
            {
                start = 0;
            }
            for (int j = end - 1; j >= start; )
            {
                Object o = src.get(j);
                if (o == null)
                {
                    if (src.compareAndSet(j, null, RESIZED))
                    {
                        j--;
                    }
                }
                else if (o == RESIZED || o == RESIZING)
                {
                    resizeContainer.zeroOutQueuePosition();
                    return;
                }
                else
                {
                    Entry<V> e = (Entry<V>) o;
                    if (src.compareAndSet(j, o, RESIZING))
                    {
                        while (e != null)
                        {
                            this.unconditionalCopy(dest, e);
                            e = e.getNext();
                        }
                        src.set(j, RESIZED);
                        j--;
                    }
                }
            }
        }
    }
}
 

public void clear()
{
    AtomicReferenceArray currentArray = this.table;
    ResizeContainer resizeContainer;
    do
    {
        resizeContainer = null;
        for (int i = 0; i < currentArray.length() - 1; i++)
        {
            Object o = currentArray.get(i);
            if (o == RESIZED || o == RESIZING)
            {
                resizeContainer = (ResizeContainer) currentArray.get(currentArray.length() - 1);
            }
            else if (o != null)
            {
                Entry<V> e = (Entry<V>) o;
                if (currentArray.compareAndSet(i, o, null))
                {
                    int removedEntries = 0;
                    while (e != null)
                    {
                        removedEntries++;
                        e = e.getNext();
                    }
                    this.addToSize(-removedEntries);
                }
            }
        }
        if (resizeContainer != null)
        {
            if (resizeContainer.isNotDone())
            {
                this.helpWithResize(currentArray);
                resizeContainer.waitForAllResizers();
            }
            currentArray = resizeContainer.nextArray;
        }
    }
    while (resizeContainer != null);
}