/*
* Copyright (C) 2007 The Guava Authors
*
* Licensed 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 com.google.common.collect;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import static com.google.common.collect.Multisets.checkNonnegative;
import com.google.common.annotations.Beta;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Serialization.FieldSetter;
import com.google.common.primitives.Ints;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.Nullable;
/**
* A multiset that supports concurrent modifications and that provides atomic
* versions of most {@code Multiset} operations (exceptions where noted). Null
* elements are not supported.
*
* @author Cliff L. Biffle
* @author mike nonemacher
* @since 2.0 (imported from Google Collections Library)
*/
public final class ConcurrentHashMultiset<E> extends AbstractMultiset<E> implements Serializable {
/*
* The ConcurrentHashMultiset's atomic operations are implemented primarily in terms of
* AtomicInteger's atomic operations, with some help from ConcurrentMap's atomic operations on
* creation and removal (including automatic removal of zeroes). If the modification of an
* AtomicInteger results in zero, we compareAndSet the value to zero; if that succeeds, we remove
* the entry from the Map. If another operation sees a zero in the map, it knows that the entry is
* about to be removed, so this operation may remove it (often by replacing it with a new
* AtomicInteger).
*/
/** The number of occurrences of each element. */
private final transient ConcurrentMap<E, AtomicInteger> countMap;
// This constant allows the deserialization code to set a final field. This
// holder class makes sure it is not initialized unless an instance is
// deserialized.
private static class FieldSettersHolder {
static final FieldSetter<ConcurrentHashMultiset> COUNT_MAP_FIELD_SETTER =
Serialization.getFieldSetter(ConcurrentHashMultiset.class, "countMap");
}
/**
* Creates a new, empty {@code ConcurrentHashMultiset} using the default
* initial capacity, load factor, and concurrency settings.
*/
public static <E> ConcurrentHashMultiset<E> create() {
// TODO(schmoe): provide a way to use this class with other (possibly arbitrary)
// ConcurrentMap implementors. One possibility is to extract most of this class into
// an AbstractConcurrentMapMultiset.
return new ConcurrentHashMultiset<E>(new ConcurrentHashMap<E, AtomicInteger>());
}
/**
* Creates a new {@code ConcurrentHashMultiset} containing the specified
* elements, using the default initial capacity, load factor, and concurrency
* settings.
*
* <p>This implementation is highly efficient when {@code elements} is itself
* a {@link Multiset}.
*
* @param elements the elements that the multiset should contain
*/
public static <E> ConcurrentHashMultiset<E> create(Iterable<? extends E> elements) {
ConcurrentHashMultiset<E> multiset = ConcurrentHashMultiset.create();
Iterables.addAll(multiset, elements);
return multiset;
}
/**
* Creates a new, empty {@code ConcurrentHashMultiset} using {@code mapMaker}
* to construct the internal backing map.
*
* <p>If this {@link MapMaker} is configured to use entry eviction of any kind, this eviction
* applies to all occurrences of a given element as a single unit. However, most updates to the
* multiset do not count as map updates at all, since we're usually just mutating the value
* stored in the map, so {@link MapMaker#expireAfterAccess} makes sense (evict the entry that
* was queried or updated longest ago), but {@link MapMaker#expireAfterWrite} doesn't, because
* the eviction time is measured from when we saw the first occurrence of the object.
*
* <p>The returned multiset is serializable but any serialization caveats
* given in {@code MapMaker} apply.
*
* <p>Finally, soft/weak values can be used but are not very useful: the values are created
* internally and not exposed externally, so no one else will have a strong reference to the
* values. Weak keys on the other hand can be useful in some scenarios.
*
* @since 7.0
*/
@Beta
public static <E> ConcurrentHashMultiset<E> create(
GenericMapMaker<? super E, ? super Number> mapMaker) {
return new ConcurrentHashMultiset<E>(mapMaker.<E, AtomicInteger>makeMap());
}
/**
* Creates an instance using {@code countMap} to store elements and their
* counts.
*
* <p>This instance will assume ownership of {@code countMap}, and other code
* should not maintain references to the map or modify it in any way.
*
* @param countMap backing map for storing the elements in the multiset and
* their counts. It must be empty.
* @throws IllegalArgumentException if {@code countMap} is not empty
*/
@VisibleForTesting ConcurrentHashMultiset(ConcurrentMap<E, AtomicInteger> countMap) {
checkArgument(countMap.isEmpty());
this.countMap = countMap;
}
// Query Operations
/**
* Returns the number of occurrences of {@code element} in this multiset.
*
* @param element the element to look for
* @return the nonnegative number of occurrences of the element
*/
@Override public int count(@Nullable Object element) {
AtomicInteger existingCounter = safeGet(element);
return (existingCounter == null) ? 0 : existingCounter.get();
}
/**
* Depending on the type of the underlying map, map.get may throw NullPointerException or
* ClassCastException, if the object is null or of the wrong type. We usually just want to treat
* those cases as if the element isn't in the map, by catching the exceptions and returning null.
*/
private AtomicInteger safeGet(Object element) {
try {
return countMap.get(element);
} catch (NullPointerException e) {
return null;
} catch (ClassCastException e) {
return null;
}
}
/**
* {@inheritDoc}
*
* <p>If the data in the multiset is modified by any other threads during this
* method, it is undefined which (if any) of these modifications will be
* reflected in the result.
*/
@Override public int size() {
long sum = 0L;
for (AtomicInteger value : countMap.values()) {
sum += value.get();
}
return Ints.saturatedCast(sum);
}
/*
* Note: the superclass toArray() methods assume that size() gives a correct
* answer, which ours does not.
*/
@Override public Object[] toArray() {
return snapshot().toArray();
}
@Override public <T> T[] toArray(T[] array) {
return snapshot().toArray(array);
}
/*
* We'd love to use 'new ArrayList(this)' or 'list.addAll(this)', but
* either of these would recurse back to us again!
*/
private List<E> snapshot() {
List<E> list = Lists.newArrayListWithExpectedSize(size());
for (Multiset.Entry<E> entry : entrySet()) {
E element = entry.getElement();
for (int i = entry.getCount(); i > 0; i--) {
list.add(element);
}
}
return list;
}
// Modification Operations
/**
* Adds a number of occurrences of the specified element to this multiset.
*
* @param element the element to add
* @param occurrences the number of occurrences to add
* @return the previous count of the element before the operation; possibly
* zero
* @throws IllegalArgumentException if {@code occurrences} is negative, or if
* the resulting amount would exceed {@link Integer#MAX_VALUE}
*/
@Override public int add(E element, int occurrences) {
if (occurrences == 0) {
return count(element);
}
checkArgument(occurrences > 0, "Invalid occurrences: %s", occurrences);
while (true) {
AtomicInteger existingCounter = safeGet(element);
if (existingCounter == null) {
existingCounter = countMap.putIfAbsent(element, new AtomicInteger(occurrences));
if (existingCounter == null) {
return 0;
}
// existingCounter != null: fall through to operate against the existing AtomicInteger
}
while (true) {
int oldValue = existingCounter.get();
if (oldValue != 0) {
checkArgument(occurrences <= Integer.MAX_VALUE - oldValue,
"Overflow adding %s occurrences to a count of %s",
occurrences, oldValue);
int newValue = oldValue + occurrences;
if (existingCounter.compareAndSet(oldValue, newValue)) {
// newValue can't == 0, so no need to check & remove
return oldValue;
}
} else {
// In the case of a concurrent remove, we might observe a zero value, which means another
// thread is about to remove (element, existingCounter) from the map. Rather than wait,
// we can just do that work here.
AtomicInteger newCounter = new AtomicInteger(occurrences);
if ((countMap.putIfAbsent(element, newCounter) == null)
|| countMap.replace(element, existingCounter, newCounter)) {
return 0;
}
break;
}
}
// If we're still here, there was a race, so just try again.
}
}
/**
* Removes a number of occurrences of the specified element from this
* multiset. If the multiset contains fewer than this number of occurrences to
* begin with, all occurrences will be removed.
*
* @param element the element whose occurrences should be removed
* @param occurrences the number of occurrences of the element to remove
* @return the count of the element before the operation; possibly zero
* @throws IllegalArgumentException if {@code occurrences} is negative
*/
@Override public int remove(@Nullable Object element, int occurrences) {
if (occurrences == 0) {
return count(element);
}
checkArgument(occurrences > 0, "Invalid occurrences: %s", occurrences);
AtomicInteger existingCounter = safeGet(element);
if (existingCounter == null) {
return 0;
}
while (true) {
int oldValue = existingCounter.get();
if (oldValue != 0) {
int newValue = Math.max(0, oldValue - occurrences);
if (existingCounter.compareAndSet(oldValue, newValue)) {
if (newValue == 0) {
// Just CASed to 0; remove the entry to clean up the map. If the removal fails,
// another thread has already replaced it with a new counter, which is fine.
countMap.remove(element, existingCounter);
}
return oldValue;
}
} else {
return 0;
}
}
}
/**
* Removes exactly the specified number of occurrences of {@code element}, or
* makes no change if this is not possible.
*
* <p>This method, in contrast to {@link #remove(Object, int)}, has no effect
* when the element count is smaller than {@code occurrences}.
*
* @param element the element to remove
* @param occurrences the number of occurrences of {@code element} to remove
* @return {@code true} if the removal was possible (including if {@code
* occurrences} is zero)
*/
public boolean removeExactly(@Nullable Object element, int occurrences) {
if (occurrences == 0) {
return true;
}
checkArgument(occurrences > 0, "Invalid occurrences: %s", occurrences);
AtomicInteger existingCounter = safeGet(element);
if (existingCounter == null) {
return false;
}
while (true) {
int oldValue = existingCounter.get();
if (oldValue < occurrences) {
return false;
}
int newValue = oldValue - occurrences;
if (existingCounter.compareAndSet(oldValue, newValue)) {
if (newValue == 0) {
// Just CASed to 0; remove the entry to clean up the map. If the removal fails,
// another thread has already replaced it with a new counter, which is fine.
countMap.remove(element, existingCounter);
}
return true;
}
}
}
/**
* Adds or removes occurrences of {@code element} such that the {@link #count}
* of the element becomes {@code count}.
*
* @return the count of {@code element} in the multiset before this call
* @throws IllegalArgumentException if {@code count} is negative
*/
@Override public int setCount(E element, int count) {
checkNonnegative(count, "count");
while (true) {
AtomicInteger existingCounter = safeGet(element);
if (existingCounter == null) {
if (count == 0) {
return 0;
} else {
existingCounter = countMap.putIfAbsent(element, new AtomicInteger(count));
if (existingCounter == null) {
return 0;
}
// existingCounter != null: fall through
}
}
while (true) {
int oldValue = existingCounter.get();
if (oldValue == 0) {
if (count == 0) {
return 0;
} else {
AtomicInteger newCounter = new AtomicInteger(count);
if ((countMap.putIfAbsent(element, newCounter) == null)
|| countMap.replace(element, existingCounter, newCounter)) {
return 0;
}
}
break;
} else {
if (existingCounter.compareAndSet(oldValue, count)) {
if (count == 0) {
// Just CASed to 0; remove the entry to clean up the map. If the removal fails,
// another thread has already replaced it with a new counter, which is fine.
countMap.remove(element, existingCounter);
}
return oldValue;
}
}
}
}
}
/**
* Sets the number of occurrences of {@code element} to {@code newCount}, but
* only if the count is currently {@code expectedOldCount}. If {@code element} does
* not appear in the multiset exactly {@code expectedOldCount} times, no changes will
* be made.
*
* @return {@code true} if the change was successful. This usually indicates
* that the multiset has been modified, but not always: in the case that
* {@code expectedOldCount == newCount}, the method will return {@code true} if
* the condition was met.
* @throws IllegalArgumentException if {@code expectedOldCount} or {@code newCount} is negative
*/
@Override public boolean setCount(E element, int expectedOldCount, int newCount) {
checkNonnegative(expectedOldCount, "oldCount");
checkNonnegative(newCount, "newCount");
AtomicInteger existingCounter = safeGet(element);
if (existingCounter == null) {
if (expectedOldCount != 0) {
return false;
} else if (newCount == 0) {
return true;
} else {
// if our write lost the race, it must have lost to a nonzero value, so we can stop
return countMap.putIfAbsent(element, new AtomicInteger(newCount)) == null;
}
}
int oldValue = existingCounter.get();
if (oldValue == expectedOldCount) {
if (oldValue == 0) {
if (newCount == 0) {
// Just observed a 0; try to remove the entry to clean up the map
countMap.remove(element, existingCounter);
return true;
} else {
AtomicInteger newCounter = new AtomicInteger(newCount);
return (countMap.putIfAbsent(element, newCounter) == null)
|| countMap.replace(element, existingCounter, newCounter);
}
} else {
if (existingCounter.compareAndSet(oldValue, newCount)) {
if (newCount == 0) {
// Just CASed to 0; remove the entry to clean up the map. If the removal fails,
// another thread has already replaced it with a new counter, which is fine.
countMap.remove(element, existingCounter);
}
return true;
}
}
}
return false;
}
// Views
@Override Set<E> createElementSet() {
final Set<E> delegate = countMap.keySet();
return new ForwardingSet<E>() {
@Override protected Set<E> delegate() {
return delegate;
}
@Override public boolean remove(Object object) {
try {
return delegate.remove(object);
} catch (NullPointerException e) {
return false;
} catch (ClassCastException e) {
return false;
}
}
};
}
private transient EntrySet entrySet;
@Override public Set<Multiset.Entry<E>> entrySet() {
EntrySet result = entrySet;
if (result == null) {
entrySet = result = new EntrySet();
}
return result;
}
@Override int distinctElements() {
return countMap.size();
}
@Override public boolean isEmpty() {
return countMap.isEmpty();
}
@Override Iterator<Entry<E>> entryIterator() {
// AbstractIterator makes this fairly clean, but it doesn't support remove(). To support
// remove(), we create an AbstractIterator, and then use ForwardingIterator to delegate to it.
final Iterator<Entry<E>> readOnlyIterator =
new AbstractIterator<Entry<E>>() {
private Iterator<Map.Entry<E, AtomicInteger>> mapEntries = countMap.entrySet().iterator();
@Override protected Entry<E> computeNext() {
while (true) {
if (!mapEntries.hasNext()) {
return endOfData();
}
Map.Entry<E, AtomicInteger> mapEntry = mapEntries.next();
int count = mapEntry.getValue().get();
if (count != 0) {
return Multisets.immutableEntry(mapEntry.getKey(), count);
}
}
}
};
return new ForwardingIterator<Entry<E>>() {
private Entry<E> last;
@Override protected Iterator<Entry<E>> delegate() {
return readOnlyIterator;
}
@Override public Entry<E> next() {
last = super.next();
return last;
}
@Override public void remove() {
checkState(last != null);
ConcurrentHashMultiset.this.setCount(last.getElement(), 0);
last = null;
}
};
}
@Override public void clear() {
countMap.clear();
}
private class EntrySet extends AbstractMultiset<E>.EntrySet {
@Override ConcurrentHashMultiset<E> multiset() {
return ConcurrentHashMultiset.this;
}
/*
* Note: the superclass toArray() methods assume that size() gives a correct
* answer, which ours does not.
*/
@Override public Object[] toArray() {
return snapshot().toArray();
}
@Override public <T> T[] toArray(T[] array) {
return snapshot().toArray(array);
}
private List<Multiset.Entry<E>> snapshot() {
List<Multiset.Entry<E>> list = Lists.newArrayListWithExpectedSize(size());
// Not Iterables.addAll(list, this), because that'll forward right back here.
Iterators.addAll(list, iterator());
return list;
}
@Override public boolean remove(Object object) {
if (object instanceof Multiset.Entry) {
Multiset.Entry<?> entry = (Multiset.Entry<?>) object;
Object element = entry.getElement();
int entryCount = entry.getCount();
if (entryCount != 0) {
// Safe as long as we never add a new entry, which we won't.
@SuppressWarnings("unchecked")
Multiset<Object> multiset = (Multiset) multiset();
return multiset.setCount(element, entryCount, 0);
}
}
return false;
}
}
/**
* @serialData the ConcurrentMap of elements and their counts.
*/
private void writeObject(ObjectOutputStream stream) throws IOException {
stream.defaultWriteObject();
stream.writeObject(countMap);
}
private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
stream.defaultReadObject();
@SuppressWarnings("unchecked") // reading data stored by writeObject
ConcurrentMap<E, Integer> deserializedCountMap =
(ConcurrentMap<E, Integer>) stream.readObject();
FieldSettersHolder.COUNT_MAP_FIELD_SETTER.set(this, deserializedCountMap);
}
private static final long serialVersionUID = 1;
}
|