@Test public void testOnlyPopulatesOnce() throws InterruptedException, BrokenBarrierException { final int threadCount = Runtime.getRuntime().availableProcessors() * 2; final CachingTier<String, String> cache = new OnHeapCachingTier<String, String>(new CountBasedBackEnd<String, Object>(threadCount * 2)); final AtomicBoolean failure = new AtomicBoolean(); final AtomicInteger invocationCounter = new AtomicInteger(); final AtomicInteger valuesRead = new AtomicInteger(); final CyclicBarrier barrier = new CyclicBarrier(threadCount + 1); final Thread[] threads = new Thread[threadCount]; for (int i = 0; i < threads.length; i++) { threads[i] = new Thread() { @Override public void run() { try { final int await = barrier.await(); cache.get(KEY, new Callable<String>() { @Override public String call() throws Exception { invocationCounter.incrementAndGet(); return "0x" + Integer.toHexString(await); } }, false); valuesRead.getAndIncrement(); } catch (Exception e) { e.printStackTrace(); failure.set(true); } } }; threads[i].start(); } barrier.await(); for (Thread thread : threads) { thread.join(); } assertThat(failure.get(), is(false)); assertThat(invocationCounter.get(), is(1)); assertThat(valuesRead.get(), is(threadCount)); }
@Override public V get(final K key, final Callable<V> source, final boolean updateStats) { if (updateStats) { getObserver.begin(); } Object cachedValue = backEnd.get(key); if (cachedValue == null) { if (updateStats) { getObserver.end(GetOutcome.MISS); } Fault<V> f = new Fault<V>(source); cachedValue = backEnd.putIfAbsent(key, f); if (cachedValue == null) { try { V value = f.get(); putObserver.begin(); if (value == null) { backEnd.remove(key, f); } else if (backEnd.replace(key, f, value)) { putObserver.end(PutOutcome.ADDED); } else { return null; } return value; } catch (Throwable e) { backEnd.remove(key, f); if (e instanceof RuntimeException) { throw (RuntimeException)e; } else { throw new CacheException(e); } } } } else { if (updateStats) { getObserver.end(GetOutcome.HIT); } } return getValue(cachedValue); }
OnHeapCachingTier implements CachingTier<K, V> { @Override public V get(final K key, final Callable<V> source, final boolean updateStats) { if (updateStats) { getObserver.begin(); } Object cachedValue = backEnd.get(key); if (cachedValue == null) { if (updateStats) { getObserver.end(GetOutcome.MISS); } Fault<V> f = new Fault<V>(source); cachedValue = backEnd.putIfAbsent(key, f); if (cachedValue == null) { try { V value = f.get(); putObserver.begin(); if (value == null) { backEnd.remove(key, f); } else if (backEnd.replace(key, f, value)) { putObserver.end(PutOutcome.ADDED); } else { return null; } return value; } catch (Throwable e) { backEnd.remove(key, f); if (e instanceof RuntimeException) { throw (RuntimeException)e; } else { throw new CacheException(e); } } } } else { if (updateStats) { getObserver.end(GetOutcome.HIT); } } return getValue(cachedValue); } }
OnHeapCachingTier implements CachingTier<K, V> { @Override public V get(final K key, final Callable<V> source, final boolean updateStats) { if (updateStats) { getObserver.begin(); } Object cachedValue = backEnd.get(key); if (cachedValue == null) { if (updateStats) { getObserver.end(GetOutcome.MISS); } Fault<V> f = new Fault<V>(source); cachedValue = backEnd.putIfAbsent(key, f); if (cachedValue == null) { try { V value = f.get(); putObserver.begin(); if (value == null) { backEnd.remove(key, f); } else if (backEnd.replace(key, f, value)) { putObserver.end(PutOutcome.ADDED); } else { return null; } return value; } catch (Throwable e) { backEnd.remove(key, f); if (e instanceof RuntimeException) { throw (RuntimeException)e; } else { throw new CacheException(e); } } } } else { if (updateStats) { getObserver.end(GetOutcome.HIT); } } return getValue(cachedValue); } OnHeapCachingTier(final HeapCacheBackEnd<K, Object> backEnd); }
OnHeapCachingTier implements CachingTier<K, V> { @Override public V get(final K key, final Callable<V> source, final boolean updateStats) { if (updateStats) { getObserver.begin(); } Object cachedValue = backEnd.get(key); if (cachedValue == null) { if (updateStats) { getObserver.end(GetOutcome.MISS); } Fault<V> f = new Fault<V>(source); cachedValue = backEnd.putIfAbsent(key, f); if (cachedValue == null) { try { V value = f.get(); putObserver.begin(); if (value == null) { backEnd.remove(key, f); } else if (backEnd.replace(key, f, value)) { putObserver.end(PutOutcome.ADDED); } else { return null; } return value; } catch (Throwable e) { backEnd.remove(key, f); if (e instanceof RuntimeException) { throw (RuntimeException)e; } else { throw new CacheException(e); } } } } else { if (updateStats) { getObserver.end(GetOutcome.HIT); } } return getValue(cachedValue); } OnHeapCachingTier(final HeapCacheBackEnd<K, Object> backEnd); static OnHeapCachingTier<Object, Element> createOnHeapCache(final Ehcache cache, final Pool onHeapPool); @Override boolean loadOnPut(); @Override V get(final K key, final Callable<V> source, final boolean updateStats); @Override V remove(final K key); @Override void clear(); @Override void addListener(final Listener<K, V> listener); @Statistic(name = "size", tags = "local-heap") @Override int getInMemorySize(); @Override int getOffHeapSize(); @Override boolean contains(final K key); @Statistic(name = "size-in-bytes", tags = "local-heap") @Override long getInMemorySizeInBytes(); @Override long getOffHeapSizeInBytes(); @Override long getOnDiskSizeInBytes(); @Override void recalculateSize(final K key); @Override Policy getEvictionPolicy(); @Override void setEvictionPolicy(final Policy policy); }
OnHeapCachingTier implements CachingTier<K, V> { @Override public V get(final K key, final Callable<V> source, final boolean updateStats) { if (updateStats) { getObserver.begin(); } Object cachedValue = backEnd.get(key); if (cachedValue == null) { if (updateStats) { getObserver.end(GetOutcome.MISS); } Fault<V> f = new Fault<V>(source); cachedValue = backEnd.putIfAbsent(key, f); if (cachedValue == null) { try { V value = f.get(); putObserver.begin(); if (value == null) { backEnd.remove(key, f); } else if (backEnd.replace(key, f, value)) { putObserver.end(PutOutcome.ADDED); } else { return null; } return value; } catch (Throwable e) { backEnd.remove(key, f); if (e instanceof RuntimeException) { throw (RuntimeException)e; } else { throw new CacheException(e); } } } } else { if (updateStats) { getObserver.end(GetOutcome.HIT); } } return getValue(cachedValue); } OnHeapCachingTier(final HeapCacheBackEnd<K, Object> backEnd); static OnHeapCachingTier<Object, Element> createOnHeapCache(final Ehcache cache, final Pool onHeapPool); @Override boolean loadOnPut(); @Override V get(final K key, final Callable<V> source, final boolean updateStats); @Override V remove(final K key); @Override void clear(); @Override void addListener(final Listener<K, V> listener); @Statistic(name = "size", tags = "local-heap") @Override int getInMemorySize(); @Override int getOffHeapSize(); @Override boolean contains(final K key); @Statistic(name = "size-in-bytes", tags = "local-heap") @Override long getInMemorySizeInBytes(); @Override long getOffHeapSizeInBytes(); @Override long getOnDiskSizeInBytes(); @Override void recalculateSize(final K key); @Override Policy getEvictionPolicy(); @Override void setEvictionPolicy(final Policy policy);  }