tomcat-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fha...@apache.org
Subject svn commit: r381703 - in /tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes: demos/ group/ group/interceptors/ io/
Date Tue, 28 Feb 2006 16:51:22 GMT
Author: fhanik
Date: Tue Feb 28 08:51:07 2006
New Revision: 381703

URL: http://svn.apache.org/viewcvs?rev=381703&view=rev
Log:
Fixed buffer usage

Modified:
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/ChannelCreator.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/LoadTest.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/GroupChannel.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ClusterData.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ObjectReader.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/XByteBuffer.java

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/ChannelCreator.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/ChannelCreator.java?rev=381703&r1=381702&r2=381703&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/ChannelCreator.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/ChannelCreator.java
Tue Feb 28 08:51:07 2006
@@ -22,6 +22,7 @@
 import org.apache.catalina.tribes.group.GroupChannel;
 import org.apache.catalina.tribes.mcast.McastService;
 import org.apache.catalina.tribes.group.interceptors.GzipInterceptor;
+import org.apache.catalina.tribes.group.interceptors.OrderInterceptor;
 
 /**
  * <p>Title: </p>
@@ -53,6 +54,7 @@
            .append("\n\t\t[-mfreq multicastfrequency]")
            .append("\n\t\t[-mdrop multicastdroptime]")
            .append("\n\t\t[-gzip]")
+           .append("\n\t\t[-order]")
            ;
        return buf;
 
@@ -74,6 +76,7 @@
         int mcastport = 45565;
         long mcastfreq = 500;
         long mcastdrop = 2000;
+        boolean order = false;
         
         for (int i = 0; i < args.length; i++) {
             if ("-bind".equals(args[i])) {
@@ -88,6 +91,8 @@
                 tcpthreadcount = Integer.parseInt(args[++i]);
             } else if ("-gzip".equals(args[i])) {
                 gzip = true;
+            } else if ("-order".equals(args[i])) {
+                order = true;
             } else if ("-ack".equals(args[i])) {
                 ack = Boolean.parseBoolean(args[++i]);
             } else if ("-ackto".equals(args[i])) {
@@ -137,6 +142,7 @@
         channel.setMembershipService(service);
 
         if (gzip) channel.addInterceptor(new GzipInterceptor());
+        if (order) channel.addInterceptor(new OrderInterceptor());
         return channel;
         
     }

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/LoadTest.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/LoadTest.java?rev=381703&r1=381702&r2=381703&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/LoadTest.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/demos/LoadTest.java
Tue Feb 28 08:51:07 2006
@@ -121,12 +121,13 @@
     public void run() {
         
         long counter = 0;
+        long total = 0;
         LoadMessage msg = new LoadMessage();
         int messageSize = LoadTest.messageSize;
         
         try {
             startTest();
-            while (counter < msgCount) {
+            while (total < msgCount) {
                 if (channel.getMembers().length == 0 || (!send)) {
                     synchronized (mutex) {
                         try {
@@ -137,7 +138,8 @@
                     }
                 } else {
                     try {
-                        msg.setMsgNr((int)++counter);
+                        msg.setMsgNr((int)++total);
+                        counter++;
                         if (debug) {
                             printArray(msg.getMessage());
                         }
@@ -334,6 +336,7 @@
                 threads = Integer.parseInt(args[++i]);
             } else if ("-count".equals(args[i])) {
                 count = Integer.parseInt(args[++i]);
+                System.out.println("Sending "+count+" messages.");
             } else if ("-pause".equals(args[i])) {
                 pause = Long.parseLong(args[++i])*1000;
             } else if ("-break".equals(args[i])) {
@@ -345,7 +348,7 @@
                 if ( "receive".equals(args[++i]) ) send = false;
             } else if ("-debug".equals(args[i])) {
                 debug = true;
-            } else //("-help".equals(args[i])) 
+            } else if ("-help".equals(args[i])) 
             {
                 usage();
                 System.exit(1);
@@ -357,6 +360,7 @@
         
         LoadTest test = new LoadTest(channel,send,count,debug,pause,stats,breakOnEx);
         LoadMessage msg = new LoadMessage();
+        
         messageSize = LoadMessage.getMessageSize(msg);
         channel.setChannelListener(test);
         channel.setMembershipListener(test);
@@ -383,6 +387,9 @@
         
         public void run() {
             System.out.println("Shutting down...");
+            SystemExit exit = new SystemExit(5000);
+            exit.setDaemon(true);
+            exit.start();
             try {
                 channel.stop(channel.DEFAULT);
                 
@@ -390,6 +397,21 @@
                 x.printStackTrace();
             }
             System.out.println("Channel stopped.");
+        }
+    }
+    public static class SystemExit extends Thread {
+        private long delay;
+        public SystemExit(long delay) {
+            this.delay = delay;
+        }
+        public void run () {
+            try {
+                Thread.sleep(delay);
+            }catch ( Exception x ) {
+                x.printStackTrace();
+            }
+            System.exit(0);
+
         }
     }
     

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/GroupChannel.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/GroupChannel.java?rev=381703&r1=381702&r2=381703&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/GroupChannel.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/GroupChannel.java
Tue Feb 28 08:51:07 2006
@@ -91,6 +91,7 @@
     public void send(Member[] destination, Serializable msg) throws ChannelException {
         if ( msg == null ) return;
         try {
+            if ( destination == null ) destination = getMembers();
             int options = 0;
             ClusterData data = new ClusterData();//generates a unique Id
             data.setAddress(getLocalMember());
@@ -103,7 +104,7 @@
                 b = XByteBuffer.serialize(msg);
             }
             data.setOptions(options);
-            XByteBuffer buffer = new XByteBuffer(b.length+128);
+            XByteBuffer buffer = new XByteBuffer(b.length+128,false);
             buffer.append(b,0,b.length);
             data.setMessage(buffer);
             getFirstInterceptor().sendMessage(destination, data, null);

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java?rev=381703&r1=381702&r2=381703&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
Tue Feb 28 08:51:07 2006
@@ -37,7 +37,7 @@
     private HashMap incounter = new HashMap();
     private HashMap incoming = new HashMap();
     private long expire = 3000;
-    private boolean forwardExpired;
+    private boolean forwardExpired = true;
 
     public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload
payload) throws ChannelException {
         for ( int i=0; i<destination.length; i++ ) {
@@ -77,6 +77,7 @@
         while ( tmp != null ) {
             //process expired messages
             if ( tmp.isExpired(expire) ) {
+                System.out.println("Found expired message");
                 //reset the head
                 if ( tmp == head ) head = tmp.next;
                 if ( getForwardExpired() ) super.messageReceived(tmp.getMessage());

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ClusterData.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ClusterData.java?rev=381703&r1=381702&r2=381703&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ClusterData.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ClusterData.java
Tue Feb 28 08:51:07 2006
@@ -198,7 +198,7 @@
         System.arraycopy(b,offset,addr,0,addr.length);
         data.setAddress(McastMember.getMember(addr));
         offset += addr.length; //addr data
-        data.message = new XByteBuffer(new byte[XByteBuffer.toInt(b,offset)]);
+        data.message = new XByteBuffer(new byte[XByteBuffer.toInt(b,offset)],false);
         offset += 4; //message length
         System.arraycopy(b,offset,data.message.getBytesDirect(),0,data.message.getLength());
         offset += data.message.getLength(); //message data

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ObjectReader.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ObjectReader.java?rev=381703&r1=381702&r2=381703&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ObjectReader.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ObjectReader.java
Tue Feb 28 08:51:07 2006
@@ -52,7 +52,7 @@
     public ObjectReader(SocketChannel channel, Selector selector, ListenCallback callback)
{
         this.channel = channel;
         this.callback = callback;
-        this.buffer = new XByteBuffer();
+        this.buffer = new XByteBuffer(true);
     }
 
     /**

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/XByteBuffer.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/XByteBuffer.java?rev=381703&r1=381702&r2=381703&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/XByteBuffer.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/XByteBuffer.java
Tue Feb 28 08:51:07 2006
@@ -85,30 +85,38 @@
     protected int bufSize = 0;
     
     /**
+     * 
+     * 
+     */
+    protected boolean discard = true;
+    
+    /**
      * Constructs a new XByteBuffer
      * @param size - the initial size of the byte buffer
      */
-    public XByteBuffer(int size) {
+    public XByteBuffer(int size, boolean discard) {
         buf = new byte[size];
+        this.discard = discard;
     }
     
-    public XByteBuffer(byte[] data) {
-        this(data,data.length+128);
+    public XByteBuffer(byte[] data,boolean discard) {
+        this(data,data.length+128,discard);
     }
     
-    public XByteBuffer(byte[] data, int size) {
+    public XByteBuffer(byte[] data, int size,boolean discard) {
         int length = Math.max(data.length,size);
         buf = new byte[length];
         System.arraycopy(data,0,buf,0,data.length);
         bufSize = data.length;
+        this.discard = discard;
     }
 
 
     /**
      * Constructs a new XByteBuffer with an initial size of 1024 bytes
      */
-    public XByteBuffer()  {
-        this(DEF_SIZE);
+    public XByteBuffer(boolean discard)  {
+        this(DEF_SIZE,discard);
     }
     
     public int getLength() {
@@ -160,11 +168,13 @@
         b.get(buf,bufSize,len);
         
         bufSize = newcount;
-
-        if (bufSize > START_DATA.length && (firstIndexOf(buf,0,START_DATA)==-1)){
-            bufSize = 0;
-            log.error("Discarded the package, invalid header");
-            return false;
+        
+        if ( discard ) {
+            if (bufSize > START_DATA.length && (firstIndexOf(buf, 0, START_DATA)
== -1)) {
+                bufSize = 0;
+                log.error("Discarded the package, invalid header");
+                return false;
+            }
         }
         return true;
 
@@ -188,10 +198,12 @@
         System.arraycopy(b, off, buf, bufSize, len);
         bufSize = newcount;
 
-        if (bufSize > START_DATA.length && (firstIndexOf(buf,0,START_DATA)==-1)){
-            bufSize = 0;
-            log.error("Discarded the package, invalid header");
-            return false;
+        if ( discard ) {
+            if (bufSize > START_DATA.length && (firstIndexOf(buf, 0, START_DATA)
== -1)) {
+                bufSize = 0;
+                log.error("Discarded the package, invalid header");
+                return false;
+            }
         }
         return true;
     }



---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org


Mime
View raw message