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}