[infinispan-commits] Infinispan SVN: r591 - in trunk: core/src/main/java/org/infinispan/config/parsing/element and 6 other directories.
infinispan-commits at lists.jboss.org
infinispan-commits at lists.jboss.org
Mon Jul 20 10:32:08 EDT 2009
Author: galder.zamarreno at jboss.com
Date: 2009-07-20 10:32:07 -0400 (Mon, 20 Jul 2009)
New Revision: 591
Modified:
trunk/cachestore/jdbc/src/test/java/org/infinispan/config/parsing/JdbcConfigurationParserTest.java
trunk/core/src/main/java/org/infinispan/config/parsing/element/LoadersElementParser.java
trunk/core/src/main/java/org/infinispan/loaders/decorators/AsyncStore.java
trunk/core/src/main/java/org/infinispan/loaders/decorators/AsyncStoreConfig.java
trunk/core/src/main/resources/schema/infinispan-config-4.0.xsd
trunk/core/src/test/java/org/infinispan/config/parsing/ConfigurationParserTest.java
trunk/core/src/test/java/org/infinispan/config/parsing/XmlFileParsingTest.java
trunk/core/src/test/java/org/infinispan/loaders/decorators/AsyncTest.java
trunk/core/src/test/java/org/infinispan/loaders/dummy/DummyInMemoryCacheStore.java
trunk/core/src/test/resources/configs/named-cache-test.xml
Log:
[ISPN-116] (Async cache store should aggregate results for a given key) Done.
Modified: trunk/cachestore/jdbc/src/test/java/org/infinispan/config/parsing/JdbcConfigurationParserTest.java
===================================================================
--- trunk/cachestore/jdbc/src/test/java/org/infinispan/config/parsing/JdbcConfigurationParserTest.java 2009-07-17 08:38:59 UTC (rev 590)
+++ trunk/cachestore/jdbc/src/test/java/org/infinispan/config/parsing/JdbcConfigurationParserTest.java 2009-07-20 14:32:07 UTC (rev 591)
@@ -31,7 +31,7 @@
" <property name=\"createTableOnStart\" value=\"false\"/>\n" +
" </properties>\n" +
" <singletonStore enabled=\"true\" pushStateWhenCoordinator=\"true\" pushStateTimeout=\"20000\"/>\n" +
- " <async enabled=\"true\" batchSize=\"15\"/>\n" +
+ " <async enabled=\"true\" threadPoolSize=\"10\" mapLockTimeout=\"10000\"/>\n" +
" </loader>\n" +
" </loaders> ";
Element e = XmlConfigHelper.stringToElement(xml);
@@ -49,10 +49,8 @@
CacheStoreConfig iclc = (CacheStoreConfig) clc.getFirstCacheLoaderConfig();
assert iclc.getCacheLoaderClassName().equals(JdbcStringBasedCacheStore.class.getName());
assert iclc.getAsyncStoreConfig().isEnabled();
- assert iclc.getAsyncStoreConfig().getBatchSize() == 15;
- assert iclc.getAsyncStoreConfig().getPollWait() == 100;
- assert iclc.getAsyncStoreConfig().getQueueSize() == 10000;
- assert iclc.getAsyncStoreConfig().getThreadPoolSize() == 1;
+ assert iclc.getAsyncStoreConfig().getMapLockTimeout() == 10000;
+ assert iclc.getAsyncStoreConfig().getThreadPoolSize() == 10;
assert iclc.isFetchPersistentState();
assert iclc.isIgnoreModifications();
assert iclc.isPurgeOnStartup();
Modified: trunk/core/src/main/java/org/infinispan/config/parsing/element/LoadersElementParser.java
===================================================================
--- trunk/core/src/main/java/org/infinispan/config/parsing/element/LoadersElementParser.java 2009-07-17 08:38:59 UTC (rev 590)
+++ trunk/core/src/main/java/org/infinispan/config/parsing/element/LoadersElementParser.java 2009-07-20 14:32:07 UTC (rev 591)
@@ -123,15 +123,9 @@
asc.setEnabled(async);
if (async) {
- String tmp = getAttributeValue(element, "batchSize");
- if (existsAttribute(tmp)) asc.setBatchSize(getInt(tmp));
+ String tmp = getAttributeValue(element, "mapLockTimeout");
+ if (existsAttribute(tmp)) asc.setMapLockTimeout(getLong(tmp));
- tmp = getAttributeValue(element, "pollWait");
- if (existsAttribute(tmp)) asc.setPollWait(getLong(tmp));
-
- tmp = getAttributeValue(element, "queueSize");
- if (existsAttribute(tmp)) asc.setQueueSize(getInt(tmp));
-
tmp = getAttributeValue(element, "threadPoolSize");
if (existsAttribute(tmp)) asc.setThreadPoolSize(getInt(tmp));
}
Modified: trunk/core/src/main/java/org/infinispan/loaders/decorators/AsyncStore.java
===================================================================
--- trunk/core/src/main/java/org/infinispan/loaders/decorators/AsyncStore.java 2009-07-17 08:38:59 UTC (rev 590)
+++ trunk/core/src/main/java/org/infinispan/loaders/decorators/AsyncStore.java 2009-07-20 14:32:07 UTC (rev 591)
@@ -1,5 +1,7 @@
package org.infinispan.loaders.decorators;
+import net.jcip.annotations.GuardedBy;
+
import org.infinispan.CacheException;
import org.infinispan.container.entries.InternalCacheEntry;
import org.infinispan.loaders.CacheLoaderException;
@@ -14,15 +16,22 @@
import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.BlockingQueue;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* The AsyncStore is a delegating CacheStore that extends AbstractDelegatingStore, overriding methods to that should not
@@ -39,73 +48,72 @@
* to define whether cache loader operations are to be asynchronous. If not specified, a cache loader operation is
* assumed synchronous and this decorator is not applied.
* <p/>
+ * Write operations affecting same key are now coalesced so that only the final state is actually stored.
+ * <p/>
*
* @author Manik Surtani
+ * @author Galder Zamarreño
* @since 4.0
*/
public class AsyncStore extends AbstractDelegatingStore {
-
private static final Log log = LogFactory.getLog(AsyncStore.class);
private static final boolean trace = log.isTraceEnabled();
+ private static final AtomicInteger threadId = new AtomicInteger(0);
+ private final AtomicBoolean stopped = new AtomicBoolean(true);
+ private final AsyncStoreConfig asyncStoreConfig;
+
+ /** Approximate count of number of modified keys. At points, it could contain negative values. */
+ private final AtomicInteger count = new AtomicInteger(0);
+ private final ReentrantLock lock = new ReentrantLock();
+ private final Condition notEmpty = lock.newCondition();
- private static AtomicInteger threadId = new AtomicInteger(0);
-
private ExecutorService executor;
- private AtomicBoolean stopped = new AtomicBoolean(true);
- private BlockingQueue<Modification> queue;
private List<Future> processorFutures;
- private AsyncStoreConfig asyncStoreConfig;
-
- public AsyncStore(CacheStore cacheStore, AsyncStoreConfig asyncStoreConfig) {
- super(cacheStore);
+ private final ReadWriteLock mapLock = new ReentrantReadWriteLock();
+ private final Lock read = mapLock.readLock();
+ private final Lock write = mapLock.writeLock();
+ @GuardedBy("mapLock") private ConcurrentMap<Object, Modification> state;
+
+ public AsyncStore(CacheStore delegate, AsyncStoreConfig asyncStoreConfig) {
+ super(delegate);
this.asyncStoreConfig = asyncStoreConfig;
}
-
+
public void store(InternalCacheEntry ed) {
- enqueue(new Store(ed));
+ enqueue(ed.getKey(), new Store(ed));
}
-
- public void clear() {
- enqueue(new Clear());
- }
-
+
public boolean remove(Object key) {
- enqueue(new Remove(key));
+ enqueue(key, new Remove(key));
return true;
}
-
+
+ public void clear() {
+ Clear clear = new Clear();
+ enqueue(clear, clear);
+ }
+
public void purgeExpired() {
- enqueue(new PurgeExpired());
+ PurgeExpired purge = new PurgeExpired();
+ enqueue(purge, purge);
}
-
- private void enqueue(final Modification mod) {
- try {
- if (stopped.get()) {
- throw new CacheException("AsyncStore stopped; no longer accepting more entries.");
- }
- log.trace("Enqueuing modification {0}", mod);
- queue.put(mod);
- } catch (Exception e) {
- throw new CacheException("Unable to enqueue asynchronous task", e);
- }
- }
-
+
@Override
public void start() throws CacheLoaderException {
- queue = new LinkedBlockingQueue<Modification>(asyncStoreConfig.getQueueSize());
+ state = new ConcurrentHashMap<Object, Modification>();
log.info("Async cache loader starting {0}", this);
stopped.set(false);
super.start();
int poolSize = asyncStoreConfig.getThreadPoolSize();
executor = Executors.newFixedThreadPool(poolSize, new ThreadFactory() {
public Thread newThread(Runnable r) {
- Thread t = new Thread(r, "AsyncStore-" + threadId.getAndIncrement());
+ Thread t = new Thread(r, "CoalescedAsyncStore-" + threadId.getAndIncrement());
t.setDaemon(true);
return t;
}
});
processorFutures = new ArrayList<Future>(poolSize);
- for (int i = 0; i < poolSize; i++) processorFutures.add(executor.submit(new AsyncProcessor()));
+ for (int i = 0; i < poolSize; i++) processorFutures.add(executor.submit(createAsyncProcessor()));
}
@Override
@@ -119,48 +127,114 @@
while (!terminated) {
terminated = executor.awaitTermination(60, TimeUnit.SECONDS);
}
- }
- catch (InterruptedException e) {
+ } catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
executor = null;
super.stop();
}
-
- protected void applyModificationsSync(List<Modification> mods) throws CacheLoaderException {
- for (Modification m : mods) {
- switch (m.getType()) {
+
+ protected void applyModificationsSync(ConcurrentMap<Object, Modification> mods) throws CacheLoaderException {
+ Set<Map.Entry<Object, Modification>> entries = mods.entrySet();
+ for (Map.Entry<Object, Modification> entry : entries) {
+ Modification mod = entry.getValue();
+ switch (mod.getType()) {
case STORE:
- Store s = (Store) m;
- super.store(s.getStoredEntry());
+ super.store(((Store)mod).getStoredEntry());
break;
+ case REMOVE:
+ super.remove(entry.getKey());
+ break;
case CLEAR:
super.clear();
break;
- case REMOVE:
- Remove r = (Remove) m;
- super.remove(r.getKey());
- break;
case PURGE_EXPIRED:
super.purgeExpired();
break;
- default:
- throw new IllegalArgumentException("Unknown modification type " + m.getType());
}
+ }
+ }
+
+ protected Runnable createAsyncProcessor() {
+ return new AsyncProcessor();
+ }
+
+ private void enqueue(Object key, Modification mod) {
+ try {
+ if (stopped.get()) {
+ throw new CacheException("AsyncStore stopped; no longer accepting more entries.");
+ }
+ if (trace) log.trace("Enqueuing modification {0}", mod);
+ Modification prev = null;
+ int c = -1;
+ boolean unlock = false;
+ try {
+ acquireLock(read);
+ unlock = true;
+ prev = state.put(key, mod); // put the key's latest state in updates
+ } finally {
+ if (unlock) read.unlock();
+ }
+ /* Increment can happen outside the lock cos worst case scenario a false not empty would
+ * be sent if the swap and decrement happened between the put and the increment. In this
+ * case, the corresponding processor would see the map empty and would wait again. This
+ * means that we're allowing count to potentially go negative but that's not a problem. */
+ if (prev == null) c = count.getAndIncrement();
+ if (c == 0) signalNotEmpty();
+ } catch (Exception e) {
+ throw new CacheException("Unable to enqueue asynchronous task", e);
}
}
-
+ private void acquireLock(Lock lock) {
+ try {
+ if (!lock.tryLock(asyncStoreConfig.getMapLockTimeout(), TimeUnit.MILLISECONDS))
+ throw new CacheException("Unable to acquire lock on update map");
+ } catch (InterruptedException ie) {
+ // restore interrupted status
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private void signalNotEmpty() {
+ lock.lock();
+ try {
+ notEmpty.signal();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private void awaitNotEmpty() throws InterruptedException {
+ lock.lockInterruptibly();
+ try {
+ try {
+ while (count.get() == 0)
+ notEmpty.await();
+ } catch (InterruptedException ie) {
+ notEmpty.signal(); // propagate to a non-interrupted thread
+ throw ie;
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private int decrementAndGet(int delta) {
+ for (;;) {
+ int current = count.get();
+ int next = current - delta;
+ if (count.compareAndSet(current, next)) return next;
+ }
+ }
+
/**
- * Processes (by batch if possible) a queue of {@link Modification}s.
- *
- * @author manik surtani
+ * Processes modifications taking the latest updates from a state map.
*/
- private class AsyncProcessor implements Runnable {
- // Modifications to invoke as a single put
- private final List<Modification> mods = new ArrayList<Modification>(asyncStoreConfig.getBatchSize());
-
+ class AsyncProcessor implements Runnable {
+ private ConcurrentMap<Object, Modification> swap = new ConcurrentHashMap<Object, Modification>();
+
public void run() {
while (!Thread.interrupted()) {
try {
@@ -172,38 +246,44 @@
}
try {
- if (trace) log.trace("Process remaining batch {0}", mods.size());
- put(mods);
- if (trace) log.trace("Process remaining queued {0}", queue.size());
- while (!queue.isEmpty()) run0();
+ if (trace) log.trace("Process remaining batch {0}", swap.size());
+ put(swap);
+ if (trace) log.trace("Process remaining queued {0}", state.size());
+ while (!state.isEmpty()) run0();
+ } catch (InterruptedException e) {
+ if (trace) log.trace("Remaining interrupted");
}
- catch (InterruptedException e) {
- log.trace("remaining interrupted");
- }
}
-
- private void run0() throws InterruptedException {
- log.trace("Checking for modifications");
- int i = queue.drainTo(mods, asyncStoreConfig.getBatchSize());
- if (i == 0) {
- Modification m = queue.take();
- mods.add(m);
+
+ void run0() throws InterruptedException {
+ if (trace) log.trace("Checking for modifications");
+ boolean unlock = false;
+ try {
+ acquireLock(write);
+ unlock = true;
+ swap = state;
+ state = new ConcurrentHashMap<Object, Modification>();
+ } finally {
+ if (unlock) write.unlock();
}
+
+ int size = swap.size();
+ if (size == 0)
+ awaitNotEmpty();
+ else
+ decrementAndGet(size);
- if (trace) log.trace("Calling put(List) with {0} modifications", mods.size());
- put(mods);
- mods.clear();
+ if (trace) log.trace("Calling put(List) with {0} modifications", size);
+ put(swap);
}
-
- private void put(List<Modification> mods) {
+
+ void put(ConcurrentMap<Object, Modification> mods) {
try {
AsyncStore.this.applyModificationsSync(mods);
- }
- catch (Exception e) {
- if (log.isWarnEnabled()) log.warn("Failed to process async modifications: " + e);
+ } catch (Exception e) {
+ if (log.isWarnEnabled()) log.warn("Failed to process async modifications", e);
if (log.isDebugEnabled()) log.debug("Exception: ", e);
}
}
}
-
}
Modified: trunk/core/src/main/java/org/infinispan/loaders/decorators/AsyncStoreConfig.java
===================================================================
--- trunk/core/src/main/java/org/infinispan/loaders/decorators/AsyncStoreConfig.java 2009-07-17 08:38:59 UTC (rev 590)
+++ trunk/core/src/main/java/org/infinispan/loaders/decorators/AsyncStoreConfig.java 2009-07-20 14:32:07 UTC (rev 591)
@@ -3,6 +3,7 @@
import org.infinispan.config.AbstractNamedCacheConfigurationBean;
import org.infinispan.config.ConfigurationAttribute;
import org.infinispan.config.ConfigurationElement;
+import org.infinispan.config.Dynamic;
/**
* Configuration for the async cache loader
@@ -13,10 +14,9 @@
@ConfigurationElement(name="async", parent="loader")
public class AsyncStoreConfig extends AbstractNamedCacheConfigurationBean {
boolean enabled;
- int batchSize = 100;
- long pollWait = 100;
- int queueSize = 10000;
int threadPoolSize = 1;
+ @Dynamic
+ long mapLockTimeout = 5000;
public boolean isEnabled() {
return enabled;
@@ -24,55 +24,36 @@
@ConfigurationAttribute(name = "enabled",
containingElement = "async",
- description="TODO")
+ description="If true, modifications are stored in the cache store asynchronously.")
public void setEnabled(boolean enabled) {
testImmutability("enabled");
this.enabled = enabled;
}
- public int getBatchSize() {
- return batchSize;
- }
-
- @ConfigurationAttribute(name = "batchSize",
- containingElement = "async",
- description="TODO")
- public void setBatchSize(int batchSize) {
- testImmutability("batchSize");
- this.batchSize = batchSize;
- }
-
- public long getPollWait() {
- return pollWait;
- }
-
- public void setPollWait(long pollWait) {
- testImmutability("pollWait");
- this.pollWait = pollWait;
- }
-
- public int getQueueSize() {
- return queueSize;
- }
-
- public void setQueueSize(int queueSize) {
- testImmutability("queueSize");
- this.queueSize = queueSize;
- }
-
public int getThreadPoolSize() {
return threadPoolSize;
}
-
@ConfigurationAttribute(name = "threadPoolSize",
containingElement = "async",
- description="TODO")
+ description="Size of the thread pool whose threads are responsible for applying the modifications.")
public void setThreadPoolSize(int threadPoolSize) {
testImmutability("threadPoolSize");
this.threadPoolSize = threadPoolSize;
}
+ public long getMapLockTimeout() {
+ return mapLockTimeout;
+ }
+
+ @ConfigurationAttribute(name = "mapLockTimeout",
+ containingElement = "async",
+ description="Lock timeout for access to map containing latest state.")
+ public void setMapLockTimeout(long stateLockTimeout) {
+ testImmutability("stateLockTimeout");
+ this.mapLockTimeout = stateLockTimeout;
+ }
+
@Override
public AsyncStoreConfig clone() {
try {
Modified: trunk/core/src/main/resources/schema/infinispan-config-4.0.xsd
===================================================================
--- trunk/core/src/main/resources/schema/infinispan-config-4.0.xsd 2009-07-17 08:38:59 UTC (rev 590)
+++ trunk/core/src/main/resources/schema/infinispan-config-4.0.xsd 2009-07-20 14:32:07 UTC (rev 591)
@@ -216,11 +216,21 @@
</xs:element>
<xs:element name="async" minOccurs="0" maxOccurs="1">
<xs:complexType>
- <xs:attribute name="enabled" type="tns:booleanType"/>
- <xs:attribute name="batchSize" type="tns:positiveNumber"/>
- <xs:attribute name="pollWait" type="tns:positiveNumber"/>
- <xs:attribute name="queueSize" type="tns:positiveNumber"/>
- <xs:attribute name="threadPoolSize" type="tns:positiveNumber"/>
+ <xs:attribute name="enabled" type="tns:booleanType">
+ <xs:annotation>
+ <xs:documentation>If true, modifications are stored in the cache store asynchronously.</xs:documentation>
+ </xs:annotation>
+ </xs:attribute>
+ <xs:attribute name="mapLockTimeout" type="tns:positiveNumber">
+ <xs:annotation>
+ <xs:documentation>Lock timeout for access to map containing latest state.</xs:documentation>
+ </xs:annotation>
+ </xs:attribute>
+ <xs:attribute name="threadPoolSize" type="tns:positiveNumber">
+ <xs:annotation>
+ <xs:documentation>Size of the thread pool whose threads are responsible for applying the modifications.</xs:documentation>
+ </xs:annotation>
+ </xs:attribute>
</xs:complexType>
</xs:element>
</xs:all>
Modified: trunk/core/src/test/java/org/infinispan/config/parsing/ConfigurationParserTest.java
===================================================================
--- trunk/core/src/test/java/org/infinispan/config/parsing/ConfigurationParserTest.java 2009-07-17 08:38:59 UTC (rev 590)
+++ trunk/core/src/test/java/org/infinispan/config/parsing/ConfigurationParserTest.java 2009-07-20 14:32:07 UTC (rev 591)
@@ -165,7 +165,7 @@
" <property name=\"location\" value=\"blahblah\"/>\n" +
" </properties>\n" +
" <singletonStore enabled=\"true\" pushStateWhenCoordinator=\"true\" pushStateTimeout=\"20000\"/>\n" +
- " <async enabled=\"true\" batchSize=\"15\"/>\n" +
+ " <async enabled=\"true\" threadPoolSize=\"10\" mapLockTimeout=\"10000\"/>\n" +
" </loader>\n" +
" </loaders> ";
Element e = XmlConfigHelper.stringToElement(xml);
@@ -183,10 +183,8 @@
CacheStoreConfig iclc = (CacheStoreConfig) clc.getFirstCacheLoaderConfig();
assert iclc.getCacheLoaderClassName().equals(FileCacheStore.class.getName());
assert iclc.getAsyncStoreConfig().isEnabled();
- assert iclc.getAsyncStoreConfig().getBatchSize() == 15;
- assert iclc.getAsyncStoreConfig().getPollWait() == 100;
- assert iclc.getAsyncStoreConfig().getQueueSize() == 10000;
- assert iclc.getAsyncStoreConfig().getThreadPoolSize() == 1;
+ assert iclc.getAsyncStoreConfig().getMapLockTimeout() == 10000;
+ assert iclc.getAsyncStoreConfig().getThreadPoolSize() == 10;
assert iclc.isFetchPersistentState();
assert iclc.isIgnoreModifications();
assert iclc.isPurgeOnStartup();
Modified: trunk/core/src/test/java/org/infinispan/config/parsing/XmlFileParsingTest.java
===================================================================
--- trunk/core/src/test/java/org/infinispan/config/parsing/XmlFileParsingTest.java 2009-07-17 08:38:59 UTC (rev 590)
+++ trunk/core/src/test/java/org/infinispan/config/parsing/XmlFileParsingTest.java 2009-07-20 14:32:07 UTC (rev 591)
@@ -126,8 +126,8 @@
assert csConf.getLocation().equals("/tmp/FileCacheStore-Location");
assert csConf.getSingletonStoreConfig().getPushStateTimeout() == 20000;
assert csConf.getSingletonStoreConfig().isPushStateWhenCoordinator() == true;
- assert csConf.getAsyncStoreConfig().getBatchSize() == 1000;
assert csConf.getAsyncStoreConfig().getThreadPoolSize() == 5;
+ assert csConf.getAsyncStoreConfig().getMapLockTimeout() == 15000;
assert csConf.getAsyncStoreConfig().isEnabled();
c = namedCaches.get("withouthJmxEnabled");
Modified: trunk/core/src/test/java/org/infinispan/loaders/decorators/AsyncTest.java
===================================================================
--- trunk/core/src/test/java/org/infinispan/loaders/decorators/AsyncTest.java 2009-07-17 08:38:59 UTC (rev 590)
+++ trunk/core/src/test/java/org/infinispan/loaders/decorators/AsyncTest.java 2009-07-20 14:32:07 UTC (rev 591)
@@ -1,11 +1,13 @@
package org.infinispan.loaders.decorators;
import org.infinispan.CacheException;
+import org.infinispan.container.entries.InternalCacheEntry;
import org.infinispan.container.entries.InternalEntryFactory;
import org.infinispan.loaders.CacheLoaderException;
import org.infinispan.loaders.dummy.DummyInMemoryCacheStore;
import org.infinispan.test.TestingUtil;
-import org.testng.annotations.AfterMethod;
+import org.infinispan.util.logging.Log;
+import org.infinispan.util.logging.LogFactory;
import org.testng.annotations.AfterTest;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;
@@ -14,17 +16,22 @@
@Test(groups = "unit", testName = "loaders.decorators.AsyncTest")
public class AsyncTest {
-
+ private static final Log log = LogFactory.getLog(AsyncTest.class);
AsyncStore store;
ExecutorService asyncExecutor;
+ DummyInMemoryCacheStore underlying;
+ AsyncStoreConfig asyncConfig;
+ DummyInMemoryCacheStore.Cfg dummyCfg;
-
@BeforeTest
public void setUp() throws CacheLoaderException {
- store = new AsyncStore(new DummyInMemoryCacheStore(), new AsyncStoreConfig());
- DummyInMemoryCacheStore.Cfg cfg = new DummyInMemoryCacheStore.Cfg();
- cfg.setStore(AsyncTest.class.getName());
- store.init(cfg, null, null);
+ underlying = new DummyInMemoryCacheStore();
+ asyncConfig = new AsyncStoreConfig();
+ asyncConfig.setThreadPoolSize(10);
+ store = new AsyncStore(underlying, asyncConfig);
+ dummyCfg = new DummyInMemoryCacheStore.Cfg();
+ dummyCfg.setStore(AsyncTest.class.getName());
+ store.init(dummyCfg, null, null);
store.start();
asyncExecutor = (ExecutorService) TestingUtil.extractField(store, "executor");
}
@@ -34,16 +41,43 @@
if (store != null) store.stop();
}
- @AfterMethod
- public void clearStore() {
- if (store != null) store.clear();
+ public void testPutRemove() throws Exception {
+ final int number = 1000;
+ String key = "testPutRemove-k-";
+ String value = "testPutRemove-v-";
+ doTestPut(number, key, value);
+ doTestRemove(number, key);
}
+
+ public void testPutClearPut() throws Exception {
+ final int number = 1000;
+ String key = "testPutClearPut-k-";
+ String value = "testPutClearPut-v-";
+ doTestPut(number, key, value);
+ doTestClear(number, key);
+ value = "testPutClearPut-v[2]-";
+ doTestPut(number, key, value);
+
+ doTestRemove(number, key);
+ }
+ public void testMultiplePutsOnSameKey() throws Exception {
+ final int number = 1000;
+ String key = "testMultiplePutsOnSameKey-k";
+ String value = "testMultiplePutsOnSameKey-v-";
+ doTestSameKeyPut(number, key, value);
+ doTestSameKeyRemove(key);
+
+ }
+
public void testRestrictionOnAddingToAsyncQueue() throws Exception {
store.remove("blah");
- for (int i = 0; i < 4; i++) store.store(InternalEntryFactory.create("k" + i, "v" + i));
-
+ final int number = 10;
+ String key = "testRestrictionOnAddingToAsyncQueue-k";
+ String value = "testRestrictionOnAddingToAsyncQueue-v-";
+ doTestPut(number, key, value);
+
// stop the cache store
store.stop();
try {
@@ -55,5 +89,92 @@
// clean up
store.start();
+ doTestRemove(number, key);
}
+
+ private void doTestPut(int number, String key, String value) throws Exception {
+ for (int i = 0; i < number; i++) store.store(InternalEntryFactory.create(key + i, value + i));
+
+ TestingUtil.sleepRandom(1000);
+
+ InternalCacheEntry[] entries = new InternalCacheEntry[number];
+ for (int i = 0; i < number; i++) {
+ entries[i] = store.load(key + i);
+ }
+
+ for (int i = 0; i < number; i++) {
+ InternalCacheEntry entry = entries[i];
+ if (entry != null) {
+ assert entry.getValue().equals(value + i);
+ } else {
+ while (entry == null) {
+ entry = store.load(key + i);
+ if (entry != null) {
+ assert entry.getValue().equals(value + i);
+ } else {
+ TestingUtil.sleepRandom(1000);
+ }
+ }
+ }
+ }
+ }
+
+ private void doTestSameKeyPut(int number, String key, String value) throws Exception {
+ for (int i = 0; i < number; i++) store.store(InternalEntryFactory.create(key, value + i));
+
+ InternalCacheEntry entry;
+ do {
+ TestingUtil.sleepRandom(1000);
+ entry = store.load(key);
+ } while (!entry.getValue().equals(value + (number-1)));
+ }
+
+ private void doTestRemove(int number, String key) throws Exception {
+ for (int i = 0; i < number; i++) store.remove(key + i);
+
+ TestingUtil.sleepRandom(1000);
+
+ InternalCacheEntry[] entries = new InternalCacheEntry[number];
+ for (int i = 0; i < number; i++) {
+ entries[i] = store.load(key + i);
+ }
+
+ for (int i = 0; i < number; i++) {
+ InternalCacheEntry entry = entries[i];
+ while (entry != null) {
+ log.info("Entry still not null {0}", entry);
+ TestingUtil.sleepRandom(1000);
+ entry = store.load(key + i);
+ }
+ }
+ }
+
+ private void doTestSameKeyRemove(String key) throws Exception {
+ store.remove(key);
+ InternalCacheEntry entry;
+ do {
+ TestingUtil.sleepRandom(1000);
+ entry = store.load(key);
+ } while (entry != null);
+ }
+
+ private void doTestClear(int number, String key) throws Exception {
+ store.clear();
+ TestingUtil.sleepRandom(1000);
+
+ InternalCacheEntry[] entries = new InternalCacheEntry[number];
+ for (int i = 0; i < number; i++) {
+ entries[i] = store.load(key + i);
+ }
+
+ for (int i = 0; i < number; i++) {
+ InternalCacheEntry entry = entries[i];
+ while (entry != null) {
+ log.info("Entry still not null {0}", entry);
+ TestingUtil.sleepRandom(1000);
+ entry = store.load(key + i);
+ }
+ }
+ }
+
}
Modified: trunk/core/src/test/java/org/infinispan/loaders/dummy/DummyInMemoryCacheStore.java
===================================================================
--- trunk/core/src/test/java/org/infinispan/loaders/dummy/DummyInMemoryCacheStore.java 2009-07-17 08:38:59 UTC (rev 590)
+++ trunk/core/src/test/java/org/infinispan/loaders/dummy/DummyInMemoryCacheStore.java 2009-07-20 14:32:07 UTC (rev 591)
@@ -22,6 +22,7 @@
public class DummyInMemoryCacheStore extends AbstractCacheStore {
private static final Log log = LogFactory.getLog(DummyInMemoryCacheStore.class);
+ private static final boolean trace = log.isTraceEnabled();
static final ConcurrentMap<String, Map> stores = new ConcurrentHashMap<String, Map>();
String storeName = "__DEFAULT_STORES__";
Map<Object, InternalCacheEntry> store;
@@ -31,7 +32,7 @@
public void store(InternalCacheEntry ed) {
if (ed != null) {
- log.trace("Store {0} in dummy map store@{1}", ed, Integer.toHexString(System.identityHashCode(store)));
+ if (trace) log.trace("Store {0} in dummy map store@{1}", ed, Integer.toHexString(System.identityHashCode(store)));
store.put(ed.getKey(), ed);
}
}
@@ -43,7 +44,7 @@
store.clear();
for (int i = 0; i < numEntries; i++) {
InternalCacheEntry e = (InternalCacheEntry) marshaller.objectFromObjectStream(ois);
- log.trace("Store {0} from stream in dummy store@{1}", e, Integer.toHexString(System.identityHashCode(store)));
+ if (trace) log.trace("Store {0} from stream in dummy store@{1}", e, Integer.toHexString(System.identityHashCode(store)));
store.put(e.getKey(), e);
}
} catch (Exception e) {
@@ -61,10 +62,12 @@
}
public void clear() {
+ if (trace) log.trace("Clear store");
store.clear();
}
public boolean remove(Object key) {
+ if (trace) log.trace("Remove {0} from dummy store", key);
return store.remove(key) != null;
}
Modified: trunk/core/src/test/resources/configs/named-cache-test.xml
===================================================================
--- trunk/core/src/test/resources/configs/named-cache-test.xml 2009-07-17 08:38:59 UTC (rev 590)
+++ trunk/core/src/test/resources/configs/named-cache-test.xml 2009-07-20 14:32:07 UTC (rev 591)
@@ -97,7 +97,7 @@
<property name="location" value="/tmp/FileCacheStore-Location"/>
</properties>
<singletonStore enabled="true" pushStateWhenCoordinator="true" pushStateTimeout="20000"/>
- <async enabled="true" batchSize="1000" threadPoolSize="5"/>
+ <async enabled="true" mapLockTimeout="15000" threadPoolSize="5"/>
</loader>
</loaders>
</namedCache>
More information about the infinispan-commits
mailing list