Tidied up locking around cursor iterators

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@479156 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java

@@ -118,14 +118,18 @@ public class DurableTopicSubscription extends PrefetchSubscription {
             }
         }
         
-        if( !keepDurableSubsActive ) {
+        if(!keepDurableSubsActive){
-        	synchronized(pending) {
+            synchronized(pending){
+                try{
                     pending.reset();
-	            while(pending.hasNext()) {
+                    while(pending.hasNext()){
-	                MessageReference node = pending.next();
+                        MessageReference node=pending.next();
                         node.decrementReferenceCount();
                         pending.remove();
                     }
+                }finally{
+                    pending.release();
+                }
             }
         }
         prefetchExtension=0;
@@ -195,22 +199,24 @@ public class DurableTopicSubscription extends PrefetchSubscription {
     /**
      * Release any references that we are holding.
      */
-    public void destroy() {
+    public void destroy(){
-    	synchronized(pending) {
+        try{
+            synchronized(pending){
                 pending.reset();
-	        while(pending.hasNext()) {
+                while(pending.hasNext()){
-	            MessageReference node = pending.next();
+                    MessageReference node=pending.next();
                     node.decrementReferenceCount();
                 }
+            }
+        }finally{
+            pending.release();
             pending.clear();
         }
-    	
+        for(Iterator iter=dispatched.iterator();iter.hasNext();){
-        for (Iterator iter = dispatched.iterator(); iter.hasNext();) {
+            MessageReference node=(MessageReference)iter.next();
-            MessageReference node = (MessageReference) iter.next();
             node.decrementReferenceCount();
         }
         dispatched.clear();
-        
     }
 
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java

@@ -123,6 +123,7 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
 	}
         
     public void add(MessageReference node) throws Exception{
+        try {
         boolean pendingEmpty = false;
         synchronized(pending){
             pendingEmpty=pending.isEmpty();
@@ -139,10 +140,15 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
                 pending.addMessageLast(node);
             }
         }
+        }catch(Throwable e) {
+            e.printStackTrace();
+            
+        }
     }
 
-    public void processMessageDispatchNotification(MessageDispatchNotification mdn) throws Exception {
+    public void processMessageDispatchNotification(MessageDispatchNotification mdn) throws Exception{
         synchronized(pending){
+            try{
                 pending.reset();
                 while(pending.hasNext()){
                     MessageReference node=pending.next();
@@ -153,7 +159,11 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
                         return;
                     }
                 }
-            throw new JMSException("Slave broker out of sync with master: Dispatched message ("+mdn.getMessageId()+") was not in the pending list: "+pending);
+            }finally{
+                pending.release();
+            }
+            throw new JMSException("Slave broker out of sync with master: Dispatched message ("+mdn.getMessageId()
+                    +") was not in the pending list: "+pending);
         }
     }
 
@@ -387,6 +397,7 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
                         dispatch(node);
                     }
                 }finally{
+                    pending.release();
                     dispatching=false;
                 }
             }

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

@@ -545,54 +545,58 @@ public class Queue implements Destination, Task {
                 }
             }
         }
-        synchronized (messages) {
+        synchronized(messages){
+            try{
                 messages.reset();
-            while(messages.hasNext()) {
+                while(messages.hasNext()){
-                try {
+                    try{
-                    MessageReference r = messages.next();
+                        MessageReference r=messages.next();
                         r.incrementReferenceCount();
-                    try {
+                        try{
-                        Message m = r.getMessage();
+                            Message m=r.getMessage();
-                        if (m != null) {
+                            if(m!=null){
                                 l.add(m);
                             }
-                    }
+                        }finally{
-                    finally {
                             r.decrementReferenceCount();
                         }
+                    }catch(IOException e){
+                        log.error("caught an exception brwsing "+this,e);
                     }
-                catch (IOException e) {
-                    log.error("caught an exception brwsing " + this,e);
                 }
+            }finally{
+                messages.release();
             }
         }
 
         return (Message[]) l.toArray(new Message[l.size()]);
     }
 
-    public Message getMessage(String messageId) {
+    public Message getMessage(String messageId){
-        synchronized (messages) {
+        synchronized(messages){
+            try{
                 messages.reset();
-            while(messages.hasNext()) {
+                while(messages.hasNext()){
-                try {
+                    try{
-                    MessageReference r = messages.next();
+                        MessageReference r=messages.next();
-                    if (messageId.equals(r.getMessageId().toString())) {
+                        if(messageId.equals(r.getMessageId().toString())){
                             r.incrementReferenceCount();
-                        try {
+                            try{
-                            Message m = r.getMessage();
+                                Message m=r.getMessage();
-                            if (m != null) {
+                                if(m!=null){
                                     return m;
                                 }
-                        }
+                            }finally{
-                        finally {
                                 r.decrementReferenceCount();
                             }
                             break;
                         }
+                    }catch(IOException e){
+                        log.error("got an exception retrieving message "+messageId);
                     }
-                catch (IOException e) {
-                    log.error("got an exception retrieving message " + messageId);
                 }
+            }finally{
+                messages.release();
             }
         }
         return null;
@@ -868,6 +872,7 @@ public class Queue implements Destination, Task {
                 int count=0;
                 result=new ArrayList(toPageIn);
                 synchronized(messages){
+                    try{
                         messages.reset();
                         while(messages.hasNext()&&count<toPageIn){
                             MessageReference node=messages.next();
@@ -876,6 +881,9 @@ public class Queue implements Destination, Task {
                             result.add(node);
                             count++;
                         }
+                    }finally{
+                        messages.release();
+                    }
                 }
                 synchronized(pagedInMessages){
                     pagedInMessages.addAll(result);

activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java

@@ -135,24 +135,28 @@ public class TopicSubscription extends AbstractSubscription{
      * Discard any expired messages from the matched list. Called from a synchronized block.
      * @throws IOException 
      */
-    protected void removeExpiredMessages() throws IOException {
+    protected void removeExpiredMessages() throws IOException{
+        try{
             matched.reset();
-        while(matched.hasNext()) {
+            while(matched.hasNext()){
                 MessageReference node=matched.next();
-            if (node.isExpired()) {
+                if(node.isExpired()){
                     matched.remove();
                     dispatched.incrementAndGet();
                     node.decrementReferenceCount();
                     break;
                 }
             }
+        }finally{
             matched.release();
         }
+    }
 
     public void processMessageDispatchNotification(MessageDispatchNotification mdn){
         synchronized(matchedListMutex){
+            try{
                 matched.reset();
-            while(matched.hasNext()) {
+                while(matched.hasNext()){
                     MessageReference node=matched.next();
                     if(node.getMessageId().equals(mdn.getMessageId())){
                         matched.remove();
@@ -161,9 +165,11 @@ public class TopicSubscription extends AbstractSubscription{
                         break;
                     }
                 }
+            }finally{
                 matched.release();
             }
         }
+    }
 
     synchronized public void acknowledge(final ConnectionContext context,final MessageAck ack) throws Exception{
         
@@ -335,23 +341,24 @@ public class TopicSubscription extends AbstractSubscription{
 
     private void dispatchMatched() throws IOException{
         synchronized(matchedListMutex){
+            try{
                 matched.reset();
-            while(matched.hasNext()) {
+                while(matched.hasNext()){
-                MessageReference message=(MessageReference) matched.next();
+                    MessageReference message=(MessageReference)matched.next();
                     matched.remove();
-                
                     // Message may have been sitting in the matched list a while
                     // waiting for the consumer to ak the message.
-        		if( message.isExpired() ) {
+                    if(message.isExpired()){
                         message.decrementReferenceCount();
                         continue; // just drop it.
                     }
-
                     dispatch(message);
                 }
+            }finally{
                 matched.release();
             }
         }
+    }
 
     private void dispatch(final MessageReference node) throws IOException{
         Message message=(Message) node;

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

@@ -110,4 +110,8 @@ public class AbstractPendingMessageCursor implements PendingMessageCursor{
     public boolean isFull() {
         return usageManager != null ? usageManager.isFull() : false;
     }
+
+    
+    public void release(){        
+    }
 }

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

@@ -18,7 +18,6 @@ import java.io.IOException;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import org.apache.activemq.broker.region.Destination;
 import org.apache.activemq.broker.region.MessageReference;
@@ -38,6 +37,7 @@ import org.apache.commons.logging.LogFactory;
  * @version $Revision$
  */
 public class FilePendingMessageCursor extends AbstractPendingMessageCursor implements UsageListener{
+
     static private final Log log=LogFactory.getLog(FilePendingMessageCursor.class);
     private Store store;
     private String name;
@@ -45,8 +45,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
     private ListContainer diskList;
     private Iterator iter=null;
     private Destination regionDestination;
-    private Lock iterLock=new ReentrantLock();
+    private ReentrantLock iterLock=new ReentrantLock();
-    private Object mutex=new Object();
 
     /**
      * @param name
@@ -60,20 +59,20 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
     /**
      * @return true if there are no pending messages
      */
-    public boolean isEmpty(){
+    public synchronized boolean isEmpty(){
-        synchronized(mutex){
         return memoryList.isEmpty()&&isDiskListEmpty();
     }
-    }
 
     /**
      * reset the cursor
      * 
      */
     public void reset(){
-        iterLock.lock();
+        try{
-        synchronized(mutex){
+            iterLock.lockInterruptibly();
-            iter=isSpaceInMemoryList()?memoryList.iterator():diskList.listIterator();
+            iter=isDiskListEmpty()?memoryList.iterator():getDiskList().listIterator();
+        }catch(InterruptedException e){
+            log.warn("Failed to get lock ",e);
         }
     }
 
@@ -81,7 +80,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
         iterLock.unlock();
     }
 
-    public void destroy(){
+    public synchronized void destroy(){
         for(Iterator i=memoryList.iterator();i.hasNext();){
             Message node=(Message)i.next();
             node.decrementReferenceCount();
@@ -92,9 +91,8 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
         }
     }
 
-    public LinkedList pageInList(int maxItems){
+    public synchronized LinkedList pageInList(int maxItems){
         LinkedList result=new LinkedList();
-        synchronized(mutex){
         int count=0;
         for(Iterator i=memoryList.iterator();i.hasNext()&&count<maxItems;){
             result.add(i.next());
@@ -109,7 +107,6 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
                 count++;
             }
         }
-        }
         return result;
     }
 
@@ -118,8 +115,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
      * 
      * @param node
      */
-    public void addMessageLast(MessageReference node){
+    public synchronized void addMessageLast(MessageReference node){
-        synchronized(mutex){
         try{
             regionDestination=node.getMessage().getRegionDestination();
             if(isSpaceInMemoryList()){
@@ -133,15 +129,13 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
             throw new RuntimeException(e);
         }
     }
-    }
 
     /**
      * add message to await dispatch
      * 
      * @param node
      */
-    public void addMessageFirst(MessageReference node){
+    public synchronized void addMessageFirst(MessageReference node){
-        synchronized(mutex){
         try{
             regionDestination=node.getMessage().getRegionDestination();
             if(isSpaceInMemoryList()){
@@ -155,22 +149,18 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
             throw new RuntimeException(e);
         }
     }
-    }
 
     /**
      * @return true if there pending messages to dispatch
      */
-    public boolean hasNext(){
+    public synchronized boolean hasNext(){
-        synchronized(mutex){
         return iter.hasNext();
     }
-    }
 
     /**
      * @return the next pending message
      */
-    public MessageReference next(){
+    public synchronized MessageReference next(){
-        synchronized(mutex){
         Message message=(Message)iter.next();
         if(!isDiskListEmpty()){
             // got from disk
@@ -179,54 +169,45 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
         }
         return message;
     }
-    }
 
     /**
      * remove the message at the cursor position
      * 
      */
-    public void remove(){
+    public synchronized void remove(){
-        synchronized(mutex){
         iter.remove();
     }
-    }
 
     /**
      * @param node
      * @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){
-        synchronized(mutex){
         memoryList.remove(node);
         if(!isDiskListEmpty()){
             getDiskList().remove(node);
         }
     }
-    }
 
     /**
      * @return the number of pending messages
      */
-    public int size(){
+    public synchronized int size(){
-        synchronized(mutex){
         return memoryList.size()+(isDiskListEmpty()?0:getDiskList().size());
     }
-    }
 
     /**
      * clear all pending messages
      * 
      */
-    public void clear(){
+    public synchronized void clear(){
-        synchronized(mutex){
         memoryList.clear();
         if(!isDiskListEmpty()){
             getDiskList().clear();
         }
     }
-    }
 
-    public boolean isFull(){
+    public synchronized boolean isFull(){
         // we always have space - as we can persist to disk
         return false;
     }
@@ -253,8 +234,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
         return hasSpace()&&isDiskListEmpty();
     }
 
-    protected void flushToDisk(){
+    protected synchronized void flushToDisk(){
-        synchronized(mutex){
         for(Iterator i=memoryList.iterator();i.hasNext();){
             MessageReference node=(MessageReference)i.next();
             node.decrementReferenceCount();
@@ -262,7 +242,6 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
         }
         memoryList.clear();
     }
-    }
 
     protected boolean isDiskListEmpty(){
         return diskList==null||diskList.isEmpty();

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

@@ -55,6 +55,13 @@ public interface PendingMessageCursor extends Service{
      */
     public void reset();
     
+    /**
+     * hint to the cursor to release any locks it might have
+     * grabbed after a reset
+     *
+     */
+    public void release();
+
     /**
      * add message to await dispatch
      * @param node

activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java

@@ -185,13 +185,19 @@ public class StoreDurableSubscriberCursor extends AbstractPendingMessageCursor{
     }
 
     public synchronized void reset(){
-        nonPersistent.reset();
         for(Iterator i=storePrefetches.iterator();i.hasNext();){
             AbstractPendingMessageCursor tsp=(AbstractPendingMessageCursor)i.next();
             tsp.reset();
         }
     }
     
+    public synchronized void release(){
+        for(Iterator i=storePrefetches.iterator();i.hasNext();){
+            AbstractPendingMessageCursor tsp=(AbstractPendingMessageCursor)i.next();
+            tsp.release();
+        }
+    }
+
     public int size(){
         return pendingCount;
     }