tomcat-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fha...@apache.org
Subject svn commit: r381446 - in /tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes: demos/ group/ io/ mcast/ tcp/
Date Mon, 27 Feb 2006 21:07:41 GMT
Author: fhanik
Date: Mon Feb 27 13:07:39 2006
New Revision: 381446

URL: http://svn.apache.org/viewcvs?rev=381446&view=rev
Log:
Optimized all serialization of all messaging. ClusterData is now transferred as a byte array
directly.


Modified:
    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/io/ClusterData.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/XByteBuffer.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMember.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/DataSender.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationTransmitter.java

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=381446&r1=381445&r2=381446&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
Mon Feb 27 13:07:39 2006
@@ -50,6 +50,13 @@
     
     static int messageSize = 0;
     
+    public static int messagesSent = 0;
+    public static long messageSendTime = 0;
+    
+    public static synchronized void addSendStats(int count, long time) {
+        messagesSent+=count;
+        messageSendTime+=time;
+    }    
     
     
     public LoadTest(ManagedChannel channel, 
@@ -108,7 +115,10 @@
                     }
                 }
                 if ( (counter % statsInterval) == 0 && (counter > 0)) {
-                    printSendStats(counter, messageSize, sendTime);
+                    //add to the global counter
+                    addSendStats(counter,sendTime);
+                    //print from the global counter
+                    printSendStats(LoadTest.messagesSent, LoadTest.messageSize, LoadTest.messageSendTime);
                 }
 
             }
@@ -122,7 +132,7 @@
         float cnt = (float)counter;
         float size = (float)messageSize;
         float time = (float)sendTime / 1000;
-        log.info("****SEND STATS*****"+
+        log.info("****SEND STATS-"+Thread.currentThread().getName()+"*****"+
                  "\n\tMessage count:"+counter+
                  "\n\tTotal bytes  :"+(long)(size*cnt)+
                  "\n\tTotal seconds:"+(time)+
@@ -179,9 +189,9 @@
         if ( (messagesReceived%statsInterval)==0 || (messagesReceived==msgCount)) {
             float bytes = (float)(((LoadMessage)msg).getMessage().length*messagesReceived);
             float seconds = ((float)(System.currentTimeMillis()-receiveStart)) / 1000f;
-            log.info("****RECEIVE STATS*****"+
+            log.info("****RECEIVE STATS-"+Thread.currentThread().getName()+"*****"+
                      "\n\tMessage count :"+(long)messagesReceived+
-                     "\n\tTotal bytes   :"+bytes+
+                     "\n\tTotal bytes   :"+(long)bytes+
                      "\n\tTime since 1st:"+seconds+" seconds"+
                      "\n\tBytes/second  :"+(bytes/seconds)+
                      "\n\tMBytes/second :"+(bytes/seconds/1024f/1024f));
@@ -237,7 +247,7 @@
         
         public byte[] getMessage() {
             byte[] data = new byte[size+4];
-            System.arraycopy(XByteBuffer.toBytes(msgNr),0,data,0,4);
+            XByteBuffer.toBytes(msgNr,data,0);
             if ( message != null ) {
                 System.arraycopy(message, 0, data, 4, message.length);
             }else {
@@ -271,6 +281,7 @@
                            "[-gzip]  \n\t\t"+
                            "[-pause nrofsecondstopausebetweensends]  \n\t\t"+
                            "[-sender pooled|fastasyncqueue]  \n\t\t"+
+                           "[-threads numberofsenderthreads]  \n\t\t"+
                            "[-break (halts execution on exception)]\n"+
                            "Example:\n\t"+
                            "java LoadTest -port 4004\n\t"+
@@ -291,6 +302,7 @@
         int count = 1000000;
         int stats = 10000;
         boolean breakOnEx = false;
+        int threads = 1;
         String sender = "pooled";
         if ( args.length == 0 ) {
             args = new String[] {"-help"};
@@ -302,6 +314,8 @@
                 sender = args[++i];
             } else if ("-port".equals(args[i])) {
                 port = Integer.parseInt(args[++i]);
+            } else if ("-threads".equals(args[i])) {
+                threads = Integer.parseInt(args[++i]);
             } else if ("-count".equals(args[i])) {
                 count = Integer.parseInt(args[++i]);
             } else if ("-pause".equals(args[i])) {
@@ -366,7 +380,15 @@
         channel.setChannelListener(test);
         channel.setMembershipListener(test);
         channel.start(channel.DEFAULT);
+        while ( threads > 1 ) {
+            Thread t = new Thread(test);
+            t.setDaemon(true);
+            t.start();
+            threads--;
+            test = new LoadTest(channel,send,count,debug,pause,stats,breakOnEx);
+        }
         test.run();
+        
         System.out.println("System test complete, sleeping to let threads finish.");
         Thread.sleep(60*1000*60);
     }    

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=381446&r1=381445&r2=381446&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
Mon Feb 27 13:07:39 2006
@@ -16,22 +16,22 @@
 package org.apache.catalina.tribes.group;
 
 
+import java.io.Serializable;
+import java.util.Iterator;
+
+import org.apache.catalina.tribes.ByteMessage;
 import org.apache.catalina.tribes.ChannelException;
 import org.apache.catalina.tribes.ChannelInterceptor;
+import org.apache.catalina.tribes.ChannelListener;
 import org.apache.catalina.tribes.ChannelMessage;
 import org.apache.catalina.tribes.ChannelReceiver;
 import org.apache.catalina.tribes.ChannelSender;
+import org.apache.catalina.tribes.ManagedChannel;
 import org.apache.catalina.tribes.Member;
 import org.apache.catalina.tribes.MembershipListener;
 import org.apache.catalina.tribes.MembershipService;
 import org.apache.catalina.tribes.io.ClusterData;
 import org.apache.catalina.tribes.io.XByteBuffer;
-import java.io.Serializable;
-import org.apache.catalina.tribes.ChannelListener;
-import org.apache.catalina.tribes.ManagedChannel;
-import java.util.Iterator;
-import java.util.UUID;
-import org.apache.catalina.tribes.ByteMessage;
 
 /**
  * The GroupChannel manages the replication channel. It coordinates
@@ -81,15 +81,6 @@
     }
     
     
-    public byte[] getUUID() {
-        UUID id = UUID.randomUUID();
-        long msb = id.getMostSignificantBits();
-        long lsb = id.getLeastSignificantBits();
-        byte[] data = new byte[16];
-        System.arraycopy(XByteBuffer.toBytes(msb),0,data,0,8);
-        System.arraycopy(XByteBuffer.toBytes(lsb),0,data,8,8);
-        return data;
-    }
     /**
      * Send a message to one or more members in the cluster
      * @param destination Member[] - the destinations, null or zero length means all
@@ -101,9 +92,8 @@
         if ( msg == null ) return;
         try {
             int options = 0;
-            ClusterData data = new ClusterData();
+            ClusterData data = new ClusterData();//generates a unique Id
             data.setAddress(getLocalMember());
-            data.setUniqueId(getUUID());
             data.setTimestamp(System.currentTimeMillis());
             byte[] b = null;
             if ( msg instanceof ByteMessage ){

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=381446&r1=381445&r2=381446&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
Mon Feb 27 13:07:39 2006
@@ -24,6 +24,7 @@
 import org.apache.catalina.tribes.mcast.McastMember;
 import java.io.ByteArrayInputStream;
 import java.io.ObjectInputStream;
+import java.util.UUID;
 
 /**
  * The cluster data class is used to transport around the byte array from
@@ -42,7 +43,15 @@
     private byte[] uniqueId ;
     private Member address;
 
-    public ClusterData() {}
+    public ClusterData() {
+        this(true);
+    }
+    
+    public ClusterData(boolean generateUUID) {
+        if ( generateUUID ) generateUUID();
+    }
+    
+    
     
     /**
      * @param type message type (class)
@@ -118,6 +127,17 @@
         this.address = address;
     }
     
+    public void generateUUID() {
+        UUID id = UUID.randomUUID();
+        long msb = id.getMostSignificantBits();
+        long lsb = id.getLeastSignificantBits();
+        byte[] data = new byte[16];
+        System.arraycopy(XByteBuffer.toBytes(msb),0,data,0,8);
+        System.arraycopy(XByteBuffer.toBytes(lsb),0,data,8,8);
+        setUniqueId(data);
+    }
+
+    
     
     /**
      * 
@@ -130,36 +150,57 @@
      * @return byte[]
      */
     public byte[] getDataPackage() throws IOException {
-        ByteArrayOutputStream bout = new ByteArrayOutputStream(getMessage().length*2);
-        ObjectOutputStream out = new ObjectOutputStream(bout);
-        out.writeInt(options);
-        out.writeLong(timestamp);
-        out.writeInt(uniqueId.length);
-        out.write(uniqueId);
         byte[] addr = ((McastMember)address).getData();
-        out.writeInt(addr.length);
-        out.write(addr);
-        out.writeInt(message.length);
-        out.write(message);
-        out.flush();
-        return bout.toByteArray();
-    }
-    
-    public static ClusterData getDataFromPackage(byte[] dataPackage) throws IOException {
-        ByteArrayInputStream bin = new ByteArrayInputStream(dataPackage);
-        ObjectInputStream in = new ObjectInputStream(bin);
-        ClusterData data = new ClusterData();
-        data.setOptions(in.readInt());
-        data.setTimestamp(in.readLong());
-        byte[] uniqueId = new byte[in.readInt()];
-        in.read(uniqueId);
-        data.setUniqueId(uniqueId);
-        byte[] addr = new byte[in.readInt()];
-        in.read(addr);
+        int length = 
+            4 + //options
+            8 + //timestamp  off=4
+            4 + //unique id length off=12
+            uniqueId.length+ //id data off=12+uniqueId.length
+            4 + //addr length off=12+uniqueId.length+4
+            addr.length+ //member data off=12+uniqueId.length+4+add.length
+            4 + //message length off=12+uniqueId.length+4+add.length+4
+            message.length;
+        byte[] data = new byte[length];
+        int offset = 0;
+        XByteBuffer.toBytes(options,data,offset);
+        offset = 4; //options
+        XByteBuffer.toBytes(timestamp,data,offset);
+        offset += 8; //timestamp
+        XByteBuffer.toBytes(uniqueId.length,data,offset);
+        offset += 4; //uniqueId.length
+        System.arraycopy(uniqueId,0,data,offset,uniqueId.length);
+        offset += uniqueId.length; //uniqueId data
+        XByteBuffer.toBytes(addr.length,data,offset);
+        offset += 4; //addr.length
+        System.arraycopy(addr,0,data,offset,addr.length);
+        offset += addr.length; //addr data
+        XByteBuffer.toBytes(message.length,data,offset);
+        offset += 4; //message.length
+        System.arraycopy(message,0,data,offset,message.length);
+        offset += message.length; //message data
+        return data;
+    }
+    
+    public static ClusterData getDataFromPackage(byte[] b) throws IOException {
+        ClusterData data = new ClusterData(false);
+        int offset = 0;
+        data.setOptions(XByteBuffer.toInt(b,offset));
+        offset += 4; //options
+        data.setTimestamp(XByteBuffer.toLong(b,offset));
+        offset += 8; //timestamp
+        data.uniqueId = new byte[XByteBuffer.toInt(b,offset)];
+        offset += 4; //uniqueId length
+        System.arraycopy(b,offset,data.uniqueId,0,data.uniqueId.length);
+        offset += data.uniqueId.length; //uniqueId data
+        byte[] addr = new byte[XByteBuffer.toInt(b,offset)];
+        offset += 4; //addr length
+        System.arraycopy(b,offset,addr,0,addr.length);
         data.setAddress(McastMember.getMember(addr));
-        byte[] message = new byte[in.readInt()];
-        in.read(message);
-        data.setMessage(message);
+        offset += addr.length; //addr data
+        data.message = new byte[XByteBuffer.toInt(b,offset)];
+        offset += 4; //message length
+        System.arraycopy(b,offset,data.message,0,data.message.length);
+        offset += data.message.length; //message data
         return data;
     }
     

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=381446&r1=381445&r2=381446&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
Mon Feb 27 13:07:39 2006
@@ -276,14 +276,17 @@
      * @return - four bytes in an array
      */
     public static byte[] toBytes(int n) {
-        byte[] b = new byte[4];
-        b[3] = (byte) (n);
+        return toBytes(n,new byte[4],0);
+    }
+
+    public static byte[] toBytes(int n,byte[] b, int offset) {
+        b[offset+3] = (byte) (n);
         n >>>= 8;
-        b[2] = (byte) (n);
+        b[offset+2] = (byte) (n);
         n >>>= 8;
-        b[1] = (byte) (n);
+        b[offset+1] = (byte) (n);
         n >>>= 8;
-        b[0] = (byte) (n);
+        b[offset+0] = (byte) (n);
         return b;
     }
 
@@ -293,22 +296,24 @@
      * @return - eight bytes in an array
      */
     public static byte[] toBytes(long n) {
-        byte[] b = new byte[8];
-        b[7] = (byte) (n);
+        return toBytes(n,new byte[8],0);
+    }
+    public static byte[] toBytes(long n, byte[] b, int offset) {
+        b[offset+7] = (byte) (n);
         n >>>= 8;
-        b[6] = (byte) (n);
+        b[offset+6] = (byte) (n);
         n >>>= 8;
-        b[5] = (byte) (n);
+        b[offset+5] = (byte) (n);
         n >>>= 8;
-        b[4] = (byte) (n);
+        b[offset+4] = (byte) (n);
         n >>>= 8;
-        b[3] = (byte) (n);
+        b[offset+3] = (byte) (n);
         n >>>= 8;
-        b[2] = (byte) (n);
+        b[offset+2] = (byte) (n);
         n >>>= 8;
-        b[1] = (byte) (n);
+        b[offset+1] = (byte) (n);
         n >>>= 8;
-        b[0] = (byte) (n);
+        b[offset+0] = (byte) (n);
         return b;
     }
 

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMember.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMember.java?rev=381446&r1=381445&r2=381446&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMember.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/mcast/McastMember.java
Mon Feb 27 13:07:39 2006
@@ -64,7 +64,8 @@
     /**
      * The name of the cluster domain from this node
      */
-    private String domain;
+    protected byte[] domain;
+    protected transient String domainname;
     
     /**
      * Counter for how many messages have been sent from this member
@@ -101,7 +102,7 @@
                        long aliveTime) throws IOException {
         setHostname(host);
         this.port = port;
-        this.domain = domain;
+        this.domain = domain.getBytes();
         this.memberAliveTime=aliveTime;
     }
 
@@ -141,7 +142,7 @@
         //host - 4 bytes
         //dlen - 4 bytes
         //domain - dlen bytes
-        byte[] domaind = getDomain().getBytes();
+        byte[] domaind = this.domain;
         byte[] addr = host;
         byte[] data = new byte[8+4+addr.length+4+domaind.length];
         long alive=System.currentTimeMillis()-getServiceStartTime();
@@ -201,7 +202,8 @@
      * @return a cluster domain to the cluster
      */
     public String getDomain() {
-        return domain;
+        if ( this.domainname == null ) this.domainname = new String(domain);
+        return this.domainname;
     }
     
     /**
@@ -256,7 +258,7 @@
      * String representation of this object
      */
     public String toString()  {
-        return "org.apache.catalina.tribes.mcast.McastMember["+getName()+","+domain+","+getHostname()+","+port+",
alive="+memberAliveTime+"]";
+        return "org.apache.catalina.tribes.mcast.McastMember["+getName()+","+getDomain()+","+getHostname()+","+port+",
alive="+memberAliveTime+"]";
     }
 
     /**
@@ -355,7 +357,7 @@
     }
 
     public void setDomain(String domain) {
-        this.domain = domain;
+        this.domain = domain.getBytes();
     }
     public void setPort(int port) {
         this.port = port;

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/DataSender.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/DataSender.java?rev=381446&r1=381445&r2=381446&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/DataSender.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/DataSender.java
Mon Feb 27 13:07:39 2006
@@ -72,6 +72,7 @@
      * current sender socket
      */
     private Socket socket = null;
+    private OutputStream socketout = null;
 
     /**
      * is Socket really connected
@@ -698,6 +699,7 @@
      */
     protected void createSocket() throws IOException, SocketException {
         socket = new Socket(getAddress(), getPort());
+        this.socketout = socket.getOutputStream();
     }
 
     /**
@@ -846,9 +848,8 @@
             isMessageTransferStarted = true ;
         }
         try {
-            OutputStream out = socket.getOutputStream();
-            out.write(XByteBuffer.createDataPackage((ClusterData)data));
-            out.flush();
+            socketout.write(XByteBuffer.createDataPackage((ClusterData)data));
+            socketout.flush();
             if (isWaitForAck()) waitForAck(ackTimeout);
         } finally {
             synchronized(this) {

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationTransmitter.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationTransmitter.java?rev=381446&r1=381445&r2=381446&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationTransmitter.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationTransmitter.java
Mon Feb 27 13:07:39 2006
@@ -376,7 +376,7 @@
             time = System.currentTimeMillis();
         }
         try {
-            String key = getKey(member);
+            Object key = getKey(member);
             IDataSender sender = (IDataSender) map.get(key);
             sendMessageData(message, sender);
         } finally {
@@ -506,7 +506,7 @@
      */
     public synchronized void add(Member member) {
         try {
-            String key = getKey(member);
+            Object key = getKey(member);
             if (!map.containsKey(key)) {
                 IDataSender sender = IDataSenderFactory.getIDataSender(
                         replicationMode, member);
@@ -524,7 +524,7 @@
      * @see org.apache.catalina.tribes.ClusterSender#remove(org.apache.catalina.tribes.Member)
      */
     public synchronized void remove(Member member) {
-        String key = getKey(member);
+        Object key = getKey(member);
         IDataSender toberemoved = (IDataSender) map.get(key);
         if (toberemoved == null)
             return;
@@ -570,8 +570,8 @@
      * @param member
      * @return concat member.host:member.port
      */
-    protected String getKey(Member member) {
-        return member.getHost() + ":" + member.getPort();
+    protected Object getKey(Member member) {
+        return member;
     }
 
     /**



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


Mime
View raw message