From e053932c5cb380de6aa2ddc7adf649a4c5f02ff3 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 30 May 2012 17:40:10 -0400 Subject: [PATCH] INT-2592 Fix Memory Leak in SimpleMessageStore Locks used to control access to a group of messages were never removed from the collection. Converted to use LockRegistry, which (by default) uses a pool of reentrant locks, using the hashcode of the group id as an index into the pool. Given that the pool is fixed, there is nothing to remove, thus avoiding the memory leak. When used within AbstractCorrelatingMessageHandler, any custom LockRegistry supplied will also be used by the message store, thus allowing the user to increase or decrease the size of the lock pool. The registry can not be changed once the SimpleMessageGroup has been used. Also, the ACMH had some protection to avoid setting the LockRegistry more than once, but this did not protect against the default lock registry being replaced after it had been used. Added additional protection to avoid this condition by setting lockRegistrySet to true in onInit(). INT-2592 Polishing PR Comments. Now, if a custom lock registry is required, it must be supplied to both the ACMH and SMS. INT-2592 polishing, fixed Javadoc in SMS --- .../AbstractCorrelatingMessageHandler.java | 41 ++-- .../integration/store/SimpleMessageStore.java | 176 ++++++++++++------ 2 files changed, 144 insertions(+), 73 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 8d2aaf671b..7746ef21b3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -1,11 +1,11 @@ /* * Copyright 2002-2012 the original author or authors. - * + * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at - * + * * http://www.apache.org/licenses/LICENSE-2.0 - * + * * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the * specific language governing permissions and limitations under the License. @@ -44,10 +44,10 @@ import org.springframework.util.CollectionUtils; /** * Abstract Message handler that holds a buffer of correlated messages in a * {@link MessageStore}. This class takes care of correlated groups of messages - * that can be completed in batches. It is useful for custom implementation of MessageHandlers that require correlation + * that can be completed in batches. It is useful for custom implementation of MessageHandlers that require correlation * and is used as a base class for Aggregator - {@link AggregatingMessageHandler} and * Resequencer - {@link ResequencingMessageHandler}, - * or custom implementations requiring correlation. + * or custom implementations requiring correlation. *

* To customize this handler inject {@link CorrelationStrategy}, * {@link ReleaseStrategy}, and {@link MessageGroupProcessor} implementations as @@ -60,6 +60,7 @@ import org.springframework.util.CollectionUtils; * @author Iwein Fuld * @author Dave Syer * @author Oleg Zhurakousky + * @author Gary Russell * @since 2.0 */ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer { @@ -83,7 +84,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH private volatile MessageChannel discardChannel = new NullChannel(); private boolean sendPartialResultOnExpiry = false; - + private volatile boolean sequenceAware = false; private volatile LockRegistry lockRegistry = new DefaultLockRegistry(); @@ -151,6 +152,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH if (beanFactory != null) { this.messagingTemplate.setBeanFactory(beanFactory); } + /* + * Disallow any further changes to the lock registry + * (checked in the setter). + */ + this.lockRegistrySet = true; } public void setDiscardChannel(MessageChannel discardChannel) { @@ -176,7 +182,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH public String getComponentType() { return "aggregator"; } - + protected MessageGroupStore getMessageStore() { return messageStore; } @@ -205,7 +211,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH logger.trace("Adding message to group [ " + messageGroup + "]"); } messageGroup = this.store(correlationKey, message); - + if (releaseStrategy.canRelease(messageGroup)) { Collection> completedMessages = null; try { @@ -213,11 +219,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH } finally { // Always clean up even if there was an exception - // processing messages + // processing messages this.afterRelease(messageGroup, completedMessages); } - } - } + } + } else { discardChannel.send(message); } @@ -228,7 +234,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH } /** - * Allows you to provide additional logic that needs to be performed after the MessageGroup was released. + * Allows you to provide additional logic that needs to be performed after the MessageGroup was released. * @param group * @param completedMessages */ @@ -273,11 +279,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH } protected int findLastReleasedSequenceNumber(Object groupId, Collection> partialSequence){ - List> sorted = new ArrayList>((Collection>)partialSequence); + List> sorted = new ArrayList>(partialSequence); Collections.sort(sorted, new SequenceNumberComparator()); - + Message lastReleasedMessage = sorted.get(partialSequence.size()-1); - + return lastReleasedMessage.getHeaders().getSequenceNumber(); } @@ -319,7 +325,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH if (logger.isDebugEnabled()) { logger.debug("Completing group with correlationKey [" + correlationKey + "]"); } - + Object result = outputProcessor.processMessageGroup(group); Collection> partialSequence = null; if (result instanceof Collection) { @@ -329,7 +335,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH this.sendReplies(result, message); return partialSequence; } - + private void verifyResultCollectionConsistsOfMessages(Collection elements){ Class commonElementType = CollectionUtils.findCommonElementType(elements); Assert.isAssignable(Message.class, commonElementType, "The expected collection of Messages contains non-Message element: " + commonElementType); @@ -393,6 +399,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH * its sequence id. This can be helpful to avoid ending up with sequences larger than their required sequence size * or sequences that are missing certain sequence numbers. */ + @Override public boolean canAdd(Message message) { if (this.size() == 0) { return true; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index 3588908c25..a098e1646f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -1,11 +1,11 @@ /* - * Copyright 2002-2011 the original author or authors. - * + * Copyright 2002-2012 the original author or authors. + * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at - * + * * http://www.apache.org/licenses/LICENSE-2.0 - * + * * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the * specific language governing permissions and limitations under the License. @@ -19,9 +19,12 @@ import java.util.Iterator; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.locks.Lock; import org.springframework.integration.Message; import org.springframework.integration.MessagingException; +import org.springframework.integration.util.DefaultLockRegistry; +import org.springframework.integration.util.LockRegistry; import org.springframework.integration.util.UpperBound; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedResource; @@ -31,27 +34,30 @@ import org.springframework.util.CollectionUtils; /** * Map-based in-memory implementation of {@link MessageStore} and {@link MessageGroupStore}. Enforces a maximum capacity for the * store. - * + * * @author Iwein Fuld * @author Mark Fisher * @author Dave Syer * @author Oleg Zhurakousky - * + * @author Gary Russell + * * @since 2.0 */ @ManagedResource public class SimpleMessageStore extends AbstractMessageGroupStore implements MessageStore, MessageGroupStore { - - private final ConcurrentMap locks = new ConcurrentHashMap(); + + private volatile LockRegistry lockRegistry; private final ConcurrentMap> idToMessage; private final ConcurrentMap groupIdToMessageGroup; - + private final UpperBound individualUpperBound; private final UpperBound groupUpperBound; - + + private volatile boolean isUsed; + /** * Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given * capacity is less than 1. The capacities are applied independently to messages stored via @@ -60,10 +66,21 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes * store another will result in an exception. */ public SimpleMessageStore(int individualCapacity, int groupCapacity) { + this(individualCapacity, groupCapacity, new DefaultLockRegistry()); + } + + /** + * See {@link #SimpleMessageStore(int, int)}. + * Also allows the provision of a custom {@link LockRegistry} + * rather than using the default. + */ + public SimpleMessageStore(int individualCapacity, int groupCapacity, LockRegistry lockRegistry) { + Assert.notNull(lockRegistry, "The LockRegistry cannot be null"); this.idToMessage = new ConcurrentHashMap>(); this.groupIdToMessageGroup = new ConcurrentHashMap(); this.individualUpperBound = new UpperBound(individualCapacity); this.groupUpperBound = new UpperBound(groupCapacity); + this.lockRegistry = lockRegistry; } /** @@ -80,12 +97,19 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes this(0); } + public void setLockRegistry(LockRegistry lockRegistry) { + Assert.notNull(lockRegistry, "The LockRegistry cannot be null"); + Assert.isTrue(!(this.isUsed), "Cannot change the lock registry after the store has been used"); + this.lockRegistry = lockRegistry; + } + @ManagedAttribute public long getMessageCount() { return idToMessage.size(); } public Message addMessage(Message message) { + this.isUsed = true; if (!individualUpperBound.tryAcquire(0)) { throw new MessagingException(this.getClass().getSimpleName() + " was out of capacity at, try constructing it with a larger capacity."); @@ -122,62 +146,112 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes throw new MessagingException(this.getClass().getSimpleName() + " was out of capacity at, try constructing it with a larger capacity."); } - Object lock = this.obtainLock(groupId); - synchronized (lock) { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); - if (group == null) { - group = new SimpleMessageGroup(groupId); - this.groupIdToMessageGroup.putIfAbsent(groupId, group); + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + if (group == null) { + group = new SimpleMessageGroup(groupId); + this.groupIdToMessageGroup.putIfAbsent(groupId, group); + } + group.add(message); + return group; } - group.add(message); - return group; + finally { + lock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); } } public void removeMessageGroup(Object groupId) { - Object lock = this.obtainLock(groupId); - synchronized (lock) { - if (!groupIdToMessageGroup.containsKey(groupId)) { - return; + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + if (!groupIdToMessageGroup.containsKey(groupId)) { + return; + } + + groupUpperBound.release(groupIdToMessageGroup.get(groupId).size()); + groupIdToMessageGroup.remove(groupId); } - - groupUpperBound.release(groupIdToMessageGroup.get(groupId).size()); - groupIdToMessageGroup.remove(groupId); + finally { + lock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); } } public MessageGroup removeMessageFromGroup(Object groupId, Message messageToRemove) { - Object lock = this.obtainLock(groupId); - synchronized (lock) { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + - "can not be located while attempting to remove Message from the MessageGroup"); - group.remove(messageToRemove); - return group; + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + + "can not be located while attempting to remove Message from the MessageGroup"); + group.remove(messageToRemove); + return group; + } + finally { + lock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); } } public Iterator iterator() { return new HashSet(groupIdToMessageGroup.values()).iterator(); } - + public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { - Object lock = this.obtainLock(groupId); - synchronized (lock) { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + - "can not be located while attempting to set 'lastReleasedSequenceNumber'"); - group.setLastReleasedMessageSequenceNumber(sequenceNumber); + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + + "can not be located while attempting to set 'lastReleasedSequenceNumber'"); + group.setLastReleasedMessageSequenceNumber(sequenceNumber); + } + finally { + lock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); } } public void completeGroup(Object groupId) { - Object lock = this.obtainLock(groupId); - synchronized (lock) { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + - "can not be located while attempting to complete the MessageGroup"); - group.complete(); + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + + "can not be located while attempting to complete the MessageGroup"); + group.complete(); + } + finally { + lock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); } } @@ -189,18 +263,8 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes if (message != null){ this.removeMessageFromGroup(groupId, message); } - } - return message; - } - - private Object obtainLock(Object groupId){ - Object lock = this.locks.get(groupId); - if (lock == null){ - Object newLock = new Object(); - Object previousLock = this.locks.putIfAbsent(groupId, newLock); - lock = (previousLock == null) ? newLock : previousLock; } - return lock; + return message; } public int messageGroupSize(Object groupId) {