tomcat-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fha...@apache.org
Subject svn commit: r381364 - in /tomcat/container/tc5.5.x/modules: groupcom/src/share/org/apache/catalina/tribes/ groupcom/src/share/org/apache/catalina/tribes/group/ groupcom/src/share/org/apache/catalina/tribes/io/ groupcom/src/share/org/apache/catalina/tri...
Date Mon, 27 Feb 2006 15:43:05 GMT
Author: fhanik
Date: Mon Feb 27 07:43:00 2006
New Revision: 381364

URL: http://svn.apache.org/viewcvs?rev=381364&view=rev
Log:
Created a load test, also implemented a "synchronized/asynchronized" option, since we are
using TCP.

Added:
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ByteMessage.java
Modified:
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/Channel.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelReceiver.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/group/ChannelCoordinator.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/ListenCallback.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
    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/ReplicationListener.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationListener.jbx
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationTransmitter.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/TcpReplicationThread.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ThreadPool.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/WorkerThread.java
    tomcat/container/tc5.5.x/modules/ha/etc/cluster-server.xml

Added: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ByteMessage.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ByteMessage.java?rev=381364&view=auto
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ByteMessage.java
(added)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ByteMessage.java
Mon Feb 27 07:43:00 2006
@@ -0,0 +1,44 @@
+/*
+ * Copyright 1999,2004-2005 The Apache Software Foundation.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes;
+
+import java.io.Serializable;
+
+/**
+ * A byte message is not serialized and deserialized by the channel
+ * @author Filip Hanik
+ * @version $Revision: 304032 $, $Date: 2005-07-27 10:11:55 -0500 (Wed, 27 Jul 2005) $
+ */
+
+public class ByteMessage implements Serializable {
+    private byte[] message;
+    
+    public ByteMessage() {
+        
+    }
+    public ByteMessage(byte[] data) {
+        message = data;
+    }
+    
+    public byte[] getMessage() {
+        return message;
+    }
+
+    public void setMessage(byte[] message) {
+        this.message = message;
+    }
+
+}
\ No newline at end of file

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/Channel.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/Channel.java?rev=381364&r1=381363&r2=381364&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/Channel.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/Channel.java
Mon Feb 27 07:43:00 2006
@@ -74,7 +74,7 @@
      * @param options int - sender options, see class documentation
      * @return ClusterMessage[] - the replies from the members, if any. 
      */
-    public void send(Member[] destination, Serializable msg, int options) throws ChannelException;
+    public void send(Member[] destination, Serializable msg) throws ChannelException;
 
     
     /**

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelReceiver.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelReceiver.java?rev=381364&r1=381363&r2=381364&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelReceiver.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/ChannelReceiver.java
Mon Feb 27 07:43:00 2006
@@ -20,7 +20,6 @@
 /**
  * Cluster Receiver Interface
  * @author Filip Hanik
- * @author Peter Rossbach
  * @version $Revision: 379904 $, $Date: 2006-02-22 15:16:25 -0600 (Wed, 22 Feb 2006) $
  */
 public interface ChannelReceiver {
@@ -35,8 +34,11 @@
      */
     public void stop();
 
-    
-    public boolean isSendAck();
+    /**
+     * returns true of the receiver is sending acks when it receives messages
+     * @return boolean
+     */
+    public boolean getSendAck();
     
     /**
      * set ack mode
@@ -44,9 +46,6 @@
      */
     public void setSendAck(boolean isSendAck);
     
-    public boolean isCompress() ;
-    public void setCompress(boolean compress);
-
     /**
      * get the listing ip interface
      * @return The host

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=381364&r1=381363&r2=381364&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
Mon Feb 27 07:43:00 2006
@@ -40,10 +40,6 @@
 
     public void sendMessage(ChannelMessage message) throws java.io.IOException;
     
-    public boolean isWaitForAck();
+    public boolean getWaitForAck();
     public void setWaitForAck(boolean isWaitForAck);
-
-    public boolean isCompress() ;
-    public void setCompress(boolean compress);
-
 }

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/ChannelCoordinator.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/ChannelCoordinator.java?rev=381364&r1=381363&r2=381364&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/ChannelCoordinator.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/group/ChannelCoordinator.java
Mon Feb 27 07:43:00 2006
@@ -83,13 +83,10 @@
         try {
             //synchronize, big time FIXME
             membershipService.setLocalMemberProperties(getClusterReceiver().getHost(), getClusterReceiver().getPort());
-            clusterReceiver.setSendAck(clusterSender.isWaitForAck());
-            clusterReceiver.setCompress(clusterSender.isCompress());
             //end FIXME
-            
-            if ( (svc & Channel.MBR_RX_SEQ) == Channel.MBR_RX_SEQ) membershipService.start(MembershipService.MBR_RX);
             if ( (svc & Channel.SND_RX_SEQ) == Channel.SND_RX_SEQ) clusterReceiver.start();
             if ( (svc & Channel.SND_TX_SEQ) == Channel.SND_TX_SEQ) clusterSender.start();
+            if ( (svc & Channel.MBR_RX_SEQ) == Channel.MBR_RX_SEQ) membershipService.start(MembershipService.MBR_RX);
             if ( (svc & Channel.MBR_TX_SEQ) == Channel.MBR_TX_SEQ) membershipService.start(MembershipService.MBR_TX);
         }catch ( ChannelException cx ) {
             throw cx;

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=381364&r1=381363&r2=381364&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 07:43:00 2006
@@ -33,6 +33,7 @@
 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
@@ -43,6 +44,8 @@
  * @version $Revision: 304032 $, $Date: 2005-07-27 10:11:55 -0500 (Wed, 27 Jul 2005) $
  */
 public class GroupChannel extends ChannelInterceptorBase implements ManagedChannel {
+    public static final int BYTE_MESSAGE = 0x0001;
+    
     private ChannelCoordinator coordinator = new ChannelCoordinator();
     private ChannelInterceptor interceptors = null;
     private MembershipListener membershipListener;
@@ -86,15 +89,23 @@
      * @param options int - sender options, see class documentation
      * @return ClusterMessage[] - the replies from the members, if any.
      */
-    public void send(Member[] destination, Serializable msg, int options) throws ChannelException
{
+    public void send(Member[] destination, Serializable msg) throws ChannelException {
         if ( msg == null ) return;
         try {
+            int options = 0;
             ClusterData data = new ClusterData();
             data.setAddress(getLocalMember());
             data.setUniqueId(UUID.randomUUID().toString());
             data.setTimestamp(System.currentTimeMillis());
+            byte[] b = null;
+            if ( msg instanceof ByteMessage ){
+                b = ((ByteMessage)msg).getMessage();
+                options = options | BYTE_MESSAGE;
+            } else {
+                b = XByteBuffer.serialize(msg);
+            }
             data.setOptions(options);
-            byte[] b = XByteBuffer.serialize(msg);
+            
             data.setMessage(b);
             getFirstInterceptor().sendMessage(destination, data, null);
         }catch ( Exception x ) {
@@ -105,7 +116,13 @@
     public void messageReceived(ChannelMessage msg) {
         if ( msg == null ) return;
         try {
-            Serializable fwd = XByteBuffer.deserialize(msg.getMessage());
+            
+            Serializable fwd = null;
+            if ( (msg.getOptions() & BYTE_MESSAGE) == BYTE_MESSAGE ) {
+                fwd = new ByteMessage(msg.getMessage());
+            } else {
+                fwd = XByteBuffer.deserialize(msg.getMessage());
+            }
             if ( channelListener != null ) channelListener.messageReceived(fwd,msg.getAddress());
         }catch ( Exception x ) {
             log.error("Unable to deserialize channel message.",x);

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ListenCallback.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ListenCallback.java?rev=381364&r1=381363&r2=381364&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ListenCallback.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/io/ListenCallback.java
Mon Feb 27 07:43:00 2006
@@ -25,7 +25,6 @@
  * when data has been received. The interface does not care about
  * objects and marshalling and just passes the bytes straight through.
  * @author Filip Hanik
- * @author Peter Rossbach
  * @version $Revision: 303987 $, $Date: 2005-07-08 15:50:30 -0500 (Fri, 08 Jul 2005) $
  */
 public interface ListenCallback
@@ -37,13 +36,4 @@
      */
      public void messageDataReceived(ChannelMessage data);
      
-    /** receiver must be send ack
-      */
-     public boolean isSendAck() ;
-     
-     /** send ack
-      *
-      */
-     public void sendAck() throws java.io.IOException ;
-
 }

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=381364&r1=381363&r2=381364&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
Mon Feb 27 07:43:00 2006
@@ -110,6 +110,14 @@
     }
     
     /**
+     * Returns the number of packages that the reader has read
+     * @return int
+     */
+    public int count() {
+        return buffer.countPackages();
+    }
+    
+    /**
      * Write Ack to sender
      * @param buf
      * @return The bytes written count

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=381364&r1=381363&r2=381364&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 07:43:00 2006
@@ -39,16 +39,12 @@
  * Transfer package:
  * <ul>
  * <li><b>START_DATA/b> - 7 bytes - <i>FLT2002</i></li>
- * <li><b>COMPRESS</b>  - 4 bytes - is message compressed flag</li>
+ * <li><b>OPTIONS</b>  - 4 bytes - message options, implementation specific</li>
  * <li><b>SIZE</b>      - 4 bytes - size of the data package</li>
  * <li><b>DATA</b>      - should be as many bytes as the prev SIZE</li>
  * <li><b>END_DATA</b>  - 7 bytes - <i>TLF2003</i></lI>
  * </ul>
- * FIXME: Why we not use a list of byte buffers?
- * FIXME: Used a pool of buffers instead, every time new generation
- *
  * @author Filip Hanik
- * @author Peter Rossbach
  * @version $Revision: 377484 $, $Date: 2006-02-13 15:00:05 -0600 (Mon, 13 Feb 2006) $
  */
 public class XByteBuffer

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=381364&r1=381363&r2=381364&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 07:43:00 2006
@@ -256,7 +256,7 @@
      * String representation of this object
      */
     public String toString()  {
-        return "org.apache.catalina.tribes.mcast.McastMember["+getName()+","+domain+","+host+","+port+",
alive="+memberAliveTime+"]";
+        return "org.apache.catalina.tribes.mcast.McastMember["+getName()+","+domain+","+getHostname()+","+port+",
alive="+memberAliveTime+"]";
     }
 
     /**

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=381364&r1=381363&r2=381364&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 07:43:00 2006
@@ -738,8 +738,8 @@
     protected void addStats(int length) {
         nrOfRequests++;
         totalBytes += length;
-        if (log.isInfoEnabled() && (nrOfRequests % 1000) == 0) {
-            log.info(sm.getString("IDataSender.stats", new Object[] {
+        if (log.isDebugEnabled() && (nrOfRequests % 1000) == 0) {
+            log.debug(sm.getString("IDataSender.stats", new Object[] {
                     getAddress().getHostAddress(), new Integer(getPort()),
                     new Long(totalBytes), new Long(nrOfRequests),
                     new Long(totalBytes / nrOfRequests),

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationListener.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationListener.java?rev=381364&r1=381363&r2=381364&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationListener.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationListener.java
Mon Feb 27 07:43:00 2006
@@ -29,12 +29,9 @@
 import org.apache.catalina.tribes.ChannelMessage;
 import org.apache.catalina.tribes.ChannelReceiver;
 import org.apache.catalina.tribes.MessageListener;
-
 import org.apache.catalina.tribes.io.ListenCallback;
 import org.apache.catalina.tribes.io.ObjectReader;
-import org.apache.catalina.tribes.io.XByteBuffer;
 import org.apache.catalina.util.StringManager;
-import java.io.Serializable;
 
 /**
  * @author Filip Hanik
@@ -74,6 +71,7 @@
 
     private Object interestOpsMutex = new Object();
     private MessageListener listener = null;
+    private boolean sync;
     public ReplicationListener() {
     }
 
@@ -119,7 +117,7 @@
      */
     public void start() {
         try {
-            pool = new ThreadPool(tcpThreadCount, TcpReplicationThread.class, interestOpsMutex);
+            pool = new ThreadPool(tcpThreadCount, TcpReplicationThread.class, interestOpsMutex,
getWorkerThreadOptions());
         } catch (Exception e) {
             log.error("ThreadPool can initilzed. Listener not started", e);
             return;
@@ -290,7 +288,7 @@
                 log.debug("No TcpReplicationThread available");
         } else {
             // invoking this wakes up the worker thread then returns
-            worker.serviceChannel(key, isSendAck());
+            worker.serviceChannel(key);
         }
     }
 
@@ -348,7 +346,7 @@
      *
      * @return True if sending ACK
      */
-    public boolean isSendAck() {
+    public boolean getSendAck() {
         return sendAck;
     }
 
@@ -373,12 +371,32 @@
         return tcpListenPort;
     }
 
+    public boolean isSync() {
+        return sync;
+    }
+
     public MessageListener getMessageListener() {
         return listener;
     }
 
     public void setTcpListenPort(int tcpListenPort) {
         this.tcpListenPort = tcpListenPort;
+    }
+
+
+    public void setSynchronized(boolean sync) {
+        this.sync = sync;
+    }
+
+    public boolean getSynchronized() {
+        return this.sync;
+    }
+    
+    public int getWorkerThreadOptions() {
+        int options = 0;
+        if ( getSynchronized() ) options = options |TcpReplicationThread.OPTION_SYNCHRONIZED;
+        if ( getSendAck() ) options = options |TcpReplicationThread.OPTION_SEND_ACK;
+        return options;
     }
 
     public void setMessageListener(MessageListener listener) {

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationListener.jbx
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationListener.jbx?rev=381364&r1=381363&r2=381364&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationListener.jbx
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ReplicationListener.jbx
Mon Feb 27 07:43:00 2006
@@ -1,18 +1,20 @@
 [PropertyInfo]
 bind,java.net.InetAddress,false,false, , ,true,<default>
 compress,boolean,false,false, , ,true,<default>
-coordinator,org.apache.catalina.cluster.MessageListener,false,false, , ,true,<default>
-coordinator,org.apache.catalina.cluster.group.ChannelInterceptorBase,false,false,coordinator,coordinator,true,<default>
 doListen,boolean,false,false, , ,false,<default>
 host,String,false,false, , ,true,<default>
 info,String,false,false, , ,true,<default>
 interestOpsMutex,Object,false,false, , ,true,<default>
+listener,org.apache.catalina.tribes.MessageListener,false,false, , ,false,<default>
 log,org.apache.commons.logging.Log,false,false, , ,false,<default>
-pool,org.apache.catalina.cluster.tcp.ThreadPool,false,false, , ,false,<default>
+messageListener,org.apache.catalina.tribes.MessageListener,false,false, , ,true,<default>
+pool,org.apache.catalina.tribes.tcp.ThreadPool,false,false, , ,false,<default>
 port,int,false,false, , ,true,<default>
 selector,java.nio.channels.Selector,false,false, , ,false,<default>
 sendAck,boolean,false,false, , ,true,<default>
 sm,org.apache.catalina.util.StringManager,false,false, , ,false,<default>
+sync,boolean,false,false,sync,sync,true,<default>
+synchronized,boolean,false,false, , ,true,<default>
 tcpListenAddress,String,false,false, , ,true,<default>
 tcpListenPort,int,false,false, , ,true,<default>
 tcpSelectorTimeout,long,false,false, , ,true,<default>

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=381364&r1=381363&r2=381364&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 07:43:00 2006
@@ -16,38 +16,26 @@
 
 package org.apache.catalina.tribes.tcp;
 
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
-import java.io.ObjectOutputStream;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
-import java.util.zip.GZIPOutputStream;
-
-import javax.management.MBeanServer;
 import javax.management.ObjectName;
 
-import org.apache.catalina.Container;
 import org.apache.catalina.tribes.ChannelMessage;
 import org.apache.catalina.tribes.ChannelSender;
 import org.apache.catalina.tribes.Member;
 import org.apache.catalina.tribes.util.IDynamicProperty;
-import org.apache.catalina.core.StandardHost;
 import org.apache.catalina.util.StringManager;
 import org.apache.tomcat.util.IntrospectionUtils;
-import org.apache.catalina.tribes.io.XByteBuffer;
-import org.apache.catalina.tribes.io.*;
 
 /**
  * Transmit message to other cluster members
  * Actual senders are created based on the replicationMode
  * type 
- * FIXME i18n log messages
- * FIXME compress data depends on message type and size 
- * TODO pause and resume senders
  * 
- * @author Peter Rossbach
  * @author Filip Hanik
+ * @author Peter Rossbach
  * @version $Revision: 379956 $ $Date: 2006-02-22 16:57:35 -0600 (Wed, 22 Feb 2006) $
  */
 public class ReplicationTransmitter implements ChannelSender,IDynamicProperty {
@@ -112,14 +100,7 @@
     /**
      * autoConnect sender when next message send
      */
-    private boolean autoConnect = false;
-
-    /**
-     * Compress message data bytes
-     */
-    private boolean compress = false;
-
-    /**
+    private boolean autoConnect = false; /**
      * doTransmitterProcessingStats
      */
     protected boolean doTransmitterProcessingStats = false;
@@ -263,22 +244,6 @@
     public ObjectName getObjectName() {
         return objectName;
     }
-
-    /**
-     * @return Returns the compress.
-     */
-    public boolean isCompress() {
-        return compress;
-    }
-
-    /**
-     * @param compressMessageData
-     *            The compress to set.
-     */
-    public void setCompress(boolean compressMessageData) {
-        this.compress = compressMessageData;
-    }
-
     /**
      * @return Returns the autoConnect.
      */
@@ -314,7 +279,7 @@
     /**
      * @return Returns the waitForAck.
      */
-    public boolean isWaitForAck() {
+    public boolean getWaitForAck() {
         return waitForAck;
     }
 

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/TcpReplicationThread.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/TcpReplicationThread.java?rev=381364&r1=381363&r2=381364&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/TcpReplicationThread.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/TcpReplicationThread.java
Mon Feb 27 07:43:00 2006
@@ -37,14 +37,14 @@
  * @version $Revision: 378050 $, $Date: 2006-02-15 12:30:02 -0600 (Wed, 15 Feb 2006) $
  */
 public class TcpReplicationThread extends WorkerThread {
+    public static final int OPTION_SEND_ACK = 0x0001;
+    public static final int OPTION_SYNCHRONIZED = 0x0002;
+
     public static final byte[] ACK_COMMAND = new byte[] {6, 2, 3};
     private static org.apache.commons.logging.Log log =
         org.apache.commons.logging.LogFactory.getLog( TcpReplicationThread.class );
     private ByteBuffer buffer = ByteBuffer.allocate (1024);
     private SelectionKey key;
-    private boolean sendAck=true;
-
-    
     TcpReplicationThread ()
     {
     }
@@ -104,10 +104,9 @@
      * to ignore read-readiness for this channel while the
      * worker thread is servicing it.
      */
-    synchronized void serviceChannel (SelectionKey key, boolean sendAck)
+    synchronized void serviceChannel (SelectionKey key)
     {
         this.key = key;
-        this.sendAck=sendAck;
         key.interestOps (key.interestOps() & (~SelectionKey.OP_READ));
         key.interestOps (key.interestOps() & (~SelectionKey.OP_WRITE));
         this.notify();		// awaken the thread
@@ -135,19 +134,42 @@
             reader.append(buffer.array(),0,count);
             buffer.clear();		// make buffer empty
         }
+        
+        int pkgcnt = reader.count();
+
+        
+
+        /**
+         * Use send ack here if you want to ack the request to the remote 
+         * server before completing the request
+         * This is considered an asynchronized request
+         */
+        if (sendAckAsync()) {
+            while ( pkgcnt > 0 ) {
+                sendAck(key,channel);
+                pkgcnt--;
+            }
+        }
+
         //check to see if any data is available
-        int pkgcnt = reader.execute();
+        pkgcnt = reader.execute();
+
         if (log.isTraceEnabled()) {
             log.trace("sending " + pkgcnt + " ack packages to " + channel.socket().getLocalPort()
);
         }
 
-        
-        if (sendAck) {
+        /**
+         * Use send ack here if you want the request to complete on this 
+         * server before sending the ack to the remote server
+         * This is considered a synchronized request
+         */
+        if (sendAckSync()) {
             while ( pkgcnt > 0 ) {
                 sendAck(key,channel);
                 pkgcnt--;
             }
-        }
+        }        
+
         
         if (count < 0) {
             // close channel on EOF, invalidates the key
@@ -166,6 +188,20 @@
         }
         
     }
+    
+    
+    public boolean sendAckSync() {
+        int options = getOptions();
+        return ((OPTION_SEND_ACK & options) == OPTION_SEND_ACK) &&
+               ((OPTION_SYNCHRONIZED & options) == OPTION_SYNCHRONIZED);
+    }
+    
+    public boolean sendAckAsync() {
+        int options = getOptions();
+        return ((OPTION_SEND_ACK & options) == OPTION_SEND_ACK) &&
+               ((OPTION_SYNCHRONIZED & options) != OPTION_SYNCHRONIZED);
+    }
+
 
     /**
      * send a reply-acknowledgement (6,2,3)

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ThreadPool.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ThreadPool.java?rev=381364&r1=381363&r2=381364&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ThreadPool.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ThreadPool.java
Mon Feb 27 07:43:00 2006
@@ -35,7 +35,7 @@
     Object mutex = new Object();
     Object interestOpsMutex = null;
 
-    ThreadPool (int poolSize, Class threadClass, Object interestOpsMutex) throws Exception
{
+    ThreadPool (int poolSize, Class threadClass, Object interestOpsMutex, int threadOptions)
throws Exception {
         // fill up the pool with worker threads
         this.interestOpsMutex = interestOpsMutex;
         for (int i = 0; i < poolSize; i++) {
@@ -46,6 +46,7 @@
             thread.setName (threadClass.getName()+"[" + (i + 1)+"]");
             thread.setDaemon(true);
             thread.setPriority(Thread.MAX_PRIORITY);
+            thread.setOptions(threadOptions);
             thread.start();
 
             idle.add (thread);

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/WorkerThread.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/WorkerThread.java?rev=381364&r1=381363&r2=381364&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/WorkerThread.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/WorkerThread.java
Mon Feb 27 07:43:00 2006
@@ -25,14 +25,22 @@
 {
     protected ThreadPool pool;
     protected boolean doRun = true;
-
+    private int options;
 
     public void setPool(ThreadPool pool) {
         this.pool = pool;
     }
-    
+
+    public void setOptions(int options) {
+        this.options = options;
+    }
+
     public ThreadPool getPool() {
         return pool;
+    }
+
+    public int getOptions() {
+        return options;
     }
 
     public void close()

Modified: tomcat/container/tc5.5.x/modules/ha/etc/cluster-server.xml
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/ha/etc/cluster-server.xml?rev=381364&r1=381363&r2=381364&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/ha/etc/cluster-server.xml (original)
+++ tomcat/container/tc5.5.x/modules/ha/etc/cluster-server.xml Mon Feb 27 07:43:00 2006
@@ -298,14 +298,15 @@
                     tcpListenAddress="auto"
                     tcpListenPort="4002"
                     tcpSelectorTimeout="100"
-                    tcpThreadCount="6"/>
+                    tcpThreadCount="6"
+                    sendAck="true"
+                    synchronized="true"/>
 
                 <Sender
                     className="org.apache.catalina.tribes.tcp.ReplicationTransmitter"
                     replicationMode="pooled"
-                    ackTimeout="150000"
-                    waitForAck="true"
-                    compress="true"/>
+                    ackTimeout="15000"
+                    waitForAck="true"/>
                     
                 <Interceptor className="org.apache.catalina.tribes.group.interceptors.GzipInterceptor"/>
             </Channel>



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


Mime
View raw message