tomcat-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fha...@apache.org
Subject svn commit: r381798 - in /tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes: ChannelInterceptor.java ChannelSender.java demos/ChannelCreator.java group/interceptors/FragmentationInterceptor.java io/XByteBuffer.java
Date Tue, 28 Feb 2006 22:26:53 GMT
Author: fhanik
Date: Tue Feb 28 14:26:50 2006
New Revision: 381798

URL: http://svn.apache.org/viewcvs?rev=381798&view=rev
Log:
Completed the fragmentation interceptor

Modified:
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelInterceptor.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelSender.java
    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/group/interceptors/FragmentationInterceptor.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/ChannelInterceptor.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelInterceptor.java?rev=381798&r1=381797&r2=381798&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelInterceptor.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelInterceptor.java
Tue Feb 28 14:26:50 2006
@@ -26,7 +26,7 @@
  * Abstract class for the interceptor base class.
  * @author Filip Hanik
  * @version $Revision: 304032 $, $Date: 2005-07-27 10:11:55 -0500 (Wed, 27 Jul 2005) $
- */
+ */   
 
 public interface ChannelInterceptor extends MembershipListener {
 

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelSender.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelSender.java?rev=381798&r1=381797&r2=381798&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelSender.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelSender.java
Tue Feb 28 14:26:50 2006
@@ -22,7 +22,7 @@
  * @author Filip Hanik
  * @version $Revision: 303950 $, $Date: 2005-06-09 15:38:30 -0500 (Thu, 09 Jun 2005) $
  *
- */
+ */    
 public interface ChannelSender
 {
 

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=381798&r1=381797&r2=381798&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 14:26:50 2006
@@ -23,6 +23,7 @@
 import org.apache.catalina.tribes.mcast.McastService;
 import org.apache.catalina.tribes.group.interceptors.GzipInterceptor;
 import org.apache.catalina.tribes.group.interceptors.OrderInterceptor;
+import org.apache.catalina.tribes.group.interceptors.FragmentationInterceptor;
 
 /**
  * <p>Title: </p>
@@ -55,7 +56,9 @@
            .append("\n\t\t[-mdrop multicastdroptime]")
            .append("\n\t\t[-gzip]")
            .append("\n\t\t[-order]")
-           .append("\n\t\t[-ordersize maxorderqueuesize]");
+           .append("\n\t\t[-ordersize maxorderqueuesize]")
+           .append("\n\t\t[-frag]")
+           .append("\n\t\t[-fragsize maxmsgsize]");
        return buf;
 
     }
@@ -78,6 +81,8 @@
         long mcastdrop = 2000;
         boolean order = false;
         int ordersize = Integer.MAX_VALUE;
+        boolean frag = false;
+        int fragsize = 1024;
         
         for (int i = 0; i < args.length; i++) {
             if ("-bind".equals(args[i])) {
@@ -97,6 +102,11 @@
             } else if ("-ordersize".equals(args[i])) {
                 ordersize = Integer.parseInt(args[++i]);
                 System.out.println("Setting OrderInterceptor.maxQueue="+ordersize);
+            } else if ("-frag".equals(args[i])) {
+                frag = true;
+            } else if ("-fragsize".equals(args[i])) {
+                fragsize = Integer.parseInt(args[++i]);
+                System.out.println("Setting FragmentationInterceptor.maxSize="+fragsize);
             } else if ("-ack".equals(args[i])) {
                 ack = Boolean.parseBoolean(args[++i]);
             } else if ("-ackto".equals(args[i])) {
@@ -146,6 +156,11 @@
         channel.setMembershipService(service);
 
         if (gzip) channel.addInterceptor(new GzipInterceptor());
+        if ( frag ) {
+            FragmentationInterceptor fi = new FragmentationInterceptor();
+            fi.setMaxSize(fragsize);
+            channel.addInterceptor(fi);
+        }
         if (order) {
             OrderInterceptor oi = new OrderInterceptor();
             oi.setMaxQueue(ordersize);

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java?rev=381798&r1=381797&r2=381798&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java
Tue Feb 28 14:26:50 2006
@@ -15,13 +15,200 @@
 
 package org.apache.catalina.tribes.group.interceptors;
 
+import java.util.Arrays;
+import java.util.HashMap;
+
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.InterceptorPayload;
+import org.apache.catalina.tribes.Member;
 import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.io.XByteBuffer;
 
 /**
  *
- *
+ * The fragmentation interceptor splits up large messages into smaller messages and assembles
them on the other end.
+ * This is very useful when you don't want large messages hogging the sending sockets
+ * and smaller messages can make it through.
+ * 
+ * <br><b>Configuration Options</b><br>
+ * OrderInteceptor.expire=<milliseconds> - how long do we keep the fragments in memory
and wait for the rest to arrive<b>default=60,000ms -> 60seconds</b>
+ * This setting is useful to avoid OutOfMemoryErrors<br>
+ * OrderInteceptor.maxSize=<max message size> - message size in bytes <b>default=1024*100
(around a tenth of a MB)</b><br>
  * @author Filip Hanik
  * @version 1.0
  */
 public class FragmentationInterceptor extends ChannelInterceptorBase {
+    protected HashMap fragpieces = new HashMap();
+    private int maxSize = 1024*100;
+    private long expire = 1000 * 60; //one minute expiration
+
+
+    public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload
payload) throws ChannelException {
+        int size = msg.getMessage().getLength();
+        boolean frag = (size>maxSize);
+        if ( frag ) {
+            frag(destination, msg, payload);
+        }
+        else {
+            byte[] flag = XByteBuffer.toBytes(frag);
+            msg.getMessage().append(flag,0,flag.length);
+            super.sendMessage(destination, msg, payload);
+        }
+    }
+    
+    public void messageReceived(ChannelMessage msg) {
+        boolean isFrag = XByteBuffer.toBoolean(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-1);
+        msg.getMessage().trim(1);
+        if ( isFrag ) {
+            defrag(msg);
+        } else {
+            super.messageReceived(msg);
+        }
+    }
+
+    
+    public synchronized FragCollection getFragCollection(FragKey key, ChannelMessage msg)
{
+        FragCollection coll = (FragCollection)fragpieces.get(key);
+        if ( coll == null ) {
+            coll = new FragCollection(msg);
+            fragpieces.put(key,coll);
+        } 
+        return coll;
+    }
+    
+    public synchronized void removeFragCollection(FragKey key) {
+        fragpieces.remove(key);
+    }
+    
+    public void defrag(ChannelMessage msg ) { 
+        FragKey key = new FragKey(msg.getUniqueId());
+        FragCollection coll = getFragCollection(key,msg);
+        coll.addMessage(msg);
+
+        if ( coll.complete() ) {
+            removeFragCollection(key);
+            ChannelMessage complete = coll.assemble();
+            super.messageReceived(complete);
+            
+        }
+    }
+
+    public void frag(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
throws ChannelException {
+        int size = msg.getMessage().getLength();
+
+        int count = ((size / maxSize )+(size%maxSize==0?0:1));
+        ChannelMessage[] messages = new ChannelMessage[count];
+        int remaining = size;
+        for ( int i=0; i<count; i++ ) {
+            ChannelMessage tmp = msg.clone();
+            int offset = (i*maxSize);
+            int length = Math.min(remaining,maxSize);
+            tmp.getMessage().clear();
+            tmp.getMessage().append(msg.getMessage().getBytesDirect(),offset,length);
+            //add the msg nr
+            tmp.getMessage().append(XByteBuffer.toBytes(i),0,4);
+            //add the total nr of messages
+            tmp.getMessage().append(XByteBuffer.toBytes(count),0,4);
+            //add true as the frag flag
+            byte[] flag = XByteBuffer.toBytes(true);
+            tmp.getMessage().append(flag,0,flag.length);
+            messages[i] = tmp;
+            remaining -= length;
+            
+        }
+        for ( int i=0; i<messages.length; i++ ) {
+            super.sendMessage(destination,messages[i],payload);
+        }
+    }
+
+    
+
+    public int getMaxSize() {
+        return maxSize;
+    }
+
+    public long getExpire() {
+        return expire;
+    }
+
+    public void setMaxSize(int maxSize) {
+        this.maxSize = maxSize;
+    }
+
+    public void setExpire(long expire) {
+        this.expire = expire;
+    }
+
+    public static class FragCollection {
+        private long received = System.currentTimeMillis();
+        private ChannelMessage msg;
+        private XByteBuffer[] frags;
+        public FragCollection(ChannelMessage msg) {
+            //get the total messages
+            int count = XByteBuffer.toInt(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-4);
+            frags = new XByteBuffer[count];
+            this.msg = msg;
+        }
+        
+        public void addMessage(ChannelMessage msg) {
+            //remove the total messages
+            msg.getMessage().trim(4);
+            //get the msg nr
+            int nr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-4);
+            //remove the msg nr
+            msg.getMessage().trim(4);
+            frags[nr] = msg.getMessage();
+            
+        }
+        
+        public boolean complete() {
+            boolean result = true;
+            for ( int i=0; (i<frags.length) && (result); i++ ) result = (frags[i]
!= null);
+            return result;
+        }
+        
+        public ChannelMessage assemble() {
+            if ( !complete() ) throw new IllegalStateException("Fragments are missing.");
+            int buffersize = 0;
+            for (int i=0; i<frags.length; i++ ) buffersize += frags[i].getLength();
+            XByteBuffer buf = new XByteBuffer(buffersize,false);
+            msg.setMessage(buf);
+            for ( int i=0; i<frags.length; i++ ) {
+                msg.getMessage().append(frags[i].getBytesDirect(),0,frags[i].getLength());
+            }
+            return msg;
+        }
+        
+        public boolean expired(long expire) {
+            return (System.currentTimeMillis()-received)>expire;
+        }
+
+        
+        
+    }
+    
+    public static class FragKey {
+        private byte[] uniqueId;
+        private long received = System.currentTimeMillis();
+        public FragKey(byte[] id ) {
+            this.uniqueId = id;
+        }
+        public int hashCode() {
+            return XByteBuffer.toInt(uniqueId,0);
+        }
+        
+        public boolean equals(Object o ) {
+            if ( o instanceof FragKey ) {
+            return Arrays.equals(uniqueId,((FragKey)o).uniqueId);
+        } else return false;
+
+        }
+        
+        public boolean expired(long expire) {
+            return (System.currentTimeMillis()-received)>expire;
+        }
+
+    }
+    
 }

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=381798&r1=381797&r2=381798&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 14:26:50 2006
@@ -331,6 +331,22 @@
             ( ( ( (long) b[off+0]) & 0xFF) << 56);
     }
 
+    
+    /**
+     * Converts an integer to four bytes
+     * @param n - the integer
+     * @return - four bytes in an array
+     */
+    public static byte[] toBytes(boolean bool) {
+        byte[] b = new byte[] {(byte)(bool?1:0)};
+        return b;
+    }
+    
+    public static boolean toBoolean(byte[] b, int offset) {
+        return b[offset] != 0;
+    }
+
+    
     /**
      * Converts an integer to four bytes
      * @param n - the integer



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


Mime
View raw message