Repository: activemq Updated Branches: refs/heads/trunk 838bbebee -> 140ce1bc8
https://issues.apache.org/jira/browse/AMQ-4485 - reenable test. concurrentStoreandDispatch case cannot be reconciled via setBatch, best we can do is trap duplicates from inflight messages as they occur. tagged async stores to support this. the revert of serialization with tx and non aysnc send brings perf back to where it was. https://issues.apache.org/jira/browse/AMQ-5266 Project: http://git-wip-us.apache.org/repos/asf/activemq/repo Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/140ce1bc Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/140ce1bc Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/140ce1bc Branch: refs/heads/trunk Commit: 140ce1bc8f447552c6a6dbfc1d3284da96386d2b Parents: c1c82be Author: gtully <gary.tu...@gmail.com> Authored: Tue Oct 7 14:49:58 2014 +0100 Committer: gtully <gary.tu...@gmail.com> Committed: Tue Oct 7 14:50:41 2014 +0100 ---------------------------------------------------------------------- .../region/cursors/AbstractStoreCursor.java | 45 +++----- .../activemq/store/kahadb/KahaDBStore.java | 114 +++++++------------ .../activemq/store/kahadb/MessageDatabase.java | 46 ++------ activemq-unit-tests/pom.xml | 1 - .../activemq/bugs/AMQ4485LowLimitTest.java | 16 ++- .../TwoBrokerQueueClientsReconnectTest.java | 4 + 6 files changed, 85 insertions(+), 141 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/activemq/blob/140ce1bc/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java ---------------------------------------------------------------------- diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java index fad666c..bef6017 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java +++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java @@ -18,7 +18,6 @@ package org.apache.activemq.broker.region.cursors; import java.util.Iterator; import java.util.LinkedList; -import java.util.concurrent.CancellationException; import java.util.concurrent.Future; import org.apache.activemq.broker.region.Destination; import org.apache.activemq.broker.region.MessageReference; @@ -100,13 +99,19 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor i clearIterator(true); recovered = true; storeHasMessages = true; + } else if (!cached) { + // a duplicate from the store (!cached) - needs to be removed/acked - otherwise it will get re dispatched on restart + if (message.isRecievedByDFBridge()) { + // expected for messages pending acks with kahadb.concurrentStoreAndDispatchQueues=true + LOG.trace("{} store replayed pending message due to concurrentStoreAndDispatchQueues {} seq: {}", this, message.getMessageId(), message.getMessageId().getFutureOrSequenceLong()); + } else { + LOG.warn("{} - cursor got duplicate from store {} seq: {}", this, message.getMessageId(), message.getMessageId().getFutureOrSequenceLong()); + duplicate(message); + } } else { - LOG.warn("{} - cursor got duplicate {} seq: {}", this, message.getMessageId(), message.getMessageId().getFutureOrSequenceLong()); - - // a duplicate from the store - needs to be removed/acked - otherwise it will get redispatched on restart - // jdbc store will store duplicates and will set entry locator to sequence long. - // REVISIT - this seems too hacky - see use case AMQ4952Test - if (!cached || message.getMessageId().getEntryLocator() instanceof Long) { + LOG.warn("{} - cursor got duplicate send {} seq: {}", this, message.getMessageId(), message.getMessageId().getFutureOrSequenceLong()); + if (message.getMessageId().getEntryLocator() instanceof Long) { + // JDBC will store a duplicate (with new sequence id) - it needs an ack (AMQ4952Test) duplicate(message); } } @@ -219,10 +224,11 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor i if (!pendingCachedIds.isEmpty() || lastCachedId != null) { LOG.trace("{} - disabling cache. current Id: {} seq: {}, batchList size: {}", new Object[]{this, node.getMessageId(), node.getMessageId().getFutureOrSequenceLong(), batchList.size()}); - collapseLastCachedIds(); + pruneLastCached(); if (lastCachedId != null) { setBatch(lastCachedId); lastCachedId = null; + pendingCachedIds.clear(); } } } @@ -240,6 +246,8 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor i Future future = (Future) futureOrLong; if (future.isCancelled()) { it.remove(); + } else { + break; } } else { // store complete - track via lastCachedId @@ -249,25 +257,8 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor i } } - private void collapseLastCachedIds() throws Exception { - for (MessageId candidate : pendingCachedIds) { - final Object futureOrLong = candidate.getFutureOrSequenceLong(); - if (futureOrLong instanceof Future) { - Future future = (Future) futureOrLong; - try { - future.get(); - // future should be replaced with sequence by this time - } catch (CancellationException ignored) { - continue; - } - } - setLastCachedId(candidate); - } - pendingCachedIds.clear(); - } - private void setLastCachedId(MessageId candidate) { - if (lastCachedId == null) { + if (lastCachedId == null || lastCachedId.getFutureOrSequenceLong() == null) { // possibly null for topics lastCachedId = candidate; } else if (Long.compare(((Long) candidate.getFutureOrSequenceLong()), ((Long) lastCachedId.getFutureOrSequenceLong())) > 0) { lastCachedId = candidate; @@ -360,7 +351,7 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor i public String toString() { return super.toString() + ":" + regionDestination.getActiveMQDestination().getPhysicalName() + ",batchResetNeeded=" + batchResetNeeded + ",storeHasMessages=" + this.storeHasMessages + ",size=" + this.size + ",cacheEnabled=" + isCacheEnabled() - + ",maxBatchSize:" + maxBatchSize + ",hasSpace:" + hasSpace() + ",pendingCachedIds.size:" + pendingCachedIds.size() + ",lastCachedId:" + lastCachedId; + + ",maxBatchSize:" + maxBatchSize + ",hasSpace:" + hasSpace() + ",pendingCachedIds.size:" + pendingCachedIds.size() + ",lastCachedId:" + lastCachedId + ",lastCachedId-seq:" + (lastCachedId != null ? lastCachedId.getFutureOrSequenceLong() : "null"); } protected abstract void doFillBatch() throws Exception; http://git-wip-us.apache.org/repos/asf/activemq/blob/140ce1bc/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java ---------------------------------------------------------------------- diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java index 54cfd7d..abdf4bf 100644 --- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java +++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java @@ -30,7 +30,6 @@ import java.util.Map.Entry; import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Callable; -import java.util.concurrent.CancellationException; import java.util.concurrent.ExecutorService; import java.util.concurrent.FutureTask; import java.util.concurrent.LinkedBlockingQueue; @@ -38,12 +37,11 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.apache.activemq.broker.ConnectionContext; -import org.apache.activemq.broker.region.Destination; -import org.apache.activemq.broker.region.RegionBroker; import org.apache.activemq.broker.scheduler.JobSchedulerStore; import org.apache.activemq.command.ActiveMQDestination; import org.apache.activemq.command.ActiveMQQueue; @@ -86,7 +84,7 @@ import org.apache.activemq.wireformat.WireFormat; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, MessageDatabase.SerialExecution<Location> { +public class KahaDBStore extends MessageDatabase implements PersistenceAdapter { static final Logger LOG = LoggerFactory.getLogger(KahaDBStore.class); private static final int MAX_ASYNC_JOBS = 10000; @@ -124,7 +122,6 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, return txid; } }; - serialExecutor = this; } @Override @@ -347,17 +344,6 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, this.forceRecoverIndex = forceRecoverIndex; } - @Override - public Location execute(Callable<Location> c) throws Exception { - if (isConcurrentStoreAndDispatchQueues()) { - FutureTask<Location> future = new FutureTask<>(c); - this.queueExecutor.execute(future); - return future.get(); - } else { - return c.call(); - } - } - public class KahaDBMessageStore extends AbstractMessageStore { protected final Map<AsyncJobKey, StoreTask> asyncTaskMap = new HashMap<AsyncJobKey, StoreTask>(); protected KahaDestination dest; @@ -383,25 +369,14 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, throws IOException { if (isConcurrentStoreAndDispatchQueues()) { StoreQueueTask result = new StoreQueueTask(this, context, message); + ListenableFuture<Object> future = result.getFuture(); + message.getMessageId().setFutureOrSequenceLong(future); + message.setRecievedByDFBridge(true); // flag message as concurrentStoreAndDispatch result.aquireLocks(); addQueueTask(this, result); - final ListenableFuture<Object> future = result.getFuture(); if (indexListener != null) { // allow concurrent dispatch by setting entry locator, - // wait for add completion to remove potential pending addition - message.getMessageId().setFutureOrSequenceLong(future); - indexListener.onAdd(new IndexListener.MessageContext(context, message, new Runnable() { - @Override - public void run() { - try { - future.get(); - trackPendingAddComplete(dest, (Long) message.getMessageId().getFutureOrSequenceLong()); - } catch (CancellationException okNothingToTrack) { - } catch (Exception e) { - LOG.warn("{} unexpected exception tracking completion of async add of {}", this, message.getMessageId(), e); - } - } - })); + indexListener.onAdd(new IndexListener.MessageContext(context, message, null)); } return future; } else { @@ -442,7 +417,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, @Override public void addMessage(final ConnectionContext context, final Message message) throws IOException { - KahaAddMessageCommand command = new KahaAddMessageCommand(); + final KahaAddMessageCommand command = new KahaAddMessageCommand(); command.setDestination(dest); command.setMessageId(message.getMessageId().toProducerKey()); command.setTransactionInfo(TransactionIdConversion.convert(transactionIdTransformer.transform(message.getTransactionId()))); @@ -450,25 +425,25 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, command.setPrioritySupported(isPrioritizedMessages()); org.apache.activemq.util.ByteSequence packet = wireFormat.marshal(message); command.setMessage(new Buffer(packet.getData(), packet.getOffset(), packet.getLength())); - store(command, isEnableJournalDiskSyncs() && message.isResponseRequired(), new IndexAware() { - @Override - public void sequenceAssignedWithIndexLocked(final long sequence) { - final Object possibleFuture = message.getMessageId().getFutureOrSequenceLong(); - message.getMessageId().setFutureOrSequenceLong(sequence); - if (indexListener != null) { - trackPendingAdd(dest, sequence); - if (possibleFuture == null) { - // sync add (for async future present from getFutureOrSequenceLong) - indexListener.onAdd(new IndexListener.MessageContext(context, message, new Runnable() { - @Override - public void run() { - trackPendingAddComplete(dest, sequence); - } - })); + store(command, isEnableJournalDiskSyncs() && message.isResponseRequired(), new IndexAware() { + // sync add? (for async, future present from getFutureOrSequenceLong) + Object possibleFuture = message.getMessageId().getFutureOrSequenceLong(); + + public void sequenceAssignedWithIndexLocked(final long sequence) { + message.getMessageId().setFutureOrSequenceLong(sequence); + if (indexListener != null) { + if (possibleFuture == null) { + trackPendingAdd(dest, sequence); + indexListener.onAdd(new IndexListener.MessageContext(context, message, new Runnable() { + @Override + public void run() { + trackPendingAddComplete(dest, sequence); + } + })); + } } } - } - }, null); + }, null); } @Override @@ -680,32 +655,23 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, @Override public void setBatch(final MessageId identity) throws IOException { + final String key = identity.toProducerKey(); + indexLock.writeLock().lock(); try { - final String key = identity.toProducerKey(); - lockAsyncJobQueue(); - - // Hopefully one day the page file supports concurrent read - // operations... but for now we must externally synchronize... - - indexLock.writeLock().lock(); - try { - pageFile.tx().execute(new Transaction.Closure<IOException>() { - @Override - public void execute(Transaction tx) throws IOException { - StoredDestination sd = getStoredDestination(dest, tx); - Long location = sd.messageIdIndex.get(tx, key); - if (location != null) { - sd.orderIndex.setBatch(tx, location); - } else { - LOG.warn("{} Location {} not found in order index for {}", this, identity.getFutureOrSequenceLong(), identity); - } + pageFile.tx().execute(new Transaction.Closure<IOException>() { + @Override + public void execute(Transaction tx) throws IOException { + StoredDestination sd = getStoredDestination(dest, tx); + Long location = sd.messageIdIndex.get(tx, key); + if (location != null) { + sd.orderIndex.setBatch(tx, location); + } else { + LOG.warn("{} {} setBatch failed, location for {} not found in messageId index for {}", this, dest.getName(), identity.getFutureOrSequenceLong(), identity); } - }); - } finally { - indexLock.writeLock().unlock(); - } + } + }); } finally { - unlockAsyncJobQueue(); + indexLock.writeLock().unlock(); } } @@ -723,7 +689,9 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, protected void lockAsyncJobQueue() { try { - this.localDestinationSemaphore.tryAcquire(this.maxAsyncJobs, 60, TimeUnit.SECONDS); + if (!this.localDestinationSemaphore.tryAcquire(this.maxAsyncJobs, 60, TimeUnit.SECONDS)) { + throw new TimeoutException(this +" timeout waiting for localDestSem:" + this.localDestinationSemaphore); + } } catch (Exception e) { LOG.error("Failed to lock async jobs for " + this.destination, e); } http://git-wip-us.apache.org/repos/asf/activemq/blob/140ce1bc/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java ---------------------------------------------------------------------- diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java index c8bc2b8..554f1d3 100644 --- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java +++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java @@ -256,11 +256,6 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe private boolean enableIndexPageCaching = true; ReentrantReadWriteLock checkpointLock = new ReentrantReadWriteLock(); - interface SerialExecution<V> { - public V execute(Callable<V> c) throws Exception; - } - SerialExecution<Location> serialExecutor; - @Override public void doStart() throws Exception { load(); @@ -962,20 +957,6 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe return store(data, sync, before, after, null); } - public Location store(final KahaCommitCommand data, final boolean sync, final IndexAware before, final Runnable after) throws IOException { - try { - return serialExecutor.execute(new Callable<Location>() { - @Override - public Location call() throws Exception { - return store(data, sync, before, after, null); - } - }); - } catch (Exception e) { - LOG.error("Failed to execute commit", e); - throw new IOException(e); - } - } - /** * All updated are are funneled through this method. The updates are converted * to a JournalMessage which is logged to the journal and then the data from @@ -1259,24 +1240,19 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe } final List<Operation> messagingTx = inflightTx; + indexLock.writeLock().lock(); try { - indexLock.writeLock().lock(); - try { - pageFile.tx().execute(new Transaction.Closure<IOException>() { - @Override - public void execute(Transaction tx) throws IOException { - for (Operation op : messagingTx) { - op.execute(tx); - } + pageFile.tx().execute(new Transaction.Closure<IOException>() { + @Override + public void execute(Transaction tx) throws IOException { + for (Operation op : messagingTx) { + op.execute(tx); } - }); - metadata.lastUpdate = location; - } finally { - indexLock.writeLock().unlock(); - } - } catch (Exception e) { - LOG.error("serial execution of commit failed", e); - throw new IOException(e); + } + }); + metadata.lastUpdate = location; + } finally { + indexLock.writeLock().unlock(); } } http://git-wip-us.apache.org/repos/asf/activemq/blob/140ce1bc/activemq-unit-tests/pom.xml ---------------------------------------------------------------------- diff --git a/activemq-unit-tests/pom.xml b/activemq-unit-tests/pom.xml index 3a49bda..1333412 100755 --- a/activemq-unit-tests/pom.xml +++ b/activemq-unit-tests/pom.xml @@ -557,7 +557,6 @@ <artifactId>maven-surefire-plugin</artifactId> <configuration> <excludes combine.children="append"> - <exclude>**/AMQ4485LowLimitTest.*</exclude> <!-- http://jira.activemq.org/jira/browse/AMQ-626 --> <exclude>**/MultipleTestsWithSpringFactoryBeanTest.*</exclude> <exclude>**/MultipleTestsWithXBeanFactoryBeanTest.*</exclude> http://git-wip-us.apache.org/repos/asf/activemq/blob/140ce1bc/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4485LowLimitTest.java ---------------------------------------------------------------------- diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4485LowLimitTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4485LowLimitTest.java index 5af1124..eb46f8f 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4485LowLimitTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4485LowLimitTest.java @@ -242,7 +242,7 @@ public class AMQ4485LowLimitTest extends JmsMultipleBrokersTestSupport { if (tally.accumulator.get() != expected) { LOG.info("Tally for: " + tally.brokerName + ", dest: " + tally.destination + " - " + tally.accumulator.get() + " != " + expected + ", " + tally.expected); if (tally.accumulator.get() > expected - 50) { - dumpQueueStat(tally.destination); + dumpQueueStat(null); } if (tally.expected.size() == 1) { startConsumer(tally.brokerName, tally.destination); @@ -260,6 +260,9 @@ public class AMQ4485LowLimitTest extends JmsMultipleBrokersTestSupport { LOG.info("done"); long duration = System.currentTimeMillis() - startTime; LOG.info("Duration:" + TimeUtils.printDuration(duration)); + + assertEquals("nothing in the dlq's", 0, dumpQueueStat(new ActiveMQQueue("ActiveMQ.DLQ"))); + } private void startConsumer(String brokerName, ActiveMQDestination destination) throws Exception { @@ -273,17 +276,20 @@ public class AMQ4485LowLimitTest extends JmsMultipleBrokersTestSupport { queueConnection.close(); } - private void dumpQueueStat(ActiveMQDestination destination) throws Exception { + private long dumpQueueStat(ActiveMQDestination destination) throws Exception { + long sumTotal = 0; Collection<BrokerItem> brokerList = brokers.values(); for (Iterator<BrokerItem> i = brokerList.iterator(); i.hasNext(); ) { BrokerService brokerService = i.next().broker; for (ObjectName objectName : brokerService.getAdminView().getQueues()) { - //if (objectName.toString().contains(destination.getQualifiedName())) { + if (destination != null && objectName.toString().contains(destination.getPhysicalName())) { QueueViewMBean qViewMBean = (QueueViewMBean) brokerService.getManagementContext().newProxyInstance(objectName, QueueViewMBean.class, false); - LOG.info(brokerService.getBrokerName() + ", " + qViewMBean.getName() + " Size: " + qViewMBean.getEnqueueCount()); - //} + LOG.info(brokerService.getBrokerName() + ", " + qViewMBean.getName() + ", Enqueue:" + qViewMBean.getEnqueueCount() + ", Size: " + qViewMBean.getQueueSize()); + sumTotal += qViewMBean.getQueueSize(); + } } } + return sumTotal; } private void startAllGWFanoutConsumers(int nBrokers) throws Exception { http://git-wip-us.apache.org/repos/asf/activemq/blob/140ce1bc/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/TwoBrokerQueueClientsReconnectTest.java ---------------------------------------------------------------------- diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/TwoBrokerQueueClientsReconnectTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/TwoBrokerQueueClientsReconnectTest.java index bfc797f..9adc2a3 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/TwoBrokerQueueClientsReconnectTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/TwoBrokerQueueClientsReconnectTest.java @@ -42,6 +42,7 @@ import org.apache.activemq.broker.region.policy.PolicyMap; import org.apache.activemq.command.ActiveMQDestination; import org.apache.activemq.network.ConditionalNetworkBridgeFilterFactory; import org.apache.activemq.network.NetworkConnector; +import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter; import org.apache.activemq.util.Wait; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -456,6 +457,9 @@ public class TwoBrokerQueueClientsReconnectTest extends JmsMultipleBrokersTestSu BrokerService brokerService = brokers.get(broker2).broker; brokerService.setPersistent(true); brokerService.setDeleteAllMessagesOnStartup(true); + // disable concurrent dispatch otherwise store duplicate suppression will be skipped b/c cursor audit is already + // disabled so verification of stats will fail - ie: duplicate will be dispatched + ((KahaDBPersistenceAdapter)brokerService.getPersistenceAdapter()).setConcurrentStoreAndDispatchQueues(false); brokerService.setPlugins(new BrokerPlugin[]{ new BrokerPluginSupport() { @Override
