• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1 /*
2  * Copyright (C) 2007 The Guava Authors
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  * http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16 
17 package com.google.common.collect;
18 
19 import static com.google.common.base.Preconditions.checkArgument;
20 import static com.google.common.base.Preconditions.checkNotNull;
21 import static com.google.common.collect.CollectPreconditions.checkNonnegative;
22 import static com.google.common.collect.CollectPreconditions.checkRemove;
23 
24 import com.google.common.annotations.Beta;
25 import com.google.common.annotations.VisibleForTesting;
26 import com.google.common.collect.Serialization.FieldSetter;
27 import com.google.common.math.IntMath;
28 import com.google.common.primitives.Ints;
29 
30 import java.io.IOException;
31 import java.io.ObjectInputStream;
32 import java.io.ObjectOutputStream;
33 import java.io.Serializable;
34 import java.util.Collection;
35 import java.util.Iterator;
36 import java.util.List;
37 import java.util.Map;
38 import java.util.Set;
39 import java.util.concurrent.ConcurrentHashMap;
40 import java.util.concurrent.ConcurrentMap;
41 import java.util.concurrent.atomic.AtomicInteger;
42 
43 import javax.annotation.Nullable;
44 
45 /**
46  * A multiset that supports concurrent modifications and that provides atomic versions of most
47  * {@code Multiset} operations (exceptions where noted). Null elements are not supported.
48  *
49  * <p>See the Guava User Guide article on <a href=
50  * "http://code.google.com/p/guava-libraries/wiki/NewCollectionTypesExplained#Multiset">
51  * {@code Multiset}</a>.
52  *
53  * @author Cliff L. Biffle
54  * @author mike nonemacher
55  * @since 2.0 (imported from Google Collections Library)
56  */
57 public final class ConcurrentHashMultiset<E> extends AbstractMultiset<E> implements Serializable {
58 
59   /*
60    * The ConcurrentHashMultiset's atomic operations are implemented primarily in terms of
61    * AtomicInteger's atomic operations, with some help from ConcurrentMap's atomic operations on
62    * creation and removal (including automatic removal of zeroes). If the modification of an
63    * AtomicInteger results in zero, we compareAndSet the value to zero; if that succeeds, we remove
64    * the entry from the Map. If another operation sees a zero in the map, it knows that the entry is
65    * about to be removed, so this operation may remove it (often by replacing it with a new
66    * AtomicInteger).
67    */
68 
69   /** The number of occurrences of each element. */
70   private final transient ConcurrentMap<E, AtomicInteger> countMap;
71 
72   // This constant allows the deserialization code to set a final field. This holder class
73   // makes sure it is not initialized unless an instance is deserialized.
74   private static class FieldSettersHolder {
75     static final FieldSetter<ConcurrentHashMultiset> COUNT_MAP_FIELD_SETTER =
76         Serialization.getFieldSetter(ConcurrentHashMultiset.class, "countMap");
77   }
78 
79   /**
80    * Creates a new, empty {@code ConcurrentHashMultiset} using the default
81    * initial capacity, load factor, and concurrency settings.
82    */
create()83   public static <E> ConcurrentHashMultiset<E> create() {
84     // TODO(schmoe): provide a way to use this class with other (possibly arbitrary)
85     // ConcurrentMap implementors. One possibility is to extract most of this class into
86     // an AbstractConcurrentMapMultiset.
87     return new ConcurrentHashMultiset<E>(new ConcurrentHashMap<E, AtomicInteger>());
88   }
89 
90   /**
91    * Creates a new {@code ConcurrentHashMultiset} containing the specified elements, using
92    * the default initial capacity, load factor, and concurrency settings.
93    *
94    * <p>This implementation is highly efficient when {@code elements} is itself a {@link Multiset}.
95    *
96    * @param elements the elements that the multiset should contain
97    */
create(Iterable<? extends E> elements)98   public static <E> ConcurrentHashMultiset<E> create(Iterable<? extends E> elements) {
99     ConcurrentHashMultiset<E> multiset = ConcurrentHashMultiset.create();
100     Iterables.addAll(multiset, elements);
101     return multiset;
102   }
103 
104   /**
105    * Creates a new, empty {@code ConcurrentHashMultiset} using {@code mapMaker}
106    * to construct the internal backing map.
107    *
108    * <p>If this {@link MapMaker} is configured to use entry eviction of any kind, this eviction
109    * applies to all occurrences of a given element as a single unit. However, most updates to the
110    * multiset do not count as map updates at all, since we're usually just mutating the value
111    * stored in the map, so {@link MapMaker#expireAfterAccess} makes sense (evict the entry that
112    * was queried or updated longest ago), but {@link MapMaker#expireAfterWrite} doesn't, because
113    * the eviction time is measured from when we saw the first occurrence of the object.
114    *
115    * <p>The returned multiset is serializable but any serialization caveats
116    * given in {@code MapMaker} apply.
117    *
118    * <p>Finally, soft/weak values can be used but are not very useful: the values are created
119    * internally and not exposed externally, so no one else will have a strong reference to the
120    * values. Weak keys on the other hand can be useful in some scenarios.
121    *
122    * @since 15.0 (source compatible (accepting the since removed {@code GenericMapMaker} class)
123    *     since 7.0)
124    */
125   @Beta
create(MapMaker mapMaker)126   public static <E> ConcurrentHashMultiset<E> create(MapMaker mapMaker) {
127     return new ConcurrentHashMultiset<E>(mapMaker.<E, AtomicInteger>makeMap());
128   }
129 
130   /**
131    * Creates an instance using {@code countMap} to store elements and their counts.
132    *
133    * <p>This instance will assume ownership of {@code countMap}, and other code
134    * should not maintain references to the map or modify it in any way.
135    *
136    * @param countMap backing map for storing the elements in the multiset and
137    *     their counts. It must be empty.
138    * @throws IllegalArgumentException if {@code countMap} is not empty
139    */
ConcurrentHashMultiset(ConcurrentMap<E, AtomicInteger> countMap)140   @VisibleForTesting ConcurrentHashMultiset(ConcurrentMap<E, AtomicInteger> countMap) {
141     checkArgument(countMap.isEmpty());
142     this.countMap = countMap;
143   }
144 
145   // Query Operations
146 
147   /**
148    * Returns the number of occurrences of {@code element} in this multiset.
149    *
150    * @param element the element to look for
151    * @return the nonnegative number of occurrences of the element
152    */
count(@ullable Object element)153   @Override public int count(@Nullable Object element) {
154     AtomicInteger existingCounter = Maps.safeGet(countMap, element);
155     return (existingCounter == null) ? 0 : existingCounter.get();
156   }
157 
158   /**
159    * {@inheritDoc}
160    *
161    * <p>If the data in the multiset is modified by any other threads during this method,
162    * it is undefined which (if any) of these modifications will be reflected in the result.
163    */
size()164   @Override public int size() {
165     long sum = 0L;
166     for (AtomicInteger value : countMap.values()) {
167       sum += value.get();
168     }
169     return Ints.saturatedCast(sum);
170   }
171 
172   /*
173    * Note: the superclass toArray() methods assume that size() gives a correct
174    * answer, which ours does not.
175    */
176 
toArray()177   @Override public Object[] toArray() {
178     return snapshot().toArray();
179   }
180 
toArray(T[] array)181   @Override public <T> T[] toArray(T[] array) {
182     return snapshot().toArray(array);
183   }
184 
185   /*
186    * We'd love to use 'new ArrayList(this)' or 'list.addAll(this)', but
187    * either of these would recurse back to us again!
188    */
snapshot()189   private List<E> snapshot() {
190     List<E> list = Lists.newArrayListWithExpectedSize(size());
191     for (Multiset.Entry<E> entry : entrySet()) {
192       E element = entry.getElement();
193       for (int i = entry.getCount(); i > 0; i--) {
194         list.add(element);
195       }
196     }
197     return list;
198   }
199 
200   // Modification Operations
201 
202   /**
203    * Adds a number of occurrences of the specified element to this multiset.
204    *
205    * @param element the element to add
206    * @param occurrences the number of occurrences to add
207    * @return the previous count of the element before the operation; possibly zero
208    * @throws IllegalArgumentException if {@code occurrences} is negative, or if
209    *     the resulting amount would exceed {@link Integer#MAX_VALUE}
210    */
add(E element, int occurrences)211   @Override public int add(E element, int occurrences) {
212     checkNotNull(element);
213     if (occurrences == 0) {
214       return count(element);
215     }
216     checkArgument(occurrences > 0, "Invalid occurrences: %s", occurrences);
217 
218     while (true) {
219       AtomicInteger existingCounter = Maps.safeGet(countMap, element);
220       if (existingCounter == null) {
221         existingCounter = countMap.putIfAbsent(element, new AtomicInteger(occurrences));
222         if (existingCounter == null) {
223           return 0;
224         }
225         // existingCounter != null: fall through to operate against the existing AtomicInteger
226       }
227 
228       while (true) {
229         int oldValue = existingCounter.get();
230         if (oldValue != 0) {
231           try {
232             int newValue = IntMath.checkedAdd(oldValue, occurrences);
233             if (existingCounter.compareAndSet(oldValue, newValue)) {
234               // newValue can't == 0, so no need to check & remove
235               return oldValue;
236             }
237           } catch (ArithmeticException overflow) {
238             throw new IllegalArgumentException("Overflow adding " + occurrences
239                 + " occurrences to a count of " + oldValue);
240           }
241         } else {
242           // In the case of a concurrent remove, we might observe a zero value, which means another
243           // thread is about to remove (element, existingCounter) from the map. Rather than wait,
244           // we can just do that work here.
245           AtomicInteger newCounter = new AtomicInteger(occurrences);
246           if ((countMap.putIfAbsent(element, newCounter) == null)
247               || countMap.replace(element, existingCounter, newCounter)) {
248             return 0;
249           }
250           break;
251         }
252       }
253 
254       // If we're still here, there was a race, so just try again.
255     }
256   }
257 
258   /**
259    * Removes a number of occurrences of the specified element from this multiset. If the multiset
260    * contains fewer than this number of occurrences to begin with, all occurrences will be removed.
261    *
262    * @param element the element whose occurrences should be removed
263    * @param occurrences the number of occurrences of the element to remove
264    * @return the count of the element before the operation; possibly zero
265    * @throws IllegalArgumentException if {@code occurrences} is negative
266    */
267   /*
268    * TODO(cpovirk): remove and removeExactly currently accept null inputs only
269    * if occurrences == 0. This satisfies both NullPointerTester and
270    * CollectionRemoveTester.testRemove_nullAllowed, but it's not clear that it's
271    * a good policy, especially because, in order for the test to pass, the
272    * parameter must be misleadingly annotated as @Nullable. I suspect that
273    * we'll want to remove @Nullable, add an eager checkNotNull, and loosen up
274    * testRemove_nullAllowed.
275    */
remove(@ullable Object element, int occurrences)276   @Override public int remove(@Nullable Object element, int occurrences) {
277     if (occurrences == 0) {
278       return count(element);
279     }
280     checkArgument(occurrences > 0, "Invalid occurrences: %s", occurrences);
281 
282     AtomicInteger existingCounter = Maps.safeGet(countMap, element);
283     if (existingCounter == null) {
284       return 0;
285     }
286     while (true) {
287       int oldValue = existingCounter.get();
288       if (oldValue != 0) {
289         int newValue = Math.max(0, oldValue - occurrences);
290         if (existingCounter.compareAndSet(oldValue, newValue)) {
291           if (newValue == 0) {
292             // Just CASed to 0; remove the entry to clean up the map. If the removal fails,
293             // another thread has already replaced it with a new counter, which is fine.
294             countMap.remove(element, existingCounter);
295           }
296           return oldValue;
297         }
298       } else {
299         return 0;
300       }
301     }
302   }
303 
304   /**
305    * Removes exactly the specified number of occurrences of {@code element}, or makes no
306    * change if this is not possible.
307    *
308    * <p>This method, in contrast to {@link #remove(Object, int)}, has no effect when the
309    * element count is smaller than {@code occurrences}.
310    *
311    * @param element the element to remove
312    * @param occurrences the number of occurrences of {@code element} to remove
313    * @return {@code true} if the removal was possible (including if {@code occurrences} is zero)
314    */
removeExactly(@ullable Object element, int occurrences)315   public boolean removeExactly(@Nullable Object element, int occurrences) {
316     if (occurrences == 0) {
317       return true;
318     }
319     checkArgument(occurrences > 0, "Invalid occurrences: %s", occurrences);
320 
321     AtomicInteger existingCounter = Maps.safeGet(countMap, element);
322     if (existingCounter == null) {
323       return false;
324     }
325     while (true) {
326       int oldValue = existingCounter.get();
327       if (oldValue < occurrences) {
328         return false;
329       }
330       int newValue = oldValue - occurrences;
331       if (existingCounter.compareAndSet(oldValue, newValue)) {
332         if (newValue == 0) {
333           // Just CASed to 0; remove the entry to clean up the map. If the removal fails,
334           // another thread has already replaced it with a new counter, which is fine.
335           countMap.remove(element, existingCounter);
336         }
337         return true;
338       }
339     }
340   }
341 
342   /**
343    * Adds or removes occurrences of {@code element} such that the {@link #count} of the
344    * element becomes {@code count}.
345    *
346    * @return the count of {@code element} in the multiset before this call
347    * @throws IllegalArgumentException if {@code count} is negative
348    */
setCount(E element, int count)349   @Override public int setCount(E element, int count) {
350     checkNotNull(element);
351     checkNonnegative(count, "count");
352     while (true) {
353       AtomicInteger existingCounter = Maps.safeGet(countMap, element);
354       if (existingCounter == null) {
355         if (count == 0) {
356           return 0;
357         } else {
358           existingCounter = countMap.putIfAbsent(element, new AtomicInteger(count));
359           if (existingCounter == null) {
360             return 0;
361           }
362           // existingCounter != null: fall through
363         }
364       }
365 
366       while (true) {
367         int oldValue = existingCounter.get();
368         if (oldValue == 0) {
369           if (count == 0) {
370             return 0;
371           } else {
372             AtomicInteger newCounter = new AtomicInteger(count);
373             if ((countMap.putIfAbsent(element, newCounter) == null)
374                 || countMap.replace(element, existingCounter, newCounter)) {
375               return 0;
376             }
377           }
378           break;
379         } else {
380           if (existingCounter.compareAndSet(oldValue, count)) {
381             if (count == 0) {
382               // Just CASed to 0; remove the entry to clean up the map. If the removal fails,
383               // another thread has already replaced it with a new counter, which is fine.
384               countMap.remove(element, existingCounter);
385             }
386             return oldValue;
387           }
388         }
389       }
390     }
391   }
392 
393   /**
394    * Sets the number of occurrences of {@code element} to {@code newCount}, but only if
395    * the count is currently {@code expectedOldCount}. If {@code element} does not appear
396    * in the multiset exactly {@code expectedOldCount} times, no changes will be made.
397    *
398    * @return {@code true} if the change was successful. This usually indicates
399    *     that the multiset has been modified, but not always: in the case that
400    *     {@code expectedOldCount == newCount}, the method will return {@code true} if
401    *     the condition was met.
402    * @throws IllegalArgumentException if {@code expectedOldCount} or {@code newCount} is negative
403    */
setCount(E element, int expectedOldCount, int newCount)404   @Override public boolean setCount(E element, int expectedOldCount, int newCount) {
405     checkNotNull(element);
406     checkNonnegative(expectedOldCount, "oldCount");
407     checkNonnegative(newCount, "newCount");
408 
409     AtomicInteger existingCounter = Maps.safeGet(countMap, element);
410     if (existingCounter == null) {
411       if (expectedOldCount != 0) {
412         return false;
413       } else if (newCount == 0) {
414         return true;
415       } else {
416         // if our write lost the race, it must have lost to a nonzero value, so we can stop
417         return countMap.putIfAbsent(element, new AtomicInteger(newCount)) == null;
418       }
419     }
420     int oldValue = existingCounter.get();
421     if (oldValue == expectedOldCount) {
422       if (oldValue == 0) {
423         if (newCount == 0) {
424           // Just observed a 0; try to remove the entry to clean up the map
425           countMap.remove(element, existingCounter);
426           return true;
427         } else {
428           AtomicInteger newCounter = new AtomicInteger(newCount);
429           return (countMap.putIfAbsent(element, newCounter) == null)
430               || countMap.replace(element, existingCounter, newCounter);
431         }
432       } else {
433         if (existingCounter.compareAndSet(oldValue, newCount)) {
434           if (newCount == 0) {
435             // Just CASed to 0; remove the entry to clean up the map. If the removal fails,
436             // another thread has already replaced it with a new counter, which is fine.
437             countMap.remove(element, existingCounter);
438           }
439           return true;
440         }
441       }
442     }
443     return false;
444   }
445 
446   // Views
447 
createElementSet()448   @Override Set<E> createElementSet() {
449     final Set<E> delegate = countMap.keySet();
450     return new ForwardingSet<E>() {
451       @Override protected Set<E> delegate() {
452         return delegate;
453       }
454 
455       @Override
456       public boolean contains(@Nullable Object object) {
457         return object != null && Collections2.safeContains(delegate, object);
458       }
459 
460       @Override
461       public boolean containsAll(Collection<?> collection) {
462         return standardContainsAll(collection);
463       }
464 
465       @Override public boolean remove(Object object) {
466         return object != null && Collections2.safeRemove(delegate, object);
467       }
468 
469       @Override public boolean removeAll(Collection<?> c) {
470         return standardRemoveAll(c);
471       }
472     };
473   }
474 
475   @Override public Set<Multiset.Entry<E>> createEntrySet() {
476     return new EntrySet();
477   }
478 
479   @Override int distinctElements() {
480     return countMap.size();
481   }
482 
483   @Override public boolean isEmpty() {
484     return countMap.isEmpty();
485   }
486 
487   @Override Iterator<Entry<E>> entryIterator() {
488     // AbstractIterator makes this fairly clean, but it doesn't support remove(). To support
489     // remove(), we create an AbstractIterator, and then use ForwardingIterator to delegate to it.
490     final Iterator<Entry<E>> readOnlyIterator =
491         new AbstractIterator<Entry<E>>() {
492           private Iterator<Map.Entry<E, AtomicInteger>> mapEntries = countMap.entrySet().iterator();
493 
494           @Override protected Entry<E> computeNext() {
495             while (true) {
496               if (!mapEntries.hasNext()) {
497                 return endOfData();
498               }
499               Map.Entry<E, AtomicInteger> mapEntry = mapEntries.next();
500               int count = mapEntry.getValue().get();
501               if (count != 0) {
502                 return Multisets.immutableEntry(mapEntry.getKey(), count);
503               }
504             }
505           }
506         };
507 
508     return new ForwardingIterator<Entry<E>>() {
509       private Entry<E> last;
510 
511       @Override protected Iterator<Entry<E>> delegate() {
512         return readOnlyIterator;
513       }
514 
515       @Override public Entry<E> next() {
516         last = super.next();
517         return last;
518       }
519 
520       @Override public void remove() {
521         checkRemove(last != null);
522         ConcurrentHashMultiset.this.setCount(last.getElement(), 0);
523         last = null;
524       }
525     };
526   }
527 
528   @Override public void clear() {
529     countMap.clear();
530   }
531 
532   private class EntrySet extends AbstractMultiset<E>.EntrySet {
533     @Override ConcurrentHashMultiset<E> multiset() {
534       return ConcurrentHashMultiset.this;
535     }
536 
537     /*
538      * Note: the superclass toArray() methods assume that size() gives a correct
539      * answer, which ours does not.
540      */
541 
542     @Override public Object[] toArray() {
543       return snapshot().toArray();
544     }
545 
546     @Override public <T> T[] toArray(T[] array) {
547       return snapshot().toArray(array);
548     }
549 
550     private List<Multiset.Entry<E>> snapshot() {
551       List<Multiset.Entry<E>> list = Lists.newArrayListWithExpectedSize(size());
552       // Not Iterables.addAll(list, this), because that'll forward right back here.
553       Iterators.addAll(list, iterator());
554       return list;
555     }
556   }
557 
558   /**
559    * @serialData the ConcurrentMap of elements and their counts.
560    */
561   private void writeObject(ObjectOutputStream stream) throws IOException {
562     stream.defaultWriteObject();
563     stream.writeObject(countMap);
564   }
565 
566   private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
567     stream.defaultReadObject();
568     @SuppressWarnings("unchecked") // reading data stored by writeObject
569     ConcurrentMap<E, Integer> deserializedCountMap =
570         (ConcurrentMap<E, Integer>) stream.readObject();
571     FieldSettersHolder.COUNT_MAP_FIELD_SETTER.set(this, deserializedCountMap);
572   }
573 
574   private static final long serialVersionUID = 1;
575 }
576