Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-3149, https://issues.apache.org/jira/browse/AMQ-3145 

expose current caching state of cursors (also via jmx), only use store async add if cache
is enabled, ensure cache is reenabled on add with 0 queue depth

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1061859 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Tully 2011-01-21 15:53:29 +00:00
parent 4e33fb114b
commit aad7e7394a
15 changed files with 211 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
activemq-core/src/main/java/org/apache/activemq/broker/jmx/DestinationViewMBean.java
View File

@ -307,9 +307,9 @@ public interface DestinationViewMBean {
public void setMaxPageSize(@MBeanInfo("pageSize") int pageSize);
/**
* @return true if caching is enabled of for the destination
* @return true if caching is allowed of for the destination
*/
@MBeanInfo("Caching is enabled")
@MBeanInfo("Caching is allowed")
public boolean isUseCache();
/**

8
activemq-core/src/main/java/org/apache/activemq/broker/jmx/QueueView.java
View File

@ -168,4 +168,12 @@ public class QueueView extends DestinationView implements QueueViewMBean {
}
return false;
}
public boolean isCacheEnabled() {
Queue queue = (Queue) destination;
if (queue.getMessages() != null){
return queue.getMessages().isCacheEnabled();
}
return false;
}
}

5
activemq-core/src/main/java/org/apache/activemq/broker/jmx/QueueViewMBean.java
View File

@ -169,4 +169,9 @@ public interface QueueViewMBean extends DestinationViewMBean {
@MBeanInfo("Number of messages available to be paged in by the cursor.")
public int cursorSize();
/**
* @return true if caching is currently enabled of for the destination
*/
@MBeanInfo("Caching is enabled")
boolean isCacheEnabled();
}

6
activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
View File

@ -667,7 +667,11 @@ public class Queue extends BaseDestination implements Task, UsageListener {
try {
if (store != null && message.isPersistent()) {
message.getMessageId().setBrokerSequenceId(getDestinationSequenceId());
result = store.asyncAddQueueMessage(context, message);
if (messages.isCacheEnabled()) {
result = store.asyncAddQueueMessage(context, message);
} else {
store.addMessage(context, message);
}
if (isReduceMemoryFootprint()) {
message.clearMarshalledState();
}

5
activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/AbstractPendingMessageCursor.java
View File

@ -45,6 +45,7 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
protected boolean enableAudit=true;
protected ActiveMQMessageAudit audit;
protected boolean useCache=true;
protected boolean cacheEnabled=true;
private boolean started=false;
protected MessageReference last = null;
protected final boolean prioritizedMessages;
@ -327,4 +328,8 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
return result;
}
public boolean isCacheEnabled() {
return cacheEnabled;
}
}

17
activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java
View File

@ -34,7 +34,6 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor i
protected final Destination regionDestination;
private final PendingList batchList;
private Iterator<MessageReference> iterator = null;
protected boolean cacheEnabled=false;
protected boolean batchResetNeeded = true;
private boolean storeHasMessages = false;
protected int size;
@ -59,9 +58,7 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor i
resetBatch();
this.size = getStoreSize();
this.storeHasMessages=this.size > 0;
if (!this.storeHasMessages&&useCache) {
cacheEnabled=true;
}
cacheEnabled = !this.storeHasMessages&&useCache;
}
}
@ -171,6 +168,12 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor i
public final synchronized void addMessageLast(MessageReference node) throws Exception {
if (!cacheEnabled && size==0 && isStarted() && useCache && hasSpace()) {
if (LOG.isDebugEnabled()) {
LOG.debug(regionDestination.getActiveMQDestination().getPhysicalName() + " enabling cache on empty add");
}
cacheEnabled=true;
}
if (cacheEnabled && hasSpace()) {
recoverMessage(node.getMessage(),true);
lastCachedId = node.getMessageId();
@ -210,12 +213,6 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor i
if (last != null) {
last.decrementReferenceCount();
}
if (size==0 && isStarted() && useCache && hasSpace() && isStoreEmpty()) {
if (LOG.isDebugEnabled()) {
LOG.debug(regionDestination.getActiveMQDestination().getPhysicalName() + " enabling cache on last remove");
}
cacheEnabled=true;
}
}

7
activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
View File

@ -63,12 +63,10 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
/**
* @param broker
* @param name
* @param prioritizedMessages
* @param store
* @param prioritizedMessages
*/
public FilePendingMessageCursor(Broker broker, String name, boolean prioritizedMessages) {
super(prioritizedMessages);
this.useCache = false;
this.broker = broker;
// the store can be null if the BrokerService has persistence
// turned off
@ -201,6 +199,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
if (hasSpace() || this.store == null) {
memoryList.add(node);
node.incrementReferenceCount();
cacheEnabled = true;
return true;
}
}
@ -248,6 +247,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
if (hasSpace()) {
memoryList.addFirst(node);
node.incrementReferenceCount();
cacheEnabled = true;
return;
}
}
@ -428,6 +428,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
}
memoryList.clear();
cacheEnabled = false;
}
}

7
activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/PendingMessageCursor.java
View File

@ -285,7 +285,7 @@ public interface PendingMessageCursor extends Service {
public void setUseCache(boolean useCache);
/**
* @return true if a cache is being used
* @return true if a cache may be used
*/
public boolean isUseCache();
@ -294,5 +294,10 @@ public interface PendingMessageCursor extends Service {
* @param id
*/
public void rollback(MessageId id);
/**
* @return true if cache is being used
*/
public boolean isCacheEnabled();
}

16
activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreQueueCursor.java
View File

@ -180,7 +180,7 @@ public class StoreQueueCursor extends AbstractPendingMessageCursor {
* Informs the Broker if the subscription needs to intervention to recover
* it's state e.g. DurableTopicSubscriber may do
*
* @see org.apache.activemq.region.cursors.PendingMessageCursor
* @see org.apache.activemq.broker.region.cursors.PendingMessageCursor
* @return true if recovery required
*/
public boolean isRecoveryRequired() {
@ -294,4 +294,18 @@ public class StoreQueueCursor extends AbstractPendingMessageCursor {
}
return currentCursor;
}
@Override
public boolean isCacheEnabled() {
cacheEnabled = isUseCache();
if (cacheEnabled) {
if (persistent != null) {
cacheEnabled &= persistent.isCacheEnabled();
}
if (nonPersistent != null) {
cacheEnabled &= nonPersistent.isCacheEnabled();
}
}
return cacheEnabled;
}
}

1
activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/VMPendingMessageCursor.java
View File

@ -37,7 +37,6 @@ public class VMPendingMessageCursor extends AbstractPendingMessageCursor {
public VMPendingMessageCursor(boolean prioritizedMessages) {
super(prioritizedMessages);
this.useCache = false;
if (this.prioritizedMessages) {
this.list= new PrioritizedPendingList();
}else {

4
activemq-core/src/test/java/org/apache/activemq/ExpiryHogTest.java
View File

@ -33,7 +33,7 @@ public class ExpiryHogTest extends JmsMultipleClientsTestSupport {
int numMessages = 4;
public void testHog() throws Exception {
public void testImmediateDispatchWhenCacheDisabled() throws Exception {
ConnectionFactory f = createConnectionFactory();
destination = createDestination();
startConsumers(f, destination);
@ -53,7 +53,7 @@ public class ExpiryHogTest extends JmsMultipleClientsTestSupport {
bs.setDestinationPolicy(policyMap);
KahaDBPersistenceAdapter ad = (KahaDBPersistenceAdapter) bs.getPersistenceAdapter();
ad.setConcurrentStoreAndDispatchQueues(false);
ad.setConcurrentStoreAndDispatchQueues(true);
return bs;
}

2
activemq-core/src/test/java/org/apache/activemq/broker/jmx/MBeanTest.java
View File

@ -156,6 +156,8 @@ public class MBeanTest extends EmbeddedBrokerTestSupport {
// check memory usage migration
assertTrue("new dest has some memory usage", queueNew.getMemoryPercentUsage() > 0);
assertEquals("old dest has no memory usage", 0, queue.getMemoryPercentUsage());
assertTrue("use cache", queueNew.isUseCache());
assertTrue("cache enabled", queueNew.isCacheEnabled());
}
public void testRetryMessages() throws Exception {

1
activemq-core/src/test/java/org/apache/activemq/broker/region/QueuePurgeTest.java
View File

@ -96,6 +96,7 @@ public class QueuePurgeTest extends TestCase {
proxy.purge();
assertEquals("Queue size is not zero, it's " + proxy.getQueueSize(), 0,
proxy.getQueueSize());
assertTrue("cache is disabled, temp store being used", !proxy.isCacheEnabled());
}
public void testRepeatedExpiryProcessingOfLargeQueue() throws Exception {

4
activemq-core/src/test/java/org/apache/activemq/broker/region/cursors/StoreQueueCursorNoDuplicateTest.java
View File

@ -34,7 +34,7 @@ import org.apache.activemq.usage.SystemUsage;
/**
* @author gtully
* @see https://issues.apache.org/activemq/browse/AMQ-2020
* https://issues.apache.org/activemq/browse/AMQ-2020
**/
public class StoreQueueCursorNoDuplicateTest extends TestCase {
ActiveMQQueue destination = new ActiveMQQueue("queue-"
@ -83,6 +83,7 @@ public class StoreQueueCursorNoDuplicateTest extends TestCase {
underTest.setSystemUsage(systemUsage);
underTest.setEnableAudit(false);
underTest.start();
assertTrue("cache enabled", underTest.isUseCache() && underTest.isCacheEnabled());
final ConnectionContext contextNotInTx = new ConnectionContext();
for (int i = 0; i < count; i++) {
@ -93,6 +94,7 @@ public class StoreQueueCursorNoDuplicateTest extends TestCase {
underTest.addMessageLast(msg);
}
assertTrue("cache is disabled as limit reached", !underTest.isCacheEnabled());
int dequeueCount = 0;
underTest.setMaxBatchSize(2);

146
activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3145Test.java Normal file
View File

@ -0,0 +1,146 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.
*/
package org.apache.activemq.bugs;
import java.util.concurrent.TimeUnit;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.jmx.QueueViewMBean;
import org.apache.activemq.broker.region.policy.FilePendingQueueMessageStoragePolicy;
import org.apache.activemq.broker.region.policy.PendingQueueMessageStoragePolicy;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class AMQ3145Test {
private static final Log LOG = LogFactory.getLog(AMQ3145Test.class);
private final String MESSAGE_TEXT = new String(new byte[1024]);
BrokerService broker;
ConnectionFactory factory;
Connection connection;
Session session;
Queue queue;
MessageConsumer consumer;
@Before
public void createBroker() throws Exception {
createBroker(true);
}
public void createBroker(boolean deleteAll) throws Exception {
broker = new BrokerService();
broker.setDeleteAllMessagesOnStartup(deleteAll);
broker.setDataDirectory("target/AMQ3145Test");
broker.setUseJmx(true);
broker.addConnector("tcp://localhost:0");
broker.start();
factory = new ActiveMQConnectionFactory(broker.getTransportConnectors().get(0).getConnectUri().toString());
connection = factory.createConnection();
connection.start();
session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
}
@After
public void tearDown() throws Exception {
if (consumer != null) {
consumer.close();
}
session.close();
connection.stop();
connection.close();
broker.stop();
}
@Test
public void testCacheDisableReEnable() throws Exception {
createProducerAndSendMessages(1);
QueueViewMBean proxy = getProxyToQueueViewMBean();
assertTrue("cache is enabled", proxy.isCacheEnabled());
tearDown();
createBroker(false);
proxy = getProxyToQueueViewMBean();
assertEquals("one pending message", 1, proxy.getQueueSize());
assertTrue("cache is disabled when there is a pending message", !proxy.isCacheEnabled());
createConsumer(1);
createProducerAndSendMessages(1);
assertTrue("cache is enabled again on next send when there are no messages", proxy.isCacheEnabled());
}
private void applyBrokerSpoolingPolicy() {
PolicyMap policyMap = new PolicyMap();
PolicyEntry defaultEntry = new PolicyEntry();
defaultEntry.setProducerFlowControl(false);
PendingQueueMessageStoragePolicy pendingQueuePolicy = new FilePendingQueueMessageStoragePolicy();
defaultEntry.setPendingQueuePolicy(pendingQueuePolicy);
policyMap.setDefaultEntry(defaultEntry);
broker.setDestinationPolicy(policyMap);
}
private QueueViewMBean getProxyToQueueViewMBean()
throws MalformedObjectNameException, JMSException {
ObjectName queueViewMBeanName = new ObjectName("org.apache.activemq"
+ ":Type=Queue,Destination=" + queue.getQueueName()
+ ",BrokerName=localhost");
QueueViewMBean proxy = (QueueViewMBean) broker.getManagementContext()
.newProxyInstance(queueViewMBeanName,
QueueViewMBean.class, true);
return proxy;
}
private void createProducerAndSendMessages(int numToSend) throws Exception {
queue = session.createQueue("test1");
MessageProducer producer = session.createProducer(queue);
for (int i = 0; i < numToSend; i++) {
TextMessage message = session.createTextMessage(MESSAGE_TEXT + i);
if (i != 0 && i % 50000 == 0) {
LOG.info("sent: " + i);
}
producer.send(message);
}
producer.close();
}
private void createConsumer(int numToConsume) throws Exception {
consumer = session.createConsumer(queue);
// wait for buffer fill out
for (int i = 0; i < numToConsume; ++i) {
Message message = consumer.receive(2000);
message.acknowledge();
}
consumer.close();
}
}