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

Fix for https://issues.apache.org/activemq/browse/AMQ-1748 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@661295 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2008-05-29 11:27:33 +00:00
parent 4dde3239ff
commit 0f1ef01d1b
3 changed files with 165 additions and 282 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1062,7 +1062,12 @@ public class Queue extends BaseDestination implements Task {
} }
final void sendMessage(final ConnectionContext context, Message msg) throws Exception { final void sendMessage(final ConnectionContext context, Message msg) throws Exception {
messages.addMessageLast(msg); if (!msg.isPersistent() && messages.getSystemUsage() != null) {
messages.getSystemUsage().getTempUsage().waitForSpace();
}
synchronized(messages) {
messages.addMessageLast(msg);
}
destinationStatistics.getEnqueues().increment(); destinationStatistics.getEnqueues().increment();
destinationStatistics.getMessages().increment(); destinationStatistics.getMessages().increment();
messageDelivered(context, msg); messageDelivered(context, msg);

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

@ -21,7 +21,6 @@ import java.util.Iterator;
import java.util.LinkedList; import java.util.LinkedList;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.activemq.broker.Broker; import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.ConnectionContext; import org.apache.activemq.broker.ConnectionContext;
@ -60,7 +59,6 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
private boolean flushRequired; private boolean flushRequired;
private AtomicBoolean started = new AtomicBoolean(); private AtomicBoolean started = new AtomicBoolean();
private MessageReference last = null; private MessageReference last = null;
private ReentrantLock lock = new ReentrantLock(true);
/** /**
* @param name * @param name
@ -95,25 +93,20 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
/** /**
* @return true if there are no pending messages * @return true if there are no pending messages
*/ */
public boolean isEmpty() { public synchronized boolean isEmpty() {
lock.lock(); if(memoryList.isEmpty() && isDiskListEmpty()){
try { return true;
if(memoryList.isEmpty() && isDiskListEmpty()){ }
return true; for (Iterator<MessageReference> iterator = memoryList.iterator(); iterator.hasNext();) {
MessageReference node = iterator.next();
if (node== QueueMessageReference.NULL_MESSAGE){
continue;
} }
for (Iterator<MessageReference> iterator = memoryList.iterator(); iterator.hasNext();) { if (!node.isDropped()) {
MessageReference node = iterator.next(); return false;
if (node== QueueMessageReference.NULL_MESSAGE){
continue;
}
if (!node.isDropped()) {
return false;
}
// We can remove dropped references.
iterator.remove();
} }
} finally { // We can remove dropped references.
lock.unlock(); iterator.remove();
} }
return isDiskListEmpty(); return isDiskListEmpty();
} }
@ -123,71 +116,48 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
/** /**
* reset the cursor * reset the cursor
*/ */
public void reset() { public synchronized void reset() {
lock.lock(); iterating = true;
try { last = null;
iterating = true; iter = isDiskListEmpty() ? memoryList.iterator() : getDiskList().listIterator();
last = null; }
iter = isDiskListEmpty() ? memoryList.iterator() : getDiskList().listIterator();
} finally { public synchronized void release() {
lock.unlock(); iterating = false;
if (flushRequired) {
flushRequired = false;
flushToDisk();
} }
} }
public void release() { public synchronized void destroy() throws Exception {
lock.lock(); stop();
try { for (Iterator<MessageReference> i = memoryList.iterator(); i.hasNext();) {
synchronized(this) { Message node = (Message)i.next();
iterating = false; node.decrementReferenceCount();
this.notifyAll(); }
} memoryList.clear();
if (flushRequired) { if (!isDiskListEmpty()) {
flushRequired = false; getDiskList().clear();
flushToDisk();
}
} finally {
lock.unlock();
} }
} }
public void destroy() throws Exception { public synchronized LinkedList<MessageReference> pageInList(int maxItems) {
lock.lock();
try {
stop();
for (Iterator<MessageReference> i = memoryList.iterator(); i.hasNext();) {
Message node = (Message)i.next();
node.decrementReferenceCount();
}
memoryList.clear();
if (!isDiskListEmpty()) {
getDiskList().clear();
}
} finally {
lock.unlock();
}
}
public LinkedList<MessageReference> pageInList(int maxItems) {
int count = 0;
LinkedList<MessageReference> result = new LinkedList<MessageReference>(); LinkedList<MessageReference> result = new LinkedList<MessageReference>();
lock.lock(); int count = 0;
try { for (Iterator<MessageReference> i = memoryList.iterator(); i.hasNext() && count < maxItems;) {
for (Iterator<MessageReference> i = memoryList.iterator(); i.hasNext() && count < maxItems;) { result.add(i.next());
result.add(i.next()); count++;
}
if (count < maxItems && !isDiskListEmpty()) {
for (Iterator<MessageReference> i = getDiskList().iterator(); i.hasNext() && count < maxItems;) {
Message message = (Message)i.next();
message.setRegionDestination(regionDestination);
message.setMemoryUsage(this.getSystemUsage().getMemoryUsage());
message.incrementReferenceCount();
result.add(message);
count++; count++;
} }
if (count < maxItems && !isDiskListEmpty()) {
for (Iterator<MessageReference> i = getDiskList().iterator(); i.hasNext() && count < maxItems;) {
Message message = (Message)i.next();
message.setRegionDestination(regionDestination);
message.setMemoryUsage(this.getSystemUsage().getMemoryUsage());
message.incrementReferenceCount();
result.add(message);
count++;
}
}
} finally {
lock.unlock();
} }
return result; return result;
} }
@ -197,108 +167,31 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
* *
* @param node * @param node
*/ */
public void addMessageLast(MessageReference node) { public synchronized void addMessageLast(MessageReference node) {
if (!node.isExpired()) { if (!node.isExpired()) {
try { try {
lock.lock(); regionDestination = node.getMessage().getRegionDestination();
try { if (isDiskListEmpty()) {
while (iterating) { if (hasSpace() || this.store==null) {
lock.unlock(); memoryList.add(node);
synchronized(this) { node.incrementReferenceCount();
try { return;
this.wait();
} catch (InterruptedException ie) {}
}
lock.lock();
} }
regionDestination = node.getMessage().getRegionDestination(); }
if (!hasSpace()) {
if (isDiskListEmpty()) { if (isDiskListEmpty()) {
if (hasSpace() || this.store==null) { expireOldMessages();
if (hasSpace()) {
memoryList.add(node); memoryList.add(node);
node.incrementReferenceCount(); node.incrementReferenceCount();
return; return;
} else {
flushToDisk();
} }
} }
if (!hasSpace()) {
if (isDiskListEmpty()) {
expireOldMessages();
if (hasSpace()) {
memoryList.add(node);
node.incrementReferenceCount();
return;
} else {
flushToDisk();
}
}
}
if (systemUsage.getTempUsage().isFull()) {
lock.unlock();
systemUsage.getTempUsage().waitForSpace();
lock.lock();
}
getDiskList().add(node);
} finally {
lock.unlock();
}
} catch (Exception e) {
LOG.error("Caught an Exception adding a message: " + node
+ " last to FilePendingMessageCursor ", e);
throw new RuntimeException(e);
}
} else {
discard(node);
}
}
/**
* add message to await dispatch
*
* @param node
*/
public void addMessageFirst(MessageReference node) {
if (!node.isExpired()) {
try {
lock.lock();
try {
while (iterating) {
lock.unlock();
synchronized(this) {
try {
this.wait();
} catch (InterruptedException ie) {}
}
lock.lock();
}
regionDestination = node.getMessage().getRegionDestination();
if (isDiskListEmpty()) {
if (hasSpace()) {
memoryList.addFirst(node);
node.incrementReferenceCount();
return;
}
}
if (!hasSpace()) {
if (isDiskListEmpty()) {
expireOldMessages();
if (hasSpace()) {
memoryList.addFirst(node);
node.incrementReferenceCount();
return;
} else {
flushToDisk();
}
}
}
if (systemUsage.getTempUsage().isFull()) {
lock.unlock();
systemUsage.getTempUsage().waitForSpace();
lock.lock();
}
node.decrementReferenceCount();
getDiskList().addFirst(node);
} finally {
lock.unlock();
} }
systemUsage.getTempUsage().waitForSpace();
getDiskList().add(node);
} catch (Exception e) { } catch (Exception e) {
LOG.error("Caught an Exception adding a message: " + node LOG.error("Caught an Exception adding a message: " + node
@ -309,38 +202,67 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
discard(node); discard(node);
} }
} }
/**
* add message to await dispatch
*
* @param node
*/
public synchronized void addMessageFirst(MessageReference node) {
if (!node.isExpired()) {
try {
regionDestination = node.getMessage().getRegionDestination();
if (isDiskListEmpty()) {
if (hasSpace()) {
memoryList.addFirst(node);
node.incrementReferenceCount();
return;
}
}
if (!hasSpace()) {
if (isDiskListEmpty()) {
expireOldMessages();
if (hasSpace()) {
memoryList.addFirst(node);
node.incrementReferenceCount();
return;
} else {
flushToDisk();
}
}
}
systemUsage.getTempUsage().waitForSpace();
node.decrementReferenceCount();
getDiskList().addFirst(node);
} catch (Exception e) {
LOG.error("Caught an Exception adding a message: " + node
+ " first to FilePendingMessageCursor ", e);
throw new RuntimeException(e);
}
} else {
discard(node);
}
}
/** /**
* @return true if there pending messages to dispatch * @return true if there pending messages to dispatch
*/ */
public boolean hasNext() { public synchronized boolean hasNext() {
boolean result; return iter.hasNext();
lock.lock();
try {
result = iter.hasNext();
} finally {
lock.unlock();
}
return result;
} }
/** /**
* @return the next pending message * @return the next pending message
*/ */
public MessageReference next() { public synchronized MessageReference next() {
Message message; Message message = (Message)iter.next();
lock.lock(); last = message;
try { if (!isDiskListEmpty()) {
message = (Message)iter.next(); // got from disk
last = message; message.setRegionDestination(regionDestination);
if (!isDiskListEmpty()) { message.setMemoryUsage(this.getSystemUsage().getMemoryUsage());
// got from disk message.incrementReferenceCount();
message.setRegionDestination(regionDestination);
message.setMemoryUsage(this.getSystemUsage().getMemoryUsage());
message.incrementReferenceCount();
}
} finally {
lock.unlock();
} }
return message; return message;
} }
@ -348,15 +270,10 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
/** /**
* remove the message at the cursor position * remove the message at the cursor position
*/ */
public void remove() { public synchronized void remove() {
lock.lock(); iter.remove();
try { if (last != null) {
iter.remove(); last.decrementReferenceCount();
if (last != null) {
last.decrementReferenceCount();
}
} finally {
lock.unlock();
} }
} }
@ -364,61 +281,36 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
* @param node * @param node
* @see org.apache.activemq.broker.region.cursors.AbstractPendingMessageCursor#remove(org.apache.activemq.broker.region.MessageReference) * @see org.apache.activemq.broker.region.cursors.AbstractPendingMessageCursor#remove(org.apache.activemq.broker.region.MessageReference)
*/ */
public void remove(MessageReference node) { public synchronized void remove(MessageReference node) {
lock.lock(); if (memoryList.remove(node)) {
try { node.decrementReferenceCount();
if (memoryList.remove(node)) { }
node.decrementReferenceCount(); if (!isDiskListEmpty()) {
} getDiskList().remove(node);
if (!isDiskListEmpty()) {
getDiskList().remove(node);
}
} finally {
lock.unlock();
} }
} }
/** /**
* @return the number of pending messages * @return the number of pending messages
*/ */
public int size() { public synchronized int size() {
int result; return memoryList.size() + (isDiskListEmpty() ? 0 : getDiskList().size());
lock.lock();
try {
result = memoryList.size() + (isDiskListEmpty() ? 0 : getDiskList().size());
} finally {
lock.unlock();
}
return result;
} }
/** /**
* clear all pending messages * clear all pending messages
*/ */
public void clear() { public synchronized void clear() {
lock.lock(); memoryList.clear();
try { if (!isDiskListEmpty()) {
memoryList.clear(); getDiskList().clear();
if (!isDiskListEmpty()) {
getDiskList().clear();
}
last=null;
} finally {
lock.unlock();
} }
last=null;
} }
public boolean isFull() { public synchronized boolean isFull() {
boolean result; // we always have space - as we can persist to disk
lock.lock(); return false;
try {
// we always have space - as we can persist to disk
// TODO: not necessarily true.
result = false;
} finally {
lock.unlock();
}
return result;
} }
public boolean hasMessagesBufferedToDeliver() { public boolean hasMessagesBufferedToDeliver() {
@ -432,8 +324,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
public void onUsageChanged(Usage usage, int oldPercentUsage, public void onUsageChanged(Usage usage, int oldPercentUsage,
int newPercentUsage) { int newPercentUsage) {
if (newPercentUsage >= getMemoryUsageHighWaterMark()) { if (newPercentUsage >= getMemoryUsageHighWaterMark()) {
lock.lock(); synchronized (this) {
try {
flushRequired = true; flushRequired = true;
if (!iterating) { if (!iterating) {
expireOldMessages(); expireOldMessages();
@ -442,8 +333,6 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
flushRequired = false; flushRequired = false;
} }
} }
} finally {
lock.unlock();
} }
} }
} }
@ -456,39 +345,31 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
return hasSpace() && isDiskListEmpty(); return hasSpace() && isDiskListEmpty();
} }
protected void expireOldMessages() { protected synchronized void expireOldMessages() {
lock.lock(); if (!memoryList.isEmpty()) {
try { LinkedList<MessageReference> tmpList = new LinkedList<MessageReference>(this.memoryList);
if (!memoryList.isEmpty()) { this.memoryList = new LinkedList<MessageReference>();
LinkedList<MessageReference> tmpList = new LinkedList<MessageReference>(this.memoryList); while (!tmpList.isEmpty()) {
this.memoryList = new LinkedList<MessageReference>(); MessageReference node = tmpList.removeFirst();
while (!tmpList.isEmpty()) { if (node.isExpired()) {
MessageReference node = tmpList.removeFirst(); discard(node);
if (node.isExpired()) { }else {
discard(node); memoryList.add(node);
}else { }
memoryList.add(node);
}
}
} }
} finally {
lock.unlock();
} }
} }
protected void flushToDisk() { protected synchronized void flushToDisk() {
lock.lock();
try { if (!memoryList.isEmpty()) {
if (!memoryList.isEmpty()) { while (!memoryList.isEmpty()) {
while (!memoryList.isEmpty()) { MessageReference node = memoryList.removeFirst();
MessageReference node = memoryList.removeFirst(); node.decrementReferenceCount();
node.decrementReferenceCount(); getDiskList().addLast(node);
getDiskList().addLast(node);
}
memoryList.clear();
} }
} finally { memoryList.clear();
lock.unlock();
} }
} }

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

@ -16,10 +16,12 @@
*/ */
package org.apache.activemq.broker.region.cursors; package org.apache.activemq.broker.region.cursors;
import org.apache.activemq.ActiveMQMessageAudit;
import org.apache.activemq.broker.Broker; import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.region.MessageReference; import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Queue; import org.apache.activemq.broker.region.Queue;
import org.apache.activemq.command.Message; import org.apache.activemq.command.Message;
import org.apache.activemq.kaha.Store;
import org.apache.activemq.usage.SystemUsage; import org.apache.activemq.usage.SystemUsage;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
@ -87,7 +89,7 @@ public class StoreQueueCursor extends AbstractPendingMessageCursor {
pendingCount = 0; pendingCount = 0;
} }
public void addMessageLast(MessageReference node) throws Exception { public synchronized void addMessageLast(MessageReference node) throws Exception {
if (node != null) { if (node != null) {
Message msg = node.getMessage(); Message msg = node.getMessage();
if (started) { if (started) {
@ -102,7 +104,7 @@ public class StoreQueueCursor extends AbstractPendingMessageCursor {
} }
} }
public void addMessageFirst(MessageReference node) throws Exception { public synchronized void addMessageFirst(MessageReference node) throws Exception {
if (node != null) { if (node != null) {
Message msg = node.getMessage(); Message msg = node.getMessage();
if (started) { if (started) {
@ -140,11 +142,6 @@ public class StoreQueueCursor extends AbstractPendingMessageCursor {
MessageReference result = currentCursor != null ? currentCursor.next() : null; MessageReference result = currentCursor != null ? currentCursor.next() : null;
return result; return result;
} }
public synchronized void release() {
nonPersistent.release();
persistent.release();
}
public synchronized void remove() { public synchronized void remove() {
if (currentCursor != null) { if (currentCursor != null) {
@ -162,7 +159,7 @@ public class StoreQueueCursor extends AbstractPendingMessageCursor {
pendingCount--; pendingCount--;
} }
public void reset() { public synchronized void reset() {
nonPersistent.reset(); nonPersistent.reset();
persistent.reset(); persistent.reset();
} }