GEODE-1456: fix race in GemFireCacheImplTest The race was caused by the GemFireCacheImpl constructor creating a real TypeRegistry. The TypeRegistry creates a region which scheduled an async create region event with the event pool. So when the test set "initialCount" this create region event might not have completed so the initialCount would be zero. The test then scheduled MAX_THREADS tasks and waits until that many complete. But the create region event could also complete causing getCompletedTaskCount() to return a value 1 more than the test expected which caused it to intermittently timeout.
The test now mocks the TypeRegistry so the only events scheduled with the pool come from the unit test. Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/11e4b256 Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/11e4b256 Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/11e4b256 Branch: refs/heads/feature/GEODE-1400 Commit: 11e4b25613a9af24c2f7efff70c8bccbde7d0a7f Parents: a2f7c6b Author: Darrel Schneider <[email protected]> Authored: Wed May 25 14:28:49 2016 -0700 Committer: Darrel Schneider <[email protected]> Committed: Thu May 26 14:13:43 2016 -0700 ---------------------------------------------------------------------- .../internal/cache/GemFireCacheImpl.java | 22 ++++++------ .../internal/cache/GemFireCacheImplTest.java | 36 ++++++++++---------- 2 files changed, 30 insertions(+), 28 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/11e4b256/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/GemFireCacheImpl.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/GemFireCacheImpl.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/GemFireCacheImpl.java index 3ff162e..76a2729 100755 --- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/GemFireCacheImpl.java +++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/GemFireCacheImpl.java @@ -749,22 +749,22 @@ public class GemFireCacheImpl implements InternalCache, ClientCache, HasCachePer // } public static GemFireCacheImpl createClient(DistributedSystem system, PoolFactory pf, CacheConfig cacheConfig) { - return basicCreate(system, true, cacheConfig, pf, true, ASYNC_EVENT_LISTENERS); + return basicCreate(system, true, cacheConfig, pf, true, ASYNC_EVENT_LISTENERS, null); } public static GemFireCacheImpl create(DistributedSystem system, CacheConfig cacheConfig) { - return basicCreate(system, true, cacheConfig, null, false, ASYNC_EVENT_LISTENERS); + return basicCreate(system, true, cacheConfig, null, false, ASYNC_EVENT_LISTENERS, null); } - public static GemFireCacheImpl createWithAsyncEventListeners(DistributedSystem system, CacheConfig cacheConfig) { - return basicCreate(system, true, cacheConfig, null, false, true); + public static GemFireCacheImpl createWithAsyncEventListeners(DistributedSystem system, CacheConfig cacheConfig, TypeRegistry typeRegistry) { + return basicCreate(system, true, cacheConfig, null, false, true, typeRegistry); } public static Cache create(DistributedSystem system, boolean existingOk, CacheConfig cacheConfig) { - return basicCreate(system, existingOk, cacheConfig, null, false, ASYNC_EVENT_LISTENERS); + return basicCreate(system, existingOk, cacheConfig, null, false, ASYNC_EVENT_LISTENERS, null); } - private static GemFireCacheImpl basicCreate(DistributedSystem system, boolean existingOk, CacheConfig cacheConfig, PoolFactory pf, boolean isClient, boolean asyncEventListeners) + private static GemFireCacheImpl basicCreate(DistributedSystem system, boolean existingOk, CacheConfig cacheConfig, PoolFactory pf, boolean isClient, boolean asyncEventListeners, TypeRegistry typeRegistry) throws CacheExistsException, TimeoutException, CacheWriterException, GatewayException, RegionExistsException @@ -772,7 +772,7 @@ public class GemFireCacheImpl implements InternalCache, ClientCache, HasCachePer try { GemFireCacheImpl instance = checkExistingCache(existingOk, cacheConfig); if (instance == null) { - instance = new GemFireCacheImpl(isClient, pf, system, cacheConfig, asyncEventListeners); + instance = new GemFireCacheImpl(isClient, pf, system, cacheConfig, asyncEventListeners, typeRegistry); instance.initialize(); } return instance; @@ -803,11 +803,13 @@ public class GemFireCacheImpl implements InternalCache, ClientCache, HasCachePer /** * Creates a new instance of GemFireCache and populates it according to the <code>cache.xml</code>, if appropriate. + * @param typeRegistry: currently only unit tests set this parameter to a non-null value */ - private GemFireCacheImpl(boolean isClient, PoolFactory pf, DistributedSystem system, CacheConfig cacheConfig, boolean asyncEventListeners) { + private GemFireCacheImpl(boolean isClient, PoolFactory pf, DistributedSystem system, CacheConfig cacheConfig, boolean asyncEventListeners, TypeRegistry typeRegistry) { this.isClient = isClient; this.clientpf = pf; this.cacheConfig = cacheConfig; // do early for bug 43213 + this.pdxRegistry = typeRegistry; // Synchronized to prevent a new cache from being created // before an old one has finished closing @@ -4387,8 +4389,8 @@ public class GemFireCacheImpl implements InternalCache, ClientCache, HasCachePer private final ArrayList<SimpleWaiter> riWaiters = new ArrayList<SimpleWaiter>(); - private TypeRegistry pdxRegistry; // never changes but is currently not - // initialized in constructor + private TypeRegistry pdxRegistry; // never changes but is currently only + // initialized in constructor by unit tests /** * update stats for completion of a registerInterest operation http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/11e4b256/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/GemFireCacheImplTest.java ---------------------------------------------------------------------- diff --git a/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/GemFireCacheImplTest.java b/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/GemFireCacheImplTest.java index 71b7fc6..46539ec 100644 --- a/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/GemFireCacheImplTest.java +++ b/geode-core/src/test/java/com/gemstone/gemfire/internal/cache/GemFireCacheImplTest.java @@ -17,6 +17,7 @@ package com.gemstone.gemfire.internal.cache; import static org.junit.Assert.*; +import static org.mockito.Mockito.*; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ThreadPoolExecutor; @@ -27,6 +28,8 @@ import org.junit.Before; import org.junit.Test; import org.junit.experimental.categories.Category; +import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem; +import com.gemstone.gemfire.pdx.internal.TypeRegistry; import com.gemstone.gemfire.test.fake.Fakes; import com.gemstone.gemfire.test.junit.categories.UnitTest; import com.jayway.awaitility.Awaitility; @@ -36,30 +39,27 @@ public class GemFireCacheImplTest { @Test public void checkThatAsyncEventListenersUseAllThreadsInPool() { - - GemFireCacheImpl gfc = GemFireCacheImpl.createWithAsyncEventListeners(Fakes.distributedSystem(), new CacheConfig()); + InternalDistributedSystem ds = Fakes.distributedSystem(); + CacheConfig cc = new CacheConfig(); + TypeRegistry typeRegistry = mock(TypeRegistry.class); + GemFireCacheImpl gfc = GemFireCacheImpl.createWithAsyncEventListeners(ds, cc, typeRegistry); try { - ThreadPoolExecutor executor = (ThreadPoolExecutor) gfc.getEventThreadPool(); - final long initialCount = executor.getCompletedTaskCount(); + ThreadPoolExecutor executor = (ThreadPoolExecutor) gfc.getEventThreadPool(); + assertEquals(0, executor.getCompletedTaskCount()); + assertEquals(0, executor.getActiveCount()); int MAX_THREADS = GemFireCacheImpl.EVENT_THREAD_LIMIT; final CountDownLatch cdl = new CountDownLatch(MAX_THREADS); for (int i = 1; i <= MAX_THREADS; i++) { - Runnable r = new Runnable() { - @Override - public void run() { - cdl.countDown(); - try { - cdl.await(); - } catch (InterruptedException e) { - } + executor.execute(() -> { + cdl.countDown(); + try { + cdl.await(); + } catch (InterruptedException e) { } - }; - executor.execute(r); + }); } - Awaitility.await().pollInterval(10, TimeUnit.MILLISECONDS).pollDelay(10, TimeUnit.MILLISECONDS).timeout(15, TimeUnit.SECONDS) - .until(() -> { - return executor.getCompletedTaskCount() == MAX_THREADS+initialCount; - }); + Awaitility.await().pollInterval(10, TimeUnit.MILLISECONDS).pollDelay(10, TimeUnit.MILLISECONDS).timeout(90, TimeUnit.SECONDS) + .until(() -> assertEquals(MAX_THREADS, executor.getCompletedTaskCount())); } finally { gfc.close(); }
