001/*
002 * Copyright (C) 2007 The Guava Authors
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 * http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016
017package com.google.common.collect;
018
019import static com.google.common.base.Preconditions.checkArgument;
020import static com.google.common.base.Preconditions.checkNotNull;
021import static com.google.common.collect.CollectPreconditions.checkNonnegative;
022import static com.google.common.collect.CollectPreconditions.checkRemove;
023
024import com.google.common.annotations.Beta;
025import com.google.common.annotations.GwtIncompatible;
026import com.google.common.annotations.VisibleForTesting;
027import com.google.common.collect.Serialization.FieldSetter;
028import com.google.common.math.IntMath;
029import com.google.common.primitives.Ints;
030import com.google.errorprone.annotations.CanIgnoreReturnValue;
031import com.google.j2objc.annotations.WeakOuter;
032import java.io.IOException;
033import java.io.ObjectInputStream;
034import java.io.ObjectOutputStream;
035import java.io.Serializable;
036import java.util.Collection;
037import java.util.Iterator;
038import java.util.List;
039import java.util.Map;
040import java.util.Set;
041import java.util.concurrent.ConcurrentHashMap;
042import java.util.concurrent.ConcurrentMap;
043import java.util.concurrent.atomic.AtomicInteger;
044import javax.annotation.Nullable;
045
046/**
047 * A multiset that supports concurrent modifications and that provides atomic versions of most
048 * {@code Multiset} operations (exceptions where noted). Null elements are not supported.
049 *
050 * <p>See the Guava User Guide article on <a href=
051 * "https://github.com/google/guava/wiki/NewCollectionTypesExplained#multiset">
052 * {@code Multiset}</a>.
053 *
054 * @author Cliff L. Biffle
055 * @author mike nonemacher
056 * @since 2.0
057 */
058@GwtIncompatible
059public final class ConcurrentHashMultiset<E> extends AbstractMultiset<E> implements Serializable {
060
061  /*
062   * The ConcurrentHashMultiset's atomic operations are implemented primarily in terms of
063   * AtomicInteger's atomic operations, with some help from ConcurrentMap's atomic operations on
064   * creation and removal (including automatic removal of zeroes). If the modification of an
065   * AtomicInteger results in zero, we compareAndSet the value to zero; if that succeeds, we remove
066   * the entry from the Map. If another operation sees a zero in the map, it knows that the entry is
067   * about to be removed, so this operation may remove it (often by replacing it with a new
068   * AtomicInteger).
069   */
070
071  /** The number of occurrences of each element. */
072  private final transient ConcurrentMap<E, AtomicInteger> countMap;
073
074  // This constant allows the deserialization code to set a final field. This holder class
075  // makes sure it is not initialized unless an instance is deserialized.
076  private static class FieldSettersHolder {
077    static final FieldSetter<ConcurrentHashMultiset> COUNT_MAP_FIELD_SETTER =
078        Serialization.getFieldSetter(ConcurrentHashMultiset.class, "countMap");
079  }
080
081  /**
082   * Creates a new, empty {@code ConcurrentHashMultiset} using the default
083   * initial capacity, load factor, and concurrency settings.
084   */
085  public static <E> ConcurrentHashMultiset<E> create() {
086    // TODO(schmoe): provide a way to use this class with other (possibly arbitrary)
087    // ConcurrentMap implementors. One possibility is to extract most of this class into
088    // an AbstractConcurrentMapMultiset.
089    return new ConcurrentHashMultiset<E>(new ConcurrentHashMap<E, AtomicInteger>());
090  }
091
092  /**
093   * Creates a new {@code ConcurrentHashMultiset} containing the specified elements, using
094   * the default initial capacity, load factor, and concurrency settings.
095   *
096   * <p>This implementation is highly efficient when {@code elements} is itself a {@link Multiset}.
097   *
098   * @param elements the elements that the multiset should contain
099   */
100  public static <E> ConcurrentHashMultiset<E> create(Iterable<? extends E> elements) {
101    ConcurrentHashMultiset<E> multiset = ConcurrentHashMultiset.create();
102    Iterables.addAll(multiset, elements);
103    return multiset;
104  }
105
106  /**
107   * Creates a new, empty {@code ConcurrentHashMultiset} using {@code mapMaker} to construct the
108   * internal backing map.
109   *
110   * <p>If this {@link MapMaker} is configured to use entry eviction of any kind, this eviction
111   * applies to all occurrences of a given element as a single unit. However, most updates to the
112   * multiset do not count as map updates at all, since we're usually just mutating the value stored
113   * in the map, so {@link MapMaker#expireAfterAccess} makes sense (evict the entry that was queried
114   * or updated longest ago), but {@link MapMaker#expireAfterWrite} doesn't, because the eviction
115   * time is measured from when we saw the first occurrence of the object.
116   *
117   * <p>The returned multiset is serializable but any serialization caveats given in {@code
118   * MapMaker} apply.
119   *
120   * <p>Finally, soft/weak values can be used but are not very useful: the values are created
121   * internally and not exposed externally, so no one else will have a strong reference to the
122   * values. Weak keys on the other hand can be useful in some scenarios.
123   *
124   * @since 15.0 (source compatible (accepting the since removed {@code GenericMapMaker} class)
125   *     since 7.0)
126   * @deprecated Use {@link #create(ConcurrentMap)} instead. This method is scheduled for deletion
127   *     in Guava 21.0.
128   */
129  @Beta
130  @Deprecated
131  public static <E> ConcurrentHashMultiset<E> create(MapMaker mapMaker) {
132    return create(mapMaker.<E, AtomicInteger>makeMap());
133  }
134
135  /**
136   * Creates a new, empty {@code ConcurrentHashMultiset} using {@code countMap} as the internal
137   * backing map.
138   *
139   * <p>This instance will assume ownership of {@code countMap}, and other code should not maintain
140   * references to the map or modify it in any way.
141   *
142   * <p>The returned multiset is serializable if the input map is.
143   *
144   * @param countMap backing map for storing the elements in the multiset and their counts. It must
145   *     be empty.
146   * @throws IllegalArgumentException if {@code countMap} is not empty
147   * @since 20.0
148   */
149  @Beta
150  public static <E> ConcurrentHashMultiset<E> create(ConcurrentMap<E, AtomicInteger> countMap) {
151    return new ConcurrentHashMultiset<E>(countMap);
152  }
153
154  @VisibleForTesting
155  ConcurrentHashMultiset(ConcurrentMap<E, AtomicInteger> countMap) {
156    checkArgument(countMap.isEmpty(), "the backing map (%s) must be empty", countMap);
157    this.countMap = countMap;
158  }
159
160  // Query Operations
161
162  /**
163   * Returns the number of occurrences of {@code element} in this multiset.
164   *
165   * @param element the element to look for
166   * @return the nonnegative number of occurrences of the element
167   */
168  @Override
169  public int count(@Nullable Object element) {
170    AtomicInteger existingCounter = Maps.safeGet(countMap, element);
171    return (existingCounter == null) ? 0 : existingCounter.get();
172  }
173
174  /**
175   * {@inheritDoc}
176   *
177   * <p>If the data in the multiset is modified by any other threads during this method,
178   * it is undefined which (if any) of these modifications will be reflected in the result.
179   */
180  @Override
181  public int size() {
182    long sum = 0L;
183    for (AtomicInteger value : countMap.values()) {
184      sum += value.get();
185    }
186    return Ints.saturatedCast(sum);
187  }
188
189  /*
190   * Note: the superclass toArray() methods assume that size() gives a correct
191   * answer, which ours does not.
192   */
193
194  @Override
195  public Object[] toArray() {
196    return snapshot().toArray();
197  }
198
199  @Override
200  public <T> T[] toArray(T[] array) {
201    return snapshot().toArray(array);
202  }
203
204  /*
205   * We'd love to use 'new ArrayList(this)' or 'list.addAll(this)', but
206   * either of these would recurse back to us again!
207   */
208  private List<E> snapshot() {
209    List<E> list = Lists.newArrayListWithExpectedSize(size());
210    for (Multiset.Entry<E> entry : entrySet()) {
211      E element = entry.getElement();
212      for (int i = entry.getCount(); i > 0; i--) {
213        list.add(element);
214      }
215    }
216    return list;
217  }
218
219  // Modification Operations
220
221  /**
222   * Adds a number of occurrences of the specified element to this multiset.
223   *
224   * @param element the element to add
225   * @param occurrences the number of occurrences to add
226   * @return the previous count of the element before the operation; possibly zero
227   * @throws IllegalArgumentException if {@code occurrences} is negative, or if
228   *     the resulting amount would exceed {@link Integer#MAX_VALUE}
229   */
230  @CanIgnoreReturnValue
231  @Override
232  public int add(E element, int occurrences) {
233    checkNotNull(element);
234    if (occurrences == 0) {
235      return count(element);
236    }
237    CollectPreconditions.checkPositive(occurrences, "occurences");
238
239    while (true) {
240      AtomicInteger existingCounter = Maps.safeGet(countMap, element);
241      if (existingCounter == null) {
242        existingCounter = countMap.putIfAbsent(element, new AtomicInteger(occurrences));
243        if (existingCounter == null) {
244          return 0;
245        }
246        // existingCounter != null: fall through to operate against the existing AtomicInteger
247      }
248
249      while (true) {
250        int oldValue = existingCounter.get();
251        if (oldValue != 0) {
252          try {
253            int newValue = IntMath.checkedAdd(oldValue, occurrences);
254            if (existingCounter.compareAndSet(oldValue, newValue)) {
255              // newValue can't == 0, so no need to check & remove
256              return oldValue;
257            }
258          } catch (ArithmeticException overflow) {
259            throw new IllegalArgumentException(
260                "Overflow adding " + occurrences + " occurrences to a count of " + oldValue);
261          }
262        } else {
263          // In the case of a concurrent remove, we might observe a zero value, which means another
264          // thread is about to remove (element, existingCounter) from the map. Rather than wait,
265          // we can just do that work here.
266          AtomicInteger newCounter = new AtomicInteger(occurrences);
267          if ((countMap.putIfAbsent(element, newCounter) == null)
268              || countMap.replace(element, existingCounter, newCounter)) {
269            return 0;
270          }
271          break;
272        }
273      }
274
275      // If we're still here, there was a race, so just try again.
276    }
277  }
278
279  /**
280   * Removes a number of occurrences of the specified element from this multiset. If the multiset
281   * contains fewer than this number of occurrences to begin with, all occurrences will be removed.
282   *
283   * @param element the element whose occurrences should be removed
284   * @param occurrences the number of occurrences of the element to remove
285   * @return the count of the element before the operation; possibly zero
286   * @throws IllegalArgumentException if {@code occurrences} is negative
287   */
288  /*
289   * TODO(cpovirk): remove and removeExactly currently accept null inputs only
290   * if occurrences == 0. This satisfies both NullPointerTester and
291   * CollectionRemoveTester.testRemove_nullAllowed, but it's not clear that it's
292   * a good policy, especially because, in order for the test to pass, the
293   * parameter must be misleadingly annotated as @Nullable. I suspect that
294   * we'll want to remove @Nullable, add an eager checkNotNull, and loosen up
295   * testRemove_nullAllowed.
296   */
297  @CanIgnoreReturnValue
298  @Override
299  public int remove(@Nullable Object element, int occurrences) {
300    if (occurrences == 0) {
301      return count(element);
302    }
303    CollectPreconditions.checkPositive(occurrences, "occurences");
304
305    AtomicInteger existingCounter = Maps.safeGet(countMap, element);
306    if (existingCounter == null) {
307      return 0;
308    }
309    while (true) {
310      int oldValue = existingCounter.get();
311      if (oldValue != 0) {
312        int newValue = Math.max(0, oldValue - occurrences);
313        if (existingCounter.compareAndSet(oldValue, newValue)) {
314          if (newValue == 0) {
315            // Just CASed to 0; remove the entry to clean up the map. If the removal fails,
316            // another thread has already replaced it with a new counter, which is fine.
317            countMap.remove(element, existingCounter);
318          }
319          return oldValue;
320        }
321      } else {
322        return 0;
323      }
324    }
325  }
326
327  /**
328   * Removes exactly the specified number of occurrences of {@code element}, or makes no
329   * change if this is not possible.
330   *
331   * <p>This method, in contrast to {@link #remove(Object, int)}, has no effect when the
332   * element count is smaller than {@code occurrences}.
333   *
334   * @param element the element to remove
335   * @param occurrences the number of occurrences of {@code element} to remove
336   * @return {@code true} if the removal was possible (including if {@code occurrences} is zero)
337   * @throws IllegalArgumentException if {@code occurrences} is negative
338   */
339  @CanIgnoreReturnValue
340  public boolean removeExactly(@Nullable Object element, int occurrences) {
341    if (occurrences == 0) {
342      return true;
343    }
344    CollectPreconditions.checkPositive(occurrences, "occurences");
345
346    AtomicInteger existingCounter = Maps.safeGet(countMap, element);
347    if (existingCounter == null) {
348      return false;
349    }
350    while (true) {
351      int oldValue = existingCounter.get();
352      if (oldValue < occurrences) {
353        return false;
354      }
355      int newValue = oldValue - occurrences;
356      if (existingCounter.compareAndSet(oldValue, newValue)) {
357        if (newValue == 0) {
358          // Just CASed to 0; remove the entry to clean up the map. If the removal fails,
359          // another thread has already replaced it with a new counter, which is fine.
360          countMap.remove(element, existingCounter);
361        }
362        return true;
363      }
364    }
365  }
366
367  /**
368   * Adds or removes occurrences of {@code element} such that the {@link #count} of the
369   * element becomes {@code count}.
370   *
371   * @return the count of {@code element} in the multiset before this call
372   * @throws IllegalArgumentException if {@code count} is negative
373   */
374  @CanIgnoreReturnValue
375  @Override
376  public int setCount(E element, int count) {
377    checkNotNull(element);
378    checkNonnegative(count, "count");
379    while (true) {
380      AtomicInteger existingCounter = Maps.safeGet(countMap, element);
381      if (existingCounter == null) {
382        if (count == 0) {
383          return 0;
384        } else {
385          existingCounter = countMap.putIfAbsent(element, new AtomicInteger(count));
386          if (existingCounter == null) {
387            return 0;
388          }
389          // existingCounter != null: fall through
390        }
391      }
392
393      while (true) {
394        int oldValue = existingCounter.get();
395        if (oldValue == 0) {
396          if (count == 0) {
397            return 0;
398          } else {
399            AtomicInteger newCounter = new AtomicInteger(count);
400            if ((countMap.putIfAbsent(element, newCounter) == null)
401                || countMap.replace(element, existingCounter, newCounter)) {
402              return 0;
403            }
404          }
405          break;
406        } else {
407          if (existingCounter.compareAndSet(oldValue, count)) {
408            if (count == 0) {
409              // Just CASed to 0; remove the entry to clean up the map. If the removal fails,
410              // another thread has already replaced it with a new counter, which is fine.
411              countMap.remove(element, existingCounter);
412            }
413            return oldValue;
414          }
415        }
416      }
417    }
418  }
419
420  /**
421   * Sets the number of occurrences of {@code element} to {@code newCount}, but only if
422   * the count is currently {@code expectedOldCount}. If {@code element} does not appear
423   * in the multiset exactly {@code expectedOldCount} times, no changes will be made.
424   *
425   * @return {@code true} if the change was successful. This usually indicates
426   *     that the multiset has been modified, but not always: in the case that
427   *     {@code expectedOldCount == newCount}, the method will return {@code true} if
428   *     the condition was met.
429   * @throws IllegalArgumentException if {@code expectedOldCount} or {@code newCount} is negative
430   */
431  @CanIgnoreReturnValue
432  @Override
433  public boolean setCount(E element, int expectedOldCount, int newCount) {
434    checkNotNull(element);
435    checkNonnegative(expectedOldCount, "oldCount");
436    checkNonnegative(newCount, "newCount");
437
438    AtomicInteger existingCounter = Maps.safeGet(countMap, element);
439    if (existingCounter == null) {
440      if (expectedOldCount != 0) {
441        return false;
442      } else if (newCount == 0) {
443        return true;
444      } else {
445        // if our write lost the race, it must have lost to a nonzero value, so we can stop
446        return countMap.putIfAbsent(element, new AtomicInteger(newCount)) == null;
447      }
448    }
449    int oldValue = existingCounter.get();
450    if (oldValue == expectedOldCount) {
451      if (oldValue == 0) {
452        if (newCount == 0) {
453          // Just observed a 0; try to remove the entry to clean up the map
454          countMap.remove(element, existingCounter);
455          return true;
456        } else {
457          AtomicInteger newCounter = new AtomicInteger(newCount);
458          return (countMap.putIfAbsent(element, newCounter) == null)
459              || countMap.replace(element, existingCounter, newCounter);
460        }
461      } else {
462        if (existingCounter.compareAndSet(oldValue, newCount)) {
463          if (newCount == 0) {
464            // Just CASed to 0; remove the entry to clean up the map. If the removal fails,
465            // another thread has already replaced it with a new counter, which is fine.
466            countMap.remove(element, existingCounter);
467          }
468          return true;
469        }
470      }
471    }
472    return false;
473  }
474
475  // Views
476
477  @Override
478  Set<E> createElementSet() {
479    final Set<E> delegate = countMap.keySet();
480    return new ForwardingSet<E>() {
481      @Override
482      protected Set<E> delegate() {
483        return delegate;
484      }
485
486      @Override
487      public boolean contains(@Nullable Object object) {
488        return object != null && Collections2.safeContains(delegate, object);
489      }
490
491      @Override
492      public boolean containsAll(Collection<?> collection) {
493        return standardContainsAll(collection);
494      }
495
496      @Override
497      public boolean remove(Object object) {
498        return object != null && Collections2.safeRemove(delegate, object);
499      }
500
501      @Override
502      public boolean removeAll(Collection<?> c) {
503        return standardRemoveAll(c);
504      }
505    };
506  }
507
508  @Override
509  public Set<Multiset.Entry<E>> createEntrySet() {
510    return new EntrySet();
511  }
512
513  @Override
514  int distinctElements() {
515    return countMap.size();
516  }
517
518  @Override
519  public boolean isEmpty() {
520    return countMap.isEmpty();
521  }
522
523  @Override
524  Iterator<Entry<E>> entryIterator() {
525    // AbstractIterator makes this fairly clean, but it doesn't support remove(). To support
526    // remove(), we create an AbstractIterator, and then use ForwardingIterator to delegate to it.
527    final Iterator<Entry<E>> readOnlyIterator =
528        new AbstractIterator<Entry<E>>() {
529          private final Iterator<Map.Entry<E, AtomicInteger>> mapEntries =
530              countMap.entrySet().iterator();
531
532          @Override
533          protected Entry<E> computeNext() {
534            while (true) {
535              if (!mapEntries.hasNext()) {
536                return endOfData();
537              }
538              Map.Entry<E, AtomicInteger> mapEntry = mapEntries.next();
539              int count = mapEntry.getValue().get();
540              if (count != 0) {
541                return Multisets.immutableEntry(mapEntry.getKey(), count);
542              }
543            }
544          }
545        };
546
547    return new ForwardingIterator<Entry<E>>() {
548      private Entry<E> last;
549
550      @Override
551      protected Iterator<Entry<E>> delegate() {
552        return readOnlyIterator;
553      }
554
555      @Override
556      public Entry<E> next() {
557        last = super.next();
558        return last;
559      }
560
561      @Override
562      public void remove() {
563        checkRemove(last != null);
564        ConcurrentHashMultiset.this.setCount(last.getElement(), 0);
565        last = null;
566      }
567    };
568  }
569
570  @Override
571  public void clear() {
572    countMap.clear();
573  }
574
575  @WeakOuter
576  private class EntrySet extends AbstractMultiset<E>.EntrySet {
577    @Override
578    ConcurrentHashMultiset<E> multiset() {
579      return ConcurrentHashMultiset.this;
580    }
581
582    /*
583     * Note: the superclass toArray() methods assume that size() gives a correct
584     * answer, which ours does not.
585     */
586
587    @Override
588    public Object[] toArray() {
589      return snapshot().toArray();
590    }
591
592    @Override
593    public <T> T[] toArray(T[] array) {
594      return snapshot().toArray(array);
595    }
596
597    private List<Multiset.Entry<E>> snapshot() {
598      List<Multiset.Entry<E>> list = Lists.newArrayListWithExpectedSize(size());
599      // Not Iterables.addAll(list, this), because that'll forward right back here.
600      Iterators.addAll(list, iterator());
601      return list;
602    }
603  }
604
605  /**
606   * @serialData the ConcurrentMap of elements and their counts.
607   */
608  private void writeObject(ObjectOutputStream stream) throws IOException {
609    stream.defaultWriteObject();
610    stream.writeObject(countMap);
611  }
612
613  private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
614    stream.defaultReadObject();
615    @SuppressWarnings("unchecked") // reading data stored by writeObject
616    ConcurrentMap<E, Integer> deserializedCountMap =
617        (ConcurrentMap<E, Integer>) stream.readObject();
618    FieldSettersHolder.COUNT_MAP_FIELD_SETTER.set(this, deserializedCountMap);
619  }
620
621  private static final long serialVersionUID = 1;
622}