Author: manik.surtani(a)jboss.com Date: 2008-10-02 10:12:36 -0400 (Thu, 02 Oct 2008) New Revision: 6827 Added: core/trunk/src/test/resources/configs/parser-test-async.xml Modified: core/trunk/src/main/java/org/jboss/cache/config/Configuration.java core/trunk/src/main/java/org/jboss/cache/config/parsing/XmlConfigurationParser.java core/trunk/src/main/java/org/jboss/cache/marshall/CommandAwareRpcDispatcher.java core/trunk/src/main/java/org/jboss/cache/util/concurrent/WithinThreadExecutor.java core/trunk/src/main/resources/schema/jbosscache-config-3.0.xsd core/trunk/src/test/java/org/jboss/cache/config/parsing/XmlConfigurationParserTest.java core/trunk/src/test/java/org/jboss/cache/config/parsing/XmlConfigurationSchemaTest.java Log: JBCACHE-1419: If async repl is used, marshalling should be done in a separate thread Modified: core/trunk/src/main/java/org/jboss/cache/config/Configuration.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/config/Configuration.java 2008-10-02 13:55:54 UTC (rev 6826) +++ core/trunk/src/main/java/org/jboss/cache/config/Configuration.java 2008-10-02 14:12:36 UTC (rev 6827) @@ -222,6 +222,7 @@ private boolean writeSkewCheck = false; private int concurrencyLevel = 500; private int listenerAsyncPoolSize = 1; + private int serializationExecutorPoolSize = 25; @Start(priority = 1) private void correctIsolationLevels() @@ -630,6 +631,18 @@ this.objectOutputStreamPoolSize = objectOutputStreamPoolSize; } + /** + * Sets the async replication serialization executor pool size for async replication. Has no effect if the + * replication queue is used. + * + * @param serializationExecutorPoolSize number of threads to use + */ + public void setSerializationExecutorPoolSize(int serializationExecutorPoolSize) + { + testImmutability("serializationExecutorPoolSize"); + this.serializationExecutorPoolSize = serializationExecutorPoolSize; + } + // ------------------------------------------------------------------------------------------------------------ // GETTERS // ------------------------------------------------------------------------------------------------------------ @@ -878,6 +891,14 @@ return url; } + /** + * @return the serialization executor pool size. + */ + public int getSerializationExecutorPoolSize() + { + return serializationExecutorPoolSize; + } + // ------------------------------------------------------------------------------------------------------------ // HELPERS // ------------------------------------------------------------------------------------------------------------ @@ -932,6 +953,7 @@ if (transactionManagerLookupClass != null ? !transactionManagerLookupClass.equals(that.transactionManagerLookupClass) : that.transactionManagerLookupClass != null) return false; if (listenerAsyncPoolSize != that.listenerAsyncPoolSize) return false; + if (serializationExecutorPoolSize != that.serializationExecutorPoolSize) return false; return true; } @@ -972,6 +994,7 @@ result = 31 * result + (useLazyDeserialization ? 1 : 0); result = 31 * result + objectInputStreamPoolSize; result = 31 * result + objectOutputStreamPoolSize; + result = 31 * result + serializationExecutorPoolSize; return result; } @@ -1042,8 +1065,9 @@ this.customInterceptors = customInterceptors; } - public BuddyManager getConsistentHashing() { - return null; - } - + public BuddyManager getConsistentHashing() + { + return null; + } + } Modified: core/trunk/src/main/java/org/jboss/cache/config/parsing/XmlConfigurationParser.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/config/parsing/XmlConfigurationParser.java 2008-10-02 13:55:54 UTC (rev 6826) +++ core/trunk/src/main/java/org/jboss/cache/config/parsing/XmlConfigurationParser.java 2008-10-02 14:12:36 UTC (rev 6827) @@ -359,6 +359,9 @@ if (existsAttribute(replQueueInterval)) config.setReplQueueInterval(getLong(replQueueInterval)); String replQueueMaxElements = getAttributeValue(element, "replQueueMaxElements"); if (existsAttribute(replQueueMaxElements)) config.setReplQueueMaxElements(getInt(replQueueMaxElements)); + String serializationExecutorPoolSize = getAttributeValue(element, "serializationExecutorPoolSize"); + if (existsAttribute(serializationExecutorPoolSize)) + config.setSerializationExecutorPoolSize(getInt(serializationExecutorPoolSize)); } private void configureLocking(Element element) Modified: core/trunk/src/main/java/org/jboss/cache/marshall/CommandAwareRpcDispatcher.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/marshall/CommandAwareRpcDispatcher.java 2008-10-02 13:55:54 UTC (rev 6826) +++ core/trunk/src/main/java/org/jboss/cache/marshall/CommandAwareRpcDispatcher.java 2008-10-02 14:12:36 UTC (rev 6827) @@ -27,15 +27,16 @@ import org.jboss.cache.commands.remote.AnnounceBuddyPoolNameCommand; import org.jboss.cache.commands.remote.AssignToBuddyGroupCommand; import org.jboss.cache.commands.remote.RemoveFromBuddyGroupCommand; +import org.jboss.cache.config.Configuration; import org.jboss.cache.factories.ComponentRegistry; import org.jboss.cache.interceptors.InterceptorChain; import org.jboss.cache.invocation.InvocationContextContainer; +import org.jboss.cache.util.concurrent.WithinThreadExecutor; import org.jgroups.Address; import org.jgroups.Channel; import org.jgroups.MembershipListener; import org.jgroups.Message; import org.jgroups.MessageListener; -import org.jgroups.blocks.GroupRequest; import org.jgroups.blocks.RpcDispatcher; import org.jgroups.blocks.RspFilter; import org.jgroups.util.Buffer; @@ -44,6 +45,14 @@ import java.io.NotSerializableException; import java.util.Vector; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; /** * A JGroups RPC dispatcher that knows how to deal with {@link org.jboss.cache.commands.ReplicableCommand}s. @@ -57,6 +66,9 @@ protected InterceptorChain interceptorChain; protected ComponentRegistry componentRegistry; protected boolean trace; + private ExecutorService replicationProcessor; + private AtomicInteger replicationProcessorCount; + private boolean asyncSerial; public CommandAwareRpcDispatcher() { @@ -71,8 +83,46 @@ this.componentRegistry = componentRegistry; this.interceptorChain = interceptorChain; trace = log.isTraceEnabled(); + + // what sort of a repl processor do we need? + Configuration c = componentRegistry.getComponent(Configuration.class); + if (c.getCacheMode().isSynchronous() || c.getSerializationExecutorPoolSize() == 0) + { + // in-process thread. Not async. + replicationProcessor = new WithinThreadExecutor(); + asyncSerial = false; + } + else + { + asyncSerial = true; + replicationProcessorCount = new AtomicInteger(0); + replicationProcessor = Executors.newFixedThreadPool(c.isUseReplQueue() ? 1 : c.getSerializationExecutorPoolSize(), + new ThreadFactory() + { + public Thread newThread(Runnable r) + { + return new Thread(r, "AsyncReplicationProcessor-" + replicationProcessorCount.incrementAndGet()); + } + } + ); + } } + @Override + public void stop() + { + replicationProcessor.shutdownNow(); + try + { + replicationProcessor.awaitTermination(60, TimeUnit.SECONDS); + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + } + super.stop(); + } + protected boolean isValid(Message req) { if (server_obj == null) @@ -95,7 +145,7 @@ * is aware of {@link org.jboss.cache.commands.ReplicableCommand} objects. */ public RspList invokeRemoteCommands(Vector<Address> dests, ReplicableCommand command, int mode, long timeout, - boolean anycasting, boolean oob, RspFilter filter) throws NotSerializableException + boolean anycasting, boolean oob, RspFilter filter) throws NotSerializableException, ExecutionException, InterruptedException { if (dests != null && dests.isEmpty()) { @@ -108,34 +158,21 @@ log.trace(new StringBuilder("dests=").append(dests).append(", command=").append(command). append(", mode=").append(mode).append(", timeout=").append(timeout)); - Buffer buf; - try + ReplicationTask replicationTask = new ReplicationTask(command, oob, dests, mode, timeout, anycasting, filter); + Future<RspList> response = replicationProcessor.submit(replicationTask); + if (asyncSerial) { - buf = req_marshaller.objectToBuffer(command); + // don't care about the response. return. + return null; } - catch (Exception e) + else { - throw new RuntimeException("Failure to marshal argument(s)", e); + RspList retval = response.get(); + if (retval.isEmpty() || containsOnlyNulls(retval)) + return null; + else + return retval; } - - Message msg = new Message(); - msg.setBuffer(buf); - if (oob) msg.setFlag(Message.OOB); - RspList retval = super.castMessage(dests, msg, mode, timeout, anycasting, filter); - if (trace) log.trace("responses: " + retval); - - // a null response is 99% likely to be due to a marshalling problem - we throw a NSE, this needs to be changed when - // JGroups supports http://jira.jboss.com/jira/browse/JGRP-193 - // the serialization problem could be on the remote end and this is why we cannot catch this above, when marshalling. - if (retval == null) - { - throw new NotSerializableException("RpcDispatcher returned a null. This is most often caused by args for " + command.getClass().getSimpleName() + " not being serializable."); - } - - if (mode == GroupRequest.GET_NONE || retval.isEmpty() || containsOnlyNulls(retval)) - return null; - else - return retval; } private boolean containsOnlyNulls(RspList l) @@ -207,4 +244,53 @@ { return getClass().getSimpleName() + "[Outgoing marshaller: " + req_marshaller + "; incoming marshaller: " + rsp_marshaller + "]"; } + + private class ReplicationTask implements Callable<RspList> + { + private ReplicableCommand command; + private boolean oob; + private Vector<Address> dests; + private int mode; + private long timeout; + private boolean anycasting; + private RspFilter filter; + + private ReplicationTask(ReplicableCommand command, boolean oob, Vector<Address> dests, int mode, long timeout, boolean anycasting, RspFilter filter) + { + this.command = command; + this.oob = oob; + this.dests = dests; + this.mode = mode; + this.timeout = timeout; + this.anycasting = anycasting; + this.filter = filter; + } + + public RspList call() throws Exception + { + Buffer buf; + try + { + buf = req_marshaller.objectToBuffer(command); + } + catch (Exception e) + { + throw new RuntimeException("Failure to marshal argument(s)", e); + } + + Message msg = new Message(); + msg.setBuffer(buf); + if (oob) msg.setFlag(Message.OOB); + RspList retval = castMessage(dests, msg, mode, timeout, anycasting, filter); + if (trace) log.trace("responses: " + retval); + + // a null response is 99% likely to be due to a marshalling problem - we throw a NSE, this needs to be changed when + // JGroups supports http://jira.jboss.com/jira/browse/JGRP-193 + // the serialization problem could be on the remote end and this is why we cannot catch this above, when marshalling. + + if (retval == null) + throw new NotSerializableException("RpcDispatcher returned a null. This is most often caused by args for " + command.getClass().getSimpleName() + " not being serializable."); + return retval; + } + } } Modified: core/trunk/src/main/java/org/jboss/cache/util/concurrent/WithinThreadExecutor.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/util/concurrent/WithinThreadExecutor.java 2008-10-02 13:55:54 UTC (rev 6826) +++ core/trunk/src/main/java/org/jboss/cache/util/concurrent/WithinThreadExecutor.java 2008-10-02 14:12:36 UTC (rev 6827) @@ -21,6 +21,8 @@ */ package org.jboss.cache.util.concurrent; +import org.jboss.cache.CacheException; + import java.util.Collection; import java.util.Collections; import java.util.List; @@ -75,7 +77,42 @@ public <T> Future<T> submit(Callable<T> task) { - throw new UnsupportedOperationException(); + try + { + final T resp = task.call(); + return new Future<T>() + { + + public boolean cancel(boolean mayInterruptIfRunning) + { + return false; + } + + public boolean isCancelled() + { + return false; + } + + public boolean isDone() + { + return true; + } + + public T get() throws InterruptedException, ExecutionException + { + return resp; + } + + public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException + { + return resp; + } + }; + } + catch (Exception e) + { + throw new CacheException(e); + } } public <T> Future<T> submit(Runnable task, T result) Modified: core/trunk/src/main/resources/schema/jbosscache-config-3.0.xsd =================================================================== --- core/trunk/src/main/resources/schema/jbosscache-config-3.0.xsd 2008-10-02 13:55:54 UTC (rev 6826) +++ core/trunk/src/main/resources/schema/jbosscache-config-3.0.xsd 2008-10-02 14:12:36 UTC (rev 6827) @@ -1,5 +1,6 @@ <?xml version="1.0" encoding="UTF-8"?> -<xs:schema attributeFormDefault="unqualified" elementFormDefault="qualified" xmlns:tns="urn:jboss:jbosscache-core:config:3.0" targetNamespace="urn:jboss:jbosscache-core:config:3.0" +<xs:schema attributeFormDefault="unqualified" elementFormDefault="qualified" + xmlns:tns="urn:jboss:jbosscache-core:config:3.0" targetNamespace="urn:jboss:jbosscache-core:config:3.0" xmlns:xs="http://www.w3.org/2001/XMLSchema" version="1.0"> <xs:element name="jbosscache" type="tns:cacheConfigurationType"/> @@ -28,7 +29,8 @@ <xs:attribute name="isolationLevel"> <xs:simpleType> <xs:restriction base="xs:string"> - <xs:pattern value="[Ss][Ee][Rr][Ii][Aa][Ll][Ii][Zz][Aa][Bb][Ll][Ee]|[Rr][Ee][Pp][Ee][Aa][Tt][Aa][Bb][Ll][Ee]_[Rr][Ee][Aa][Dd]|[Rr][Ee][Aa][Dd]_[Cc][Oo][Mm][Mm][Ii][Tt][Tt][Ee][Dd]|[Nn][Oo][Nn][Ee]|\$\{.*\}"/> + <xs:pattern + value="[Ss][Ee][Rr][Ii][Aa][Ll][Ii][Zz][Aa][Bb][Ll][Ee]|[Rr][Ee][Pp][Ee][Aa][Tt][Aa][Bb][Ll][Ee]_[Rr][Ee][Aa][Dd]|[Rr][Ee][Aa][Dd]_[Cc][Oo][Mm][Mm][Ii][Tt][Tt][Ee][Dd]|[Nn][Oo][Nn][Ee]|\$\{.*\}"/> </xs:restriction> </xs:simpleType> </xs:attribute> @@ -37,7 +39,8 @@ <xs:attribute name="nodeLockingScheme"> <xs:simpleType> <xs:restriction base="xs:string"> - <xs:pattern value="[Mm][Vv][Cc][Cc]|[Oo][Pp][Tt][Ii][Mm][Ii][Ss][Tt][Ii][Cc]|[Pp][Ee][Ss][Ss][Ii][Mm][Ii][Ss][Tt][Ii][Cc]|\$\{.*\}"/> + <xs:pattern + value="[Mm][Vv][Cc][Cc]|[Oo][Pp][Tt][Ii][Mm][Ii][Ss][Tt][Ii][Cc]|[Pp][Ee][Ss][Ss][Ii][Mm][Ii][Ss][Tt][Ii][Cc]|\$\{.*\}"/> </xs:restriction> </xs:simpleType> </xs:attribute> @@ -64,7 +67,8 @@ <xs:attribute name="hookBehavior"> <xs:simpleType> <xs:restriction base="xs:string"> - <xs:pattern value="[Dd][Ee][Ff][Aa][Uu][Ll][Tt]|[Rr][Ee][Gg][Ii][Ss][Tt][Ee][Rr]|[Dd][Oo][Nn][Tt]_[Rr][Ee][Gg][Ii][Ss][Tt][Ee][Rr]|\$\{.*\}"/> + <xs:pattern + value="[Dd][Ee][Ff][Aa][Uu][Ll][Tt]|[Rr][Ee][Gg][Ii][Ss][Tt][Ee][Rr]|[Dd][Oo][Nn][Tt]_[Rr][Ee][Gg][Ii][Ss][Tt][Ee][Rr]|\$\{.*\}"/> </xs:restriction> </xs:simpleType> </xs:attribute> @@ -155,11 +159,11 @@ <xs:attribute name="replTimeout" type="tns:positiveInteger"/> </xs:complexType> - <xs:complexType name="asyncType"> <xs:attribute name="useReplQueue" type="tns:booleanType"/> <xs:attribute name="replQueueInterval" type="tns:positiveInteger"/> <xs:attribute name="replQueueMaxElements" type="tns:positiveInteger"/> + <xs:attribute name="serializationExecutorPoolSize" type="tns:positiveInteger"/> </xs:complexType> <xs:complexType name="evictionType"> Modified: core/trunk/src/test/java/org/jboss/cache/config/parsing/XmlConfigurationParserTest.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/config/parsing/XmlConfigurationParserTest.java 2008-10-02 13:55:54 UTC (rev 6826) +++ core/trunk/src/test/java/org/jboss/cache/config/parsing/XmlConfigurationParserTest.java 2008-10-02 14:12:36 UTC (rev 6827) @@ -1,12 +1,18 @@ package org.jboss.cache.config.parsing; -import org.jboss.cache.config.*; +import org.jboss.cache.config.BuddyReplicationConfig; +import org.jboss.cache.config.CacheLoaderConfig; +import org.jboss.cache.config.Configuration; +import org.jboss.cache.config.ConfigurationException; +import org.jboss.cache.config.CustomInterceptorConfig; +import org.jboss.cache.config.EvictionConfig; +import org.jboss.cache.config.EvictionRegionConfig; import org.jboss.cache.config.parsing.custominterceptors.AaaCustomInterceptor; import org.jboss.cache.config.parsing.custominterceptors.BbbCustomInterceptor; import org.jboss.cache.eviction.LRUAlgorithmConfig; import org.jboss.cache.eviction.MRUAlgorithmConfig; import org.jboss.cache.lock.IsolationLevel; -import org.testng.annotations.BeforeMethod; +import org.testng.annotations.BeforeTest; import org.testng.annotations.Test; import java.util.List; @@ -21,18 +27,18 @@ @Test(groups = "functional") public class XmlConfigurationParserTest { - Configuration config; + Configuration syncConfig, asyncConfig; - @BeforeMethod + @BeforeTest public void setUp() { - XmlConfigurationParser parser = new XmlConfigurationParser(false, null); - config = parser.parseFile("configs/parser-test.xml"); + syncConfig = new XmlConfigurationParser(false, null).parseFile("configs/parser-test.xml"); + asyncConfig = new XmlConfigurationParser(false, null).parseFile("configs/parser-test-async.xml"); } public void testParseOldConfigFile() { - System.setProperty("jbosscache.config.validate","false"); + System.setProperty("jbosscache.config.validate", "false"); XmlConfigurationParser parser = new XmlConfigurationParser(); try { @@ -43,124 +49,132 @@ { //expectd } - finally { - System.setProperty("jbosscache.config.validate","true"); + finally + { + System.setProperty("jbosscache.config.validate", "true"); } } public void testTransactionManagerLookupClass() { - assert config.getTransactionManagerLookupClass().equals("org.jboss.cache.transaction.GenericTransactionManagerLookup"); + assert syncConfig.getTransactionManagerLookupClass().equals("org.jboss.cache.transaction.GenericTransactionManagerLookup"); } public void testIsolationLevel() { - assert config.getIsolationLevel().equals(IsolationLevel.REPEATABLE_READ); + assert syncConfig.getIsolationLevel().equals(IsolationLevel.REPEATABLE_READ); } public void testCacheMode() { - assert config.getCacheMode().equals(Configuration.CacheMode.REPL_SYNC); + assert syncConfig.getCacheMode().equals(Configuration.CacheMode.REPL_SYNC); + assert asyncConfig.getCacheMode().equals(Configuration.CacheMode.REPL_ASYNC); } + public void testAsyncSerializationExecutorSize() + { + assert asyncConfig.getSerializationExecutorPoolSize() == 250; + } + public void testUseReplQueue() { - assert !config.isUseReplQueue(); + assert !syncConfig.isUseReplQueue(); + assert !asyncConfig.isUseReplQueue(); } public void testClusterName() { - assert config.getClusterName().equals("JBossCache-Cluster"); + assert syncConfig.getClusterName().equals("JBossCache-Cluster"); } public void testGetClusterConfig() { - assert config.getClusterConfig().indexOf("MERGE2") >= 0; + assert syncConfig.getClusterConfig().indexOf("MERGE2") >= 0; } public void testFetchInMemoryState() { - assert config.isFetchInMemoryState(); + assert syncConfig.isFetchInMemoryState(); } public void testStateRetrievalTimeout() { - assert config.getStateRetrievalTimeout() == 15124; + assert syncConfig.getStateRetrievalTimeout() == 15124; } public void testSyncReplTimeout() { - assert config.getSyncReplTimeout() == 15421; + assert syncConfig.getSyncReplTimeout() == 15421; } public void testLockAcquisitionTimeout() { - assert config.getLockAcquisitionTimeout() == 10234; + assert syncConfig.getLockAcquisitionTimeout() == 10234; } public void testUseLazyDeserialization() { - assert config.isUseLazyDeserialization(); + assert syncConfig.isUseLazyDeserialization(); } public void testObjectInputStreamPoolSize() { - assert 12 == config.getObjectInputStreamPoolSize(); + assert 12 == syncConfig.getObjectInputStreamPoolSize(); } public void testObjectOutputStreamPoolSize() { - assert 14 == config.getObjectOutputStreamPoolSize(); + assert 14 == syncConfig.getObjectOutputStreamPoolSize(); } public void testShutdownHookBehavior() { - assert Configuration.ShutdownHookBehavior.REGISTER == config.getShutdownHookBehavior(); + assert Configuration.ShutdownHookBehavior.REGISTER == syncConfig.getShutdownHookBehavior(); } public void testSyncRollbackPhase() { - assert config.isSyncRollbackPhase(); + assert syncConfig.isSyncRollbackPhase(); } public void testSyncCommitPhase() { - assert config.isSyncCommitPhase(); + assert syncConfig.isSyncCommitPhase(); } public void testUseReplicationVersion() { - assert config.getReplicationVersion() == 124; + assert syncConfig.getReplicationVersion() == 124; } public void testGetMultiplexerStack() { - assert "file_name".equals(config.getMultiplexerStack()); + assert "file_name".equals(syncConfig.getMultiplexerStack()); } public void testMarshallerClass() { - assert "some.Clazz".equals(config.getMarshallerClass()); + assert "some.Clazz".equals(syncConfig.getMarshallerClass()); } public void testLockParentForChildInsertRemove() { - assert config.isLockParentForChildInsertRemove(); + assert syncConfig.isLockParentForChildInsertRemove(); } public void testInactiveOnStartup() { - assert config.isInactiveOnStartup(); + assert syncConfig.isInactiveOnStartup(); } public void testExposeManagementStatistics() { - assert !config.getExposeManagementStatistics(); + assert !syncConfig.getExposeManagementStatistics(); } public void testCacheLoaderConfiguration() { - CacheLoaderConfig clc = config.getCacheLoaderConfig(); + CacheLoaderConfig clc = syncConfig.getCacheLoaderConfig(); assert null != clc; assert clc.isPassivation(); assert clc.isShared(); @@ -178,7 +192,7 @@ public void testBuddyReplicationConfig() { - BuddyReplicationConfig brConfig = config.getBuddyReplicationConfig(); + BuddyReplicationConfig brConfig = syncConfig.getBuddyReplicationConfig(); assert brConfig.isEnabled(); BuddyReplicationConfig.BuddyLocatorConfig locatorConfig = brConfig.getBuddyLocatorConfig(); assert "org.jboss.cache.buddyreplication.NextMemberBuddyLocator".equals(locatorConfig.getBuddyLocatorClass()); @@ -193,12 +207,12 @@ public void testUseRegionBasedMarshalling() { - assert config.isUseRegionBasedMarshalling(); + assert syncConfig.isUseRegionBasedMarshalling(); } public void testEvictionPolicyConfig() { - EvictionConfig evictionConfig = config.getEvictionConfig(); + EvictionConfig evictionConfig = syncConfig.getEvictionConfig(); assert "org.jboss.cache.eviction.LRUAlgorithm".equals(evictionConfig.getDefaultEvictionRegionConfig().getEvictionAlgorithmConfig().getEvictionAlgorithmClassName()); assert 200000 == evictionConfig.getDefaultEvictionRegionConfig().getEventQueueSize(); assert 5 == evictionConfig.getWakeupInterval(); @@ -229,7 +243,7 @@ public void testCustomInterceptors() { - List<CustomInterceptorConfig> interceptorConfigs = config.getCustomInterceptors(); + List<CustomInterceptorConfig> interceptorConfigs = syncConfig.getCustomInterceptors(); assert interceptorConfigs.size() == 5; assert interceptorConfigs.get(0).getInterceptor() instanceof AaaCustomInterceptor; assert interceptorConfigs.get(1).getInterceptor() instanceof BbbCustomInterceptor; @@ -246,7 +260,7 @@ public void testSingletonStore() { - CacheLoaderConfig.IndividualCacheLoaderConfig clc = config.getCacheLoaderConfig().getFirstCacheLoaderConfig(); + CacheLoaderConfig.IndividualCacheLoaderConfig clc = syncConfig.getCacheLoaderConfig().getFirstCacheLoaderConfig(); assert clc != null; CacheLoaderConfig.IndividualCacheLoaderConfig.SingletonStoreConfig singlStoreConf = clc.getSingletonStoreConfig(); assert singlStoreConf != null; @@ -259,17 +273,17 @@ public void testMvccAttributes() { - assert !config.isWriteSkewCheck(); - assert config.getConcurrencyLevel() == 21; + assert !syncConfig.isWriteSkewCheck(); + assert syncConfig.getConcurrencyLevel() == 21; } public void testListenerAsyncThreads() { - assert config.getListenerAsyncPoolSize() == 5; + assert syncConfig.getListenerAsyncPoolSize() == 5; } public void testInvocationBatching() { - assert config.isInvocationBatchingEnabled(); + assert syncConfig.isInvocationBatchingEnabled(); } } Modified: core/trunk/src/test/java/org/jboss/cache/config/parsing/XmlConfigurationSchemaTest.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/config/parsing/XmlConfigurationSchemaTest.java 2008-10-02 13:55:54 UTC (rev 6826) +++ core/trunk/src/test/java/org/jboss/cache/config/parsing/XmlConfigurationSchemaTest.java 2008-10-02 14:12:36 UTC (rev 6827) @@ -30,6 +30,7 @@ "mixedPolicy-eviction.xml", "mux.xml", "parser-test.xml", + "parser-test-async.xml", "policyPerRegion-eviction.xml", "replSync.xml", "string-property-replaced.xml", @@ -42,7 +43,7 @@ public void testSimpleFile() { ExceptionCountingErrorHandler handler = new ExceptionCountingErrorHandler(); - System.setProperty("jbosscache.config.schemaLocation","src/main/resources/jbosscache-config-3.0.xsd"); + System.setProperty("jbosscache.config.schemaLocation", "src/main/resources/jbosscache-config-3.0.xsd"); XmlConfigurationParser parser = new XmlConfigurationParser(handler); for (String file : testFiles) { Copied: core/trunk/src/test/resources/configs/parser-test-async.xml (from rev 6825, core/trunk/src/test/resources/configs/parser-test.xml) =================================================================== --- core/trunk/src/test/resources/configs/parser-test-async.xml (rev 0) +++ core/trunk/src/test/resources/configs/parser-test-async.xml 2008-10-02 14:12:36 UTC (rev 6827) @@ -0,0 +1,117 @@ +<?xml version="1.0" encoding="UTF-8"?> + +<!-- file used for functional test of the xml parser --> + +<jbosscache xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xmlns="urn:jboss:jbosscache-core:config:3.0"> + + + <locking isolationLevel="REPEATABLE_READ" lockParentForChildInsertRemove="true" lockAcquisitionTimeout="10234" + nodeLockingScheme="mvcc" writeSkewCheck="false" concurrencyLevel="21"/> + + <transaction transactionManagerLookupClass="org.jboss.cache.transaction.GenericTransactionManagerLookup" + syncRollbackPhase="true" syncCommitPhase="true"/> + + <!-- serialization related configuration, used for replication and cache loading --> + <serialization objectInputStreamPoolSize="12" objectOutputStreamPoolSize="14" version="1.2.4" + marshallerClass="some.Clazz" useLazyDeserialization="true" useRegionBasedMarshalling="true"/> + + <replication> + <async useReplQueue="false" serializationExecutorPoolSize="250"/> + <buddy enabled="true" poolName="myBuddyPoolReplicationGroup" communicationTimeout="2000"> + <dataGravitation auto="true" removeOnFind="true" searchBackupTrees="true"/> + <locator class="org.jboss.cache.buddyreplication.NextMemberBuddyLocator"> + <properties> + numBuddies = 1 + ignoreColocatedBuddies = true + </properties> + </locator> + </buddy> + </replication> + + <stateRetrieval timeout="15124" fetchInMemoryState="true"/> + <startup regionsInactiveOnStartup="true"/> + <shutdown hookBehavior="REGISTER"/> + + <transport clusterName="JBossCache-Cluster" multiplexerStack="file_name"> + <jgroupsConfig> + <PING timeout="2000" num_initial_members="3"/> + <MERGE2 max_interval="30000" min_interval="10000"/> + <FD_SOCK/> + <FD timeout="10000" max_tries="5" shun="true"/> + <VERIFY_SUSPECT timeout="1500"/> + <pbcast.NAKACK use_mcast_xmit="false" gc_lag="0" + retransmit_timeout="300,600,1200,2400,4800" + discard_delivered_msgs="true"/> + <UNICAST timeout="300,600,1200,2400,3600"/> + <pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000" + max_bytes="400000"/> + <pbcast.GMS print_local_addr="true" join_timeout="5000" shun="false" + view_bundling="true" view_ack_collection_timeout="5000"/> + <FRAG2 frag_size="60000"/> + <pbcast.STREAMING_STATE_TRANSFER use_reading_thread="true"/> + <!-- <pbcast.STATE_TRANSFER/> --> + <pbcast.FLUSH timeout="0"/> + </jgroupsConfig> + </transport> + + + <jmxStatistics enabled="false"/> + + <eviction wakeUpInterval="5"> + <default algorithmClass="org.jboss.cache.eviction.LRUAlgorithm" eventQueueSize="200000"> + <attribute name="maxNodes">5000</attribute> + <attribute name="timeToLive">1000</attribute> + </default> + <region name="/org/jboss/data"> + <attribute name="timeToLive">1002</attribute> + </region> + <region name="/org/jboss/xyz" algorithmClass="org.jboss.cache.eviction.MRUAlgorithm" eventQueueSize="21"> + <attribute name="maxNodes">2103</attribute> + <attribute name="minTimeToLive">22</attribute> + </region> + </eviction> + + <!-- this should be deprecated in 3.0 and should be replaced with CacheLoaderConfig--> + <loaders passivation="true" shared="true"> + <preload> + <node fqn="/a/b/c"/> + <node fqn="/f/r/s"/> + </preload> + + <!-- we can now have multiple cache loaders, which get chained --> + <loader class="org.jboss.cache.loader.JDBCCacheLoader" async="true" fetchPersistentState="true" + ignoreModifications="true" purgeOnStartup="true"> + <properties> + cache.jdbc.table.name=jbosscache + cache.jdbc.table.create=true + cache.jdbc.table.drop=true + </properties> + <singletonStore enabled="true" class="org.jboss.cache.loader.SingletonStoreCacheLoader"> + <properties> + pushStateWhenCoordinator=true + pushStateWhenCoordinatorTimeout=20000 + </properties> + </singletonStore> + </loader> + </loaders> + + <!-- this is new behavior added within 3.x only. it support configuring custom interceptors through configurations --> + <customInterceptors> + <interceptor position="first" class="org.jboss.cache.config.parsing.custominterceptors.AaaCustomInterceptor"> + <attribute name="attrOne">value1</attribute> + <attribute name="attrTwo">value2</attribute> + <attribute name="attrThree">value3</attribute> + </interceptor> + <interceptor position="last" class="org.jboss.cache.config.parsing.custominterceptors.BbbCustomInterceptor"/> + <interceptor index="3" class="org.jboss.cache.config.parsing.custominterceptors.AaaCustomInterceptor"/> + <interceptor before="org.jboss.cache.interceptors.CallInterceptor" + class="org.jboss.cache.config.parsing.custominterceptors.BbbCustomInterceptor"/> + <interceptor after="org.jboss.cache.interceptors.CallInterceptor" + class="org.jboss.cache.config.parsing.custominterceptors.AaaCustomInterceptor"/> + </customInterceptors> + + <!-- the number of threads to use for asynchronous cache listeners - defaults to 1 --> + <listeners asyncPoolSize="5"/> + <invocationBatching enabled="true"/> +</jbosscache>