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.Multisets.checkNonnegative; 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; 029 030import java.io.IOException; 031import java.io.ObjectInputStream; 032import java.io.ObjectOutputStream; 033import java.io.Serializable; 034import java.util.Collection; 035import java.util.Iterator; 036import java.util.List; 037import java.util.Map; 038import java.util.Set; 039import java.util.concurrent.ConcurrentHashMap; 040import java.util.concurrent.ConcurrentMap; 041import java.util.concurrent.atomic.AtomicInteger; 042 043import javax.annotation.Nullable; 044 045/** 046 * A multiset that supports concurrent modifications and that provides atomic versions of most 047 * {@code Multiset} operations (exceptions where noted). Null elements are not supported. 048 * 049 * <p>See the Guava User Guide article on <a href= 050 * "http://code.google.com/p/guava-libraries/wiki/NewCollectionTypesExplained#Multiset"> 051 * {@code Multiset}</a>. 052 * 053 * @author Cliff L. Biffle 054 * @author mike nonemacher 055 * @since 2.0 (imported from Google Collections Library) 056 */ 057public final class ConcurrentHashMultiset<E> extends AbstractMultiset<E> implements Serializable { 058 059 /* 060 * The ConcurrentHashMultiset's atomic operations are implemented primarily in terms of 061 * AtomicInteger's atomic operations, with some help from ConcurrentMap's atomic operations on 062 * creation and removal (including automatic removal of zeroes). If the modification of an 063 * AtomicInteger results in zero, we compareAndSet the value to zero; if that succeeds, we remove 064 * the entry from the Map. If another operation sees a zero in the map, it knows that the entry is 065 * about to be removed, so this operation may remove it (often by replacing it with a new 066 * AtomicInteger). 067 */ 068 069 /** The number of occurrences of each element. */ 070 private final transient ConcurrentMap<E, AtomicInteger> countMap; 071 072 // This constant allows the deserialization code to set a final field. This holder class 073 // makes sure it is not initialized unless an instance is deserialized. 074 private static class FieldSettersHolder { 075 static final FieldSetter<ConcurrentHashMultiset> COUNT_MAP_FIELD_SETTER = 076 Serialization.getFieldSetter(ConcurrentHashMultiset.class, "countMap"); 077 } 078 079 /** 080 * Creates a new, empty {@code ConcurrentHashMultiset} using the default 081 * initial capacity, load factor, and concurrency settings. 082 */ 083 public static <E> ConcurrentHashMultiset<E> create() { 084 // TODO(schmoe): provide a way to use this class with other (possibly arbitrary) 085 // ConcurrentMap implementors. One possibility is to extract most of this class into 086 // an AbstractConcurrentMapMultiset. 087 return new ConcurrentHashMultiset<E>(new ConcurrentHashMap<E, AtomicInteger>()); 088 } 089 090 /** 091 * Creates a new {@code ConcurrentHashMultiset} containing the specified elements, using 092 * the default initial capacity, load factor, and concurrency settings. 093 * 094 * <p>This implementation is highly efficient when {@code elements} is itself a {@link Multiset}. 095 * 096 * @param elements the elements that the multiset should contain 097 */ 098 public static <E> ConcurrentHashMultiset<E> create(Iterable<? extends E> elements) { 099 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 7.0 123 */ 124 @Beta 125 public static <E> ConcurrentHashMultiset<E> create( 126 GenericMapMaker<? super E, ? super Number> 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 */ 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 */ 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 */ 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 177 @Override public Object[] toArray() { 178 return snapshot().toArray(); 179 } 180 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 */ 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 */ 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 */ 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 */ 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 */ 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 */ 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 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 private transient EntrySet entrySet; 476 477 @Override public Set<Multiset.Entry<E>> entrySet() { 478 EntrySet result = entrySet; 479 if (result == null) { 480 entrySet = result = new EntrySet(); 481 } 482 return result; 483 } 484 485 @Override int distinctElements() { 486 return countMap.size(); 487 } 488 489 @Override public boolean isEmpty() { 490 return countMap.isEmpty(); 491 } 492 493 @Override Iterator<Entry<E>> entryIterator() { 494 // AbstractIterator makes this fairly clean, but it doesn't support remove(). To support 495 // remove(), we create an AbstractIterator, and then use ForwardingIterator to delegate to it. 496 final Iterator<Entry<E>> readOnlyIterator = 497 new AbstractIterator<Entry<E>>() { 498 private Iterator<Map.Entry<E, AtomicInteger>> mapEntries = countMap.entrySet().iterator(); 499 500 @Override protected Entry<E> computeNext() { 501 while (true) { 502 if (!mapEntries.hasNext()) { 503 return endOfData(); 504 } 505 Map.Entry<E, AtomicInteger> mapEntry = mapEntries.next(); 506 int count = mapEntry.getValue().get(); 507 if (count != 0) { 508 return Multisets.immutableEntry(mapEntry.getKey(), count); 509 } 510 } 511 } 512 }; 513 514 return new ForwardingIterator<Entry<E>>() { 515 private Entry<E> last; 516 517 @Override protected Iterator<Entry<E>> delegate() { 518 return readOnlyIterator; 519 } 520 521 @Override public Entry<E> next() { 522 last = super.next(); 523 return last; 524 } 525 526 @Override public void remove() { 527 checkState(last != null); 528 ConcurrentHashMultiset.this.setCount(last.getElement(), 0); 529 last = null; 530 } 531 }; 532 } 533 534 @Override public void clear() { 535 countMap.clear(); 536 } 537 538 private class EntrySet extends AbstractMultiset<E>.EntrySet { 539 @Override ConcurrentHashMultiset<E> multiset() { 540 return ConcurrentHashMultiset.this; 541 } 542 543 /* 544 * Note: the superclass toArray() methods assume that size() gives a correct 545 * answer, which ours does not. 546 */ 547 548 @Override public Object[] toArray() { 549 return snapshot().toArray(); 550 } 551 552 @Override public <T> T[] toArray(T[] array) { 553 return snapshot().toArray(array); 554 } 555 556 private List<Multiset.Entry<E>> snapshot() { 557 List<Multiset.Entry<E>> list = Lists.newArrayListWithExpectedSize(size()); 558 // Not Iterables.addAll(list, this), because that'll forward right back here. 559 Iterators.addAll(list, iterator()); 560 return list; 561 } 562 } 563 564 /** 565 * @serialData the ConcurrentMap of elements and their counts. 566 */ 567 private void writeObject(ObjectOutputStream stream) throws IOException { 568 stream.defaultWriteObject(); 569 stream.writeObject(countMap); 570 } 571 572 private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException { 573 stream.defaultReadObject(); 574 @SuppressWarnings("unchecked") // reading data stored by writeObject 575 ConcurrentMap<E, Integer> deserializedCountMap = 576 (ConcurrentMap<E, Integer>) stream.readObject(); 577 FieldSettersHolder.COUNT_MAP_FIELD_SETTER.set(this, deserializedCountMap); 578 } 579 580 private static final long serialVersionUID = 1; 581}