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