From b9c32d6aa5c209338ef86eff407d7399b5c6cf51 Mon Sep 17 00:00:00 2001 From: Derek Hulley Date: Mon, 24 Oct 2011 15:57:57 +0000 Subject: [PATCH] Performance improvement during property writes - Touch node and copy node caches *before* writing updated cache entry - Added concurrency tests for aspect updates git-svn-id: https://svn.alfresco.com/repos/alfresco-enterprise/alfresco/HEAD/root@31446 c4b6b30b-aa2e-2d43-bbcb-ca4b014f7261 --- .../repo/domain/node/AbstractNodeDAOImpl.java | 55 +-- .../repo/node/BaseNodeServiceTest_model.xml | 21 + .../repo/node/ConcurrentNodeServiceTest.java | 375 ++++++------------ 3 files changed, 169 insertions(+), 282 deletions(-) diff --git a/source/java/org/alfresco/repo/domain/node/AbstractNodeDAOImpl.java b/source/java/org/alfresco/repo/domain/node/AbstractNodeDAOImpl.java index 9757421b31..cf3d0d8269 100644 --- a/source/java/org/alfresco/repo/domain/node/AbstractNodeDAOImpl.java +++ b/source/java/org/alfresco/repo/domain/node/AbstractNodeDAOImpl.java @@ -1606,7 +1606,16 @@ public abstract class AbstractNodeDAOImpl implements NodeDAO, BatchingDAO } // The node is remaining in the current store - int count = updateNode(nodeUpdate); + int count = 0; + Throwable concurrencyException = null; + try + { + count = updateNode(nodeUpdate); + } + catch (Throwable e) + { + concurrencyException = e; + } // Do concurrency check if (count != 1) { @@ -1614,7 +1623,7 @@ public abstract class AbstractNodeDAOImpl implements NodeDAO, BatchingDAO nodesCache.removeByKey(nodeId); nodesCache.removeByValue(nodeUpdate); - throw new ConcurrencyFailureException("Failed to update node " + nodeId); + throw new ConcurrencyFailureException("Failed to update node " + nodeId, concurrencyException); } else { @@ -1622,7 +1631,6 @@ public abstract class AbstractNodeDAOImpl implements NodeDAO, BatchingDAO nodeUpdate.lock(); nodesCache.setValue(nodeId, nodeUpdate); // The node's version has moved on so no need to invalidate caches - // TODO: Should we copy values between the cache keys? } // Done @@ -1969,7 +1977,18 @@ public abstract class AbstractNodeDAOImpl implements NodeDAO, BatchingDAO boolean modifyProps = propsToDelete.size() > 0 || propsToAdd.size() > 0; boolean updated = modifyProps || nodeUpdate.isUpdateAnything(); - // Touch to bring into current txn + // Now we know that the node needs modification or not; grab the lock + if (nodeUpdate.isUpdateAnything()) + { + // We have to explicitly update the node (sys:locale or cm:auditable) + updateNodeImpl(node, nodeUpdate); + } + else if (modifyProps) + { + // Touch the node; all caches are fine + touchNode(nodeId, null, false, false, false); + } + if (modifyProps) { // Clean up content properties @@ -2032,12 +2051,6 @@ public abstract class AbstractNodeDAOImpl implements NodeDAO, BatchingDAO // Update cache setNodePropertiesCached(nodeId, propsToCache); } - // Touch to bring into current transaction - if (updated) - { - // We have to explicitly update the node (sys:locale or cm:auditable) - updateNodeImpl(node, nodeUpdate); - } // Done if (isDebugEnabled && updated) @@ -2099,12 +2112,12 @@ public abstract class AbstractNodeDAOImpl implements NodeDAO, BatchingDAO if (deleteCount > 0) { + // Touch the node; all caches are fine + touchNode(nodeId, null, false, false, false); // Update cache Map cachedProps = getNodePropertiesCached(nodeId); cachedProps.keySet().removeAll(propertyQNames); setNodePropertiesCached(nodeId, cachedProps); - // Touch the node; all caches are fine - touchNode(nodeId, null, false, false, false); } // Done return deleteCount > 0; @@ -2333,11 +2346,7 @@ public abstract class AbstractNodeDAOImpl implements NodeDAO, BatchingDAO executeBatch(); } - // Manually update the cache - Set newAspectQNames = new HashSet(existingAspectQNames); - newAspectQNames.addAll(aspectQNamesToAdd); - setNodeAspectsCached(nodeId, newAspectQNames); - + // Bring the node into the transaction if (aspectQNamesToAdd.contains(ContentModel.ASPECT_ROOT)) { // This is a special case. The presence of the aspect affects the path @@ -2349,6 +2358,11 @@ public abstract class AbstractNodeDAOImpl implements NodeDAO, BatchingDAO // Touch the node; all caches are fine touchNode(nodeId, null, false, false, false); } + + // Manually update the cache + Set newAspectQNames = new HashSet(existingAspectQNames); + newAspectQNames.addAll(aspectQNamesToAdd); + setNodeAspectsCached(nodeId, newAspectQNames); // Done return true; @@ -2746,13 +2760,6 @@ public abstract class AbstractNodeDAOImpl implements NodeDAO, BatchingDAO // Persist it assoc.setId(assocId); - // Primary associations accompany new nodes, so we only have to bring the - // node into the current transaction for secondary associations - if (!isPrimary) - { - updateNode(childNodeId, null, null); - } - // Done if (isDebugEnabled) { diff --git a/source/java/org/alfresco/repo/node/BaseNodeServiceTest_model.xml b/source/java/org/alfresco/repo/node/BaseNodeServiceTest_model.xml index 1a205197b4..13d349f638 100644 --- a/source/java/org/alfresco/repo/node/BaseNodeServiceTest_model.xml +++ b/source/java/org/alfresco/repo/node/BaseNodeServiceTest_model.xml @@ -468,6 +468,27 @@ + + + + + + + + + + + + + + + + + + + + + diff --git a/source/java/org/alfresco/repo/node/ConcurrentNodeServiceTest.java b/source/java/org/alfresco/repo/node/ConcurrentNodeServiceTest.java index ea872d279d..ae0cbf4522 100644 --- a/source/java/org/alfresco/repo/node/ConcurrentNodeServiceTest.java +++ b/source/java/org/alfresco/repo/node/ConcurrentNodeServiceTest.java @@ -19,35 +19,25 @@ package org.alfresco.repo.node; import java.io.InputStream; +import java.io.Serializable; import java.util.ArrayList; -import java.util.HashMap; import java.util.List; import java.util.Map; - -import javax.transaction.UserTransaction; +import java.util.Set; import junit.framework.TestCase; -import org.alfresco.model.ContentModel; +import org.alfresco.repo.cache.SimpleCache; import org.alfresco.repo.dictionary.DictionaryDAO; import org.alfresco.repo.dictionary.M2Model; -import org.alfresco.repo.search.impl.lucene.fts.FullTextSearchIndexer; import org.alfresco.repo.security.authentication.AuthenticationComponent; import org.alfresco.repo.security.authentication.AuthenticationUtil; -import org.alfresco.repo.tagging.TaggingServiceImpl; import org.alfresco.repo.transaction.RetryingTransactionHelper; import org.alfresco.repo.transaction.RetryingTransactionHelper.RetryingTransactionCallback; import org.alfresco.service.ServiceRegistry; -import org.alfresco.service.cmr.repository.ChildAssociationRef; -import org.alfresco.service.cmr.repository.MLText; import org.alfresco.service.cmr.repository.NodeRef; import org.alfresco.service.cmr.repository.NodeService; import org.alfresco.service.cmr.repository.StoreRef; -import org.alfresco.service.cmr.search.ResultSet; -import org.alfresco.service.cmr.search.SearchService; -import org.alfresco.service.namespace.DynamicNamespacePrefixResolver; -import org.alfresco.service.namespace.NamespacePrefixResolver; -import org.alfresco.service.namespace.NamespaceService; import org.alfresco.service.namespace.QName; import org.alfresco.service.transaction.TransactionService; import org.alfresco.util.ApplicationContextHelper; @@ -79,7 +69,6 @@ public class ConcurrentNodeServiceTest extends TestCase private NodeService nodeService; private TransactionService transactionService; - private RetryingTransactionHelper retryingTransactionHelper; private NodeRef rootNodeRef; private AuthenticationComponent authenticationComponent; @@ -103,10 +92,10 @@ public class ConcurrentNodeServiceTest extends TestCase model = M2Model.createModel(modelStream); dictionaryDao.putModel(model); - nodeService = (NodeService) ctx.getBean("dbNodeService"); - transactionService = (TransactionService) ctx.getBean("transactionComponent"); - retryingTransactionHelper = (RetryingTransactionHelper) ctx.getBean("retryingTransactionHelper"); - this.authenticationComponent = (AuthenticationComponent) ctx.getBean("authenticationComponent"); + ServiceRegistry serviceRegistry = (ServiceRegistry) ctx.getBean(ServiceRegistry.SERVICE_REGISTRY); + nodeService = serviceRegistry.getNodeService(); + transactionService = serviceRegistry.getTransactionService(); + authenticationComponent = (AuthenticationComponent) ctx.getBean("authenticationComponent"); this.authenticationComponent.setSystemUserAsCurrentUser(); @@ -121,7 +110,7 @@ public class ConcurrentNodeServiceTest extends TestCase return null; } }; - retryingTransactionHelper.doInTransaction(createRootNodeCallback); + transactionService.getRetryingTransactionHelper().doInTransaction(createRootNodeCallback); } @Override @@ -130,184 +119,6 @@ public class ConcurrentNodeServiceTest extends TestCase authenticationComponent.clearCurrentSecurityContext(); super.tearDown(); } - - protected Map buildNodeGraph() throws Exception - { - return BaseNodeServiceTest.buildNodeGraph(nodeService, rootNodeRef); - } - - protected Map commitNodeGraph() throws Exception - { - RetryingTransactionCallback> buildGraphCallback = - new RetryingTransactionCallback>() - { - public Map execute() throws Exception - { - - Map answer = buildNodeGraph(); - return answer; - } - }; - return retryingTransactionHelper.doInTransaction(buildGraphCallback); - } - - public void xtest1() throws Exception - { - testConcurrent(); - } - - public void xtest2() throws Exception - { - testConcurrent(); - } - - public void xtest3() throws Exception - { - testConcurrent(); - } - - public void xtest4() throws Exception - { - testConcurrent(); - } - - public void xtest5() throws Exception - { - testConcurrent(); - } - - public void xtest6() throws Exception - { - testConcurrent(); - } - - public void xtest7() throws Exception - { - testConcurrent(); - } - - public void xtest8() throws Exception - { - testConcurrent(); - } - - public void xtest9() throws Exception - { - testConcurrent(); - } - - public void xtest10() throws Exception - { - testConcurrent(); - } - - public void testConcurrent() throws Exception - { - Map assocRefs = commitNodeGraph(); - Thread runner = null; - - for (int i = 0; i < COUNT; i++) - { - runner = new Nester("Concurrent-" + i, runner, REPEATS); - } - if (runner != null) - { - runner.start(); - - try - { - runner.join(); - System.out.println("Query thread has waited for " + runner.getName()); - } - catch (InterruptedException e) - { - e.printStackTrace(); - } - } - - /* - * Builds a graph of child associations as follows: - * 
-         * Level 0:     root
-         * Level 1:     root_p_n1   root_p_n2
-         * Level 2:     n1_p_n3     n2_p_n4     n1_n4       n2_p_n5     n1_n8
-         * Level 3:     n3_p_n6     n4_n6       n5_p_n7
-         * Level 4:     n6_p_n8     n7_n8
-         *
- */ - RetryingTransactionCallback testCallback = new RetryingTransactionCallback() - { - public Object execute() throws Exception - { - // There are two nodes at the base level in each test - assertEquals(2 * ((COUNT * REPEATS) + 1), nodeService.getChildAssocs(rootNodeRef).size()); - - SearchService searcher = (SearchService) ctx.getBean(ServiceRegistry.SEARCH_SERVICE.getLocalName()); - assertEquals( - 2 * ((COUNT * REPEATS) + 1), - searcher.selectNodes(rootNodeRef, "/*", null, getNamespacePrefixResolver(""), false).size()); - - return null; - } - }; - retryingTransactionHelper.doInTransaction(testCallback); - } - - /** - * Daemon thread - */ - private class Nester extends Thread - { - Thread waiter; - - int repeats; - - Nester(String name, Thread waiter, int repeats) - { - super(name); - this.setDaemon(true); - this.waiter = waiter; - this.repeats = repeats; - } - - public void run() - { - authenticationComponent.setSystemUserAsCurrentUser(); - - if (waiter != null) - { - System.out.println("Starting " + waiter.getName()); - waiter.start(); - } - try - { - System.out.println("Start " + this.getName()); - for (int i = 0; i < repeats; i++) - { - Map assocRefs = commitNodeGraph(); - System.out.println(" " + this.getName() + " " + i); - } - System.out.println("End " + this.getName()); - } - catch (Exception e) - { - e.printStackTrace(); - } - if (waiter != null) - { - try - { - waiter.join(); - System.out.println("Thread " - + this.getName() + " has waited for " + (waiter == null ? "null" : waiter.getName())); - } - catch (InterruptedException e) - { - System.err.println(e); - } - } - } - } /** * Tests that when multiple threads try to edit different @@ -316,7 +127,7 @@ public class ConcurrentNodeServiceTest extends TestCase * * @since 3.4 */ - public void testMultiThreadedNodePropertiesWrites() throws Exception + public void testMultiThreaded_PropertyWrites() throws Exception { final List threads = new ArrayList(); final int loops = 200; @@ -328,11 +139,13 @@ public class ConcurrentNodeServiceTest extends TestCase QName.createQName("test2", "MadeUp2"), QName.createQName("test3", "MadeUp3"), QName.createQName("test4", "MadeUp4"), - QName.createQName("test5", "MadeUp5") + QName.createQName("test4", "MadeUp5"), }; - for(QName prop : properties) + final int[] propCounts = new int[properties.length]; + for (int propNum = 0; propNum < properties.length; propNum++) { - final QName property = prop; + final QName property = properties[propNum]; + final int propNumberFinal = propNum; // Zap the property if it is there transactionService.getRetryingTransactionHelper().doInTransaction( @@ -364,7 +177,7 @@ public class ConcurrentNodeServiceTest extends TestCase // Loop, incrementing each time // If we miss an update, then at the end it'll be obvious - AuthenticationUtil.setFullyAuthenticatedUser(AuthenticationUtil.getSystemUserName()); + AuthenticationUtil.setRunAsUserSystem(); for (int i = 0; i < loops; i++) { RetryingTransactionCallback callback = new RetryingTransactionCallback() @@ -381,14 +194,21 @@ public class ConcurrentNodeServiceTest extends TestCase } // Increment by one. Really should be this! current++; - // Save the new value nodeService.setProperty(rootNodeRef, property, Integer.valueOf(current)); return current; } }; - RetryingTransactionHelper txnHelper = transactionService.getRetryingTransactionHelper(); - txnHelper.setMaxRetries(loops); - txnHelper.doInTransaction(callback, false, true); + try + { + RetryingTransactionHelper txnHelper = transactionService.getRetryingTransactionHelper(); + txnHelper.setMaxRetries(loops); + Integer newCount = txnHelper.doInTransaction(callback, false, true); + System.out.println("Set value: " + Thread.currentThread().getName() + " " + newCount); + } + catch (Throwable e) + { + logger.error("Failed to set value: ", e); + } } // Report us as finished @@ -411,63 +231,102 @@ public class ConcurrentNodeServiceTest extends TestCase { t.join(); } + + // Force a clear to see if we are seeing old values + SimpleCache nodesCache = (SimpleCache) ctx.getBean("node.nodesSharedCache"); + nodesCache.clear(); + SimpleCache propsCache = (SimpleCache) ctx.getBean("node.propertiesSharedCache"); + propsCache.clear(); - // Check each property in turn - RetryingTransactionCallback checkCallback = new RetryingTransactionCallback() + Map nodeProperties = nodeService.getProperties(rootNodeRef); + List errors = new ArrayList(); + for (int i =0; i < properties.length; i++) { - @Override - public Void execute() throws Throwable + Integer value = (Integer) nodeProperties.get(properties[i]); + if (value == null) { - HashMap values = new HashMap(); - for(QName prop : properties) - { - Object val = nodeService.getProperty(rootNodeRef, prop); - Integer value = -1; - if(val instanceof MLText) - { - value = Integer.valueOf( ((MLText)val).getValues().iterator().next() ); - } - else - { - value = (Integer)val; - } - - values.put(prop,value); - } - - List errors = new ArrayList(); - for(QName prop : properties) - { - Integer value = values.get(prop); - if (value == null || !value.equals(new Integer(loops))) - { - errors.add("\n Prop " + prop + " : " + value); - } - if (errors.size() > 0) - { - StringBuilder sb = new StringBuilder(); - sb.append("Incorrect counts recieved for " + loops + " loops."); - for (String error : errors) - { - sb.append(error); - } - fail(sb.toString()); - } - } - return null; + errors.add("\n Prop " + properties[i] + " : " + value); } - }; - transactionService.getRetryingTransactionHelper().doInTransaction(checkCallback, true); + if (!value.equals(new Integer(loops))) + { + // Check against the number of loops actually executed + if (propCounts[i] == value.intValue()) + { + System.out.println("Actually, not all counts done."); + } + errors.add("\n Prop " + properties[i] + " : " + value); + } + } + if (errors.size() > 0) + { + StringBuilder sb = new StringBuilder(); + sb.append("Incorrect counts recieved for " + loops + " loops."); + for (String error : errors) + { + sb.append(error); + } + fail(sb.toString()); + } } - - private NamespacePrefixResolver getNamespacePrefixResolver(String defaultURI) + + /** + * Adds 'residual' aspects that are named according to the thread. Multiple threads should all + * get their changes in. + */ + public void testMultithreaded_AspectWrites() throws Exception { - DynamicNamespacePrefixResolver nspr = new DynamicNamespacePrefixResolver(null); - nspr.registerNamespace(NamespaceService.SYSTEM_MODEL_PREFIX, NamespaceService.SYSTEM_MODEL_1_0_URI); - nspr.registerNamespace(NamespaceService.CONTENT_MODEL_PREFIX, NamespaceService.CONTENT_MODEL_1_0_URI); - nspr.registerNamespace(NamespaceService.APP_MODEL_PREFIX, NamespaceService.APP_MODEL_1_0_URI); - nspr.registerNamespace("namespace", "namespace"); - nspr.registerNamespace(NamespaceService.DEFAULT_PREFIX, defaultURI); - return nspr; + final Thread[] threads = new Thread[2]; + final int loops = 10; + + for (int i = 0; i < threads.length; i++) + { + final String name = "Thread-" + i + "-"; + Runnable runnable = new Runnable() + { + @Override + public void run() + { + AuthenticationUtil.setRunAsUserSystem(); + for (int loop = 0; loop < loops; loop++) + { + final String nameWithLoop = name + loop; + RetryingTransactionCallback runCallback = new RetryingTransactionCallback() + { + @Override + public Void execute() throws Throwable + { + // Add another aspect to the node + QName qname = QName.createQName(NAMESPACE, nameWithLoop); + nodeService.addAspect(rootNodeRef, qname, null); + return null; + } + }; + transactionService.getRetryingTransactionHelper().doInTransaction(runCallback); + } + } + }; + threads[i] = new Thread(runnable, name); + } + // Start all the threads + for (int i = 0; i < threads.length; i++) + { + threads[i].start(); + } + // Wait for them all to finish + for (int i = 0; i < threads.length; i++) + { + threads[i].join(); + } + // Check the aspects + Set aspects = nodeService.getAspects(rootNodeRef); + for (int i = 0; i < threads.length; i++) + { + for (int j = 0; j < loops; j++) + { + String nameWithLoop = "Thread-" + i + "-" + j; + QName qname = QName.createQName(NAMESPACE, nameWithLoop); + assertTrue("Missing aspect: "+ nameWithLoop, aspects.contains(qname)); + } + } } }