[gatein-commits] gatein SVN: r7645 - in epp/portal/branches/EPP_5_2_Branch/component/common/src: test/java/org/exoplatform/commons/cache/future and 1 other directory.

Author: theute Date: 2011-10-04 04:09:21 -0400 (Tue, 04 Oct 2011) New Revision: 7645 Added: epp/portal/branches/EPP_5_2_Branch/component/common/src/main/java/org/exoplatform/commons/cache/future/Retrieval.java Modified: epp/portal/branches/EPP_5_2_Branch/component/common/src/main/java/org/exoplatform/commons/cache/future/FutureCache.java epp/portal/branches/EPP_5_2_Branch/component/common/src/test/java/org/exoplatform/commons/cache/future/GetTestCase.java Log: JBEPP-1255: Detect reentrancy in future cache to prevent self deadlock Modified: epp/portal/branches/EPP_5_2_Branch/component/common/src/main/java/org/exoplatform/commons/cache/future/FutureCache.java =================================================================== --- epp/portal/branches/EPP_5_2_Branch/component/common/src/main/java/org/exoplatform/commons/cache/future/FutureCache.java 2011-10-04 08:04:40 UTC (rev 7644) +++ epp/portal/branches/EPP_5_2_Branch/component/common/src/main/java/org/exoplatform/commons/cache/future/FutureCache.java 2011-10-04 08:09:21 UTC (rev 7645) @@ -38,10 +38,10 @@ { /** . */ - private final Loader<K, V, C> loader; + final Loader<K, V, C> loader; /** . */ - private final ConcurrentMap<K, FutureTask<V>> futureEntries; + private final ConcurrentMap<K, Retrieval<K, V, C>> futureEntries; /** . */ private final Logger log = LoggerFactory.getLogger(FutureCache.class); @@ -49,7 +49,7 @@ public FutureCache(Loader<K, V, C> loader) { this.loader = loader; - this.futureEntries = new ConcurrentHashMap<K, FutureTask<V>>(); + this.futureEntries = new ConcurrentHashMap<K, Retrieval<K, V, C>>(); } protected abstract V get(K key); @@ -75,65 +75,67 @@ if (value == null) { // Create our future - FutureTask<V> future = new FutureTask<V>(new Callable<V>() - { - public V call() throws Exception - { - // Retrieve the value from the loader - V value = loader.retrieve(context, key); + Retrieval<K, V, C> retrieval = new Retrieval<K, V, C>(context, key, this); - // - if (value != null) - { - // Cache it, it is made available to other threads (unless someone removes it) - put(key, value); - - // Return value - return value; - } - else - { - return null; - } - } - }); - // This boolean means we inserted in the local boolean inserted = true; // try { - FutureTask<V> phantom = futureEntries.putIfAbsent(key, future); + Retrieval<K, V, C> phantom = futureEntries.putIfAbsent(key, retrieval); // Use the value that could have been inserted by another thread if (phantom != null) { - future = phantom; + retrieval = phantom; inserted = false; } else { - future.run(); + try + { + retrieval.current = Thread.currentThread(); + retrieval.future.run(); + } + catch (Exception e) + { + log.error("Retrieval of resource " + key + " threw an exception", e); + } + finally + { + retrieval.current = null; + } } // Returns the value - value = future.get(); + if (retrieval.current == Thread.currentThread()) + { + throw new IllegalStateException("Reentrancy detected when obtaining key " + key + " with context " + context + " detected"); + } + else + { + try + { + value = retrieval.future.get(); + } + catch (ExecutionException e) + { + log.error("Computing of resource " + key + " threw an exception", e.getCause()); + } + catch (InterruptedException e) + { + // We should handle interruped exception in some manner + log.error("Retrieval of resource " + key + " threw an exception", e); + } + } } - catch (ExecutionException e) - { - log.error("Computing of resource " + key + " threw an exception", e.getCause()); - } - catch (Exception e) - { - log.error("Retrieval of resource " + key + " threw an exception", e); - } finally { // Clean up the per key map but only if our insertion succeeded and with our future if (inserted) { - futureEntries.remove(key, future); + futureEntries.remove(key, retrieval); } } } Copied: epp/portal/branches/EPP_5_2_Branch/component/common/src/main/java/org/exoplatform/commons/cache/future/Retrieval.java (from rev 7615, portal/trunk/component/common/src/main/java/org/exoplatform/commons/cache/future/Retrieval.java) =================================================================== --- epp/portal/branches/EPP_5_2_Branch/component/common/src/main/java/org/exoplatform/commons/cache/future/Retrieval.java (rev 0) +++ epp/portal/branches/EPP_5_2_Branch/component/common/src/main/java/org/exoplatform/commons/cache/future/Retrieval.java 2011-10-04 08:09:21 UTC (rev 7645) @@ -0,0 +1,53 @@ +package org.exoplatform.commons.cache.future; + +import java.util.concurrent.Callable; +import java.util.concurrent.FutureTask; + +/** @author <a href="mailto:julien.viet@exoplatform.com">Julien Viet</a> */ +class Retrieval<K, V, C> implements Callable<V> +{ + + /** . */ + private final C context; + + /** . */ + private final K key; + + /** . */ + private final FutureCache<K, V, C> cache; + + /** . */ + final FutureTask<V> future; + + /** Avoid reentrancy. */ + transient Thread current; + + public Retrieval(C context, K key, FutureCache<K, V, C> cache) + { + this.key = key; + this.context = context; + this.future = new FutureTask<V>(this); + this.cache = cache; + this.current = null; + } + + public V call() throws Exception + { + // Retrieve the value from the loader + V value = cache.loader.retrieve(context, key); + + // + if (value != null) + { + // Cache it, it is made available to other threads (unless someone removes it) + cache.put(key, value); + + // Return value + return value; + } + else + { + return null; + } + } +} Modified: epp/portal/branches/EPP_5_2_Branch/component/common/src/test/java/org/exoplatform/commons/cache/future/GetTestCase.java =================================================================== --- epp/portal/branches/EPP_5_2_Branch/component/common/src/test/java/org/exoplatform/commons/cache/future/GetTestCase.java 2011-10-04 08:04:40 UTC (rev 7644) +++ epp/portal/branches/EPP_5_2_Branch/component/common/src/test/java/org/exoplatform/commons/cache/future/GetTestCase.java 2011-10-04 08:09:21 UTC (rev 7645) @@ -23,6 +23,7 @@ import junit.framework.TestCase; import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicBoolean; /** * @author <a href="mailto:julien.viet@exoplatform.com">Julien Viet</a> @@ -68,4 +69,32 @@ }, "foo")); Assert.assertFalse(futureCache.data.containsKey("foo")); } + + public void testReentrancy() + { + final FutureMap<Callable<String>> futureCache = new FutureMap<Callable<String>>(new StringLoader()); + String res = futureCache.get(new Callable<String>() + { + public String call() throws Exception + { + try + { + futureCache.get(new Callable<String>() + { + public String call() throws Exception + { + // Should not go there + throw new AssertionError(); + } + }, "foo"); + return "fail"; + } + catch (IllegalStateException expected) + { + return "pass"; + } + } + }, "foo"); + assertEquals("pass", res); + } }