activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chir...@apache.org
Subject svn commit: r387562 - /incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
Date Tue, 21 Mar 2006 16:08:41 GMT
Author: chirino
Date: Tue Mar 21 08:08:40 2006
New Revision: 387562

URL: http://svn.apache.org/viewcvs?rev=387562&view=rev
Log:
removed the preLoadLimit logic since it was causing the RoundRobingDispatchPolicyTests to
fail intermitently.  The preLoadLimit was an additional prefetch limit, and when it kicked
in, 
it would cause the round robin distribution to stop sending round robin.

Modified:
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java?rev=387562&r1=387561&r2=387562&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
Tue Mar 21 08:08:40 2006
@@ -46,8 +46,6 @@
     final protected LinkedList dispatched=new LinkedList();
     
     protected int prefetchExtension=0;
-    int preLoadLimit=1024*100;
-    int preLoadSize=0;
     boolean dispatching=false;
     
     long enqueueCounter;
@@ -65,6 +63,8 @@
             dispatch(node);
         }else{
             synchronized(pending){
+                if( pending.isEmpty() )
+                    log.info("Prefetch limit.");
                 pending.addLast(node);
             }
         }
@@ -79,7 +79,6 @@
                     try{
                         MessageDispatch md=createMessageDispatch(node,node.getMessage());
                         dispatched.addLast(node);
-                        incrementPreloadSize(node.getSize());
                         node.decrementReferenceCount();
                     }catch(Exception e){
                         log.error("Problem processing MessageDispatchNotification: "+mdn,e);
@@ -101,6 +100,7 @@
                 final MessageReference node=(MessageReference) iter.next();
                 MessageId messageId=node.getMessageId();
                 if(ack.getFirstMessageId()==null||ack.getFirstMessageId().equals(messageId)){
+System.out.println("in range: "+messageId);
                     inAckRange=true;
                 }
                 if(inAckRange){
@@ -202,7 +202,7 @@
     }
 
     protected boolean isFull(){
-        return dispatched.size()-prefetchExtension>=info.getPrefetchSize()||preLoadSize>preLoadLimit;
+        return dispatched.size()-prefetchExtension>=info.getPrefetchSize();
     }
     
     synchronized public int getPendingQueueSize(){
@@ -252,7 +252,6 @@
             dispatchCounter++;
             MessageDispatch md=createMessageDispatch(node,message);
             dispatched.addLast(node);            
-            incrementPreloadSize(node.getMessage().getSize());
             if(info.isDispatchAsync()){
                 md.setConsumer(new Runnable(){
                     public void run(){
@@ -275,7 +274,6 @@
 
     synchronized private void onDispatch(final MessageReference node,final Message message){
         boolean wasFull=isFull();
-        decrementPreloadSize(message.getSize());
         node.decrementReferenceCount();
         if(node.getRegionDestination()!=null){
             node.getRegionDestination().getDestinationStatistics().onMessageDequeue(message);
@@ -288,16 +286,6 @@
                 }
             }
         }
-    }
-
-    private int incrementPreloadSize(int size){
-        preLoadSize+=size;
-        return preLoadSize;
-    }
-
-    private int decrementPreloadSize(int size){
-        preLoadSize-=size;
-        return preLoadSize;
     }
 
     /**



Mime
View raw message