Author: manik.surtani(a)jboss.com Date: 2008-08-29 00:57:46 -0400 (Fri, 29 Aug 2008) New Revision: 6641 Modified: core/trunk/src/main/java/org/jboss/cache/eviction/EvictionTimerTask.java core/trunk/src/test/java/org/jboss/cache/eviction/EvictionTestsBase.java core/trunk/src/test/java/org/jboss/cache/eviction/ExpirationPolicyTest.java core/trunk/src/test/java/org/jboss/cache/util/internals/EvictionController.java Log: Better timing control between eviction events Modified: core/trunk/src/main/java/org/jboss/cache/eviction/EvictionTimerTask.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/eviction/EvictionTimerTask.java 2008-08-29 04:23:28 UTC (rev 6640) +++ core/trunk/src/main/java/org/jboss/cache/eviction/EvictionTimerTask.java 2008-08-29 04:57:46 UTC (rev 6641) @@ -34,12 +34,14 @@ private static AtomicInteger tcount = new AtomicInteger(); private long wakeupInterval; ScheduledExecutorService scheduledExecutor; + private Task task; public EvictionTimerTask() { // synchronized set because we need to maintain thread safety // for dynamic configuration purposes. processedRegions = new ConcurrentHashSet<Region>(); + task = new Task(); } public void init(long wakeupInterval) @@ -106,16 +108,7 @@ return new Thread(r, "EvictionTimer-" + tcount.getAndIncrement()); } }); - scheduledExecutor.scheduleWithFixedDelay(new Runnable() - { - public void run() - { - // Run the eviction thread. - // This thread will synchronize the set of regions and iterate through every MarshRegion registered w/ the - // Eviction thread. It also synchronizes on each individual region as it is being processed. - processRegions(); - } - }, wakeupInterval, wakeupInterval, TimeUnit.MILLISECONDS); + scheduledExecutor.scheduleWithFixedDelay(task, wakeupInterval, wakeupInterval, TimeUnit.MILLISECONDS); } private void processRegions() @@ -146,6 +139,17 @@ } } } + + public class Task implements Runnable + { + public void run() + { + // Run the eviction thread. + // This thread will synchronize the set of regions and iterate through every MarshRegion registered w/ the + // Eviction thread. It also synchronizes on each individual region as it is being processed. + processRegions(); + } + } } Modified: core/trunk/src/test/java/org/jboss/cache/eviction/EvictionTestsBase.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/eviction/EvictionTestsBase.java 2008-08-29 04:23:28 UTC (rev 6640) +++ core/trunk/src/test/java/org/jboss/cache/eviction/EvictionTestsBase.java 2008-08-29 04:57:46 UTC (rev 6641) @@ -13,6 +13,7 @@ import org.jboss.cache.config.EvictionConfig; import org.jboss.cache.config.EvictionRegionConfig; import org.jboss.cache.util.TestingUtil; +import org.jboss.cache.util.internals.EvictionController; import org.jboss.cache.util.internals.EvictionWatcher; import java.util.concurrent.TimeUnit; @@ -48,6 +49,10 @@ */ public boolean waitForEviction(Cache cache, long timeToWait, TimeUnit unit, Fqn... fqnsToEvict) throws InterruptedException { - return new EvictionWatcher(cache, fqnsToEvict).waitForEviction(timeToWait, unit); + EvictionController ec = new EvictionController(cache); + EvictionController.Signaller signaller = ec.getEvictionThreadSignaller(); + boolean evicted = new EvictionWatcher(cache, fqnsToEvict).waitForEviction(timeToWait, unit); + signaller.waitForEvictionThreadCompletion(timeToWait, unit); + return evicted; } } Modified: core/trunk/src/test/java/org/jboss/cache/eviction/ExpirationPolicyTest.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/eviction/ExpirationPolicyTest.java 2008-08-29 04:23:28 UTC (rev 6640) +++ core/trunk/src/test/java/org/jboss/cache/eviction/ExpirationPolicyTest.java 2008-08-29 04:57:46 UTC (rev 6641) @@ -16,6 +16,7 @@ import org.jboss.cache.config.EvictionConfig; import org.jboss.cache.config.EvictionRegionConfig; import org.jboss.cache.util.TestingUtil; +import org.jboss.cache.util.internals.EvictionController; import static org.testng.AssertJUnit.*; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; @@ -68,45 +69,63 @@ public void testUpdateToFuture() throws Exception { - log.info("update 1 from future to past"); - cache.put(fqn1, ExpirationAlgorithmConfig.EXPIRATION_KEY, future); - TestingUtil.sleepThread(200); - assertNotNull(cache.getNode(fqn1)); - cache.put(fqn1, ExpirationAlgorithmConfig.EXPIRATION_KEY, future + 250); - TestingUtil.sleepThread(500); - assertNotNull(cache.getNode(fqn1)); - TestingUtil.sleepThread(100); - assertNull(cache.getNode(fqn1)); - cache.removeNode(Fqn.ROOT); + try + { + log.info("update 1 from future to past"); + cache.put(fqn1, ExpirationAlgorithmConfig.EXPIRATION_KEY, future); + TestingUtil.sleepThread(200); + new EvictionController(cache).startEviction(); + assertNotNull(cache.getNode(fqn1)); + cache.put(fqn1, ExpirationAlgorithmConfig.EXPIRATION_KEY, future + 250); + TestingUtil.sleepThread(500); + new EvictionController(cache).startEviction(); + assertNotNull(cache.getNode(fqn1)); + TestingUtil.sleepThread(100); + new EvictionController(cache).startEviction(); + assertNull(cache.getNode(fqn1)); + } + finally + { + cache.removeNode(Fqn.ROOT); + } } + @Test(invocationCount = 10) public void testEviction() throws Exception { cache.put(fqn1, ExpirationAlgorithmConfig.EXPIRATION_KEY, future); cache.put(fqn2, ExpirationAlgorithmConfig.EXPIRATION_KEY, past); cache.put(fqn3, ExpirationAlgorithmConfig.EXPIRATION_KEY, future); cache.put(fqn4, "foo", "bar"); - assert waitForEviction(cache, 30, TimeUnit.SECONDS, fqn2) : "Eviction event not received!"; + + waitForEviction(cache, 30, TimeUnit.SECONDS, fqn2); assertNotNull(cache.getNode(fqn1)); assertNull(cache.getNode(fqn2)); assertNotNull(cache.getNode(fqn3)); assertNotNull(cache.getNode(fqn4)); log.info("should remove 1 and 3 now"); - assert waitForEviction(cache, 30, TimeUnit.SECONDS, fqn1, fqn3) : "Eviction event not received!"; + waitForEviction(cache, 30, TimeUnit.SECONDS, fqn1, fqn3); assertNull(cache.getNode(fqn1)); assertNull(cache.getNode(fqn3)); } public void testUpdate() throws Exception { - log.info("update 1 from future to past"); - cache.put(fqn1, ExpirationAlgorithmConfig.EXPIRATION_KEY, future); - assertNotNull(cache.getNode(fqn1)); - cache.put(fqn1, ExpirationAlgorithmConfig.EXPIRATION_KEY, past); - assert waitForEviction(cache, 1, TimeUnit.SECONDS, fqn1) : "Eviction event not received!"; - assertNull(cache.getNode(fqn1)); - cache.removeNode(Fqn.ROOT); + try + { + log.info("update 1 from future to past"); + cache.put(fqn1, ExpirationAlgorithmConfig.EXPIRATION_KEY, future); + new EvictionController(cache).startEviction(); + assertNotNull(cache.getNode(fqn1)); + cache.put(fqn1, ExpirationAlgorithmConfig.EXPIRATION_KEY, past); + new EvictionController(cache).startEviction(); + assertNull(cache.getNode(fqn1)); + } + finally + { + cache.removeNode(Fqn.ROOT); + } } public void testMaxNodes() throws Exception Modified: core/trunk/src/test/java/org/jboss/cache/util/internals/EvictionController.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/util/internals/EvictionController.java 2008-08-29 04:23:28 UTC (rev 6640) +++ core/trunk/src/test/java/org/jboss/cache/util/internals/EvictionController.java 2008-08-29 04:57:46 UTC (rev 6641) @@ -8,10 +8,14 @@ import org.jboss.cache.config.EvictionConfig; import org.jboss.cache.config.EvictionRegionConfig; import org.jboss.cache.eviction.EvictionTimerTask; +import org.jboss.cache.eviction.EvictionTimerTask.Task; import org.jboss.cache.eviction.LRUAlgorithmConfig; import org.jboss.cache.util.TestingUtil; +import java.lang.reflect.Field; import java.lang.reflect.Method; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; /** * when used on a cache will disable defaul eviction behavior and it will supply means of kicking off evction @@ -115,4 +119,72 @@ } } } + + public Signaller getEvictionThreadSignaller() + { + final Signaller s = new Signaller(); + Task signallingTask = timerTask.new Task() + { + public void run() + { + s.getToken(); + try + { + super.run(); + } + finally + { + s.releaseToken(); + } + } + }; + + try + { + Class ettClass = EvictionTimerTask.class; + Field f = ettClass.getDeclaredField("task"); + f.setAccessible(true); + f.set(timerTask, signallingTask); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + timerTask.init(originalWakeupInterval); + return s; + } + + public static class Signaller + { + Semaphore s = new Semaphore(1); + + public boolean waitForEvictionThreadCompletion(long time, TimeUnit unit) throws InterruptedException + { + try + { + return s.tryAcquire(time, unit); + } + finally + { + s.release(); + } + } + + void getToken() + { + try + { + s.acquire(); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + } + + void releaseToken() + { + s.release(); + } + } }