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